fix start up issues; should be more stable

app.py

@@ -1,6 +1,7 @@
 import asyncio
 from collections import deque
 import os
+import sys
 import threading
 import time
 import traceback
@@ -16,6 +17,7 @@ from sample_utils.turn import get_ice_servers
 import json
 from typing import List
 import subprocess
+import ray
 
 st.set_page_config(layout="wide")
 
@@ -27,25 +29,50 @@ load_dotenv()
 
 webrtc_ctx = None
 
-# Initialize Ray
-import ray
+
+@st.cache_resource
 def init_ray():
     try:
         subprocess.check_output(["ray", "start", "--head"])
     except Exception as e:
-        print (e)
-    # 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")
-
-
-if not ray.is_initialized():
-    init_ray()
+        print (f"app.py init_ray: {e}")
+    # Connect to a running Ray cluster
+    while not ray.is_initialized():
+        time.sleep(0.1)
+        ray_address = os.getenv('RAY_ADDRESS')
+        if ray_address:
+            ray.init(ray_address, namespace="project_charles")
+        else:
+            ray.init(namespace="project_charles")
+
+@st.cache_resource
+def get_charles_actor():
+    charles_actor_instance = None
+    charles_actor_proc = subprocess.Popen([sys.executable, "charles_actor.py"])
+    while charles_actor_instance == None:
+        try:
+            charles_actor_instance = ray.get_actor("CharlesActor")
+        except ValueError as e:
+            time.sleep(0.1) # give the subprocess a chance to start
+    return charles_actor_instance
+
+@st.cache_resource
+def get_streamlit_av_queue():
+    from streamlit_av_queue import StreamlitAVQueue
+    streamlit_av_queue_instance = StreamlitAVQueue()
+    return streamlit_av_queue_instance
 
 async def main():
+    # Initialize Ray
+    ray_status = init_ray()
+    while not ray.is_initialized():
+        await asyncio.sleep(0.1)
+    # get ray actors
+    charles_actor = get_charles_actor()
+    await asyncio.sleep(0.1)
+    streamlit_av_queue = get_streamlit_av_queue()
+    await asyncio.sleep(0.1)
+
     st.title("Project Charles")
 
     col1, col2 = st.columns(2)
@@ -58,26 +85,22 @@ async def main():
             looking = st.checkbox("Look", value=False)
         charles_actor_debug_output = st.empty()
         environment_state_ouput = st.empty()
-        with nested_col3:
-            if st.button('Reboot Charles'):
-                st.write('Killing RAY...')
-                subprocess.check_output(["ray", "start", "--head"])
-                st.write('Restarting RAY...')
-                init_ray()
-                charles_actor = None
-                st.write('Reboot Charles')
+        # with nested_col3:
+        #     if st.button('Reboot Charles'):
+        #         st.write('Killing RAY...')
+        #         subprocess.check_output(["ray", "start", "--head"])
+        #         st.write('Restarting RAY...')
+        #         init_ray()
+        #         charles_actor = None
+        #         st.write('Reboot Charles')
 
     with col2:
-        if "streamlit_av_queue" not in st.session_state:
-            from streamlit_av_queue import StreamlitAVQueue
-            st.session_state.streamlit_av_queue = StreamlitAVQueue()
-
         playing = st.checkbox("Playing", value=True)
         webrtc_ctx = webrtc_streamer(
             key="charles",
             desired_playing_state=playing,
-            queued_audio_frames_callback=st.session_state.streamlit_av_queue.queued_audio_frames_callback,
-            queued_video_frames_callback=st.session_state.streamlit_av_queue.queued_video_frames_callback,
+            queued_audio_frames_callback=streamlit_av_queue.queued_audio_frames_callback,
+            queued_video_frames_callback=streamlit_av_queue.queued_video_frames_callback,
             mode=WebRtcMode.SENDRECV,
             media_stream_constraints={
                 "video": True, 
@@ -97,49 +120,30 @@ async def main():
     if not webrtc_ctx.state.playing:
         exit
 
-    from charles_actor import CharlesActor
-    charles_actor = None
-
-
     try:
         while True:
-            if "streamlit_av_queue" in st.session_state:
-                st.session_state.streamlit_av_queue.set_looking_listening(looking, listening)
             if not webrtc_ctx.state.playing:
                 system_one_audio_status.write("Camera has stopped.")
                 await asyncio.sleep(0.1)
                 continue
             if charles_actor is None:
-                try:
-                    charles_actor = ray.get_actor("CharlesActor")
-                    system_one_audio_status.write("Charles is here.")
-                except ValueError as e:
-                    if 'charles_actor_proc' in locals() or 'charles_actor_proc' in globals() and charles_actor_proc is not None:
-                        system_one_audio_status.write("Waiting for instance of Charles...")
-                    else:
-                        await asyncio.sleep(0.5) # give ray a chance to start
-                        charles_actor_proc = subprocess.Popen(["python3", "charles_actor.py"])
-                        system_one_audio_status.write("Creating an instance of Charles...")
-                        await asyncio.sleep(0.5)
-                    # import platform
-                    # print (f"platform:{platform.system()}")
-                    # if platform.system() == 'Linux':
-                    #     charles_actor_proc = subprocess.Popen(["python3", "charles_actor.py"])
-                    #     system_one_audio_status.write("Creating Charles.")
-                    # else:
-                    #     system_one_audio_status.write("Charles is sleeping.")
-                    pass
-            if charles_actor is not None: 
-                try:
-                    # new_environment_state = await charles_actor.get_environment_state.remote()
-                    # environment_state_ouput.markdown(f"{new_environment_state}")
-                    charles_debug_str = await charles_actor.get_charles_actor_debug_output.remote()
-                    charles_actor_debug_output.markdown(charles_debug_str)
-                    state = await charles_actor.get_state.remote()
-                    system_one_audio_status.write(state)
-                except Exception as e:
-                    # assume we disconnected
-                    charles_actor = None
+                system_one_audio_status.write("Looking for Charles actor...")
+                charles_actor = get_charles_actor()
+                if charles_actor is None:
+                    await asyncio.sleep(0.1)
+                    continue
+                system_one_audio_status.write("Found Charles actor.")
+            try:
+                # new_environment_state = await charles_actor.get_environment_state.remote()
+                # environment_state_ouput.markdown(f"{new_environment_state}")
+                streamlit_av_queue.set_looking_listening(looking, listening)
+                charles_debug_str = await charles_actor.get_charles_actor_debug_output.remote()
+                charles_actor_debug_output.markdown(charles_debug_str)
+                state = await charles_actor.get_state.remote()
+                system_one_audio_status.write(state)
+            except Exception as e:
+                # assume we disconnected
+                charles_actor = None
             await asyncio.sleep(0.1)
 
     except Exception as e:
@@ -155,7 +159,5 @@ if __name__ == "__main__":
         if  webrtc_ctx is not None:
             del webrtc_ctx
             webrtc_ctx = None
-        if "streamlit_av_queue" in st.session_state:
-            del st.session_state.streamlit_av_queue    
     finally:
         pass

charles_actor.py

@@ -7,6 +7,7 @@ from clip_transform import CLIPTransform
 from environment_state_actor import EnvironmentStateActor, EnvironmentState
 from agent_state_actor import AgentStateActor
 import asyncio
+import subprocess
 
 @ray.remote
 class CharlesActor:
@@ -186,20 +187,24 @@ class CharlesActor:
             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}. {vector_debug}"
 
-async def main():
-    if not ray.is_initialized():
-        # Try to connect to a running Ray cluster
+def init_ray():
+    try:
+        subprocess.check_output(["ray", "start", "--head"])
+    except Exception as e:
+        print (f"charles_actor.py init_ray: {e}")
+    # Connect to a running Ray cluster
+    while not ray.is_initialized():
+        time.sleep(0.1)
         ray_address = os.getenv('RAY_ADDRESS')
-        import subprocess
-        try:
-            subprocess.check_output(["ray", "start", "--head"])
-        except Exception as e:
-            print (e)
         if ray_address:
             ray.init(ray_address, namespace="project_charles")
         else:
             ray.init(namespace="project_charles")
 
+async def main():
+    if not ray.is_initialized():
+        init_ray()
+
     charles_actor = CharlesActor.options(
         name="CharlesActor", 
         get_if_exists=True,

streamlit_av_queue.py

@@ -15,7 +15,7 @@ class StreamlitAVQueue:
         self._output_channels = 2
         self._audio_bit_rate = audio_bit_rate
         self._listening = True
-        self._looking = True
+        self._looking = False
         self._lock = threading.Lock()
         self.queue_actor = WebRtcAVQueueActor.options(
             name="WebRtcAVQueueActor",