refactor: in/out_audio/video to audio/video_input/output

charles_actor.py

@@ -33,15 +33,15 @@ class CharlesActor:
         self._state = "000 - creating StreamlitAVQueue"
         from streamlit_av_queue import StreamlitAVQueue
         self._streamlit_av_queue = StreamlitAVQueue()
-        self._out_audio_queue = await self._streamlit_av_queue.get_out_audio_queue()
-        self._out_video_queue = await self._streamlit_av_queue.get_out_video_queue()
+        self._audio_output_queue = await self._streamlit_av_queue.get_audio_output_queue()
+        self._video_output_queue = await self._streamlit_av_queue.get_video_output_queue()
 
         print("001 - create RespondToPromptActor")
         self._state = "001 - creating RespondToPromptActor"
         from respond_to_prompt_actor import RespondToPromptActor
         self._environment_state_actor = EnvironmentStateActor.remote()
         self._agent_state_actor = AgentStateActor.remote()
-        self._respond_to_prompt_actor = RespondToPromptActor.remote(self._environment_state_actor, self._out_audio_queue)
+        self._respond_to_prompt_actor = RespondToPromptActor.remote(self._environment_state_actor, self._audio_output_queue)
 
         print("002 - create SpeechToTextVoskActor")
         self._state = "002 - creating SpeechToTextVoskActor"
@@ -114,7 +114,7 @@ class CharlesActor:
             env_state = await self._environment_state_actor.begin_next_step.remote()
             self._environment_state = env_state
             self._agent_state_actor.begin_step.remote()
-            audio_frames = await self._streamlit_av_queue.get_in_audio_frames_async()    
+            audio_frames = await self._streamlit_av_queue.get_audio_input_frames_async()    
             video_frames = await self._streamlit_av_queue.get_video_frames_async()
 
             if len(audio_frames) > 0:
@@ -211,15 +211,15 @@ class CharlesActor:
             await asyncio.sleep(0.01)
 
             # add observations to the environment state
-            count = len(self._out_audio_queue)
+            count = len(self._audio_output_queue)
             is_talking = bool(count > 0)
             has_spoken_for_this_prompt = has_spoken_for_this_prompt or is_talking
             frame = self._animator.update(is_talking)
-            if self._out_video_queue.full():
-                evicted_item = await self._out_video_queue.get_async()
+            if self._video_output_queue.full():
+                evicted_item = await self._video_output_queue.get_async()
                 del evicted_item
             frame_ref = ray.put(frame)
-            await self._out_video_queue.put_async(frame_ref)                
+            await self._video_output_queue.put_async(frame_ref)                
 
             loops+=1
             self._state = f"Processed {total_video_frames} video frames and {total_audio_frames} audio frames, loops: {loops}. loops per second: {loops/(time.time()-start_time):.2f}. Is speaking: {is_talking}({count}). {vector_debug}"

respond_to_prompt_actor.py

@@ -144,14 +144,14 @@ class RespondToPromptActor:
     def __init__(
             self, 
             environment_state_actor:EnvironmentStateActor, 
-            out_audio_queue):
+            audio_output_queue):
         voice_id="2OviOUQc1JsQRQgNkVBj"
         self.prompt_queue = Queue(maxsize=100)
         self.llm_sentence_queue = Queue(maxsize=100)
         self.speech_chunk_queue = Queue(maxsize=100)
         self.environment_state_actor = environment_state_actor
 
-        self.ffmpeg_converter_actor = FFMpegConverterActor.remote(out_audio_queue)
+        self.ffmpeg_converter_actor = FFMpegConverterActor.remote(audio_output_queue)
         
         self.prompt_to_llm = PromptToLLMActor.remote(
             self.environment_state_actor,

streamlit_av_queue.py

@@ -23,7 +23,7 @@ class StreamlitAVQueue:
             name="WebRtcAVQueueActor", 
             get_if_exists=True,
             ).remote()
-        self._out_video_frame = None 
+        self._video_output_frame = None 
         
     def set_looking_listening(self, looking, listening: bool):
         with self._lock:
@@ -38,16 +38,16 @@ class StreamlitAVQueue:
         try:
             with self._lock:
                 should_look = self._looking
-            next_out_video_frame = await self.queue_actor.get_out_video_frame.remote()
-            if next_out_video_frame is not None:
-                self._out_video_frame = next_out_video_frame
+            next_video_output_frame = await self.queue_actor.get_video_output_frame.remote()
+            if next_video_output_frame is not None:
+                self._video_output_frame = next_video_output_frame
             for i, frame in enumerate(frames):
                 user_image = frame.to_ndarray(format="rgb24")
                 if should_look:
                     shared_tensor_ref = ray.put(user_image)
-                    await self.queue_actor.enqueue_in_video_frame.remote(shared_tensor_ref)
-                if self._out_video_frame is not None:
-                    frame = self._out_video_frame
+                    await self.queue_actor.enqueue_video_input_frame.remote(shared_tensor_ref)
+                if self._video_output_frame is not None:
+                    frame = self._video_output_frame
                     # resize user image to 1/4 size
                     user_frame = cv2.resize(user_image, (user_image.shape[1]//4, user_image.shape[0]//4), interpolation=cv2.INTER_AREA)
                     # flip horizontally
@@ -85,7 +85,7 @@ class StreamlitAVQueue:
                     sound_chunk += sound
                 shared_buffer = np.array(sound_chunk.get_array_of_samples())
                 shared_buffer_ref = ray.put(shared_buffer)
-                await self.queue_actor.enqueue_in_audio_frame.remote(shared_buffer_ref)
+                await self.queue_actor.enqueue_audio_input_frame.remote(shared_buffer_ref)
         except Exception as e:
             print (e)
             
@@ -97,7 +97,7 @@ 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'
-                frame_as_bytes = await self.queue_actor.get_out_audio_frame.remote()
+                frame_as_bytes = await self.queue_actor.get_audio_output_frame.remote()
                 if frame_as_bytes:
                     # print(f"frame_as_bytes: {len(frame_as_bytes)}")
                     assert len(frame_as_bytes) == frame.samples * frame.format.bytes
@@ -115,16 +115,16 @@ class StreamlitAVQueue:
             print (e)
         return new_frames
 
-    async def get_in_audio_frames_async(self) -> List[av.AudioFrame]:
-        shared_buffers = await self.queue_actor.get_in_audio_frames.remote()
+    async def get_audio_input_frames_async(self) -> List[av.AudioFrame]:
+        shared_buffers = await self.queue_actor.get_audio_input_frames.remote()
         return shared_buffers
 
     async def get_video_frames_async(self) -> List[av.AudioFrame]:
-        shared_tensors = await self.queue_actor.get_in_video_frames.remote()
+        shared_tensors = await self.queue_actor.get_video_input_frames.remote()
         return shared_tensors
     
-    def get_out_audio_queue(self)->Queue:
-        return self.queue_actor.get_out_audio_queue.remote()
+    def get_audio_output_queue(self)->Queue:
+        return self.queue_actor.get_audio_output_queue.remote()
     
-    def get_out_video_queue(self)->Queue:
-        return self.queue_actor.get_out_video_queue.remote()
+    def get_video_output_queue(self)->Queue:
+        return self.queue_actor.get_video_output_queue.remote()

webrtc_av_queue_actor.py

@@ -8,58 +8,58 @@ import numpy as np
 @ray.remote
 class WebRtcAVQueueActor:
     def __init__(self):
-        self.in_audio_queue = Queue(maxsize=3000)  # Adjust the size as needed
-        self.in_video_queue = Queue(maxsize=10)  # Adjust the size as needed
-        self.out_audio_queue = Queue(maxsize=3000)  # Adjust the size as needed
-        self.out_video_queue = Queue(maxsize=10)  # Adjust the size as needed
+        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=3000)  # Adjust the size as needed
+        self.video_output_queue = Queue(maxsize=10)  # Adjust the size as needed
 
 
-    async def enqueue_in_video_frame(self, shared_tensor_ref):
-        if self.in_video_queue.full():
-            evicted_item = await self.in_video_queue.get_async()
+    async def enqueue_video_input_frame(self, shared_tensor_ref):
+        if self.video_input_queue.full():
+            evicted_item = await self.video_input_queue.get_async()
             del evicted_item
-        await self.in_video_queue.put_async(shared_tensor_ref)
+        await self.video_input_queue.put_async(shared_tensor_ref)
 
-    async def enqueue_in_audio_frame(self, shared_buffer_ref):
-        if self.in_audio_queue.full():
-            evicted_item = await self.in_audio_queue.get_async()
+    async def enqueue_audio_input_frame(self, shared_buffer_ref):
+        if self.audio_input_queue.full():
+            evicted_item = await self.audio_input_queue.get_async()
             del evicted_item
-        await self.in_audio_queue.put_async(shared_buffer_ref)
+        await self.audio_input_queue.put_async(shared_buffer_ref)
 
-    async def get_in_audio_frames(self):
+    async def get_audio_input_frames(self):
         audio_frames = []
-        if self.in_audio_queue.empty():
+        if self.audio_input_queue.empty():
             return audio_frames
-        while not self.in_audio_queue.empty():
-            shared_tensor_ref = await self.in_audio_queue.get_async()
+        while not self.audio_input_queue.empty():
+            shared_tensor_ref = await self.audio_input_queue.get_async()
             audio_frames.append(shared_tensor_ref)
         return audio_frames
 
-    async def get_in_video_frames(self):
+    async def get_video_input_frames(self):
         video_frames = []
-        if self.in_video_queue.empty():
+        if self.video_input_queue.empty():
             return video_frames
-        while not self.in_video_queue.empty():
-            shared_tensor_ref = await self.in_video_queue.get_async()
+        while not self.video_input_queue.empty():
+            shared_tensor_ref = await self.video_input_queue.get_async()
             video_frames.append(shared_tensor_ref)
         return video_frames
     
-    def get_out_audio_queue(self)->Queue:
-        return self.out_audio_queue
+    def get_audio_output_queue(self)->Queue:
+        return self.audio_output_queue
     
-    def get_out_video_queue(self)->Queue:
-        return self.out_video_queue
+    def get_video_output_queue(self)->Queue:
+        return self.video_output_queue
     
-    async def get_out_audio_frame(self):
-        if self.out_audio_queue.empty():
+    async def get_audio_output_frame(self):
+        if self.audio_output_queue.empty():
             return None
-        frame = await self.out_audio_queue.get_async()
+        frame = await self.audio_output_queue.get_async()
         return frame
 
-    async def get_out_video_frame(self):
-        if self.out_video_queue.empty():
+    async def get_video_output_frame(self):
+        if self.video_output_queue.empty():
             return None
         frame = None
-        while not self.out_video_queue.empty():
-            frame = await self.out_video_queue.get_async()
+        while not self.video_output_queue.empty():
+            frame = await self.video_output_queue.get_async()
         return frame