Charles running between two applications

.vscode/launch.json

@@ -1,6 +1,17 @@
 {
-    "version": "0.1.0",
+    // Use IntelliSense to learn about possible attributes.
+    // Hover to view descriptions of existing attributes.
+    // For more information, visit: https://go.microsoft.com/fwlink/?linkid=830387
+    "version": "0.2.0",
     "configurations": [
+        {
+            "name": "Python: Current File",
+            "type": "python",
+            "request": "launch",
+            "program": "${file}",
+            "console": "integratedTerminal"
+            // "justMyCode": true
+        },
         {
             "name": "debug streamlit",
             "type": "python",
@@ -11,6 +22,6 @@
                 // "app.py"
                 "d_app.py"
             ]
-        }
+        }        
     ]
-    }
+}

charles_actor.py

@@ -0,0 +1,120 @@
+import ray
+import time
+import asyncio
+import os
+
+@ray.remote
+class CharlesActor:
+    def __init__(self):
+        self._needs_init = True
+        self._system_one_audio_history_output = ""
+        self._state = "Initializing"
+        
+    def get_state(self):
+        return self._state
+    
+    
+    def get_system_one_audio_history_output(self):
+        return self._system_one_audio_history_output
+    
+    async def _initalize_resources(self):
+        # Initialize resources
+        print("000")
+        from streamlit_av_queue import StreamlitAVQueue
+        self._streamlit_av_queue = StreamlitAVQueue()
+
+        print("002")
+        from speech_to_text_vosk import SpeechToTextVosk
+        self._speech_to_text_vosk = SpeechToTextVosk()
+
+        from chat_pipeline import ChatPipeline
+        self._chat_pipeline = ChatPipeline()
+        await self._chat_pipeline.start()
+
+        self._debug_queue = [
+            # "hello, how are you today?",
+            # "hmm, interesting, tell me more about that.",
+        ]
+        print("010")
+        self._needs_init = True
+        self._state = "Initialized"
+        
+    async def start(self):
+        if self._needs_init:
+            await self._initalize_resources()
+            
+        system_one_audio_history = []
+        
+        self._state = "Waiting for input"
+        total_video_frames = 0
+        total_audio_frames = 0
+        loops = 0
+
+        while True:
+            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()
+            if len(audio_frames) > 0:
+                total_audio_frames += len(audio_frames)
+                # Concatenate all audio frames into a single buffer
+                audio_buffer = b"".join([buffer.tobytes() for buffer in audio_frames])
+                self._speech_to_text_vosk.add_speech_bytes(audio_buffer)
+            prompt, speaker_finished = self._speech_to_text_vosk.get_text()
+            if speaker_finished and len(prompt) > 0:
+                print(f"Prompt: {prompt}")
+                system_one_audio_history.append(prompt)
+                if len(system_one_audio_history) > 10:
+                    system_one_audio_history = system_one_audio_history[-10:]
+                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
+                await self._chat_pipeline.enqueue(prompt)
+            video_frames = self._streamlit_av_queue.get_video_frames()
+            if len(video_frames) > 0:
+                total_video_frames += len(video_frames)
+            #     for video_frame in video_frames:
+            #         system_one_video_output.image(video_frame.to_ndarray())
+            #         pass
+            
+            # update debug output
+            if (total_video_frames >0 or total_audio_frames > 0):
+                self._state = f"Processed {total_video_frames} video frames and {total_audio_frames} audio frames"
+            await asyncio.sleep(0.1)
+            loops+=1
+            self._state = f"Processed {total_video_frames} video frames and {total_audio_frames} audio frames, loops: {loops}"
+
+if __name__ == "__main__":
+    if not ray.is_initialized():
+        # Try to connect to a running Ray cluster
+        ray_address = os.getenv('RAY_ADDRESS')
+        if ray_address:
+            ray.init(ray_address, namespace="project_charles")
+        else:
+            ray.init(namespace="project_charles")
+
+    charles_actor = CharlesActor.options(
+        name="CharlesActor", 
+        get_if_exists=True,
+        ).remote() 
+    future = charles_actor.start.remote()
+
+    try:
+        while True:
+            ready, _ = ray.wait([future], timeout=0)
+            if ready:
+                # The start method has terminated. You can fetch the result (if any) with ray.get().
+                # If the method raised an exception, it will be re-raised here.
+                try:
+                    result = ray.get(future)
+                    print(f"The start method has terminated with result: {result}")
+                except Exception as e:
+                    print(f"The start method raised an exception: {e}")
+                break
+            else:
+                # 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)}")
+    except KeyboardInterrupt:
+        print("Script was manually terminated")

d_app.py

@@ -27,7 +27,12 @@ webrtc_ctx = None
 # Initialize Ray
 import ray
 if not ray.is_initialized():
-    ray.init()
+    # Try to connect to a running Ray cluster
+    ray_address = os.getenv('RAY_ADDRESS')
+    if ray_address:
+        ray.init(ray_address, namespace="project_charles")
+    else:
+        ray.init(namespace="project_charles")
 
 
 
@@ -46,35 +51,12 @@ async def main():
     system_one_audio_history_output = st.empty()
 
     # Initialize resources if not already done
-    print("000")
     system_one_audio_status.write("Initializing streaming")
     if "streamlit_av_queue" not in st.session_state:
-        print("001")
         from streamlit_av_queue import StreamlitAVQueue
         st.session_state.streamlit_av_queue = StreamlitAVQueue()
 
-    if "speech_to_text_vosk" not in st.session_state:
-        print("002")
-        from speech_to_text_vosk import SpeechToTextVosk
-        st.session_state.speech_to_text_vosk = SpeechToTextVosk()
-
-    from chat_pipeline import ChatPipeline
-    if "chat_pipeline" not in st.session_state:
-        print("003")
-        # from chat_pipeline import ChatPipeline
-        # st.session_state.chat_pipeline = ChatPipeline()
-        # await st.session_state.chat_pipeline.start()
-    st.session_state.chat_pipeline = ChatPipeline()
-    await st.session_state.chat_pipeline.start()
-
-    if "debug_queue" not in st.session_state:
-        st.session_state.debug_queue = [
-    #         "hello, how are you today?",
-    #         "hmm, interesting, tell me more about that.",
-        ]
-
     system_one_audio_status.write("resources referecned")
-    print("010")
 
 
 
@@ -93,6 +75,10 @@ async def main():
         exit
 
     system_one_audio_status.write("Initializing speech")
+    
+    from charles_actor import CharlesActor
+    charles_actor = None
+
 
     try:
         while True:
@@ -100,62 +86,24 @@ async def main():
                 system_one_audio_status.write("Stopped.")
                 await asyncio.sleep(0.1)
                 continue
-            system_one_audio_status.write("Streaming.")
-            if len(st.session_state.debug_queue) > 0:
-                prompt = st.session_state.debug_queue.pop(0)
-                await st.session_state.chat_pipeline.enqueue(prompt)
-            audio_frames = st.session_state.streamlit_av_queue.get_audio_frames()
-            if len(audio_frames) > 0:
-                # Concatenate all audio frames into a single buffer
-                audio_buffer = b"".join([buffer.tobytes() for buffer in audio_frames])
-                st.session_state.speech_to_text_vosk.add_speech_bytes(audio_buffer)
-            prompt, speaker_finished = st.session_state.speech_to_text_vosk.get_text()
-            if speaker_finished and len(prompt) > 0:
-                print(f"Prompt: {prompt}")
-                system_one_audio_history.append(prompt)
-                if len(system_one_audio_history) > 10:
-                    system_one_audio_history = system_one_audio_history[-10:]
-                table_content = "| System 1 Audio History |\n| --- |\n"
-                table_content += "\n".join([f"| {item} |" for item in reversed(system_one_audio_history)])
-                system_one_audio_history_output.markdown(table_content)
-                await st.session_state.chat_pipeline.enqueue(prompt)
-            video_frames = st.session_state.streamlit_av_queue.get_video_frames()
-            if len(video_frames) > 0:
-                # for video_frame in video_frames:
-                    # system_one_video_output.image(video_frame.to_ndarray())
+            if charles_actor is None:
+                try:
+                    charles_actor = ray.get_actor("CharlesActor")
+                    system_one_audio_status.write("Charles is here.")
+                except ValueError as e:
+                    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)
             await asyncio.sleep(0.1)
 
-    # try:
-    #     prompts = [
-    #         "hello, how are you today?",
-    #         "tell me about your shadow self?",
-    #         "hmm, interesting, tell me more about that.",
-    #         "wait, that is so interesting, what else?",
-    #     ]
-    #     for prompt in prompts:
-    #         system_one_audio_history.append(prompt)
-    #         if len(system_one_audio_history) > 10:
-    #             system_one_audio_history = system_one_audio_history[-10:]
-    #         table_content = "| System 1 Audio History |\n| --- |\n"
-    #         table_content += "\n".join([f"| {item} |" for item in reversed(system_one_audio_history)])
-    #         system_one_audio_history_output.markdown(table_content)
-    #         await chat_pipeline.enqueue(prompt)
-
     except Exception as e:
         print(f"An error occurred: {e}")
         traceback.print_exc()
         raise e
 
 
-    # while True:
-    #     if webrtc_ctx.state.playing:
-    #         system_one_audio_status.write("Streaming.")
-    #     else:
-    #         system_one_audio_status.write("Stopped.")
-    #     await asyncio.sleep(0.5)
-
-
 if __name__ == "__main__":
     try:
         asyncio.run(main())
@@ -164,12 +112,6 @@ if __name__ == "__main__":
             del webrtc_ctx
             webrtc_ctx = None
         if "streamlit_av_queue" in st.session_state:
-            del st.session_state.streamlit_av_queue
-
-        if "speech_to_text_vosk" in st.session_state:
-            del st.session_state.speech_to_text_vosk
-
-        if "chat_pipeline" in st.session_state:
-            del st.session_state.chat_pipeline        
+            del st.session_state.streamlit_av_queue    
     finally:
         pass

input_av_queue_actor.py

@@ -12,9 +12,15 @@ class InputAVQueueActor:
         self.video_queue = Queue(maxsize=100)  # Adjust the size as needed
 
     def enqueue_video_frame(self, shared_tensor_ref):
+        if self.video_queue.full():
+            evicted_item = self.video_queue.get()
+            del evicted_item
         self.video_queue.put(shared_tensor_ref)
 
     def enqueue_audio_frame(self, shared_buffer_ref):
+        if self.audio_queue.full():
+            evicted_item = self.audio_queue.get()
+            del evicted_item
         self.audio_queue.put(shared_buffer_ref)
 
 

streamlit_av_queue.py

@@ -13,11 +13,11 @@ import torch
 class StreamlitAVQueue:
     def __init__(self, audio_bit_rate=16000):
         self._audio_bit_rate = audio_bit_rate
-        try:
-            self.queue_actor = ray.get_actor("InputAVQueueActor")
-        except ValueError as e:
-            self.queue_actor = InputAVQueueActor.options(name="InputAVQueueActor").remote() 
-
+        self.queue_actor = InputAVQueueActor.options(
+            name="InputAVQueueActor", 
+            get_if_exists=True,
+            ).remote() 
+            
     async def queued_video_frames_callback(
                 self,
                 frames: List[av.AudioFrame],