fix use of ray.get inside aync

charles_actor.py

@@ -54,7 +54,7 @@ class CharlesActor:
             if len(self._debug_queue) > 0:
                 prompt = self._debug_queue.pop(0)
                 await self._chat_pipeline.enqueue(prompt)
-            audio_frames = self._streamlit_av_queue.get_audio_frames()
+            audio_frames = await self._streamlit_av_queue.get_audio_frames_async()
             if len(audio_frames) > 0:
                 total_audio_frames += len(audio_frames)
                 # Concatenate all audio frames into a single buffer
@@ -70,7 +70,7 @@ class CharlesActor:
                 table_content += "\n".join([f"| {item} |" for item in reversed(system_one_audio_history)])
                 self._system_one_audio_history_output = table_content
                 await self._chat_pipeline.enqueue(prompt)
-            video_frames = self._streamlit_av_queue.get_video_frames()
+            video_frames = await self._streamlit_av_queue.get_video_frames_async()
             if len(video_frames) > 0:
                 total_video_frames += len(video_frames)
             #     for video_frame in video_frames:
@@ -115,6 +115,6 @@ if __name__ == "__main__":
                 # The start method is still running. You can poll for debug information here.
                 time.sleep(1)
                 state = charles_actor.get_state.remote()
-                print(f"Charles is in state: {ray.get(state)}")
+                # print(f"Charles is in state: {ray.get(state)}")
     except KeyboardInterrupt:
         print("Script was manually terminated")

streamlit_av_queue.py

@@ -65,11 +65,11 @@ class StreamlitAVQueue:
             new_frame.sample_rate = frame.sample_rate
             new_frames.append(new_frame)    
         return new_frames
-    
-    def get_audio_frames(self) -> List[av.AudioFrame]:
-        shared_buffers = ray.get(self.queue_actor.get_audio_frames.remote())
+
+    async def get_audio_frames_async(self) -> List[av.AudioFrame]:
+        shared_buffers = await self.queue_actor.get_audio_frames.remote()
         return shared_buffers
-    
-    def get_video_frames(self) -> List[av.AudioFrame]:
-        shared_tensors = ray.get(self.queue_actor.get_video_frames.remote())
-        return shared_tensors
+
+    async def get_video_frames_async(self) -> List[av.AudioFrame]:
+        shared_tensors = await self.queue_actor.get_video_frames.remote()
+        return shared_tensors