remove charles dependancy on streamlit_av_queue

app_interface_actor.py

@@ -20,6 +20,7 @@ class AppInterfaceActor:
             get_if_exists=True,
         ).remote()
 
+# functions for UI to enqueue input, dequeue output
     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()
@@ -32,40 +33,45 @@ class AppInterfaceActor:
             del evicted_item
         await self.audio_input_queue.put_async(shared_buffer_ref)
 
-    async def get_audio_input_frames(self):
+    async def dequeue_audio_output_frame_async(self):
+        if self.audio_output_queue.empty():
+            return None
+        frame = await self.audio_output_queue.get_async()
+        return frame
+
+    async def dequeue_video_output_frames_async(self):
+        video_frames = []
+        if self.video_output_queue.empty():
+            return video_frames
+        while not self.video_output_queue.empty():
+            shared_tensor = await self.video_output_queue.get_async()
+            video_frames.append(shared_tensor)
+        return video_frames
+
+# functions for application to dequeue input, enqueue output    
+    def get_audio_output_queue(self)->Queue:
+        return self.audio_output_queue
+    
+    async def enqueue_video_output_frame(self, shared_tensor_ref):
+        if self.video_output_queue.full():
+            evicted_item = await self.video_output_queue.get_async()
+            del evicted_item
+        await self.video_output_queue.put_async(shared_tensor_ref)
+    
+    async def dequeue_audio_input_frames_async(self):
         audio_frames = []
         if self.audio_input_queue.empty():
             return audio_frames
         while not self.audio_input_queue.empty():
-            shared_tensor_ref = await self.audio_input_queue.get_async()
-            audio_frames.append(shared_tensor_ref)
+            shared_tensor = await self.audio_input_queue.get_async()
+            audio_frames.append(shared_tensor)
         return audio_frames
 
-    async def get_video_input_frames(self):
+    async def dequeue_video_input_frames_async(self):
         video_frames = []
         if self.video_input_queue.empty():
             return video_frames
         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_audio_output_queue(self)->Queue:
-        return self.audio_output_queue
-    
-    def get_video_output_queue(self)->Queue:
-        return self.video_output_queue
-    
-    async def get_audio_output_frame(self):
-        if self.audio_output_queue.empty():
-            return None
-        frame = await self.audio_output_queue.get_async()
-        return frame
-
-    async def get_video_output_frame(self):
-        if self.video_output_queue.empty():
-            return None
-        frame = None
-        while not self.video_output_queue.empty():
-            frame = await self.video_output_queue.get_async()
-        return frame
+            shared_tensor = await self.video_input_queue.get_async()
+            video_frames.append(shared_tensor)
+        return video_frames

charles_actor.py

@@ -29,12 +29,11 @@ class CharlesActor:
     
     async def _initalize_resources(self):
         # Initialize resources
-        print("000 - create StreamlitAVQueue")
-        self._state = "000 - creating StreamlitAVQueue"
-        from streamlit_av_queue import StreamlitAVQueue
-        self._streamlit_av_queue = StreamlitAVQueue()
-        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("000 - create AppInterfaceActor")
+        self._state = "000 - creating AppInterfaceActor"
+        from app_interface_actor import AppInterfaceActor
+        self._app_interface_actor = AppInterfaceActor.get_singleton()
+        self._audio_output_queue = await self._app_interface_actor.get_audio_output_queue.remote()
 
         print("001 - create RespondToPromptActor")
         self._state = "001 - creating RespondToPromptActor"
@@ -114,8 +113,8 @@ 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_audio_input_frames_async()    
-            video_frames = await self._streamlit_av_queue.get_video_frames_async()
+            audio_frames = await self._app_interface_actor.dequeue_audio_input_frames_async.remote()    
+            video_frames = await self._app_interface_actor.dequeue_video_input_frames_async.remote()
 
             if len(audio_frames) > 0:
                 total_audio_frames += len(audio_frames)
@@ -215,11 +214,8 @@ class CharlesActor:
             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._video_output_queue.full():
-                evicted_item = await self._video_output_queue.get_async()
-                del evicted_item
             frame_ref = ray.put(frame)
-            await self._video_output_queue.put_async(frame_ref)                
+            await self._app_interface_actor.enqueue_video_output_frame.remote(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}"

streamlit_av_queue.py

@@ -35,9 +35,9 @@ class StreamlitAVQueue:
         try:
             with self._lock:
                 should_look = self._looking
-            next_video_output_frame = await self.app_interface_actor.get_video_output_frame.remote()
-            if next_video_output_frame is not None:
-                self._video_output_frame = next_video_output_frame
+            video_output_frames = await self.app_interface_actor.dequeue_video_output_frames_async.remote()
+            if len(video_output_frames) > 0:
+                self._video_output_frame = video_output_frames[-1]
             for i, frame in enumerate(frames):
                 user_image = frame.to_ndarray(format="rgb24")
                 if should_look:
@@ -94,7 +94,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.app_interface_actor.get_audio_output_frame.remote()
+                frame_as_bytes = await self.app_interface_actor.dequeue_audio_output_frame_async.remote()
                 if frame_as_bytes:
                     # print(f"frame_as_bytes: {len(frame_as_bytes)}")
                     assert len(frame_as_bytes) == frame.samples * frame.format.bytes
@@ -111,17 +111,3 @@ class StreamlitAVQueue:
         except Exception as e:
             print (e)
         return new_frames
-
-    async def get_audio_input_frames_async(self) -> List[av.AudioFrame]:
-        shared_buffers = await self.app_interface_actor.get_audio_input_frames.remote()
-        return shared_buffers
-
-    async def get_video_frames_async(self) -> List[av.AudioFrame]:
-        shared_tensors = await self.app_interface_actor.get_video_input_frames.remote()
-        return shared_tensors
-    
-    def get_audio_output_queue(self)->Queue:
-        return self.app_interface_actor.get_audio_output_queue.remote()
-    
-    def get_video_output_queue(self)->Queue:
-        return self.app_interface_actor.get_video_output_queue.remote()