more demo cleanup, allow bundled frames in output_queue

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()
+        if self.camera_thread and self.camera_thread.is_alive():
-        self.output_queue.put(OutputQueueFrame(FrameType.END_STREAM, None))
+            self.camera_thread.join()
-        self.frame_consumer_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()
+                frames: OutputQueueFrame | list[OutputQueueFrame] = self.output_queue.get()
-                if frame.frame_type == FrameType.END_STREAM:
+                if type(frames) != list:
-                    self.logger.info("Stopping frame consumer thread")
+                    frames = [frames]
-                    return
 
-                # if interrupted, we just pull frames off the queue and discard them
+                for frame in frames:
-                if not self.is_interrupted.is_set():
+                    if frame.frame_type == FrameType.END_STREAM:
-                    if frame:
+                        self.logger.info("Stopping frame consumer thread")
-                        if frame.frame_type == FrameType.AUDIO_FRAME:
+                        return
-                            chunk = frame.frame_data
 
-                            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)
+                                all_audio_frames.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:
+                                b.extend(chunk)
-                        self.is_interrupted.clear()
+                                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:

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(
+    async def get_all_audio(text):
-        [
+        all_audio = bytearray()
-            {
+        async for audio in tts.run_tts(text):
-                "role": "system",
+            all_audio.append(audio)
-                "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."
+
-            }
+        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()
-
+        months = [
-        inference_text = await inference_text_process
+            "January",
-
+            "February",
-        tts_iterator = tts.run_tts(inference_text)
+            "March",
-        (image, audio) = await asyncio.gather(
+            "April",
-            *[dalle.run_image_gen(inference_text, "1024x1024"), anext(tts_iterator)]
+            "May",
-        )
+            "June",
-        transport.output_queue.put(OutputQueueFrame(FrameType.IMAGE_FRAME, image[1]))
+            "July",
-        transport.output_queue.put(OutputQueueFrame(FrameType.AUDIO_FRAME, audio))
+            "August",
-        async for audio in tts_iterator:
+            "September",
-            transport.output_queue.put(
+            "October",
-                OutputQueueFrame(FrameType.AUDIO_FRAME, audio)
+            "November",
-            )
+        ]
-
+        await asyncio.gather(*[show_month(month) for month in months])
-        await asyncio.sleep(meeting_duration_minutes * 60)
     finally:
         transport.stop()
     print("Done")