fix async issues and flush ffmpeg actor

charles_actor.py

@@ -62,7 +62,7 @@ class CharlesActor:
         while True:
             if len(self._debug_queue) > 0:
                 prompt = self._debug_queue.pop(0)
-                self._respond_to_prompt_actor.enqueue_prompt.remote(prompt)
+                await self._respond_to_prompt_actor.enqueue_prompt.remote(prompt)
             audio_frames = await self._streamlit_av_queue.get_in_audio_frames_async()    
             if len(audio_frames) > 0:
                 total_audio_frames += len(audio_frames)
@@ -75,7 +75,7 @@ class CharlesActor:
             if len(process_speech_to_text_future) > 0:
                 ready, _ = ray.wait([process_speech_to_text_future[0]], timeout=0)
                 if ready:
-                    prompt, speaker_finished = ray.get(process_speech_to_text_future[0])
+                    prompt, speaker_finished = await process_speech_to_text_future[0]
                     del process_speech_to_text_future[0]
 
                     if speaker_finished and len(prompt) > 0:
@@ -86,7 +86,7 @@ class CharlesActor:
                         table_content = "| System 1 Audio History |\n| --- |\n"
                         table_content += "\n".join([f"| {item} |" for item in reversed(system_one_audio_history)])
                         self._system_one_audio_history_output = table_content
-                        self._respond_to_prompt_actor.enqueue_prompt.remote(prompt)
+                        await self._respond_to_prompt_actor.enqueue_prompt.remote(prompt)
                         
             # video_frames = await self._streamlit_av_queue.get_video_frames_async()
             # if len(video_frames) > 0:

d_app.py

@@ -104,8 +104,12 @@ async def main():
                     system_one_audio_status.write("Charles is sleeping.")
                     pass
             if charles_actor is not None: 
-                audio_history = ray.get(charles_actor.get_system_one_audio_history_output.remote())
-                system_one_audio_history_output.markdown(audio_history)
+                try:
+                    audio_history = await charles_actor.get_system_one_audio_history_output.remote()
+                    system_one_audio_history_output.markdown(audio_history)
+                except Exception as e:
+                    # assume we disconnected
+                    charles_actor = None
             await asyncio.sleep(0.1)
 
     except Exception as e:

ffmpeg_converter_actor.py

@@ -57,9 +57,9 @@ class FFMpegConverterActor:
     # def has_processed_all_data(self):
     #     return self.process.poll() is not None
 
-    def flush_output_queue(self):
+    async def flush_output_queue(self):
         while not self.output_queue.empty():
-            self.output_queue.get()   
+            await self.output_queue.get_async()   
 
     def close(self):
         self.input_pipe.close()

respond_to_prompt_actor.py

@@ -26,13 +26,13 @@ class PromptToLLMActor:
                 print(f"{sentence}")
                 await self.output_queue.put_async(sentence)
                 
-    def cancel(self):
+    async def cancel(self):
         if self.cancel_event:
             self.cancel_event.set()
         while not self.input_queue.empty():
-            self.input_queue.get()
+            await self.input_queue.get_async()
         while not self.output_queue.empty():
-            self.output_queue.get()
+            await self.output_queue.get_async()
 
 @ray.remote
 class LLMSentanceToSpeechActor:
@@ -51,13 +51,13 @@ class LLMSentanceToSpeechActor:
             async for chunk in self.chat_service.get_speech_chunks_async(sentance, self.cancel_event):
                 await self.output_queue.put_async(chunk)
 
-    def cancel(self):
+    async def cancel(self):
         if self.cancel_event:
             self.cancel_event.set()
         while not self.input_queue.empty():
-            self.input_queue.get()
+            await self.input_queue.get_async()
         while not self.output_queue.empty():
-            self.output_queue.get()
+            await self.output_queue.get_async()
 
 
 @ray.remote
@@ -74,9 +74,9 @@ class SpeechToSpeakerActor:
             # print (f"Got audio chunk {len(audio_chunk)}")
             self.chat_service.enqueue_speech_bytes_to_play([audio_chunk])
             
-    def cancel(self):
+    async def cancel(self):
         while not self.input_queue.empty():
-            self.input_queue.get()            
+            await self.input_queue.get_async()            
 
 @ray.remote
 class SpeechToConverterActor:
@@ -115,18 +115,18 @@ class RespondToPromptActor:
         self.llm_sentence_to_speech.run.remote()
         self.speech_output.run.remote()
             
-    def enqueue_prompt(self, prompt):
+    async def enqueue_prompt(self, prompt):
         print("flush anything queued")
         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([
+        await asyncio.gather(
             prompt_to_llm_future,
             llm_sentence_to_speech_future,
             speech_to_speaker_future,
             ffmpeg_converter_future,
-        ])
+        )
         self.prompt_queue.put(prompt)
         print("Enqueued prompt")