Update src/streamlit_app.py

src/streamlit_app.py

@@ -6,7 +6,7 @@ import base64
 import io
 import threading
 import traceback
-import atexit
+import atexit # Correctly imported
 import time
 import logging
 from dotenv import load_dotenv
@@ -25,7 +25,7 @@ from streamlit_webrtc import (
     AudioProcessorBase,
     VideoProcessorBase,
 )
-# from aiortc import RTCIceServer, RTCConfiguration # RTCConfiguration object not needed directly
+# from aiortc import RTCIceServer, RTCConfiguration # Not needed directly
 
 # --- Configuration ---
 load_dotenv()
@@ -45,6 +45,7 @@ MEDIA_TO_GEMINI_QUEUE_MAXSIZE = 30
 VIDEO_FPS_TO_GEMINI = 2
 VIDEO_API_RESIZE = (1024, 1024)
 
+# !!! IMPORTANT: Verify this model name is correct for the Live API !!!
 MODEL_NAME = "models/gemini-2.0-flash-live-001"
 
 MEDICAL_ASSISTANT_SYSTEM_PROMPT = """You are an AI Medical Assistant. Your primary function is to analyze visual information from the user's camera or screen and respond via voice.
@@ -91,12 +92,16 @@ else:
     logging.critical("GEMINI_API_KEY not found.")
     st.stop()
 
+# Gemini LiveConnectConfig - Simplified for debugging ConnectionClosedError
+# Removed speech_config temporarily. If connection works now, the issue was voice config.
+# If it still fails, check MODEL_NAME and API Key permissions.
 LIVE_CONNECT_CONFIG = types.LiveConnectConfig(
     response_modalities=["audio", "text"],
-    speech_config=types.SpeechConfig(
-        voice_config=types.VoiceConfig(prebuilt_voice_config=types.PrebuiltVoiceConfig(voice_name="Zephyr"))
-    ),
+    # speech_config=types.SpeechConfig( # Temporarily commented out for debugging
+    #     voice_config=types.VoiceConfig(prebuilt_voice_config=types.PrebuiltVoiceConfig(voice_name="Zephyr"))
+    # ),
 )
+logging.info(f"Using LiveConnectConfig: {LIVE_CONNECT_CONFIG}") # Log the config being used
 
 # --- Backend Gemini Interaction Loop ---
 class GeminiInteractionLoop:
@@ -105,8 +110,6 @@ class GeminiInteractionLoop:
         self.async_event_loop = None
         self.is_running = True
         self.playback_stream = None
-        # Queues will be initialized in run_main_loop and assigned to global vars
-        # This class will use the global queue variables directly
 
     async def send_text_input_to_gemini(self, user_text):
         if not user_text or not self.gemini_session or not self.is_running:
@@ -121,33 +124,24 @@ class GeminiInteractionLoop:
     async def stream_media_to_gemini(self):
         logging.info("Task started: Stream media from WebRTC queues to Gemini.")
         async def get_media_from_queues():
-            # Ensure queues are initialized before trying to get from them
             if video_frames_to_gemini_q is None or audio_chunks_to_gemini_q is None:
-                await asyncio.sleep(0.1) # Wait for queues to be initialized
-                return None
+                await asyncio.sleep(0.1); return None
             try:
                 video_frame = await asyncio.wait_for(video_frames_to_gemini_q.get(), timeout=0.02)
-                video_frames_to_gemini_q.task_done()
-                return video_frame
+                video_frames_to_gemini_q.task_done(); return video_frame
             except asyncio.TimeoutError: pass
             except Exception as e: logging.error(f"Error getting video from queue: {e}", exc_info=True)
-
             try:
                 audio_chunk = await asyncio.wait_for(audio_chunks_to_gemini_q.get(), timeout=0.02)
-                audio_chunks_to_gemini_q.task_done()
-                return audio_chunk
+                audio_chunks_to_gemini_q.task_done(); return audio_chunk
             except asyncio.TimeoutError: return None
-            except Exception as e:
-                logging.error(f"Error getting audio from queue: {e}", exc_info=True)
-                return None
+            except Exception as e: logging.error(f"Error getting audio from queue: {e}", exc_info=True); return None
         try:
             while self.is_running:
-                if not self.gemini_session:
-                    await asyncio.sleep(0.1); continue
+                if not self.gemini_session: await asyncio.sleep(0.1); continue
                 media_data = await get_media_from_queues()
                 if media_data and self.gemini_session and self.is_running:
-                    try:
-                        await self.gemini_session.send(input=media_data)
+                    try: await self.gemini_session.send(input=media_data)
                     except Exception as e: logging.error(f"Error sending media chunk to Gemini: {e}", exc_info=True)
                 elif not media_data: await asyncio.sleep(0.05)
         except asyncio.CancelledError: logging.info("Task cancelled: stream_media_to_gemini.")
@@ -157,21 +151,19 @@ class GeminiInteractionLoop:
         logging.info("Task started: Process responses from Gemini.")
         try:
             while self.is_running:
-                if not self.gemini_session:
-                    await asyncio.sleep(0.1); continue
-                if audio_from_gemini_playback_q is None: # Wait for queue init
-                    await asyncio.sleep(0.1); continue
+                if not self.gemini_session: await asyncio.sleep(0.1); continue
+                if audio_from_gemini_playback_q is None: await asyncio.sleep(0.1); continue
                 try:
                     turn_response = self.gemini_session.receive()
                     async for chunk in turn_response:
                         if not self.is_running: break
                         if audio_data := chunk.data:
-                            if not audio_from_gemini_playback_q.full():
-                                audio_from_gemini_playback_q.put_nowait(audio_data)
+                            if not audio_from_gemini_playback_q.full(): audio_from_gemini_playback_q.put_nowait(audio_data)
                             else: logging.warning("Audio playback queue full, discarding Gemini audio data.")
                         if text_response := chunk.text:
                             logging.info(f"Gemini text response: {text_response[:100]}")
                             if 'chat_messages' not in st.session_state: st.session_state.chat_messages = []
+                            # This direct update might cause issues if multiple updates happen before rerun
                             st.session_state.chat_messages = st.session_state.chat_messages + [{"role": "assistant", "content": text_response}]
                 except types.generation_types.StopCandidateException: logging.info("Gemini response stream ended normally.")
                 except Exception as e:
@@ -182,10 +174,10 @@ class GeminiInteractionLoop:
 
     async def play_gemini_audio(self):
         logging.info("Task started: Play Gemini audio responses.")
+        # Note: This task might fail in environments without proper ALSA setup (like HF Spaces default)
+        # Error handling below will log the error but allow other tasks to continue.
         try:
-            # Wait for the playback queue to be initialized
-            while audio_from_gemini_playback_q is None and self.is_running:
-                await asyncio.sleep(0.1)
+            while audio_from_gemini_playback_q is None and self.is_running: await asyncio.sleep(0.1)
             if not self.is_running: return
 
             self.playback_stream = await asyncio.to_thread(
@@ -195,49 +187,48 @@ class GeminiInteractionLoop:
             while self.is_running:
                 try:
                     audio_chunk = await asyncio.wait_for(audio_from_gemini_playback_q.get(), timeout=1.0)
-                    if audio_chunk:
-                        await asyncio.to_thread(self.playback_stream.write, audio_chunk)
+                    if audio_chunk: await asyncio.to_thread(self.playback_stream.write, audio_chunk)
                     if audio_chunk: audio_from_gemini_playback_q.task_done()
                 except asyncio.TimeoutError: continue
-                except Exception as e: logging.error(f"Error playing audio: {e}", exc_info=True); await asyncio.sleep(0.01)
+                except Exception as e: logging.error(f"Error playing audio chunk: {e}", exc_info=True); await asyncio.sleep(0.01)
         except Exception as e:
-            logging.error(f"Failed to open PyAudio playback stream: {e}", exc_info=True)
-            self.is_running = False
+            logging.error(f"Failed to open or use PyAudio playback stream: {e}", exc_info=True)
+            # Don't set self.is_running = False here, just log the playback failure
         finally:
             if self.playback_stream:
                 logging.info("Stopping and closing PyAudio playback stream.")
-                await asyncio.to_thread(self.playback_stream.stop_stream)
-                await asyncio.to_thread(self.playback_stream.close)
+                try:
+                    await asyncio.to_thread(self.playback_stream.stop_stream)
+                    await asyncio.to_thread(self.playback_stream.close)
+                except Exception as e_close:
+                     logging.error(f"Error closing playback stream: {e_close}", exc_info=True)
                 self.playback_stream = None
             logging.info("Task finished: play_gemini_audio.")
 
     def signal_stop(self):
         logging.info("Signal to stop GeminiInteractionLoop received.")
         self.is_running = False
-        # Use global queue variables directly
         for q in [video_frames_to_gemini_q, audio_chunks_to_gemini_q, audio_from_gemini_playback_q]:
-            if q: # Check if queue is initialized
+            if q:
                 try: q.put_nowait(None)
                 except asyncio.QueueFull: logging.warning(f"Queue was full when trying to put sentinel for stop signal.")
                 except Exception as e: logging.error(f"Error putting sentinel in queue: {e}", exc_info=True)
 
     async def run_main_loop(self):
-        global video_frames_to_gemini_q, audio_chunks_to_gemini_q, audio_from_gemini_playback_q # Allow modification of global vars
+        global video_frames_to_gemini_q, audio_chunks_to_gemini_q, audio_from_gemini_playback_q
 
         self.async_event_loop = asyncio.get_running_loop()
         self.is_running = True
         logging.info("GeminiInteractionLoop run_main_loop starting...")
 
-        # Initialize queues here, within the asyncio loop of this thread
         video_frames_to_gemini_q = asyncio.Queue(maxsize=MEDIA_TO_GEMINI_QUEUE_MAXSIZE)
         audio_chunks_to_gemini_q = asyncio.Queue(maxsize=MEDIA_TO_GEMINI_QUEUE_MAXSIZE)
         audio_from_gemini_playback_q = asyncio.Queue(maxsize=AUDIO_PLAYBACK_QUEUE_MAXSIZE)
         logging.info("Asyncio queues initialized in GeminiInteractionLoop.")
 
-        if client is None:
-            logging.critical("Gemini client is None in run_main_loop. Aborting.")
-            return
+        if client is None: logging.critical("Gemini client is None in run_main_loop. Aborting."); return
 
+        session_task_group = None # Define before try block
         try:
             async with client.aio.live.connect(model=MODEL_NAME, config=LIVE_CONNECT_CONFIG) as session:
                 self.gemini_session = session
@@ -251,24 +242,32 @@ class GeminiInteractionLoop:
                     self.is_running = False; return
 
                 async with asyncio.TaskGroup() as tg:
+                    session_task_group = tg # Assign task group
                     logging.info("Creating async tasks for Gemini interaction...")
                     tg.create_task(self.stream_media_to_gemini(), name="stream_media_to_gemini")
                     tg.create_task(self.process_gemini_responses(), name="process_gemini_responses")
                     tg.create_task(self.play_gemini_audio(), name="play_gemini_audio")
                     logging.info("All Gemini interaction tasks created in TaskGroup.")
                 logging.info("Gemini TaskGroup finished execution.")
+
         except asyncio.CancelledError: logging.info("GeminiInteractionLoop.run_main_loop() was cancelled.")
-        except ExceptionGroup as eg:
-            logging.error(f"ExceptionGroup caught in GeminiInteractionLoop: {eg}")
-            for i, exc in enumerate(eg.exceptions):
-                logging.error(f"  Task Exception {i+1}/{len(eg.exceptions)}: {type(exc).__name__}: {exc}", exc_info=exc)
+        # Removed ExceptionGroup handling as it's not available in Python 3.9
         except Exception as e:
-            logging.error(f"General Exception in GeminiInteractionLoop: {type(e).__name__}: {e}", exc_info=True)
+            # Log general exceptions, including the ConnectionClosedError if it happens here
+            logging.error(f"Exception in GeminiInteractionLoop run_main_loop: {type(e).__name__}: {e}", exc_info=True)
         finally:
             logging.info("GeminiInteractionLoop.run_main_loop() finishing...")
-            self.is_running = False
-            self.gemini_session = None
-            # Clear queues on exit to prevent issues if loop restarts
+            self.is_running = False # Ensure flag is set
+            # Cancel any potentially lingering tasks if TaskGroup didn't exit cleanly (though it should)
+            if session_task_group and not session_task_group._closed:
+                 logging.warning("TaskGroup did not close cleanly, attempting cancellation.")
+                 try:
+                     session_task_group.cancel() # Attempt cancellation if needed
+                 except Exception as e_cancel:
+                     logging.error(f"Error cancelling task group: {e_cancel}")
+
+            self.gemini_session = None # Session closed by async with
+            # Clear global queues
             video_frames_to_gemini_q = None
             audio_chunks_to_gemini_q = None
             audio_from_gemini_playback_q = None
@@ -281,15 +280,10 @@ class VideoProcessor(VideoProcessorBase):
         self.last_gemini_send_time = time.monotonic()
 
     async def _process_and_queue_frame_async(self, frame_ndarray):
-        if video_frames_to_gemini_q is None: # Wait for queue to be initialized
-            logging.debug("VideoProcessor: video_frames_to_gemini_q is None, waiting...")
-            return
-
+        if video_frames_to_gemini_q is None: return
         self.frame_counter += 1
         current_time = time.monotonic()
-        if (current_time - self.last_gemini_send_time) < (1.0 / VIDEO_FPS_TO_GEMINI):
-            return
-
+        if (current_time - self.last_gemini_send_time) < (1.0 / VIDEO_FPS_TO_GEMINI): return
         self.last_gemini_send_time = current_time
         try:
             img_rgb = cv2.cvtColor(frame_ndarray, cv2.COLOR_BGR2RGB)
@@ -299,7 +293,6 @@ class VideoProcessor(VideoProcessorBase):
             pil_img.save(image_io, format="jpeg")
             image_bytes = image_io.getvalue()
             api_data = {"mime_type": "image/jpeg", "data": base64.b64encode(image_bytes).decode()}
-            
             if video_frames_to_gemini_q.full():
                 try: await asyncio.wait_for(video_frames_to_gemini_q.get(), timeout=0.01)
                 except asyncio.TimeoutError: logging.warning("Video queue full, frame dropped."); return
@@ -308,27 +301,19 @@ class VideoProcessor(VideoProcessorBase):
 
     async def recv(self, frame):
         img_bgr = frame.to_ndarray(format="bgr24")
-        # Ensure an event loop is running in the current thread for create_task
         try:
             loop = asyncio.get_running_loop()
             loop.create_task(self._process_and_queue_frame_async(img_bgr))
-        except RuntimeError: # No running loop in this thread (should not happen with streamlit-webrtc async_processing=True)
-            logging.error("VideoProcessor.recv: No running asyncio loop in current thread for create_task.")
-            # Fallback or log error, direct call might block WebRTC thread
-            # await self._process_and_queue_frame_async(img_bgr) # Potentially blocking
+        except RuntimeError: logging.error("VideoProcessor.recv: No running asyncio loop in current thread for create_task.")
         return frame
 
 class AudioProcessor(AudioProcessorBase):
     async def _process_and_queue_audio_async(self, audio_frames):
-        if audio_chunks_to_gemini_q is None: # Wait for queue to be initialized
-            logging.debug("AudioProcessor: audio_chunks_to_gemini_q is None, waiting...")
-            return
-
+        if audio_chunks_to_gemini_q is None: return
         for frame in audio_frames:
             audio_data = frame.planes[0].to_bytes()
             mime_type = f"audio/L16;rate={frame.sample_rate};channels={frame.layout.channels}"
             api_data = {"data": audio_data, "mime_type": mime_type}
-
             try:
                 if audio_chunks_to_gemini_q.full():
                     try: await asyncio.wait_for(audio_chunks_to_gemini_q.get(), timeout=0.01)
@@ -340,9 +325,7 @@ class AudioProcessor(AudioProcessorBase):
         try:
             loop = asyncio.get_running_loop()
             loop.create_task(self._process_and_queue_audio_async(frames))
-        except RuntimeError:
-            logging.error("AudioProcessor.recv: No running asyncio loop in current thread for create_task.")
-            # await self._process_and_queue_audio_async(frames) # Potentially blocking
+        except RuntimeError: logging.error("AudioProcessor.recv: No running asyncio loop in current thread for create_task.")
         return frames
 
 # --- Streamlit UI and Application Logic ---
@@ -363,6 +346,8 @@ def run_streamlit_app():
 
     st.title("Live AI Medical Assistant")
     st.markdown("Utilizing Gemini Live API via WebRTC on Hugging Face Spaces")
+    st.info("Remember: This AI cannot provide medical diagnoses. Always consult a healthcare professional for medical advice.")
+
 
     with st.sidebar:
         st.header("Session Control")
@@ -402,11 +387,6 @@ def run_streamlit_app():
             }
         }
         
-        # Only render WebRTC streamer if queues are expected to be initialized by the Gemini loop
-        # This is a bit of a race condition check, might need refinement
-        # A better way would be for Gemini loop to signal when queues are ready.
-        # For now, we assume if session is active, loop is trying to start and init queues.
-        
         webrtc_ctx = webrtc_streamer(
             key=st.session_state.webrtc_component_key,
             mode=WebRtcMode.SENDONLY,
@@ -414,8 +394,8 @@ def run_streamlit_app():
                 "iceServers": [{"urls": ["stun:stun.l.google.com:19302"]}]
             },
             media_stream_constraints=MEDIA_STREAM_CONSTRAINTS,
-            video_processor_factory=VideoProcessor, # Pass the class
-            audio_processor_factory=AudioProcessor, # Pass the class
+            video_processor_factory=VideoProcessor,
+            audio_processor_factory=AudioProcessor,
             async_processing=True,
         )
 
@@ -429,12 +409,16 @@ def run_streamlit_app():
         st.info("Click 'Start Session' in the sidebar to enable the live feed and assistant.")
 
     st.subheader("Chat with Assistant")
-    chat_placeholder = st.container() 
-    with chat_placeholder:
-        for msg in st.session_state.get('chat_messages', []):
+    # Use a container with a fixed height for scrollable chat
+    chat_container = st.container()
+    with chat_container:
+        # Display messages from session state
+        messages = st.session_state.get('chat_messages', [])
+        for msg in messages:
             with st.chat_message(msg["role"]):
                 st.write(msg["content"])
-    
+
+    # Chat input outside the container
     user_chat_input = st.chat_input(
         "Type your message...",
         key="user_chat_input_box",
@@ -442,25 +426,39 @@ def run_streamlit_app():
     )
 
     if user_chat_input:
-        current_messages = st.session_state.get('chat_messages', [])
-        current_messages.append({"role": "user", "content": user_chat_input})
-        st.session_state.chat_messages = current_messages
+        # Append user message immediately for responsiveness
+        st.session_state.chat_messages = st.session_state.get('chat_messages', []) + [{"role": "user", "content": user_chat_input}]
         
+        # Send to backend
         loop_instance = st.session_state.get('gemini_loop_instance')
         if loop_instance and loop_instance.async_event_loop and loop_instance.gemini_session:
             if loop_instance.async_event_loop.is_running():
-                asyncio.run_coroutine_threadsafe(
+                # Use run_coroutine_threadsafe to call async func from Streamlit thread
+                future = asyncio.run_coroutine_threadsafe(
                     loop_instance.send_text_input_to_gemini(user_chat_input),
                     loop_instance.async_event_loop
                 )
+                try:
+                    future.result(timeout=2) # Optional: wait briefly for confirmation
+                except TimeoutError:
+                    logging.warning("Timed out waiting for send_text_input_to_gemini confirmation.")
+                except Exception as e:
+                     logging.error(f"Error calling send_text_input_to_gemini: {e}", exc_info=True)
+
             else: st.error("Session event loop is not running. Cannot send message.")
         elif not loop_instance or not st.session_state.gemini_session_active:
             st.error("Session is not active. Please start a session to send messages.")
         else: st.warning("Session components not fully ready. Please wait a moment.")
+        
+        # Rerun to display the user message and potentially any quick text response from AI
         st.rerun()
 
 if __name__ == "__main__":
+    # Final check before running the app
     if client is None:
+        # Log critical error if client is still None (should have been caught earlier)
         logging.critical("Gemini client could not be initialized. Application cannot start.")
+        # Display error in Streamlit if possible (might not render if client init failed early)
+        st.error("CRITICAL ERROR: Gemini client failed to initialize. Check API Key and logs.")
     else:
         run_streamlit_app()