samios/packaging/scripts/voice-bridge.py

#!/usr/bin/env python3
"""
SamiOS Voice Command Bridge
Listens for voice commands from SamiType, routes them to Hermes or dictation mode.

Usage:
    python3 voice-bridge.py [--version]

Requires:
    - SamiType STT running in streaming mode (Unix socket or WebSocket)
    - xdotool (for dictation mode typing)
    - Hermes API access (for command mode)
"""

import json
import subprocess
import socket
import os
import sys
import time
import threading
from enum import Enum


def _resolve_version() -> str:
    """Walk up from this script looking for ./VERSION (single source of
    truth shared with the samios CLI, installer, and profiledef.sh).
    Falls back to /etc/samios-version, then a placeholder."""
    here = os.path.dirname(os.path.abspath(__file__))
    d = here
    while d != "/":
        candidate = os.path.join(d, "VERSION")
        if os.path.isfile(candidate):
            try:
                with open(candidate) as f:
                    return f.readline().strip()
            except OSError:
                pass
        d = os.path.dirname(d)
    try:
        with open("/etc/samios-version") as f:
            return f.readline().strip()
    except OSError:
        return "0.0.0-unknown"


VERSION = _resolve_version()
del _resolve_version


class VoiceMode(Enum):
    IDLE = "idle"
    COMMAND = "command"
    DICTATION = "dictation"

class VoiceBridge:
    def __init__(self):
        self.mode = VoiceMode.IDLE
        self.running = False
        self.last_transcript = ""
        self.context = {
            "hostname": os.uname().nodename,
            "user": os.environ.get("USER", "sami"),
        }
        
        # Dictation commands (spoken → action)
        self.dictation_commands = {
            "new paragraph": "Return",
            "new line": "Shift+Return",
            "comma": ",",
            "period": ".",
            "question mark": "?",
            "exclamation mark": "!",
            "colon": ":",
            "semicolon": ";",
            "open quote": '"',
            "close quote": '"',
            "open paren": "(",
            "close paren": ")",
            "tab": "Tab",
            "scratch that": "Ctrl+Z",
            "undo that": "Ctrl+Z",
            "stop dictation": "EXIT_DICTATION",
        }
    
    def start(self):
        """Start the voice bridge"""
        self.running = True
        print(f"[VoiceBridge] SamiOS v{VERSION}")
        print("[VoiceBridge] Ready. Say 'Computer' + command, or use hotkey.")
        
        # Connect to SamiType STT
        # For now, read from stdin (will be SamiType socket later)
        while self.running:
            try:
                line = input()
                if not line.strip():
                    continue
                data = json.loads(line) if line.startswith("{") else {"text": line}
                self.handle_transcript(data)
            except EOFError:
                break
            except KeyboardInterrupt:
                break
    
    def handle_transcript(self, data: dict):
        """Handle incoming transcript from SamiType"""
        text = data.get("text", "").strip()
        confidence = data.get("confidence", 1.0)
        
        if not text:
            return
        
        # Check mode
        if self.mode == VoiceMode.DICTATION:
            self._handle_dictation(text, confidence)
        elif text.lower().startswith("computer"):
            command = text[8:].strip()  # Remove "computer"
            self._handle_command(command, confidence)
        elif text.lower() == "start dictation":
            self._enter_dictation()
        else:
            # Not a command and not in dictation mode — ignore
            pass
    
    def _handle_command(self, command: str, confidence: float):
        """Route command to Hermes API"""
        print(f"[COMMAND] → Hermes: {command}")
        
        # Show low-confidence warning
        if confidence < 0.8:
            print(f"[WARNING] Low confidence ({confidence:.0%}) — verify: {command}")
        
        # Send to Hermes
        result = self._send_to_hermes(command)
        
        # Display result
        if result:
            print(f"[HERMES] {result.get('summary', 'Done.')}")
            # TODO: TTS via SamiType
        else:
            print("[HERMES] No response")
    
    def _handle_dictation(self, text: str, confidence: float):
        """Type text into focused application"""
        lower = text.lower().strip()
        
        # Check for dictation commands
        if lower in self.dictation_commands:
            action = self.dictation_commands[lower]
            if action == "EXIT_DICTATION":
                self._exit_dictation()
                return
            self._send_key(action)
            return
        
        # Type the text
        self._type_text(text)
        
        # Show correction prompt if low confidence
        if confidence < 0.85:
            print(f"[CORRECT?] {text} (confidence: {confidence:.0%})")
    
    def _enter_dictation(self):
        """Enter dictation mode"""
        self.mode = VoiceMode.DICTATION
        print("[MODE] Dictation active — speaking will type into focused app")
        print("[MODE] Say 'stop dictation' to exit")
    
    def _exit_dictation(self):
        """Exit dictation mode"""
        self.mode = VoiceMode.IDLE
        print("[MODE] Dictation stopped")
    
    def _type_text(self, text: str):
        """Type text into the focused application"""
        try:
            subprocess.run(
                ["xdotool", "type", "--clearmodifiers", "--delay", "0", text + " "],
                check=True
            )
        except FileNotFoundError:
            # Fallback: xsel/xclip
            try:
                proc = subprocess.Popen(["xsel", "-b", "-i"], stdin=subprocess.PIPE)
                proc.communicate(text.encode())
                subprocess.run(["xdotool", "key", "ctrl+v"], check=True)
            except Exception as e:
                print(f"[ERROR] Cannot type text: {e}")
    
    def _send_key(self, keysym: str):
        """Send a key combination"""
        try:
            subprocess.run(["xdotool", "key", keysym], check=True)
        except Exception as e:
            print(f"[ERROR] Cannot send key {keysym}: {e}")
    
    def _send_to_hermes(self, command: str) -> dict:
        """Send command to Hermes API"""
        import urllib.request
        
        payload = json.dumps({
            "text": command,
            "source": "voice",
            "context": self.context,
        }).encode()
        
        try:
            req = urllib.request.Request(
                "http://localhost:8765/command",
                data=payload,
                headers={"Content-Type": "application/json"},
                method="POST"
            )
            with urllib.request.urlopen(req, timeout=30) as resp:
                return json.loads(resp.read())
        except Exception as e:
            print(f"[ERROR] Hermes API: {e}")
            # Fallback: CLI
            try:
                result = subprocess.run(
                    ["hermes", "ask", command],
                    capture_output=True, text=True, timeout=60
                )
                return {"summary": result.stdout.strip()}
            except Exception:
                return None

def main():
    if "--version" in sys.argv or "-V" in sys.argv:
        print(f"SamiOS voice-bridge v{VERSION}")
        sys.exit(0)
    bridge = VoiceBridge()
    try:
        bridge.start()
    except KeyboardInterrupt:
        print("\n[VoiceBridge] Stopped.")

if __name__ == "__main__":
    main()