From 07b1f1e2d6721530f82c4a33dd8374d5af7248df Mon Sep 17 00:00:00 2001 From: kassam Date: Thu, 9 Jul 2026 17:48:38 +0400 Subject: [PATCH] Update 2026-07-09 17:48:37 --- config/voice_config.json | 26 ++- dashboard/routes/_safe_io.py | 32 ++- dashboard/routes/audio_control.py | 47 ++++- dashboard/routes/live_subprocess.py | 231 +++++++++++++++++++++ dashboard/routes/voice.py | 6 +- dashboard/static/index.html | 94 ++++++++- gemini/script.py | 304 ++++++++++++++++++++++++++-- gemini/subprocess.py | 16 +- vision/camera.py | 41 ++++ voice/audio_devices.py | 77 ++++++- voice/audio_io.py | 111 +++++++++- voice/movement_dispatch.py | 5 + voice/sanad_voice.py | 3 +- 13 files changed, 932 insertions(+), 61 deletions(-) diff --git a/config/voice_config.json b/config/voice_config.json index 00e151d..ac683eb 100644 --- a/config/voice_config.json +++ b/config/voice_config.json @@ -1,6 +1,5 @@ { "_description": "Tunables for voice/* modules. Loaded via core.config_loader.load('voice').", - "sanad_voice": { "_comment": "voice/sanad_voice.py — main live voice subprocess. Gemini API credentials (api_key, model, voice_name) come from core_config.json's gemini_defaults — single source of truth.", "mic_gain": 1.0, @@ -10,9 +9,10 @@ "session_timeout_sec": 660, "max_reconnect_delay_sec": 30, "max_consecutive_errors": 10, - "no_messages_timeout_sec": 30 + "no_messages_timeout_sec": 180, + "pure_voice": false, + "internet_search": false }, - "mic_udp": { "_comment": "G1 built-in mic — UDP multicast subscriber", "group": "239.168.123.161", @@ -21,41 +21,40 @@ "read_timeout_sec": 0.04, "socket_timeout_sec": 1.0 }, - "speaker": { "_comment": "G1 built-in speaker — AudioClient.PlayStream wrapper", "app_name": "sanad", "begin_stream_pause_sec": 0.15, - "wait_finish_margin_sec": 0.3 + "wait_finish_margin_sec": 0.3, + "jitter_prebuffer_sec": 0.3, + "jitter_first_flush_sec": 0.6, + "jitter_max_lead_sec": 0.6 }, - "vad": { "_comment": "Gemini Live server-side voice-activity-detection config", "start_sensitivity": "START_SENSITIVITY_HIGH", "end_sensitivity": "END_SENSITIVITY_LOW", "prefix_padding_ms": 20, - "silence_duration_ms": 200 + "silence_duration_ms": 200, + "direct_answer": false }, - "barge_in": { "threshold": 500, "loud_chunks_needed": 3, "cooldown_sec": 0.3, "echo_suppress_below": 500, - "ai_speak_grace_sec": 0.15 + "ai_speak_grace_sec": 0.15, + "external_barge_in": false }, - "recording": { "enabled": true, "dir_relative": "data/recordings" }, - "typed_replay": { "_comment": "voice/typed_replay.py — max_text_len comes from dashboard.api_input", "monitor_chunk_size": 512, "monitor_tail_sec": 0.2 }, - "live_voice_loop": { "_comment": "voice/live_voice_loop.py — arm phrase dispatcher. arm_txt filename comes from core.script_files.arm_phrases", "trigger_log_size": 100, @@ -63,7 +62,6 @@ "deferred_default": false, "trigger_enabled_default": false }, - "local_tts": { "_comment": "voice/local_tts.py — offline Coqui TTS", "model_subdir": "speecht5_tts_clartts_ar", @@ -72,4 +70,4 @@ "sample_rate": 16000, "channels": 1 } -} +} \ No newline at end of file diff --git a/dashboard/routes/_safe_io.py b/dashboard/routes/_safe_io.py index 633ea7c..6bb53fd 100644 --- a/dashboard/routes/_safe_io.py +++ b/dashboard/routes/_safe_io.py @@ -58,13 +58,43 @@ def check_upload_size(content: bytes, max_bytes: int = MAX_UPLOAD_BYTES) -> None def atomic_write_bytes(path: Path, data: bytes) -> None: - """Write bytes atomically via tempfile + os.replace.""" + """Write bytes atomically via tempfile + os.replace. + + Fallback: a bind-MOUNTED single file (docker `-v host.txt:/app/x.txt`) + cannot be renamed over — os.replace raises EBUSY (errno 16) because the + mount pins the inode. In that case fall back to an in-place truncate+write + (not atomic, but the only way to persist through a file mount).""" path.parent.mkdir(parents=True, exist_ok=True) + # Carry the target's mode/owner onto the replacement — mkstemp creates + # 0600 root, and dashboards run as root in containers over host-owned + # bind mounts: without this every save locks the host user out (root:600). + try: + st = path.stat() + except OSError: + st = None fd, tmp = tempfile.mkstemp(prefix=f".{path.name}.", suffix=".tmp", dir=str(path.parent)) try: with os.fdopen(fd, "wb") as f: f.write(data) + try: + if st is not None: + os.chmod(tmp, st.st_mode & 0o7777) + os.chown(tmp, st.st_uid, st.st_gid) + else: + os.chmod(tmp, 0o664) + except OSError: + pass # chown needs root; keep the write even if it fails os.replace(tmp, path) + except OSError as exc: + try: + os.unlink(tmp) + except OSError: + pass + if exc.errno == 16: # EBUSY — path is a bind-mounted file + with open(path, "wb") as f: + f.write(data) + return + raise except Exception: try: os.unlink(tmp) diff --git a/dashboard/routes/audio_control.py b/dashboard/routes/audio_control.py index cd6f02e..82abe2f 100644 --- a/dashboard/routes/audio_control.py +++ b/dashboard/routes/audio_control.py @@ -627,6 +627,37 @@ def _kill_audio_daemon(flavour: str) -> dict: "stderr": "timeout (>5s)"} +def _remote_pulse_mode() -> bool: + """True when this process talks to ANOTHER machine-scope PulseAudio — + the containerized deployments (P1–P4) run as root with PULSE_SERVER + pointed at the host's socket. Restarting that daemon is not ours to do.""" + return bool(os.environ.get("PULSE_SERVER")) or os.path.exists("/.dockerenv") + + +def _cycle_remote_pulse() -> dict: + """Suspend→resume every sink/source on the remote (host) PulseAudio. + A suspend cycle forces the ALSA device rebind and clears wedged + streams — the useful part of a daemon restart, minus the restart.""" + cycled: list[str] = [] + for kind, flag in (("sinks", "suspend-sink"), ("sources", "suspend-source")): + try: + r = subprocess.run(["pactl", "list", "short", kind], check=False, + capture_output=True, text=True, timeout=5.0) + names = [ln.split("\t")[1] for ln in (r.stdout or "").splitlines() + if "\t" in ln] + except (FileNotFoundError, subprocess.SubprocessError): + names = [] + for name in names: + if name.endswith(".monitor"): + continue + for state in ("1", "0"): + subprocess.run(["pactl", flag, name, state], check=False, + capture_output=True, text=True, timeout=5.0) + cycled.append(name) + return {"cmd": "remote pulse suspend/resume cycle", "returncode": 0, + "stderr": "", "cycled": cycled} + + def _wait_for_pactl(deadline_s: float = 5.0, interval_s: float = 0.2) -> bool: """Poll `pactl info` until it returns 0 or the deadline expires.""" import time as _time @@ -646,7 +677,8 @@ async def reset_audio_subsystem(): is being selected. **Does NOT recover a kernel-side missing USB capture descriptor** — for that symptom use /api/audio/usb-reset. """ - if os.geteuid() == 0: + remote_pulse = _remote_pulse_mode() + if os.geteuid() == 0 and not remote_pulse: raise HTTPException( 403, "Refusing to reset audio as root — Sanad must run as the " "unitree user so the per-user PulseAudio session is reachable.", @@ -687,9 +719,16 @@ async def reset_audio_subsystem(): if acquired and play_lock is not None: play_lock.release() - flavour = _detect_pa_flavour() - kill_info = _kill_audio_daemon(flavour) - came_back = _wait_for_pactl(deadline_s=5.0) + if remote_pulse: + # Containerized: the daemon is the HOST's — never kill it. + # Cycle its devices over the socket instead. + flavour = "pulse-remote" + kill_info = _cycle_remote_pulse() + came_back = _wait_for_pactl(deadline_s=5.0) + else: + flavour = _detect_pa_flavour() + kill_info = _kill_audio_daemon(flavour) + came_back = _wait_for_pactl(deadline_s=5.0) if not came_back and flavour == "pulse": # autospawn may be disabled — try an explicit start. try: diff --git a/dashboard/routes/live_subprocess.py b/dashboard/routes/live_subprocess.py index 43e1610..636c24d 100644 --- a/dashboard/routes/live_subprocess.py +++ b/dashboard/routes/live_subprocess.py @@ -41,6 +41,237 @@ async def set_record(on: bool = Query(...)): return {"ok": True, "record_enabled": st.record_enabled} +_VOICE_OPTS_PATH = BASE_DIR / "data" / ".voice_options.json" + + +def _read_voice_opts() -> dict: + import json + try: + with open(_VOICE_OPTS_PATH, encoding="utf-8") as f: + d = json.load(f) + return d if isinstance(d, dict) else {} + except Exception: + return {} + + +@router.get("/interrupt-option") +async def get_interrupt_option(): + """Voice barge-in on external speakers (JBL / Beats Pill). Resolution: + dashboard toggle (data/.voice_options.json) > SANAD_EXTERNAL_BARGE_IN env + > voice_config.json barge_in.external_barge_in > off.""" + import os + opts = _read_voice_opts() + if "external_barge_in" in opts: + return {"enabled": bool(opts["external_barge_in"]), "source": "dashboard"} + env = os.environ.get("SANAD_EXTERNAL_BARGE_IN", "").strip().lower() + if env in ("1", "true", "yes"): + return {"enabled": True, "source": "env"} + try: + from Project.Sanad.core.config_loader import section + if bool((section("voice", "barge_in") or {}).get("external_barge_in", False)): + return {"enabled": True, "source": "config"} + except Exception: + pass + return {"enabled": False, "source": "default"} + + +@router.post("/interrupt-option") +async def set_interrupt_option(on: bool = Query(...)): + """Persist the interrupt option (survives rebuilds — data/ is mounted) and + restart the live session so the voice child picks it up.""" + import json + import tempfile + import os as _os + opts = _read_voice_opts() + opts["external_barge_in"] = bool(on) + _VOICE_OPTS_PATH.parent.mkdir(parents=True, exist_ok=True) + fd, tmp = tempfile.mkstemp(dir=str(_VOICE_OPTS_PATH.parent), + prefix=".voice_options.", suffix=".tmp") + with _os.fdopen(fd, "w", encoding="utf-8") as f: + json.dump(opts, f, indent=2) + _os.replace(tmp, _VOICE_OPTS_PATH) + + restarted = False + from Project.Sanad.main import live_sub + if live_sub is not None: + try: + if (live_sub.status() or {}).get("running"): + await asyncio.to_thread(live_sub.stop) + await asyncio.to_thread(live_sub.start) + restarted = True + except Exception: + pass + return {"ok": True, "enabled": bool(on), "restarted": restarted} + + +@router.get("/direct-answer-option") +async def get_direct_answer_option(): + """'Direct answer' = Gemini replies as soon as the speaker pauses (Live-API + end-of-speech sensitivity HIGH). Resolution: dashboard > env > config.""" + import os + opts = _read_voice_opts() + if "direct_answer" in opts: + return {"enabled": bool(opts["direct_answer"]), "source": "dashboard"} + if os.environ.get("SANAD_DIRECT_ANSWER", "").strip().lower() in ("1", "true", "yes"): + return {"enabled": True, "source": "env"} + try: + from Project.Sanad.core.config_loader import section + if bool((section("voice", "vad") or {}).get("direct_answer", False)): + return {"enabled": True, "source": "config"} + except Exception: + pass + return {"enabled": False, "source": "default"} + + +@router.post("/direct-answer-option") +async def set_direct_answer_option(on: bool = Query(...)): + """Persist the direct-answer option and restart the live session.""" + import json + import tempfile + import os as _os + opts = _read_voice_opts() + opts["direct_answer"] = bool(on) + _VOICE_OPTS_PATH.parent.mkdir(parents=True, exist_ok=True) + fd, tmp = tempfile.mkstemp(dir=str(_VOICE_OPTS_PATH.parent), + prefix=".voice_options.", suffix=".tmp") + with _os.fdopen(fd, "w", encoding="utf-8") as f: + json.dump(opts, f, indent=2) + _os.replace(tmp, _VOICE_OPTS_PATH) + restarted = False + from Project.Sanad.main import live_sub + if live_sub is not None: + try: + if (live_sub.status() or {}).get("running"): + await asyncio.to_thread(live_sub.stop) + await asyncio.to_thread(live_sub.start) + restarted = True + except Exception: + pass + return {"ok": True, "enabled": bool(on), "restarted": restarted} + + +@router.get("/pure-voice-option") +async def get_pure_voice_option(): + """PURE VOICE = the live session built exactly like standalone Sanadv1: + no tools, no state injections, persona only, no lip-sync markers — the + fastest conversation pipe. Resolution: dashboard > env > config.""" + import os + opts = _read_voice_opts() + if "pure_voice" in opts: + return {"enabled": bool(opts["pure_voice"]), "source": "dashboard"} + if os.environ.get("SANAD_PURE_VOICE", "").strip().lower() in ("1", "true", "yes"): + return {"enabled": True, "source": "env"} + try: + from Project.Sanad.core.config_loader import section + if bool((section("voice", "sanad_voice") or {}).get("pure_voice", False)): + return {"enabled": True, "source": "config"} + except Exception: + pass + return {"enabled": False, "source": "default"} + + +@router.post("/pure-voice-option") +async def set_pure_voice_option(on: bool = Query(...)): + """Persist the pure-voice option and restart the live session.""" + import json + import tempfile + import os as _os + opts = _read_voice_opts() + opts["pure_voice"] = bool(on) + _VOICE_OPTS_PATH.parent.mkdir(parents=True, exist_ok=True) + fd, tmp = tempfile.mkstemp(dir=str(_VOICE_OPTS_PATH.parent), + prefix=".voice_options.", suffix=".tmp") + with _os.fdopen(fd, "w", encoding="utf-8") as f: + json.dump(opts, f, indent=2) + _os.replace(tmp, _VOICE_OPTS_PATH) + restarted = False + from Project.Sanad.main import live_sub + if live_sub is not None: + try: + if (live_sub.status() or {}).get("running"): + await asyncio.to_thread(live_sub.stop) + await asyncio.to_thread(live_sub.start) + restarted = True + except Exception: + pass + return {"ok": True, "enabled": bool(on), "restarted": restarted} + + +@router.get("/internet-search-option") +async def get_internet_search_option(): + """INTERNET SEARCH = the live session gets the search_internet tool so Gemini + can look up current facts (prices, latest phones, live offers, news) via + Google Search grounding, announcing the search first. Resolution: + dashboard > env > config.""" + import os + opts = _read_voice_opts() + if "internet_search" in opts: + return {"enabled": bool(opts["internet_search"]), "source": "dashboard"} + if os.environ.get("SANAD_INTERNET_SEARCH", "").strip().lower() in ("1", "true", "yes"): + return {"enabled": True, "source": "env"} + try: + from Project.Sanad.core.config_loader import section + if bool((section("voice", "sanad_voice") or {}).get("internet_search", False)): + return {"enabled": True, "source": "config"} + except Exception: + pass + return {"enabled": False, "source": "default"} + + +@router.post("/internet-search-option") +async def set_internet_search_option(on: bool = Query(...)): + """Persist the internet-search option and restart the live session so the + tool set (and persona addendum) is rebaked.""" + import json + import tempfile + import os as _os + opts = _read_voice_opts() + opts["internet_search"] = bool(on) + _VOICE_OPTS_PATH.parent.mkdir(parents=True, exist_ok=True) + fd, tmp = tempfile.mkstemp(dir=str(_VOICE_OPTS_PATH.parent), + prefix=".voice_options.", suffix=".tmp") + with _os.fdopen(fd, "w", encoding="utf-8") as f: + json.dump(opts, f, indent=2) + _os.replace(tmp, _VOICE_OPTS_PATH) + restarted = False + from Project.Sanad.main import live_sub + if live_sub is not None: + try: + if (live_sub.status() or {}).get("running"): + await asyncio.to_thread(live_sub.stop) + await asyncio.to_thread(live_sub.start) + restarted = True + except Exception: + pass + return {"ok": True, "enabled": bool(on), "restarted": restarted} + + +@router.post("/answer-now") +async def answer_now(): + """Force Gemini to reply immediately instead of waiting out the pause + detection (no-op while it is already speaking).""" + sub = _sub_or_503() + if not (sub.status() or {}).get("running"): + raise HTTPException(409, "Live session not running") + if not hasattr(sub, "send_answer_now"): + raise HTTPException(501, "engine build lacks send_answer_now") + await asyncio.to_thread(sub.send_answer_now) + return {"ok": True} + + +@router.post("/interrupt") +async def interrupt_speech(): + """Instantly stop the AI mid-sentence and go back to listening (works + regardless of the barge-in option; no-op if it isn't speaking).""" + sub = _sub_or_503() + if not (sub.status() or {}).get("running"): + raise HTTPException(409, "Live session not running") + if not hasattr(sub, "send_interrupt"): + raise HTTPException(501, "engine build lacks send_interrupt") + await asyncio.to_thread(sub.send_interrupt) + return {"ok": True} + + @router.post("/start") async def start_subprocess(): live_sub = _sub_or_503() diff --git a/dashboard/routes/voice.py b/dashboard/routes/voice.py index a3d1b88..8edf2f3 100644 --- a/dashboard/routes/voice.py +++ b/dashboard/routes/voice.py @@ -173,10 +173,12 @@ async def update_api_key(payload: ApiKeyPayload): raise HTTPException(400, "API key cannot be empty.") if len(key) < 20: raise HTTPException(400, "API key looks too short.") - if not key.startswith("AIza"): + # Google AI Studio issues classic "AIza..." keys and (since late 2025) + # new-format "AQ...." keys — both are plain x-goog-api-key API keys. + if not key.startswith(("AIza", "AQ.")): raise HTTPException( 400, - "Gemini API keys normally start with 'AIza'. " + "Gemini API keys start with 'AIza' (classic) or 'AQ.' (new format). " "Double-check you're pasting a Google AI Studio key.", ) diff --git a/dashboard/static/index.html b/dashboard/static/index.html index 50dac2c..14b2c84 100644 --- a/dashboard/static/index.html +++ b/dashboard/static/index.html @@ -302,7 +302,7 @@
- + –
@@ -314,7 +314,7 @@
- Controls the ACTIVE speaker — G1 chest (DDS) + the selected PulseAudio sink (JBL/Anker). Applies live. + Controls the ACTIVE speaker — G1 chest (DDS) + the selected PulseAudio sink (JBL / Anker / Beats Pill / any plugged). Applies live.
@@ -405,9 +405,15 @@
+ + + Pure: … + Direct: … + Interrupt: … + Search: … Pause: Auto @@ -2068,6 +2074,90 @@ function updateLiveHoldUI(hold){ pm.className='badge '+(hold?'badge-warn':'badge-ok'); } } +let _pureOpt=null; +function _paintPureBadge(){ + const b=document.getElementById('ls-pure'); + if(!b)return; + b.textContent='Pure: '+(_pureOpt===null?'…':(_pureOpt?'ON':'OFF')); + b.className='badge '+(_pureOpt?'badge-ok':''); +} +async function refreshPureOpt(){ + try{const r=await api('GET','/api/live-subprocess/pure-voice-option');_pureOpt=!!r.enabled;_paintPureBadge();}catch(e){} +} +async function togglePureOpt(){ + const want=!_pureOpt; + try{ + const r=await api('POST','/api/live-subprocess/pure-voice-option?on='+want); + _pureOpt=!!r.enabled;_paintPureBadge(); + toast('Pure voice '+(_pureOpt?'ON — Sanadv1-style session':'OFF — full features')+(r.restarted?' (session restarted)':'')); + }catch(e){toast('Pure toggle failed: '+(e.message||e));} +} +setTimeout(refreshPureOpt,1000); +let _searchOpt=null; +function _paintSearchBadge(){ + const b=document.getElementById('ls-search'); + if(!b)return; + b.textContent='Search: '+(_searchOpt===null?'…':(_searchOpt?'ON':'OFF')); + b.className='badge '+(_searchOpt?'badge-ok':''); +} +async function refreshSearchOpt(){ + try{const r=await api('GET','/api/live-subprocess/internet-search-option');_searchOpt=!!r.enabled;_paintSearchBadge();}catch(e){} +} +async function toggleSearchOpt(){ + const want=!_searchOpt; + try{ + const r=await api('POST','/api/live-subprocess/internet-search-option?on='+want); + _searchOpt=!!r.enabled;_paintSearchBadge(); + toast('Internet search '+(_searchOpt?'ON — Sanad can look up current info':'OFF — pure voice, no search')+(r.restarted?' (session restarted)':'')); + }catch(e){toast('Search toggle failed: '+(e.message||e));} +} +setTimeout(refreshSearchOpt,1050); +let _dirOpt=null; +function _paintDirectBadge(){ + const b=document.getElementById('ls-direct'); + if(!b)return; + b.textContent='Direct: '+(_dirOpt===null?'…':(_dirOpt?'ON':'OFF')); + b.className='badge '+(_dirOpt?'badge-ok':''); +} +async function refreshDirectOpt(){ + try{const r=await api('GET','/api/live-subprocess/direct-answer-option');_dirOpt=!!r.enabled;_paintDirectBadge();}catch(e){} +} +async function toggleDirectOpt(){ + const want=!_dirOpt; + try{ + const r=await api('POST','/api/live-subprocess/direct-answer-option?on='+want); + _dirOpt=!!r.enabled;_paintDirectBadge(); + toast('Direct answer '+(_dirOpt?'ON':'OFF')+(r.restarted?' — live session restarted':'')); + }catch(e){toast('Direct toggle failed: '+(e.message||e));} +} +setTimeout(refreshDirectOpt,950); +let _intOpt=null; +function _paintInterruptBadge(){ + const b=document.getElementById('ls-interrupt'); + if(!b)return; + b.textContent='Interrupt: '+(_intOpt===null?'…':(_intOpt?'ON':'OFF')); + b.className='badge '+(_intOpt?'badge-ok':''); +} +async function refreshInterruptOpt(){ + try{const r=await api('GET','/api/live-subprocess/interrupt-option');_intOpt=!!r.enabled;_paintInterruptBadge();}catch(e){} +} +async function answerNow(){ + try{await api('POST','/api/live-subprocess/answer-now');toast('Answering now…');} + catch(e){toast('Answer-now failed: '+(e.message||e));} +} +async function instantInterrupt(){ + try{await api('POST','/api/live-subprocess/interrupt');toast('Interrupted — listening');} + catch(e){toast('Interrupt failed: '+(e.message||e));} +} +async function toggleInterruptOpt(){ + const want=!_intOpt; + try{ + const r=await api('POST','/api/live-subprocess/interrupt-option?on='+want); + _intOpt=!!r.enabled;_paintInterruptBadge(); + toast('Interrupt '+(_intOpt?'ON':'OFF')+(r.restarted?' — live session restarted':'')); + }catch(e){toast('Interrupt toggle failed: '+(e.message||e));} +} +setTimeout(refreshInterruptOpt,900); async function refreshPlaybackStatus(){ try{ const s=await api('GET','/api/records/playback-status'); diff --git a/gemini/script.py b/gemini/script.py index e745a21..4940792 100644 --- a/gemini/script.py +++ b/gemini/script.py @@ -55,11 +55,23 @@ _MODEL = os.environ.get( "SANAD_GEMINI_MODEL", "gemini-2.5-flash-native-audio-preview-12-2025", ) -_MIC_GAIN = _SV.get("mic_gain", 1.0) +# Mic gain: amplifies mic samples before they reach Gemini — the single +# biggest lever for slow/failed speech detection at a distance (weak audio +# transcribes but often yields NO reply turn). Env override for per-robot +# tuning; watch for clipping if the speaker stands close (>3.0 rarely wise). +try: + _MIC_GAIN = float(os.environ.get("SANAD_MIC_GAIN", "") or _SV.get("mic_gain", 1.0)) +except ValueError: + _MIC_GAIN = _SV.get("mic_gain", 1.0) _SESSION_TIMEOUT = _SV.get("session_timeout_sec", 660) _MAX_RECONNECT_DELAY = _SV.get("max_reconnect_delay_sec", 30) _MAX_CONSECUTIVE_ERRORS = _SV.get("max_consecutive_errors", 10) -_NO_MESSAGES_TIMEOUT = _SV.get("no_messages_timeout_sec", 30) +# How long with ZERO messages from Gemini before we declare the session dead. +# In a QUIET room Gemini legitimately sends nothing — 30s was killing healthy +# sessions every ~40s (reconnect churn + unsolicited greeting + a 1-3s deaf +# window each time). REAL deaths are caught fast by the send path ("mic send +# failed: 1011 ..."), so this can be patient. +_NO_MESSAGES_TIMEOUT = _SV.get("no_messages_timeout_sec", 180) # Extra mic-gate time after the AI stops, on loud external-speaker profiles # (JBL) — covers the speaker buffer + room reverb so it doesn't hear its tail. _ECHO_TAIL_SEC = _SV.get("echo_tail_sec", 0.6) @@ -77,6 +89,67 @@ _JBL_BARGE_CHUNKS = _SV.get("jbl_barge_chunks", 9) # FADED (a gap between words/numbers). In that window barge-in drops to a low, # sensitive bar so the user can interrupt; while audio is flowing it stays high. _JBL_BLEED_FADE_SEC = _SV.get("jbl_bleed_fade_sec", 0.5) +# OPTION: allow voice barge-in on loud external speakers (JBL / Beats Pill). +# The mic feed to Gemini STAYS fully gated while the AI speaks (echo-safe); +# this only enables the LOCAL energy detector to cut the speaker, using the +# stricter external parameters (threshold×margin, longer grace, more sustained +# chunks) so speaker bleed can't false-trigger. Env > config; default OFF. +_EXTERNAL_BARGE_IN = ( + os.environ.get("SANAD_EXTERNAL_BARGE_IN", "").strip().lower() in ("1", "true", "yes") + or bool(_BI.get("external_barge_in", False)) +) +# OPTION: "direct answer" — reply as soon as the speaker pauses. Flips the +# Live-API end-of-speech sensitivity from LOW (conservative: waits to be sure +# you finished, feels slow) to HIGH (immediate turn-taking; may cut off very +# hesitant speakers mid-thought). Env > config; dashboard file has final say. +_DIRECT_ANSWER = ( + os.environ.get("SANAD_DIRECT_ANSWER", "").strip().lower() in ("1", "true", "yes") + or bool(_VAD.get("direct_answer", False)) +) + +# OPTION: PURE VOICE — make the live session EXACTLY like standalone Sanadv1: +# no tools (no function-calling deliberation), no state injections, persona +# only (no face addendum), no lip-sync markers, no frame/state loops. The +# fastest possible conversation pipe; mask/nav/vision hooks are dormant. +_PURE_VOICE = ( + os.environ.get("SANAD_PURE_VOICE", "").strip().lower() in ("1", "true", "yes") + or bool(_SV.get("pure_voice", False)) +) + +# OPTION: INTERNET SEARCH (Google Search grounding). When ON, the live session +# gets ONE extra tool, `search_internet(query)`, so the model can look up +# CURRENT facts (today's prices, the newest phones, live Etisalat offers, news). +# The tool description tells the model to SAY it will check first, then call the +# tool — so the robot announces "let me search the internet", the parent runs +# the grounded generateContent call in a worker thread, and the model speaks the +# result. Default OFF → pure voice stays tool-free and Sanadv1-fast; only when +# a search actually fires does the (announced) grounding delay occur. +_INTERNET_SEARCH = ( + os.environ.get("SANAD_INTERNET_SEARCH", "").strip().lower() in ("1", "true", "yes") + or bool(_SV.get("internet_search", False)) +) +# Model used for the grounded web-search call (generateContent, NOT the live +# native-audio model). gemini-3.5-flash grounds well and cites the source site. +_SEARCH_MODEL = os.environ.get("SANAD_SEARCH_MODEL", "gemini-3.5-flash") + +# The dashboard toggles (data/.voice_options.json, written by the +# /api/live-subprocess/*-option endpoints, which then restart this child) +# have the FINAL say over env/config. +try: + import json as _vo_json + with open(BASE_DIR / "data" / ".voice_options.json", encoding="utf-8") as _vo_f: + _vo = _vo_json.load(_vo_f) + if isinstance(_vo, dict): + if "external_barge_in" in _vo: + _EXTERNAL_BARGE_IN = bool(_vo["external_barge_in"]) + if "direct_answer" in _vo: + _DIRECT_ANSWER = bool(_vo["direct_answer"]) + if "pure_voice" in _vo: + _PURE_VOICE = bool(_vo["pure_voice"]) + if "internet_search" in _vo: + _INTERNET_SEARCH = bool(_vo["internet_search"]) +except Exception: + pass _CHUNK_BYTES = CHUNK_SIZE * 2 _SILENCE_PCM = b"\x00" * _CHUNK_BYTES @@ -88,6 +161,15 @@ _SILENCE_PCM = b"\x00" * _CHUNK_BYTES # safe to read from the asyncio loops. _INPUT_PAUSED = threading.Event() +# Set by an "interrupt:" stdin command (the dashboard's Interrupt-now button). +# The mic loop consumes it: stops the speaker mid-sentence, back to listening. +_MANUAL_INTERRUPT = threading.Event() + +# Set by an "answernow:" stdin command (the dashboard's Answer-now button). +# The mic loop feeds ~0.6s of silence so the server VAD closes the user's turn +# immediately → Gemini answers right away instead of waiting out the pause. +_FORCE_ANSWER = threading.Event() + # ── Recognition (camera + face gallery) tunables ── _RECOG_STATE_PATH = Path(os.environ.get( "SANAD_RECOGNITION_STATE_PATH", @@ -195,6 +277,72 @@ def _nav_function_declarations() -> list: ] +def _search_function_declaration() -> list: + """Live tool: search the internet (Google Search grounding) for current facts.""" + S, T = types.Schema, types.Type + return [ + types.FunctionDeclaration( + name="search_internet", + description=( + "Search the LIVE internet for CURRENT, real-time facts you cannot " + "otherwise know: today's prices, the newest phone / device models, " + "current Etisalat / e& offers and promotions, product availability, " + "or recent news. IMPORTANT: BEFORE you call this, first SAY one " + "short sentence out loud telling the user you will check — e.g. " + "\"Let me search the internet for that.\" / " + "\"دعني أبحث لك على الإنترنت.\" — " + "then call this with a clear English search query. After the " + "result returns, answer the user naturally from it. Do NOT say the " + "tool name out loud. Only use this for genuinely current / " + "time-sensitive questions, not for things you already know." + ), + parameters=S(type=T.OBJECT, properties={ + "query": S(type=T.STRING, + description="The web-search query, phrased in English."), + }, required=["query"]), + ), + ] + + +# Appended to the persona ONLY when internet search is enabled, so the model +# knows it can (and how to) look things up. Kept short — the tool description +# carries the detail. +_SEARCH_PROMPT_ADDENDUM = ( + "\n\nYou can look things up on the live internet with the search_internet " + "tool. Use it whenever the user asks about current prices, the latest phones " + "or devices, today's Etisalat/e& offers, availability, or recent news — " + "anything you can't be certain is up to date. ALWAYS say a short 'let me " + "check the internet' line first (in the user's language), then call the tool, " + "then answer from the result. Keep answers concise and, for exact prices, add " + "that a sales advisor can confirm the final figure. " + "CRITICAL: after the search result returns, CONTINUE where you left off — " + "NEVER repeat or re-say any sentence you already spoke, including the 'let me " + "check the internet' lead-in. If your reply is complete, stop and wait." +) + + +def _run_grounding_search(query: str, api_key: str) -> dict: + """Blocking Google-Search-grounded generateContent call (run in a thread). + Returns {ok, answer} for the model to speak, or {ok: False, reason}.""" + try: + client = genai.Client(api_key=api_key) + resp = client.models.generate_content( + model=_SEARCH_MODEL, + contents=query, + config=types.GenerateContentConfig( + tools=[types.Tool(google_search=types.GoogleSearch())], + ), + ) + answer = (getattr(resp, "text", "") or "").strip() + if not answer: + return {"ok": False, "reason": "empty_result"} + # Cap length so the spoken reply stays reasonable. + return {"ok": True, "answer": answer[:1500]} + except Exception as exc: + log.warning("internet search failed: %s", exc) + return {"ok": False, "reason": "search_failed", "detail": str(exc)[:200]} + + # Emotions Gemini can show on the LED face (a subset of the mask's frames that # read as feelings — the talk/blink/gaze frames are driven automatically). _FACE_EMOTIONS = ("smile", "laugh", "heart", "love", "sad", "surprised", @@ -221,7 +369,10 @@ _FACE_PROMPT_ADDENDUM = ( "Instagram, or to see/show your social media, ALWAYS call show_social with " "'bu_sunaidah' (@bu.sunaidah) or 'yslootahtech' (@yslootahtech) to display " "the QR on your face. These tools are silent — never say the tool name, the " - "emotion, or any bracket marker out loud." + "emotion, or any bracket marker out loud. CRITICAL: after any tool returns, " + "CONTINUE exactly where you left off — NEVER repeat or re-say sentences you " + "already spoke before the tool call. If your reply was already complete, do " + "not speak again; wait silently for the user." ) @@ -239,7 +390,8 @@ def _face_function_declarations() -> list: "surprised when astonished, confused when you didn't understand, wink " "when joking, sad when empathizing, cool when playful, sleepy when " "tired, angry only rarely. Your mouth already lip-syncs on its own — " - "this is ONLY the emotion, not the mouth." + "this is ONLY the emotion, not the mouth. After this tool returns, never " + "repeat what you already said — continue your sentence or stay silent." ), parameters=S(type=T.OBJECT, properties={ "emotion": S(type=T.STRING, enum=list(_FACE_EMOTIONS), @@ -301,7 +453,7 @@ _PROFILE_LOCK = threading.Lock() _PROFILE_PENDING: dict = {"id": None, "reason": ""} _VALID_PROFILES = {"builtin", "anker", "anker_powerconf", - "hollyland_builtin", "jbl_builtin_mic"} + "hollyland_builtin", "jbl_builtin_mic", "beats_pill"} def _stdin_watcher() -> None: @@ -334,6 +486,10 @@ def _stdin_watcher() -> None: with _LATEST_FRAME_LOCK: _LATEST_FRAME["bytes"] = data _LATEST_FRAME["ts"] = time.time() + elif line.startswith("interrupt:"): + _MANUAL_INTERRUPT.set() + elif line.startswith("answernow:"): + _FORCE_ANSWER.set() elif line.startswith("state:"): try: payload = json.loads(line[len("state:"):]) @@ -417,7 +573,13 @@ class GeminiBrain: self._swap_lock: Optional[asyncio.Lock] = None # built in run() self._recorder = recorder self._voice = voice_name or GEMINI_VOICE - self._system_prompt = (system_prompt or "") + _FACE_PROMPT_ADDENDUM + # PURE VOICE: persona only — no face/tool addendum (Sanadv1 parity). + self._system_prompt = ((system_prompt or "") if _PURE_VOICE + else (system_prompt or "") + _FACE_PROMPT_ADDENDUM) + # Internet-search addendum is independent of pure voice — the search + # tool is available in either mode when the option is enabled. + if _INTERNET_SEARCH: + self._system_prompt += _SEARCH_PROMPT_ADDENDUM self._api_key = GEMINI_API_KEY self._stop_flag = asyncio.Event() # per-session state (reset in the outer reconnect loop) @@ -511,6 +673,10 @@ class GeminiBrain: if self._swap_lock is None: self._swap_lock = asyncio.Lock() + # PURE VOICE is TOGGLE-AWARE: all loops run (cheap when + # their feature is off) — pure-ness is enforced per-feature + # (announce guards + tools) so enabling Camera/Movement in + # the dashboard works live, and disabling returns to pure. try: await asyncio.wait_for( asyncio.gather( @@ -573,10 +739,15 @@ class GeminiBrain: types.StartSensitivity, _VAD.get("start_sensitivity", "START_SENSITIVITY_HIGH"), ), - end_of_speech_sensitivity=getattr( - types.EndSensitivity, - _VAD.get("end_sensitivity", "END_SENSITIVITY_LOW"), - ), + # "Direct answer" option: HIGH = reply the moment the + # speaker pauses; otherwise the configured (LOW) default + # waits longer to be sure the speaker finished. + end_of_speech_sensitivity=( + types.EndSensitivity.END_SENSITIVITY_HIGH + if _DIRECT_ANSWER else getattr( + types.EndSensitivity, + _VAD.get("end_sensitivity", "END_SENSITIVITY_LOW"), + )), prefix_padding_ms=_VAD.get("prefix_padding_ms", 20), silence_duration_ms=_VAD.get("silence_duration_ms", 200), ), @@ -591,11 +762,28 @@ class GeminiBrain: # stop_navigation). Disable with SANAD_NAV_TOOLS=0. # Native function-calling: nav tools (if enabled) + the always-on # expressive-face / social-QR tools (set_expression / show_social). - tools=[types.Tool(function_declarations=( - (_nav_function_declarations() if _NAV_TOOLS_ENABLED else []) - + _face_function_declarations()))], + # PURE VOICE: no tools (the reply-latency tax) — EXCEPT nav tools + # when movement is enabled (walking commands need them). Face + # tools stay off in pure. + tools=self._build_tools(), ) + def _build_tools(self) -> Any: + """Assemble the live-session function tools. PURE VOICE keeps this None + (the reply-latency tax) UNLESS movement is on (nav tools) or internet + search is on (the search tool). Face tools stay off in pure voice.""" + decls: list = [] + # Nav tools: whenever NOT (pure and not moving) and nav is enabled. + if _NAV_TOOLS_ENABLED and (self._movement_enabled or not _PURE_VOICE): + decls += _nav_function_declarations() + # Face/social tools: only outside pure voice. + if not _PURE_VOICE: + decls += _face_function_declarations() + # Internet-search tool: whenever the option is enabled (either mode). + if _INTERNET_SEARCH: + decls += _search_function_declaration() + return [types.Tool(function_declarations=decls)] if decls else None + # ─── state helpers ──────────────────────────────────── def _reset_turn_state(self) -> None: @@ -635,6 +823,7 @@ class GeminiBrain: loop = asyncio.get_event_loop() loud_count = 0 last_activity = time.time() + force_silence_until = 0.0 # Answer-now button: silence burst window while not self._done.is_set() and not self._stop_flag.is_set(): try: @@ -649,6 +838,27 @@ class GeminiBrain: data = samples.tobytes() energy = _audio_energy(data) now = time.time() + + # Dashboard Answer-now button: feed the server VAD a silence burst + # so it closes the user's turn NOW → immediate reply. + if _FORCE_ANSWER.is_set(): + _FORCE_ANSWER.clear() + if not self._speaking: + force_silence_until = now + 0.6 + log.info("ANSWER NOW (dashboard) — forcing turn end") + + # Dashboard Interrupt-now button: cut speech instantly AND stay + # SILENT until the user speaks — arming the greeting suppression + # also swallows any in-flight or reconnect-greeting audio, so a + # press while "listening" can never be followed by Sanad talking + # on his own. + if _MANUAL_INTERRUPT.is_set(): + _MANUAL_INTERRUPT.clear() + log.info("MANUAL INTERRUPT (dashboard)%s", + "" if self._speaking else " — idle; muted until user speaks") + self._interrupt("manual") + self._barge_block_until = now + cooldown + self._suppress_greeting = True # On the JBL (loud external speaker) the head mic hears the robot's # OWN voice as loud as the user. We FULLY gate the mic to Gemini while # it speaks (+ a short echo tail) so it NEVER hears itself, and we @@ -657,24 +867,38 @@ class GeminiBrain: # model. (Reliable JBL interrupt needs AEC; the only PulseAudio mic is # dead, so that's separate work.) The chest speaker (builtin) keeps # light quiet-frame suppression + working barge-in (firmware AEC). - full_gate = "jbl" in (self._current_profile_id or "") + # The Beats Pill (USB) is the same regime as the JBL: loud external + # speaker with a raw AEC-less mic inches away — the mic feed to + # Gemini is ALWAYS fully gated while speaking (echo-safe). Voice + # barge-in on external speakers is an OPTION (_EXTERNAL_BARGE_IN): + # the LOCAL detector cuts the speaker using stricter parameters. + external = any(t in (self._current_profile_id or "") + for t in ("jbl", "beats")) + full_gate = external + allow_barge = (not external) or _EXTERNAL_BARGE_IN - # Barge-in: sustained user energy cuts the AI — chest profile only. - if self._speaking and not full_gate and now >= self._barge_block_until: - if (now - self._ai_speak_start) >= grace: - if energy > threshold: + # Barge-in: sustained user energy cuts the AI. + if self._speaking and allow_barge and now >= self._barge_block_until: + eff_grace = _JBL_BARGE_GRACE if external else grace + eff_threshold = threshold * _JBL_BLEED_MARGIN if external else threshold + eff_chunks = _JBL_BARGE_CHUNKS if external else chunks_needed + if (now - self._ai_speak_start) >= eff_grace: + if energy > eff_threshold: loud_count += 1 else: loud_count = max(0, loud_count - 1) - if loud_count > chunks_needed: - log.info("BARGE-IN (e=%d)", energy) + if loud_count > eff_chunks: + log.info("BARGE-IN (e=%d, external=%s)", energy, external) self._interrupt("barge-in") loud_count = 0 self._barge_block_until = now + cooldown # Echo suppression: mask the mic so the model doesn't hear its own bleed. send_data = data - if _INPUT_PAUSED.is_set(): + if now < force_silence_until: + # Answer-now: mask the mic so the server VAD sees pure silence. + send_data = _SILENCE_PCM + elif _INPUT_PAUSED.is_set(): # Paused for a record playback — feed silence so Gemini neither # hears the record nor keeps talking over it. send_data = _SILENCE_PCM @@ -835,7 +1059,8 @@ class GeminiBrain: # mouth-open level (0..3) from this chunk's RMS, # throttled. Parsed by GeminiSubprocess._reader_loop. _mnow = time.time() - if _mnow - getattr(self, "_mouth_t", 0.0) >= 0.08: + if not _PURE_VOICE and \ + _mnow - getattr(self, "_mouth_t", 0.0) >= 0.08: _rms = (float(np.sqrt(np.mean( audio.astype(np.float32) ** 2))) if audio.size else 0.0) # Lower thresholds bias the mouth more open @@ -892,6 +1117,10 @@ class GeminiBrain: async def _announce_vision_state(self, session: Any, enabled: bool, is_toggle: bool) -> None: + if _PURE_VOICE and not enabled and not is_toggle: + # PURE: skip the "disabled" standing-context injection — only + # ENABLED features get announced/used (toggles always announce). + return if is_toggle and enabled: text = ( "[VISION ON] Your camera was just enabled — you can now see " @@ -935,6 +1164,10 @@ class GeminiBrain: async def _announce_facerec_state(self, session: Any, enabled: bool, is_toggle: bool) -> None: + if _PURE_VOICE and not enabled and not is_toggle: + # PURE: skip the "disabled" standing-context injection — only + # ENABLED features get announced/used (toggles always announce). + return if is_toggle and enabled: text = ( "[FACE RECOGNITION ON] Face recognition was just enabled — " @@ -979,6 +1212,10 @@ class GeminiBrain: async def _announce_zonerec_state(self, session: Any, enabled: bool, is_toggle: bool) -> None: + if _PURE_VOICE and not enabled and not is_toggle: + # PURE: skip the "disabled" standing-context injection — only + # ENABLED features get announced/used (toggles always announce). + return if is_toggle and enabled: text = ( "[ZONE RECOGNITION ON] You were just given the zones and places " @@ -1140,6 +1377,13 @@ class GeminiBrain: "places": r.get("places", [])} if name == "stop_navigation": return await asyncio.to_thread(_nav_api, "POST", "/api/nav/cancel", None) + if name == "search_internet": + query = str(args.get("query") or "").strip() + if not query: + return {"ok": False, "reason": "no_query"} + log.info("[[SEARCH:%s]]", query) # breadcrumb for the dashboard + return await asyncio.to_thread( + _run_grounding_search, query, self._api_key) if name == "set_expression": emotion = str(args.get("emotion") or "").strip().lower() if emotion not in _FACE_EMOTIONS: @@ -1165,6 +1409,10 @@ class GeminiBrain: async def _announce_movement_state(self, session: Any, enabled: bool, is_toggle: bool) -> None: + if _PURE_VOICE and not enabled and not is_toggle: + # PURE: skip the "disabled" standing-context injection — only + # ENABLED features get announced/used (toggles always announce). + return if is_toggle and enabled: text = ( "[MOVEMENT ON] Walking is now enabled — you can move when the " @@ -1490,6 +1738,12 @@ class GeminiBrain: await self._announce_movement_state( session, self._movement_enabled, is_toggle=True, ) + if _PURE_VOICE: + # PURE: tools are baked per-session (nav tools only while + # movement is on) — cycle the session so the toggle takes + # effect on the function-calling side too. + log.info("pure-voice: cycling session to rebake nav tools") + self._done.set() # Auto-record toggle — flip the recorder live (no session restart). if new_state.record_enabled != last_state.record_enabled: @@ -1520,6 +1774,12 @@ class GeminiBrain: await asyncio.sleep(max(period, backoff)) if not self._vision_enabled: continue + # Never compete with the INCOMING audio stream while the AI is + # speaking — a frame upload on the shared websocket delays audio + # chunks, which breaks the chest speaker (tiny firmware buffer; + # the pulse/Pill path hides it behind its own buffering). + if self._speaking or (time.time() - self._last_ai_audio) < 0.5: + continue with _LATEST_FRAME_LOCK: data = _LATEST_FRAME.get("bytes") ts = _LATEST_FRAME.get("ts", 0.0) diff --git a/gemini/subprocess.py b/gemini/subprocess.py index 862436a..2fba29a 100644 --- a/gemini/subprocess.py +++ b/gemini/subprocess.py @@ -38,7 +38,9 @@ _FRAME_FORWARD_INTERVAL_S = float(_LS_CFG.get("frame_forward_interval_sec", 0.5) # Audio profile watcher — poll pactl for the Anker USB device at this # interval, send "profile:" to the child on every state change. -_AUDIO_WATCH_INTERVAL_S = float(_LS_CFG.get("audio_watch_interval_sec", 1.5)) +# 5s: each poll shells out to pactl several times (detect_plugged_profiles) — +# 1.5s was a constant subprocess storm for a rare event (device hot-plug). +_AUDIO_WATCH_INTERVAL_S = float(_LS_CFG.get("audio_watch_interval_sec", 5.0)) # The Anker profile id, as defined in voice/audio_devices.py. When this # profile is fully plugged (both sink + source present), we switch the @@ -562,7 +564,7 @@ class GeminiSubprocess: if the process isn't running or stdin is closed.""" pid = (profile_id or "").strip().lower() if pid not in {"builtin", "anker", "anker_powerconf", - "hollyland_builtin", "jbl_builtin_mic"}: + "hollyland_builtin", "jbl_builtin_mic", "beats_pill"}: log.warning("send_profile: ignoring unknown profile %r", profile_id) return payload: dict[str, Any] = {"id": pid} @@ -580,6 +582,16 @@ class GeminiSubprocess: owns the chest speaker, then resumes. No-op if not running.""" self._send_stdin("pause:%d\n" % (1 if paused else 0)) + def send_interrupt(self) -> None: + """Instantly stop the AI's current speech (dashboard Interrupt-now + button) — the child stops the speaker and returns to listening.""" + self._send_stdin("interrupt:1\n") + + def send_answer_now(self) -> None: + """Force Gemini to answer immediately (dashboard Answer-now button) — + the child feeds the server VAD a short silence burst to close the turn.""" + self._send_stdin("answernow:1\n") + def _audio_watcher(self) -> None: """Background thread — poll pactl for the Anker USB device, signal the child on every plug/unplug edge transition. diff --git a/vision/camera.py b/vision/camera.py index 003122e..33b62ad 100644 --- a/vision/camera.py +++ b/vision/camera.py @@ -23,6 +23,7 @@ from __future__ import annotations import base64 import os +import re import threading import time from typing import Optional @@ -386,6 +387,46 @@ class CameraDaemon: log.info("USB camera unavailable: %s", exc) return None + # Pinned DEVICE PATH — strongest pin. A /dev/v4l/by-path/... symlink is + # keyed to the physical USB port + interface, so it survives the + # re-enumeration that silently moves raw /dev/videoN indices around + # (observed on the D435i: colour node drifted 12→6→5 across replugs). + # Resolve the symlink to the current /dev/videoN and open that. + dev_path = os.environ.get("SANAD_CAMERA_DEVICE", "").strip() + if dev_path: + # Glob patterns allowed: a D435i that renegotiates USB3→USB2 + # changes its bus/port prefix, but the INTERFACE suffix is + # invariant — e.g. "/dev/v4l/by-path/*-usb-*:1.3-video-index0" + # always names the RGB capture node wherever it enumerates. + import glob as _glob + if any(c in dev_path for c in "*?["): + candidates = sorted(_glob.glob(dev_path)) + if not candidates: + log.warning("USB camera: SANAD_CAMERA_DEVICE pattern %s " + "matched nothing — falling back to index pin " + "/ scan", dev_path) + else: + candidates = [dev_path] + for cand in candidates: + real = os.path.realpath(cand) + m = re.match(r"^/dev/video(\d+)$", real) + if not m: + log.warning("USB camera: %s does not resolve to a " + "/dev/videoN node (got %s)", cand, real) + continue + backend = self._open_usb_index(int(m.group(1)), w, h, f, cv2) + if backend is not None: + fw, fh = backend["frame_wh"] + log.info("USB camera: pinned by path %s → /dev/video%d " + "(%dx%d, %s)", cand, backend["index"], fw, fh, + "colour" if backend["is_color"] else "grayscale/IR") + return backend + log.warning("USB camera: pinned path %s (→ %s) unusable", + cand, real) + if candidates: + log.warning("USB camera: no SANAD_CAMERA_DEVICE candidate " + "usable — falling back to index pin / scan") + # Pinned index — accept whatever it is (colour or not). explicit = os.environ.get("SANAD_CAMERA_USB_INDEX", "").strip() if explicit.isdigit(): diff --git a/voice/audio_devices.py b/voice/audio_devices.py index 9a075b9..ba7942e 100644 --- a/voice/audio_devices.py +++ b/voice/audio_devices.py @@ -50,6 +50,20 @@ class AudioProfile: description: str = "" sink_sample_rate: int = 0 # 0 = use device default source_sample_rate: int = 0 + # True → the profile's REAL mic is the G1 firmware UDP-multicast stream + # (voice/audio_io), so a missing PulseAudio source must not disqualify it. + # Some G1 units never expose the chest mic to the Jetson's PulseAudio + # (I2S capture reads silence) while the UDP mic works fine — detection + # then falls back to a synthetic "udp:" source (see apply_selection). + udp_mic_fallback: bool = False + # True → the profile's REAL speaker is the G1 firmware chest speaker driven + # over DDS (voice/audio_io BuiltinSpeaker.PlayStream), NOT a PulseAudio sink. + # The Jetson's platform-sound card is input-only on this robot (it has no + # output sink at all), so a missing PulseAudio sink must NOT disqualify the + # chest profile — the chest is always usable. Detection then falls back to + # a synthetic "dds:" sink (see apply_selection, which skips + # `pactl set-default-sink` for it, exactly like the "udp:" source). + dds_speaker_fallback: bool = False # Built-in device profiles. @@ -77,6 +91,8 @@ PROFILES: list[AudioProfile] = [ sink_pattern="platform-sound", source_pattern="alsa_input.platform-sound", description="Jetson / G1 built-in audio chip. (Default)", + udp_mic_fallback=True, # chest mic = G1 firmware UDP stream, not pulse + dds_speaker_fallback=True, # chest speaker = G1 DDS PlayStream, not pulse ), AudioProfile( id="hollyland_builtin", @@ -92,6 +108,18 @@ PROFILES: list[AudioProfile] = [ source_pattern="powerconf,anker", description="Anker PowerConf USB conference unit — mic + speaker on the same device.", ), + AudioProfile( + id="beats_pill", + label="Beats Pill (mic + speaker)", + # USB-C: kernel names the card "Pill"; pulse sinks/sources carry + # "Beats_Pill" (auto-probed or the manual module-alsa-sink/-source + # loads from /etc/pulse/default.pa — pulse fails to probe the Pill's + # output profile on some kernels). Bluetooth: bluez sink. + sink_pattern="beats_pill,beats,pill", + source_pattern="beats_pill,beats,pill", + description="Beats Pill (USB-C or Bluetooth) — speaker + speakerphone mic.", + udp_mic_fallback=True, # mic falls back to the G1 UDP mic if absent + ), AudioProfile( id="jbl_builtin_mic", label="JBL speaker + built-in mic", @@ -101,6 +129,7 @@ PROFILES: list[AudioProfile] = [ # The JBL has NO microphone → input stays on the G1 built-in mic. source_pattern="alsa_input.platform-sound", description="JBL Bluetooth speaker for output + the G1 built-in microphone for input (the JBL has no mic).", + udp_mic_fallback=True, # builtin mic = G1 UDP stream on units without the alsa source ), ] @@ -488,6 +517,10 @@ def ensure_card_input_capable(card_pattern: str) -> bool: return switched_any +# Profiles that already logged the (permanent) UDP-mic fallback notice. +_UDP_FALLBACK_LOGGED: set[str] = set() + + def detect_plugged_profiles() -> list[dict[str, Any]]: """Return all profiles whose sink AND source are currently plugged in. @@ -507,6 +540,17 @@ def detect_plugged_profiles() -> list[dict[str, Any]]: refreshed_sources = False for prof in PROFILES: sink = find_first_match(sinks, prof.sink_pattern) + if not sink and prof.dds_speaker_fallback: + # The profile's real speaker is the G1 chest driven over DDS + # (AudioClient.PlayStream), not a PulseAudio sink. On this robot the + # Jetson audio card is input-only (no output sink exists at all), so + # a missing pulse sink must NOT disqualify the chest — it is always + # usable. Synthesise a "dds:" sink so the profile stays detected and + # selectable; apply_selection() skips set-default-sink for "dds:". + sink = { + "name": "dds:g1-chest-speaker", + "description": "G1 firmware chest speaker (DDS, no PulseAudio sink)", + } if not sink: continue src = find_first_match(sources, prof.source_pattern, exclude_monitors=True) @@ -535,6 +579,24 @@ def detect_plugged_profiles() -> list[dict[str, Any]]: log.info("detect_plugged_profiles: %s source resolved via " "PyAudio fallback (pactl missed it): %s", prof.id, src.get("name", "?")) + if src is None and prof.udp_mic_fallback: + # The profile's real mic is the G1 firmware UDP-multicast stream + # (voice/audio_io) — no PulseAudio source needed. Some G1 units + # never expose the chest mic to the Jetson's pulse (I2S capture + # reads silence); the profile is still fully functional, so a + # synthetic source keeps it selectable. apply_selection() skips + # `pactl set-default-source` for "udp:" names (like "pyaudio:"). + src = { + "name": "udp:g1-multicast-mic", + "description": "G1 firmware UDP-multicast mic (no PulseAudio source)", + } + via = "udp-builtin" + # This path is the steady state on chest-mic robots and the + # detector polls every few seconds — log once, not per poll. + if prof.id not in _UDP_FALLBACK_LOGGED: + _UDP_FALLBACK_LOGGED.add(prof.id) + log.info("detect_plugged_profiles: %s using UDP-mic fallback " + "(no matching pulse source)", prof.id) if sink and src: detected.append({ "profile": asdict(prof), @@ -726,7 +788,11 @@ def apply_selection(sink: str, source: str) -> dict[str, Any]: """ errors: list[str] = [] if sink: - if not set_default_sink(sink): + if sink.startswith("dds:"): + log.info("apply_selection: sink is synthetic (%s) — skipping " + "pactl set-default-sink. The G1 chest speaker plays via " + "DDS (AudioClient.PlayStream), not PulseAudio.", sink) + elif not set_default_sink(sink): errors.append(f"set-default-sink failed: {sink}") else: try: @@ -734,10 +800,11 @@ def apply_selection(sink: str, source: str) -> dict[str, Any]: except (FileNotFoundError, subprocess.SubprocessError): pass if source: - if source.startswith("pyaudio:"): - log.info("apply_selection: source is PyAudio-direct (%s) — " - "skipping pactl set-default-source. Live mic path " - "uses PortAudio device match; pactl defaults stay put.", + if source.startswith(("pyaudio:", "udp:")): + log.info("apply_selection: source is synthetic (%s) — " + "skipping pactl set-default-source. PyAudio-direct mics " + "use PortAudio device match; udp: is the G1 firmware " + "multicast mic. pactl defaults stay put.", source) elif not set_default_source(source): errors.append(f"set-default-source failed: {source}") diff --git a/voice/audio_io.py b/voice/audio_io.py index c2b0b09..59c208d 100644 --- a/voice/audio_io.py +++ b/voice/audio_io.py @@ -31,6 +31,7 @@ import json import socket import struct import subprocess +import queue import threading import time from abc import ABC, abstractmethod @@ -211,7 +212,16 @@ class BuiltinMic(Mic): class BuiltinSpeaker(Speaker): - """G1 robot's built-in speaker via AudioClient.PlayStream (16 kHz mono).""" + """G1 robot's built-in speaker via AudioClient.PlayStream (16 kHz mono). + + JITTER-BUFFERED: chunks are enqueued and a dedicated writer thread sends + them to the firmware, prebuffering ~0.3s before the first RPC and never + running more than ~0.6s ahead of real time. Gemini's audio arrives in + BURSTS (WiFi jitter, shared-websocket contention) — the old direct + per-chunk RPC mapped every arrival gap to an audible dropout on the + chest speaker's tiny firmware buffer, while the pulse (Pill/Anker) path + hid it behind PulseAudio's own buffering. This gives the chest the same + tolerance.""" HARDWARE_RATE = 16_000 @@ -224,10 +234,16 @@ class BuiltinSpeaker(Speaker): self._app_name = app_name or _SP_CFG.get("app_name", "sanad") self._begin_pause = _SP_CFG.get("begin_stream_pause_sec", 0.15) self._finish_margin = _SP_CFG.get("wait_finish_margin_sec", 0.3) + # Jitter buffer tuning (see class docstring). + self._prebuffer_sec = _SP_CFG.get("jitter_prebuffer_sec", 0.30) + self._first_flush_sec = _SP_CFG.get("jitter_first_flush_sec", 0.60) + self._max_lead_sec = _SP_CFG.get("jitter_max_lead_sec", 0.60) self._stop_flag = threading.Event() self._stream_id: Optional[str] = None self._total_sent = 0.0 self._play_start = 0.0 + self._q: "queue.Queue[Optional[bytes]]" = queue.Queue() + self._writer: Optional[threading.Thread] = None def _stop_play_api(self) -> None: try: @@ -241,13 +257,79 @@ class BuiltinSpeaker(Speaker): except Exception: log.warning("BuiltinSpeaker AUDIO_STOP_PLAY failed") + def _drain_queue(self) -> None: + try: + while True: + self._q.get_nowait() + except queue.Empty: + pass + def begin_stream(self) -> None: + # Tear down any previous writer cleanly before starting a new turn. + self._stop_flag.set() + self._drain_queue() + if self._writer is not None and self._writer.is_alive(): + self._q.put(None) + self._writer.join(timeout=1.0) self._stop_flag.clear() + self._drain_queue() self._stop_play_api() time.sleep(self._begin_pause) self._stream_id = f"s_{int(time.time() * 1000)}" self._total_sent = 0.0 self._play_start = time.time() + self._writer = threading.Thread( + target=self._writer_loop, args=(self._stream_id,), + name="builtin-speaker-writer", daemon=True) + self._writer.start() + + def _writer_loop(self, stream_id: str) -> None: + rate = float(self.HARDWARE_RATE) + sent_sec = 0.0 + started = False + t0 = time.time() + batch: list = [] + buffered = 0.0 + while not self._stop_flag.is_set() and self._stream_id == stream_id: + try: + item = self._q.get(timeout=0.05) + except queue.Empty: + item = () + if item is None: + return + if not started: + if item: + batch.append(item) + buffered += len(item) / 2 / rate + if buffered >= self._prebuffer_sec or ( + batch and (time.time() - t0) >= self._first_flush_sec): + self._play_start = time.time() + started = True + for b in batch: + if self._stop_flag.is_set() or self._stream_id != stream_id: + return + try: + self._ac.PlayStream(self._app_name, stream_id, b) + except Exception as exc: + log.warning("PlayStream failed: %s", exc) + sent_sec += len(b) / 2 / rate + batch = [] + continue + if not item: + continue + # Pace: never run more than max_lead ahead of real time — keeps + # the firmware queue short (fast interrupts) while OUR queue is + # the elastic reservoir that absorbs bursty arrival. + lead = sent_sec - (time.time() - self._play_start) + if lead > self._max_lead_sec: + time.sleep(min(lead - self._max_lead_sec, 0.25)) + if self._stop_flag.is_set() or self._stream_id != stream_id: + return + try: + self._ac.PlayStream(self._app_name, stream_id, item) + except Exception as exc: + log.warning("PlayStream failed: %s", exc) + sent_sec += len(item) / 2 / rate def send_chunk(self, pcm: PCMLike, source_rate: int) -> None: if self._stop_flag.is_set(): @@ -256,20 +338,22 @@ class BuiltinSpeaker(Speaker): if arr.size < 10: return hw = _resample_int16(arr, source_rate, self.HARDWARE_RATE) - self._ac.PlayStream(self._app_name, self._stream_id, hw.tobytes()) + self._q.put(hw.tobytes()) self._total_sent += len(hw) / self.HARDWARE_RATE def wait_finish(self) -> None: - elapsed = time.time() - self._play_start - remaining = self._total_sent - elapsed + self._finish_margin - waited = 0.0 - while waited < remaining and not self._stop_flag.is_set(): + # Wait for the queue to drain AND real-time playback to elapse. + while not self._stop_flag.is_set(): + elapsed = time.time() - self._play_start + remaining = self._total_sent - elapsed + self._finish_margin + if self._q.empty() and remaining <= 0: + break time.sleep(0.1) - waited += 0.1 self._stop_play_api() def stop(self) -> None: self._stop_flag.set() + self._drain_queue() self._stop_play_api() @property @@ -815,9 +899,13 @@ _PROFILE_ALIASES = { "hollyland_builtin": "hollyland_builtin", "jbl": "jbl_builtin_mic", "jbl_builtin_mic": "jbl_builtin_mic", + "beats": "beats_pill", + "beats_pill": "beats_pill", + "pill": "beats_pill", } -SUPPORTED_PROFILES = ("builtin", "anker", "hollyland_builtin", "jbl_builtin_mic") +SUPPORTED_PROFILES = ("builtin", "anker", "hollyland_builtin", "jbl_builtin_mic", + "beats_pill") @dataclass @@ -885,6 +973,13 @@ class AudioIO: # pacat is used because PyAudio's 'pulse' device is unavailable in # this env. Neither backend needs the AudioClient. return BuiltinMic(), PulseStreamSpeaker(label="JBL", sink_pattern="jbl,bluez") + if resolved == "beats_pill": + # Beats Pill speaker via pacat → its PulseAudio sink + the G1 + # built-in DDS mic — same regime as the JBL (loud external speaker, + # full mic-gate while speaking; see gemini/script.py). The Pill's + # own USB mic stays available to pulse for dashboard recordings. + return BuiltinMic(), PulseStreamSpeaker( + label="Beats Pill", sink_pattern="beats_pill,beats,pill") raise AssertionError(f"unhandled resolved profile: {resolved!r}") @classmethod diff --git a/voice/movement_dispatch.py b/voice/movement_dispatch.py index d9d010c..7252a7a 100644 --- a/voice/movement_dispatch.py +++ b/voice/movement_dispatch.py @@ -108,6 +108,11 @@ class MovementDispatcher: def _load(self): try: data = json.loads(self._instruction_path.read_text(encoding="utf-8")) + except FileNotFoundError: + # Normal on robots with no movement phrase pack — not an error. + log.info("no %s — movement dispatcher inert (0 phrases)", + self._instruction_path.name) + data = {} except Exception as exc: log.error("could not load %s: %s — dispatcher inert", self._instruction_path, exc) data = {} diff --git a/voice/sanad_voice.py b/voice/sanad_voice.py index 4746c46..c6c30c6 100644 --- a/voice/sanad_voice.py +++ b/voice/sanad_voice.py @@ -146,7 +146,8 @@ _MOVEMENT_PROMPT_RULES = ( "(you receive a [MOVEMENT ON] or [MOVEMENT STATUS] note). When movement is " "OFF, never confirm a motion — tell the user to enable movement from the " "dashboard.\n" - "When movement is ON and the user addresses you by name (Bousandah / بوسنده) AND " + "When movement is ON and the user addresses you by name (Sanad / سند / " + "Bousandah / بوسنده) AND " "asks you to move, reply with ONE short confirmation phrase per requested " "motion, in the SAME language, in the order asked. Use these EXACT shapes — " "they are what triggers the motion:\n"