From 7a6136fa9d3e3c6d1dc86200103e604c2f1ada8f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Thu, 30 May 2024 16:28:17 -0700 Subject: [PATCH] make all transports read/write functions async --- examples/websocket-server/server.py | 7 +- src/pipecat/pipeline/pipeline.py | 5 +- src/pipecat/pipeline/runner.py | 2 +- src/pipecat/processors/frame_processor.py | 8 +- src/pipecat/transports/base_input.py | 94 ++++--------- src/pipecat/transports/base_output.py | 130 +++++++----------- src/pipecat/transports/base_transport.py | 4 +- src/pipecat/transports/local/tk.py | 22 ++- .../transports/network/websocket_server.py | 100 +++++--------- src/pipecat/transports/services/daily.py | 95 +++++++------ src/pipecat/vad/silero.py | 4 +- src/pipecat/vad/vad_analyzer.py | 6 +- 12 files changed, 195 insertions(+), 282 deletions(-) diff --git a/examples/websocket-server/server.py b/examples/websocket-server/server.py index fa9fdc057..cf6ff0289 100644 --- a/examples/websocket-server/server.py +++ b/examples/websocket-server/server.py @@ -31,7 +31,12 @@ logger.add(sys.stderr, level="DEBUG") async def main(): async with aiohttp.ClientSession() as session: - transport = WebsocketServerTransport(params=WebsocketServerParams(add_wav_header=True)) + transport = WebsocketServerTransport( + params=WebsocketServerParams( + audio_out_enabled=True, + add_wav_header=True + ) + ) vad = SileroVAD(audio_passthrough=True) diff --git a/src/pipecat/pipeline/pipeline.py b/src/pipecat/pipeline/pipeline.py index 5d1f9d3cf..87e917845 100644 --- a/src/pipecat/pipeline/pipeline.py +++ b/src/pipecat/pipeline/pipeline.py @@ -4,8 +4,6 @@ # SPDX-License-Identifier: BSD 2-Clause License # -import asyncio - from typing import Callable, Coroutine, List from pipecat.frames.frames import Frame @@ -67,7 +65,8 @@ class Pipeline(FrameProcessor): await self._sink.process_frame(frame, FrameDirection.UPSTREAM) async def _cleanup_processors(self): - await asyncio.gather(*[p.cleanup() for p in self._processors]) + for p in self._processors: + await p.cleanup() def _link_processors(self): prev = self._processors[0] diff --git a/src/pipecat/pipeline/runner.py b/src/pipecat/pipeline/runner.py index fe8cd8b6e..2d1b8100a 100644 --- a/src/pipecat/pipeline/runner.py +++ b/src/pipecat/pipeline/runner.py @@ -46,7 +46,7 @@ class PipelineRunner: return self._running def _setup_sigint(self): - loop = asyncio.get_event_loop() + loop = asyncio.get_running_loop() loop.add_signal_handler( signal.SIGINT, lambda *args: asyncio.create_task(self._sigint_handler()) diff --git a/src/pipecat/processors/frame_processor.py b/src/pipecat/processors/frame_processor.py index 8b6a66e21..793aa8eb1 100644 --- a/src/pipecat/processors/frame_processor.py +++ b/src/pipecat/processors/frame_processor.py @@ -5,7 +5,7 @@ # import asyncio -from asyncio import AbstractEventLoop + from enum import Enum from pipecat.frames.frames import ErrorFrame, Frame @@ -21,12 +21,12 @@ class FrameDirection(Enum): class FrameProcessor: - def __init__(self): + def __init__(self, loop: asyncio.AbstractEventLoop | None = None): self.id: int = obj_id() self.name = f"{self.__class__.__name__}#{obj_count(self)}" self._prev: "FrameProcessor" | None = None self._next: "FrameProcessor" | None = None - self._loop: AbstractEventLoop = asyncio.get_event_loop() + self._loop: asyncio.AbstractEventLoop = loop or asyncio.get_running_loop() async def cleanup(self): pass @@ -36,7 +36,7 @@ class FrameProcessor: processor._prev = self logger.debug(f"Linking {self} -> {self._next}") - def get_event_loop(self) -> AbstractEventLoop: + def get_event_loop(self) -> asyncio.AbstractEventLoop: return self._loop async def process_frame(self, frame: Frame, direction: FrameDirection): diff --git a/src/pipecat/transports/base_input.py b/src/pipecat/transports/base_input.py index 71733c59c..b1c015d9e 100644 --- a/src/pipecat/transports/base_input.py +++ b/src/pipecat/transports/base_input.py @@ -5,9 +5,6 @@ # import asyncio -import queue - -from concurrent.futures import ThreadPoolExecutor from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.frames.frames import ( @@ -33,14 +30,11 @@ class BaseInputTransport(FrameProcessor): self._params = params - self._running = False self._allow_interruptions = False - self._in_executor = ThreadPoolExecutor(max_workers=5) - # Create audio input queue if needed. if self._params.audio_in_enabled or self._params.vad_enabled: - self._audio_in_queue = queue.Queue() + self._audio_in_queue = asyncio.Queue() # 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. @@ -52,45 +46,27 @@ class BaseInputTransport(FrameProcessor): # for example. self._allow_interruptions = frame.allow_interruptions - if self._running: - return - - self._running = True - if self._params.audio_in_enabled or self._params.vad_enabled: loop = self.get_event_loop() - self._audio_in_thread = loop.run_in_executor( - self._in_executor, self._audio_in_thread_handler) - self._audio_out_thread = loop.run_in_executor( - self._in_executor, self._audio_out_thread_handler) + self._audio_task = 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 tasks to finish. if self._params.audio_in_enabled or self._params.vad_enabled: - await self._audio_in_thread - await self._audio_out_thread + self._audio_task.cancel() self._push_frame_task.cancel() - def vad_analyze(self, audio_frames: bytes) -> VADState: + async def vad_analyze(self, audio_frames: bytes) -> VADState: pass - def read_raw_audio_frames(self, frame_count: int) -> bytes: + async def read_raw_audio_frames(self, frame_count: int) -> bytes: pass # # Frame processor # - async def cleanup(self): - pass - async def process_frame(self, frame: Frame, direction: FrameDirection): if isinstance(frame, CancelFrame): await self.stop() @@ -150,8 +126,8 @@ class BaseInputTransport(FrameProcessor): # Audio input # - 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: @@ -160,51 +136,39 @@ 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_in_thread_handler(self): + async def _audio_task_handler(self): + vad_state: VADState = VADState.QUIET + sample_rate = self._params.audio_in_sample_rate num_channels = self._params.audio_in_channels num_frames = int(sample_rate / 100) # 10ms of audio - while self._running: + + while True: try: - audio_frames = self.read_raw_audio_frames(num_frames) + audio_frames = await self.read_raw_audio_frames(num_frames) if len(audio_frames) > 0: frame = AudioRawFrame( audio=audio_frames, sample_rate=sample_rate, num_channels=num_channels) - self._audio_in_queue.put(frame) + + 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 = await self._handle_vad(frame.audio, vad_state) + audio_passthrough = self._params.vad_audio_passthrough + + # Push audio downstream if passthrough. + if audio_passthrough: + await self._internal_push_frame(frame) + except asyncio.CancelledError: + break except BaseException as e: logger.error(f"Error reading audio frames: {e}") - - def _audio_out_thread_handler(self): - vad_state: VADState = VADState.QUIET - while self._running: - try: - frame = self._audio_in_queue.get(timeout=1) - - 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) - 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.get_event_loop()) - future.result() - - self._audio_in_queue.task_done() - except queue.Empty: - pass - except BaseException as e: - logger.error(f"Error pushing audio frames: {e}") diff --git a/src/pipecat/transports/base_output.py b/src/pipecat/transports/base_output.py index 146c58496..1d9ba8626 100644 --- a/src/pipecat/transports/base_output.py +++ b/src/pipecat/transports/base_output.py @@ -1,17 +1,11 @@ # # Copyright (c) 2024, Daily - # # SPDX-License-Identifier: BSD 2-Clause License # import asyncio import itertools -import queue -import time -import threading - -from concurrent.futures import ThreadPoolExecutor from PIL import Image from typing import List @@ -40,22 +34,19 @@ class BaseOutputTransport(FrameProcessor): self._params = params - self._running = False self._allow_interruptions = False - self._out_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. + # Create media task queues. if self._params.camera_out_enabled: - self._camera_out_queue = queue.Queue() - self._sink_queue = queue.Queue() + self._camera_out_queue = asyncio.Queue() + self._sink_queue = asyncio.Queue() self._stopped_event = asyncio.Event() - self._is_interrupted = threading.Event() + self._is_interrupted = asyncio.Event() async def start(self, frame: StartFrame): # Make sure we have the latest params. Note that this transport might @@ -63,52 +54,40 @@ class BaseOutputTransport(FrameProcessor): # for example. self._allow_interruptions = frame.allow_interruptions - if self._running: - return - - self._running = True - loop = self.get_event_loop() if self._params.camera_out_enabled: - self._camera_out_thread = loop.run_in_executor( - self._out_executor, self._camera_out_thread_handler) + self._camera_out_task = loop.create_task(self._camera_out_task_handler()) - self._sink_thread = loop.run_in_executor(self._out_executor, self._sink_thread_handler) + self._sink_task = loop.create_task(self._sink_task_handler()) # 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 stop(self): - if not self._running: - return - - # This will exit all threads. - self._running = False - self._stopped_event.set() - def send_message(self, frame: TransportMessageFrame): + async def cleanup(self): + if self._params.camera_out_enabled: + self._camera_out_task.cancel() + + self._sink_task.cancel() + self._push_frame_task.cancel() + + async def send_message(self, frame: TransportMessageFrame): 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 # # Frame processor # - async def cleanup(self): - # Wait on the threads to finish. - if self._params.camera_out_enabled: - await self._camera_out_thread - - await self._sink_thread - async def process_frame(self, frame: Frame, direction: FrameDirection): # # Out-of-band frames like (CancelFrame or StartInterruptionFrame) are @@ -117,7 +96,7 @@ class BaseOutputTransport(FrameProcessor): # if isinstance(frame, StartFrame): await self.start(frame) - self._sink_queue.put(frame) + await self._sink_queue.put(frame) # EndFrame is managed in the queue handler. elif isinstance(frame, CancelFrame): await self.stop() @@ -126,11 +105,11 @@ class BaseOutputTransport(FrameProcessor): await self._handle_interruptions(frame) await self.push_frame(frame, direction) else: - self._sink_queue.put(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 - # know when the internal threads will finish. + # know when the internal tasks will finish. if isinstance(frame, CancelFrame) or isinstance(frame, EndFrame): await self._stopped_event.wait() @@ -145,7 +124,7 @@ class BaseOutputTransport(FrameProcessor): elif isinstance(frame, StopInterruptionFrame): self._is_interrupted.clear() - def _sink_thread_handler(self): + async def _sink_task_handler(self): # 10ms bytes bytes_size_10ms = int(self._params.audio_out_sample_rate / 100) * \ self._params.audio_out_channels * 2 @@ -155,37 +134,34 @@ class BaseOutputTransport(FrameProcessor): # Audio accumlation buffer buffer = bytearray() - while self._running: + while True: try: - frame = self._sink_queue.get(timeout=1) + frame = await self._sink_queue.get() if not self._is_interrupted.is_set(): if isinstance(frame, AudioRawFrame): if self._params.audio_out_enabled: buffer.extend(frame.audio) - buffer = self._send_audio_truncated(buffer, smallest_write_size) + buffer = await self._send_audio_truncated(buffer, smallest_write_size) elif isinstance(frame, ImageRawFrame) and self._params.camera_out_enabled: - self._set_camera_image(frame) + await self._set_camera_image(frame) elif isinstance(frame, SpriteFrame) and self._params.camera_out_enabled: - self._set_camera_images(frame.images) + await self._set_camera_images(frame.images) elif isinstance(frame, TransportMessageFrame): - self.send_message(frame) + await self.send_message(frame) else: - future = asyncio.run_coroutine_threadsafe( - self._internal_push_frame(frame), self.get_event_loop()) - future.result() + await self._internal_push_frame(frame) else: # If we get interrupted just clear the output buffer. buffer = bytearray() if isinstance(frame, EndFrame): # Send all remaining audio before stopping (multiple of 10ms of audio). - self._send_audio_truncated(buffer, bytes_size_10ms) - future = asyncio.run_coroutine_threadsafe(self.stop(), self.get_event_loop()) - future.result() + await self._send_audio_truncated(buffer, bytes_size_10ms) + await self.stop() self._sink_queue.task_done() - except queue.Empty: - pass + except asyncio.CancelledError: + break except BaseException as e: logger.error(f"Error processing sink queue: {e}") @@ -219,7 +195,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: @@ -229,32 +205,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(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"Error writing to camera: {e}") @@ -265,13 +241,9 @@ class BaseOutputTransport(FrameProcessor): async def send_audio(self, frame: AudioRawFrame): await self.process_frame(frame, FrameDirection.DOWNSTREAM) - def _send_audio_truncated(self, buffer: bytearray, smallest_write_size: int) -> bytearray: - try: - truncated_length: int = len(buffer) - (len(buffer) % smallest_write_size) - if truncated_length: - self.write_raw_audio_frames(bytes(buffer[:truncated_length])) - buffer = buffer[truncated_length:] - return buffer - except BaseException as e: - logger.error(f"Error writing audio frames: {e}") - return buffer + async def _send_audio_truncated(self, buffer: bytearray, smallest_write_size: int) -> bytearray: + truncated_length: int = len(buffer) - (len(buffer) % smallest_write_size) + if truncated_length: + await self.write_raw_audio_frames(bytes(buffer[:truncated_length])) + buffer = buffer[truncated_length:] + return buffer diff --git a/src/pipecat/transports/base_transport.py b/src/pipecat/transports/base_transport.py index 921e08c83..7034b81eb 100644 --- a/src/pipecat/transports/base_transport.py +++ b/src/pipecat/transports/base_transport.py @@ -41,8 +41,8 @@ class TransportParams(BaseModel): class BaseTransport(ABC): - def __init__(self, loop: asyncio.AbstractEventLoop): - self._loop = loop + def __init__(self, loop: asyncio.AbstractEventLoop | None): + self._loop = loop or asyncio.get_running_loop() self._event_handlers: dict = {} @abstractmethod diff --git a/src/pipecat/transports/local/tk.py b/src/pipecat/transports/local/tk.py index 782c01dae..5cc13604c 100644 --- a/src/pipecat/transports/local/tk.py +++ b/src/pipecat/transports/local/tk.py @@ -6,6 +6,8 @@ import asyncio +from concurrent.futures import ThreadPoolExecutor + import numpy as np import tkinter as tk @@ -38,6 +40,8 @@ class TkInputTransport(BaseInputTransport): def __init__(self, py_audio: pyaudio.PyAudio, params: TransportParams): super().__init__(params) + self._executor = ThreadPoolExecutor(max_workers=5) + self._in_stream = py_audio.open( format=py_audio.get_format_from_width(2), channels=params.audio_in_channels, @@ -45,8 +49,11 @@ class TkInputTransport(BaseInputTransport): frames_per_buffer=params.audio_in_sample_rate, input=True) - def read_raw_audio_frames(self, frame_count: int) -> bytes: - return self._in_stream.read(frame_count, exception_on_overflow=False) + async def read_raw_audio_frames(self, frame_count: int) -> bytes: + def read_audio(frame_count: int) -> bytes: + return self._in_stream.read(frame_count, exception_on_overflow=False) + + return await self.get_event_loop().run_in_executor(self._executor, read_audio, frame_count) async def start(self, frame: StartFrame): await super().start(frame) @@ -70,6 +77,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, @@ -83,10 +92,13 @@ 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) + async def write_raw_audio_frames(self, frames: bytes): + def write_audio(frames: bytes): + self._out_stream.write(frames) - def write_frame_to_camera(self, frame: ImageRawFrame): + await self.get_event_loop().run_in_executor(self._executor, write_audio, frames) + + async 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): diff --git a/src/pipecat/transports/network/websocket_server.py b/src/pipecat/transports/network/websocket_server.py index f765af615..3f22ccde5 100644 --- a/src/pipecat/transports/network/websocket_server.py +++ b/src/pipecat/transports/network/websocket_server.py @@ -4,7 +4,6 @@ # SPDX-License-Identifier: BSD 2-Clause License # - import asyncio import io import wave @@ -18,12 +17,11 @@ from pipecat.frames.frames import ( CancelFrame, EndFrame, Frame, - StartFrame, - TTSStartedFrame, - TTSStoppedFrame) + StartFrame) from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.serializers.base_serializer import FrameSerializer from pipecat.serializers.protobuf import ProtobufFrameSerializer +from pipecat.transports.base_output import BaseOutputTransport from pipecat.transports.base_transport import BaseTransport, TransportParams from loguru import logger @@ -134,24 +132,15 @@ class WebsocketServerInputTransport(FrameProcessor): break -class WebsocketServerOutputTransport(FrameProcessor): +class WebsocketServerOutputTransport(BaseOutputTransport): def __init__(self, params: WebsocketServerParams): - super().__init__() + super().__init__(params) self._params = params - self._websocket = None - self._audio_buffer = bytes() - self._websocket: websockets.WebSocketServerProtocol | None = None - - loop = self.get_event_loop() - self._send_queue_task = loop.create_task(self._send_queue_task_handler()) - self._send_queue = asyncio.Queue() - self._audio_buffer = bytes() - self._in_tts_audio = False async def set_client_connection(self, websocket: websockets.WebSocketServerProtocol): if self._websocket: @@ -159,61 +148,34 @@ class WebsocketServerOutputTransport(FrameProcessor): logger.warning("Only one client allowed, using new connection") self._websocket = websocket - async def _send_queue_task_handler(self): - running = True - while running: - frame = await self._send_queue.get() - if self._websocket and frame: - # We send WAV data so we can easily decoded in the browser. - if self._params.add_wav_header: - content = io.BytesIO() - ww = wave.open(content, "wb") - ww.setsampwidth(2) - ww.setnchannels(frame.num_channels) - ww.setframerate(frame.sample_rate) - ww.writeframes(frame.audio) - ww.close() - content.seek(0) - wav_frame = AudioRawFrame( - content.read(), - sample_rate=frame.sample_rate, - num_channels=frame.num_channels) - frame = wav_frame - proto = self._params.serializer.serialize(frame) - await self._websocket.send(proto) + 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( + audio=self._audio_buffer[:self._params.audio_frame_size], + sample_rate=self._params.audio_out_sample_rate, + num_channels=self._params.audio_out_channels + ) - async def _stop(self): - self._send_queue_task.cancel() + if self._params.add_wav_header: + content = io.BytesIO() + ww = wave.open(content, "wb") + ww.setsampwidth(2) + ww.setnchannels(frame.num_channels) + ww.setframerate(frame.sample_rate) + ww.writeframes(frame.audio) + ww.close() + content.seek(0) + wav_frame = AudioRawFrame( + content.read(), + sample_rate=frame.sample_rate, + num_channels=frame.num_channels) + frame = wav_frame - async def process_frame(self, frame: Frame, direction: FrameDirection): - if isinstance(frame, CancelFrame): - await self._stop() - await self.push_frame(frame, direction) - elif isinstance(frame, TTSStartedFrame): - self._in_tts_audio = True - elif isinstance(frame, AudioRawFrame): - if self._in_tts_audio: - self._audio_buffer += frame.audio - while len(self._audio_buffer) >= self._params.audio_frame_size: - frame = AudioRawFrame( - audio=self._audio_buffer[:self._params.audio_frame_size], - sample_rate=self._params.audio_out_sample_rate, - num_channels=self._params.audio_out_channels - ) - await self._send_queue.put(frame) - self._audio_buffer = self._audio_buffer[self._params.audio_frame_size:] - elif isinstance(frame, TTSStoppedFrame): - self._in_tts_audio = False - if self._audio_buffer: - frame = AudioRawFrame( - audio=self._audio_buffer, - sample_rate=self._params.audio_out_sample_rate, - num_channels=self._params.audio_out_channels - ) - await self._send_queue.put(frame) - self._audio_buffer = bytes() - else: - await self.push_frame(frame, direction) + proto = self._params.serializer.serialize(frame) + await self._websocket.send(proto) + + self._audio_buffer = self._audio_buffer[self._params.audio_frame_size:] class WebsocketServerTransport(BaseTransport): @@ -223,7 +185,7 @@ class WebsocketServerTransport(BaseTransport): host: str = "localhost", port: int = 8765, params: WebsocketServerParams = WebsocketServerParams(), - loop: asyncio.AbstractEventLoop = asyncio.get_event_loop()): + loop: asyncio.AbstractEventLoop | None = None): super().__init__(loop) self._host = host self._port = port diff --git a/src/pipecat/transports/services/daily.py b/src/pipecat/transports/services/daily.py index 99421f2ef..1b4022727 100644 --- a/src/pipecat/transports/services/daily.py +++ b/src/pipecat/transports/services/daily.py @@ -182,9 +182,6 @@ class DailyTransportClient(EventHandler): def participant_id(self) -> str: return self._participant_id - 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) @@ -464,8 +461,10 @@ class DailyInputTransport(BaseInputTransport): self._client = client + self._executor = ThreadPoolExecutor(max_workers=5) + self._video_renderers = {} - self._camera_in_queue = queue.Queue() + self._camera_in_queue = asyncio.Queue() self._vad_analyzer = params.vad_analyzer if params.vad_enabled and not params.vad_analyzer: @@ -474,38 +473,34 @@ class DailyInputTransport(BaseInputTransport): num_channels=self._params.audio_in_channels) async def start(self, frame: StartFrame): - if self._running: - return # Join the room. await self._client.join() - # This will set _running=True + # Create camera in task + self._camera_in_task = self.get_event_loop().create_task(self._camera_in_task_handler()) await super().start(frame) - # Create camera in thread (runs if _running is true). - self._camera_in_thread = self._loop.run_in_executor( - self._in_executor, self._camera_in_thread_handler) async def stop(self): - if not self._running: - return # Leave the room. await self._client.leave() - # This will set _running=False + # The task will stop. + self._camera_in_task.cancel() await super().stop() - # The thread will stop. - await self._camera_in_thread async def cleanup(self): - await super().cleanup() await self._client.cleanup() + await super().cleanup() - def vad_analyze(self, audio_frames: bytes) -> VADState: + async def vad_analyze(self, audio_frames: bytes) -> VADState: state = VADState.QUIET if self._vad_analyzer: - state = self._vad_analyzer.analyze_audio(audio_frames) + state = await self._vad_analyzer.analyze_audio(audio_frames) return state - def read_raw_audio_frames(self, frame_count: int) -> bytes: - return self._client.read_raw_audio_frames(frame_count) + async def read_raw_audio_frames(self, frame_count: int) -> bytes: + def read_audio(frame_count: int) -> bytes: + return self._client.read_raw_audio_frames(frame_count) + + return await self.get_event_loop().run_in_executor(self._executor, read_audio, frame_count) # # FrameProcessor @@ -580,20 +575,19 @@ class DailyInputTransport(BaseInputTransport): image=buffer, size=size, format=format) - self._camera_in_queue.put(frame) + asyncio.run_coroutine_threadsafe( + self._camera_in_queue.put(frame), self.get_event_loop()) self._video_renderers[participant_id]["timestamp"] = curr_time - def _camera_in_thread_handler(self): - while self._running: + async def _camera_in_task_handler(self): + while True: try: - frame = self._camera_in_queue.get(timeout=1) - future = asyncio.run_coroutine_threadsafe( - self._internal_push_frame(frame), self.get_event_loop()) - future.result() + frame = await self._camera_in_queue.get() + await self._internal_push_frame(frame) self._camera_in_queue.task_done() - except queue.Empty: - pass + except asyncio.CancelledError: + break except BaseException as e: logger.error(f"Error capturing video: {e}") @@ -605,34 +599,39 @@ class DailyOutputTransport(BaseOutputTransport): self._client = client + self._executor = ThreadPoolExecutor(max_workers=5) + async def start(self, frame: StartFrame): - if self._running: - return - # This will set _running=True - await super().start(frame) # Join the room. await self._client.join() + await super().start(frame) async def stop(self): - if not self._running: - return - # This will set _running=False - await super().stop() # Leave the room. await self._client.leave() + await super().stop() async def cleanup(self): - await super().cleanup() await self._client.cleanup() + await super().cleanup() - def send_message(self, frame: DailyTransportMessageFrame): - self._client.send_message(frame) + async def send_message(self, frame: DailyTransportMessageFrame): + def send_message(frame: DailyTransportMessageFrame): + self._client.send_message(frame) - def write_raw_audio_frames(self, frames: bytes): - self._client.write_raw_audio_frames(frames) + await self.get_event_loop().run_in_executor(self._executor, send_message, frame) - def write_frame_to_camera(self, frame: ImageRawFrame): - self._client.write_frame_to_camera(frame) + async def write_raw_audio_frames(self, frames: bytes): + def write_audio(frames: bytes): + self._client.write_raw_audio_frames(frames) + + await self.get_event_loop().run_in_executor(self._executor, write_audio, frames) + + async def write_frame_to_camera(self, frame: ImageRawFrame): + def write_image(frame: ImageRawFrame): + self._client.write_frame_to_camera(frame) + + await self.get_event_loop().run_in_executor(self._executor, write_image, frame) class DailyTransport(BaseTransport): @@ -643,7 +642,7 @@ class DailyTransport(BaseTransport): token: str | None, bot_name: str, params: DailyParams, - loop: asyncio.AbstractEventLoop = asyncio.get_event_loop()): + loop: asyncio.AbstractEventLoop | None = None): super().__init__(loop) callbacks = DailyCallbacks( @@ -663,7 +662,8 @@ class DailyTransport(BaseTransport): ) self._params = params - self._client = DailyTransportClient(room_url, token, bot_name, params, callbacks, loop) + self._client = DailyTransportClient( + room_url, token, bot_name, params, callbacks, self._loop) self._input: DailyInputTransport | None = None self._output: DailyOutputTransport | None = None @@ -831,6 +831,5 @@ class DailyTransport(BaseTransport): 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) + self._call_event_handler(event_name, *args, **kwargs), self._loop) future.result() diff --git a/src/pipecat/vad/silero.py b/src/pipecat/vad/silero.py index 78ea4f9b7..65a4283f5 100644 --- a/src/pipecat/vad/silero.py +++ b/src/pipecat/vad/silero.py @@ -46,7 +46,7 @@ class SileroVADAnalyzer(VADAnalyzer): def num_frames_required(self) -> int: return int(self.sample_rate / 100) * 4 # 40ms - def voice_confidence(self, buffer) -> float: + async def voice_confidence(self, buffer) -> float: try: audio_int16 = np.frombuffer(buffer, np.int16) # Divide by 32768 because we have signed 16-bit data. @@ -88,7 +88,7 @@ class SileroVAD(FrameProcessor): async def _analyze_audio(self, frame: AudioRawFrame): # Check VAD and push event if necessary. We just care about changes # from QUIET to SPEAKING and vice versa. - new_vad_state = self._vad_analyzer.analyze_audio(frame.audio) + new_vad_state = await self._vad_analyzer.analyze_audio(frame.audio) if new_vad_state != self._processor_vad_state and new_vad_state != VADState.STARTING and new_vad_state != VADState.STOPPING: new_frame = None diff --git a/src/pipecat/vad/vad_analyzer.py b/src/pipecat/vad/vad_analyzer.py index cd49d282b..e26e5a283 100644 --- a/src/pipecat/vad/vad_analyzer.py +++ b/src/pipecat/vad/vad_analyzer.py @@ -58,14 +58,14 @@ class VADAnalyzer: pass @abstractmethod - def voice_confidence(self, buffer) -> float: + async def voice_confidence(self, buffer) -> float: pass def _get_smoothed_volume(self, audio: bytes) -> float: volume = calculate_audio_volume(audio, self._sample_rate) return exp_smoothing(volume, self._prev_volume, self._smoothing_factor) - def analyze_audio(self, buffer) -> VADState: + async def analyze_audio(self, buffer) -> VADState: self._vad_buffer += buffer num_required_bytes = self._vad_frames_num_bytes @@ -75,7 +75,7 @@ class VADAnalyzer: audio_frames = self._vad_buffer[:num_required_bytes] self._vad_buffer = self._vad_buffer[num_required_bytes:] - confidence = self.voice_confidence(audio_frames) + confidence = await self.voice_confidence(audio_frames) volume = self._get_smoothed_volume(audio_frames) self._prev_volume = volume