Update src/streamlit_app.py

src/streamlit_app.py

@@ -11,41 +11,51 @@ import time
 import logging
 from dotenv import load_dotenv
 
-import cv2 # For image processing
-import pyaudio # For audio PLAYBACK
+import cv2
+import pyaudio
 import PIL.Image
 
 from google import genai
 from google.genai import types
 
-# streamlit-webrtc components
 from streamlit_webrtc import (
     webrtc_streamer,
     WebRtcMode,
     AudioProcessorBase,
     VideoProcessorBase,
 )
-# from aiortc import RTCIceServer, RTCConfiguration # Not needed directly
 
-# --- Configuration ---
 load_dotenv()
-logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(threadName)s - %(levelname)s - %(message)s')
-logging.info("Application starting up...")
+
 
 # Audio configuration
-PYAUDIO_FORMAT = pyaudio.paInt16
-PYAUDIO_CHANNELS = 1
-WEBRTC_REQUESTED_AUDIO_CHANNELS = 1
-WEBRTC_REQUESTED_SEND_SAMPLE_RATE = 16000
-GEMINI_AUDIO_RECEIVE_SAMPLE_RATE = 24000
-PYAUDIO_PLAYBACK_CHUNK_SIZE = 1024
-AUDIO_PLAYBACK_QUEUE_MAXSIZE = 50
-MEDIA_TO_GEMINI_QUEUE_MAXSIZE = 30
+
+FORMAT = pyaudio.paInt16
+CHANNELS = 1
+SEND_SAMPLE_RATE = 16000
+RECEIVE_SAMPLE_RATE = 24000
+CHUNK_SIZE = 1024
+
+# Map PyAudio format to a more descriptive name for clarity.
+PYAUDIO_FORMAT = FORMAT  # pyaudio.paInt16
+PYAUDIO_CHANNELS = CHANNELS
+PYAUDIO_PLAYBACK_CHUNK_SIZE = CHUNK_SIZE
+GEMINI_AUDIO_RECEIVE_SAMPLE_RATE = RECEIVE_SAMPLE_RATE
 
 # Video configuration
+
 VIDEO_FPS_TO_GEMINI = 2
 VIDEO_API_RESIZE = (1024, 1024)
 
+# Queue sizes
+MEDIA_TO_GEMINI_QUEUE_MAXSIZE = 10
+AUDIO_PLAYBACK_QUEUE_MAXSIZE = 10
+
+# WebRTC settings
+WEBRTC_REQUESTED_SEND_SAMPLE_RATE = SEND_SAMPLE_RATE
+WEBRTC_REQUESTED_AUDIO_CHANNELS = CHANNELS
+
+
 # !!! IMPORTANT: Verify this model name is correct for the Live API !!!
 MODEL_NAME = "models/gemini-2.0-flash-live-001"
 logging.info(f"Using Gemini Model: {MODEL_NAME}")
@@ -68,13 +78,16 @@ Example of a disclaimer you might use: "As an AI assistant, I can describe what
 pya = None
 try:
     pya = pyaudio.PyAudio()
+
     def cleanup_pyaudio():
         logging.info("Terminating PyAudio instance.")
-        if pya: pya.terminate()
+        if pya:
+            pya.terminate()
     atexit.register(cleanup_pyaudio)
     logging.info("PyAudio initialized successfully.")
 except Exception as e_pyaudio:
-    logging.warning(f"PyAudio initialization failed (expected in some server environments): {e_pyaudio}")
+    logging.warning(
+        f"PyAudio initialization failed (expected in some server environments): {e_pyaudio}")
     pya = None
 
 # --- Global Queues - Declare as None, initialize later ---
@@ -94,17 +107,19 @@ if GEMINI_API_KEY:
         logging.critical(f"Gemini client initialization failed: {e}", exc_info=True)
         st.stop()
 else:
-    st.error("GEMINI_API_KEY not found in environment variables. Please set it for the application to run.")
+    st.error(
+        "GEMINI_API_KEY not found in environment variables. Please set it for the application to run.")
     logging.critical("GEMINI_API_KEY not found.")
     st.stop()
 
 LIVE_CONNECT_CONFIG = types.LiveConnectConfig(
-    response_modalities=["audio", "text", "video"], # Requesting audio response
+    response_modalities=["audio", "text", "video"],  # Requesting audio response
     speech_config=types.SpeechConfig(
         voice_config=types.VoiceConfig(
-            prebuilt_voice_config=types.PrebuiltVoiceConfig(voice_name="Zephyr") # Using Puck voice
+            # Using Puck voice
+            prebuilt_voice_config=types.PrebuiltVoiceConfig(voice_name="Zephyr")
         )
-    ) # <---------------------------------- CORRECTED: Added missing closing parenthesis
+    )  # <---------------------------------- CORRECTED: Added missing closing parenthesis
 )
 logging.info(f"Attempting connection with LiveConnectConfig: {LIVE_CONNECT_CONFIG}")
 
@@ -119,7 +134,8 @@ class GeminiInteractionLoop:
 
     async def send_text_input_to_gemini(self, user_text):
         if not user_text or not self.gemini_session or not self.is_running:
-            logging.warning("Cannot send text. Session not active, no text, or not running.")
+            logging.warning(
+                "Cannot send text. Session not active, no text, or not running.")
             return
         try:
             logging.info(f"Sending text to Gemini: '{user_text[:50]}...'")
@@ -127,96 +143,144 @@ class GeminiInteractionLoop:
             # For now, keeping session.send as it was working functionally
             await self.gemini_session.send(input=user_text, end_of_turn=True)
         except Exception as e:
-            logging.error(f"Error sending text message to Gemini: {e}", exc_info=True)
+            logging.error(
+                f"Error sending text message to Gemini: {e}", exc_info=True)
 
     async def stream_media_to_gemini(self):
         logging.info("Task started: Stream media from WebRTC queues to Gemini.")
+
         async def get_media_from_queues():
             if video_frames_to_gemini_q is None or audio_chunks_to_gemini_q is None:
-                await asyncio.sleep(0.1); 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)
-                if video_frame is None: return None # Sentinel received
-                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)
+                if video_frame is None:
+                    return None  # Sentinel received
+                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)
-                if audio_chunk is None: return None # Sentinel received
-                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
+                if audio_chunk is None:
+                    return None  # Sentinel received
+                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
 
         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 is None and not self.is_running: break # Sentinel and stop signal
+                if media_data is None and not self.is_running:
+                    break  # Sentinel and stop signal
                 if media_data and self.gemini_session and self.is_running:
                     try:
                         # Use the specific method as suggested by the deprecation warning if possible
                         # For now, keeping session.send as it was working functionally
                         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) # No data, yield
-        except asyncio.CancelledError: logging.info("Task cancelled: stream_media_to_gemini.")
-        finally: logging.info("Task finished: stream_media_to_gemini.")
+                    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)  # No data, yield
+        except asyncio.CancelledError:
+            logging.info("Task cancelled: stream_media_to_gemini.")
+        finally:
+            logging.info("Task finished: stream_media_to_gemini.")
 
     async def process_gemini_responses(self):
         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: 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 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)
-                            else: logging.warning("Audio playback queue full, discarding Gemini 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 = []
-                            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.")
+                            if 'chat_messages' not in st.session_state:
+                                st.session_state.chat_messages = []
+                            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:
-                    if self.is_running: logging.error(f"Error receiving from Gemini: {e}", exc_info=True)
+                    if self.is_running:
+                        logging.error(
+                            f"Error receiving from Gemini: {e}", exc_info=True)
                     await asyncio.sleep(0.1)
-        except asyncio.CancelledError: logging.info("Task cancelled: process_gemini_responses.")
-        finally: logging.info("Task finished: process_gemini_responses.")
+        except asyncio.CancelledError:
+            logging.info("Task cancelled: process_gemini_responses.")
+        finally:
+            logging.info("Task finished: process_gemini_responses.")
 
     async def play_gemini_audio(self):
         logging.info("Task started: Play Gemini audio responses.")
         if pya is None:
-             logging.warning("PyAudio not available. Audio playback task will not run.")
-             return
+            logging.warning(
+                "PyAudio not available. Audio playback task will not run.")
+            return
 
         try:
-            while audio_from_gemini_playback_q is None and self.is_running: await asyncio.sleep(0.1)
-            if not self.is_running: return
+            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(
                 pya.open, format=PYAUDIO_FORMAT, channels=PYAUDIO_CHANNELS, rate=GEMINI_AUDIO_RECEIVE_SAMPLE_RATE, output=True, frames_per_buffer=PYAUDIO_PLAYBACK_CHUNK_SIZE
             )
-            logging.info(f"PyAudio playback stream opened at {GEMINI_AUDIO_RECEIVE_SAMPLE_RATE} Hz.")
+            logging.info(
+                f"PyAudio playback stream opened at {GEMINI_AUDIO_RECEIVE_SAMPLE_RATE} Hz.")
             while self.is_running:
                 try:
                     audio_chunk = await asyncio.wait_for(audio_from_gemini_playback_q.get(), timeout=1.0)
-                    if audio_chunk is None and not self.is_running: break # Sentinel and stop signal
-                    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 chunk: {e}", exc_info=True); await asyncio.sleep(0.01)
+                    if audio_chunk is None and not self.is_running:
+                        break  # Sentinel and stop signal
+                    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 chunk: {e}", exc_info=True)
+                    await asyncio.sleep(0.01)
         except Exception as e:
-            logging.error(f"Failed to open or use PyAudio playback stream (might be expected in this environment): {e}", exc_info=True)
+            logging.error(
+                f"Failed to open or use PyAudio playback stream (might be expected in this environment): {e}", exc_info=True)
         finally:
             if self.playback_stream:
                 logging.info("Stopping and closing PyAudio playback stream.")
                 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)
+                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.")
 
@@ -227,9 +291,14 @@ class GeminiInteractionLoop:
                                   ("audio_in_q", audio_chunks_to_gemini_q),
                                   ("audio_out_q", audio_from_gemini_playback_q)]:
             if q_obj_ref:
-                try: q_obj_ref.put_nowait(None)
-                except asyncio.QueueFull: logging.warning(f"Queue {q_name} was full when trying to put sentinel for stop signal.")
-                except Exception as e: logging.error(f"Error putting sentinel in {q_name}: {e}", exc_info=True)
+                try:
+                    q_obj_ref.put_nowait(None)
+                except asyncio.QueueFull:
+                    logging.warning(
+                        f"Queue {q_name} was full when trying to put sentinel for stop signal.")
+                except Exception as e:
+                    logging.error(
+                        f"Error putting sentinel in {q_name}: {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
@@ -238,59 +307,77 @@ class GeminiInteractionLoop:
         self.is_running = True
         logging.info("GeminiInteractionLoop run_main_loop starting...")
 
-        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)
+        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
 
         try:
             async with client.aio.live.connect(model=MODEL_NAME, config=LIVE_CONNECT_CONFIG) as session:
                 self.gemini_session = session
-                logging.info(f"Gemini session established with API for model {MODEL_NAME}.")
+                logging.info(
+                    f"Gemini session established with API for model {MODEL_NAME}.")
                 try:
                     logging.info("Sending system prompt to Gemini...")
                     await self.gemini_session.send(input=MEDICAL_ASSISTANT_SYSTEM_PROMPT, end_of_turn=False)
                     logging.info("System prompt sent successfully.")
                 except Exception as e:
-                    logging.error(f"Failed to send system prompt: {e}", exc_info=True)
-                    self.is_running = False; return
+                    logging.error(
+                        f"Failed to send system prompt: {e}", exc_info=True)
+                    self.is_running = False
+                    return
 
                 # Using asyncio.gather for Python 3.9 compatibility
                 tasks = []
                 try:
                     logging.info("Creating async tasks for Gemini interaction...")
-                    tasks.append(asyncio.create_task(self.stream_media_to_gemini(), name="stream_media_to_gemini"))
-                    tasks.append(asyncio.create_task(self.process_gemini_responses(), name="process_gemini_responses"))
-                    tasks.append(asyncio.create_task(self.play_gemini_audio(), name="play_gemini_audio"))
+                    tasks.append(asyncio.create_task(
+                        self.stream_media_to_gemini(), name="stream_media_to_gemini"))
+                    tasks.append(asyncio.create_task(
+                        self.process_gemini_responses(), name="process_gemini_responses"))
+                    tasks.append(asyncio.create_task(
+                        self.play_gemini_audio(), name="play_gemini_audio"))
                     logging.info("All Gemini interaction tasks created.")
                     # Wait for tasks to complete or raise an exception
                     done, pending = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
                     # Check results of completed tasks for errors
                     for future in done:
                         try:
-                            future.result() # Raise exception if task failed
+                            future.result()  # Raise exception if task failed
                         except Exception as task_exc:
-                            logging.error(f"Task {future.get_name()} failed: {task_exc}", exc_info=True)
+                            logging.error(
+                                f"Task {future.get_name()} failed: {task_exc}", exc_info=True)
                             # Optionally cancel remaining tasks if one fails critically
-                            for p_task in pending: p_task.cancel()
+                            for p_task in pending:
+                                p_task.cancel()
                     # If loop completes normally (e.g., user stops), pending tasks will be handled by finally block
-                except Exception as e_gather: # Catch errors during task creation/gathering
-                    logging.error(f"Error during task management: {e_gather}", exc_info=True)
+                except Exception as e_gather:  # Catch errors during task creation/gathering
+                    logging.error(
+                        f"Error during task management: {e_gather}", exc_info=True)
                     for task in tasks:
-                        if not task.done(): task.cancel()
+                        if not task.done():
+                            task.cancel()
                     # Wait for cancellations to complete
                     await asyncio.gather(*tasks, return_exceptions=True)
                 logging.info("Gemini interaction tasks finished or cancelled.")
 
-        except asyncio.CancelledError: logging.info("GeminiInteractionLoop.run_main_loop() was cancelled.")
-        except Exception as e: # General catch-all, including ConnectionClosedError
-            logging.error(f"Exception in GeminiInteractionLoop run_main_loop: {type(e).__name__}: {e}", exc_info=True)
+        except asyncio.CancelledError:
+            logging.info("GeminiInteractionLoop.run_main_loop() was cancelled.")
+        except Exception as e:  # General catch-all, including ConnectionClosedError
+            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.signal_stop() # Ensure sentinels are sent
+            self.signal_stop()  # Ensure sentinels are sent
             # Clean up any remaining tasks (important if gather didn't complete)
             # current_tasks = [t for t in asyncio.all_tasks(self.async_event_loop) if t is not asyncio.current_task()]
             # if current_tasks:
@@ -302,7 +389,8 @@ class GeminiInteractionLoop:
             video_frames_to_gemini_q = None
             audio_chunks_to_gemini_q = None
             audio_from_gemini_playback_q = None
-            logging.info("GeminiInteractionLoop finished and global queues set to None.")
+            logging.info(
+                "GeminiInteractionLoop finished and global queues set to None.")
 
 
 # --- WebRTC Media Processors ---
@@ -312,10 +400,12 @@ 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: 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)
@@ -324,40 +414,57 @@ class VideoProcessor(VideoProcessorBase):
             image_io = io.BytesIO()
             pil_img.save(image_io, format="jpeg")
             image_bytes = image_io.getvalue()
-            api_data = {"mime_type": "image/jpeg", "data": base64.b64encode(image_bytes).decode()}
+            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
+                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
             video_frames_to_gemini_q.put_nowait(api_data)
-        except Exception as e: logging.error(f"Error processing/queueing video frame: {e}", exc_info=True)
+        except Exception as e:
+            logging.error(
+                f"Error processing/queueing video frame: {e}", exc_info=True)
 
     async def recv(self, frame):
         img_bgr = frame.to_ndarray(format="bgr24")
         try:
             loop = asyncio.get_running_loop()
             loop.create_task(self._process_and_queue_frame_async(img_bgr))
-        except RuntimeError: logging.error("VideoProcessor.recv: No running asyncio loop in current thread for create_task.")
+        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: 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)
-                    except asyncio.TimeoutError: logging.warning("Audio queue full, chunk dropped."); continue
+                    try:
+                        await asyncio.wait_for(audio_chunks_to_gemini_q.get(), timeout=0.01)
+                    except asyncio.TimeoutError:
+                        logging.warning("Audio queue full, chunk dropped.")
+                        continue
                 audio_chunks_to_gemini_q.put_nowait(api_data)
-            except Exception as e: logging.error(f"Error queueing audio chunk: {e}", exc_info=True)
-    
+            except Exception as e:
+                logging.error(
+                    f"Error queueing audio chunk: {e}", exc_info=True)
+
     async def recv(self, frames):
         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.")
+        except RuntimeError:
+            logging.error(
+                "AudioProcessor.recv: No running asyncio loop in current thread for create_task.")
         return frames
 
 # --- Streamlit UI and Application Logic ---