fix audio issue when starting a new responce

app_interface_actor.py

@@ -1,16 +1,32 @@
+import time
 import ray
-from ray.util.queue import Queue
+from ray.util.queue import Queue, Empty
 from ray.actor import ActorHandle
-import torch
 import numpy as np
 import pid_helper
 
+# Ray Queue's take ~.5 seconds to splin up; 
+# this class creates a pool of queues to cycle through
+class QueueFactory:
+    def __init__(self, max_size:int):
+        self.queues:[Queue] = []
+        self.queue_size = 5
+        self.max_size = max_size
+        while len(self.queues) < self.queue_size:
+            self.queues.append(Queue(maxsize=max_size))
+
+    def get_queue(self)->Queue:
+        queue = self.queues.pop(0)
+        self.queues.append(Queue(maxsize=self.max_size))
+        return queue
+        
 @ray.remote
 class AppInterfaceActor:
     def __init__(self):
         self.audio_input_queue = Queue(maxsize=3000)  # Adjust the size as needed
         self.video_input_queue = Queue(maxsize=10)  # Adjust the size as needed
-        self.audio_output_queue = Queue(maxsize=50)  # Adjust the size as needed
+        self.audio_output_queue_factory = QueueFactory(max_size=50)
+        self.audio_output_queue = self.audio_output_queue_factory.get_queue()
         self.video_output_queue = Queue(maxsize=10)  # Adjust the size as needed
         self.debug_str = ""
         self.state = "Initializing"
@@ -37,9 +53,14 @@ class AppInterfaceActor:
         await self.audio_input_queue.put_async(shared_buffer_ref)
 
     async def dequeue_audio_output_frame_async(self):
-        if self.audio_output_queue.empty():
-            return None
-        frame = await self.audio_output_queue.get_async()
+        start_time = time.time()
+        try:
+            frame = await self.audio_output_queue.get_async(block=False)
+        except Empty:
+            frame = None
+        elapsed_time = time.time() - start_time
+        if elapsed_time > 0.1:
+            print (f"dequeue_audio_output_frame_async time: {elapsed_time}. was empty: {frame is None}. frame type: {type(frame) if frame else str(0)}")
         return frame
 
     async def dequeue_video_output_frames_async(self):
@@ -51,9 +72,14 @@ class AppInterfaceActor:
             video_frames.append(shared_tensor)
         return video_frames
 
-# functions for application to dequeue input, enqueue output    
+# functions for application to dequeue input, enqueue output
     def get_audio_output_queue(self)->Queue:
         return self.audio_output_queue
+
+    async def cycle_output_queue(self)->Queue:
+        self.audio_output_queue.shutdown(grace_period_s=0.1)
+        self.audio_output_queue = self.audio_output_queue_factory.get_queue()
+        return self.audio_output_queue
     
     async def enqueue_video_output_frame(self, shared_tensor_ref):
         if self.video_output_queue.full():

charles_app.py

@@ -1,4 +1,3 @@
-import json
 import ray
 import time
 import asyncio
@@ -62,12 +61,12 @@ class CharlesApp:
         self.set_state("010 - Initialized")
 
     async def cancel_response_task(self):
-        if self._respond_to_prompt_task is None:
-            return
-        await self._respond_to_prompt.terminate()
-        self._respond_to_prompt_task.cancel()
-        self._respond_to_prompt_task = None
-        self._respond_to_prompt = None
+        if self._respond_to_prompt_task is not None:
+            await self._respond_to_prompt.terminate()
+            self._respond_to_prompt_task.cancel()
+            self._respond_to_prompt_task = None
+            self._respond_to_prompt = None
+            self._audio_output_queue = await self._app_interface_actor.cycle_output_queue.remote()
         
     async def start(self):
         if self._needs_init:
@@ -164,7 +163,7 @@ class CharlesApp:
                         if additional_prompt is not None:
                             prompt = additional_prompt + ". " + prompt                        
                         human_preview_text = f"👨❓ {prompt}"
-                        # await self.cancel_response_task() # TODO re-enable to interupt when user speaks
+                        await self.cancel_response_task()
 
 
             # i choose to add each line of responce one at a time as them come in

ffmpeg_converter.py

@@ -1,7 +1,6 @@
-
 import asyncio
 import ray
-from ray.util.queue import Queue
+from ray.util.queue import Queue, Full
 
 class FFMpegConverter:
     def __init__(self, output_queue: Queue, buffer_size: int = 1920, output_format: str='s16le'):
@@ -24,9 +23,19 @@ class FFMpegConverter:
                 # If the pipe is broken, restart the process.
                 await self.start_process()
                 continue
+            if self.running == False:
+                return
             # print(f"FFMpegConverter: read {len(chunk)} bytes")
             chunk_ref = ray.put(chunk)
-            await self.output_queue.put_async(chunk_ref)       
+            keep_trying = True
+            while keep_trying:
+                try:
+                    await self.output_queue.put_async(chunk_ref, timeout=0.01)
+                    keep_trying = False
+                except Full:
+                    if self.running == False:
+                        return                    
+                    await asyncio.sleep(0.01)
 
     async def start_process(self):
         cmd = [
@@ -52,6 +61,8 @@ class FFMpegConverter:
         # print (f"input_pipe: {self.input_pipe}")
 
     async def push_chunk(self, chunk):
+        if self.running == False:
+            return
         try:
             self.input_pipe.write(chunk)
             await self.input_pipe.drain()
@@ -67,5 +78,3 @@ class FFMpegConverter:
             self.process.stdin.transport.close()
             self.process.kill()
             self.process.terminate()
-        # while not self.output_queue.empty():
-        #     await self.output_queue.get_async()

respond_to_prompt_async.py

@@ -1,10 +1,7 @@
 from asyncio import Queue, TaskGroup
 import asyncio
-from contextlib import asynccontextmanager
-
 import ray
 from chat_service import ChatService
-# from local_speaker_service import LocalSpeakerService
 from text_to_speech_service import TextToSpeechService
 from response_state_manager import ResponseStateManager
 from ffmpeg_converter import FFMpegConverter
@@ -97,23 +94,23 @@ class RespondToPromptAsync:
 
     async def terminate(self):
         # Cancel tasks
+        all_tasks = []
         if self.task_group_tasks:
             for task in self.task_group_tasks:
                 task.cancel()
+            all_tasks.extend(self.task_group_tasks)
         for task in self.sentence_tasks:
             task.cancel()
+        all_tasks.extend(self.sentence_tasks)
+        await asyncio.gather(*all_tasks, return_exceptions=True)
 
         # Close FFmpeg converter actor
         if self.ffmpeg_converter_task:
-            self.ffmpeg_converter_task.cancel()
             await self.ffmpeg_converter.close()
+            self.ffmpeg_converter_task.cancel()
         # ray.kill(self.ffmpeg_converter)
 
         # Flush all queues
-        # TODO re-enable to interupt when user speaks
-        # while not self.audio_output_queue.empty():
-        #     await self.audio_output_queue.get_async()
-        #     # await self.audio_output_queue.get_async(block=False)
         while not self.llm_sentence_queue.empty():
             self.llm_sentence_queue.get_nowait()
         while not self.speech_chunk_queue.empty():

streamlit_av_queue.py

@@ -104,7 +104,12 @@ class StreamlitAVQueue:
                 # print (f"frame: {frame.format.name}, {frame.layout.name}, {frame.sample_rate}, {frame.samples}")
                 assert frame.format.bytes == 2
                 assert frame.format.name == 's16'
+                import time
+                start_time = time.time()
                 frame_as_bytes = await self.app_interface_actor.dequeue_audio_output_frame_async.remote()
+                elapsed_time = time.time() - start_time
+                if elapsed_time > 0.1:
+                    print (f"app_interface_actor.dequeue_audio_output_frame_async() elapsed_time: {elapsed_time}")
                 if frame_as_bytes:
                     # print(f"frame_as_bytes: {len(frame_as_bytes)}")
                     assert len(frame_as_bytes) == frame.samples * frame.format.bytes