"""Gemini WebSocket client for real-time voice interaction. Provides: - Bidirectional audio streaming (mic → Gemini → speaker) - Text-to-speech via typed input - Voice-command detection through transcription parsing - System instruction injection for persona control """ from __future__ import annotations import asyncio import base64 import inspect import json import urllib.error import urllib.request from typing import Any import websockets from Project.Sanad.config import ( GEMINI_API_KEY, GEMINI_MODEL, GEMINI_VOICE, GEMINI_WS_TIMEOUT, GEMINI_WS_URI, ) from Project.Sanad.core.config_loader import section as _cfg_section from Project.Sanad.core.event_bus import bus from Project.Sanad.core.logger import get_logger log = get_logger("gemini_client") _GC = _cfg_section("gemini", "client") # Default system prompt — SINGLE SOURCE in core.gemini_defaults _DEFAULT_SYSTEM_PROMPT = _cfg_section("core", "gemini_defaults").get( "default_system_prompt", "You are Sanad (Bousandah), a wise and friendly Emirati assistant. " "Speak in UAE dialect (Khaleeji). Be helpful and concise." ) # TTS / typed-replay system prompt. The voice_client speaks TYPED text, so it # must read the text VERBATIM in its OWN language — NOT answer it and NOT force # Khaleeji. This is what makes the Live native-audio model return AUDIO instead # of "thinking" text. Copied from SanadR1/Sanadv3 so the lite dashboard speaks # with the exact same engine + behavior as the robots. TTS_SYSTEM_PROMPT = _cfg_section("core", "gemini_defaults").get( "tts_system_prompt", "You are a pure multilingual text-to-speech voice. The instant the user " "sends text, speak it aloud word for word in the SAME language it is " "written in, then stop. Output ONLY that spoken audio — no thinking, no " "commentary, no acknowledgements, no headers, no explanations, no " "greetings, no extra words. Never translate and never change the language: " "English stays English, Arabic stays Arabic, Urdu stays Urdu, Indonesian " "stays Indonesian. Your speech must be identical to the user's text, " "nothing more and nothing less." ) # Per-voice system-prompt overrides, keyed by prebuilt voice name. Empty by # default, and INTENTIONALLY so for Charon (Unitree G1) and Puck (Unitree R1): # those two must keep speaking with the robots' own verbatim TTS prompt, since # sounding identical to Sanadv3/SanadR1 is the whole point. Only voices with no # robot to match — Agibot x2 / Kore — get a different instruction here. # Configured in core_config.json → gemini_defaults.voice_system_prompts. _VOICE_SYSTEM_PROMPTS = _cfg_section("core", "gemini_defaults").get( "voice_system_prompts", {}) or {} _RECV_TIMEOUT_SEC = _GC.get("recv_timeout_sec", 30) _RECONNECT_MAX_ATTEMPTS = _GC.get("reconnect_max_attempts", 3) _RECONNECT_INITIAL_DELAY_SEC = _GC.get("reconnect_initial_delay_sec", 1.0) _RECONNECT_MAX_DELAY_SEC = _GC.get("reconnect_max_delay_sec", 10.0) # Dedicated text-to-speech model (stateless REST generateContent). Far more # reliable than the Live native-audio model for pure TTS: the Live model often # returns its "thinking" reasoning text instead of speech for short prompts, # whereas this returns audio-only. Same prebuilt voices (Charon/Puck/Kore/...). GEMINI_TTS_MODEL = _GC.get("tts_model", "gemini-2.5-flash-preview-tts") _GEMINI_REST_BASE = _GC.get( "rest_base", "https://generativelanguage.googleapis.com/v1beta") # Spoken-directive preamble so the TTS model READS the text aloud rather than # answering it (bare short inputs like "مرحبا" otherwise 400 "Model tried to # generate text"), and nudges an Emirati/Gulf dialect. The directive is NOT # spoken (verified: audio length ~unchanged with/without it). Retries cover the # model's occasional empty response. All config-overridable via gemini_config.json. _TTS_PREAMBLE = _GC.get("tts_preamble", "بِاللهجة الإماراتية: ") _TTS_MAX_ATTEMPTS = int(_GC.get("tts_max_attempts", 3)) # BCP-47 language/accent code for the TTS speechConfig. "ar-AE" = UAE Arabic — # the structured, reliable lever for an Emirati accent. "" omits the field. _TTS_LANGUAGE_CODE = _GC.get("tts_language_code", "ar-AE") # Sampling temperature. 0.0 makes the TTS DETERMINISTIC — the same text renders # the same audio every time (stable tone). The default (~1.0) re-performs with # random prosody on every call, which is why the tone kept drifting between # generations. The retry temperature is used only if temp-0 returns an empty # part for a given input (so we can still get audio for it). _TTS_TEMPERATURE = float(_GC.get("tts_temperature", 0.0)) _TTS_RETRY_TEMPERATURE = float(_GC.get("tts_retry_temperature", 0.6)) # Per-attempt HTTP timeout for the TTS call. A normal TTS response is ~2s; a # stuck call would otherwise hang the whole generation and hold the typed-replay # single-flight lock (blocking the UI), so keep this tight. _TTS_TIMEOUT = float(_GC.get("tts_timeout_sec", 20)) # Ask the Live session to transcribe its own spoken output. Costs nothing # extra and gives typed_replay a way to verify the whole text was actually # read aloud. Set false in gemini_config.json to go back to blind reads. _OUTPUT_TRANSCRIPTION = bool(_GC.get("output_transcription", True)) # How long to keep reading after `generationComplete` while waiting for # `turnComplete`. Only the tail of an already-generated turn arrives in this # window, so it is short. _POST_GENERATION_GRACE_SEC = float(_GC.get("post_generation_grace_sec", 3.0)) class GeminiQuotaExhausted(RuntimeError): """The API key itself is out of credits / over quota. Distinct from every other failure because retrying is pointless: the Live socket, the reconnect chain and the REST TTS fallback all fail the same way. Without this, one Generate & Play burns 4 Live attempts (each with a reconnect chain) plus 3 REST retries before answering — long enough to blow the Apache proxy timeout, so the user sees a bare 503 and the cron health-check restarts a perfectly healthy app. """ # Substrings that mean "this key cannot make calls right now". Deliberately # narrow: a transient 429 rate-limit IS worth retrying, an empty wallet is not. _QUOTA_MARKERS = ( "credits are depleted", "prepayment credits", "billing", "exceeded your current quota", ) def _quota_message(exc: BaseException) -> str: """Return a clean reason string if `exc` is a credits/quota failure, else "".""" text = f"{exc}".lower() if not any(m in text for m in _QUOTA_MARKERS): return "" if "credits are depleted" in text or "prepayment credits" in text: return ("Gemini API credits are depleted — top up billing in AI Studio " "for this project's key. No voice can be generated until then.") return ("Gemini API quota/billing error — this API key cannot generate " "audio right now.") def _tts_rest_blocking(url: str, body: dict, timeout: float) -> bytes: """Blocking POST to the TTS generateContent endpoint → PCM bytes. Runs in a worker thread (via asyncio.to_thread) so the event loop is never blocked. Returns the first inline audio part decoded from base64; raises RuntimeError carrying the API's message on an HTTP/transport error. """ data = json.dumps(body).encode("utf-8") req = urllib.request.Request( url, data=data, headers={"Content-Type": "application/json"}, method="POST") try: with urllib.request.urlopen(req, timeout=timeout) as resp: raw = resp.read().decode("utf-8") except urllib.error.HTTPError as exc: detail = "" try: detail = json.loads(exc.read().decode("utf-8")).get("error", {}).get("message", "") except Exception: pass raise RuntimeError(f"Gemini TTS HTTP {exc.code}: {detail or exc.reason}") except urllib.error.URLError as exc: raise RuntimeError(f"Gemini TTS request failed: {exc.reason}") payload = json.loads(raw) if isinstance(payload, dict) and payload.get("error"): msg = payload["error"].get("message", payload["error"]) raise RuntimeError(f"Gemini TTS error: {msg}") for cand in payload.get("candidates", []): for part in cand.get("content", {}).get("parts", []): inline = part.get("inlineData") or part.get("inline_data") if inline and inline.get("data"): return base64.b64decode(inline["data"]) return b"" class GeminiVoiceClient: """Manages one WebSocket session to the Gemini Bidi audio API. Concurrency model: - `_send_lock` serializes ALL websocket writes. - `_session_lock` ensures only one consumer (live loop OR typed replay) owns the receive stream at a time. Acquired by send_text and receive_stream context managers. - `_owner` records who currently holds the session lock for diagnostics. """ def __init__(self, system_prompt: str = ""): self.system_prompt = system_prompt or _DEFAULT_SYSTEM_PROMPT self._ws: Any = None self._connected = False self._send_lock = asyncio.Lock() self._session_lock = asyncio.Lock() self._connect_lock = asyncio.Lock() # serializes reconnect attempts self._owner: str | None = None self._reconnect_attempts = 0 # Transcript of the audio returned by the most recent send_text(), # when output transcription is enabled. Written under the session # lock, so it always belongs to the call that just returned. self.last_output_transcript = "" @property def connected(self) -> bool: return self._connected @property def session_owner(self) -> str | None: return self._owner def _ws_kwargs(self) -> dict[str, Any]: kwargs: dict[str, Any] = {"max_size": None, "open_timeout": 30} try: sig = inspect.signature(websockets.connect) key = "extra_headers" if "extra_headers" in sig.parameters else "additional_headers" except Exception: key = "extra_headers" kwargs[key] = {"Content-Type": "application/json"} return kwargs async def _safe_close_ws(self) -> None: """Best-effort close of the current socket, guarded by a timeout. A half-dead socket whose close() hangs must never wedge the event loop here, so the close itself is bounded. """ ws, self._ws = self._ws, None if ws is None: return try: await asyncio.wait_for(ws.close(), timeout=5) except Exception: pass async def connect(self): uri = f"{GEMINI_WS_URI}?key={GEMINI_API_KEY}" # The system instruction is baked into this handshake alongside the # voice, so per-voice prompts can only be applied here. Voices with no # override (Charon/Puck) keep self.system_prompt untouched. system_prompt = _VOICE_SYSTEM_PROMPTS.get(GEMINI_VOICE) or self.system_prompt try: self._ws = await websockets.connect(uri, **self._ws_kwargs()) setup = { "setup": { "model": GEMINI_MODEL, "generationConfig": { "responseModalities": ["AUDIO"], "speechConfig": { "voiceConfig": { "prebuiltVoiceConfig": {"voiceName": GEMINI_VOICE} } }, }, "systemInstruction": {"parts": [{"text": system_prompt}]}, } } if _OUTPUT_TRANSCRIPTION: # Ask the server to transcribe the audio it actually speaks. # This is the only reliable way to tell a COMPLETE read from a # cut-off one: the native-audio model regularly speaks the # first few words, stops, and emits reasoning text instead of # the rest — and the audio it returns looks perfectly healthy. # typed_replay compares this transcript against the requested # text and retries when words are missing. setup["setup"]["outputAudioTranscription"] = {} # Guard the app-level setup handshake with a timeout. websockets' # own open_timeout only covers the HTTP upgrade, NOT this send/ACK. # A socket that opens but never ACKs would otherwise block this # await forever and freeze uvicorn's single event loop — every HTTP # request with it. That is exactly what took the site down. await asyncio.wait_for( self._ws.send(json.dumps(setup)), timeout=GEMINI_WS_TIMEOUT) await asyncio.wait_for(self._ws.recv(), timeout=GEMINI_WS_TIMEOUT) # ACK self._connected = True self._reconnect_attempts = 0 log.info("Connected to Gemini (%s, voice=%s%s)", GEMINI_MODEL, GEMINI_VOICE, ", voice-specific prompt" if GEMINI_VOICE in _VOICE_SYSTEM_PROMPTS else "") await bus.emit("voice.connected") except asyncio.TimeoutError: self._connected = False await self._safe_close_ws() log.warning("Gemini setup handshake timed out after %ss", GEMINI_WS_TIMEOUT) raise except Exception as exc: self._connected = False await self._safe_close_ws() reason = _quota_message(exc) if reason: # Out of credits — say so once, plainly, and stop. Retrying # only stacks up latency until the proxy times out. log.error("Gemini refused the connection: %s", reason) raise GeminiQuotaExhausted(reason) from exc log.exception("Failed to connect to Gemini") raise async def disconnect(self): await self._safe_close_ws() self._connected = False self._owner = None log.info("Disconnected from Gemini") await bus.emit("voice.disconnected") async def _ensure_connected(self): """Reconnect if dropped, with bounded retries. Serialized via _connect_lock so concurrent callers don't trigger duplicate handshakes. """ # Fast path — no lock needed if self._connected and self._ws is not None: return True async with self._connect_lock: # Re-check inside the lock (another coroutine may have just connected) if self._connected and self._ws is not None: return True max_attempts = _RECONNECT_MAX_ATTEMPTS delay = _RECONNECT_INITIAL_DELAY_SEC for attempt in range(max_attempts): try: log.warning("Reconnecting to Gemini (attempt %d/%d)", attempt + 1, max_attempts) await self.connect() return True except GeminiQuotaExhausted: # No amount of reconnecting refills the account — surface it # to the caller immediately instead of sleeping through the # whole backoff chain on every attempt. raise except Exception: self._reconnect_attempts += 1 await asyncio.sleep(delay) delay = min(delay * 2, _RECONNECT_MAX_DELAY_SEC) log.error("Reconnect failed after %d attempts", max_attempts) await bus.emit("voice.error", reason="reconnect_failed") return False async def send_audio_chunk(self, pcm_b64: str) -> bool: """Send a base64-encoded PCM audio chunk (mic input). Returns False on failure so the caller can react instead of silently no-op'ing forever (the original bug). """ if not self._connected or self._ws is None: return False msg = { "realtimeInput": { "mediaChunks": [ {"mimeType": "audio/pcm;rate=16000", "data": pcm_b64} ] } } try: async with self._send_lock: await asyncio.wait_for( self._ws.send(json.dumps(msg)), timeout=GEMINI_WS_TIMEOUT) return True except websockets.exceptions.ConnectionClosed: log.warning("send_audio_chunk: connection closed") self._connected = False await bus.emit("voice.error", reason="connection_closed") return False except asyncio.TimeoutError: log.warning("send_audio_chunk: send timed out after %ss", GEMINI_WS_TIMEOUT) self._connected = False await bus.emit("voice.error", reason="send_timeout") return False except Exception: log.exception("send_audio_chunk failed") return False async def send_text(self, text: str, owner: str = "send_text") -> tuple[bytes, list[str]]: """Send text, receive audio response. Returns (audio_bytes, text_parts). Acquires the session lock for the entire request/response cycle so no other consumer can steal frames from the receive side. If the connection drops mid-request, reconnects once and retries. """ if not await self._ensure_connected(): raise RuntimeError("Not connected to Gemini and reconnect failed.") async with self._session_lock: self._owner = owner try: return await self._send_text_inner(text) except (websockets.exceptions.ConnectionClosed, asyncio.TimeoutError) as exc: # A 1011 close carrying a billing message is the server telling # us the key is empty — the close reason is the only place that # information appears, so read it before treating this as a # routine drop and reconnecting into the same wall. reason = _quota_message(exc) if reason: self._connected = False log.error("Gemini closed the session: %s", reason) raise GeminiQuotaExhausted(reason) from exc log.warning("send_text: connection died/stalled on send — reconnecting once") self._connected = False if not await self._ensure_connected(): raise RuntimeError("Reconnect after send failure also failed.") return await self._send_text_inner(text) finally: self._owner = None async def _drain_socket(self) -> int: """Discard frames left over from an earlier turn. Returns how many. If a previous turn ended without consuming everything the server sent, those frames sit in the socket and the NEXT send_text() reads them as its own reply — you get the tail of the last sentence instead of the new one, arriving implausibly fast. Clearing them first makes every turn start from a known-empty stream. """ dropped = 0 while dropped < 500: try: await asyncio.wait_for(self._ws.recv(), timeout=0.01) except (asyncio.TimeoutError, asyncio.CancelledError): break except Exception: break dropped += 1 if dropped: log.warning("drained %d stale frame(s) from the previous turn", dropped) return dropped async def _send_text_inner(self, text: str) -> tuple[bytes, list[str]]: """Inner send/receive loop — caller must hold _session_lock.""" request = { "client_content": { "turns": [{"role": "user", "parts": [{"text": text}]}], "turn_complete": True, } } async with self._send_lock: await self._drain_socket() await asyncio.wait_for( self._ws.send(json.dumps(request)), timeout=GEMINI_WS_TIMEOUT) audio_chunks: list[bytes] = [] text_parts: list[str] = [] transcript_parts: list[str] = [] self.last_output_transcript = "" # `generationComplete` means the model stopped GENERATING — the server # still has audio and transcript to deliver, and only `turnComplete` # ends the turn. Breaking on the former truncated the tail of every # sentence AND left those frames in the socket for the next turn to # mis-read. Wait for turnComplete, with a short grace period after # generationComplete so a turn that never sends it can't stall us. gen_done = False while True: timeout = _POST_GENERATION_GRACE_SEC if gen_done else GEMINI_WS_TIMEOUT try: raw = await asyncio.wait_for(self._ws.recv(), timeout=timeout) except asyncio.TimeoutError: if gen_done: break # tail delivered, server just never closed the turn log.warning("send_text: recv timed out") break except websockets.exceptions.ConnectionClosed: log.warning("send_text: connection closed mid-stream") self._connected = False break try: resp = json.loads(raw) except json.JSONDecodeError: log.warning("send_text: bad JSON from server") continue if "error" in resp: log.error("Gemini error: %s", resp["error"]) await bus.emit("voice.error", reason=str(resp["error"])) break sc = resp.get("serverContent", {}) mt = sc.get("modelTurn", {}) for part in mt.get("parts", []): inline = part.get("inlineData") if inline and inline.get("data"): audio_chunks.append(base64.b64decode(inline["data"])) tp = part.get("text") if isinstance(tp, str) and tp.strip(): text_parts.append(tp.strip()) input_tr = sc.get("inputTranscription", {}) if input_tr.get("text"): await bus.emit("voice.user_said", text=input_tr["text"]) # Transcript of the audio the model is speaking, streamed in # fragments alongside it. Accumulated verbatim; the caller decides # whether it covers the requested text. out_tr = sc.get("outputTranscription", {}) if out_tr.get("text"): transcript_parts.append(out_tr["text"]) if sc.get("turnComplete"): break if sc.get("generationComplete"): gen_done = True audio_bytes = b"".join(audio_chunks) self.last_output_transcript = "".join(transcript_parts).strip() if audio_bytes: await bus.emit("voice.gemini_spoke", audio_len=len(audio_bytes)) return audio_bytes, text_parts def acquire_session(self, owner: str) -> "_SessionGuard": """Return an async context manager for exclusive session ownership. Use as `async with client.acquire_session("live_voice"):`. While held, no other consumer may call send_text or receive_stream. """ return _SessionGuard(self, owner) async def receive_stream(self): """Yield server events. Caller MUST hold the session lock.""" if self._owner is None: raise RuntimeError( "receive_stream requires session lock — use acquire_session() first" ) if not self._connected or self._ws is None: return try: async for raw in self._ws: try: resp = json.loads(raw) except json.JSONDecodeError: continue yield resp.get("serverContent", {}) except websockets.exceptions.ConnectionClosed: log.warning("receive_stream: connection closed") self._connected = False await bus.emit("voice.error", reason="connection_closed") async def raw_send(self, payload: dict): """Low-level send for the live loop. Always use through send lock.""" if not self._connected or self._ws is None: return False try: async with self._send_lock: await asyncio.wait_for( self._ws.send(json.dumps(payload)), timeout=GEMINI_WS_TIMEOUT) return True except Exception: log.exception("raw_send failed") return False async def synthesize_tts(self, text: str, voice: str | None = None) -> bytes: """Text-to-speech via the dedicated Gemini TTS model (stateless REST). Returns raw PCM bytes (24 kHz, 16-bit, mono). Reliable audio-only output — unlike the Live native-audio model, which frequently returns its reasoning text instead of speech for short prompts. Voice defaults to the current GEMINI_VOICE (hot-swappable from the dashboard). Runs the blocking HTTP call in a worker thread so the event loop stays free. """ text = (text or "").strip() if not text: return b"" if not GEMINI_API_KEY: raise RuntimeError("No Gemini API key configured.") voice = voice or GEMINI_VOICE url = (f"{_GEMINI_REST_BASE}/models/{GEMINI_TTS_MODEL}" f":generateContent?key={GEMINI_API_KEY}") prompt = f"{_TTS_PREAMBLE}{text}" if _TTS_PREAMBLE else text speech_config = { "voiceConfig": {"prebuiltVoiceConfig": {"voiceName": voice}} } if _TTS_LANGUAGE_CODE: speech_config["languageCode"] = _TTS_LANGUAGE_CODE body = { "contents": [{"parts": [{"text": prompt}]}], "generationConfig": { "responseModalities": ["AUDIO"], "speechConfig": speech_config, }, } # Deterministic first (temperature 0) so the same text renders the same # audio → stable tone across generations. Only bump the temperature on a # retry if temp-0 returned an empty part for this input (some very short # inputs need it); those retries are the only ones whose tone can vary. temps = [_TTS_TEMPERATURE] + [_TTS_RETRY_TEMPERATURE] * max(0, _TTS_MAX_ATTEMPTS - 1) last_exc = None for attempt, temp in enumerate(temps, 1): body["generationConfig"]["temperature"] = temp try: audio = await asyncio.to_thread( _tts_rest_blocking, url, body, _TTS_TIMEOUT) except Exception as exc: reason = _quota_message(exc) if reason: log.error("Gemini REST TTS refused: %s", reason) raise GeminiQuotaExhausted(reason) from exc last_exc = exc log.warning("TTS attempt %d/%d (temp=%.1f) failed: %s", attempt, len(temps), temp, exc) continue if audio: if attempt > 1: log.info("TTS succeeded on attempt %d (temp=%.1f)", attempt, temp) return audio log.warning("TTS attempt %d/%d (temp=%.1f) returned no audio", attempt, len(temps), temp) if last_exc is not None: raise last_exc return b"" def status(self) -> dict[str, Any]: return { "connected": self._connected, "model": GEMINI_MODEL, "voice": GEMINI_VOICE, "session_owner": self._owner, "reconnect_attempts": self._reconnect_attempts, } class _SessionGuard: """Async context manager for exclusive session ownership. Always releases owner + lock on exit, even on exceptions. """ def __init__(self, client: GeminiVoiceClient, owner: str): self._client = client self._owner = owner self._held = False async def __aenter__(self): await self._client._session_lock.acquire() self._held = True self._client._owner = self._owner return self._client async def __aexit__(self, exc_type, exc, tb): try: self._client._owner = None finally: if self._held: self._client._session_lock.release() self._held = False return False # don't suppress exceptions