add missing async / await

respond_to_prompt_actor.py

@@ -9,7 +9,7 @@ import asyncio
 
 @ray.remote
 class PromptToLLMActor:
-    def __init__(self, input_queue, output_queue):
+    def __init__(self, input_queue:Queue, output_queue:Queue):
         load_dotenv()
         self.input_queue = input_queue
         self.output_queue = output_queue
@@ -78,7 +78,7 @@ class SpeechToSpeakerActor:
 
 @ray.remote
 class SpeechToConverterActor:
-    def __init__(self, input_queue, ffmpeg_converter_actor):
+    def __init__(self, input_queue:Queue, ffmpeg_converter_actor:Queue):
         load_dotenv()
         self.input_queue = input_queue
         self.ffmpeg_converter_actor = ffmpeg_converter_actor
@@ -89,9 +89,9 @@ class SpeechToConverterActor:
             # print (f"Got audio chunk {len(audio_chunk)}")
             await self.ffmpeg_converter_actor.push_chunk.remote(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
@@ -117,14 +117,14 @@ class RespondToPromptActor:
         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()
+        speech_output_future = self.speech_output.cancel.remote()
         ffmpeg_converter_future = self.ffmepg_converter_actor.flush_output_queue.remote()
         await asyncio.gather(
             prompt_to_llm_future,
             llm_sentence_to_speech_future,
-            speech_to_speaker_future,
+            speech_output_future,
             ffmpeg_converter_future,
         )
-        self.prompt_queue.put(prompt)
+        await self.prompt_queue.put_async(prompt)
         print("Enqueued prompt")
         

streamlit_av_queue.py

@@ -27,7 +27,7 @@ class StreamlitAVQueue:
             for frame in frames:
                 shared_tensor = torch.from_numpy(frame.to_ndarray())
                 shared_tensor_ref = ray.put(shared_tensor)
-                self.queue_actor.enqueue_in_video_frame.remote(shared_tensor_ref)
+                await self.queue_actor.enqueue_in_video_frame.remote(shared_tensor_ref)
         except Exception as e:
             print (e)
         return frames
@@ -51,7 +51,7 @@ class StreamlitAVQueue:
                     sound_chunk += sound
                 shared_buffer = np.array(sound_chunk.get_array_of_samples())
                 shared_buffer_ref = ray.put(shared_buffer)
-                self.queue_actor.enqueue_in_audio_frame.remote(shared_buffer_ref)
+                await self.queue_actor.enqueue_in_audio_frame.remote(shared_buffer_ref)
         except Exception as e:
             print (e)
             

webrtc_av_queue_actor.py

@@ -13,34 +13,34 @@ class WebRtcAVQueueActor:
         self.out_audio_queue = Queue(maxsize=100)  # Adjust the size as needed
 
 
-    def enqueue_in_video_frame(self, shared_tensor_ref):
+    async def enqueue_in_video_frame(self, shared_tensor_ref):
         if self.in_video_queue.full():
-            evicted_item = self.in_video_queue.get()
+            evicted_item = await self.in_video_queue.get_async()
             del evicted_item
-        self.in_video_queue.put(shared_tensor_ref)
+        await self.in_video_queue.put_async(shared_tensor_ref)
 
-    def enqueue_in_audio_frame(self, shared_buffer_ref):
+    async def enqueue_in_audio_frame(self, shared_buffer_ref):
         if self.in_audio_queue.full():
-            evicted_item = self.in_audio_queue.get()
+            evicted_item = await self.in_audio_queue.get_async()
             del evicted_item
-        self.in_audio_queue.put(shared_buffer_ref)
+        await self.in_audio_queue.put_async(shared_buffer_ref)
 
 
-    def get_in_audio_frames(self):
+    async def get_in_audio_frames(self):
         audio_frames = []
         if self.in_audio_queue.empty():
             return audio_frames
         while not self.in_audio_queue.empty():
-            shared_tensor_ref = self.in_audio_queue.get()
+            shared_tensor_ref = await self.in_audio_queue.get_async()
             audio_frames.append(shared_tensor_ref)
         return audio_frames
 
-    def get_in_video_frames(self):
+    async def get_in_video_frames(self):
         video_frames = []
         if self.in_video_queue.empty():
             return video_frames
         while not self.in_video_queue.empty():
-            shared_tensor_ref = self.in_video_queue.get()
+            shared_tensor_ref = await self.in_video_queue.get_async()
             video_frames.append(shared_tensor_ref)
         return video_frames