From 115b744861af7a723b8b77130bfe8be50dc22723 Mon Sep 17 00:00:00 2001 From: Moishe Lettvin Date: Tue, 16 Jan 2024 21:11:38 -0500 Subject: [PATCH] input/output -> receive/send queues, for transport service --- .../services/daily_transport_service.py | 37 ++++++++++--------- .../02-llm-say-one-thing.py | 2 +- .../06-listen-and-respond.py | 4 +- 3 files changed, 22 insertions(+), 21 deletions(-) diff --git a/src/dailyai/services/daily_transport_service.py b/src/dailyai/services/daily_transport_service.py index 55b6ad001..3147ecfa1 100644 --- a/src/dailyai/services/daily_transport_service.py +++ b/src/dailyai/services/daily_transport_service.py @@ -35,10 +35,11 @@ class DailyTransportService(EventHandler): self.duration: float = duration self.expiration = time.time() + duration * 60 - # This queue is used to marshal frames from the async output queue to the thread that emits audio & video. + # This queue is used to marshal frames from the async send queue to the thread that emits audio & video. # We need this to maintain the asynchronous behavior of asyncio queues -- to give async functions - # a chance to run while waiting for queue items -- but also to maintain thread safety. - self.threadsafe_output_queue = Queue() + # a chance to run while waiting for queue items -- but also to maintain thread safety and have a threaded + # handler to send frames, to ensure that sending isn't subject to pauses in the async thread. + self.threadsafe_send_queue = Queue() self.is_interrupted = Event() self.stop_threads = Event() @@ -49,8 +50,8 @@ class DailyTransportService(EventHandler): self.camera_height = 768 self.camera_enabled = False - self.output_queue = asyncio.Queue() - self.media_queue = asyncio.Queue() + self.send_queue = asyncio.Queue() + self.receive_queue = asyncio.Queue() self.other_participant_has_joined = False @@ -187,21 +188,21 @@ class DailyTransportService(EventHandler): self.my_participant_id = self.client.participants()["local"]["id"] - async def get_media_frames(self): + async def get_receive_frames(self): while True: - frame = await self.media_queue.get() + frame = await self.receive_queue.get() yield frame if frame.frame_type == FrameType.END_STREAM: break - def get_async_output_queue(self): - return self.output_queue + def get_async_send_queue(self): + return self.send_queue async def marshal_frames(self): while True: - frame = await self.output_queue.get() - self.threadsafe_output_queue.put(frame) - self.output_queue.task_done() + frame = await self.send_queue.get() + self.threadsafe_send_queue.put(frame) + self.send_queue.task_done() if frame.frame_type == FrameType.END_STREAM: break @@ -224,8 +225,8 @@ class DailyTransportService(EventHandler): self.stop_threads.set() - await self.media_queue.put(QueueFrame(FrameType.END_STREAM, None)) - await self.output_queue.put(QueueFrame(FrameType.END_STREAM, None)) + await self.receive_queue.put(QueueFrame(FrameType.END_STREAM, None)) + await self.send_queue.put(QueueFrame(FrameType.END_STREAM, None)) await async_output_queue_marshal_task if self.camera_thread and self.camera_thread.is_alive(): @@ -264,7 +265,7 @@ class DailyTransportService(EventHandler): def on_transcription_message(self, message:dict): if self.loop: frame = QueueFrame(FrameType.TRANSCRIPTION, message) - asyncio.run_coroutine_threadsafe(self.media_queue.put(frame), self.loop) + asyncio.run_coroutine_threadsafe(self.receive_queue.put(frame), self.loop) def on_transcription_stopped(self, stopped_by, stopped_by_error): pass @@ -295,7 +296,7 @@ class DailyTransportService(EventHandler): all_audio_frames = bytearray() while True: try: - frames_or_frame: QueueFrame | list[QueueFrame] = self.threadsafe_output_queue.get() + frames_or_frame: QueueFrame | list[QueueFrame] = self.threadsafe_send_queue.get() if type(frames_or_frame) == QueueFrame: frames: list[QueueFrame] = [frames_or_frame] elif type(frames_or_frame) == list: @@ -306,7 +307,7 @@ class DailyTransportService(EventHandler): for frame in frames: if frame.frame_type == FrameType.END_STREAM: self.logger.info("Stopping frame consumer thread") - self.threadsafe_output_queue.task_done() + self.threadsafe_send_queue.task_done() return # if interrupted, we just pull frames off the queue and discard them @@ -337,7 +338,7 @@ class DailyTransportService(EventHandler): if frame.frame_type == FrameType.START_STREAM: self.is_interrupted.clear() - self.threadsafe_output_queue.task_done() + self.threadsafe_send_queue.task_done() except Empty: try: if len(b): diff --git a/src/samples/theoretical-to-real/02-llm-say-one-thing.py b/src/samples/theoretical-to-real/02-llm-say-one-thing.py index a0667b708..b1688eace 100644 --- a/src/samples/theoretical-to-real/02-llm-say-one-thing.py +++ b/src/samples/theoretical-to-real/02-llm-say-one-thing.py @@ -21,7 +21,7 @@ async def main(room_url): llm_to_tts_queue = asyncio.Queue() tts = ElevenLabsTTSService( - llm_to_tts_queue, transport.get_async_output_queue(), voice_id="29vD33N1CtxCmqQRPOHJ" + llm_to_tts_queue, transport.get_async_send_queue(), voice_id="29vD33N1CtxCmqQRPOHJ" ) llm = AzureLLMService(text_to_llm_queue, llm_to_tts_queue) diff --git a/src/samples/theoretical-to-real/06-listen-and-respond.py b/src/samples/theoretical-to-real/06-listen-and-respond.py index a6e8213f2..d4bcf492f 100644 --- a/src/samples/theoretical-to-real/06-listen-and-respond.py +++ b/src/samples/theoretical-to-real/06-listen-and-respond.py @@ -32,7 +32,7 @@ async def main(room_url:str, token): ] sentence = "" - async for frame in transport.get_media_frames(): + async for frame in transport.get_receive_frames(): if frame.frame_type != FrameType.TRANSCRIPTION: continue @@ -50,7 +50,7 @@ async def main(room_url:str, token): async for response in llm.run_llm_async_sentences(messages): full_response += response async for audio in tts.run_tts(response): - await transport.output_queue.put(QueueFrame(FrameType.AUDIO, audio)) + await transport.send_queue.put(QueueFrame(FrameType.AUDIO, audio)) messages.append({"role": "assistant", "content": full_response})