Bring frame-voice into Frametop as voice/, with its history

Imported from DeeJanuz/frame-voice (branch frametop-keys, 1d88af4).
The next commits make it a Frametop module.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

git-subtree-dir: voice
git-subtree-mainline: b29adb50cc
git-subtree-split: 1d88af4c6e

voice/.gitignore

@@ -0,0 +1,4 @@
+.venv/
+vendor/
+models/
+__pycache__/

voice/README.md

@@ -0,0 +1,159 @@
+# frame-voice
+
+Local voice input for the Steam Frame. Nothing leaves the headset.
+
+- **Dictation**: tap a key (or hold a controller button) and speak. Words are typed as you talk into whatever has focus (desktop apps, search bars, terminals, game-mode windows) through a virtual uinput keyboard, transcribed on the headset by Moonshine.
+- **Wake words**: say one (default "Hey Computer"; add your own in the settings app), then a command. The command is transcribed and handed to a hook script and to any event subscribers, or runs a program you set up for it (see Voice commands). This is the entry point for a voice assistant.
+
+## Pieces
+
+| Part | What it does |
+| --- | --- |
+| `frame-voice-whisper.service` | `whisper-server` (whisper.cpp, Vulkan on the Adreno 750) with `ggml-small.en-q5_1` loaded, on `127.0.0.1:8178` |
+| `frame-voice.service` | `frame_voice.py`: mic capture, hotkey, wake word, typing, control socket |
+| `frame-voice-vr-input.service` | `frame_voice_vr_input.py`: controller push-to-talk, reads left d-pad down from SteamVR's raw input stream |
+| `frame-voice-config` | Settings app (GTK 4 / libadwaita): start and stop the services, start at system startup, pick push-to-talk buttons by pressing them. In the app menu as *Frame Voice Settings* |
+| `frame-voice-ctl` | CLI for the control socket |
+| `hooks/on-command` | Runs after a wake word with the command text as `$1` |
+
+The mic is captured once with `pw-record`. The wake word spotter (sherpa-onnx open-vocabulary KWS, about 5% of one core) listens continuously. Silero VAD ends commands on silence and trims clips before transcription.
+
+## Triggers
+
+| Where | How |
+| --- | --- |
+| Keyboard or mouse | Tap **Right Ctrl** (default) to start, tap again to stop. It counts only on its own: shortcuts with it (Ctrl+C), or with another modifier held when you press it (Shift+Right Ctrl), don't start or stop dictation. Hold mode is available in the settings app. |
+| Steam Frame controllers | Hold left **d-pad down** (default), speak, release. Works in games, the Steam UI and the desktop with no per-app setup. The button is observed, not captured, so whatever else it is bound to still happens. |
+| Desktop mode shortcut | KDE System Settings > Shortcuts > add a command shortcut running `frame-voice-ctl toggle`. |
+| Scripts, the assistant | `frame-voice-ctl toggle`, `start`, `stop`, `cancel`, `type "text"` |
+| Voice | "<wake word>, ..." runs a matching voice command (see Voice commands), otherwise `hooks/on-command`. "Hey Computer, type hello world" types "hello world" (example in the hook). |
+
+Recordings shorter than 2 s (`min_seconds`) are discarded with the cancel tone, so brief or accidental presses never type anything.
+
+*Frame Voice Settings* (`./frame-voice-config`) starts and stops all three services (**Running**) and sets whether they start at system startup (see Services). Change the buttons there too: **Add** waits for you to press a controller button, key or mouse button (not left/right click, not the controller's system button). The app also sets hold or tap mode per trigger, switches live typing and sounds, and manages wake words: add any phrase in English words (they're spelled out for the spotter when loaded, no training), remove them, or turn listening off. Changes apply within a couple of seconds. Everything lives in `~/.config/frame-voice/settings.json`, along with voice commands (see Voice commands); both services apply changes without a restart. Delete the file to go back to the defaults.
+
+Live text works by agreement, with the approach and model borrowed from the Faceclaw app for Even Realities G2 glasses (Moonshine base through sherpa-onnx, on the CPU): the speech since your last pause is re-decoded every 0.8 s, and words two decodes in a row agree on are typed, except the last one, whose punctuation can still change. One word the two decodes disagree on (a filler heard as "a" once and "uh" the next time) doesn't hold back the words after it once three of them agree again. At a 0.8 s pause, or at the quietest point once a decode would pass 8 s of audio (Moonshine returns nothing past about 9 s), the window is decoded one last time and its remaining words are typed.
+
+Windows join up like one recording. Each one is decoded after the last 1.5 s of the one before it (`context_seconds`), so Moonshine hears a sentence that goes on after a pause as the continuation it is. Every decode is lined up with the words already typed, by content rather than by word count, and only what comes after them is new; a phrase you really say twice stays twice. The punctuation after a pause waits until the next window is heard: a period Moonshine put there only because the audio stopped is dropped when the next decode runs on and the word can't end a sentence ("the", "to", "we are"), so "increase the. Width" comes out as "increase the width". A window cut without a pause ends mid-speech, where Moonshine can garble or repeat the cut-off words, so only words the last live decode agrees with are typed then; the rest (up to 5) go to the next window, whose context is made long enough to hold them. After a silence, 0.5 s before the speech is kept, and anything loud in the last second the speech detector didn't flag (a short "I" before a gap), since its speech flag comes about 0.27 s late.
+
+On test clips (two TTS voices, 1 s pauses placed mid-sentence and between sentences, replayed in real time) every mid-sentence seam came out without a stray period or capital. The previous version put a period in 4 of the 6, deleted the phrase said twice, doubled words where a window was cut without a pause, and in one run dropped a whole phrase. The last word of each phrase is typed 0.3-1.8 s after you say it, about the same as before. Moonshine adds fewer commas than whisper. `engine = "whisper"` switches live text to the whisper-server (it dropped words when re-decoded this way in testing). Wake-word commands still use whisper.
+
+Live text never types while Ctrl, Alt, Shift or Super is held on any keyboard: the letters would reach apps as shortcuts. It waits for the release. That is why the keyboard default is tap mode: holding Right Ctrl to talk would hold back the text until you let go. (Pressing and releasing Ctrl on the virtual keyboard to "unhold" it did not work: letters still arrived as Ctrl shortcuts in the Frametop desktop.) When typing holds back, the log says which key on which device it waits for (`typing waits for release of ...`).
+
+Only if you use Frametop: keyboards grabbed by Frametop's input relay (while typing goes to its VR desktop) can't be read directly, so the relay sends their keys to the abstract socket `@frametop_keys`, and frame-voice treats them like keys from any other keyboard. The relay only does this with `SHARE_KEYS=1` in `~/.config/frametop.conf`; without it, the push-to-talk key does nothing while typing goes to the desktop. frame-voice accepts these keys only from your own user, and skips its own typing keyboard, which the relay grabs too. With `SHARE_KEYS=1`, everything typed into the desktop, passwords included, goes to that socket, where any local process could listen while frame-voice isn't running. Without Frametop nothing sends to the socket; keyboards aren't grabbed, and frame-voice reads them directly.
+
+Audio cues are off by default (`[sounds] enabled = true` turns them on): rising tone = recording, falling tone = processing, single high tone = wake word heard, low tone = cancelled or nothing heard.
+
+## Control and events
+
+```
+./frame-voice-ctl status
+./frame-voice-ctl wake off        # stop wake word listening; the mic fully closes when idle
+./frame-voice-ctl wake on
+./frame-voice-ctl wake            # act as if a wake word was heard
+./frame-voice-ctl type "text"     # type into the focused app
+./frame-voice-ctl events          # stream JSON events: wake, command, dictation, command_timeout, error
+```
+
+The socket is `/run/user/<uid>/frame-voice/control.sock` (systemd's runtime dir; the clients ignore `$XDG_RUNTIME_DIR`, which Frametop's desktop points elsewhere), mode 0600. It can type into any focused window, so keep it owner-only.
+
+An assistant can either replace `hooks/on-command` or subscribe with `frame-voice-ctl events` and act on `{"event": "command", "text": ...}`.
+
+## Voice commands
+
+A command that starts with a set phrase can run a program. Add a `"commands"` list to `~/.config/frame-voice/settings.json` by hand (the settings app doesn't edit it, and keeps it when it saves). It is empty by default. Example: "Hey Computer, flag that" or "Hey Computer, flag that, fingers crossed" runs a hand tracking tool's flag command.
+
+```json
+  "commands": [
+    {
+      "say": ["flag that", "flag this", "flag it", "flag"],
+      "run": ["~/.local/bin/ft-handsctl", "flag"]
+    }
+  ]
+```
+
+| Key | Meaning |
+| --- | --- |
+| `say` | Phrases. Case and punctuation don't matter. |
+| `run` | The program and its arguments. `~` is expanded. It runs directly, never through a shell, so spoken words can't become shell syntax. |
+| `rest` | Optional. `"arg"` (default): the words after the phrase are added as one more argument when there are any. `"none"`: they never are. |
+
+Matching works on whole words. A command matches when its words equal a phrase or start with it, so "Flag that." and "Flag that, fingers crossed." match "flag that", but "Flagging that" and "Please flag that" don't. If phrases of several entries match, the longest wins, then the first entry. The rest keeps your words as transcribed, without the punctuation that follows the phrase or a final period: the second example runs `ft-handsctl flag "fingers crossed"`.
+
+A matched command runs instead of `hooks/on-command`. The `command` event is still sent, with `"matched"` (the phrase) and `"ran"` (the argv) added. The program gets these environment variables:
+
+| Variable | Value |
+| --- | --- |
+| `FRAME_VOICE_EVENT` | `voice_command` |
+| `FRAME_VOICE_TEXT` | The whole transcribed command |
+| `FRAME_VOICE_PHRASE` | The phrase that matched, as written in `say` |
+| `FRAME_VOICE_REST` | The words after it, or empty |
+| `FRAME_VOICE_WAKE_UNIX` | When the wake word was heard, in Unix seconds (e.g. `1791582811.212`) |
+
+frame-voice doesn't wait for the program. A malformed entry is skipped with a warning in the log (`journalctl --user -u frame-voice`). Changes to the list apply within a couple of seconds, like the other settings. Updating frame-voice from a version without voice commands needs one `systemctl --user restart frame-voice`.
+
+## Config
+
+Copy `config.example.toml` to `~/.config/frame-voice/config.toml`. Every key is optional. Then `systemctl --user restart frame-voice`.
+
+Wake phrase tuning: raise `threshold` for fewer false triggers, raise `boost` if it misses you. Phrases can be anything spelled in English words; they are tokenized at startup, no training needed.
+
+## Services
+
+**Running** in the settings app starts or stops all three units; **Start at system startup** enables or disables them. Or by hand:
+
+```
+systemctl --user status frame-voice-whisper frame-voice frame-voice-vr-input
+journalctl --user -u frame-voice -f
+systemctl --user stop frame-voice frame-voice-whisper frame-voice-vr-input
+```
+
+The units are linked from `systemd/`. Don't turn startup off with a plain `systemctl --user disable`: it removes those links too, and when a unit file goes missing systemd drops the typing keyboard (see below), even while the service runs. The app does what this does:
+
+```
+systemctl --user disable --no-reload frame-voice-whisper frame-voice frame-voice-vr-input
+systemctl --user link --no-reload ~/Desktop/Projects/frame-voice/systemd/*.service
+systemctl --user daemon-reload
+```
+
+Startup matters for typing into game-mode windows: SteamVR only uses the typing keyboard if it existed when SteamVR started (see Build notes). Enabled, `frame-voice` starts before SteamVR, and stopping and starting it keeps the same keyboard. Disabled, it starts only when you start it, after SteamVR, so restart SteamVR once after each start to type into game-mode windows: systemd keeps a stopped unit's fd store only while the unit is enabled, so each start makes a new keyboard. Typing into Frametop's desktop works either way.
+
+## Build notes
+
+whisper.cpp is in `vendor/whisper.cpp`, built with:
+
+```
+cmake -B build -G Ninja -DCMAKE_BUILD_TYPE=Release -DGGML_VULKAN=ON -DGGML_NATIVE=OFF \
+  -DGGML_CPU_ARM_ARCH=armv9.2-a+dotprod+i8mm+sve2+fp16 \
+  -DCMAKE_PREFIX_PATH=$PWD/../prefix \
+  -DCMAKE_C_FLAGS="-mtune=cortex-a720 -I$PWD/../prefix/include" \
+  -DCMAKE_CXX_FLAGS="-mtune=cortex-a720 -I$PWD/../prefix/include"
+cmake --build build -j5 --target whisper-server whisper-cli whisper-bench
+```
+
+`vendor/prefix` holds SPIRV-Headers, which SteamOS does not ship. Tests: `.venv/bin/python -m unittest` in the repo (no audio stack needed; with the system `python3` the push-to-talk key tests are skipped, since they need `evdev`). The Python side lives in `.venv` (`evdev sherpa-onnx numpy sentencepiece websockets`). `models/sherpa-onnx-moonshine-base-en-quantized-2026-02-27` holds the Moonshine files from huggingface.co/csukuangfj2 (`encoder_model.ort`, `decoder_model_merged.ort`, `tokens.txt`). The settings app uses the system Python for GTK (`gi`) and the venv for the rest; the `frame-voice-config` launcher sets that up.
+
+The whisper service sets `DISABLE_GAMESCOPE_WSI=1`. Without it the Gamescope Vulkan layer aborts whisper-server at startup.
+
+Typing goes through a uinput keyboard that SteamVR must have open: SteamVR opens `/dev/input` devices only when it starts, and gamescope gets keyboard input from SteamVR. `frame-voice.service` is ordered `Before=steamvr.service` and parks the device's fd in systemd's fd store, so daemon restarts keep the same device. After the very first install, restart SteamVR once. The device is a full keyboard (fd store name `typer2`; the first version, `typer`, had only the typing keys and is replaced automatically, which also needs one SteamVR restart). Multidesk's input relay picks the device up immediately for its desktop screens. To drop the device: `systemctl --user clean --what=fdstore frame-voice`.
+
+Controller input: SteamVR sends OpenVR actions only to the app with input focus, and while the 2D Steam or desktop screens are up the compositor's laser mouse holds focus, so a background or overlay OpenVR app never sees the button there. `frame_voice_vr_input.py` instead subscribes to vrserver's local web socket (`ws://127.0.0.1:27062`, mailbox `input_server`, `request_input_state_updates`), the stream the controller binding page uses for its live view. It is undocumented and could change with a SteamVR update. `frame-voice-vr-input.service` has no `After=steamvr.service`: with `WantedBy=default.target` that is an ordering cycle at boot.
+
+## Measurements (2026-09-25, jfk.wav)
+
+Server with the model loaded, `small.en-q5_1` on Vulkan:
+
+| Clip | Latency |
+| --- | --- |
+| 3.2 s phrase | 0.4 s |
+| 11 s sentence | 1.2 s |
+
+Whisper always encodes a 30 s window. The daemon shrinks `audio_ctx` to the clip length (floor 512), which is most of that speedup. The CPU backend took over 20 s. This process only gets CPUs 0-4; the faster cores are reserved.
+
+Bigger models (2026-09-26, 9 real dictations replayed, 512 window, VR running): `medium.en-q5_0` and `large-v3-turbo-q5_0` were no more accurate than `small.en` on this voice (the one difference: "T3 Code Write" vs "right"), took 2-3x longer (about 3 s and 3.5 s per piece versus 1 s), and loading them next to VR made the headset stutter. Turbo also added stray "..." on streamed pieces. Below a 512 window both loop and take 10-20 s. `small.en` stays the default; both bigger models are still in `models/` (about 1.1 GB).
+
+## Limits
+
+- Typing assumes a US keyboard layout and ASCII. Accents are stripped, other characters dropped.
+- Transcription uses the GPU. A heavy VR game may stutter briefly while a clip transcribes. `ggml-base.en.bin` is faster and lighter if that matters.
+- Not yet verified: typing into SteamVR dashboard/overlay text fields and into apps inside VR games.

voice/config.example.toml

@@ -0,0 +1,39 @@
+# Copy to ~/.config/frame-voice/config.toml and edit. Every key is optional.
+# Push-to-talk buttons, hold/tap modes, live typing, sounds, wake words and wake on/off
+# are set in the Frame Voice Settings app (frame-voice-config), saved in settings.json.
+
+[audio]
+source = ""            # PipeWire node name; empty = default mic
+
+[whisper]
+url = "http://127.0.0.1:8178/inference"
+language = "en"
+prompt = "SteamOS, Steam Frame, Godot."   # vocabulary hints
+min_audio_ctx = 512        # smallest encoder window; lower makes medium/turbo loop on short clips
+
+[dictation]
+hold_delay = 0.0           # seconds held before recording starts (0 = on press)
+min_seconds = 2.0          # recordings shorter than this are discarded
+max_seconds = 120
+key_delay_ms = 4
+trailing_space = true
+streaming = true           # transcribe pieces while you talk; only the last one waits for release
+engine = "moonshine"       # live engine: moonshine (CPU, fast) or whisper (GPU server)
+moonshine_threads = 2
+step_seconds = 0.8         # re-decode the current window this often
+pause_seconds = 0.8        # a pause this long closes the window
+max_window_seconds = 8.0   # without a pause, close at the quietest point by this much audio, context included (moonshine max ~9)
+context_seconds = 1.5      # decode each window after this much of the previous one, so a sentence can carry on across a pause
+save_clips = 0             # keep the last N dictations' audio in ~/.local/state/frame-voice/clips
+
+[wake]
+threshold = 0.25           # higher = fewer false triggers, more misses
+boost = 1.5                # higher = easier to trigger
+command_timeout = 5.0      # seconds to start speaking after the wake chime
+end_silence = 0.8          # silence that ends a command
+max_command_seconds = 20
+wake_hook = ""             # run with the keyword as $1 as soon as the wake word fires
+command_hook = "hooks/on-command"   # run with the transcribed command as $1
+
+[sounds]
+volume = 0.5

voice/frame-voice-config

@@ -0,0 +1,4 @@
+#!/bin/sh
+# GTK comes from the system Python; evdev and websockets from the project venv.
+dir=$(dirname "$(readlink -f "$0")")
+PYTHONPATH="$dir:$dir/.venv/lib/python3.12/site-packages" exec /usr/bin/python3 "$dir/frame_voice_config.py" "$@"

voice/frame-voice-ctl

@@ -0,0 +1,23 @@
+#!/usr/bin/env python3
+"""Control the frame-voice daemon: frame-voice-ctl toggle|start|stop|cancel|status|type TEXT|wake [on|off|toggle]|events"""
+import os
+import socket
+import sys
+from pathlib import Path
+
+# Not $XDG_RUNTIME_DIR: Frametop's desktop points it at its own directory.
+path = Path(f"/run/user/{os.getuid()}") / "frame-voice/control.sock"
+if len(sys.argv) < 2 or sys.argv[1] in ("-h", "--help"):
+    sys.exit(__doc__)
+sock = socket.socket(socket.AF_UNIX)
+try:
+    sock.connect(str(path))
+except OSError as err:
+    sys.exit(f"frame-voice not reachable at {path}: {err}")
+sock.sendall((" ".join(sys.argv[1:]) + "\n").encode())
+f = sock.makefile()
+if sys.argv[1] == "events":
+    for line in f:
+        print(line, end="", flush=True)
+else:
+    print(f.readline(), end="")

voice/frame_voice_config.py

@@ -0,0 +1,522 @@
+#!/usr/bin/env python3
+"""Frame Voice settings (GTK 4 / libadwaita): pick push-to-talk buttons by pressing them.
+
+"Add" waits for a controller button (read from vrserver's input stream, like
+frame-voice-vr-input) or a keyboard key / mouse button (read from evdev, like frame-voice)
+and saves ~/.config/frame-voice/settings.json. Both services pick changes up on their
+own. The Service switches start, stop, enable and disable the three systemd user units.
+Run through the `frame-voice-config` launcher: GTK comes from the system Python, evdev
+and websockets from the project venv.
+"""
+import asyncio
+import json
+import os
+import secrets
+import select
+import socket
+import subprocess
+import sys
+import threading
+import time
+import urllib.request
+from pathlib import Path
+
+import gi
+
+gi.require_version("Gtk", "4.0")
+gi.require_version("Adw", "1")
+import evdev  # noqa: E402
+import websockets  # noqa: E402
+from evdev import ecodes as e  # noqa: E402
+from gi.repository import Adw, GLib, Gtk  # noqa: E402
+
+import settings  # noqa: E402
+
+# systemd's runtime dir, where the services and their socket live. Not $XDG_RUNTIME_DIR: Frametop's
+# desktop points that (and the session bus) at its own, where systemctl --user can't reach them.
+USER_RUNTIME = Path(f"/run/user/{os.getuid()}")
+SOCKET_PATH = USER_RUNTIME / "frame-voice/control.sock"
+VRSERVER = "http://127.0.0.1:27062"
+VR_HEADERS = {"Referer": f"{VRSERVER}/dashboard/controllerbinding.html"}
+TYPER_NAME = "frame-voice keyboard"
+CAPTURE_SECONDS = 15
+
+# Clicking "Add" would otherwise bind the mouse button that clicked it.
+IGNORED_KEYS = {e.BTN_LEFT, e.BTN_RIGHT, e.BTN_TOUCH, e.BTN_TOOL_FINGER, e.BTN_TOOL_PEN,
+                e.BTN_TOOL_DOUBLETAP, e.BTN_TOOL_TRIPLETAP, e.BTN_STYLUS}
+# The system button opens the SteamVR dashboard; ft_pointer is Frametop's virtual pointer (its 3D mouse).
+IGNORED_INPUTS = {"/input/system/click"}
+IGNORED_CONTROLLERS = {"ft_pointer"}
+
+INPUT_NAMES = {"dpad_up": "D-pad up", "dpad_down": "D-pad down", "dpad_left": "D-pad left",
+               "dpad_right": "D-pad right", "thumbstick": "Thumbstick press", "trackpad": "Trackpad press",
+               "trigger": "Trigger", "grip": "Grip", "bumper": "Bumper", "thumbrest": "Thumb rest",
+               "view": "View", "menu": "Menu", "system": "System"}
+KEY_NAMES = {"CTRL": "Ctrl", "ALT": "Alt", "SHIFT": "Shift", "META": "Super", "SIDE": "side button",
+             "EXTRA": "extra button", "FORWARD": "forward button", "BACK": "back button",
+             "MIDDLE": "middle button", "CAPSLOCK": "Caps Lock", "SCROLLLOCK": "Scroll Lock",
+             "PAUSE": "Pause", "ESC": "Esc", "SPACE": "Space", "ENTER": "Enter"}
+
+
+def key_label(name):
+    base = name.split("_", 1)[1] if "_" in name else name
+    if name.startswith("BTN_"):
+        return "Mouse " + KEY_NAMES.get(base, base.lower().replace("_", " "))
+    for side in ("LEFT", "RIGHT"):
+        rest = base[len(side):]
+        if base.startswith(side) and rest in KEY_NAMES:
+            return f"{side.title()} {KEY_NAMES[rest]}"
+    return KEY_NAMES.get(base, base.title() if len(base) > 1 else base)
+
+
+def controller_label(device, input_path):
+    hand = device.rsplit("/", 1)[-1]
+    part = input_path.split("/")[2] if input_path.count("/") >= 2 else input_path
+    name = INPUT_NAMES.get(part, part.upper() if len(part) == 1 else part.replace("_", " ").capitalize())
+    return f"{hand.title() if hand in ('left', 'right') else device} controller: {name}"
+
+
+MODE_LABELS = {"toggle": "Tap to start, tap to stop", "hold": "Hold to talk"}
+MODE_HINTS = {"toggle": "Tap to start, tap again to stop.", "hold": "Hold to talk, release to stop."}
+
+
+def key_note(name, mode):
+    if name in settings.MODIFIERS:
+        if mode == "hold":
+            return "While it's held, text waits for release (it would turn letters into shortcuts)"
+        return "Counts only on its own: not in shortcuts like Ctrl+C, or with another modifier held"
+    if name.startswith("KEY_") and (len(name) == 5 or name in ("KEY_SPACE", "KEY_ENTER")):
+        return "This key also types into apps"
+    return name
+
+
+def device_label(root_path, controller_type):
+    name = controller_type.replace("_", " ").title().replace("Frame", "Steam Frame")
+    hand = root_path.rsplit("/", 1)[-1]
+    # A controller without a hand role (Frametop's pointer holds one) is known by its serial.
+    return f"{hand.title()} {name}" if hand in ("left", "right") else f"{name} ({hand})"
+
+
+def vr_devices():
+    """Connected controllers as SteamVR reports them, or None when SteamVR is not running."""
+    req = urllib.request.Request(f"{VRSERVER}/input/getstate.json", headers=VR_HEADERS)
+    try:
+        with urllib.request.urlopen(req, timeout=3) as resp:
+            state = json.load(resp)
+    except OSError:
+        return None
+    return [{"device": d["root_path"], "controller_type": d["controller_type"],
+             "label": device_label(d["root_path"], d["controller_type"])}
+            for d in state.get("devices", [])
+            if d.get("class") == "TrackedDeviceClass_Controller" and d.get("root_path")
+            and d.get("controller_type") not in IGNORED_CONTROLLERS]
+
+
+UNIT_DIR = Path(__file__).resolve().parent / "systemd"
+UNITS = ("frame-voice-whisper.service", "frame-voice.service", "frame-voice-vr-input.service")
+
+
+def systemctl(*args):
+    env = {**os.environ, "XDG_RUNTIME_DIR": str(USER_RUNTIME), "DBUS_SESSION_BUS_ADDRESS": f"unix:path={USER_RUNTIME}/bus"}
+    try:
+        return subprocess.run(["systemctl", "--user", *args], capture_output=True, text=True, timeout=90, env=env)
+    except (OSError, subprocess.TimeoutExpired) as err:
+        return subprocess.CompletedProcess(args, 1, "", str(err))
+
+
+def service_state():
+    """unit -> {"LoadState", "ActiveState", "UnitFileState"}, in UNITS order."""
+    out = systemctl("show", "-p", "Id,LoadState,ActiveState,UnitFileState", *UNITS).stdout
+    blocks = [dict(line.split("=", 1) for line in block.splitlines() if "=" in line) for block in out.split("\n\n")]
+    return {unit: next((b for b in blocks if b.get("Id") == unit), {}) for unit in UNITS}
+
+
+def service_action(action):
+    """systemctl start, stop, enable or disable all three units. Returns an error or None.
+
+    SteamVR only sees the typing keyboard if it existed when SteamVR started. It lives in
+    frame-voice's fd store, which systemd keeps across a stop only while the unit is enabled,
+    and drops when the unit file goes missing, even while running. The units are linked from
+    the repo and `systemctl disable` removes those links too, so they are linked again before
+    systemd reloads."""
+    steps = [(action, *UNITS)]
+    if action in ("start", "enable"):  # links missing (never installed, or removed by hand)
+        state = service_state()
+        missing = [str(UNIT_DIR / u) for u in UNITS if state[u].get("LoadState") != "loaded"]
+        steps[:0] = [("link", *missing)] if missing else []
+    elif action == "disable":
+        steps = [("disable", "--no-reload", *UNITS), ("link", "--no-reload", *(str(UNIT_DIR / u) for u in UNITS)),
+                 ("daemon-reload",)]
+    for args in steps:
+        result = systemctl(*args)
+        if result.returncode:
+            return (result.stderr.strip() or f"systemctl {args[0]} failed").splitlines()[0]
+    return None
+
+
+def notify_daemon():
+    try:
+        with socket.socket(socket.AF_UNIX) as sock:
+            sock.settimeout(2)
+            sock.connect(str(SOCKET_PATH))
+            sock.sendall(b"reload\n")
+            sock.recv(4096)
+        return True
+    except OSError:
+        return False
+
+
+class Capture:
+    """One capture at a time; a new request or /cancel ends the previous one."""
+
+    def __init__(self):
+        self.lock = threading.Lock()
+        self.cancel = threading.Event()
+
+    def begin(self):
+        self.cancel.set()
+        self.lock.acquire()
+        self.cancel = threading.Event()
+        return self.cancel
+
+    def end(self):
+        self.lock.release()
+
+    def keyboard(self):
+        cancel = self.begin()
+        devices = []
+        try:
+            for path in evdev.list_devices():
+                try:
+                    dev = evdev.InputDevice(path)
+                except OSError:
+                    continue
+                if dev.name == TYPER_NAME or e.EV_KEY not in dev.capabilities():
+                    dev.close()
+                    continue
+                devices.append(dev)
+            for dev in devices:  # drop anything already queued
+                try:
+                    while dev.read_one():
+                        pass
+                except OSError:
+                    pass
+            deadline = time.monotonic() + CAPTURE_SECONDS
+            while time.monotonic() < deadline and not cancel.is_set():
+                ready, _, _ = select.select(devices, [], [], 0.2)
+                for dev in ready:
+                    try:
+                        events = list(dev.read())
+                    except OSError:
+                        continue
+                    for ev in events:
+                        if ev.type != e.EV_KEY or ev.value != 1 or ev.code in IGNORED_KEYS:
+                            continue
+                        if ev.code == e.KEY_ESC:
+                            return {"cancelled": True}
+                        name = e.KEY.get(ev.code) or e.BTN.get(ev.code)
+                        if isinstance(name, list):
+                            name = name[0]
+                        if name:
+                            return {"code": name, "label": key_label(name), "source": dev.name}
+            return {"cancelled": True} if cancel.is_set() else {"timeout": True}
+        finally:
+            for dev in devices:
+                dev.close()
+            self.end()
+
+    def controller(self):
+        cancel = self.begin()
+        try:
+            devices = vr_devices()
+            if devices is None:
+                return {"error": "SteamVR is not running."}
+            if not devices:
+                return {"error": "No controller is connected. Wake a controller and try again."}
+            return asyncio.run(self._controller(cancel, devices))
+        except (OSError, websockets.WebSocketException) as err:
+            return {"error": f"Could not read controllers from SteamVR: {err}"}
+        finally:
+            self.end()
+
+    async def _controller(self, cancel, devices):
+        mailbox = f"frame_voice_config_{os.getpid()}_{secrets.token_hex(3)}"
+        types = {d["device"]: d["controller_type"] for d in devices}
+        async with websockets.connect(VRSERVER.replace("http", "ws"), origin=VRSERVER, max_size=None,
+                                      additional_headers=VR_HEADERS) as ws:
+            await ws.send(f"mailbox_open {mailbox}")
+            for device in types:
+                message = {"type": "request_input_state_updates", "device_path": device, "returnAddress": mailbox}
+                await ws.send(f"mailbox_send input_server {json.dumps(message)}")
+            released = {}  # device -> inputs seen up; a button held when capture starts doesn't count
+            deadline = time.monotonic() + CAPTURE_SECONDS
+            while time.monotonic() < deadline and not cancel.is_set():
+                try:
+                    message = json.loads(await asyncio.wait_for(ws.recv(), 0.2))
+                except asyncio.TimeoutError:
+                    continue
+                device = message.get("device")
+                if message.get("type") != "update_component_states" or device not in types:
+                    continue
+                seen = released.setdefault(device, set())
+                for name, value in message.get("components", {}).items():
+                    if not name.endswith("/click") or name in IGNORED_INPUTS:
+                        continue
+                    if not value:
+                        seen.add(name)
+                    elif name in seen:
+                        return {"device": device, "input": name, "controller_type": types[device],
+                                "label": controller_label(device, name)}
+            return {"cancelled": True} if cancel.is_set() else {"timeout": True}
+
+
+class Window(Adw.ApplicationWindow):
+    def __init__(self, app):
+        super().__init__(application=app, title="Frame Voice", default_width=560, default_height=640)
+        self.capture = Capture()
+        self.data = settings.load()
+        self.rows = []
+        self.dialog = None
+
+        self.service_busy = False  # a start/stop/enable/disable is running
+        self.polling = False
+        self.syncing = False  # switches being set from the service state, not by the user
+
+        self.controller_status = "Checking for controllers…"
+        self.controller_group = Adw.PreferencesGroup(title="Controller")
+        self.keyboard_group = Adw.PreferencesGroup(title="Keyboard and mouse")
+        for group, kind in ((self.controller_group, "controller"), (self.keyboard_group, "keyboard")):
+            add = Gtk.Button(label="Add", css_classes=["flat"])
+            add.connect("clicked", lambda _b, k=kind: self.start_capture(k))
+            group.set_header_suffix(add)
+        page = Adw.PreferencesPage()
+        self.wake_group = self._wake_group()
+        for group in (self._service_group(), self.controller_group, self.keyboard_group, self.wake_group,
+                      self._behavior_group()):
+            page.add(group)
+        self.toasts = Adw.ToastOverlay(child=page)
+        view = Adw.ToolbarView(content=self.toasts)
+        view.add_top_bar(Adw.HeaderBar())
+        self.set_content(view)
+        self.refresh()
+        threading.Thread(target=self._load_controllers, daemon=True).start()
+        self._poll_service()
+        GLib.timeout_add_seconds(2, self._poll_service)
+
+    def _service_group(self):
+        group = Adw.PreferencesGroup(title="Service")
+        self.running_row = Adw.SwitchRow(title="Running", subtitle="Checking…")
+        self.running_row.connect("notify::active", lambda r, _p: self.syncing or self.run_service(
+            "start" if r.get_active() else "stop"))
+        self.startup_row = Adw.SwitchRow(title="Start at system startup")
+        self.startup_row.connect("notify::active", lambda r, _p: self.syncing or self.run_service(
+            "enable" if r.get_active() else "disable"))
+        group.add(self.running_row)
+        group.add(self.startup_row)
+        return group
+
+    def _poll_service(self):
+        if not self.polling and not self.service_busy:
+            self.polling = True
+
+            def work():
+                state = service_state()
+                GLib.idle_add(self._polled, state)
+            threading.Thread(target=work, daemon=True).start()
+        return True  # keep polling: systemctl or a reboot can change it too
+
+    def _polled(self, state):
+        self.polling = False
+        self.show_service(state)
+
+    def show_service(self, state):
+        if self.service_busy:
+            return
+        active = {u: state[u].get("ActiveState", "unknown") for u in UNITS}
+        on = active["frame-voice.service"] in ("active", "activating", "reloading")
+        failed = [u for u in UNITS if active[u] == "failed"]
+        stopped = [u for u in UNITS if active[u] not in ("active", "activating", "reloading")]
+        if failed:
+            subtitle = f"{failed[0].removesuffix('.service')} failed. Log: journalctl --user -u {failed[0]}"
+        elif on and stopped:
+            subtitle = "Partly running: " + ", ".join(u.removesuffix(".service") for u in stopped) + " stopped"
+        elif on:
+            subtitle = "Dictation, wake words and controller push-to-talk are on"
+        else:
+            subtitle = "Stopped"
+        enabled = state["frame-voice.service"].get("UnitFileState") == "enabled"
+        self.syncing = True
+        self.running_row.set_active(on)
+        self.startup_row.set_active(enabled)
+        self.syncing = False
+        self.running_row.set_subtitle(subtitle)
+        self.startup_row.set_subtitle(
+            "Starts before SteamVR, so typing reaches game-mode windows" if enabled else
+            "Off: restart SteamVR after starting Frame Voice to type into game-mode windows")
+
+    def run_service(self, action):
+        self.service_busy = True
+        row = self.running_row if action in ("start", "stop") else self.startup_row
+        row.set_subtitle({"start": "Starting…", "stop": "Stopping…", "enable": "Enabling…", "disable": "Disabling…"}[action])
+        self.running_row.set_sensitive(False)
+        self.startup_row.set_sensitive(False)
+
+        def work():
+            error = service_action(action)
+            GLib.idle_add(self._service_done, action, error, service_state())
+        threading.Thread(target=work, daemon=True).start()
+
+    def _service_done(self, action, error, state):
+        self.service_busy = False
+        self.running_row.set_sensitive(True)
+        self.startup_row.set_sensitive(True)
+        self.show_service(state)
+        done = {"start": "Frame Voice started.", "stop": "Frame Voice stopped.",
+                "enable": "Frame Voice will start at system startup.",
+                "disable": "Frame Voice won't start at system startup."}
+        self.toast(f"Couldn't {action} Frame Voice: {error}" if error else done[action])
+
+    def _wake_group(self):
+        group = Adw.PreferencesGroup(
+            title="Wake words",
+            description="Say one, then a command; the command goes to hooks/on-command and the event "
+                        "stream. Any English words work. Two or more words trigger by accident less.")
+        switch = Adw.SwitchRow(title="Listen for wake words", subtitle="While not dictating",
+                               active=self.data["wake_word"])
+        switch.connect("notify::active", lambda r, _p: self.set_option("wake_word", r.get_active()))
+        group.add(switch)
+        self.wake_entry = Adw.EntryRow(title="Add a wake word, like \u201cHey Computer\u201d", show_apply_button=True)
+        self.wake_entry.connect("apply", self.add_wake_phrase)
+        return group
+
+    def add_wake_phrase(self, entry):
+        phrase = settings.clean_phrase(entry.get_text())
+        entry.set_text("")
+        if not phrase:
+            self.toast("Use letters and spaces.")
+        elif phrase in self.data["wake_phrases"]:
+            self.toast(f"\u201c{phrase.title()}\u201d is already a wake word.")
+        else:
+            self.data["wake_phrases"].append(phrase)
+            self.save(f"Added \u201c{phrase.title()}\u201d.")
+
+    def _behavior_group(self):
+        group = Adw.PreferencesGroup(title="Behavior",
+                                     description="Recordings shorter than 2 seconds are discarded.")
+        modes = list(MODE_LABELS)
+        for key, title in (("keyboard_mode", "Keyboard and mouse"), ("controller_mode", "Controller")):
+            row = Adw.ComboRow(title=title, model=Gtk.StringList.new([MODE_LABELS[m] for m in modes]),
+                               selected=modes.index(self.data[key]))
+            row.connect("notify::selected", lambda r, _p, k=key: self.set_option(k, modes[r.get_selected()]))
+            group.add(row)
+        for key, title, subtitle in (
+                ("live_typing", "Type while talking", "Text appears every few seconds instead of after you stop"),
+                ("sounds", "Sounds", "Tones when recording starts and stops")):
+            row = Adw.SwitchRow(title=title, subtitle=subtitle, active=self.data[key])
+            row.connect("notify::active", lambda r, _p, k=key: self.set_option(k, r.get_active()))
+            group.add(row)
+        return group
+
+    def set_option(self, key, value):
+        self.data[key] = value
+        self.save("Saved.")
+
+    def _load_controllers(self):
+        devices = vr_devices()
+        if devices is None:
+            text = "SteamVR is not running."
+        elif not devices:
+            text = "No controller connected. Wake one to add its buttons."
+        else:
+            text = "Connected: " + ", ".join(d["label"] for d in devices) + "."
+        self.controller_status = text
+        GLib.idle_add(self.refresh)
+
+    def refresh(self):
+        for group, row in self.rows:
+            group.remove(row)
+        self.rows = []
+        entries = [(self.controller_group, controller_label(c["device"], c["input"]), f"{c['device']} {c['input']}", ("controller", i))
+                   for i, c in enumerate(self.data["controller"])]
+        self.controller_group.set_description(MODE_HINTS[self.data["controller_mode"]] + " " + self.controller_status)
+        self.keyboard_group.set_description(MODE_HINTS[self.data["keyboard_mode"]])
+        entries += [(self.keyboard_group, key_label(k), key_note(k, self.data["keyboard_mode"]), ("keyboard", i))
+                    for i, k in enumerate(self.data["keyboard"])]
+        entries += [(self.wake_group, p.title(), "", ("wake_phrases", i)) for i, p in enumerate(self.data["wake_phrases"])]
+        for group, title, subtitle, (kind, index) in entries:
+            row = Adw.ActionRow(title=title)
+            if subtitle:
+                row.set_subtitle(subtitle)
+            remove = Gtk.Button(icon_name="user-trash-symbolic", valign=Gtk.Align.CENTER,
+                                css_classes=["flat"], tooltip_text="Remove")
+            remove.connect("clicked", lambda _b, k=kind, i=index: self.remove(k, i))
+            row.add_suffix(remove)
+            group.add(row)
+            self.rows.append((group, row))
+        for group, kind in ((self.controller_group, "controller"), (self.keyboard_group, "keyboard")):
+            if not self.data[kind]:
+                row = Adw.ActionRow(title="Nothing bound", subtitle="Use Add to pick a button", sensitive=False)
+                group.add(row)
+                self.rows.append((group, row))
+        self.wake_group.add(self.wake_entry)  # keep the entry below the phrases
+        self.rows.append((self.wake_group, self.wake_entry))
+
+    def save(self, message):
+        settings.save(self.data)
+        live = notify_daemon()
+        self.refresh()
+        self.toast(message + (" Active now." if live else " frame-voice is not running."))
+
+    def toast(self, text):
+        self.toasts.add_toast(Adw.Toast(title=text, timeout=3))
+
+    def remove(self, kind, index):
+        del self.data[kind][index]
+        self.save("Removed.")
+
+    def start_capture(self, kind):
+        what = "controller button" if kind == "controller" else "key or mouse button"
+        self.dialog = Adw.AlertDialog(heading=f"Press a {what}",
+                                      body=f"Press the {what} you want to use for dictation.")
+        self.dialog.set_extra_child(Gtk.Spinner(spinning=True, height_request=32))
+        self.dialog.add_response("cancel", "Cancel")
+        self.dialog.connect("response", lambda *_: self.capture.cancel.set())
+        self.dialog.present(self)
+
+        def work():
+            result = self.capture.controller() if kind == "controller" else self.capture.keyboard()
+            GLib.idle_add(self.capture_done, kind, result)
+        threading.Thread(target=work, daemon=True).start()
+
+    def capture_done(self, kind, result):
+        if self.dialog:
+            self.dialog.force_close()
+            self.dialog = None
+        if "error" in result:
+            self.toast(result["error"])
+        elif "timeout" in result:
+            self.toast("Nothing pressed.")
+        elif kind == "controller" and "input" in result:
+            entry = {"device": result["device"], "input": result["input"]}
+            if entry in self.data["controller"]:
+                self.toast(f"{result['label']} is already bound.")
+            else:
+                self.data["controller"].append(entry)
+                self.save(f"Added {result['label']}.")
+        elif kind == "keyboard" and "code" in result:
+            if result["code"] in self.data["keyboard"]:
+                self.toast(f"{result['label']} is already bound.")
+            else:
+                self.data["keyboard"].append(result["code"])
+                self.save(f"Added {result['label']}.")
+        return False
+
+
+def main():
+    app = Adw.Application(application_id="local.framevoice.Config")
+    app.connect("activate", lambda a: (a.get_active_window() or Window(a)).present())
+    return app.run(sys.argv)
+
+
+if __name__ == "__main__":
+    sys.exit(main())

voice/frame_voice_vr_input.py

@@ -0,0 +1,152 @@
+#!/usr/bin/env python3
+"""Controller push-to-talk for frame-voice: hold a bound button (or tap it, in toggle mode).
+
+SteamVR only gives controller input to the app that has focus, and while the 2D Steam
+or desktop screens are up the compositor's laser mouse holds it, so an OpenVR action
+never fires there. Instead this reads raw component state from vrserver's local web
+socket (127.0.0.1:27062), the same stream the controller binding page uses for its live
+view. It sees the button whatever has focus. Press -> `start vr`, release -> `stop` on
+the frame-voice control socket; the daemon applies the 2 s minimum.
+
+The button is observed, not captured: whatever else is bound to it still happens.
+Buttons come from ~/.config/frame-voice/settings.json (see settings.py and the
+frame-voice-config app); changes apply within a second.
+"""
+import asyncio
+import json
+import logging
+import os
+import socket
+import time
+import urllib.request
+from pathlib import Path
+
+import websockets
+
+import settings
+
+SOCKET_PATH = Path(os.environ.get("XDG_RUNTIME_DIR", f"/run/user/{os.getuid()}")) / "frame-voice/control.sock"
+VRSERVER = "http://127.0.0.1:27062"
+MAILBOX = f"frame_voice_ptt_{os.getpid()}"
+VR_HEADERS = {"Referer": f"{VRSERVER}/dashboard/controllerbinding.html"}
+DEVICE_POLL = 5  # seconds between checks for controllers connecting or leaving
+
+log = logging.getLogger("frame-voice-vr-input")
+
+
+def send(command):
+    try:
+        with socket.socket(socket.AF_UNIX) as sock:
+            sock.settimeout(2)
+            sock.connect(str(SOCKET_PATH))
+            sock.sendall((command + "\n").encode())
+            sock.recv(4096)
+    except OSError as err:
+        log.warning("frame-voice not reachable (%s): %s", command, err)
+
+
+async def connect():
+    # vrserver serves the binding page, and its socket, to local clients from that page.
+    return await websockets.connect(VRSERVER.replace("http", "ws"), origin=VRSERVER, max_size=None,
+                                    additional_headers=VR_HEADERS)
+
+
+def connected():
+    """Root paths of the controllers SteamVR has right now (vrserver only streams changes,
+    so a quiet stream says nothing about whether a controller is still there)."""
+    req = urllib.request.Request(f"{VRSERVER}/input/getstate.json", headers=VR_HEADERS)
+    with urllib.request.urlopen(req, timeout=3) as resp:
+        return {d["root_path"] for d in json.load(resp).get("devices", []) if d.get("root_path")}
+
+
+async def subscribe(ws, device, mailbox, on=True):
+    kind = "request_input_state_updates" if on else "cancel_input_state_updates"
+    message = {"type": kind, "device_path": device, "returnAddress": mailbox}
+    await ws.send(f"mailbox_send input_server {json.dumps(message)}")
+
+
+def wanted():
+    """device -> set of inputs, from settings.json."""
+    out = {}
+    for b in settings.load()["controller"]:
+        out.setdefault(b["device"], set()).add(b["input"])
+    return out
+
+
+async def run_session():
+    """One connection to vrserver. Returns or raises when SteamVR goes away."""
+    async with await connect() as ws:
+        await ws.send(f"mailbox_open {MAILBOX}")
+        stamp, inputs, mode = settings.mtime(), wanted(), settings.load()["controller_mode"]
+        for device in inputs:
+            await subscribe(ws, device, MAILBOX)
+        log.info("connected to SteamVR, watching %s (%s)", describe(inputs), mode)
+        held = False
+        pressed = set()  # (device, input) currently down
+        present = await asyncio.to_thread(connected)
+        next_poll = time.monotonic() + DEVICE_POLL
+        try:
+            while True:
+                try:
+                    raw = await asyncio.wait_for(ws.recv(), 1)
+                except asyncio.TimeoutError:
+                    raw = None
+                now = time.monotonic()
+                if raw is not None:
+                    message = json.loads(raw)
+                    device = message.get("device")
+                    if message.get("type") == "update_component_states" and device in inputs:
+                        components = message.get("components", {})
+                        for name in inputs[device]:
+                            if components.get(name):
+                                pressed.add((device, name))
+                            elif name in components:
+                                pressed.discard((device, name))
+                if settings.mtime() != stamp:
+                    stamp, new, mode = settings.mtime(), wanted(), settings.load()["controller_mode"]
+                    for device in inputs.keys() - new.keys():
+                        await subscribe(ws, device, MAILBOX, on=False)
+                    for device in new.keys() - inputs.keys():
+                        await subscribe(ws, device, MAILBOX)
+                    inputs = new
+                    pressed = {(d, i) for d, i in pressed if i in inputs.get(d, ())}
+                    log.info("settings changed, watching %s (%s)", describe(inputs), mode)
+                if now >= next_poll:
+                    next_poll = now + DEVICE_POLL
+                    current = await asyncio.to_thread(connected)
+                    for device in inputs.keys() & (current - present):
+                        log.info("%s connected", device)
+                        await subscribe(ws, device, MAILBOX, on=False)
+                        await subscribe(ws, device, MAILBOX)
+                    for device in inputs.keys() & (present - current):
+                        log.info("%s disconnected", device)
+                        pressed = {(d, i) for d, i in pressed if d != device}
+                    present = current
+                if pressed and not held:
+                    held = True
+                    send("toggle vr" if mode == "toggle" else "start vr")
+                elif held and not pressed:
+                    held = False
+                    if mode == "hold":
+                        send("stop")
+        finally:
+            if held and mode == "hold":
+                send("stop")
+
+
+def describe(inputs):
+    return ", ".join(f"{d} {i}" for d, names in inputs.items() for i in sorted(names)) or "nothing"
+
+
+def main():
+    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
+    while True:
+        try:
+            asyncio.run(run_session())
+        except (OSError, websockets.WebSocketException) as err:
+            log.info("SteamVR not reachable (%s); retrying", err)
+        time.sleep(5)
+
+
+if __name__ == "__main__":
+    main()

voice/hooks/on-command

@@ -0,0 +1,14 @@
+#!/usr/bin/env bash
+# Called after a wake word with the transcribed command as $1.
+# Replace this with a call into the assistant. Runs as your user, no shell interpolation of $1.
+set -euo pipefail
+text="$1"
+log_dir="${XDG_STATE_HOME:-$HOME/.local/state}/frame-voice"
+mkdir -p "$log_dir"
+printf '%s\t%s\n' "$(date -Is)" "$text" >> "$log_dir/commands.log"
+
+# Built-in example: "<wake word>, type hello world" types into the focused field.
+shopt -s nocasematch
+if [[ "$text" =~ ^(type|dictate)[[:space:],:]+(.*)$ ]]; then
+    exec "$(dirname "$0")/../frame-voice-ctl" type "${BASH_REMATCH[2]}"
+fi

voice/settings.py

@@ -0,0 +1,97 @@
+"""Push-to-talk bindings and settings shared by frame-voice, frame-voice-vr-input and frame-voice-config.
+
+~/.config/frame-voice/settings.json (written by the frame-voice-config app):
+  {"keyboard": ["KEY_RIGHTCTRL", "BTN_SIDE"],
+   "controller": [{"device": "/user/hand/left", "input": "/input/dpad_down/click"}],
+   "keyboard_mode": "toggle", "controller_mode": "hold",
+   "live_typing": true, "sounds": false, "wake_word": true,
+   "wake_phrases": ["HEY COMPUTER"],
+   "commands": [{"say": ["flag that", "flag"], "run": ["~/.local/bin/my-tool", "flag"]}]}
+
+keyboard: evdev key or mouse button names, read from every physical keyboard and mouse.
+controller: a SteamVR device path and one of its boolean inputs, read from vrserver.
+*_mode: "hold" records while held; "toggle" starts on one press and stops on the next.
+wake_phrases: what to say before a command; any English words, spelled out at load time.
+commands: wake word commands that run a program (see voice_commands.py). Edited by hand;
+the settings app keeps them as they are.
+The options override config.toml.
+"""
+import json
+import os
+import re
+from pathlib import Path
+
+# Not under $XDG_CONFIG_HOME: Frametop's desktop points it at ~/.config/frametop, and the
+# services (started by systemd) and the settings app must read the same file.
+PATH = Path(os.environ.get("FRAME_VOICE_SETTINGS", "~/.config/frame-voice/settings.json")).expanduser()
+DEFAULT = {
+    "keyboard": ["KEY_RIGHTCTRL"],
+    "controller": [{"device": "/user/hand/left", "input": "/input/dpad_down/click"}],
+    # Toggle for keyboards: live text can't be typed while a modifier like Ctrl is held.
+    "keyboard_mode": "toggle",
+    "controller_mode": "hold",
+    "live_typing": True,
+    "sounds": False,
+    "wake_word": True,
+    "wake_phrases": ["HEY COMPUTER"],
+}
+MODES = ("hold", "toggle")
+OPTIONS = ("live_typing", "sounds", "wake_word")
+# Held modifiers turn typed letters into shortcuts, so live typing waits until they are released.
+MODIFIERS = {"KEY_LEFTCTRL", "KEY_RIGHTCTRL", "KEY_LEFTALT", "KEY_RIGHTALT",
+             "KEY_LEFTSHIFT", "KEY_RIGHTSHIFT", "KEY_LEFTMETA", "KEY_RIGHTMETA"}
+
+
+def clean_phrase(text):
+    """A wake phrase as the keyword spotter wants it: uppercase words, letters and apostrophes."""
+    return " ".join(re.sub(r"[^A-Z' ]", " ", text.upper()).split())
+
+
+def load():
+    try:
+        data = json.loads(PATH.read_text())
+    except (FileNotFoundError, ValueError):
+        data = {}
+    if not isinstance(data, dict):
+        data = {}
+    out = json.loads(json.dumps(DEFAULT))
+    if isinstance(data.get("keyboard"), list):
+        out["keyboard"] = [k for k in data["keyboard"] if isinstance(k, str)]
+    if isinstance(data.get("controller"), list):
+        out["controller"] = [c for c in data["controller"] if isinstance(c, dict)
+                             and isinstance(c.get("device"), str) and isinstance(c.get("input"), str)]
+    for key in ("keyboard_mode", "controller_mode"):
+        if data.get(key) in MODES:
+            out[key] = data[key]
+    for key in OPTIONS:
+        if isinstance(data.get(key), bool):
+            out[key] = data[key]
+    if isinstance(data.get("wake_phrases"), list):
+        phrases = (clean_phrase(p) for p in data["wake_phrases"] if isinstance(p, str))
+        out["wake_phrases"] = list(dict.fromkeys(p for p in phrases if p))
+    # Checked by voice_commands.parse(), which warns about each bad entry.
+    out["commands"] = data.get("commands", [])
+    return out
+
+
+def save(data):
+    """Write the keys the settings app manages (DEFAULT). Every other key, "commands"
+    included, keeps its value from the file, so hand edits survive a save from the app."""
+    try:
+        on_disk = json.loads(PATH.read_text())
+    except (FileNotFoundError, ValueError):
+        on_disk = {}
+    if not isinstance(on_disk, dict):
+        on_disk = {}
+    merged = {**on_disk, **{k: data[k] for k in DEFAULT if k in data}}
+    PATH.parent.mkdir(parents=True, exist_ok=True)
+    tmp = PATH.with_suffix(".tmp")
+    tmp.write_text(json.dumps(merged, indent=2) + "\n")
+    tmp.replace(PATH)
+
+
+def mtime():
+    try:
+        return PATH.stat().st_mtime_ns
+    except FileNotFoundError:
+        return 0

voice/systemd/frame-voice-vr-input.service

@@ -0,0 +1,14 @@
+[Unit]
+Description=frame-voice controller push-to-talk
+# No After=steamvr.service: default.target is ordered after its wants and steamvr after
+# the graphical session, so that would be an ordering cycle. The script retries until
+# SteamVR is up.
+After=frame-voice.service
+
+[Service]
+ExecStart=%h/Desktop/Projects/frame-voice/.venv/bin/python %h/Desktop/Projects/frame-voice/frame_voice_vr_input.py
+Restart=always
+RestartSec=5
+
+[Install]
+WantedBy=default.target steamvr.service

voice/systemd/frame-voice-whisper.service

@@ -0,0 +1,15 @@
+[Unit]
+Description=whisper.cpp server for frame-voice
+
+[Service]
+# The Gamescope WSI layer asserts on compute-only Vulkan devices; whisper needs no swapchain.
+Environment=DISABLE_GAMESCOPE_WSI=1
+# The binary's RUNPATH is the directory it was built in; this keeps it working if the
+# project folder moves.
+Environment=LD_LIBRARY_PATH=%h/Desktop/Projects/frame-voice/vendor/whisper.cpp/build/bin
+ExecStart=%h/Desktop/Projects/frame-voice/vendor/whisper.cpp/build/bin/whisper-server -m %h/Desktop/Projects/frame-voice/models/ggml-small.en-q5_1.bin --host 127.0.0.1 --port 8178 -t 3 -fa
+Restart=on-failure
+Nice=5
+
+[Install]
+WantedBy=default.target

voice/systemd/frame-voice.service

@@ -0,0 +1,22 @@
+[Unit]
+Description=frame-voice dictation and wake word daemon
+Wants=frame-voice-whisper.service pipewire.service
+After=frame-voice-whisper.service pipewire.service
+# SteamVR opens input devices only at startup, and gamescope gets keyboard input through
+# SteamVR, so the typing keyboard must exist before SteamVR starts.
+Before=steamvr.service
+
+[Service]
+# READY=1 is sent after the virtual keyboard exists.
+Type=notify
+NotifyAccess=main
+ExecStart=%h/Desktop/Projects/frame-voice/.venv/bin/python %h/Desktop/Projects/frame-voice/frame_voice.py
+Restart=on-failure
+RestartSec=2
+# systemd keeps the virtual keyboard's fd across restarts, so SteamVR keeps the same device.
+# Clear with: systemctl --user clean --what=fdstore frame-voice
+FileDescriptorStoreMax=1
+FileDescriptorStorePreserve=yes
+
+[Install]
+WantedBy=default.target steamvr.service

voice/tests/test_commands.py

@@ -0,0 +1,123 @@
+"""Voice command matching and the settings round trip. Run: python3 -m unittest"""
+import json
+import os
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+import voice_commands as vc  # noqa: E402
+
+FLAG = {"say": ["flag that", "flag this", "flag it", "flag"], "run": ["~/.local/bin/tool", "flag"]}
+TOOL = os.path.expanduser("~/.local/bin/tool")
+
+
+def parsed(*entries):
+    commands, warnings = vc.parse(list(entries))
+    return commands
+
+
+class Normalise(unittest.TestCase):
+    def test_case_punctuation_space(self):
+        self.assertEqual(vc.normalise("  Flag   that,  fingers-crossed! "), "flag that fingers crossed")
+        self.assertEqual(vc.normalise("..."), "")
+
+
+class Matching(unittest.TestCase):
+    def setUp(self):
+        self.commands = parsed(FLAG)
+
+    def check(self, text, phrase, rest, argv):
+        m = vc.match(text, self.commands)
+        self.assertIsNotNone(m, text)
+        self.assertEqual((m.phrase, m.rest, m.argv), (phrase, rest, argv))
+
+    def test_whisper_forms(self):
+        self.check("Flag that.", "flag that", "", [TOOL, "flag"])
+        self.check(" flag that", "flag that", "", [TOOL, "flag"])
+        self.check("FLAG THAT!", "flag that", "", [TOOL, "flag"])
+
+    def test_rest_is_one_argument_with_user_words(self):
+        self.check("Flag that, fingers crossed.", "flag that", "fingers crossed",
+                   [TOOL, "flag", "fingers crossed"])
+        self.check("Flag that: the left hand is off...", "flag that", "the left hand is off",
+                   [TOOL, "flag", "the left hand is off"])
+        self.check("Flag that, is it wrong?", "flag that", "is it wrong?", [TOOL, "flag", "is it wrong?"])
+
+    def test_longest_phrase_wins(self):
+        self.check("Flag it.", "flag it", "", [TOOL, "flag"])
+        self.check("Flag, the pinch.", "flag", "the pinch", [TOOL, "flag", "the pinch"])
+        long = parsed(FLAG, {"say": ["flag that hand"], "run": ["other"]})
+        self.assertEqual(vc.match("Flag that hand now.", long).argv, ["other", "now"])
+        self.assertEqual(vc.match("Flag that, now.", long).argv, [TOOL, "flag", "now"])
+
+    def test_ties_go_to_first_entry(self):
+        both = parsed({"say": ["go"], "run": ["a"]}, {"say": ["Go!"], "run": ["b"]})
+        self.assertEqual(vc.match("go", both).argv, ["a"])
+
+    def test_no_match(self):
+        for text in ("Flagging that.", "Please flag that.", "", "fla", "Type hello world."):
+            self.assertIsNone(vc.match(text, self.commands), text)
+        self.assertIsNone(vc.match("flag", []))
+
+    def test_rest_none(self):
+        cmds = parsed({**FLAG, "rest": "none"})
+        m = vc.match("Flag that, fingers crossed.", cmds)
+        self.assertEqual((m.rest, m.argv), ("fingers crossed", [TOOL, "flag"]))
+
+    def test_shell_text_stays_one_argument(self):
+        m = vc.match("flag that; rm -rf $HOME `x`", self.commands)
+        self.assertEqual(m.argv, [TOOL, "flag", "rm -rf $HOME `x`"])
+
+
+class Parsing(unittest.TestCase):
+    def test_malformed_entries_are_skipped_with_warnings(self):
+        entries = [
+            "flag",
+            {"run": ["x"]},
+            {"say": "flag", "run": "x --y"},
+            {"say": ["flag"], "run": []},
+            {"say": ["flag"], "run": ["", "x"]},
+            {"say": ["flag"], "run": ["x", 3]},
+            {"say": [1, None], "run": ["x"]},
+            {"say": ["...", " "], "run": ["x"]},
+            {"say": ["flag"], "run": ["x"], "rest": "all"},
+            {"say": "single", "run": ["ok"]},
+        ]
+        commands, warnings = vc.parse(entries)
+        self.assertEqual(len(commands), 1)
+        self.assertEqual(len(warnings), len(entries) - 1)
+        self.assertEqual(vc.match("Single.", commands).argv, ["ok"])
+
+    def test_not_a_list(self):
+        self.assertEqual(vc.parse(None), ([], []))
+        commands, warnings = vc.parse({"say": ["x"]})
+        self.assertEqual((commands, len(warnings)), ([], 1))
+
+
+class SettingsRoundTrip(unittest.TestCase):
+    def test_app_save_keeps_commands_and_unknown_keys(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            path = Path(tmp) / "settings.json"
+            path.write_text(json.dumps({"wake_phrases": ["HEY COMPUTER"], "commands": [FLAG], "extra": 1}))
+            os.environ["FRAME_VOICE_SETTINGS"] = str(path)
+            sys.modules.pop("settings", None)
+            import settings
+            try:
+                data = settings.load()
+                self.assertEqual(data["commands"], [FLAG])
+                data["sounds"] = True
+                data["commands"] = []  # a stale copy must not overwrite the file's list
+                settings.save(data)
+                saved = json.loads(path.read_text())
+                self.assertEqual((saved["commands"], saved["extra"], saved["sounds"]), ([FLAG], 1, True))
+                path.write_text("{}")
+                self.assertEqual(settings.load()["commands"], [])
+            finally:
+                del os.environ["FRAME_VOICE_SETTINGS"]
+                sys.modules.pop("settings", None)
+
+
+if __name__ == "__main__":
+    unittest.main()

voice/tests/test_hotkey.py

@@ -0,0 +1,121 @@
+"""The push-to-talk key starts or stops dictation only when pressed on its own.
+Needs the venv's Python (evdev): .venv/bin/python -m unittest"""
+import asyncio
+import sys
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+try:
+    import frame_voice as fv  # noqa: E402
+except ImportError as err:
+    raise unittest.SkipTest(f"needs the venv's Python: {err}")
+
+e = fv.e
+
+
+class Dev:
+    def __init__(self, path):
+        self.path = path
+
+
+KEYBOARD, OTHER, MOUSE = Dev("/dev/input/event3"), Dev("/dev/input/event7"), Dev("/dev/input/event9")
+
+
+class Daemon(fv.FrameVoice):
+    """on_key() and the state it reads, without models, mic or typing."""
+
+    def __init__(self, hotkeys, mode):
+        self.hotkeys = {e.ecodes[name]: name for name in hotkeys}
+        self.kb_mode = mode
+        self.dcfg = {"hold_delay": 0.0}
+        self.hk_down = self.hk_chord = False
+        self.hk_timer = None
+        self.mods_down = set()
+        self.mods_released = asyncio.Event()
+        self._held_logged = None
+        self.stream = None
+        self.recording_source = None
+        self.calls = []
+
+    async def toggle_dictation(self, source):
+        self.calls.append("toggle")
+
+    def start_dictation(self, source):
+        self.calls.append("start")
+        self.recording_source = source
+
+    async def stop_dictation(self):
+        self.calls.append("stop")
+        self.recording_source = None
+
+    def cancel_dictation(self, chime=True):
+        self.calls.append("cancel")
+        self.recording_source = None
+
+
+class Hotkey(unittest.IsolatedAsyncioTestCase):
+    async def press(self, daemon, *events):
+        """events: (device, key name, 1 for down or 0 for up)."""
+        for dev, name, value in events:
+            daemon.on_key(dev, e.ecodes[name], value)
+            await asyncio.sleep(0)
+        await asyncio.sleep(0.01)  # the tasks and the hold timer it started
+        return daemon.calls
+
+    async def test_tap_alone_toggles(self):
+        d = Daemon(["KEY_RIGHTCTRL"], "toggle")
+        self.assertEqual(await self.press(d, (KEYBOARD, "KEY_RIGHTCTRL", 1), (KEYBOARD, "KEY_RIGHTCTRL", 0)),
+                         ["toggle"])
+
+    async def test_modifier_held_before(self):
+        for other in ("KEY_LEFTSHIFT", "KEY_RIGHTSHIFT", "KEY_LEFTALT", "KEY_LEFTMETA", "KEY_LEFTCTRL"):
+            d = Daemon(["KEY_RIGHTCTRL"], "toggle")
+            calls = await self.press(d, (KEYBOARD, other, 1), (KEYBOARD, "KEY_RIGHTCTRL", 1),
+                                     (KEYBOARD, "KEY_RIGHTCTRL", 0), (KEYBOARD, other, 0))
+            self.assertEqual(calls, [], other)
+
+    async def test_modifier_released_first(self):
+        d = Daemon(["KEY_RIGHTCTRL"], "toggle")
+        self.assertEqual(await self.press(d, (KEYBOARD, "KEY_LEFTSHIFT", 1), (KEYBOARD, "KEY_RIGHTCTRL", 1),
+                                          (KEYBOARD, "KEY_LEFTSHIFT", 0), (KEYBOARD, "KEY_RIGHTCTRL", 0)), [])
+
+    async def test_modifier_held_on_another_keyboard(self):
+        d = Daemon(["KEY_RIGHTCTRL"], "toggle")
+        self.assertEqual(await self.press(d, (OTHER, "KEY_LEFTALT", 1), (KEYBOARD, "KEY_RIGHTCTRL", 1),
+                                          (KEYBOARD, "KEY_RIGHTCTRL", 0), (OTHER, "KEY_LEFTALT", 0)), [])
+
+    async def test_modifier_pressed_during(self):
+        d = Daemon(["KEY_RIGHTCTRL"], "toggle")
+        self.assertEqual(await self.press(d, (KEYBOARD, "KEY_RIGHTCTRL", 1), (KEYBOARD, "KEY_LEFTSHIFT", 1),
+                                          (KEYBOARD, "KEY_LEFTSHIFT", 0), (KEYBOARD, "KEY_RIGHTCTRL", 0)), [])
+
+    async def test_shortcut_with_the_key(self):
+        d = Daemon(["KEY_RIGHTCTRL"], "toggle")
+        self.assertEqual(await self.press(d, (KEYBOARD, "KEY_RIGHTCTRL", 1), (KEYBOARD, "KEY_C", 1),
+                                          (KEYBOARD, "KEY_C", 0), (KEYBOARD, "KEY_RIGHTCTRL", 0)), [])
+
+    async def test_tap_alone_after_a_shortcut(self):
+        d = Daemon(["KEY_RIGHTCTRL"], "toggle")
+        await self.press(d, (KEYBOARD, "KEY_LEFTSHIFT", 1), (KEYBOARD, "KEY_RIGHTCTRL", 1),
+                         (KEYBOARD, "KEY_RIGHTCTRL", 0), (KEYBOARD, "KEY_LEFTSHIFT", 0))
+        self.assertEqual(await self.press(d, (KEYBOARD, "KEY_RIGHTCTRL", 1), (KEYBOARD, "KEY_RIGHTCTRL", 0)),
+                         ["toggle"])
+
+    async def test_mouse_button_with_ctrl(self):
+        d = Daemon(["BTN_SIDE"], "toggle")
+        self.assertEqual(await self.press(d, (KEYBOARD, "KEY_LEFTCTRL", 1), (MOUSE, "BTN_SIDE", 1),
+                                          (MOUSE, "BTN_SIDE", 0), (KEYBOARD, "KEY_LEFTCTRL", 0)), [])
+        self.assertEqual(await self.press(d, (MOUSE, "BTN_SIDE", 1), (MOUSE, "BTN_SIDE", 0)), ["toggle"])
+
+    async def test_hold_mode(self):
+        d = Daemon(["KEY_RIGHTCTRL"], "hold")
+        self.assertEqual(await self.press(d, (KEYBOARD, "KEY_LEFTSHIFT", 1), (KEYBOARD, "KEY_RIGHTCTRL", 1)), [])
+        await self.press(d, (KEYBOARD, "KEY_RIGHTCTRL", 0), (KEYBOARD, "KEY_LEFTSHIFT", 0))
+        self.assertEqual(d.calls, [])
+        self.assertEqual(await self.press(d, (KEYBOARD, "KEY_RIGHTCTRL", 1)), ["start"])
+        self.assertEqual(await self.press(d, (KEYBOARD, "KEY_RIGHTCTRL", 0)), ["start", "stop"])
+
+
+if __name__ == "__main__":
+    unittest.main()

voice/voice_commands.py

@@ -0,0 +1,92 @@
+"""Voice commands: run a program when a wake word command starts with a set phrase.
+
+From "commands" in ~/.config/frame-voice/settings.json:
+  [{"say": ["flag that", "flag"], "run": ["~/.local/bin/my-tool", "flag"], "rest": "arg"}]
+
+say:  phrases. A command matches when its words equal a phrase or start with it.
+      Case and punctuation don't matter. The longest matching phrase of all entries wins.
+run:  the program and its arguments, run directly (never through a shell). "~" is expanded.
+rest: "arg" (default) appends the words after the phrase as one extra argument when there
+      are any. "none" never appends them.
+
+Pure standard library, so the tests can import it without the audio stack.
+"""
+import os
+import re
+from typing import NamedTuple
+
+REST_MODES = ("arg", "none")
+WORD = re.compile(r"\w+")
+
+
+class Command(NamedTuple):
+    phrases: tuple  # (as written, normalised word tuple)
+    run: tuple
+    rest: str
+
+
+class Match(NamedTuple):
+    phrase: str  # as written in settings.json
+    rest: str  # the user's words after the phrase, "" when none
+    argv: list
+
+
+def words(text):
+    """Lowercase words with punctuation as separators: "Flag that, OK." -> ("flag", "that", "ok")."""
+    return tuple(w.lower() for w in WORD.findall(text))
+
+
+def normalise(text):
+    return " ".join(words(text))
+
+
+def parse(entries):
+    """Valid commands from the settings list, and a warning for each entry skipped."""
+    commands, warnings = [], []
+    if not isinstance(entries, list):
+        if entries is not None:
+            warnings.append(f"commands: expected a list, got {type(entries).__name__}")
+        return commands, warnings
+    for i, entry in enumerate(entries):
+        problem = None
+        if not isinstance(entry, dict):
+            problem = "not an object"
+        else:
+            say = entry.get("say")
+            if isinstance(say, str):
+                say = [say]
+            run = entry.get("run")
+            rest = entry.get("rest", "arg")
+            if not isinstance(say, list) or not all(isinstance(p, str) for p in say):
+                problem = '"say" must be a list of phrases'
+            elif not isinstance(run, list) or not run or not all(isinstance(a, str) for a in run) or not run[0]:
+                problem = '"run" must be a list: program, then arguments'
+            elif rest not in REST_MODES:
+                problem = f'"rest" must be one of {", ".join(REST_MODES)}'
+            else:
+                phrases = tuple((p.strip(), words(p)) for p in say if words(p))
+                if not phrases:
+                    problem = '"say" has no words'
+        if problem:
+            warnings.append(f"commands[{i}] skipped: {problem}")
+            continue
+        commands.append(Command(phrases, tuple(os.path.expanduser(a) for a in run), rest))
+    return commands, warnings
+
+
+def match(text, commands):
+    """The command `text` starts with, or None. Ties go to the earlier entry."""
+    found = list(WORD.finditer(text or ""))
+    heard = tuple(m.group().lower() for m in found)
+    best = None
+    for cmd in commands:
+        for phrase, pw in cmd.phrases:
+            if heard[:len(pw)] == pw and (best is None or len(pw) > len(best[2])):
+                best = (cmd, phrase, pw)
+    if best is None:
+        return None
+    cmd, phrase, pw = best
+    rest = text[found[len(pw) - 1].end():]
+    rest = re.sub(r"^\W+", "", rest).rstrip(" \t\n.")
+    argv = list(cmd.run) + ([rest] if rest and cmd.rest == "arg" else [])
+    return Match(phrase, rest, argv)