#!/usr/bin/env python3
"""ft-voiced: Frametop Voice, local dictation and wake words (frametop-voice.service).

One PipeWire capture stream (pw-record) feeds two consumers:
  * dictation: push-to-talk -> record -> Moonshine (live, on the CPU) or whisper-server -> the
    input relay types the text where typing goes
  * wake word: sherpa-onnx keyword spotter -> capture the spoken command -> whisper-server ->
    a voice command (voice_commands.py), or the command hook and event subscribers

Push-to-talk comes from the input relay (a key combination or mouse button with the dictate or
dictate_hold action: "toggle|start|stop|cancel <source> [quiet]" on @frametop_voice, taken only
from this user; quiet: no cancel tone) and from Frame controller buttons, read from SteamVR's web
socket (input/vrws.py), so they still do what they do in games. Text goes to the relay as "type
TEXT" on @frametop_relay, any language's characters: it types them with the keys of the desktop's
keyboard layout, and only while no modifier is held, so letters never turn into shortcuts.

The language ("language" in the settings) picks the live engine: English is Moonshine on the
CPU, any other language whisper-server on the GPU, with the model the settings choose
(voice_settings.model_for, loaded by voice/whisper-run).

Runs in Frametop's container (scripts/in-box: numpy, sherpa-onnx and pw-record are there),
named ft-voiced for pkill. Programs it starts for voice commands and hooks run on the host
(distrobox-host-exec), as the relay's commands do. Settings: voice_settings.py. Models:
~/.local/share/frametop/voice/models (voice/models.sh). Control socket for ft-voice:
/run/user/UID/frametop-voice/control.sock.
"""
import asyncio
import ctypes
import io
import json
import logging
import os
import re
import shutil
import signal
import socket
import struct
import sys
import threading
import time
import unicodedata
import urllib.request
import uuid
import wave
from pathlib import Path

import numpy as np
import sentencepiece
import sherpa_onnx

ROOT = Path(__file__).resolve().parent
sys.path[:0] = [str(ROOT), str(ROOT.parent / "input"),  # its modules, vrws (SteamVR's web socket),
                str(ROOT.parent / "commands")]  # and your Frametop commands ("run NAME")
import ft_commands
import voice_commands
import voice_settings
import vrws

RATE = 16000
CHUNK = 1536  # 96 ms of 16 kHz mono: a whole number of 512-sample VAD windows
MODELS = Path(os.environ.get("FRAMETOP_VOICE_MODELS", "~/.local/share/frametop/voice/models")).expanduser()
KWS_DIR = MODELS / "sherpa-onnx-kws-zipformer-gigaspeech-3.3M-2024-01-01"
VAD_MODEL = MODELS / "silero_vad.onnx"
# Not $XDG_RUNTIME_DIR: Frametop's desktop points it at its own directory.
RUNTIME_DIR = Path(f"/run/user/{os.getuid()}/frametop-voice")
STATE_DIR = Path(os.environ.get("XDG_STATE_HOME", Path.home() / ".local/state")) / "frametop/voice"
CLIP_DIR = STATE_DIR / "clips"
SOCKET_PATH = RUNTIME_DIR / "control.sock"
RELAY = "\0frametop_relay"  # the input relay: it types ("type TEXT")
VOICE = "\0frametop_voice"  # push-to-talk from the relay
TYPE_CHUNK = 2000  # bytes of text per "type" datagram, at most
# Programs started for voice commands and hooks run on the host: in the container they'd miss
# the host's tools. distrobox-host-exec passes no environment, so it goes through env there.
IN_BOX = Path("/run/.containerenv").exists() and shutil.which("distrobox-host-exec") is not None
HOST = ["distrobox-host-exec"] if IN_BOX else []
# Audio kept before speech that starts after a silence: the VAD flags speech ~0.27 s late.
LEAD_IN_SECONDS = 0.5
# Committed words a window's decode is lined up against: more than its context can hold.
ALIGN_WORDS = 12
LOOKUP_EVERY = 5.0  # seconds between looking up the controllers' root paths (they change)

log = logging.getLogger("ft-voiced")

DEFAULTS = {
    "audio": {"source": ""},
    "whisper": {
        "url": "http://127.0.0.1:8178/inference",
        "prompt": "SteamOS, Steam Frame, Godot.",  # for English only: it would pull others to English
        # Smallest encoder window (50 per second of audio). medium and large-v3-turbo loop
        # and take 10-20 s on short clips below 512; 512 covers pieces up to ~10 s.
        "min_audio_ctx": 512,
    },
    "dictation": {
        # Shorter recordings are dropped, not transcribed: a press that was a slip. A single word
        # takes ~0.3-0.5 s; a recording without speech in it types nothing anyway (the VAD).
        "min_seconds": 0.3,
        "live_after_seconds": 2.0,  # no live typing before this far in (as when it was min_seconds)
        "max_seconds": 120,
        "trailing_space": True,
        # Transcribe while recording: pieces split at pauses go to whisper as soon as they end,
        # so only the last piece is left after release.
        "streaming": True,
        # Live text by agreement: the current window of speech is re-decoded every
        # step_seconds and words two decodes in a row agree on are typed. A window closes at a
        # pause (its last words are typed then) or at its quietest point by max_window_seconds.
        # moonshine runs on the CPU (~0.1-0.3 s per decode, returns nothing past ~9 s of audio);
        # "whisper" uses the whisper-server on the GPU (better punctuation, ~0.5-0.8 s per decode).
        "engine": "moonshine",
        "moonshine_model": "sherpa-onnx-moonshine-base-en-quantized-2026-02-27",  # in MODELS
        "moonshine_threads": 2,
        "step_seconds": 0.8,
        "pause_seconds": 0.8,  # shorter pauses inside a sentence ended it with "?"/"." in testing
        "max_window_seconds": 8.0,  # audio per decode, context included
        # Each window is decoded after this much of the previous one's end, so a sentence that
        # goes on after a pause isn't heard as a new one (no stray "." and capital at the seam).
        "context_seconds": 1.5,
        # Keep the audio of the last N dictations in ~/.local/state/frametop/voice/clips for
        # comparing models and settings. 0 = never write audio to disk.
        "save_clips": 0,
    },
    "wake": {
        "threshold": 0.25,
        "boost": 1.5,
        "command_timeout": 5.0,
        "end_silence": 0.8,
        "max_command_seconds": 20,
        "wake_hook": "",  # run with the keyword as $1 as soon as the wake word fires (relative: to voice/)
        "command_hook": "hooks/on-command",  # run with the transcribed command as $1
    },
    # On or off: "sounds" in the settings. Each tone has silence around it (lead, tail), so an
    # output that wakes up late or a stream that ends early cuts only silence.
    "sounds": {"volume": 0.5, "lead_seconds": 0.12, "tail_seconds": 0.2},
}


def load_config(advanced):
    """DEFAULTS, with the settings' "advanced" overrides."""
    cfg = {k: dict(v) for k, v in DEFAULTS.items()}
    for section, values in advanced.items():
        if section in cfg:
            cfg[section].update(values)
    if advanced:
        log.info("advanced settings: %s", json.dumps(advanced))
    return cfg


def resolve(p, base=ROOT):
    p = Path(p).expanduser()
    return p if p.is_absolute() else base / p


def set_process_name(name):
    """The name pkill -x and pgrep -x see (frametop-voice.service stops it by name: in the
    container, it isn't in the unit's cgroup)."""
    try:
        ctypes.CDLL(None).prctl(15, name.encode(), 0, 0, 0)  # PR_SET_NAME
    except (OSError, AttributeError):
        pass


# ---------------------------------------------------------------- typing

ASCII_FIXUPS = str.maketrans({"\u2018": "'", "\u2019": "'", "\u201c": '"', "\u201d": '"',
                              "\u2013": "-", "\u2014": "-", "\u2026": "...", "\n": " ", "\t": " "})


def to_text(text):
    """Text for the relay to type: curly quotes and dashes made plain (most layouts have no key
    for them), letters composed (NFC: "ü" as one character, the way a layout's key types it),
    control characters dropped. Accents and other scripts stay: the relay types what the
    desktop's layout has keys for, and logs the rest."""
    text = unicodedata.normalize("NFC", text.translate(ASCII_FIXUPS))
    return "".join(c for c in text if c.isprintable())


def type_chunks(text, size=TYPE_CHUNK):
    """text in pieces of at most size bytes (UTF-8), split after a space where one is near."""
    while text:
        if len(text.encode()) <= size:
            yield text
            return
        end = size
        while len(text[:end].encode()) > size:
            end -= 1
        cut = text.rfind(" ", end // 2, end) + 1 or end
        yield text[:cut]
        text = text[cut:]


def drop_repeats(text):
    """Collapse sentences whisper repeated back to back ("Ask not. Ask not. Ask not.")."""
    out = []
    for sentence in re.split(r"(?<=[.!?])\s+", text):
        if not out or sentence.strip(" .!?,").lower() != out[-1].strip(" .!?,").lower():
            out.append(sentence)
    return " ".join(out)


SAME_WORD = {"okay": "ok"}  # spellings Moonshine switches between from one decode to the next


def norm(word):
    word = re.sub(r"[^\w']", "", word.lower())
    return SAME_WORD.get(word, word)


def split_marks(word):
    """("the.", ...) -> ("the", "."): a word and the punctuation after it."""
    m = re.fullmatch(r"(.*?)([.,!?;:]*)", word)
    return m[1], m[2]


SENTENCE_END = (".", "?", "!")
# Words a sentence rarely ends on, so a pause after one is a pause inside a sentence.
DANGLING = set("""a an the to of for in on at by with from into onto about as than and or but nor
    because if when while where whether that which whose my our your their its i we they he she
    i'm we're you're they're it's are were be been being am have has had will would can
    could should shall may might must very""".split())


def settle_mark(word, held, heard):
    """Punctuation after `word`, the last one before a pause. held: what the window decoded
    alone put there (every window it ends reads as a finished sentence). heard: what the next
    decode put there after hearing what followed (None if it didn't line up), which in turn
    tends to run finished sentences together."""
    if heard is None:
        return held
    if heard.endswith(SENTENCE_END) or norm(word) in DANGLING:
        return heard
    return held or heard


def common_prefix(a, b):
    """How many leading words two hypotheses share, ignoring case and punctuation."""
    n = 0
    for x, y in zip(a, b):
        if norm(x) != norm(y):
            break
        n += 1
    return n


def agreement(prev, cur):
    """How many leading words of `cur` the previous hypothesis backs up. One word that
    differs (a filler heard as "A" one time and "Uh," the next) doesn't hold back the rest
    when the three words after it agree again; cur's version of it wins."""
    i = j = 0
    while True:
        k = common_prefix(prev[i:], cur[j:])
        i, j = i + k, j + k
        if i >= len(prev) or j >= len(cur):
            return j
        for di, dj in ((1, 1), (0, 1), (1, 0)):
            if common_prefix(prev[i + di:], cur[j + dj:]) >= 3:
                i, j = i + di, j + dj
                break
        else:
            return j


def align(ref, hyp):
    """Where the words after `ref` start in `hyp`, a decode whose audio starts inside the
    audio of ref's last words. Lines ref's end up with hyp's start (same word +2, anything
    else -1, ref may start anywhere); on a tie the earlier start wins, so a phrase said twice
    in a row isn't taken for the copy already typed. Returns (start, hyp's form of ref's last
    word or None when that word didn't line up), or None when nothing lines up."""
    a, b = [norm(w) for w in ref], [norm(w) for w in hyp]
    row = [-j for j in range(len(b) + 1)]
    for i in range(1, len(a) + 1):
        nxt = [0]
        for j in range(1, len(b) + 1):
            nxt.append(max(row[j - 1] + (2 if a[i - 1] == b[j - 1] else -1), row[j] - 1, nxt[j - 1] - 1))
        row = nxt
    start = max(range(len(b) + 1), key=lambda j: (row[j], -j))
    if not a or row[start] < 2:
        return None
    return start, hyp[start - 1] if start and b[start - 1] == a[-1] else None


def feed_vad(vad, samples):
    """Silero VAD only handles input correctly in whole 512-sample windows."""
    size = vad.config.silero_vad.window_size
    for i in range(0, len(samples) - size + 1, size):
        vad.accept_waveform(samples[i:i + size])


# ---------------------------------------------------------------- sounds

def make_sounds(volume, lead=0.12, tail=0.2):
    """Short tones so state changes are audible inside VR, with lead and tail seconds of silence
    around them. Returns name -> wav path."""
    specs = {"start": [660, 880], "stop": [880, 660], "wake": [988], "cancel": [330, 262]}
    out = {}
    RUNTIME_DIR.mkdir(parents=True, exist_ok=True)
    for name, freqs in specs.items():
        parts = [np.zeros(int(RATE * lead))]
        for f in freqs:
            t = np.arange(int(RATE * 0.11)) / RATE
            env = np.minimum(1, np.minimum(t, t[::-1]) * 200)
            parts.append(np.sin(2 * np.pi * f * t) * env * volume)
        parts.append(np.zeros(int(RATE * tail)))
        path = RUNTIME_DIR / f"{name}.wav"
        path.write_bytes(to_wav(np.concatenate(parts).astype(np.float32)))
        out[name] = path
    return out


def to_wav(samples):
    buf = io.BytesIO()
    with wave.open(buf, "wb") as w:
        w.setnchannels(1)
        w.setsampwidth(2)
        w.setframerate(RATE)
        w.writeframes((np.clip(samples, -1, 1) * 32767).astype(np.int16).tobytes())
    return buf.getvalue()


# ---------------------------------------------------------------- daemon

class Voice:
    def __init__(self):
        self.settings_mtime = voice_settings.mtime()
        self.named_commands = []  # your Frametop commands, "run NAME" (voice_commands.named)
        self.named_mtime = None
        b = voice_settings.load()  # defaults when the file doesn't exist yet
        self.cfg = cfg = load_config(b["advanced"])
        self.dcfg, self.wcfg = cfg["dictation"], cfg["wake"]
        self.relay = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)  # to type ("type TEXT")
        self.relay.settimeout(2)  # the relay's queue full for that long: it's stuck
        self.relay_said = False  # logged that the relay isn't there (once until it is)
        self.type_lock = asyncio.Lock()
        self.sounds = {}  # apply_options() makes them when enabled
        self.chime_task = None  # the tone playing or waiting; tones play one after another
        self.start_chime = None  # dictation's start tone waits for the mic's first audio (a timer)
        self.language = voice_settings.DEFAULT["language"]  # apply_options() sets the saved ones
        self.model = voice_settings.DEFAULT["model"]
        self.subscribers = set()

        self.recording = None  # list of chunks while dictating
        self.recording_source = None
        self.stream = None  # streaming state of the current dictation, see start_dictation
        self.whisper_lock = asyncio.Lock()  # one request at a time, in order
        self.command = None  # dict while capturing a post-wake command
        self.voice_commands = []  # from "commands" in the settings, see voice_commands.py
        self.wake_enabled = voice_settings.DEFAULT["wake_word"]  # apply_options() sets the saved value
        self.whisper_wanted = None  # whisper-server asked to run (True) or stop (False)
        self.mic_task = None
        self.mic_proc = None
        self.audio_session_ready = False
        self.tasks = set()  # keep references so fire-and-forget tasks are not collected early

        self.applied = {}  # settings options last applied, so a save only changes what changed
        self.controller = (tuple(b["controller"]), b["controller_mode"])  # push-to-talk buttons, mode
        self.controller_changed = threading.Event()  # the reader (controller_loop) looks again
        self.loop = None  # the event loop, for the controller reader's thread

        self.kws = self.kws_stream = None  # built from the wake phrases in the settings
        self.wake_phrases = []
        self.cmd_vad = self._make_vad(self.wcfg["end_silence"])
        self.clip_vad = self._make_vad(0.5)
        self.piece_vad = self._make_vad(0.1)  # per-chunk speech/silence labels for live windows
        self.moonshine = None
        if self.dcfg["engine"] == "moonshine":
            m = resolve(self.dcfg["moonshine_model"], MODELS)
            self.moonshine = sherpa_onnx.OfflineRecognizer.from_moonshine_v2(
                encoder=str(m / "encoder_model.ort"), decoder=str(m / "decoder_model_merged.ort"),
                tokens=str(m / "tokens.txt"), num_threads=self.dcfg["moonshine_threads"])
            log.info("live dictation engine: moonshine (%s)", m.name)
        self.apply_options(b)

    # --- models

    def _make_kws(self, phrases):
        """Open-vocabulary spotter for the phrases: they are spelled with the model's BPE
        tokens at load time, so any English words work without training."""
        self.wake_phrases = list(phrases or ())
        if not phrases:
            log.info("wake phrases: none")
            return None
        sp = sentencepiece.SentencePieceProcessor(model_file=str(KWS_DIR / "bpe.model"))
        lines = []
        for phrase in phrases:
            tokens = " ".join(sp.encode(phrase, out_type=str))
            lines.append(f"{tokens} :{self.wcfg['boost']} #{self.wcfg['threshold']} @{phrase.replace(' ', '_')}")
        RUNTIME_DIR.mkdir(parents=True, exist_ok=True)
        kw_file = RUNTIME_DIR / "keywords.txt"
        kw_file.write_text("\n".join(lines) + "\n")
        log.info("wake phrases: %s", ", ".join(phrases))
        return sherpa_onnx.KeywordSpotter(
            tokens=str(KWS_DIR / "tokens.txt"),
            encoder=str(KWS_DIR / "encoder-epoch-12-avg-2-chunk-16-left-64.int8.onnx"),
            decoder=str(KWS_DIR / "decoder-epoch-12-avg-2-chunk-16-left-64.int8.onnx"),
            joiner=str(KWS_DIR / "joiner-epoch-12-avg-2-chunk-16-left-64.int8.onnx"),
            keywords_file=str(kw_file),
            num_threads=1,
        )

    def _make_vad(self, min_silence):
        config = sherpa_onnx.VadModelConfig()
        config.silero_vad.model = str(VAD_MODEL)
        config.silero_vad.min_silence_duration = min_silence
        config.silero_vad.min_speech_duration = 0.2
        config.sample_rate = RATE
        return sherpa_onnx.VoiceActivityDetector(config, buffer_size_in_seconds=180)

    # --- state

    def state(self):
        return {
            "dictating": self.recording is not None,
            "awaiting_command": self.command is not None,
            "wake_enabled": self.wake_enabled,
            "mic_open": self.mic_proc is not None and self.mic_proc.returncode is None,
        }

    def broadcast(self, event, **fields):
        msg = (json.dumps({"event": event, "time": time.time(), **fields}) + "\n").encode()
        for w in list(self.subscribers):
            try:
                w.write(msg)
            except Exception:
                self.subscribers.discard(w)

    def chime(self, name):
        """Play a tone, after any tone still playing (two at once garble each other). The mic
        stays open until it's done (update_mic): on the Frame, the speaker and the mic go
        through one echo-cancelling filter, and closing the mic while a tone plays can cut it."""
        if name in self.sounds:
            self.chime_task = asyncio.create_task(self._play(name, self.chime_task))
            self.tasks.add(self.chime_task)
            self.chime_task.add_done_callback(self.tasks.discard)

    async def _play(self, name, prev):
        if prev and not prev.done():
            await asyncio.wait([prev])
        # SteamOS runs every playback stream through a per-stream spatializer that takes
        # ~0.35 s to set up, late enough to bury a start tone under the first word. Its
        # WirePlumber script skips streams named like filter chains, so use such a name.
        props = '{ node.name = "frametop-voice-chime-filter-chain" media.role = "Notification" }'
        try:
            await self._run(["pw-play", "-P", props, str(self.sounds[name])], quiet=True)
        finally:
            if self.chime_task is asyncio.current_task():
                self.chime_task = None
                self.update_mic()  # a mic left open for the tone closes now

    def chime_playing(self):
        return self.chime_task is not None and not self.chime_task.done()

    def chime_start(self):
        """Dictation's start tone, once the mic sends audio (on_audio), or after 0.6 s without:
        a tone while the mic is still opening can be cut, and says "talk" too early."""
        if self.start_chime:
            self.start_chime.cancel()
            self.start_chime = None
            self.chime("start")

    async def _run(self, argv, env=None, quiet=False, host=False):
        """Run a program and wait for it. host: on the host (voice commands, hooks), with env
        passed through env(1) there."""
        if host and HOST:
            argv = [*HOST, "env", *(f"{k}={v}" for k, v in (env or {}).items()), *argv]
        try:
            proc = await asyncio.create_subprocess_exec(
                *argv, env={**os.environ, **(env or {})},
                stdout=asyncio.subprocess.DEVNULL if quiet else None,
                stderr=asyncio.subprocess.DEVNULL if quiet else None)
            rc = await proc.wait()
            if rc and not quiet:
                log.warning("%s exited %s", argv[0], rc)
        except OSError as err:
            log.warning("could not run %s: %s", argv[0], err)

    def spawn(self, coro):
        task = asyncio.create_task(coro)
        self.tasks.add(task)
        task.add_done_callback(self.tasks.discard)

    def run_hook(self, key, *args, env=None):
        hook = self.wcfg.get(key)
        if not hook:
            return
        path = resolve(hook)
        if not os.access(path, os.X_OK):
            log.warning("hook %s not executable: %s", key, path)
            return
        self.spawn(self._run([str(path), *args], env=env, host=True))

    # --- microphone

    def mic_needed(self):
        return (self.wake_enabled and self.kws is not None) or self.recording is not None or self.command is not None

    def reset_kws(self):
        if self.kws:
            self.kws.reset_stream(self.kws_stream)

    async def _rebuild_kws(self, phrases):
        kws = await asyncio.to_thread(self._make_kws, phrases)
        self.kws, self.kws_stream = kws, kws.create_stream() if kws else None
        self.update_mic()

    def update_mic(self):
        if self.mic_needed() and (self.mic_task is None or self.mic_task.done()):
            self.mic_task = asyncio.create_task(self._capture())
        elif not self.mic_needed() and self.mic_proc and self.mic_proc.returncode is None:
            if self.chime_playing():
                return  # closes when the tone is done (_play), see chime
            self.mic_proc.terminate()  # fully close the mic when nothing is listening

    async def _wait_for_audio_session(self):
        """Don't open the mic while SteamOS is still building its mic filter chain.

        At boot this service can start before WirePlumber has built the chain. A capture
        stream that connects mid-setup can leave the echo-cancel link stuck, and the mic then
        reads silence until PipeWire restarts. The unit is ordered after WirePlumber, but that
        only means it started, so wait here until it's been up for 10 s.
        """
        while True:
            proc = await asyncio.create_subprocess_exec(
                *HOST, "systemctl", "--user", "show", "wireplumber", "-p", "ActiveState",
                "-p", "ActiveEnterTimestampMonotonic", stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.DEVNULL)
            props = dict(line.split("=", 1) for line in (await proc.communicate())[0].decode().split())
            if props.get("ActiveState") == "active":
                started = int(props.get("ActiveEnterTimestampMonotonic", "0")) / 1e6
                settle = started + 10 - time.clock_gettime(time.CLOCK_MONOTONIC)
                if settle > 0:
                    await asyncio.sleep(settle)
                return
            await asyncio.sleep(1)

    async def _capture(self):
        if not self.audio_session_ready:
            log.info("waiting for WirePlumber before opening the mic")
            await self._wait_for_audio_session()
            self.audio_session_ready = True
        while self.mic_needed():
            argv = ["pw-record", "--raw", "--rate", str(RATE), "--channels", "1",
                    "--format", "s16", "--latency", "50ms", "-"]
            if self.cfg["audio"]["source"]:
                argv[1:1] = ["--target", self.cfg["audio"]["source"]]
            self.mic_proc = await asyncio.create_subprocess_exec(
                *argv, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.DEVNULL)
            log.info("mic open")
            try:
                while True:
                    data = await self.mic_proc.stdout.readexactly(CHUNK * 2)
                    self.on_audio(np.frombuffer(data, np.int16).astype(np.float32) / 32768)
            except asyncio.IncompleteReadError:
                pass
            await self.mic_proc.wait()
            log.info("mic closed")
            if self.mic_needed():
                await asyncio.sleep(1)

    def on_audio(self, chunk):
        if self.recording is not None:
            self.recording.append(chunk)
            if self.start_chime:
                self.chime_start()
            if self.stream:
                feed_vad(self.piece_vad, chunk)
                self.stream["speech"].append(self.piece_vad.is_speech_detected())
                self.stream["level"].append(float(np.sqrt(np.mean(chunk ** 2))))
                while not self.piece_vad.empty():
                    self.piece_vad.pop()
                self.stream["wake"].set()
            if len(self.recording) * CHUNK / RATE >= self.dcfg["max_seconds"]:
                asyncio.create_task(self.stop_dictation())
        elif self.command is not None:
            self._command_audio(chunk)
        elif self.wake_enabled and self.kws:
            self.kws_stream.accept_waveform(RATE, chunk)
            while self.kws.is_ready(self.kws_stream):
                self.kws.decode_stream(self.kws_stream)
                keyword = self.kws.get_result(self.kws_stream)
                if keyword:
                    self.reset_kws()
                    self.on_wake(keyword.replace("_", " "))
                    break

    # --- transcription

    def trim_to_speech(self, samples):
        """Cut leading/trailing silence. Returns None when there is no speech at all."""
        vad = self.clip_vad
        vad.reset()
        feed_vad(vad, samples)
        vad.flush()
        first = last = None
        while not vad.empty():
            seg = vad.front
            first = seg.start if first is None else first
            last = seg.start + len(seg.samples)
            vad.pop()
        if first is None:
            return None
        pad = int(0.25 * RATE)
        return samples[max(0, first - pad):last + pad]

    async def transcribe(self, samples):
        speech = self.trim_to_speech(samples)
        if speech is None:
            log.info("no speech detected; skipping transcription")
            return ""
        return await self.transcribe_speech(speech)

    async def transcribe_speech(self, speech, context="", quiet=False):
        """Transcribe audio that is already speech. `context` is the text said just before."""
        dur = len(speech) / RATE
        # Whisper's encoder always runs a 30 s window (1500 frames, 50/s). Shrinking it to the
        # clip length is the main latency win on the Adreno GPU. It must still cover the whole
        # clip, and very small contexts make the decoder loop, hence the floor.
        audio_ctx = 0 if dur > 28 else min(1500, max(self.cfg["whisper"]["min_audio_ctx"], int(dur * 50 * 1.1) + 64))
        async with self.whisper_lock:
            t0 = time.monotonic()
            text = await asyncio.to_thread(self._post, to_wav(speech), audio_ctx, context)
        text = re.sub(r"\s+", " ", re.sub(r"\[[^\]]*\]|\([^)]*\)", "", text)).strip()
        text = drop_repeats(text)
        (log.debug if quiet else log.info)("transcribed %.1fs audio in %.2fs: %r", dur, time.monotonic() - t0, text)
        return text

    def _post(self, wav_bytes, audio_ctx, context=""):
        w = self.cfg["whisper"]
        # temperature_inc 0: no fallback re-decodes, which cost 10+ s on short pieces with turbo.
        fields = {"response_format": "json", "temperature": "0", "temperature_inc": "0",
                  "no_timestamps": "true", "language": self.language, "audio_ctx": str(audio_ctx)}
        prompt = " ".join(filter(None, [w["prompt"] if self.language == "en" else "", context]))
        if prompt:
            fields["prompt"] = prompt
        boundary = uuid.uuid4().hex
        parts = [f'--{boundary}\r\nContent-Disposition: form-data; name="{k}"\r\n\r\n{v}\r\n'.encode()
                 for k, v in fields.items()]
        parts.append(f'--{boundary}\r\nContent-Disposition: form-data; name="file"; filename="clip.wav"\r\n'
                     f"Content-Type: audio/wav\r\n\r\n".encode() + wav_bytes + b"\r\n")
        parts.append(f"--{boundary}--\r\n".encode())
        req = urllib.request.Request(w["url"], data=b"".join(parts), headers={
            "Content-Type": f"multipart/form-data; boundary={boundary}"})
        with urllib.request.urlopen(req, timeout=120) as resp:
            return json.load(resp).get("text", "")

    async def type_text(self, text):
        """Have the input relay type text where typing goes. In order: one at a time."""
        text = to_text(text)
        if not text:
            return
        async with self.type_lock:
            await asyncio.to_thread(self._send_text, text)

    def _send_text(self, text):
        for piece in type_chunks(text):
            try:
                self.relay.sendto(b"type " + piece.encode(), RELAY)
            except OSError as err:
                if not self.relay_said:
                    log.warning("can't type: the input relay isn't there (frametop-input-relay.service): %s", err)
                    self.relay_said = True
                return
        self.relay_said = False

    # --- dictation

    def start_dictation(self, source):
        if self.recording is not None:
            return
        if self.command is not None:
            self.command = None  # dictation wins over a pending wake command
        self.recording, self.recording_source = [], source
        if self.dcfg["streaming"]:
            self.piece_vad.reset()
            # win: first chunk of the current window. lead: the end of the previous window,
            # decoded ahead of it. deferred: lead holds a word not committed yet (a window cut
            # without a pause). words: committed, text: the same as typed, held: punctuation
            # after the last word, waiting to hear if the sentence goes on. typed: chars typed.
            self.stream = {"source": source, "t0": time.monotonic(), "chunks": self.recording,
                           "speech": [], "level": [], "wake": asyncio.Event(), "final": False,
                           "win": 0, "decoded_at": 0, "lead": np.zeros(0, np.float32),
                           "deferred": False, "prev": [], "base": 0,
                           "words": [], "text": "", "held": None, "typed": 0}
            self.stream["worker"] = asyncio.create_task(self._stream_worker(self.stream))
        self.update_mic()
        if "start" in self.sounds:
            self.start_chime = self.loop.call_later(0.6, self.chime_start)
        self.broadcast("dictation_start", source=source)
        log.info("dictation start (%s)", source)

    def drop_start_chime(self):
        """A dictation over before its mic sent audio gets no start tone after all."""
        if self.start_chime:
            self.start_chime.cancel()
            self.start_chime = None

    def cancel_dictation(self, chime=True):
        if self.recording is None:
            return
        self.recording = self.recording_source = None
        if self.stream:
            self.stream["worker"].cancel()
            self.stream = None
        self.drop_start_chime()
        if chime:
            self.chime("cancel")
        self.update_mic()
        self.broadcast("dictation_cancel")
        log.info("dictation cancelled")

    async def stop_dictation(self):
        if self.recording is None:
            return
        chunks, self.recording, self.recording_source = self.recording, None, None
        stream, self.stream = self.stream, None
        self.drop_start_chime()
        seconds = len(chunks) * CHUNK / RATE
        too_short = seconds < self.dcfg["min_seconds"]
        self.chime("cancel" if too_short else "stop")  # before the mic closes: it waits for the tone
        self.update_mic()
        if too_short:
            if stream:
                stream["worker"].cancel()
            log.info("dictation discarded: %.1fs is under the %.1fs minimum", seconds, self.dcfg["min_seconds"])
            self.broadcast("dictation_cancel", reason="too_short", seconds=seconds)
            return
        if stream:
            per = CHUNK / RATE
            log.info("recorded %.1fs: %.1fs labeled speech, peak level %.4f", seconds,
                     sum(stream["speech"]) * per, max(stream["level"], default=0))
        if self.dcfg["save_clips"]:
            self.save_clip(np.concatenate(chunks))
        try:
            if stream:
                stream["final"] = True
                stream["wake"].set()
                await stream["worker"]
                stream["text"] += stream["held"] or ""
                stream["held"] = None
                text = stream["text"]
                if not text:
                    log.info("no speech detected")
            else:
                text = await self.transcribe(np.concatenate(chunks))
        except Exception as err:
            log.error("transcription failed: %s", err)
            self.chime("cancel")
            self.broadcast("error", message=str(err))
            return
        self.broadcast("dictation", text=text)
        if stream:
            await self._flush_typing(stream, final=True)
        elif text:
            await self.type_text(text + (" " if self.dcfg["trailing_space"] else ""))

    def save_clip(self, samples):
        CLIP_DIR.mkdir(parents=True, exist_ok=True)
        (CLIP_DIR / time.strftime("%Y%m%d-%H%M%S.wav")).write_bytes(to_wav(samples))
        for old in sorted(CLIP_DIR.glob("*.wav"))[:-self.dcfg["save_clips"]]:
            old.unlink()

    async def _decode(self, audio, context):
        """Text for a window of audio from the live engine, without the "..." whisper puts on
        text that stops mid-sentence."""
        if self.live_moonshine():
            async with self.whisper_lock:
                text = await asyncio.to_thread(self._moonshine_decode, audio)
                if not text.strip() and len(audio) <= 3 * RATE:
                    # trim_to_speech's VAD is shared, so it runs here; the decodes in a thread
                    text = await asyncio.to_thread(self._moonshine_retry, audio, self.trim_to_speech(audio))
            text = drop_repeats(re.sub(r"\s+", " ", text).strip())
            words = text.split()
            if len(words) >= 3 and len(audio) <= 3 * RATE and len({norm(w) for w in words}) == 1:
                # a lone short word Moonshine looped on ("Way, way, way,"); costs a spoken "no, no, no"
                log.info("short window: %r is one word looped; keeping one", text)
                text = words[-1][:1].upper() + words[-1][1:] if words[0][:1].isupper() else words[-1]
        else:
            text = await self.transcribe_speech(audio, context, quiet=True)
        text = re.sub(r"([,;:!?])\1+", r"\1", text)  # "so,," -> "so,"
        text = re.sub(r"(?<![\w'])i(?=['\s.,!?;:]|$)", "I", text)  # audio starting mid-phrase can give "i"
        return re.sub(r"^(\.\.\.|\u2026)\s*|\s*(\.\.\.|\u2026)$", "", text)

    def _moonshine_retry(self, audio, speech):
        """Moonshine often hears nothing in a lone short word with silence around it, and which
        framing it does hear differs from clip to clip. Try a few: cut to the speech, then each
        after 0.2 s of silence. The first text wins. `speech` is trim_to_speech(audio)."""
        if speech is None:
            return ""
        lead = np.zeros(int(0.2 * RATE), np.float32)
        tries = [speech] if len(speech) < len(audio) else []
        tries += [np.concatenate([lead, speech])]
        for n, clip in enumerate(tries, 1):
            text = self._moonshine_decode(clip)
            if text.strip():
                log.info("short window heard nothing; try %d of %d: %r", n, len(tries), text)
                return text
        log.info("short window heard nothing in %d more tries", len(tries))
        return ""

    def _moonshine_decode(self, audio):
        # Quiet speech is scaled up to a fixed peak first, as Faceclaw does (helped at low levels).
        peak = float(np.abs(audio).max()) or 1.0
        audio = audio * min(10.0, 0.5 / peak)
        stream = self.moonshine.create_stream()
        stream.accept_waveform(RATE, audio)
        self.moonshine.decode_stream(stream)
        return stream.result.text

    async def _stream_worker(self, stream):
        """Live transcription by agreement.

        The current window (speech since the last pause) is re-decoded every step_seconds of
        new audio, after the last context_seconds of the previous window, so it is heard as the
        continuation it is. Each decode is lined up with the words already committed; of the
        words after them, those two decodes in a row agree on are committed and typed, except
        the last, whose punctuation can still change. At a pause, or at the window's quietest
        point once max_window_seconds of audio is decoded, the window closes: it is decoded
        once more and the rest of its words are committed. The punctuation after a pause waits
        for the next window, which hears whether the sentence went on.
        """
        per = CHUNK / RATE
        step = max(1, round(self.dcfg["step_seconds"] / per))
        while True:
            await stream["wake"].wait()
            stream["wake"].clear()
            speech, win = stream["speech"], stream["win"]
            end = len(speech)
            window = speech[win:end]
            last = max((i for i, v in enumerate(window) if v), default=None)
            if last is None and stream["deferred"]:
                last = -1  # no new speech, but the context holds a word still to commit
            full = len(stream["lead"]) / RATE + len(window) * per >= self.dcfg["max_window_seconds"]
            if stream["final"]:
                if last is None and not self._short_speech(stream, end):
                    return
                if full:  # Moonshine returns nothing for more than ~9 s of audio
                    await self._close_window(stream, self._quiet_cut(stream, end), forced=True)
                    stream["wake"].set()
                    continue
                await self._close_window(stream, end)
                return
            if last is None:
                if len(window) * per > 1.0:
                    stream["win"] = stream["decoded_at"] = self._lead_in(stream, end)
                continue
            silence = len(window) - (last + 1)
            if silence * per >= self.dcfg["pause_seconds"]:
                await self._close_window(stream, end - silence + int(0.15 / per))
            elif full:
                await self._close_window(stream, self._quiet_cut(stream, end), forced=True)
            elif end - stream["decoded_at"] >= step:
                stream["decoded_at"] = end
                hyp = (await self._decode(self._window_audio(stream, end), self._context(stream))).split()
                found = self._uncommitted(stream, hyp)
                if found is None:
                    log.info("decode skipped, doesn't line up with the words typed: %s", " ".join(hyp))
                    continue
                new, boundary = found
                before = self._uncommitted(stream, stream["prev"])
                stream["prev"] = hyp
                agreed = agreement(before[0], new) if before else 0
                if agreed > 1:
                    self._commit(stream, new[:agreed - 1], boundary)

    def _short_speech(self, stream, end):
        """Speech the live labels missed in the window: they come ~0.27 s late, so a lone short
        word ("yes") released right away can end before any chunk is labeled speech. The window's
        audio through the VAD to its end (trim_to_speech flushes it) says."""
        if end <= stream["win"]:
            return False
        found = self.trim_to_speech(np.concatenate(stream["chunks"][stream["win"]:end])) is not None
        if found:
            log.info("speech the live labels missed (a short word): decoding it")
        return found

    async def _close_window(self, stream, cut, forced=False):
        """Decode the window once more and commit the rest of its words. After a pause the last
        word's punctuation is held for the next window. Without one (forced), the audio stops
        mid-speech, and Moonshine can garble or repeat the cut-off words ("on a dot on a dot"),
        so only words the last live decode agrees on are committed; the rest (up to 5) are
        carried in the next window's context, sized to hold them."""
        audio = self._window_audio(stream, cut)
        hyp = (await self._decode(audio, self._context(stream))).split()
        found = self._uncommitted(stream, hyp)
        if found is None:
            # Go by position within this window's own audio, as before context was added.
            log.info("window decode doesn't line up with the words typed: %s", " ".join(hyp))
            hyp = (await self._decode(np.concatenate(stream["chunks"][stream["win"]:cut]), self._context(stream))).split()
            found = hyp[len(stream["words"]) - stream["base"]:], None
        new, boundary = found
        start = len(hyp) - len(new)
        context = self.dcfg["context_seconds"]
        if forced:
            before = self._uncommitted(stream, stream["prev"])
            keep = max(0, agreement(before[0], new) - 1 if before else 0, len(new) - 5)
            context = max(context, min(3.5, 0.45 * (len(new) - keep + 2)))  # ~0.3-0.4 s per word
            new = new[:keep]
        self._commit(stream, new, boundary, hold=not forced)
        log.info("window %.1fs + %.1fs context: %s", (cut - stream["win"]) * CHUNK / RATE,
                 len(stream["lead"]) / RATE, " ".join(hyp[:start] + ["|"] + hyp[start:]) if start else " ".join(hyp))
        stream.update(win=cut, decoded_at=cut, lead=self._lead(audio, context), deferred=forced, prev=[],
                      base=len(stream["words"]))

    def _window_audio(self, stream, end):
        return np.concatenate([stream["lead"], *stream["chunks"][stream["win"]:end]])

    def _lead(self, audio, seconds):
        """Context for the next window: about `seconds` from the end of this one, from the
        quietest 30 ms nearby, so it rarely starts inside a word."""
        n = int(seconds * RATE)
        if n <= 0:
            return audio[:0]
        if len(audio) <= n:
            return audio
        frame = RATE * 30 // 1000
        starts = range(max(0, len(audio) - n - RATE // 2), len(audio) - n + RATE // 4, frame)
        return audio[min(starts, key=lambda i: float(np.mean(audio[i:i + frame] ** 2))):]

    def _lead_in(self, stream, end):
        """Start of a window with no speech yet: LEAD_IN_SECONDS before now, or before a sound
        in the last second the VAD didn't flag (a lone short word like "I" is too brief)."""
        per = CHUNK / RATE
        win, level = stream["win"], stream["level"]
        start = end - round(LEAD_IN_SECONDS / per)
        floor = float(np.percentile(level[win:end], 20))
        loud = next((i for i in range(max(win, end - round(1.0 / per)), end)
                     if level[i] > max(3 * floor, 0.005)), None)
        if loud is not None:
            start = min(start, loud - round(0.2 / per))
        return max(win, start)

    def _quiet_cut(self, stream, end):
        """Where to close a window with no pause: after the quietest chunk of its second half,
        preferring ones the VAD calls non-speech (a breath between words, not inside one)."""
        speech, level = stream["speech"], stream["level"]
        first = stream["win"] + (end - stream["win"]) // 2
        return min(range(first, end), key=lambda i: (speech[i], level[i], -i)) + 1

    def _uncommitted(self, stream, hyp):
        """hyp's words after the committed ones, and hyp's form of the last committed word
        (None when it didn't line up). None when hyp doesn't line up with them at all."""
        # Without context, only words committed in this window are in hyp. With it, a decode too
        # short to line up ("Post. The" for "...pose. The") is skipped rather than guessed at.
        ref = stream["words"][max(0, stream["base"] - ALIGN_WORDS) if len(stream["lead"]) else stream["base"]:]
        if not ref:
            return hyp, None
        found = align(ref, hyp)
        return (hyp[found[0]:], found[1]) if found else None

    def _context(self, stream):
        return " ".join(stream["words"][-30:])

    def _commit(self, stream, words, boundary=None, hold=False):
        """Queue words for typing. boundary: this decode's form of the last committed word,
        which settles punctuation held back at a pause. hold: keep the last new word's
        punctuation back until the next window hears whether the sentence goes on."""
        if not words:
            return
        piece = ""
        if stream["held"] is not None:
            heard = None if boundary is None else split_marks(boundary)[1]
            mark = settle_mark(stream["words"][-1] if stream["words"] else "", stream["held"], heard)
            piece, stream["held"] = mark, None
            if mark.endswith(SENTENCE_END) and words[0][:1].islower():
                words = [words[0][:1].upper() + words[0][1:], *words[1:]]
        if hold:
            word, stream["held"] = split_marks(words[-1])
            words = words[:-1] + ([word] if word else [])
        if words:
            piece += (" " if stream["text"] else "") + " ".join(words)
        stream["words"].extend(words)
        stream["text"] += piece
        if not piece:
            return
        self.broadcast("dictation_piece", text=piece.strip())
        task = asyncio.create_task(self._flush_typing(stream))
        self.tasks.add(task)
        task.add_done_callback(self.tasks.discard)

    async def _flush_typing(self, stream, final=False):
        """Type committed text not typed yet: live while recording when allowed, else at the end.
        (The relay holds it back while a modifier is held.)"""
        if not final:
            live = (self.dcfg["live_typing"] and stream is self.stream
                    and time.monotonic() - stream["t0"] >= self.dcfg["live_after_seconds"])
            if not live:
                return
        text = stream["text"][stream["typed"]:]
        stream["typed"] = len(stream["text"])
        if final and stream["text"] and self.dcfg["trailing_space"]:
            text += " "
        if text:
            if not final:
                log.info("typing %d words live (%.1fs in)", len(text.split()), time.monotonic() - stream["t0"])
            await self.type_text(text)

    async def toggle_dictation(self, source):
        if self.recording is None:
            self.start_dictation(source)
        else:
            await self.stop_dictation()

    # --- wake word

    def on_wake(self, keyword):
        log.info("wake word: %s", keyword)
        self.chime("wake")
        self.broadcast("wake", keyword=keyword)
        self.run_hook("wake_hook", keyword, env={"FRAME_VOICE_EVENT": "wake"})
        self.listen_for_command(time.time())

    def listen_for_command(self, wake_unix, again=False):
        self.command = {"chunks": [], "t0": time.monotonic(), "heard": False, "wake_unix": wake_unix,
                        "again": again}
        self.cmd_vad.reset()
        self.update_mic()

    def _command_audio(self, chunk):
        cmd = self.command
        cmd["chunks"].append(chunk)
        feed_vad(self.cmd_vad, chunk)
        while not self.cmd_vad.empty():
            self.cmd_vad.pop()
        speaking = self.cmd_vad.is_speech_detected()
        cmd["heard"] |= speaking
        elapsed = time.monotonic() - cmd["t0"]
        if cmd["heard"] and not speaking or elapsed > self.wcfg["max_command_seconds"]:
            self.command = None
            asyncio.create_task(self.finish_command(np.concatenate(cmd["chunks"]), cmd["wake_unix"], cmd["again"]))
        elif not cmd["heard"] and elapsed > self.wcfg["command_timeout"]:
            self.command = None
            log.info("wake: no command heard")
            self.chime("cancel")
            self.broadcast("command_timeout")
        if self.command is None:
            self.reset_kws()
            self.update_mic()

    async def finish_command(self, samples, wake_unix, again=False):
        self.chime("stop")
        try:
            text = await self.transcribe(samples)
        except Exception as err:
            log.error("transcription failed: %s", err)
            self.broadcast("error", message=str(err))
            return
        # The capture can hold only the wake phrase's end ("Hey, Computer."): the spotter fires
        # late, and a pause after it (waiting for the chime) ends the capture before the command.
        # Then listen once more.
        text, had_wake = voice_commands.strip_wake(text, self.wake_phrases)
        if had_wake and not text.strip(" .!?,") and not again and self.command is None and self.recording is None:
            log.info("wake: only the wake phrase heard, listening again")
            self.chime("wake")
            self.listen_for_command(wake_unix, again=True)
            return
        found = voice_commands.match(text, self.voice_commands + self.named_commands) if text else None
        if found is None:
            self.broadcast("command", text=text)
            if text:
                self.run_hook("command_hook", text, env={"FRAME_VOICE_EVENT": "command"})
            return
        # A voice command runs instead of the command hook. The argv goes straight to
        # exec, so the transcribed words never reach a shell.
        log.info("voice command %r: %s", found.phrase, found.argv)
        self.broadcast("command", text=text, matched=found.phrase, ran=found.argv)
        self.spawn(self._run(found.argv, host=True, env={
            "FRAME_VOICE_EVENT": "voice_command",
            "FRAME_VOICE_TEXT": text,
            "FRAME_VOICE_REST": found.rest,
            "FRAME_VOICE_PHRASE": found.phrase,
            "FRAME_VOICE_WAKE_UNIX": f"{wake_unix:.3f}",
        }))

    # --- push-to-talk

    def push_to_talk(self, command, source, quiet=False):
        """toggle, start, stop or cancel, from the relay (a key combination or mouse button) or
        a controller button. quiet: a cancel without its tone (a hold-to-talk key that turned out
        to be part of a shortcut)."""
        if command == "toggle":
            self.spawn(self.toggle_dictation(source))
        elif command == "start":
            self.start_dictation(source)
        elif command == "stop":
            self.spawn(self.stop_dictation())
        elif command == "cancel":
            self.cancel_dictation(chime=not quiet)

    def listen_relay(self):
        """Push-to-talk from the input relay, on @frametop_voice. Only from this user: the socket
        is abstract, so any local process could send to it."""
        sock = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM | socket.SOCK_NONBLOCK)
        sock.setsockopt(socket.SOL_SOCKET, socket.SO_PASSCRED, 1)  # the kernel adds the sender's uid
        try:
            sock.bind(VOICE)
        except OSError as err:
            log.warning("push-to-talk: can't listen on @frametop_voice (%s)", err)
            sock.close()
            return None
        creds = struct.Struct("3i")  # struct ucred: pid, uid, gid

        def readable():
            while True:
                try:
                    data, anc, _, _ = sock.recvmsg(256, socket.CMSG_SPACE(creds.size))
                except BlockingIOError:
                    return
                uid = next((creds.unpack(d[:creds.size])[1] for level, kind, d in anc
                            if level == socket.SOL_SOCKET and kind == socket.SCM_CREDENTIALS), None)
                words = data.decode(errors="replace").split()
                if uid != os.getuid() or not words or words[0] not in ("toggle", "start", "stop", "cancel"):
                    continue
                self.push_to_talk(words[0], words[1] if len(words) > 1 else "relay", words[2:3] == ["quiet"])

        self.loop.add_reader(sock.fileno(), readable)
        return sock

    def controller_loop(self):
        """Push-to-talk on Frame controller buttons ("controller" in the settings), read from
        SteamVR's web socket in this thread. Waits while SteamVR isn't up, and while no button
        is set."""
        said = None  # what was last logged about the connection
        while True:
            if self.controller[0]:
                try:
                    self._controller_session()
                    said = None
                except (OSError, ValueError) as err:  # urllib's errors are OSErrors
                    if said != "down":
                        log.info("controller push-to-talk: SteamVR isn't reachable (%s); trying every 5 s", err)
                        said = "down"
            self.controller_changed.wait(LOOKUP_EVERY)
            self.controller_changed.clear()

    def _controller_session(self):
        """One connection to SteamVR's web socket. Returns when the buttons are taken away;
        raises OSError when SteamVR goes."""
        ws = vrws.VrSocket()
        held, mode = False, None
        try:
            ws.open(f"frametop_voice_{os.getpid()}")
            sides = {}  # root path -> side; a controller's root path changes (input/vrws.py)
            pressed = set()
            next_poll = 0.0
            watching = None
            while True:
                buttons, mode = self.controller
                if not buttons:
                    return
                if (buttons, mode) != watching:
                    watching = (buttons, mode)
                    pressed &= set(buttons)
                    if held and not pressed:
                        held = False
                        if mode == "hold":
                            self._from_controller("stop")
                    log.info("controller push-to-talk: %s (%s)", ", ".join(buttons), mode)
                now = time.monotonic()
                if now >= next_poll:
                    next_poll = now + LOOKUP_EVERY
                    found = vrws.controllers()
                    if found != sides:
                        for path in sides.keys() - found.keys():
                            ws.subscribe(path, on=False)
                        for path in found.keys() - sides.keys():
                            ws.subscribe(path)
                        sides = found
                        pressed = {b for b in pressed if b.split("/")[0] in sides.values()}
                # (side, component) -> button: "left/dpad_down" is /input/dpad_down/click on the left one.
                wanted = {(b.split("/", 1)[0], f"/input/{b.split('/', 1)[1]}/click"): b for b in buttons}
                for text in ws.messages(timeout=1.0):
                    # About 160 messages a second, nearly all capacitive sensing: parse only
                    # those that carry one of the buttons.
                    if not any(c in text for _, c in wanted):
                        continue
                    try:
                        msg = json.loads(text)
                    except ValueError:
                        continue
                    side = sides.get(msg.get("device"))
                    components = msg.get("components") or {}
                    for (s, c), b in wanted.items():
                        if s == side and c in components:
                            (pressed.add if components[c] else pressed.discard)(b)
                    # After each message: a quick press and release can come in one batch.
                    if pressed and not held:
                        held = True
                        self._from_controller("toggle" if mode == "toggle" else "start")
                    elif held and not pressed:
                        held = False
                        if mode == "hold":
                            self._from_controller("stop")
        finally:
            if held and mode == "hold":
                self._from_controller("stop")
            ws.close()

    def _from_controller(self, command):
        self.loop.call_soon_threadsafe(self.push_to_talk, command, "controller")

    # --- settings

    async def watch_settings(self):
        while True:
            await asyncio.sleep(2)
            if voice_settings.mtime() != self.settings_mtime:
                self.load_settings()
            self.load_named_commands()

    def load_named_commands(self):
        """Your Frametop commands, read again when their file changes. They run through ft-cmd on
        the host (the relay links it into ~/.local/bin)."""
        try:
            mtime = os.stat(ft_commands.COMMANDS_PATH).st_mtime_ns
        except OSError:
            mtime = 0
        if mtime == self.named_mtime:
            return
        self.named_mtime = mtime
        self.named_commands = voice_commands.named(ft_commands.load(), ft_commands.CLI_LINK)
        log.info("Frametop commands: %s", ", ".join(c.phrases[0][0] for c in self.named_commands) or "none")

    def load_settings(self):
        self.settings_mtime = voice_settings.mtime()
        self.apply_options(voice_settings.load())

    def apply_options(self, b):
        """Settings from Frametop Input Settings. Only changed ones are applied, so a save doesn't
        undo e.g. `ft-voice wake off`."""
        changed = {k: b[k] for k in ("controller", "controller_mode", "wake_phrases", "commands", "advanced",
                                     "language", "model", *voice_settings.OPTIONS)
                   if self.applied.get(k) != b[k]}
        first = not self.applied
        self.applied.update(changed)
        if "commands" in changed:
            self.voice_commands, warnings = voice_commands.parse(changed.pop("commands"))
            for warning in warnings:
                log.warning("settings: %s", warning)
            log.info("voice commands: %s", ", ".join(
                f"{c.phrases[0][0]!r} -> {c.run[0]}" for c in self.voice_commands) or "none")
        if "advanced" in changed:
            if not first:
                log.info("advanced settings changed: they apply when frametop-voice restarts")
            changed.pop("advanced")
        if "controller" in changed or "controller_mode" in changed:
            self.controller = (tuple(b["controller"]), b["controller_mode"])
            self.controller_changed.set()
        if "live_typing" in changed:
            self.dcfg["live_typing"] = changed["live_typing"]
        if "sounds" in changed:
            sc = self.cfg["sounds"]
            self.sounds = make_sounds(sc["volume"], sc["lead_seconds"], sc["tail_seconds"]) if changed["sounds"] else {}
        try:
            asyncio.get_running_loop()
            running = True
        except RuntimeError:
            running = False  # at startup; run() opens the mic and starts whisper-server
        if "language" in changed or "model" in changed:
            model = voice_settings.model_for(b)
            restart = model != voice_settings.model_for({"language": self.language, "model": self.model})
            self.language, self.model = b["language"], b["model"]
            if not voice_settings.model_installed(model):
                log.warning("whisper model %s isn't there (%s): whisper-server loads small.en instead; "
                            "fetch it with voice/models.sh model %s", model, voice_settings.model_path(model), model)
            log.info("language %s: %s live, whisper model %s", self.language,
                     "moonshine" if self.live_moonshine() else "whisper", model)
            if running:
                self.update_whisper(restart=restart and not first)
        if "wake_word" in changed and changed["wake_word"] != self.wake_enabled:
            self.wake_enabled = changed["wake_word"]
            self.reset_kws()
            if running:
                self.update_mic()
                self.update_whisper()
        if "wake_phrases" in changed:
            if running:
                self.spawn(self._rebuild_kws(changed["wake_phrases"]))
            else:
                self.kws = self._make_kws(changed["wake_phrases"])
                self.kws_stream = self.kws.create_stream() if self.kws else None
        if changed:
            log.info("settings: %s", ", ".join(f"{k}={v}" for k, v in changed.items()))

    def live_moonshine(self):
        """Live dictation with Moonshine: it knows only English."""
        return self.moonshine is not None and self.language == "en"

    def update_whisper(self, restart=False):
        """Run whisper-server (frametop-voice-whisper.service: ~300 MB of the GPU's memory, more
        for a bigger model) only while something uses it: wake word commands, or dictation set
        to use it (another language than English, or the engine). restart: the model changed."""
        want = (self.wake_enabled or not self.live_moonshine() or self.dcfg["engine"] == "whisper"
                or not self.dcfg["streaming"])
        if restart and want and self.whisper_wanted:
            log.info("whisper-server: restarting for the model the settings chose")
            self.spawn(self._run(["systemctl", "--user", "restart", "frametop-voice-whisper.service"],
                                 host=True, quiet=True))
            return
        if want != self.whisper_wanted:
            if want or self.whisper_wanted is not None:  # at startup, not running is the start
                log.info("whisper-server: %s", "starting" if want else "stopping (nothing uses it)")
            self.whisper_wanted = want
            self.spawn(self._run(["systemctl", "--user", "start" if want else "stop",
                                  "frametop-voice-whisper.service"], host=True, quiet=True))

    # --- control socket

    async def handle_client(self, reader, writer):
        try:
            while line := await reader.readline():
                cmd, _, arg = line.decode().strip().partition(" ")
                reply = await self.command_reply(cmd, arg, writer)
                if reply is None:
                    return  # subscribed to events; keep the connection open
                writer.write((json.dumps(reply) + "\n").encode())
                await writer.drain()
        except (ConnectionError, UnicodeDecodeError):
            pass
        finally:
            if writer not in self.subscribers:
                writer.close()

    async def command_reply(self, cmd, arg, writer):
        match cmd:
            case "toggle":
                await self.toggle_dictation(arg or "socket")
            case "start":
                self.start_dictation(arg or "socket")
            case "stop":
                await self.stop_dictation()
            case "cancel":
                self.cancel_dictation()
            case "type":
                await self.type_text(arg)
            case "wake":
                if arg in ("on", "off", "toggle"):
                    self.wake_enabled = {"on": True, "off": False}.get(arg, not self.wake_enabled)
                    self.reset_kws()
                    self.update_mic()
                else:
                    self.on_wake("manual")
            case "status":
                pass
            case "reload":
                self.load_settings()
            case "events":
                self.subscribers.add(writer)
                writer.write((json.dumps({"event": "subscribed", **self.state()}) + "\n").encode())
                return None
            case _:
                return {"ok": False, "error": f"unknown command {cmd!r}"}
        return {"ok": True, **self.state()}

    async def run(self):
        self.loop = asyncio.get_running_loop()
        RUNTIME_DIR.mkdir(parents=True, exist_ok=True, mode=0o700)
        SOCKET_PATH.unlink(missing_ok=True)
        old_umask = os.umask(0o177)  # socket is owner-only: it can type into any focused app
        server = await asyncio.start_unix_server(self.handle_client, path=str(SOCKET_PATH))
        os.umask(old_umask)
        log.info("control socket: %s", SOCKET_PATH)
        log.info("dictation: min %.1fs, live typing %s", self.dcfg["min_seconds"], self.dcfg["live_typing"])
        relay = self.listen_relay()
        threading.Thread(target=self.controller_loop, name="controller", daemon=True).start()
        self.update_whisper()
        self.update_mic()
        settings = asyncio.create_task(self.watch_settings())
        stop = asyncio.Event()
        for sig in (signal.SIGTERM, signal.SIGINT):
            self.loop.add_signal_handler(sig, stop.set)
        await stop.wait()
        settings.cancel()
        if relay:
            self.loop.remove_reader(relay.fileno())
            relay.close()
        server.close()
        if self.mic_proc and self.mic_proc.returncode is None:
            self.mic_proc.terminate()
        SOCKET_PATH.unlink(missing_ok=True)


def main():
    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
    set_process_name("ft-voiced")
    m = resolve(DEFAULTS["dictation"]["moonshine_model"], MODELS)
    missing = [str(p) for p in (KWS_DIR, VAD_MODEL, m) if not p.exists()]
    if missing:
        log.error("models missing (%s): fetch them with voice/models.sh", ", ".join(missing))
        sys.exit(1)
    asyncio.run(Voice().run())


if __name__ == "__main__":
    main()
