Add voice-bridge.py prototype + systemd service, LibreOffice in package list
SamiOS CI / lint-and-test (push) Successful in 6s
SamiOS CI / lint-and-test (push) Successful in 6s
- voice-bridge.py: SamiType → Hermes router with dictation mode - Handles command mode (→Hermes) and dictation mode (→xdotool typing) - Dictation commands: punctuation, formatting, scratch/undo, stop - Fallback to hermes CLI if API unavailable - systemd service unit for always-on voice
This commit is contained in:
@@ -0,0 +1,16 @@
|
||||
[Unit]
|
||||
Description=SamiOS Voice Bridge (SamiType → Hermes)
|
||||
After=pipewire.service graphical-session.target
|
||||
Wants=pipewire.service
|
||||
|
||||
[Service]
|
||||
Type=simple
|
||||
ExecStart=/usr/bin/python3 /usr/local/bin/voice-bridge.py
|
||||
Restart=on-failure
|
||||
RestartSec=3
|
||||
Environment=DISPLAY=:0
|
||||
Environment=XDG_RUNTIME_DIR=/mnt/wslg/runtime-dir
|
||||
User=sami
|
||||
|
||||
[Install]
|
||||
WantedBy=default.target
|
||||
@@ -0,0 +1,209 @@
|
||||
#!/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
|
||||
|
||||
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
|
||||
|
||||
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("[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():
|
||||
bridge = VoiceBridge()
|
||||
try:
|
||||
bridge.start()
|
||||
except KeyboardInterrupt:
|
||||
print("\n[VoiceBridge] Stopped.")
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user