From acf6dc0a304c303f1172cda069af936660e57590 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Wed, 15 May 2024 15:23:03 -0700 Subject: [PATCH] transports: more start and stop fixes --- CHANGELOG.md | 7 ++- src/pipecat/pipeline/task.py | 2 + src/pipecat/transports/base_input.py | 31 +++++------ src/pipecat/transports/base_output.py | 32 +++++------ src/pipecat/transports/local/audio.py | 16 ++++++ src/pipecat/transports/local/tk.py | 20 ++++++- src/pipecat/transports/services/daily.py | 67 +++++++++++++++--------- 7 files changed, 115 insertions(+), 60 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index e1220a3aa..97cdc8420 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,9 +7,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added + +- Added `DailyTransport` event `on_participant_left`. + ### Fixed -- Fixed `DailyInputTransport` and `DailyOutputTransport` stop/cleanup ordering. +- Fixed transport start and stop. In some situation processors would halt or not + shutdown properly. ## [0.0.13] - 2024-05-14 diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index 70504d2ca..2ecbe84ca 100644 --- a/src/pipecat/pipeline/task.py +++ b/src/pipecat/pipeline/task.py @@ -53,6 +53,8 @@ class PipelineTask: async def run(self): await asyncio.gather(self._process_task_queue(), self._process_up_queue()) + await self._source.cleanup() + await self._pipeline.cleanup() async def queue_frame(self, frame: Frame): await self._task_queue.put(frame) diff --git a/src/pipecat/transports/base_input.py b/src/pipecat/transports/base_input.py index 786e991fa..aaa1e0a29 100644 --- a/src/pipecat/transports/base_input.py +++ b/src/pipecat/transports/base_input.py @@ -6,7 +6,6 @@ import asyncio import queue -import threading from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.frames.frames import ( @@ -35,10 +34,6 @@ class BaseInputTransport(FrameProcessor): # Start media threads. if self._params.audio_in_enabled or self._params.vad_enabled: self._audio_in_queue = queue.Queue() - self._audio_in_thread = threading.Thread(target=self._audio_in_thread_handler) - self._audio_out_thread = threading.Thread(target=self._audio_out_thread_handler) - - self._stopped_event = asyncio.Event() async def start(self): if self._running: @@ -47,14 +42,21 @@ class BaseInputTransport(FrameProcessor): self._running = True if self._params.audio_in_enabled or self._params.vad_enabled: - self._audio_in_thread.start() - self._audio_out_thread.start() + loop = asyncio.get_running_loop() + self._audio_in_thread = loop.run_in_executor(None, self._audio_in_thread_handler) + self._audio_out_thread = loop.run_in_executor(None, self._audio_out_thread_handler) async def stop(self): + if not self._running: + return + # This will exit all threads. self._running = False - self._stopped_event.set() + # Wait for the threads to finish. + if self._params.audio_in_enabled or self._params.vad_enabled: + await self._audio_in_thread + await self._audio_out_thread def vad_analyze(self, audio_frames: bytes) -> VADState: pass @@ -67,25 +69,18 @@ class BaseInputTransport(FrameProcessor): # async def cleanup(self): - if self._params.audio_in_enabled or self._params.vad_enabled: - self._audio_in_thread.join() - self._audio_out_thread.join() + pass async def process_frame(self, frame: Frame, direction: FrameDirection): if isinstance(frame, StartFrame): - await self.push_frame(frame, direction) await self.start() - elif isinstance(frame, CancelFrame) or isinstance(frame, EndFrame): await self.push_frame(frame, direction) + elif isinstance(frame, CancelFrame) or isinstance(frame, EndFrame): await self.stop() + await self.push_frame(frame, direction) else: await self.push_frame(frame, direction) - # If we are finishing, wait here until we have stopped, otherwise we - # might close things too early upstream. - if isinstance(frame, CancelFrame) or isinstance(frame, EndFrame): - await self._stopped_event.wait() - # # Audio input # diff --git a/src/pipecat/transports/base_output.py b/src/pipecat/transports/base_output.py index 7b33308a2..c9e3678b1 100644 --- a/src/pipecat/transports/base_output.py +++ b/src/pipecat/transports/base_output.py @@ -8,7 +8,6 @@ import asyncio import itertools import queue -import threading import time from typing import List @@ -41,13 +40,10 @@ class BaseOutputTransport(FrameProcessor): # framerate. self._camera_images = None - # Start media threads. + # Create media threads queues. if self._params.camera_out_enabled: self._camera_out_queue = queue.Queue() - self._camera_out_thread = threading.Thread(target=self._camera_out_thread_handler) - self._sink_queue = queue.Queue() - self._sink_thread = threading.Thread(target=self._sink_thread_handler) self._stopped_event = asyncio.Event() @@ -57,12 +53,17 @@ class BaseOutputTransport(FrameProcessor): self._running = True - if self._params.camera_out_enabled: - self._camera_out_thread.start() + loop = asyncio.get_running_loop() - self._sink_thread.start() + if self._params.camera_out_enabled: + self._camera_out_thread = loop.run_in_executor(None, self._camera_out_thread_handler) + + self._sink_thread = loop.run_in_executor(None, self._sink_thread_handler) async def stop(self): + if not self._running: + return + # This will exit all threads. self._running = False @@ -82,26 +83,28 @@ class BaseOutputTransport(FrameProcessor): # async def cleanup(self): + # Wait on the threads to finish. if self._params.camera_out_enabled: - self._camera_out_thread.join() + await self._camera_out_thread - self._sink_thread.join() + await self._sink_thread async def process_frame(self, frame: Frame, direction: FrameDirection): if isinstance(frame, StartFrame): - await self.push_frame(frame, direction) await self.start() + await self.push_frame(frame, direction) # EndFrame is managed in the queue handler. elif isinstance(frame, CancelFrame): - await self.push_frame(frame, direction) await self.stop() + await self.push_frame(frame, direction) elif self._frame_managed_by_sink(frame): self._sink_queue.put(frame) else: await self.push_frame(frame, direction) # If we are finishing, wait here until we have stopped, otherwise we might - # close things too early upstream. + # close things too early upstream. We need this event because we don't + # know when the internal threads will finish. if isinstance(frame, CancelFrame) or isinstance(frame, EndFrame): await self._stopped_event.wait() @@ -110,7 +113,6 @@ class BaseOutputTransport(FrameProcessor): or isinstance(frame, ImageRawFrame) or isinstance(frame, SpriteFrame) or isinstance(frame, TransportMessageFrame) - or isinstance(frame, CancelFrame) or isinstance(frame, EndFrame)) def _sink_thread_handler(self): @@ -120,7 +122,7 @@ class BaseOutputTransport(FrameProcessor): while self._running: try: frame = self._sink_queue.get(timeout=1) - if isinstance(frame, CancelFrame) or isinstance(frame, EndFrame): + 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()) diff --git a/src/pipecat/transports/local/audio.py b/src/pipecat/transports/local/audio.py index c0038d250..ee85bcf61 100644 --- a/src/pipecat/transports/local/audio.py +++ b/src/pipecat/transports/local/audio.py @@ -37,6 +37,14 @@ class LocalAudioInputTransport(BaseInputTransport): def read_raw_audio_frames(self, frame_count: int) -> bytes: return self._in_stream.read(frame_count, exception_on_overflow=False) + async def start(self): + await super().start() + self._in_stream.start_stream() + + async def stop(self): + await super().stop() + self._in_stream.stop_stream() + async def cleanup(self): # This is not very pretty (taken from PyAudio docs). while self._in_stream.is_active(): @@ -60,6 +68,14 @@ class LocalAudioOutputTransport(BaseOutputTransport): def write_raw_audio_frames(self, frames: bytes): self._out_stream.write(frames) + async def start(self): + await super().start() + self._out_stream.start_stream() + + async def stop(self): + await super().stop() + self._out_stream.stop_stream() + async def cleanup(self): # This is not very pretty (taken from PyAudio docs). while self._out_stream.is_active(): diff --git a/src/pipecat/transports/local/tk.py b/src/pipecat/transports/local/tk.py index 5a32e85d7..4b1d4e297 100644 --- a/src/pipecat/transports/local/tk.py +++ b/src/pipecat/transports/local/tk.py @@ -48,6 +48,14 @@ class TkInputTransport(BaseInputTransport): def read_raw_audio_frames(self, frame_count: int) -> bytes: return self._in_stream.read(frame_count, exception_on_overflow=False) + async def start(self): + await super().start() + self._in_stream.start_stream() + + async def stop(self): + await super().stop() + self._in_stream.stop_stream() + async def cleanup(self): # This is not very pretty (taken from PyAudio docs). while self._in_stream.is_active(): @@ -79,7 +87,17 @@ class TkOutputTransport(BaseOutputTransport): self._out_stream.write(frames) def write_frame_to_camera(self, frame: ImageRawFrame): - asyncio.run_coroutine_threadsafe(self._write_frame_to_tk(frame), self.get_event_loop()) + future = asyncio.run_coroutine_threadsafe( + self._write_frame_to_tk(frame), self.get_event_loop()) + future.result() + + async def start(self): + await super().start() + self._out_stream.start_stream() + + async def stop(self): + await super().stop() + self._out_stream.stop_stream() async def cleanup(self): # This is not very pretty (taken from PyAudio docs). diff --git a/src/pipecat/transports/services/daily.py b/src/pipecat/transports/services/daily.py index d3ede3b31..fc29fa81d 100644 --- a/src/pipecat/transports/services/daily.py +++ b/src/pipecat/transports/services/daily.py @@ -7,7 +7,6 @@ import asyncio import inspect import queue -import threading import time import types @@ -108,7 +107,7 @@ class DailyCallbacks(BaseModel): on_error: Callable[[str], None] -class DailySession(EventHandler): +class DailyTransportClient(EventHandler): _daily_initialized: bool = False @@ -413,34 +412,44 @@ class DailySession(EventHandler): class DailyInputTransport(BaseInputTransport): - def __init__(self, session: DailySession, params: DailyParams): + def __init__(self, client: DailyTransportClient, params: DailyParams): super().__init__(params) - self._session = session + self._client = client self._video_renderers = {} self._camera_in_queue = queue.Queue() - self._camera_in_thread = threading.Thread(target=self._camera_in_thread_handler) async def start(self): + if self._running: + return + # Join the room. + await self._client.join() + # This will set _running=True await super().start() - self._camera_in_thread.start() - await self._session.join() + # Create camera in thread (runs if _running is true). + loop = asyncio.get_running_loop() + self._camera_in_thread = loop.run_in_executor(None, self._camera_in_thread_handler) async def stop(self): - await self._session.leave() + if not self._running: + return + # Leave the room. + await self._client.leave() + # This will set _running=False await super().stop() + # The thread will stop. + await self._camera_in_thread async def cleanup(self): - self._camera_in_thread.join() - await self._session.cleanup() await super().cleanup() + await self._client.cleanup() def vad_analyze(self, audio_frames: bytes) -> VADState: - return self._session.vad_analyze(audio_frames) + return self._client.vad_analyze(audio_frames) def read_raw_audio_frames(self, frame_count: int) -> bytes: - return self._session.read_raw_audio_frames(frame_count) + return self._client.read_raw_audio_frames(frame_count) # # FrameProcessor @@ -468,7 +477,7 @@ class DailyInputTransport(BaseInputTransport): "render_next_frame": False, } - self._session.capture_participant_video( + self._client.capture_participant_video( participant_id, self._on_participant_video_frame, framerate, @@ -519,28 +528,36 @@ class DailyInputTransport(BaseInputTransport): class DailyOutputTransport(BaseOutputTransport): - def __init__(self, session: DailySession, params: DailyParams): + def __init__(self, client: DailyTransportClient, params: DailyParams): super().__init__(params) - self._session = session + self._client = client async def start(self): + if self._running: + return + # This will set _running=True await super().start() - await self._session.join() + # Join the room. + await self._client.join() async def stop(self): - await self._session.leave() + if not self._running: + return + # This will set _running=False await super().stop() + # Leave the room. + await self._client.leave() async def cleanup(self): - await self._session.cleanup() await super().cleanup() + await self._client.cleanup() def write_raw_audio_frames(self, frames: bytes): - self._session.write_raw_audio_frames(frames) + self._client.write_raw_audio_frames(frames) def write_frame_to_camera(self, frame: ImageRawFrame): - self._session.write_frame_to_camera(frame) + self._client.write_frame_to_camera(frame) class DailyTransport(BaseTransport): @@ -556,7 +573,7 @@ class DailyTransport(BaseTransport): ) self._params = params - self._session = DailySession(room_url, token, bot_name, params, callbacks) + self._client = DailyTransportClient(room_url, token, bot_name, params, callbacks) self._input: DailyInputTransport | None = None self._output: DailyOutputTransport | None = None self._loop = asyncio.get_running_loop() @@ -577,12 +594,12 @@ class DailyTransport(BaseTransport): def input(self) -> FrameProcessor: if not self._input: - self._input = DailyInputTransport(self._session, self._params) + self._input = DailyInputTransport(self._client, self._params) return self._input def output(self) -> FrameProcessor: if not self._output: - self._output = DailyOutputTransport(self._session, self._params) + self._output = DailyOutputTransport(self._client, self._params) return self._output # @@ -591,7 +608,7 @@ class DailyTransport(BaseTransport): @property def participant_id(self) -> str: - return self._session.participant_id + return self._client.participant_id async def send_image(self, frame: ImageRawFrame | SpriteFrame): if self._output: @@ -602,7 +619,7 @@ class DailyTransport(BaseTransport): await self._output.process_frame(frame, FrameDirection.DOWNSTREAM) def capture_participant_transcription(self, participant_id: str): - self._session.capture_participant_transcription( + self._client.capture_participant_transcription( participant_id, self._on_transcription_message )