"""Reflexes — instant local handling for trivial commands. "Volume up", "pause the music", "open Discord" should feel like a light switch, not a conversation: no LLM round-trip, no thinking pause. The engine calls try_reflex() with each final transcript BEFORE sending it to the brain; a match executes immediately (media keys / Start-menu launch) and returns a short spoken confirmation. No match returns None and the utterance flows to Claude as usual. Deliberately conservative: only short, simple, unambiguous phrasings match (no "and"/"then"/questions) — anything with nuance belongs to the brain. """ from __future__ import annotations import re import subprocess import sys # macOS media keys have no pyautogui names — post real NX_SYSDEFINED events # (what the keyboard's F7-F9 send). Volume goes through osascript instead: # exact percentage control and no permission requirement. _NX = {"playpause": 16, "nexttrack": 17, "prevtrack": 18} def _mac_media_key(code: int) -> None: import Quartz from AppKit import NSEvent def _post(down: bool) -> None: flags = 0xA00 if down else 0xB00 data1 = (code << 16) | flags ev = (NSEvent. otherEventWithType_location_modifierFlags_timestamp_windowNumber_context_subtype_data1_data2_( 14, (0, 0), flags, 0, 0, None, 8, data1, -1)) Quartz.CGEventPost(Quartz.kCGHIDEventTap, ev.CGEvent()) _post(True) _post(False) def _osa(script: str) -> None: subprocess.run(["osascript", "-e", script], capture_output=True, timeout=8) def _press(key: str, times: int = 1) -> None: if sys.platform == "darwin": if key == "volumeup": _osa("set volume output volume " f"((output volume of (get volume settings)) + {6 * times})") elif key == "volumedown": _osa("set volume output volume " f"((output volume of (get volume settings)) - {6 * times})") elif key == "volumemute": _osa("set volume output muted " "(not (output muted of (get volume settings)))") else: for _ in range(times): _mac_media_key(_NX[key]) return import pyautogui # lazy for _ in range(times): pyautogui.press(key) # "open chrome" must resolve the way a human means it, not literal app names _MAC_APP_ALIASES = { "chrome": "Google Chrome", "google chrome": "Google Chrome", "vscode": "Visual Studio Code", "vs code": "Visual Studio Code", "code": "Visual Studio Code", "visual studio": "Visual Studio Code", "teams": "Microsoft Teams", "word": "Microsoft Word", "excel": "Microsoft Excel", "powerpoint": "Microsoft PowerPoint", "outlook": "Microsoft Outlook", "settings": "System Settings", "system settings": "System Settings", "preferences": "System Settings", "browser": "Google Chrome", "the browser": "Google Chrome", "file explorer": "Finder", "files": "Finder", "explorer": "Finder", } def _launch(name: str) -> tuple[bool, str]: """Open an app by spoken name. Returns (ok, resolved_name).""" if sys.platform != "darwin": from voice.desktop_control import shell_launch return shell_launch(name) n = name.strip().rstrip(".!?") for cand in dict.fromkeys( [_MAC_APP_ALIASES.get(n.lower()), n, n.title()]): if not cand: continue r = subprocess.run(["open", "-a", cand], capture_output=True, timeout=8) if r.returncode == 0: return True, cand return False, name # --- patterns --------------------------------------------------------------- _FILLER = r"(?:please\s+|jarvis[,\s]+|can\s+you\s+|could\s+you\s+)?" _VOL_UP = re.compile( rf"^{_FILLER}(?:turn\s+(?:the\s+)?volume\s+up|volume\s+up" rf"|raise\s+(?:the\s+)?volume|louder|turn\s+it\s+up)" rf"(?P.{{0,12}})$", re.I) _VOL_DOWN = re.compile( rf"^{_FILLER}(?:turn\s+(?:the\s+)?volume\s+down|volume\s+down" rf"|lower\s+(?:the\s+)?volume|quieter|turn\s+it\s+down)" rf"(?P.{{0,12}})$", re.I) _VOL_MUTE = re.compile( rf"^{_FILLER}(?:mute|unmute)\s+(?:the\s+)?(?:sound|audio|volume|music|pc)" rf"[.!]?$", re.I) _MEDIA_PAUSE = re.compile( rf"^{_FILLER}(?:pause|stop|resume|play|unpause)\s+(?:the\s+)?" rf"(?:music|song|media|playback)[.!]?$|^{_FILLER}(?:pause|resume)[.!]?$", re.I) _MEDIA_NEXT = re.compile( rf"^{_FILLER}(?:next|skip)(?:\s+(?:the\s+)?(?:song|track|music))?[.!]?$", re.I) _MEDIA_PREV = re.compile( rf"^{_FILLER}(?:previous|last)\s+(?:song|track)[.!]?$", re.I) _OPEN = re.compile( rf"^{_FILLER}(?:open|launch|start)\s+(?P[\w .&'+-]{{2,32}})[.!]?$", re.I) # words that mean the "open" is part of something bigger → not a reflex _COMPLEX = re.compile(r"\b(and|then|after|before|search|find|go\s+to|type|" r"click|tell|send|message|play\s+some)\b", re.I) def _vol_times(mod: str) -> int: m = (mod or "").lower() if "lot" in m or "way" in m or "max" in m: return 10 if "bit" in m or "little" in m or "slight" in m: return 2 return 4 def _run_intent(intent: str, text: str) -> str | None: """Execute a classified intent. Returns the confirmation to speak.""" from voice.intent import app_name, multiplier if intent == "volume_up": _press("volumeup", 2 * multiplier(text)) return "Volume up, sir." if intent == "volume_down": _press("volumedown", 2 * multiplier(text)) return "Volume down, sir." if intent in ("mute_sound", "unmute_sound"): _press("volumemute") return "Done, sir." if intent == "pause_media": _press("playpause") return "Paused, sir." if intent == "resume_media": _press("playpause") return "Playing, sir." if intent == "next_track": _press("nexttrack") return "Next track, sir." if intent == "prev_track": _press("prevtrack") return "Going back, sir." if intent == "open_app": name = app_name(text) if not name: return None ok, matched = _launch(name) return f"Opening {matched}, sir." if ok else None return None def try_reflex(text: str) -> str | None: """Execute a trivial command instantly. Returns the spoken confirmation, or None when the utterance isn't a reflex (send it to the brain). Primary path: the SEMANTIC intent router (voice/intent.py) — matches meaning ("I can't hear this" -> volume_up) and stands down when unsure. The regexes below are only the fallback when model2vec is missing. """ t = (text or "").strip() if not t: return None # multi-step phrasing is never a reflex, however confident the router is if _COMPLEX.search(t): return None try: from voice.intent import get_router router = get_router() if router is not None: intent, score = router.classify(t) if intent is not None: out = _run_intent(intent, t) if out is not None: print(f" [reflex: {intent} ({score:.2f})]") return out return None # router available but unsure -> brain decides except Exception: # noqa: BLE001 — router problems must never block pass # ---- regex fallback (no model2vec) ---- if len(t.split()) > 7: return None try: m = _VOL_UP.match(t) if m and not _COMPLEX.search(m.group("mod") or ""): _press("volumeup", _vol_times(m.group("mod"))) return "Volume up, sir." m = _VOL_DOWN.match(t) if m and not _COMPLEX.search(m.group("mod") or ""): _press("volumedown", _vol_times(m.group("mod"))) return "Volume down, sir." if _VOL_MUTE.match(t): _press("volumemute") return "Done, sir." if _MEDIA_PAUSE.match(t): _press("playpause") return "Done, sir." if _MEDIA_NEXT.match(t): _press("nexttrack") return "Next track, sir." if _MEDIA_PREV.match(t): _press("prevtrack") return "Going back, sir." m = _OPEN.match(t) if m: ok, matched = _launch(m.group("app").strip()) if ok: return f"Opening {matched}, sir." return None # couldn't resolve — let the brain figure it out except Exception: # noqa: BLE001 — a reflex must never break the loop return None return None