From cd2c9700ad9866d810d54c23df57f28fdeb39caf Mon Sep 17 00:00:00 2001 From: Moishe Lettvin Date: Thu, 4 Jan 2024 17:54:13 -0500 Subject: [PATCH] more demo cleanup, allow bundled frames in output_queue --- .../services/daily_transport_service.py | 68 ++++++++++--------- src/samples/05-sync-speech-and-text.py | 66 +++++++++++------- 2 files changed, 80 insertions(+), 54 deletions(-) diff --git a/src/dailyai/services/daily_transport_service.py b/src/dailyai/services/daily_transport_service.py index 63567b1b7..dc7e7c0ae 100644 --- a/src/dailyai/services/daily_transport_service.py +++ b/src/dailyai/services/daily_transport_service.py @@ -147,9 +147,11 @@ class DailyTransportService(EventHandler): def stop(self): self.stop_threads.set() - self.camera_thread.join() - self.output_queue.put(OutputQueueFrame(FrameType.END_STREAM, None)) - self.frame_consumer_thread.join() + if self.camera_thread and self.camera_thread.is_alive(): + self.camera_thread.join() + if self.frame_consumer_thread and self.frame_consumer_thread.is_alive(): + self.output_queue.put(OutputQueueFrame(FrameType.END_STREAM, None)) + self.frame_consumer_thread.join() self.client.leave() def call_joined(self, join_data, client_error): @@ -223,38 +225,42 @@ class DailyTransportService(EventHandler): all_audio_frames = bytearray() while True: try: - frame: OutputQueueFrame = self.output_queue.get() - if frame.frame_type == FrameType.END_STREAM: - self.logger.info("Stopping frame consumer thread") - return + frames: OutputQueueFrame | list[OutputQueueFrame] = self.output_queue.get() + if type(frames) != list: + frames = [frames] - # if interrupted, we just pull frames off the queue and discard them - if not self.is_interrupted.is_set(): - if frame: - if frame.frame_type == FrameType.AUDIO_FRAME: - chunk = frame.frame_data + for frame in frames: + if frame.frame_type == FrameType.END_STREAM: + self.logger.info("Stopping frame consumer thread") + return - all_audio_frames.extend(chunk) + # if interrupted, we just pull frames off the queue and discard them + if not self.is_interrupted.is_set(): + if frame: + if frame.frame_type == FrameType.AUDIO_FRAME: + chunk = frame.frame_data - b.extend(chunk) - l = len(b) - (len(b) % smallest_write_size) - if l: - self.mic.write_frames(bytes(b[:l])) - b = b[l:] - elif frame.frame_type == FrameType.IMAGE_FRAME: - self.set_image(frame.frame_data) - elif len(b): - self.mic.write_frames(bytes(b)) - b = bytearray() - else: - if self.interrupt_time: - self.logger.info( - f"Lag to stop stream after interruption {time.perf_counter() - self.interrupt_time}" - ) - self.interrupt_time = None + all_audio_frames.extend(chunk) - if frame.frame_type == FrameType.START_STREAM: - self.is_interrupted.clear() + b.extend(chunk) + l = len(b) - (len(b) % smallest_write_size) + if l: + self.mic.write_frames(bytes(b[:l])) + b = b[l:] + elif frame.frame_type == FrameType.IMAGE_FRAME: + self.set_image(frame.frame_data) + elif len(b): + self.mic.write_frames(bytes(b)) + b = bytearray() + else: + if self.interrupt_time: + self.logger.info( + f"Lag to stop stream after interruption {time.perf_counter() - self.interrupt_time}" + ) + self.interrupt_time = None + + if frame.frame_type == FrameType.START_STREAM: + self.is_interrupted.clear() self.output_queue.task_done() except Empty: diff --git a/src/samples/05-sync-speech-and-text.py b/src/samples/05-sync-speech-and-text.py index de3f0cc80..1c04acc96 100644 --- a/src/samples/05-sync-speech-and-text.py +++ b/src/samples/05-sync-speech-and-text.py @@ -26,32 +26,52 @@ async def main(room_url, token): tts = AzureTTSService() dalle = AzureImageGenServiceREST() - inference_text_process = llm.run_llm( - [ - { - "role": "system", - "content": f"Describe a nature photograph suitable for use in a calendar, for the month of January. Include only the image description with no preamble." - } - ] - ) + async def get_all_audio(text): + all_audio = bytearray() + async for audio in tts.run_tts(text): + all_audio.append(audio) + + return all_audio + + async def show_month(month): + print(f"Running llm for {month}") + inference_text = await llm.run_llm( + [ + { + "role": "system", + "content": f"Describe a nature photograph suitable for use in a calendar, for the month of {month}. Include only the image description with no preamble." + } + ] + ) + print(f"got llm for {month}") + + (image, audio) = await asyncio.gather( + *[dalle.run_image_gen(inference_text, "1024x1024"), get_all_audio(inference_text)] + ) + print(f"Got audio and video for {month}") + transport.output_queue.put( + [ + OutputQueueFrame(FrameType.IMAGE_FRAME, image[1]), + OutputQueueFrame(FrameType.AUDIO_FRAME, audio), + ] + ) try: transport.run() - - inference_text = await inference_text_process - - tts_iterator = tts.run_tts(inference_text) - (image, audio) = await asyncio.gather( - *[dalle.run_image_gen(inference_text, "1024x1024"), anext(tts_iterator)] - ) - transport.output_queue.put(OutputQueueFrame(FrameType.IMAGE_FRAME, image[1])) - transport.output_queue.put(OutputQueueFrame(FrameType.AUDIO_FRAME, audio)) - async for audio in tts_iterator: - transport.output_queue.put( - OutputQueueFrame(FrameType.AUDIO_FRAME, audio) - ) - - await asyncio.sleep(meeting_duration_minutes * 60) + months = [ + "January", + "February", + "March", + "April", + "May", + "June", + "July", + "August", + "September", + "October", + "November", + ] + await asyncio.gather(*[show_month(month) for month in months]) finally: transport.stop() print("Done")