add flush_output_queue

ffmpeg_converter_actor.py

@@ -5,8 +5,8 @@ from ray.util.queue import Queue
 
 @ray.remote
 class FFMpegConverterActor:
-    def __init__(self, queue: Queue, buffer_size: int = 1920, output_format: str='s16le'):
-        self.queue = queue
+    def __init__(self, output_queue: Queue, buffer_size: int = 1920, output_format: str='s16le'):
+        self.output_queue = output_queue
         self.buffer_size = buffer_size
         self.output_format = output_format
 
@@ -19,7 +19,7 @@ class FFMpegConverterActor:
         while True:
             chunk = await self.output_pipe.readexactly(self.buffer_size)
             # print(f"FFMpegConverterActor: read {len(chunk)} bytes")
-            await self.queue.put_async(chunk)       
+            await self.output_queue.put_async(chunk)       
 
     async def start_process(self):
         cmd = [
@@ -57,6 +57,10 @@ class FFMpegConverterActor:
     # def has_processed_all_data(self):
     #     return self.process.poll() is not None
 
+    def flush_output_queue(self):
+        while not self.output_queue.empty():
+            self.output_queue.get()   
+
     def close(self):
         self.input_pipe.close()
         self.output_pipe.close()

respond_to_prompt_actor.py

@@ -103,6 +103,7 @@ class RespondToPromptActor:
         self.prompt_queue = Queue(maxsize=100)
         self.llm_sentence_queue = Queue(maxsize=100)
         self.speech_chunk_queue = Queue(maxsize=100)
+        self.ffmepg_converter_actor = ffmpeg_converter_actor
         
         self.prompt_to_llm = PromptToLLMActor.remote(self.prompt_queue, self.llm_sentence_queue, voice_id)
         self.llm_sentence_to_speech = LLMSentanceToSpeechActor.remote(self.llm_sentence_queue, self.speech_chunk_queue, voice_id)
@@ -119,10 +120,12 @@ class RespondToPromptActor:
         prompt_to_llm_future = self.prompt_to_llm.cancel.remote()
         llm_sentence_to_speech_future = self.llm_sentence_to_speech.cancel.remote()
         speech_to_speaker_future = self.speech_output.cancel.remote()
+        ffmpeg_converter_future = self.ffmepg_converter_actor.flush_output_queue.remote()
         ray.get([
             prompt_to_llm_future,
             llm_sentence_to_speech_future,
             speech_to_speaker_future,
+            ffmpeg_converter_future,
         ])
         self.prompt_queue.put(prompt)
         print("Enqueued prompt")