wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #241 from pipecat-ai/aleix/transports-async

Browse Source 
transports: fully use asyncio in all read/write operations
This commit is contained in:
Aleix Conchillo Flaqué
2024-06-21 16:00:08 -07:00
committed by GitHub
parent c544b2474b d3e410b2ac
commit dc64e57f63
12 changed files with 335 additions and 346 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
CHANGELOG.md
View File

@@ -20,6 +20,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
- Added new `AzureSTTService`. This allows you to use Azure Speech-To-Text.
### Performance
- Convert `BaseOutputTransport` and `BaseOutputTransport` to fully use asyncio
and remove the use of threads.
### Other
- Added `twilio-chatbot`. This is an example that shows how to integrate Twilio

1
dev-requirements.txt
View File

@@ -2,6 +2,7 @@ autopep8~=2.1.0
build~=1.2.1
grpcio-tools~=1.62.2
pip-tools~=7.4.1
pyright~=1.1.367
pytest~=8.2.0
setuptools~=69.5.1
setuptools_scm~=8.1.0

3
src/pipecat/pipeline/task.py
View File

@@ -71,6 +71,8 @@ class PipelineTask:
await self._source.process_frame(CancelFrame(), FrameDirection.DOWNSTREAM)
self._process_down_task.cancel()
self._process_up_task.cancel()
await self._process_down_task
await self._process_up_task
async def run(self):
self._process_up_task = asyncio.create_task(self._process_up_queue())
@@ -122,6 +124,7 @@ class PipelineTask:
await self._pipeline.cleanup()
# We just enqueue None to terminate the task gracefully.
self._process_up_task.cancel()
await self._process_up_task
async def _process_up_queue(self):
while True:

3
src/pipecat/services/azure.py
View File

@@ -146,10 +146,11 @@ class AzureSTTService(AIService):
async def cancel(self, frame: CancelFrame):
self._speech_recognizer.stop_continuous_recognition_async()
self._push_frame_task.cancel()
await self._push_frame_task
def _create_push_task(self):
self._push_frame_task = self.get_event_loop().create_task(self._push_frame_task_handler())
self._push_queue = asyncio.Queue()
self._push_frame_task = self.get_event_loop().create_task(self._push_frame_task_handler())
async def _push_frame_task_handler(self):
running = True

3
src/pipecat/services/deepgram.py
View File

@@ -134,10 +134,11 @@ class DeepgramSTTService(AIService):
async def cancel(self, frame: CancelFrame):
await self._connection.finish()
self._push_frame_task.cancel()
await self._push_frame_task
def _create_push_task(self):
self._push_frame_task = self.get_event_loop().create_task(self._push_frame_task_handler())
self._push_queue = asyncio.Queue()
self._push_frame_task = self.get_event_loop().create_task(self._push_frame_task_handler())
async def _push_frame_task_handler(self):
running = True

65
src/pipecat/transports/base_input.py
View File

@@ -5,7 +5,6 @@
#
import asyncio
import queue
from concurrent.futures import ThreadPoolExecutor
@@ -33,8 +32,6 @@ class BaseInputTransport(FrameProcessor):
self._params = params
self._running = False
self._executor = ThreadPoolExecutor(max_workers=5)
# Create push frame task. This is the task that will push frames in
@@ -42,34 +39,21 @@ class BaseInputTransport(FrameProcessor):
self._create_push_task()
async def start(self, frame: StartFrame):
if self._running:
return
self._running = True
# Create audio input queue and thread if needed.
# Create audio input queue and task if needed.
if self._params.audio_in_enabled or self._params.vad_enabled:
self._audio_in_queue = queue.Queue()
self._audio_thread = self._loop.run_in_executor(
self._executor, self._audio_thread_handler)
self._audio_in_queue = asyncio.Queue()
self._audio_task = self.get_event_loop().create_task(self._audio_task_handler())
async def stop(self):
if not self._running:
return
# This will exit all threads.
self._running = False
# Wait for the threads to finish.
# Wait for the task to finish.
if self._params.audio_in_enabled or self._params.vad_enabled:
await self._audio_thread
self._push_frame_task.cancel()
self._audio_task.cancel()
await self._audio_task
def vad_analyzer(self) -> VADAnalyzer | None:
return self._params.vad_analyzer
def push_audio_frame(self, frame: AudioRawFrame):
async def push_audio_frame(self, frame: AudioRawFrame):
if self._params.audio_in_enabled or self._params.vad_enabled:
self._audio_in_queue.put_nowait(frame)
@@ -78,7 +62,8 @@ class BaseInputTransport(FrameProcessor):
#
async def cleanup(self):
pass
self._push_frame_task.cancel()
await self._push_frame_task
async def process_frame(self, frame: Frame, direction: FrameDirection):
await super().process_frame(frame, direction)
@@ -102,8 +87,8 @@ class BaseInputTransport(FrameProcessor):
def _create_push_task(self):
loop = self.get_event_loop()
self._push_frame_task = loop.create_task(self._push_frame_task_handler())
self._push_queue = asyncio.Queue()
self._push_frame_task = loop.create_task(self._push_frame_task_handler())
async def _internal_push_frame(
self,
@@ -129,6 +114,7 @@ class BaseInputTransport(FrameProcessor):
if isinstance(frame, UserStartedSpeakingFrame):
logger.debug("User started speaking")
self._push_frame_task.cancel()
await self._push_frame_task
self._create_push_task()
await self.push_frame(StartInterruptionFrame())
elif isinstance(frame, UserStoppedSpeakingFrame):
@@ -140,15 +126,16 @@ class BaseInputTransport(FrameProcessor):
# Audio input
#
def _vad_analyze(self, audio_frames: bytes) -> VADState:
async def _vad_analyze(self, audio_frames: bytes) -> VADState:
state = VADState.QUIET
vad_analyzer = self.vad_analyzer()
if vad_analyzer:
state = vad_analyzer.analyze_audio(audio_frames)
state = await self.get_event_loop().run_in_executor(
self._executor, vad_analyzer.analyze_audio, audio_frames)
return state
def _handle_vad(self, audio_frames: bytes, vad_state: VADState):
new_vad_state = self._vad_analyze(audio_frames)
async def _handle_vad(self, audio_frames: bytes, vad_state: VADState):
new_vad_state = await self._vad_analyze(audio_frames)
if new_vad_state != vad_state and new_vad_state != VADState.STARTING and new_vad_state != VADState.STOPPING:
frame = None
if new_vad_state == VADState.SPEAKING:
@@ -157,33 +144,29 @@ class BaseInputTransport(FrameProcessor):
frame = UserStoppedSpeakingFrame()
if frame:
future = asyncio.run_coroutine_threadsafe(
self._handle_interruptions(frame), self.get_event_loop())
future.result()
await self._handle_interruptions(frame)
vad_state = new_vad_state
return vad_state
def _audio_thread_handler(self):
async def _audio_task_handler(self):
vad_state: VADState = VADState.QUIET
while self._running:
while True:
try:
frame: AudioRawFrame = self._audio_in_queue.get(timeout=1)
frame: AudioRawFrame = await self._audio_in_queue.get()
audio_passthrough = True
# Check VAD and push event if necessary. We just care about
# changes from QUIET to SPEAKING and vice versa.
if self._params.vad_enabled:
vad_state = self._handle_vad(frame.audio, vad_state)
vad_state = await self._handle_vad(frame.audio, vad_state)
audio_passthrough = self._params.vad_audio_passthrough
# Push audio downstream if passthrough.
if audio_passthrough:
future = asyncio.run_coroutine_threadsafe(
self._internal_push_frame(frame), self._loop)
future.result()
except queue.Empty:
pass
await self._internal_push_frame(frame)
except asyncio.CancelledError:
break
except BaseException as e:
logger.error(f"{self} error reading audio frames: {e}")

172
src/pipecat/transports/base_output.py
View File

@@ -7,11 +7,6 @@
import asyncio
import itertools
import queue
import time
import threading
from concurrent.futures import ThreadPoolExecutor
from PIL import Image
from typing import List
@@ -42,67 +37,51 @@ class BaseOutputTransport(FrameProcessor):
self._params = params
self._running = False
self._executor = ThreadPoolExecutor(max_workers=5)
# These are the images that we should send to the camera at our desired
# framerate.
self._camera_images = None
# Create media threads queues.
if self._params.camera_out_enabled:
self._camera_out_queue = queue.Queue()
self._sink_queue = queue.Queue()
self._sink_thread = None
self._stopped_event = asyncio.Event()
self._is_interrupted = threading.Event()
# We will write 20ms audio at a time. If we receive long audio frames we
# will chunk them. This will help with interruption handling.
audio_bytes_10ms = int(self._params.audio_out_sample_rate / 100) * \
self._params.audio_out_channels * 2
self._audio_chunk_size = audio_bytes_10ms * 2
self._stopped_event = asyncio.Event()
# Create sink frame task. This is the task that will actually write
# audio or video frames. We write audio/video in a task so we can keep
# generating frames upstream while, for example, the audio is playing.
self._create_sink_task()
# Create push frame task. This is the task that will push frames in
# order. We also guarantee that all frames are pushed in the same task.
self._create_push_task()
async def start(self, frame: StartFrame):
if self._running:
return
self._running = True
loop = self.get_event_loop()
# Create queues and threads.
# Create media threads queues.
if self._params.camera_out_enabled:
self._camera_out_thread = loop.run_in_executor(
self._executor, self._camera_out_thread_handler)
self._sink_thread = loop.run_in_executor(self._executor, self._sink_thread_handler)
self._camera_out_queue = asyncio.Queue()
self._camera_out_task = self.get_event_loop().create_task(self._camera_out_task_handler())
async def stop(self):
if not self._running:
return
# This will exit all threads.
self._running = False
# Wait on the threads to finish.
if self._params.camera_out_enabled:
self._camera_out_task.cancel()
await self._camera_out_task
self._stopped_event.set()
def send_message(self, frame: TransportMessageFrame):
async def send_message(self, frame: TransportMessageFrame):
pass
def send_metrics(self, frame: MetricsFrame):
async def send_metrics(self, frame: MetricsFrame):
pass
def write_frame_to_camera(self, frame: ImageRawFrame):
async def write_frame_to_camera(self, frame: ImageRawFrame):
pass
def write_raw_audio_frames(self, frames: bytes):
async def write_raw_audio_frames(self, frames: bytes):
pass
#
@@ -110,12 +89,12 @@ class BaseOutputTransport(FrameProcessor):
#
async def cleanup(self):
# Wait on the threads to finish.
if self._params.camera_out_enabled:
await self._camera_out_thread
if self._sink_task:
self._sink_task.cancel()
await self._sink_task
if self._sink_thread:
await self._sink_thread
self._push_frame_task.cancel()
await self._push_frame_task
async def process_frame(self, frame: Frame, direction: FrameDirection):
await super().process_frame(frame, direction)
@@ -128,7 +107,7 @@ class BaseOutputTransport(FrameProcessor):
if isinstance(frame, StartFrame):
await self.start(frame)
await self.push_frame(frame, direction)
# EndFrame is managed in the queue handler.
# EndFrame is managed in the sink queue handler.
elif isinstance(frame, CancelFrame):
await self.stop()
await self.push_frame(frame, direction)
@@ -136,14 +115,14 @@ class BaseOutputTransport(FrameProcessor):
await self._handle_interruptions(frame)
await self.push_frame(frame, direction)
elif isinstance(frame, MetricsFrame):
self.send_metrics(frame)
await self.send_metrics(frame)
await self.push_frame(frame, direction)
elif isinstance(frame, SystemFrame):
await self.push_frame(frame, direction)
elif isinstance(frame, AudioRawFrame):
await self._handle_audio(frame)
else:
self._sink_queue.put_nowait(frame)
await self._sink_queue.put(frame)
# If we are finishing, wait here until we have stopped, otherwise we might
# close things too early upstream. We need this event because we don't
@@ -156,50 +135,51 @@ class BaseOutputTransport(FrameProcessor):
return
if isinstance(frame, StartInterruptionFrame):
self._is_interrupted.set()
# Stop sink task.
self._sink_task.cancel()
await self._sink_task
self._create_sink_task()
# Stop push task.
self._push_frame_task.cancel()
await self._push_frame_task
self._create_push_task()
elif isinstance(frame, StopInterruptionFrame):
self._is_interrupted.clear()
async def _handle_audio(self, frame: AudioRawFrame):
audio = frame.audio
for i in range(0, len(audio), self._audio_chunk_size):
chunk = AudioRawFrame(audio[i: i + self._audio_chunk_size],
sample_rate=frame.sample_rate, num_channels=frame.num_channels)
self._sink_queue.put_nowait(chunk)
await self._sink_queue.put(chunk)
def _sink_thread_handler(self):
def _create_sink_task(self):
loop = self.get_event_loop()
self._sink_queue = asyncio.Queue()
self._sink_task = loop.create_task(self._sink_task_handler())
async def _sink_task_handler(self):
# Audio accumlation buffer
buffer = bytearray()
while self._running:
while True:
try:
frame = self._sink_queue.get(timeout=1)
if not self._is_interrupted.is_set():
if isinstance(frame, AudioRawFrame) and self._params.audio_out_enabled:
buffer.extend(frame.audio)
buffer = self._maybe_send_audio(buffer)
elif isinstance(frame, ImageRawFrame) and self._params.camera_out_enabled:
self._set_camera_image(frame)
elif isinstance(frame, SpriteFrame) and self._params.camera_out_enabled:
self._set_camera_images(frame.images)
elif isinstance(frame, TransportMessageFrame):
self.send_message(frame)
else:
future = asyncio.run_coroutine_threadsafe(
self._internal_push_frame(frame), self.get_event_loop())
future.result()
frame = await self._sink_queue.get()
if isinstance(frame, AudioRawFrame) and self._params.audio_out_enabled:
buffer.extend(frame.audio)
buffer = await self._maybe_send_audio(buffer)
elif isinstance(frame, ImageRawFrame) and self._params.camera_out_enabled:
await self._set_camera_image(frame)
elif isinstance(frame, SpriteFrame) and self._params.camera_out_enabled:
await self._set_camera_images(frame.images)
elif isinstance(frame, TransportMessageFrame):
await self.send_message(frame)
else:
# If we get interrupted just clear the output buffer.
buffer = bytearray()
await self._internal_push_frame(frame)
if isinstance(frame, EndFrame):
future = asyncio.run_coroutine_threadsafe(self.stop(), self.get_event_loop())
future.result()
await self.stop()
self._sink_queue.task_done()
except queue.Empty:
pass
except asyncio.CancelledError:
break
except BaseException as e:
logger.error(f"{self} error processing sink queue: {e}")
@@ -209,8 +189,8 @@ class BaseOutputTransport(FrameProcessor):
def _create_push_task(self):
loop = self.get_event_loop()
self._push_frame_task = loop.create_task(self._push_frame_task_handler())
self._push_queue = asyncio.Queue()
self._push_frame_task = loop.create_task(self._push_frame_task_handler())
async def _internal_push_frame(
self,
@@ -233,7 +213,7 @@ class BaseOutputTransport(FrameProcessor):
async def send_image(self, frame: ImageRawFrame | SpriteFrame):
await self.process_frame(frame, FrameDirection.DOWNSTREAM)
def _draw_image(self, frame: ImageRawFrame):
async def _draw_image(self, frame: ImageRawFrame):
desired_size = (self._params.camera_out_width, self._params.camera_out_height)
if frame.size != desired_size:
@@ -243,32 +223,32 @@ class BaseOutputTransport(FrameProcessor):
f"{frame} does not have the expected size {desired_size}, resizing")
frame = ImageRawFrame(resized_image.tobytes(), resized_image.size, resized_image.format)
self.write_frame_to_camera(frame)
await self.write_frame_to_camera(frame)
def _set_camera_image(self, image: ImageRawFrame):
async def _set_camera_image(self, image: ImageRawFrame):
if self._params.camera_out_is_live:
self._camera_out_queue.put_nowait(image)
await self._camera_out_queue.put(image)
else:
self._camera_images = itertools.cycle([image])
def _set_camera_images(self, images: List[ImageRawFrame]):
async def _set_camera_images(self, images: List[ImageRawFrame]):
self._camera_images = itertools.cycle(images)
def _camera_out_thread_handler(self):
while self._running:
async def _camera_out_task_handler(self):
while True:
try:
if self._params.camera_out_is_live:
image = self._camera_out_queue.get(timeout=1)
self._draw_image(image)
image = await self._camera_out_queue.get()
await self._draw_image(image)
self._camera_out_queue.task_done()
elif self._camera_images:
image = next(self._camera_images)
self._draw_image(image)
time.sleep(1.0 / self._params.camera_out_framerate)
await self._draw_image(image)
await asyncio.sleep(1.0 / self._params.camera_out_framerate)
else:
time.sleep(1.0 / self._params.camera_out_framerate)
except queue.Empty:
pass
await asyncio.sleep(1.0 / self._params.camera_out_framerate)
except asyncio.CancelledError:
break
except Exception as e:
logger.error(f"{self} error writing to camera: {e}")
@@ -279,12 +259,8 @@ class BaseOutputTransport(FrameProcessor):
async def send_audio(self, frame: AudioRawFrame):
await self.process_frame(frame, FrameDirection.DOWNSTREAM)
def _maybe_send_audio(self, buffer: bytearray) -> bytearray:
try:
if len(buffer) >= self._audio_chunk_size:
self.write_raw_audio_frames(bytes(buffer[:self._audio_chunk_size]))
buffer = buffer[self._audio_chunk_size:]
return buffer
except BaseException as e:
logger.error(f"{self} error writing audio frames: {e}")
return buffer
async def _maybe_send_audio(self, buffer: bytearray) -> bytearray:
if len(buffer) >= self._audio_chunk_size:
await self.write_raw_audio_frames(bytes(buffer[:self._audio_chunk_size]))
buffer = buffer[self._audio_chunk_size:]
return buffer

30
src/pipecat/transports/local/audio.py
View File

@@ -6,6 +6,8 @@
import asyncio
from concurrent.futures import ThreadPoolExecutor
from pipecat.frames.frames import AudioRawFrame, StartFrame
from pipecat.processors.frame_processor import FrameProcessor
from pipecat.transports.base_input import BaseInputTransport
@@ -43,26 +45,20 @@ class LocalAudioInputTransport(BaseInputTransport):
await super().start(frame)
self._in_stream.start_stream()
async def stop(self):
await super().stop()
self._in_stream.stop_stream()
async def cleanup(self):
await super().cleanup()
self._in_stream.stop_stream()
# This is not very pretty (taken from PyAudio docs).
while self._in_stream.is_active():
await asyncio.sleep(0.1)
self._in_stream.close()
await super().cleanup()
def _audio_in_callback(self, in_data, frame_count, time_info, status):
if not self._running:
return (None, pyaudio.paAbort)
frame = AudioRawFrame(audio=in_data,
sample_rate=self._params.audio_in_sample_rate,
num_channels=self._params.audio_in_channels)
self.push_audio_frame(frame)
asyncio.run_coroutine_threadsafe(self.push_audio_frame(frame), self.get_event_loop())
return (None, pyaudio.paContinue)
@@ -72,19 +68,29 @@ class LocalAudioOutputTransport(BaseOutputTransport):
def __init__(self, py_audio: pyaudio.PyAudio, params: TransportParams):
super().__init__(params)
self._executor = ThreadPoolExecutor(max_workers=5)
self._out_stream = py_audio.open(
format=py_audio.get_format_from_width(2),
channels=params.audio_out_channels,
rate=params.audio_out_sample_rate,
output=True)
def write_raw_audio_frames(self, frames: bytes):
self._out_stream.write(frames)
async def start(self, frame: StartFrame):
await super().start(frame)
self._out_stream.start_stream()
async def cleanup(self):
await super().cleanup()
self._out_stream.stop_stream()
# This is not very pretty (taken from PyAudio docs).
while self._out_stream.is_active():
await asyncio.sleep(0.1)
self._out_stream.close()
async def write_raw_audio_frames(self, frames: bytes):
await self.get_event_loop().run_in_executor(self._executor, self._out_stream.write, frames)
class LocalAudioTransport(BaseTransport):

33
src/pipecat/transports/local/tk.py
View File

@@ -6,6 +6,8 @@
import asyncio
from concurrent.futures import ThreadPoolExecutor
import numpy as np
import tkinter as tk
@@ -53,25 +55,20 @@ class TkInputTransport(BaseInputTransport):
await super().start(frame)
self._in_stream.start_stream()
async def stop(self):
await super().stop()
self._in_stream.stop_stream()
async def cleanup(self):
await super().cleanup()
self._in_stream.stop_stream()
# This is not very pretty (taken from PyAudio docs).
while self._in_stream.is_active():
await asyncio.sleep(0.1)
self._in_stream.close()
def _audio_in_callback(self, in_data, frame_count, time_info, status):
if not self._running:
return (None, pyaudio.paAbort)
frame = AudioRawFrame(audio=in_data,
sample_rate=self._params.audio_in_sample_rate,
num_channels=self._params.audio_in_channels)
self.push_audio_frame(frame)
asyncio.run_coroutine_threadsafe(self.push_audio_frame(frame), self.get_event_loop())
return (None, pyaudio.paContinue)
@@ -81,6 +78,8 @@ class TkOutputTransport(BaseOutputTransport):
def __init__(self, tk_root: tk.Tk, py_audio: pyaudio.PyAudio, params: TransportParams):
super().__init__(params)
self._executor = ThreadPoolExecutor(max_workers=5)
self._out_stream = py_audio.open(
format=py_audio.get_format_from_width(2),
channels=params.audio_out_channels,
@@ -94,16 +93,24 @@ class TkOutputTransport(BaseOutputTransport):
self._image_label = tk.Label(tk_root, image=photo)
self._image_label.pack()
def write_raw_audio_frames(self, frames: bytes):
self._out_stream.write(frames)
def write_frame_to_camera(self, frame: ImageRawFrame):
self.get_event_loop().call_soon(self._write_frame_to_tk, frame)
async def start(self, frame: StartFrame):
await super().start(frame)
self._out_stream.start_stream()
async def cleanup(self):
await super().cleanup()
self._out_stream.stop_stream()
# This is not very pretty (taken from PyAudio docs).
while self._out_stream.is_active():
await asyncio.sleep(0.1)
self._out_stream.close()
async def write_raw_audio_frames(self, frames: bytes):
await self.get_event_loop().run_in_executor(self._executor, self._out_stream.write, frames)
async def write_frame_to_camera(self, frame: ImageRawFrame):
self.get_event_loop().call_soon(self._write_frame_to_tk, frame)
def _write_frame_to_tk(self, frame: ImageRawFrame):
width = frame.size[0]
height = frame.size[1]

8
src/pipecat/transports/network/fastapi_websocket.py
View File

@@ -75,7 +75,7 @@ class FastAPIWebsocketInputTransport(BaseInputTransport):
continue
if isinstance(frame, AudioRawFrame):
self.push_audio_frame(frame)
await self.push_audio_frame(frame)
await self._callbacks.on_client_disconnected(self._websocket)
@@ -89,7 +89,7 @@ class FastAPIWebsocketOutputTransport(BaseOutputTransport):
self._params = params
self._audio_buffer = bytes()
def write_raw_audio_frames(self, frames: bytes):
async def write_raw_audio_frames(self, frames: bytes):
self._audio_buffer += frames
while len(self._audio_buffer) >= self._params.audio_frame_size:
frame = AudioRawFrame(
@@ -115,9 +115,7 @@ class FastAPIWebsocketOutputTransport(BaseOutputTransport):
payload = self._params.serializer.serialize(frame)
if payload:
future = asyncio.run_coroutine_threadsafe(
self._websocket.send_text(payload), self.get_event_loop())
future.result()
await self._websocket.send_text(payload)
self._audio_buffer = self._audio_buffer[self._params.audio_frame_size:]

8
src/pipecat/transports/network/websocket_server.py
View File

@@ -93,7 +93,7 @@ class WebsocketServerInputTransport(BaseInputTransport):
continue
if isinstance(frame, AudioRawFrame):
self.push_audio_frame(frame)
await self.push_audio_frame(frame)
else:
await self._internal_push_frame(frame)
@@ -123,7 +123,7 @@ class WebsocketServerOutputTransport(BaseOutputTransport):
logger.warning("Only one client allowed, using new connection")
self._websocket = websocket
def write_raw_audio_frames(self, frames: bytes):
async def write_raw_audio_frames(self, frames: bytes):
self._audio_buffer += frames
while len(self._audio_buffer) >= self._params.audio_frame_size:
frame = AudioRawFrame(
@@ -149,9 +149,7 @@ class WebsocketServerOutputTransport(BaseOutputTransport):
proto = self._params.serializer.serialize(frame)
future = asyncio.run_coroutine_threadsafe(
self._websocket.send(proto), self.get_event_loop())
future.result()
await self._websocket.send(proto)
self._audio_buffer = self._audio_buffer[self._params.audio_frame_size:]

350
src/pipecat/transports/services/daily.py
View File

@@ -6,11 +6,10 @@
import aiohttp
import asyncio
import queue
import time
from dataclasses import dataclass
from typing import Any, Callable, Mapping
from typing import Any, Awaitable, Callable, Mapping
from concurrent.futures import ThreadPoolExecutor
from daily import (
@@ -108,20 +107,27 @@ class DailyParams(TransportParams):
class DailyCallbacks(BaseModel):
on_joined: Callable[[Mapping[str, Any]], None]
on_left: Callable[[], None]
on_error: Callable[[str], None]
on_app_message: Callable[[Any, str], None]
on_call_state_updated: Callable[[str], None]
on_dialin_ready: Callable[[str], None]
on_dialout_answered: Callable[[Any], None]
on_dialout_connected: Callable[[Any], None]
on_dialout_stopped: Callable[[Any], None]
on_dialout_error: Callable[[Any], None]
on_dialout_warning: Callable[[Any], None]
on_first_participant_joined: Callable[[Mapping[str, Any]], None]
on_participant_joined: Callable[[Mapping[str, Any]], None]
on_participant_left: Callable[[Mapping[str, Any], str], None]
on_joined: Callable[[Mapping[str, Any]], Awaitable[None]]
on_left: Callable[[], Awaitable[None]]
on_error: Callable[[str], Awaitable[None]]
on_app_message: Callable[[Any, str], Awaitable[None]]
on_call_state_updated: Callable[[str], Awaitable[None]]
on_dialin_ready: Callable[[str], Awaitable[None]]
on_dialout_answered: Callable[[Any], Awaitable[None]]
on_dialout_connected: Callable[[Any], Awaitable[None]]
on_dialout_stopped: Callable[[Any], Awaitable[None]]
on_dialout_error: Callable[[Any], Awaitable[None]]
on_dialout_warning: Callable[[Any], Awaitable[None]]
on_first_participant_joined: Callable[[Mapping[str, Any]], Awaitable[None]]
on_participant_joined: Callable[[Mapping[str, Any]], Awaitable[None]]
on_participant_left: Callable[[Mapping[str, Any], str], Awaitable[None]]
def completion_callback(future):
def _callback(*args):
if not future.cancelled():
future.get_loop().call_soon_threadsafe(future.set_result, *args)
return _callback
class DailyTransportClient(EventHandler):
@@ -161,7 +167,6 @@ class DailyTransportClient(EventHandler):
self._joined = False
self._joining = False
self._leaving = False
self._sync_response = {k: queue.Queue() for k in ["join", "leave"]}
self._executor = ThreadPoolExecutor(max_workers=5)
@@ -174,10 +179,16 @@ class DailyTransportClient(EventHandler):
color_format=self._params.camera_out_color_format)
self._mic: VirtualMicrophoneDevice = Daily.create_microphone_device(
"mic", sample_rate=self._params.audio_out_sample_rate, channels=self._params.audio_out_channels)
"mic",
sample_rate=self._params.audio_out_sample_rate,
channels=self._params.audio_out_channels,
non_blocking=True)
self._speaker: VirtualSpeakerDevice = Daily.create_speaker_device(
"speaker", sample_rate=self._params.audio_in_sample_rate, channels=self._params.audio_in_channels)
"speaker",
sample_rate=self._params.audio_in_sample_rate,
channels=self._params.audio_in_channels,
non_blocking=True)
Daily.select_speaker_device("speaker")
@property
@@ -187,30 +198,39 @@ class DailyTransportClient(EventHandler):
def set_callbacks(self, callbacks: DailyCallbacks):
self._callbacks = callbacks
def send_message(self, frame: DailyTransportMessageFrame):
self._client.send_app_message(frame.message, frame.participant_id)
async def send_message(self, frame: DailyTransportMessageFrame):
future = self._loop.create_future()
self._client.send_app_message(
frame.message,
frame.participant_id,
completion=completion_callback(future))
await future
def read_next_audio_frame(self) -> AudioRawFrame | None:
async def read_next_audio_frame(self) -> AudioRawFrame | None:
sample_rate = self._params.audio_in_sample_rate
num_channels = self._params.audio_in_channels
if self._other_participant_has_joined:
num_frames = int(sample_rate / 100) * 2 # 20ms of audio
audio = self._speaker.read_frames(num_frames)
future = self._loop.create_future()
self._speaker.read_frames(num_frames, completion=completion_callback(future))
audio = await future
return AudioRawFrame(audio=audio, sample_rate=sample_rate, num_channels=num_channels)
else:
# If no one has ever joined the meeting `read_frames()` would block,
# instead we just wait a bit. daily-python should probably return
# silence instead.
time.sleep(0.01)
await asyncio.sleep(0.01)
return None
def write_raw_audio_frames(self, frames: bytes):
self._mic.write_frames(frames)
async def write_raw_audio_frames(self, frames: bytes):
future = self._loop.create_future()
self._mic.write_frames(frames, completion=completion_callback(future))
await future
def write_frame_to_camera(self, frame: ImageRawFrame):
async def write_frame_to_camera(self, frame: ImageRawFrame):
self._camera.write_frame(frame.image)
async def join(self):
@@ -218,13 +238,10 @@ class DailyTransportClient(EventHandler):
if self._joined or self._joining:
return
self._joining = True
await self._loop.run_in_executor(self._executor, self._join)
def _join(self):
logger.info(f"Joining {self._room_url}")
self._joining = True
# For performance reasons, never subscribe to video streams (unless a
# video renderer is registered).
self._client.update_subscription_profiles({
@@ -236,10 +253,42 @@ class DailyTransportClient(EventHandler):
self._client.set_user_name(self._bot_name)
try:
(data, error) = await self._join()
if not error:
self._joined = True
self._joining = False
logger.info(f"Joined {self._room_url}")
if self._token and self._params.transcription_enabled:
logger.info(
f"Enabling transcription with settings {self._params.transcription_settings}")
self._client.start_transcription(
self._params.transcription_settings.model_dump())
await self._callbacks.on_joined(data["participants"]["local"])
else:
error_msg = f"Error joining {self._room_url}: {error}"
logger.error(error_msg)
await self._callbacks.on_error(error_msg)
except asyncio.TimeoutError:
error_msg = f"Time out joining {self._room_url}"
logger.error(error_msg)
await self._callbacks.on_error(error_msg)
async def _join(self):
future = self._loop.create_future()
def handle_join_response(data, error):
if not future.cancelled():
future.get_loop().call_soon_threadsafe(future.set_result, (data, error))
self._client.join(
self._room_url,
self._token,
completion=self._call_joined,
completion=handle_join_response,
client_settings={
"inputs": {
"camera": {
@@ -275,33 +324,7 @@ class DailyTransportClient(EventHandler):
},
})
self._handle_join_response()
def _handle_join_response(self):
try:
(data, error) = self._sync_response["join"].get(timeout=10)
if not error:
self._joined = True
self._joining = False
logger.info(f"Joined {self._room_url}")
if self._token and self._params.transcription_enabled:
logger.info(
f"Enabling transcription with settings {self._params.transcription_settings}")
self._client.start_transcription(
self._params.transcription_settings.model_dump())
self._callbacks.on_joined(data["participants"]["local"])
else:
error_msg = f"Error joining {self._room_url}: {error}"
logger.error(error_msg)
self._callbacks.on_error(error_msg)
self._sync_response["join"].task_done()
except queue.Empty:
error_msg = f"Time out joining {self._room_url}"
logger.error(error_msg)
self._callbacks.on_error(error_msg)
return await asyncio.wait_for(future, timeout=10)
async def leave(self):
# Transport not joined, ignore.
@@ -311,34 +334,36 @@ class DailyTransportClient(EventHandler):
self._joined = False
self._leaving = True
await self._loop.run_in_executor(self._executor, self._leave)
def _leave(self):
logger.info(f"Leaving {self._room_url}")
if self._params.transcription_enabled:
self._client.stop_transcription()
self._client.leave(completion=self._call_left)
self._handle_leave_response()
def _handle_leave_response(self):
try:
error = self._sync_response["leave"].get(timeout=10)
error = await self._leave()
if not error:
self._leaving = False
logger.info(f"Left {self._room_url}")
self._callbacks.on_left()
await self._callbacks.on_left()
else:
error_msg = f"Error leaving {self._room_url}: {error}"
logger.error(error_msg)
self._callbacks.on_error(error_msg)
self._sync_response["leave"].task_done()
except queue.Empty:
await self._callbacks.on_error(error_msg)
except asyncio.TimeoutError:
error_msg = f"Time out leaving {self._room_url}"
logger.error(error_msg)
self._callbacks.on_error(error_msg)
await self._callbacks.on_error(error_msg)
async def _leave(self):
future = self._loop.create_future()
def handle_leave_response(error):
if not future.cancelled():
future.get_loop().call_soon_threadsafe(future.set_result, error)
self._client.leave(completion=handle_leave_response)
return await asyncio.wait_for(future, timeout=10)
async def cleanup(self):
await self._loop.run_in_executor(self._executor, self._cleanup)
@@ -400,28 +425,28 @@ class DailyTransportClient(EventHandler):
#
def on_app_message(self, message: Any, sender: str):
self._callbacks.on_app_message(message, sender)
self._call_async_callback(self._callbacks.on_app_message, message, sender)
def on_call_state_updated(self, state: str):
self._callbacks.on_call_state_updated(state)
self._call_async_callback(self._callbacks.on_call_state_updated, state)
def on_dialin_ready(self, sip_endpoint: str):
self._callbacks.on_dialin_ready(sip_endpoint)
self._call_async_callback(self._callbacks.on_dialin_ready, sip_endpoint)
def on_dialout_answered(self, data: Any):
self._callbacks.on_dialout_answered(data)
self._call_async_callback(self._callbacks.on_dialout_answered, data)
def on_dialout_connected(self, data: Any):
self._callbacks.on_dialout_connected(data)
self._call_async_callback(self._callbacks.on_dialout_connected, data)
def on_dialout_stopped(self, data: Any):
self._callbacks.on_dialout_stopped(data)
self._call_async_callback(self._callbacks.on_dialout_stopped, data)
def on_dialout_error(self, data: Any):
self._callbacks.on_dialout_error(data)
self._call_async_callback(self._callbacks.on_dialout_error, data)
def on_dialout_warning(self, data: Any):
self._callbacks.on_dialout_warning(data)
self._call_async_callback(self._callbacks.on_dialout_warning, data)
def on_participant_joined(self, participant):
id = participant["id"]
@@ -429,15 +454,15 @@ class DailyTransportClient(EventHandler):
if not self._other_participant_has_joined:
self._other_participant_has_joined = True
self._callbacks.on_first_participant_joined(participant)
self._call_async_callback(self._callbacks.on_first_participant_joined, participant)
self._callbacks.on_participant_joined(participant)
self._call_async_callback(self._callbacks.on_participant_joined, participant)
def on_participant_left(self, participant, reason):
id = participant["id"]
logger.info(f"Participant left {id}")
self._callbacks.on_participant_left(participant, reason)
self._call_async_callback(self._callbacks.on_participant_left, participant, reason)
def on_transcription_message(self, message: Mapping[str, Any]):
participant_id = ""
@@ -446,7 +471,7 @@ class DailyTransportClient(EventHandler):
if participant_id in self._transcription_renderers:
callback = self._transcription_renderers[participant_id]
callback(participant_id, message)
self._call_async_callback(callback, participant_id, message)
def on_transcription_error(self, message):
logger.error(f"Transcription error: {message}")
@@ -461,18 +486,19 @@ class DailyTransportClient(EventHandler):
# Daily (CallClient callbacks)
#
def _call_joined(self, data, error):
self._sync_response["join"].put((data, error))
def _call_left(self, error):
self._sync_response["leave"].put(error)
def _video_frame_received(self, participant_id, video_frame):
callback = self._video_renderers[participant_id]
callback(participant_id,
video_frame.buffer,
(video_frame.width, video_frame.height),
video_frame.color_format)
self._call_async_callback(
callback,
participant_id,
video_frame.buffer,
(video_frame.width,
video_frame.height),
video_frame.color_format)
def _call_async_callback(self, callback, *args):
future = asyncio.run_coroutine_threadsafe(callback(*args), self._loop)
future.result()
class DailyInputTransport(BaseInputTransport):
@@ -491,8 +517,6 @@ class DailyInputTransport(BaseInputTransport):
num_channels=self._params.audio_in_channels)
async def start(self, frame: StartFrame):
if self._running:
return
# Parent start.
await super().start(frame)
# Join the room.
@@ -500,19 +524,17 @@ class DailyInputTransport(BaseInputTransport):
# Create audio task. It reads audio frames from Daily and push them
# internally for VAD processing.
if self._params.audio_in_enabled or self._params.vad_enabled:
self._audio_in_thread = self._loop.run_in_executor(
self._executor, self._audio_in_thread_handler)
self._audio_in_task = self.get_event_loop().create_task(self._audio_in_task_handler())
async def stop(self):
if not self._running:
return
# Parent stop. This will set _running to False.
# Parent stop.
await super().stop()
# Leave the room.
await self._client.leave()
# Stop audio thread.
if self._params.audio_in_enabled or self._params.vad_enabled:
await self._audio_in_thread
self._audio_in_task.cancel()
await self._audio_in_task
async def cleanup(self):
await super().cleanup()
@@ -535,26 +557,25 @@ class DailyInputTransport(BaseInputTransport):
# Frames
#
def push_transcription_frame(self, frame: TranscriptionFrame | InterimTranscriptionFrame):
future = asyncio.run_coroutine_threadsafe(
self._internal_push_frame(frame), self.get_event_loop())
future.result()
async def push_transcription_frame(self, frame: TranscriptionFrame | InterimTranscriptionFrame):
await self._internal_push_frame(frame)
def push_app_message(self, message: Any, sender: str):
async def push_app_message(self, message: Any, sender: str):
frame = DailyTransportMessageFrame(message=message, participant_id=sender)
future = asyncio.run_coroutine_threadsafe(
self._internal_push_frame(frame), self.get_event_loop())
future.result()
await self._internal_push_frame(frame)
#
# Audio in
#
def _audio_in_thread_handler(self):
while self._running:
frame = self._client.read_next_audio_frame()
if frame:
self.push_audio_frame(frame)
async def _audio_in_task_handler(self):
while True:
try:
frame = await self._client.read_next_audio_frame()
if frame:
await self.push_audio_frame(frame)
except asyncio.CancelledError:
break
#
# Camera in
@@ -584,7 +605,7 @@ class DailyInputTransport(BaseInputTransport):
if participant_id in self._video_renderers:
self._video_renderers[participant_id]["render_next_frame"] = True
def _on_participant_video_frame(self, participant_id: str, buffer, size, format):
async def _on_participant_video_frame(self, participant_id: str, buffer, size, format):
render_frame = False
curr_time = time.time()
@@ -604,9 +625,7 @@ class DailyInputTransport(BaseInputTransport):
image=buffer,
size=size,
format=format)
future = asyncio.run_coroutine_threadsafe(
self._internal_push_frame(frame), self.get_event_loop())
future.result()
await self._internal_push_frame(frame)
self._video_renderers[participant_id]["timestamp"] = curr_time
@@ -619,17 +638,13 @@ class DailyOutputTransport(BaseOutputTransport):
self._client = client
async def start(self, frame: StartFrame):
if self._running:
return
# Parent start.
await super().start(frame)
# Join the room.
await self._client.join()
async def stop(self):
if not self._running:
return
# Parent stop. This will set _running to False.
# Parent stop.
await super().stop()
# Leave the room.
await self._client.leave()
@@ -638,10 +653,10 @@ class DailyOutputTransport(BaseOutputTransport):
await super().cleanup()
await self._client.cleanup()
def send_message(self, frame: DailyTransportMessageFrame):
self._client.send_message(frame)
async def send_message(self, frame: DailyTransportMessageFrame):
await self._client.send_message(frame)
def send_metrics(self, frame: MetricsFrame):
async def send_metrics(self, frame: MetricsFrame):
ttfb = [{"name": n, "time": t} for n, t in frame.ttfb.items()]
message = DailyTransportMessageFrame(message={
"type": "pipecat-metrics",
@@ -649,13 +664,13 @@ class DailyOutputTransport(BaseOutputTransport):
"ttfb": ttfb
},
})
self._client.send_message(message)
await self._client.send_message(message)
def write_raw_audio_frames(self, frames: bytes):
self._client.write_raw_audio_frames(frames)
async def write_raw_audio_frames(self, frames: bytes):
await self._client.write_raw_audio_frames(frames)
def write_frame_to_camera(self, frame: ImageRawFrame):
self._client.write_frame_to_camera(frame)
async def write_frame_to_camera(self, frame: ImageRawFrame):
await self._client.write_frame_to_camera(frame)
class DailyTransport(BaseTransport):
@@ -774,24 +789,24 @@ class DailyTransport(BaseTransport):
self._input.capture_participant_video(
participant_id, framerate, video_source, color_format)
def _on_joined(self, participant):
self._call_async_event_handler("on_joined", participant)
async def _on_joined(self, participant):
await self._call_event_handler("on_joined", participant)
def _on_left(self):
self._call_async_event_handler("on_left")
async def _on_left(self):
await self._call_event_handler("on_left")
def _on_error(self, error):
async def _on_error(self, error):
# TODO(aleix): Report error to input/output transports. The one managing
# the client should report the error.
pass
def _on_app_message(self, message: Any, sender: str):
async def _on_app_message(self, message: Any, sender: str):
if self._input:
self._input.push_app_message(message, sender)
self._call_async_event_handler("on_app_message", message, sender)
await self._input.push_app_message(message, sender)
await self._call_event_handler("on_app_message", message, sender)
def _on_call_state_updated(self, state: str):
self._call_async_event_handler("on_call_state_updated", state)
async def _on_call_state_updated(self, state: str):
await self._call_event_handler("on_call_state_updated", state)
async def _handle_dialin_ready(self, sip_endpoint: str):
if not self._params.dialin_settings:
@@ -824,36 +839,36 @@ class DailyTransport(BaseTransport):
except BaseException as e:
logger.error(f"Error handling dialin-ready event ({url}): {e}")
def _on_dialin_ready(self, sip_endpoint):
async def _on_dialin_ready(self, sip_endpoint):
if self._params.dialin_settings:
asyncio.run_coroutine_threadsafe(self._handle_dialin_ready(sip_endpoint), self._loop)
self._call_async_event_handler("on_dialin_ready", sip_endpoint)
await self._handle_dialin_ready(sip_endpoint)
await self._call_event_handler("on_dialin_ready", sip_endpoint)
def _on_dialout_answered(self, data):
self._call_async_event_handler("on_dialout_answered", data)
async def _on_dialout_answered(self, data):
await self._call_event_handler("on_dialout_answered", data)
def _on_dialout_connected(self, data):
self._call_async_event_handler("on_dialout_connected", data)
async def _on_dialout_connected(self, data):
await self._call_event_handler("on_dialout_connected", data)
def _on_dialout_stopped(self, data):
self._call_async_event_handler("on_dialout_stopped", data)
async def _on_dialout_stopped(self, data):
await self._call_event_handler("on_dialout_stopped", data)
def _on_dialout_error(self, data):
self._call_async_event_handler("on_dialout_error", data)
async def _on_dialout_error(self, data):
await self._call_event_handler("on_dialout_error", data)
def _on_dialout_warning(self, data):
self._call_async_event_handler("on_dialout_warning", data)
async def _on_dialout_warning(self, data):
await self._call_event_handler("on_dialout_warning", data)
def _on_participant_joined(self, participant):
self._call_async_event_handler("on_participant_joined", participant)
async def _on_participant_joined(self, participant):
await self._call_event_handler("on_participant_joined", participant)
def _on_participant_left(self, participant, reason):
self._call_async_event_handler("on_participant_left", participant, reason)
async def _on_participant_left(self, participant, reason):
await self._call_event_handler("on_participant_left", participant, reason)
def _on_first_participant_joined(self, participant):
self._call_async_event_handler("on_first_participant_joined", participant)
async def _on_first_participant_joined(self, participant):
await self._call_event_handler("on_first_participant_joined", participant)
def _on_transcription_message(self, participant_id, message):
async def _on_transcription_message(self, participant_id, message):
text = message["text"]
timestamp = message["timestamp"]
is_final = message["rawResponse"]["is_final"]
@@ -864,9 +879,4 @@ class DailyTransport(BaseTransport):
frame = InterimTranscriptionFrame(text, participant_id, timestamp)
if self._input:
self._input.push_transcription_frame(frame)
def _call_async_event_handler(self, event_name: str, *args, **kwargs):
future = asyncio.run_coroutine_threadsafe(
self._call_event_handler(event_name, *args, **kwargs), self._loop)
future.result()
await self._input.push_transcription_frame(frame)