"""LiveVoiceLoop — voice-to-arm phrase trigger dispatcher. Listens to user transcriptions from the GeminiSubprocess and, when a configured wake phrase is matched, fires the corresponding arm action via `motion.sanad_arm_controller.ARM`. Mode toggle ("DEFERRED TRIGGER"): - fire_on_wake_match=True fires the arm instantly on phrase match (fast, no coordination with AI speech) - fire_on_wake_match=False marks a pending action that fires when the AI starts/finishes its reply (visually nicer — robot answers, then moves) This is Option-D integration: parallel to skill_registry, uses the full gemini_interact phrase dictionary (sanad_arm.txt, 29 arm IDs, hundreds of Arabic phrase variants). """ from __future__ import annotations import threading import time from collections import deque from datetime import datetime from types import SimpleNamespace from typing import Any from Project.Sanad.config import SCRIPTS_DIR, BASE_DIR from Project.Sanad.core.config_loader import section as _cfg_section from Project.Sanad.core.logger import get_logger from Project.Sanad.voice.text_utils import ( load_arm_phrase_dispatch, maybe_trigger_arm, ) log = get_logger("live_voice_loop") _LV_CFG = _cfg_section("voice", "live_voice_loop") # Filename from core.script_files (single source) — resolved under SCRIPTS_DIR _SCRIPTS = _cfg_section("core", "script_files") _ARM_TXT_NAME = _SCRIPTS.get("arm_phrases", "sanad_arm.txt") SANAD_ARM_TXT = SCRIPTS_DIR / _ARM_TXT_NAME TRIGGER_LOG_SIZE = _LV_CFG.get("trigger_log_size", 100) POLL_INTERVAL_SEC = _LV_CFG.get("poll_interval_sec", 0.1) DEFERRED_DEFAULT = _LV_CFG.get("deferred_default", False) TRIGGER_ENABLED_DEFAULT = bool(_LV_CFG.get("trigger_enabled_default", False)) class LiveVoiceLoop: """Polls GeminiSubprocess transcripts → fires arm actions.""" def __init__(self, voice_client, arm, wake_mgr, audio_mgr): self.voice_client = voice_client self.arm = arm # Sanad's motion/arm_controller (not used for trigger) self.wake_mgr = wake_mgr self.audio_mgr = audio_mgr self._running = False self._poll_thread: threading.Thread | None = None self._stop_event = threading.Event() # Deferred-trigger toggle (fire on phrase match vs fire after AI responds) self.deferred_mode = DEFERRED_DEFAULT # Master arm-trigger gate — when False, transcripts are still # captured (you can watch them on the dashboard) but NO arm # actions fire. Defaults to OFF so the robot doesn't move # unexpectedly until the operator opts in. self.trigger_enabled = TRIGGER_ENABLED_DEFAULT # Trigger history (dashboard log) self.triggers: deque[dict[str, Any]] = deque(maxlen=TRIGGER_LOG_SIZE) self.last_heard: str = "" self.last_action: str = "" # ASR dispatch state (SimpleNamespace — maybe_trigger_arm mutates attrs) self.state = SimpleNamespace() # Load sanad_arm.txt on first construction self.wake_dispatch: dict[int, set[str]] = {} self.option_by_id: dict[int, Any] = {} self.sanad_arm: Any = None self._load_dispatch() # Guards the cross-thread transcript cursor + pending-trigger state. # set_trigger_enabled() runs on the FastAPI event-loop thread while the # poll daemon mutates the same fields — without this lock a concurrent # update can tear the cursor / pending flags. self._trigger_lock = threading.Lock() # Transcript consumption is tracked by POSITION, not by content, so a # repeated identical command (e.g. "wave" said twice) re-fires. We # remember the last deque snapshot we processed and only dispatch the # newly-appended tail. (live_sub.user_transcript is an append-only, # left-evicting deque, so the new lines are always at the right.) self._last_snapshot: list[str] = [] # ── phrase dispatch loader ──────────────────────────────────── def _load_dispatch(self): try: from Project.Sanad.motion.sanad_arm_controller import ARM, OPTION_LIST, OPTION_BY_ID self.sanad_arm = ARM self.option_by_id = OPTION_BY_ID # Voice-trigger policy: SDK built-ins only. # JSONL replays (option.file set) are dashboard-only — voice # phrase blocks for laugh/bird/change_battery/move_* never # reach `wake_dispatch`, so a matched phrase for one of those # silently no-ops in voice mode. sdk_only_options = [o for o in OPTION_LIST if not getattr(o, "file", "")] filtered_out = [o.name for o in OPTION_LIST if getattr(o, "file", "")] if SANAD_ARM_TXT.exists(): self.wake_dispatch = load_arm_phrase_dispatch(SANAD_ARM_TXT, sdk_only_options) log.info( "loaded %d arm-action phrase sets from %s " "(SDK-only filter: %d/%d options)", len(self.wake_dispatch), SANAD_ARM_TXT.name, len(sdk_only_options), len(OPTION_LIST), ) # Make the silent no-op observable: file-backed replays # (laugh/bird/change_battery/move_*) are excluded from voice # dispatch, so a spoken phrase for one of these does nothing. if filtered_out: log.warning( "voice arm dispatch EXCLUDES %d file-backed action(s) " "(dashboard-only, no voice trigger): %s", len(filtered_out), ", ".join(filtered_out), ) else: log.warning("sanad_arm.txt missing at %s — arm trigger disabled", SANAD_ARM_TXT) # Fold in operator-editable WakePhraseManager entries so dashboard # CRUD (data/wake_phrases.json) actually affects voice triggering. self._merge_wake_phrases() except Exception as exc: log.warning("arm dispatch unavailable: %s", exc) self.sanad_arm = None self.wake_dispatch = {} def _merge_wake_phrases(self) -> None: """Merge WakePhraseManager phrases into wake_dispatch. Without this, dashboard wake-phrase edits were a silent no-op — the manager was stored on self.wake_mgr but never consulted. We only fold in entries whose action_id resolves unambiguously to a voice-eligible (SDK-only, non file-backed) arm option, matching either the integer id or the option name; anything else is skipped with a warning so a mistyped/file-backed action can't misfire the arm.""" if self.wake_mgr is None: return try: from Project.Sanad.motion.sanad_arm_controller import ( OPTION_BY_ID, OPTION_BY_NAME, ) amap = self.wake_mgr.action_phrase_map() except Exception as exc: log.warning("wake_phrase merge unavailable: %s", exc) return merged = skipped = 0 for action_id_str, phrases in amap.items(): opt = None key = str(action_id_str).strip() # Resolve by integer id first, then by option name. if key.isdigit(): opt = OPTION_BY_ID.get(int(key)) if opt is None: opt = OPTION_BY_NAME.get(key.lower()) if opt is None or getattr(opt, "file", ""): # Unknown action, or a file-backed replay (voice-excluded). skipped += 1 log.warning("wake phrase action %r not voice-eligible — skipped", action_id_str) continue bucket = self.wake_dispatch.setdefault(opt.id, set()) bucket.update(p for p in phrases if p) merged += 1 if merged or skipped: log.info("merged WakePhraseManager entries (%d actions merged, %d skipped)", merged, skipped) # ── lifecycle ──────────────────────────────────────────────── async def start(self) -> None: if self._running: return self._stop_event.clear() self._running = True self._poll_thread = threading.Thread( target=self._poll_loop, daemon=True, name="live_voice_loop") self._poll_thread.start() log.info("LiveVoiceLoop started (deferred=%s, dispatch=%d)", self.deferred_mode, len(self.wake_dispatch)) async def stop(self) -> None: self._stop_event.set() self._running = False log.info("LiveVoiceLoop stopped") def set_deferred(self, enabled: bool) -> None: self.deferred_mode = bool(enabled) def set_trigger_enabled(self, enabled: bool) -> None: """Master arm-trigger gate. When False, phrase matches are ignored. Toggle semantics (no queue memory across the gate): - Always clears any in-flight pending trigger so a late fallback fire can't happen after disable/enable. - On enable: snapshots every transcript currently in the live_sub deque as already-seen. Only NEW speech after this moment will dispatch — phrases said while the gate was off don't suddenly fire when you turn it back on. """ self.trigger_enabled = bool(enabled) with self._trigger_lock: # Drop pending fallback timer — a queued "fire in 0.6s" from # before the toggle must not leak across. self.state._pending_arm_wave = False self.state._pending_arm_wave_fired = False self.state._pending_arm_trigger_fn = None self.state._pending_arm_fallback_time = 0.0 snapshotted = 0 if self.trigger_enabled: # On enable, mark everything currently in the deque as already # consumed (by position) so phrases said while the gate was off # don't suddenly fire when it's turned back on. Only NEW speech # after this moment will dispatch. try: from Project.Sanad.main import live_sub if live_sub is not None: self._last_snapshot = list(live_sub.user_transcript) snapshotted = len(self._last_snapshot) except Exception as exc: log.warning("set_trigger_enabled: snapshot failed: %s", exc) log.info("trigger_enabled=%s (pending cleared, %d transcripts marked seen)", self.trigger_enabled, snapshotted) # ── poll loop ──────────────────────────────────────────────── def _poll_loop(self): """Poll GeminiSubprocess.user_transcript for new user texts AND fire any deferred-mode arm trigger whose fallback timer elapsed. Without the pending-check, a deferred trigger (`fire_on_wake_match= False`) would only fire when the NEXT transcript arrives — so if the user says one sentence and stops, the arm never moves. """ while not self._stop_event.is_set(): self._check_transcripts() self._check_pending_trigger() self._stop_event.wait(POLL_INTERVAL_SEC) def _check_pending_trigger(self): """Fire a queued deferred trigger if its fallback time has passed.""" # Master gate — same check as _dispatch if not self.trigger_enabled: return # Read-and-claim the pending trigger under the lock so a concurrent # set_trigger_enabled() (FastAPI thread) can't clear it mid-fire and # cause a stray or lost deferred arm action. with self._trigger_lock: if not getattr(self.state, "_pending_arm_wave", False): return if getattr(self.state, "_pending_arm_wave_fired", False): return fn = getattr(self.state, "_pending_arm_trigger_fn", None) if fn is None: return fallback_at = float(getattr(self.state, "_pending_arm_fallback_time", 0.0) or 0.0) if fallback_at <= 0.0 or time.time() < fallback_at: return # Gate on arm idle — skip fire if a motion is already running if self.sanad_arm is not None and getattr(self.sanad_arm, "_is_busy", False): return # Claim it now (still under the lock) so it fires exactly once. self.state._pending_arm_wave_fired = True self.state._pending_arm_wave = False self.state._pending_arm_trigger_fn = None try: fn() except Exception as exc: log.warning("deferred arm trigger failed: %s", exc) @staticmethod def _new_tail(prev: list[str], curr: list[str]) -> list[str]: """Return the items appended to ``curr`` since ``prev`` was taken. ``curr`` is a snapshot of an append-only, left-evicting deque. The new lines are the suffix of ``curr`` that wasn't present at the end of ``prev``. We find the largest overlap k where the tail of ``prev`` equals the head of ``curr`` and return everything after it. This is position-based (not content-based), so a repeated identical command is treated as a genuinely new line and re-fires.""" if not prev: return list(curr) max_k = min(len(prev), len(curr)) for k in range(max_k, 0, -1): if prev[-k:] == curr[:k]: return list(curr[k:]) # No overlap — the buffer rolled over entirely between polls; treat the # whole current snapshot as new. return list(curr) def _check_transcripts(self): try: from Project.Sanad.main import live_sub except Exception: return if live_sub is None: return curr = list(live_sub.user_transcript) with self._trigger_lock: new_lines = self._new_tail(self._last_snapshot, curr) self._last_snapshot = curr # Dispatch only the newly-appended tail (outside the lock — _dispatch # may spawn an arm replay thread). for text in new_lines: self.last_heard = text self._dispatch(text) def _dispatch(self, transcript_text: str) -> None: if not self.wake_dispatch or self.sanad_arm is None: return # Master gate — skip arm triggering entirely when disabled if not self.trigger_enabled: return # Arm ⇄ locomotion interlock — refuse voice gestures while the robot # may be walking. The authoritative check is sanad_arm._blocked() at # fire time (covers the deferred path too); refuse early here so the # block is observable and we don't queue a deferred fire that would be # silently dropped later. try: if self.sanad_arm._blocked(): log.info("arm trigger refused — locomotion active (movement_active)") return except Exception: pass # Gate trigger on arm idle if getattr(self.sanad_arm, "_is_busy", False): return fire_now = not self.deferred_mode for action_id, phrases in self.wake_dispatch.items(): fn = self._make_trigger_fn(action_id) fired = maybe_trigger_arm( self.state, transcript_text, phrases, fire_on_wake_match=fire_now, arm_trigger_fn=fn, ) if fired: self._record_trigger(action_id, transcript_text, fire_now) break def _make_trigger_fn(self, action_id: int): def _fire(): try: self.sanad_arm.trigger_action_by_id(action_id) except Exception as exc: log.warning("arm trigger failed (id=%d): %s", action_id, exc) return _fire def _record_trigger(self, action_id: int, user_text: str, fired_now: bool): opt = self.option_by_id.get(action_id) action_name = opt.name if opt else f"id={action_id}" self.last_action = action_name self.triggers.append({ "time": datetime.now().strftime("%H:%M:%S"), "user_text": user_text, "action_id": action_id, "action_name": action_name, "mode": "instant" if fired_now else "deferred", }) log.info("arm trigger %s (id=%d) for: %r [%s]", action_name, action_id, user_text, "instant" if fired_now else "deferred") # ── status (dashboard) ─────────────────────────────────────── def status(self) -> dict[str, Any]: pending = "" pend_name = "" if getattr(self.state, "_pending_arm_wave", False): if self.state._pending_arm_trigger_fn is not None: # We can't introspect the action id from fn (closure), but # the last triggered line in self.triggers is likely the one. pend_name = self.triggers[-1].get("action_name", "") if self.triggers else "" pending = f"pending: {pend_name}" return { "running": self._running, "deferred_mode": self.deferred_mode, "trigger_enabled": self.trigger_enabled, "last_heard": self.last_heard, "pending_action": pending, "last_action": self.last_action, "audio_attached": self.audio_mgr is not None, "arm_attached": self.sanad_arm is not None, "gemini_connected": bool( self.voice_client and self.voice_client.connected), "dispatch_actions": len(self.wake_dispatch), "triggers": list(self.triggers)[-30:], }