pipeline.py
Rama main · réplica de lectura del repositorio original.
#!/usr/bin/env python3
"""computah pipeline: wake word -> speech-to-text -> brain -> text-to-speech.
The load-bearing core of a local voice assistant, built to run mic-free by
feeding it audio files. No microphone, no network LLM API, no PyTorch.
Stages:
detect_wake(wav) openWakeWord (ONNX) — is the wake phrase present?
transcribe(audio) faster-whisper (CTranslate2, int8) — what was said?
brain(text) dispatches to a persistent assistant session (the bridge)
or, as a dev fallback, the local `claude` CLI subprocess.
speak(text, out) Piper TTS (ONNX) — render the reply to a WAV.
All four are CPU-only and fit in the Pi's RAM. The wake word is changeable
through config.json (see set_wake_word / available_wake_models). The brain
backend is selected by the brain_backend config key; deployment-specific and
sensitive bridge settings live in config.local.json (gitignored), which
overrides config.json.
"""
from __future__ import annotations
import argparse
import collections
import itertools
import json
import logging
import os
import re
import shutil
import subprocess
import sys
import tempfile
import time
import warnings
import wave
from math import gcd
from pathlib import Path
from typing import Callable, NamedTuple
import brain_bridge
# Silence onnxruntime's CUDA-provider chatter before it imports. There is no GPU
# on the Pi; CPU fallback is expected and correct.
os.environ.setdefault("ORT_LOGGING_LEVEL", "3")
warnings.filterwarnings("ignore", message=".*CUDAExecutionProvider.*")
import numpy as np # noqa: E402
import soundfile as sf # noqa: E402
from scipy.signal import resample_poly # noqa: E402
PROJECT_DIR = Path(__file__).resolve().parent
CONFIG_PATH = PROJECT_DIR / "config.json"
# Gitignored overrides for this deployment (persona, ssh host, reply path).
# Keeps infra names and paths out of the published config.json.
LOCAL_CONFIG_PATH = PROJECT_DIR / "config.local.json"
VOICES_DIR = PROJECT_DIR / "voices"
WHISPER_DIR = PROJECT_DIR / "whisper_models"
CUSTOM_MODELS_DIR = PROJECT_DIR / "models" # custom-trained wake words live here
def _resolve_claude_bin() -> str:
"""Locate the claude CLI: prefer PATH, otherwise fall back per OS.
The fallback applies only when claude is not on PATH. On POSIX (the Pi) the
common per-user install is ~/.local/bin/claude. On Windows that POSIX path is
meaningless, so defer to the bare name and let the OS resolver -- or the
graceful FileNotFoundError path in _brain_cli -- handle absence.
"""
found = shutil.which("claude")
if found:
return found
if os.name == "posix":
return str(Path.home() / ".local/bin/claude")
return "claude"
# Prefer the claude binary on PATH; otherwise a platform-appropriate fallback.
CLAUDE_BIN = _resolve_claude_bin()
# openWakeWord ships its pretrained models inside the installed package.
import openwakeword # noqa: E402
_OWW_MODELS_DIR = Path(openwakeword.__file__).parent / "resources" / "models"
# Files that are feature extractors / VAD, not wake-word classifiers.
_NON_WAKE = {"melspectrogram", "embedding_model", "silero_vad"}
# Pretrained classifiers that are not actually wake *phrases* (they are intent
# models bundled with the library). Keep them out of the wake-word library.
_NON_PHRASE = {"timer", "weather"}
DEFAULTS = {
"wake_word": "hey_jarvis",
"wake_threshold": 0.5,
"log_level": "INFO",
"whisper_model": "tiny.en",
"whisper_compute": "int8",
# CTranslate2 uses its own CPU default when cpu_threads is zero. Set a positive
# count in config.local.json to benchmark a fixed thread count on the target Pi.
"whisper_cpu_threads": 0,
# faster-whisper's transcription-time VAD is separate from the capture-time
# Silero gate below.
# Keep it off by default to preserve current output; enable it for Pi measurements.
"whisper_vad_filter": False,
# Mishear guard: gate a transcript on faster-whisper's confidence before it
# reaches the brain, so a garbled command never drives an action. avg_logprob is
# the gate (it must stay at or above the floor); following faster-whisper's own
# no-speech rule, a high no_speech_prob only marks a reject as silence when the
# decode is also unconfident, so a clear command is never dropped for it alone.
# The defaults mirror faster-whisper's log_prob and no_speech thresholds. Set
# stt_confidence_guard false to dispatch every turn.
"stt_confidence_guard": True,
"stt_min_avg_logprob": -1.0,
"stt_max_no_speech_prob": 0.6,
# Capture-time speech gate (#54): after a wake fires, energy endpointing decides
# when the command ends, but energy alone cannot tell a sustained noise from speech,
# so a >=240 ms cough/clap/hum after a bare wake could prepend the wake-word pre-roll
# as a phantom command. capture_request runs the silero VAD over the POST-FIRE audio
# and keeps the pre-roll only if the peak per-chunk speech probability reaches this
# threshold. Lower to confirm quieter/marginal speech; raise to reject more noise.
"capture_vad_threshold": 0.5,
# Request endpointing knobs (#15). After a wake fires, capture_request records the
# command and stops once the trailing quiet reaches endpoint_silence_ms; max_request_ms
# caps a stuck stream so it cannot record forever. Both are milliseconds, rounded to
# whole 80 ms capture frames at use. The defaults equal the built-in frame constants
# (_ENDPOINT_SILENCE_FRAMES=10 -> 800 ms, _MAX_REQUEST_FRAMES=100 -> 8000 ms), so the
# default config leaves capture behavior unchanged; raise endpoint_silence_ms if the
# endpoint clips slow talkers, lower it for a snappier turn. The resolution is one
# 80 ms frame: a value rounds to the nearest 80 ms and never falls below one frame.
"endpoint_silence_ms": 800,
"max_request_ms": 8000,
# Wake-acknowledgment chime (issue #41). Opt-in, default off: on the half-duplex
# PowerConf the cue cannot play while the mic captures, so when it is on a command
# spoken in one breath with the wake word loses its leading audio (it lands in the
# cue window) -- which regresses the pre-roll no-pause case. Off keeps that primary
# flow intact; enable it for a wait-for-cue style. Gating the cue on a pause so both
# can coexist is the deferred fix.
"wake_chime": False,
"voice_model": "en_US-lessac-medium",
# Live audio device selection by case-insensitive name substring (see audio.py).
# Empty string means the system default device. Per-host values belong in
# config.local.json -- e.g. Legion: mic "shure", output "desktop speakers";
# the Pi: "powerconf" over USB.
"mic_device": "",
"output_device": "",
# ALSA PCM for live_driver.py's `aplay -D` when `-o` is omitted (issue #50).
# Kept distinct from output_device above: that is a sounddevice friendly-substring
# (e.g. "desktop speakers") for the audio.py path, and passing it to `aplay -D`
# would fail. This holds a literal ALSA PCM name, e.g.
# "plughw:CARD=PowerConf,DEV=0". Empty string means the ALSA default device.
"live_output_pcm": "",
# Brain backend: "cli" (the claude -p dev fallback, default so a fresh clone
# runs standalone) or "bridge" (a persistent assistant session).
"brain_backend": "cli",
"claude_model": "haiku",
"claude_timeout_s": 60,
# Bridge settings (used when brain_backend == "bridge"). Put real values in
# config.local.json — persona/host/path are deployment-specific and stay out
# of the published config.json.
"brain_persona": "assistant",
"brain_transport": "local", # "local", "ssh", or "sim" (tests)
"brain_host": "", # ssh host alias, required for transport "ssh"
"brain_bot_spren_bin": "bot-spren",
"brain_bot_spren_workdir": "", # bot-spren --working-dir; set so the send lands
# in the session's inbox, not a dead-letter one
"brain_reply_path": "", # path to the persona's FileOutbound reply file
"brain_inbox_path": "", # inbox the session consumes; blank disables the landing
# check. Set an independently verified path to detect a dead-lettered send (#44).
"brain_timeout_s": 120,
"brain_poll_s": 0.5,
}
WAKE_LOGGER = logging.getLogger("computah.wake")
STT_LOGGER = logging.getLogger("computah.stt")
BRAIN_LOGGER = logging.getLogger("computah.brain")
TTS_LOGGER = logging.getLogger("computah.tts")
_LOG_LEVELS = {"DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"}
VOICE_SYSTEM_PROMPT = (
"You are a local voice assistant. Answer in one or two short, plain spoken "
"sentences. No markdown, no bullet points, no code blocks, no emoji."
)
SYL_VOICE_SYSTEM_PROMPT = (
"You are Syl, a local voice assistant and dispatcher. Apply these instructions "
"only to voice events; do not carry their spoken-output constraints into later "
"text or Telegram events. For voice events, keep only pure conversation and "
"quick answers inline. If Joe explicitly asks you to hand work off, or if a "
"request needs tools, multiple steps, or more than a quick answer, it requires "
"delegation. Before delegating a request that creates, changes, sends, or deletes "
"external state, read the intended action back and wait for confirmation on a "
"later voice turn. Do not start the subagent or act until Joe confirms. For other "
"delegated work, and after a required confirmation, start an ephemeral subagent "
"in the background immediately. Use the default tier for routine work and a "
"higher model and effort for hard work, and give the subagent the relevant "
"conversation context. After confirmation when required, do not do delegated "
"work inline or wait for the subagent before replying. Acknowledge the handoff "
"in one short spoken sentence so the voice loop can listen again. Accept "
"follow-up corrections while the subagent runs and redirect it when needed. "
"When a subagent finishes, keep the result in this session's context. At the "
"start of each new voice turn, check for finished background work and eventually "
"summarize each unreported result once. Keep the total spoken reply at most 500 "
"characters. Summarize only the results that fit with the current answer and "
"leave other results unreported for later voice turns. Mark a result as reported "
"only after its summary is included. Never emit a second reply for the original "
"turn because the voice bridge accepts exactly one reply per turn. For replies "
"to voice events, use plain spoken sentences with no markdown, bullet points, "
"code blocks, or emoji."
)
def _bridge_voice_system_prompt(persona: object) -> str:
"""Return Syl's dispatcher policy only for the Syl bridge persona."""
if isinstance(persona, str) and persona.casefold() == "syl":
return SYL_VOICE_SYSTEM_PROMPT
return VOICE_SYSTEM_PROMPT
# Spoken when the mishear guard rejects a low-confidence transcript. Short, so the
# user simply repeats the command; the guard never sends this turn to the brain.
STT_REPROMPT = "Sorry, I didn't catch that. Please say that again."
# Module-level caches so repeated calls in one process do not reload models.
_oww_cache: dict[str, object] = {}
_whisper_cache: dict[tuple[str, str, int], object] = {}
_piper_cache: dict[str, object] = {}
# --------------------------------------------------------------------------- #
# Config + wake-word library
# --------------------------------------------------------------------------- #
def configure_logging(level: str) -> None:
"""Configure the live service's stream-only logger once per process."""
root = logging.getLogger("computah")
if not root.handlers:
handler = logging.StreamHandler()
handler.setFormatter(
logging.Formatter("%(asctime)s %(levelname)s %(name)s %(message)s")
)
root.addHandler(handler)
root.setLevel(level)
root.propagate = False
def load_config() -> dict:
"""Read config, filling any missing key from DEFAULTS.
Layering, lowest to highest precedence: DEFAULTS, then config.json (committed,
non-sensitive), then config.local.json (gitignored deployment overrides).
"""
cfg = dict(DEFAULTS)
for path, label in (
(CONFIG_PATH, "config.json"),
(LOCAL_CONFIG_PATH, "config.local.json"),
):
if not path.exists():
continue
try:
cfg.update(json.loads(path.read_text()))
except json.JSONDecodeError as e:
print(
f"warning: {label} is not valid JSON ({e}); ignoring it",
file=sys.stderr,
)
return validate_config(cfg)
def _supported_whisper_compute_types() -> frozenset:
"""Compute types whisper_compute may take: the "auto"/"default" meta types that let
CTranslate2 choose, plus the concrete quantizations the CPU backend actually supports.
_get_whisper forces device="cpu", and CTranslate2 raises rather than silently falling
back when an explicitly requested type is not supported on the backend (float16,
bfloat16, int16 on a Pi CPU all raise), so a name that merely spells a real type is not
enough -- it has to be one this box can load. get_supported_compute_types adapts to the
host, so the same check is correct on an x86 build that supports more types.
"""
types = {"default", "auto"}
try:
import ctranslate2
types |= set(ctranslate2.get_supported_compute_types("cpu"))
except Exception:
# ctranslate2 missing or too old to report support: cannot validate the type here,
# so defer to _get_whisper rather than reject a name that may well load.
return frozenset()
return frozenset(types)
def _is_number(value: object) -> bool:
"""A real int/float, not a bool. bool is an int subclass, and a JSON true/false is
never a valid numeric knob, so it must be rejected here rather than read as 1/0."""
return isinstance(value, (int, float)) and not isinstance(value, bool)
def _is_positive_int(value: object) -> bool:
return isinstance(value, int) and not isinstance(value, bool) and value > 0
def validate_config(cfg: dict) -> dict:
"""Reject out-of-range or wrong-type config values, falling each bad key back to
its default with a clear stderr message naming the key.
Runs after the DEFAULTS/config.json/config.local.json merge in load_config, so a
typo in a committed or local config fails fast and legibly at load rather than deep
in detect_wake, _get_whisper, or speak on an unattended box. Only keys with a
knowable valid set are checked; free-form keys (whisper_model, device names, paths)
are left to the components that consume them, and unknown extra keys are preserved.
"""
def fall_back(key: str, reason: str) -> None:
print(
f"warning: config {key}={cfg[key]!r} is invalid ({reason}); "
f"using default {DEFAULTS[key]!r}",
file=sys.stderr,
)
cfg[key] = DEFAULTS[key]
# Unit-interval floats: a confidence threshold outside [0, 1] can never fire, or
# always fires, which is worse than a legible reject at load. stt_max_no_speech_prob
# is faster-whisper's no-speech probability, bounded the same way as the two thresholds.
for key in ("wake_threshold", "capture_vad_threshold", "stt_max_no_speech_prob"):
if key in cfg and not (_is_number(cfg[key]) and 0.0 <= cfg[key] <= 1.0):
fall_back(key, "must be a number in [0, 1]")
# Positive-int timeouts: a zero or negative timeout would return instantly or hang.
for key in ("claude_timeout_s", "brain_timeout_s"):
if key in cfg and not _is_positive_int(cfg[key]):
fall_back(key, "must be a positive integer")
# CTranslate2 defines zero CPU threads as its automatic setting. Positive values
# pin a count for target-device measurements; negatives and booleans are invalid.
if "whisper_cpu_threads" in cfg and not (
isinstance(cfg["whisper_cpu_threads"], int)
and not isinstance(cfg["whisper_cpu_threads"], bool)
and cfg["whisper_cpu_threads"] >= 0
):
fall_back("whisper_cpu_threads", "must be a non-negative integer")
if "whisper_vad_filter" in cfg and not isinstance(cfg["whisper_vad_filter"], bool):
fall_back("whisper_vad_filter", "must be true or false")
if "log_level" in cfg:
if not isinstance(cfg["log_level"], str):
fall_back("log_level", "must be a string level name")
else:
log_level = cfg["log_level"].upper()
if log_level not in _LOG_LEVELS:
fall_back("log_level", f"must be one of {sorted(_LOG_LEVELS)}")
else:
cfg["log_level"] = log_level
# Whisper compute type: a typo, or a type this CPU backend can't load, otherwise
# surfaces deep in _get_whisper. Reject non-strings before the membership test, since a
# JSON array/object is unhashable and would raise TypeError instead of falling back.
if "whisper_compute" in cfg:
if not isinstance(cfg["whisper_compute"], str):
fall_back("whisper_compute", "must be a string compute type")
else:
supported = _supported_whisper_compute_types()
if supported and cfg["whisper_compute"] not in supported:
fall_back("whisper_compute", f"must be one of {sorted(supported)}")
# Wake word must resolve to an installed model, reported here instead of at first
# detection. A non-string can never name a model, so it falls back unconditionally;
# otherwise the check is deferred on an empty library (a fresh clone without models
# cannot validate anything) rather than spuriously rewriting a word a later install
# will supply.
if "wake_word" in cfg:
if not isinstance(cfg["wake_word"], str):
fall_back("wake_word", "must be a string model name")
else:
lib = available_wake_models()
if lib and cfg["wake_word"] not in lib:
print(
f"warning: config wake_word={cfg['wake_word']!r} is not an installed "
f"model (available: {sorted(lib)}); using default {DEFAULTS['wake_word']!r}",
file=sys.stderr,
)
cfg["wake_word"] = DEFAULTS["wake_word"]
# live_output_pcm is spliced straight into the `aplay -D` argv by live_driver. A
# non-string from config.local.json (a bare true, a number, a JSON array) would reach
# subprocess.run and raise TypeError, silencing every reply on the hardware path with
# no legible cause. Reject it at load so a typo falls back to the ALSA default instead.
if "live_output_pcm" in cfg and not isinstance(cfg["live_output_pcm"], str):
fall_back("live_output_pcm", "must be a string ALSA PCM name")
return cfg
def _strip_version(filename: str) -> str:
"""alexa_v0.1.onnx -> alexa, hey_jarvis_v0.1.onnx -> hey_jarvis."""
stem = filename[:-5] if filename.endswith(".onnx") else filename
if "_v" in stem:
stem = stem.rsplit("_v", 1)[0]
return stem
def available_wake_models() -> dict[str, str]:
"""Map friendly wake-word name -> absolute .onnx path for installed models.
Excludes the feature-extraction/VAD models and the non-phrase intent models
(timer, weather). These are the words you can switch between instantly.
"""
out: dict[str, str] = {}
for f in sorted(_OWW_MODELS_DIR.glob("*.onnx")):
# Filter feature/VAD models on the raw filename (before version-stripping,
# since "silero_vad" would otherwise split on its "_v").
if f.stem in _NON_WAKE:
continue
name = _strip_version(f.name)
if name in _NON_PHRASE:
continue
out[name] = str(f)
# Custom-trained models in the repo's models/ dir are named directly by file
# stem (e.g. computah.onnx -> "computah") and override built-ins on collision.
if CUSTOM_MODELS_DIR.is_dir():
for f in sorted(CUSTOM_MODELS_DIR.glob("*.onnx")):
out[f.stem] = str(f)
return out
def _resolve_wake_path(name: str) -> str:
lib = available_wake_models()
if name not in lib:
raise ValueError(f"unknown wake word {name!r}. available: {sorted(lib)}")
return lib[name]
def _read_base_config() -> dict:
"""Read only config.json — no DEFAULTS, no config.local.json overlay.
set_wake_word writes back to config.json, so it must round-trip the committed
base alone. Writing the merged dict (load_config) would persist DEFAULTS and,
worse, the gitignored config.local.json overlay (persona, ssh host, reply
path) into the tracked file, defeating the committed/local split.
"""
if CONFIG_PATH.exists():
try:
return json.loads(CONFIG_PATH.read_text())
except json.JSONDecodeError:
pass
return {}
def _read_local_config() -> dict:
"""Read only config.local.json — the gitignored deployment overlay.
set_wake_word(local=True) writes back here, so it must round-trip the local
overlay alone and leave the committed config.json untouched, the mirror of
what _read_base_config does for the base. A malformed overlay raises rather
than falling back to {}: this file also carries the deployment's brain_*
settings, so writing an empty object back over it would silently erase them.
Failing here leaves the file intact for the operator to fix. (load_config's
own read stays tolerant, so a bad overlay degrades the runtime instead of
crashing it — only the write path refuses to clobber.)
"""
if not LOCAL_CONFIG_PATH.exists():
return {}
try:
return json.loads(LOCAL_CONFIG_PATH.read_text())
except json.JSONDecodeError as e:
raise ValueError(
f"config.local.json is not valid JSON ({e}); refusing to overwrite "
"it and risk losing the deployment overrides it carries. Fix the "
"file and retry."
) from e
def set_wake_word(name: str, *, local: bool = False) -> dict:
"""Persist a new active wake word and return the effective config.
By default the change lands in the committed config.json, updating the
built-in default a fresh clone runs with. With local=True it lands in the
gitignored config.local.json instead, so a deployment can activate its own
(often gitignored) wake-word model without dirtying the tracked file. Either
way the target file is round-tripped alone so the committed/local split holds.
"""
_resolve_wake_path(name) # validate before writing
if local:
overlay = _read_local_config()
overlay["wake_word"] = name
LOCAL_CONFIG_PATH.write_text(json.dumps(overlay, indent=2) + "\n")
else:
base = _read_base_config()
base["wake_word"] = name
CONFIG_PATH.write_text(json.dumps(base, indent=2) + "\n")
return load_config()
# --------------------------------------------------------------------------- #
# Audio helpers
# --------------------------------------------------------------------------- #
def _load_pcm16(wav_path: str, target_sr: int = 16000) -> np.ndarray:
"""Load a WAV as mono int16 PCM at target_sr (openWakeWord's input format)."""
audio, sr = sf.read(wav_path, dtype="float32", always_2d=False)
if audio.ndim > 1:
audio = audio.mean(axis=1)
if sr != target_sr:
g = gcd(int(target_sr), int(sr))
audio = resample_poly(audio, target_sr // g, sr // g)
pcm = np.clip(audio, -1.0, 1.0) * 32767.0
return pcm.astype(np.int16)
# --------------------------------------------------------------------------- #
# Stage 1: wake-word detection
# --------------------------------------------------------------------------- #
def _get_oww_model(model_path: str):
if model_path not in _oww_cache:
from openwakeword.model import Model
model = Model(wakeword_model_paths=[model_path])
# A fresh model's preprocessor holds blank audio/feature buffers. Snapshot
# them now so _reset_oww can restore a true clean slate cheaply (see there).
pp = model.preprocessor
model._blank_buffers = (
pp.melspectrogram_buffer.copy(),
pp.feature_buffer.copy(),
)
_oww_cache[model_path] = model
return _oww_cache[model_path]
def _reset_oww(model) -> None:
"""Reset an openWakeWord model to a clean slate between independent streams.
Model.reset() clears only the prediction (score) buffer; the preprocessor's
raw-audio, melspectrogram, and feature buffers (~10 s of history) survive. A
warm model reused across turns therefore carries the previous turn's audio
forward and can false-fire on the next stream. This restores the pristine
buffers snapshotted at model creation — a clean slate without rebuilding the
feature buffer (~0.6 s of embedding compute) on every turn.
"""
model.reset()
pp = model.preprocessor
pp.raw_data_buffer.clear()
pp.accumulated_samples = 0
blank = getattr(model, "_blank_buffers", None)
if blank is None:
# A model not built by _get_oww_model (e.g. a mic adapter's own Model).
# Snapshot on first reset: a freshly constructed model is in the blank
# state, which is the only safe moment to capture it.
blank = (pp.melspectrogram_buffer.copy(), pp.feature_buffer.copy())
model._blank_buffers = blank
mel, feat = blank
pp.melspectrogram_buffer = mel.copy()
pp.feature_buffer = feat.copy()
# Leading and trailing silence to fill a freshly reset model's ~2s context window.
# A one-shot clip must fill that window before the real audio; live audio fills it
# naturally. Named here so every one-shot scoring path (detect_wake and the
# threshold-tuning eval) pads identically and cannot drift apart.
_WAKE_LEAD_SAMPLES = 24000 # 1.5s leading silence
_WAKE_TAIL_SAMPLES = 8000 # 0.5s trailing silence
def wake_frame_scores(wav_path: str, model_name: str | None = None) -> list[float]:
"""Per-frame wake scores for a one-shot clip, in order.
Resets the active model to a clean slate and pads the clip with leading (plus a
short trailing) silence so the detector's ~2s context window is full before the
real audio: the wake word can sit at the very start of a clip of any length, and
a freshly reset detector would otherwise under-score the opening. Silence scores
~0, so the pad never raises the score for a clip that already had context. The
clip is scanned in 80ms (1280-sample) frames and each frame's highest score is
kept. detect_wake takes the peak of these for a fire decision; the
threshold-tuning eval keeps the whole list so it can apply any threshold and count
fires over the clip.
"""
model_name = model_name or load_config()["wake_word"]
model = _get_oww_model(_resolve_wake_path(model_name))
_reset_oww(model) # full clean slate between independent clips
pcm = _load_pcm16(wav_path)
lead = np.zeros(_WAKE_LEAD_SAMPLES, dtype=np.int16)
tail = np.zeros(_WAKE_TAIL_SAMPLES, dtype=np.int16)
pcm = np.concatenate([lead, pcm, tail])
step = 1280 # 80ms at 16kHz — openWakeWord's frame size
scores: list[float] = []
for i in range(0, len(pcm) - step + 1, step):
preds = model.predict(pcm[i : i + step])
scores.append(max(float(s) for s in preds.values()))
return scores
def detect_wake(
wav_path: str, model_name: str | None = None, threshold: float | None = None
) -> tuple[bool, str, float]:
"""Run audio through the active openWakeWord model.
Returns (fired, model_name, peak_score). fired is peak_score >= threshold. The
clip is scored frame by frame (see wake_frame_scores) and the highest score over
the clip is the detection score.
"""
cfg = load_config()
model_name = model_name or cfg["wake_word"]
threshold = cfg["wake_threshold"] if threshold is None else threshold
scores = wake_frame_scores(wav_path, model_name)
peak = max(scores) if scores else 0.0
return peak >= threshold, model_name, peak
# --------------------------------------------------------------------------- #
# Stage 1b: live streaming — continuous detection + request capture
# --------------------------------------------------------------------------- #
# These drive a turn from a frame stream instead of a whole file. A live mic and
# iter_wav_frames yield the same thing — consecutive 80 ms int16 frames — so the
# detection and capture logic is identical and testable with no hardware. The
# physical mic and speaker adapters land with the microphone (issues #9, #11).
FRAME_SIZE = 1280 # 80 ms at 16 kHz — openWakeWord's frame size
_SILENCE_RMS = 250.0 # int16 RMS below this is room tone, not speech
_ENDPOINT_SILENCE_FRAMES = 10 # ~0.8 s of trailing quiet ends a captured request
_NO_SPEECH_ONSET_FRAMES = 20 # ~1.6 s; abandon a wake that no speech follows
_SPEECH_ONSET_FRAMES = 3 # consecutive above-threshold frames before energy counts
# as speech onset; a click/cough/echo is 1-2 frames, so a
# sustained run keeps a transient from tripping onset and
# prepending the wake-word pre-roll as a phantom (#54)
_MAX_REQUEST_FRAMES = 100 # ~8 s cap so a stuck stream cannot record forever
_PREROLL_FRAMES = 2 # ~0.16 s of audio kept before the wake fire and
# prepended to the request, so a command spoken with
# no pause after the wake word is not clipped by
# detection latency (issue #30). Tuned on the deployed
# PowerConf with a live voice test: at 8 the wake word
# itself bled into the transcript ("computer file an
# issue..."); at 2 (~160 ms) the no-pause command keeps a
# clip-margin while the wake word stays out. The command
# onset never clipped at any size here, so the real risk
# was over-capture, not under-capture. Larger recovers
# more leading audio but bleeds more of the wake word in.
_WAKE_HISTORY_FRAMES = 40 # ~3.2 s retained only for empty-capture recovery (#53).
# This is deliberately separate from the two-frame pre-roll above: normal requests
# must not receive the wake phrase, while a command fully consumed during detection
# needs enough context for transcription to distinguish "wake" from "wake + command."
_CUE_PEEK_FRAMES = _SPEECH_ONSET_FRAMES + 1 # ~0.32 s window peeked after the wake
# word to tell a no-pause command from a pause before the
# acknowledgment cue plays (issue #55). This is the shortest
# silence that counts as a pause: a voiced run still open at
# the window's end is read past it (#106), so a late onset is
# not missed. Larger demands a longer pause and delays the cue
# on the paused case -- the cue plays only after this window
# is read, at 80 ms/frame.
_FRAME_MS = FRAME_SIZE / 16000 * 1000 # 80.0 ms of audio per capture frame
class WakeAudioBuffer:
"""Keep normal pre-roll and longer recovery history from one detection stream.
Detectors only need an append-capable object, so this preserves their existing
caller-owned-buffer contract while maintaining two independently bounded views.
`preroll` remains the short, wake-tail-safe seed for normal capture. `history` is
transcribed only after capture found no post-fire speech, when it can prove that a
fully consumed command followed the wake phrase (#53).
"""
def __init__(self) -> None:
self.preroll = collections.deque(maxlen=_PREROLL_FRAMES)
self.history = collections.deque(maxlen=_WAKE_HISTORY_FRAMES)
def append(self, frame: np.ndarray) -> None:
self.preroll.append(frame)
self.history.append(frame)
def extend(self, frames) -> None:
for frame in frames:
self.append(frame)
def clear(self) -> None:
self.preroll.clear()
self.history.clear()
class _EmptyCapture(np.ndarray):
"""A zero-length PCM array that records why capture rejected the turn."""
_EMPTY_NO_ONSET = "no_speech_onset"
_EMPTY_VAD_REJECTED = "vad_rejected"
def _empty_capture(reason: str) -> np.ndarray:
pcm = np.zeros(0, dtype=np.int16).view(_EmptyCapture)
pcm.empty_reason = reason
return pcm
def _ms_to_frames(ms: float) -> int:
"""Round a millisecond duration to a whole number of 80 ms capture frames.
Capture endpointing counts frames, so a config value in ms is converted here. The
floor of 1 keeps a small positive value from rounding to zero frames, which would
make an endpoint or cap check fire on the first frame."""
return max(1, round(ms / _FRAME_MS))
def iter_wav_frames(wav_path: str):
"""Yield consecutive 80 ms int16 frames from a WAV: a file-backed stand-in for
a live mic. A microphone source yields frames of the same size and dtype, so
everything downstream is identical."""
pcm = _load_pcm16(wav_path)
for i in range(0, len(pcm) - FRAME_SIZE + 1, FRAME_SIZE):
yield pcm[i : i + FRAME_SIZE]
def _frame_rms(frame: np.ndarray) -> float:
return float(np.sqrt(np.mean(frame.astype(np.float32) ** 2)))
def stream_detect_wake(frames, model, threshold: float, preroll=None) -> float | None:
"""Watch a continuous frame stream for the wake word.
detect_wake handles a one-shot file by resetting the model and padding the clip
so its ~2 s context window fills. A live stream fills that window naturally, so
here the model is reset ONCE and fed frame by frame with no padding. Stops at
and consumes the first frame whose score crosses threshold, returning the peak
score so far; returns None if the stream ends with no detection. Leaves the
iterator positioned right after the wake word so the caller can capture the
request that follows.
`preroll`, if given, is a bounded collection (a deque(maxlen=N)) that each frame
is appended to as it is consumed, so on a fire it holds the most recent N frames
-- including the firing frame. capture_request prepends those to the request,
recovering the leading audio that detection latency consumes, so a command spoken
with no pause after the wake word is not clipped (issue #30).
"""
_reset_oww(model)
peak = 0.0
for frame in frames:
if preroll is not None:
preroll.append(frame)
score = max(float(s) for s in model.predict(frame).values())
peak = max(peak, score)
if score >= threshold:
return peak
return None
_vad_model = None
def _get_vad():
"""Cache the bundled silero VAD across calls (one resident instance, like
_get_oww_model / _get_whisper). The ONNX ships with openWakeWord, so there is no
download and no extra dependency."""
global _vad_model
if _vad_model is None:
from openwakeword.vad import VAD
_vad_model = VAD()
return _vad_model
def _confirm_speech(pcm: np.ndarray, threshold: float) -> bool:
"""True if `pcm` (post-fire command audio, int16 16 kHz) holds real speech.
Runs the silero VAD over 480-sample (30 ms) chunks and keeps the PEAK chunk
probability, not the mean the model returns over a whole buffer -- a short command
must not be diluted to a non-speech average by trailing silence or noise in the
capture. Reset the RNN state once, then feed chunks in order so state carries across
them. Pass the command audio ONLY, never the pre-roll: the pre-roll holds the wake
word, which is speech, so including it would always confirm and defeat the gate.
silero rejects coughs, claps, echoes, and sustained noise where an energy threshold
cannot (#54).
"""
vad = _get_vad()
vad.reset_states()
frame = 480
pad = (-len(pcm)) % frame # silero needs a length that is a multiple of frame
if pad:
pcm = np.concatenate([pcm, np.zeros(pad, dtype=np.int16)])
peak = 0.0
for i in range(0, len(pcm), frame):
prob = float(vad.predict(pcm[i : i + frame], frame_size=frame))
if prob > peak:
peak = prob
return peak >= threshold
def peek_cue_gate(frames, vad_threshold=None):
"""Peek the post-wake audio to decide whether the acknowledgment cue plays (#55).
The cue (#41) and no-pause capture (#30) conflict on a half-duplex device: the cue
cannot play while the mic captures, so a command spoken in one breath with the wake
word ("computah file an issue") lands in the cue window and is lost. Reading a short
window first tells the two cases apart -- a no-pause command already has speech here,
a user waiting for the cue has room tone -- so the cue can be skipped when it would
clip a command and played when it is safe.
Pulls _CUE_PEEK_FRAMES frames off `frames` (more while a voiced run is open at the
window's end, see below), stopping the instant a sustained
voiced run appears (_SPEECH_ONSET_FRAMES in a row -- the same onset test
capture_request uses, so a lone click or cough edge does not count as speech).
When `vad_threshold` is given, the sustained-energy candidate must also pass the
same Silero VAD used by capture_request. This keeps a fan or room rumble from
suppressing the cue when the user has paused.
Returns (play_cue, peeked): play_cue is False when speech is already present (skip
the cue, capture the command intact), True when the window is silent (safe to play
the cue, then capture). `peeked` is every frame pulled; because `frames` is a
one-shot iterator, the caller MUST feed `peeked` back to capture ahead of the rest
of the stream on the no-pause branch, so peeking never drops command audio. On the
silent branch `peeked` is pre-cue room tone the caller drops with the cue bleed.
A voiced run still open when the window closes is read until it resolves (#106). A command
that starts in the last frame or two of a very short pause has fewer than
_SPEECH_ONSET_FRAMES voiced frames inside the window; deciding there would read it as
a pause, play the cue, and drop that leading fragment with the cue bleed. The fragment
cannot be stitched back, because the mic is flushed while the cue plays. So the peek
keeps reading until the run either completes an onset (skip the cue) or breaks on a
quiet frame (a transient; play it). The extension is at most _SPEECH_ONSET_FRAMES - 1
frames (~160 ms), it delays the cue only when the window ends on sound, and a silent
window still decides at _CUE_PEEK_FRAMES. Skipping the cue on any voiced frame would
also avoid the clip, but a click or cough at the edge would then suppress the cue;
resolving the run keeps that transient rejection for a bounded delay.
"""
peeked: list[np.ndarray] = []
voiced_run = 0
speech = False
while len(peeked) < _CUE_PEEK_FRAMES or voiced_run:
frame = next(frames, None)
if frame is None:
break # stream ended inside the peek; decide on what was seen
peeked.append(frame)
if _frame_rms(frame) < _SILENCE_RMS:
voiced_run = 0
else:
voiced_run += 1
if voiced_run >= _SPEECH_ONSET_FRAMES:
speech = True
break
if speech and vad_threshold is not None:
speech = _confirm_speech(np.concatenate(peeked), vad_threshold)
return (not speech), peeked
def capture_request(
frames,
preroll=None,
vad_threshold=None,
endpoint_silence_ms=None,
max_request_ms=None,
) -> np.ndarray:
"""Collect request frames after a wake fire until a trailing-silence endpoint.
Energy-based endpointing: once speech has been seen, stop after a window of
consecutive quiet frames. If no speech ever starts, stop after the shorter
_NO_SPEECH_ONSET_FRAMES window so a false or abandoned wake frees the listener in
~1.6 s instead of holding it for the full max-request cap. Returns the captured
int16 PCM, or an empty array if no speech followed the wake word. Transcribing
that silence would feed whisper room tone, which it tends to hallucinate words
for, so the empty return lets the caller ignore the turn instead.
`preroll`, if given, is a list of frames captured just before the wake fired
(see stream_detect_wake). They are prepended to the returned PCM so a request
spoken immediately after the wake word keeps its leading audio, which detection
latency would otherwise have consumed (issue #30). The pre-roll takes no part in
the speech-onset / endpoint decision -- that runs only over the frames after the
wake word. Onset requires a sustained run of above-threshold frames
(_SPEECH_ONSET_FRAMES), not a single loud frame, so a transient after the wake --
a click, a cough edge, a speaker echo -- does not count as speech. A wake with no
sustained speech after it still returns empty, so the wake word held in the pre-roll
is never prepended and transcribed as a phantom request (#54).
The energy onset is a cheap first filter: it rejects a 1-2 frame transient without
running a model and bounds how long an abandoned wake holds the mic. It cannot tell
a *sustained* noise (a long cough, a fan, room rumble past _SPEECH_ONSET_FRAMES)
from speech, though, so when `vad_threshold` is given the captured command audio is
confirmed by the silero VAD (_confirm_speech) before the pre-roll is kept. Energy
that the VAD rejects is treated as an abandoned wake and returns empty, closing the
phantom path the energy gate alone leaves open (#54). With `vad_threshold` None the
VAD step is skipped and only the energy gate runs.
`endpoint_silence_ms` and `max_request_ms` (#15), when given, override the built-in
trailing-silence window and runaway cap; each is milliseconds, rounded to whole 80 ms
frames. Left None they fall back to _ENDPOINT_SILENCE_FRAMES and _MAX_REQUEST_FRAMES,
so a caller that does not pass config -- the file-fed tests -- keeps the original
behavior. run_turn threads the config values (endpoint_silence_ms, max_request_ms)
through so the live loop is tunable without a code change.
"""
endpoint_frames = (
_ENDPOINT_SILENCE_FRAMES
if endpoint_silence_ms is None
else _ms_to_frames(endpoint_silence_ms)
)
max_frames = (
_MAX_REQUEST_FRAMES if max_request_ms is None else _ms_to_frames(max_request_ms)
)
captured: list[np.ndarray] = []
quiet = 0
voiced_run = 0
speech_seen = False
for frame in frames:
captured.append(frame)
if _frame_rms(frame) < _SILENCE_RMS:
quiet += 1
voiced_run = 0
else:
quiet = 0
voiced_run += 1
if voiced_run >= _SPEECH_ONSET_FRAMES:
speech_seen = True # sustained energy, not a lone transient
if speech_seen and quiet >= endpoint_frames:
break
if not speech_seen and quiet >= _NO_SPEECH_ONSET_FRAMES:
break # nothing said after the wake — abandon fast, don't wait the cap
if len(captured) >= max_frames:
break
if not speech_seen:
# No sustained speech after the wake fired (silence, or only a transient like a
# click or cough). Never prepend the short pre-roll here: it always holds the wake
# phrase, so doing that based on energy would turn a bare wake into a phantom
# request. A caller can transcribe a separate, longer detection history;
# recover_consumed_command then requires a wake-prefixed transcript with a
# nonempty suffix before dispatch (#53).
return _empty_capture(_EMPTY_NO_ONSET)
command = np.concatenate(captured)
if vad_threshold is not None and not _confirm_speech(command, vad_threshold):
# Energy was sustained but the VAD says it is not speech (a long cough, a fan,
# room rumble). Treat it as an abandoned wake: drop the pre-roll so the wake word
# it holds is never prepended and transcribed as a phantom command (#54).
return _empty_capture(_EMPTY_VAD_REJECTED)
if preroll:
return np.concatenate([*preroll, command])
return command
# --------------------------------------------------------------------------- #
# Stage 2: speech-to-text
# --------------------------------------------------------------------------- #
def _get_whisper(model: str, compute: str, cpu_threads: int):
key = (model, compute, cpu_threads)
if key not in _whisper_cache:
from faster_whisper import WhisperModel
_whisper_cache[key] = WhisperModel(
model,
device="cpu",
compute_type=compute,
cpu_threads=cpu_threads,
download_root=str(WHISPER_DIR),
)
return _whisper_cache[key]
class Transcript(NamedTuple):
"""A transcription plus the faster-whisper signals the mishear guard reads.
avg_logprob is the mean per-token log-probability (closer to 0 is more
confident); no_speech_prob is how silence-like the audio looked (higher means
less likely to be real speech).
"""
text: str
avg_logprob: float
no_speech_prob: float
def _aggregate_segments(segments) -> Transcript:
"""Collapse faster-whisper segments into one Transcript.
avg_logprob is averaged across segments weighted by their duration, so a long
confident clause outweighs a short hesitant one; no_speech_prob is the max
across segments, the most conservative summary -- any silence-like span pulls
confidence down. Pure over the segment objects (it reads only .text, .start,
.end, .avg_logprob, .no_speech_prob), so it is unit-testable with lightweight
stand-ins and no model.
"""
segs = list(segments)
text = " ".join(s.text for s in segs).strip()
if not segs:
# No segments at all: empty text that fails the guard on both signals.
return Transcript(text, float("-inf"), 1.0)
weights = [max(float(s.end) - float(s.start), 1e-3) for s in segs]
total = sum(weights)
avg_logprob = sum(float(s.avg_logprob) * w for s, w in zip(segs, weights)) / total
no_speech_prob = max(float(s.no_speech_prob) for s in segs)
return Transcript(text, avg_logprob, no_speech_prob)
def _whisper_audio_input(
audio: str | os.PathLike[str] | np.ndarray,
) -> str | np.ndarray:
"""Return an input in the format faster-whisper expects.
File callers stay compatible. Live callers can pass captured 16 kHz mono PCM
directly, avoiding a temporary WAV: signed 16-bit samples are normalized to
float32 in [-1, 1), while floating-point arrays must already be normalized.
"""
if not isinstance(audio, np.ndarray):
return os.fspath(audio)
if audio.ndim != 1:
raise ValueError("transcription audio must be a one-dimensional mono array")
if audio.dtype == np.int16:
return audio.astype(np.float32) / 32768.0
if np.issubdtype(audio.dtype, np.floating):
normalized = audio.astype(np.float32, copy=False)
if not np.all(np.isfinite(normalized)):
raise ValueError("transcription audio must contain only finite samples")
if normalized.size and float(np.max(np.abs(normalized))) > 1.0:
raise ValueError("floating-point transcription audio must be normalized")
return normalized
raise TypeError("transcription arrays must use int16 or a floating-point dtype")
def transcribe_detailed(
audio: str | os.PathLike[str] | np.ndarray,
) -> Transcript:
"""Transcribe a WAV path or normalized 16 kHz mono array with faster-whisper.
Returns the text plus the confidence signals the mishear guard needs.
transcribe() wraps this for callers that want only the text.
"""
cfg = load_config()
model = _get_whisper(
cfg["whisper_model"],
cfg["whisper_compute"],
cfg["whisper_cpu_threads"],
)
segments, _info = model.transcribe(
_whisper_audio_input(audio),
beam_size=1,
language="en",
vad_filter=cfg["whisper_vad_filter"],
# Each capture is one independent command. For audio that spans decoding
# windows, carrying one window's text into the next adds work and can
# reinforce an early mishearing without useful continuity for this use case.
condition_on_previous_text=False,
)
return _aggregate_segments(segments)
_WAKE_TRANSCRIPT_ALIASES = {
# Whisper commonly writes the custom phonetic wake "computah" as this standard
# spelling. Keep aliases explicit rather than fuzzy-matching arbitrary words: this
# path is a last-chance action dispatch after an otherwise abandoned wake.
"computah": (("computah",), ("computer",)),
}
def recover_consumed_command(detection_frames, wake_word: str) -> Transcript | None:
"""Recover a command consumed before post-fire speech onset could be confirmed.
Transcribe the bounded detection history after normal post-fire capture found no
speech. Dispatch remains fail-closed: the transcript must start with the configured
wake phrase (or an explicit spelling alias) and contain at least one later word. A
bare wake and a false detector fire therefore both return None. The returned command
retains the original decoder confidence so callers apply the normal guard before it
can reach the brain.
"""
detection = list(detection_frames)
if not detection:
return None
audio = np.concatenate(detection)
heard = transcribe_detailed(audio)
words = list(re.finditer(r"[a-z0-9]+(?:'[a-z0-9]+)?", heard.text, re.IGNORECASE))
tokens = tuple(match.group(0).casefold() for match in words)
# `wake_word` is the configured friendly model name, not a versioned model file.
# Using _strip_version here would truncate legitimate names at any `_v` (for
# example `hey_vera` -> `hey`) and weaken the false-wake prefix check.
canonical = wake_word.replace("_", " ").replace("-", " ")
phrases = _WAKE_TRANSCRIPT_ALIASES.get(
canonical.casefold(),
(tuple(re.findall(r"[a-z0-9]+", canonical.casefold())),),
)
for phrase in phrases:
if phrase and tokens[: len(phrase)] == phrase and len(tokens) > len(phrase):
command = heard.text[words[len(phrase)].start() :].strip()
return Transcript(command, heard.avg_logprob, heard.no_speech_prob)
return None
def transcribe(audio: str | os.PathLike[str] | np.ndarray) -> str:
"""Transcribe a WAV path or normalized 16 kHz mono array. Returns the text."""
return transcribe_detailed(audio).text
def transcript_confident(
t: Transcript, *, min_avg_logprob: float, max_no_speech_prob: float
) -> tuple[bool, str]:
"""Is a transcript trustworthy enough to act on?
The brain acts on voice commands -- it files issues, changes things -- so a
misheard transcript must be stopped before it dispatches. The decoder's
avg_logprob is the gate: below the floor the words are unreliable and the turn
is rejected. Following faster-whisper's own no-speech rule, a confidently decoded
command is never rejected for a high no_speech_prob alone (a confident decode
overrides it); no_speech_prob only distinguishes a silence-derived reject (it
over the ceiling and avg_logprob under the floor) from a plain garbled one, which
sharpens the logged reason without ever re-prompting a clear command. Returns
(ok, reason); reason names why a turn was dropped.
"""
if not t.text.strip():
return False, "empty transcript"
if t.avg_logprob < min_avg_logprob:
kind = "silence" if t.no_speech_prob > max_no_speech_prob else "low confidence"
return False, (
f"{kind}: avg_logprob {t.avg_logprob:.2f} under floor {min_avg_logprob:.2f}"
)
return True, "ok"
def guard_transcript(heard: Transcript, cfg: dict) -> tuple[bool, str]:
"""Apply the configured mishear guard to a transcription.
Returns (ok, reason): ok is True when the guard is disabled or the transcript
clears it, and False when a low-confidence transcript should be re-prompted
rather than dispatched. Both live paths -- run_turn here and live_driver's own
turn loop -- gate through this one function, so the brain cannot be reached
unguarded on one path while the other is protected.
"""
if not cfg["stt_confidence_guard"]:
return True, "guard disabled"
return transcript_confident(
heard,
min_avg_logprob=cfg["stt_min_avg_logprob"],
max_no_speech_prob=cfg["stt_max_no_speech_prob"],
)
# --------------------------------------------------------------------------- #
# Stage 3: the brain (persistent session via the bridge, or the claude CLI)
# --------------------------------------------------------------------------- #
# A reply is model output going straight to Piper, so VOICE_SYSTEM_PROMPT asking
# for short plain text is a request, not a guarantee: a confused or truncated
# model still emits markdown, emoji, control bytes, or pages of prose. These
# bound what can reach speak() regardless of what the model did.
MAX_SPOKEN_CHARS = 600
EMPTY_REPLY_FALLBACK = "Sorry, I don't have an answer for that."
# A whole ANSI escape sequence, not just its ESC byte: _brain_cli returns the
# claude CLI's stdout, so colour codes are a realistic reply. Dropping only the
# ESC would leave "[31m" behind to be read aloud. The CSI branch is the ECMA-48
# grammar rather than the parameter bytes seen so far: any of 0x30-0x3f may appear
# before the final byte, so a colon-separated true-colour SGR ("\x1b[38:2::255:0:0m")
# is one sequence, and a narrower pattern leaves its parameters to be spoken.
_ANSI_ESCAPE_RE = re.compile(
r"\x1b(?:\[[0-?]*[ -/]*[@-~]?" # CSI (colour, cursor moves), terminated or not
r"|\][^\x07\x1b]*(?:\x07|\x1b\\)?" # OSC (window title), terminated or not
r"|[@-Z\\-_])" # two-character escapes
)
_CONTROL_CHARS_RE = re.compile(r"[\x00-\x08\x0b\x0c\x0e-\x1f\x7f-\x9f]")
# U+20E3 is the enclosing keycap: "1" + U+FE0F + U+20E3 is one glyph, so taking only the
# variation selector leaves the digit still wearing a combining mark, which is the emoji
# this class promises never reaches Piper. U+FE0E is its text-presentation twin.
_EMOJI_RE = re.compile(
"[\U0001f000-\U0001faff\U00002600-\U000027bf\U00002b00-\U00002bff"
"\U00002190-\U000021ff\ufe0e\ufe0f\u20e3\u200d]"
)
_LINK_OPEN_RE = re.compile(r"!?\[")
_WHITESPACE_RE = re.compile(r"\s")
# Quote and list markers, the two containers a fenced block can sit inside. Headings are not
# containers and are handled separately, after fences.
#
# A list marker needs a space to be a marker at all ("-foo" is a word), but a quote marker does
# not: ">" alone opens a quote, so ">```" is a fenced block inside one. Requiring a space after
# ">" left that marker in place, the fence never reached a line start, and the block's body was
# spoken -- the exact failure the ordering above exists to prevent. So the two take separate
# branches rather than sharing one whitespace rule. The space stays optional-but-single, which
# is what markdown counts as the marker; any further indent belongs to the content.
#
# The trailing + takes a line's whole marker run in one match, which is what lets the caller do
# a single pass. Stripping one marker per pass and re-running until the text stopped changing
# read the same, and cost a pass per marker with a full rescan each time: ">"*8000 took 0.29s
# and scaled as the square, so a reply that was nothing but quote markers could stall a loop
# that is always listening. Each repetition consumes at least its marker character, so the run
# cannot match empty and the + cannot spin.
_CONTAINER_RE = re.compile(r"(?m)^(?:[ \t]{0,3}(?:(?:[-*+]|\d+[.)])[ \t]+|>[ \t]?))+")
# A complete fenced block: an opener at a line start (up to 3 spaces, as markdown allows),
# closed by a run of the SAME character at least as long. Either character opens a block, so a
# stray "~~~" inside a backtick block cannot close it early.
#
# (?!\2) forces the opener to take its whole run. Without it the pattern could give a character
# back -- matching only three of a "~~~~" opener, letting .*? eat the fourth, and then accepting
# a later "~~~" as a closer that markdown says is too short to close anything, so what is still
# code gets spoken.
#
# Anchoring to a line start is what keeps a reply that merely MENTIONS ``` mid-sentence from
# reading as a block, whereupon the unterminated-fence rule would drop every word after it: an
# answer explaining how to open a code block would lose the explanation, silently, and nothing
# downstream can tell speech from speech-that-was-cut. An inline run left behind is not spoken
# anyway -- it pairs off as a code span, or the final tick sweep takes it.
_FENCED_BLOCK_RE = re.compile(
r"(?ms)^[ ]{0,3}((`|~)\2{2,})(?!\2).*?^[ ]{0,3}\1\2*[ \t]*$"
)
# Emphasis is a matched pair of identical delimiter runs hugging its content, which is
# the only thing markdown reads as emphasis. An unpaired run is literal text, and a
# reply about code is made of unpaired runs (*args, **kwargs, _private, VALUE_) whose
# delimiter is part of the name being spoken. A run that whitespace keeps from hugging
# anything (5 * 6) opens nothing either, and \w on the outer edges keeps an underscore
# inside a word (snake_case) from reading as a delimiter at all.
# One pattern per delimiter, each barred from crossing its own character: an unpaired
# run then gives up at the next candidate rather than rescanning the rest of the
# reply, which keeps a long reply -- the case this whole stage exists for -- linear.
_EMPHASIS_RES = (
re.compile(r"(?<!\w)(\*{1,3})(?!\s)([^*]+)(?<!\s)\1(?!\w)"),
re.compile(r"(?<!\w)(_{1,3})(?!\s)([^_]+)(?<!\s)\1(?!\w)"),
re.compile(r"(?<!\w)(~~)(?!\s)([^~]+)(?<!\s)\1(?!\w)"),
)
def _tick_run_end(text: str, start: int) -> int:
"""Index just past the backtick run beginning at `start`."""
i = start
while i < len(text) and text[i] == "`":
i += 1
return i
def _iter_code_spans(text: str):
"""Yield (start, end, contents) for each code span, left to right.
Markdown closes a backtick run with the next run of exactly the same length, and a run
that never finds its match is literal text rather than an opener. Comparing runs is a
direct reading of that rule; the pattern this replaces, (`+)(.+?)\\1, could only
approximate it by trying every opener length against every closing position, which cost
cubic time in a run's length -- 4,000 ticks took 2.3s and 8,000 about 18s. A reply is
model output arriving at a loop that is always listening, so a long enough run of one
character was enough to stop it answering at all.
Runs are indexed by length up front, and each length keeps a cursor that only moves
forward, so a run is examined once no matter how many openers look past it. Searching
forward per opener instead was still superlinear, because an opener that nothing closes
rescans every run behind it: 400 runs of distinct lengths -- "`"*n + "x" for n in 1..400,
80KB -- took 2.0s, which is the same stall by another route.
Scanning runs also settles the case the pattern read backwards: in "`a``b`" the outer
single ticks pair, so the inner pair is content, not a delimiter.
"""
runs: list[tuple[int, int]] = [] # (start, length), left to right
by_length: dict[int, list[int]] = {} # length -> indices into runs
pos = 0
while True:
start = text.find("`", pos)
if start < 0:
break
pos = _tick_run_end(text, start)
by_length.setdefault(pos - start, []).append(len(runs))
runs.append((start, pos - start))
seek = dict.fromkeys(by_length, 0)
i = 0
while i < len(runs):
start, length = runs[i]
candidates = by_length[length]
# Only ever forward: i never goes back, so a candidate already behind it is behind
# every later opener too, and skipping it here is skipping it for good.
k = seek[length]
while k < len(candidates) and candidates[k] <= i:
k += 1
seek[length] = k
if k == len(candidates):
i += 1 # nothing of its length closes it; the tick sweep takes it
continue
close = runs[candidates[k]][0]
yield start, close + length, text[start + length : close]
i = candidates[k] + 1
def _strip_emphasis(text: str) -> str:
"""Drop matched emphasis pairs, leaving unpaired delimiter runs as literal text."""
# Content cannot cross its own delimiter, so a pass takes the innermost pair and
# emphasis nested in emphasis needs another; each pass drops two runs, so this
# settles.
while True:
stripped = text
for pattern in _EMPHASIS_RES:
stripped = pattern.sub(r"\2", stripped)
if stripped == text:
return text
text = stripped
def _strip_links(text: str) -> str:
"""Reduce [label](dest) and  to the label, at any nesting depth.
A destination may carry balanced parens ("/wiki/Foo_(bar)"), and stopping at the first
")" leaves the outer one to be spoken as punctuation after the label. Counting them is
the part a pattern cannot do: every nesting depth a pattern spells out has a next one,
and a link the pattern misses entirely is a whole URL read aloud -- a worse failure than
the stray punctuation the nesting was added to fix. An unbalanced destination takes the
rest of the text with it, which is the call the unterminated-fence rule already makes:
the reply was cut mid-link, and what follows an unclosed "(" is the URL.
"""
closes = _label_ends(text)
out: list[str] = []
pos = 0
while True:
head = _LINK_OPEN_RE.search(text, pos)
if head is None:
out.append(text[pos:])
return "".join(out)
label_end = closes.get(head.end() - 1, -1)
if label_end < 0 or not text.startswith("(", label_end + 1):
# A "[" that opens no link is ordinary text ("[1]", a stray bracket). Emit it
# and resume just past it: the next "[" on the line may still open a real link,
# and treating this one as a failed head would hide it.
out.append(text[pos : head.end()])
pos = head.end()
continue
out.append(text[pos : head.start()])
out.append(text[head.end() : label_end])
pos = _end_of_destination(text, label_end + 2)
def _label_ends(text: str) -> dict[int, int]:
"""Map each "[" to the "]" that closes it, for the whole text in one pass.
A label may hold brackets of its own -- "[the [docs]](url)" is one link whose label
markdown reads as "the [docs]". Reading a label as "everything up to the first ]" stops
at the inner bracket, finds no "(" after it, and does not see the link at all, so the
whole URL is spoken: a worse failure than the stray punctuation the destination counting
already exists to avoid.
Pairing with a stack, once, is what keeps that from costing a rescan per bracket.
Counting forward from each "[" instead re-walks the tail for every bracket that nothing
closes -- "["*8000 + "](url)" took 5.2s that way, the same stall the tick and container
rules each call out, and the reply is model output arriving at a loop that is always
listening.
A backslash escape reads differently on each bracket, because the two choices that miss
a link are not the same choice. CommonMark says "\\[" opens nothing, but every link this
pass fails to see is a destination spoken in full, so an escaped "[" is still a head
worth reading -- an escape is the reply's own markup, not a request to hear a URL. An
escaped "]" is skipped for the same reason from the other side: it does not close the
label, so "[a\\]b](url)" pairs at the real "]" and speaks its label instead of its URL.
Both rules err toward finding the link, which is the only direction that is quiet.
"""
closes: dict[int, int] = {}
opens: list[int] = []
i = 0
while i < len(text):
c = text[i]
if c == "\\" and i + 1 < len(text):
if text[i + 1] == "[":
opens.append(i + 1)
i += 2
continue
if c == "[":
opens.append(i)
elif c == "]" and opens:
closes[opens.pop()] = i
i += 1
return closes
def _end_of_destination(text: str, start: int) -> int:
"""Index just past the destination opened by the "(" before `start`.
Counting rather than matching is what a pattern cannot do, and a backslash escape is a
character in the destination rather than structure -- reading "\\(" as a nesting level runs
the count off the end of a link that was balanced all along.
"""
depth = 1
i = start
while i < len(text):
c = text[i]
if c == "\\" and i + 1 < len(text):
i += 2
continue
if c == "(":
depth += 1
elif c == ")":
depth -= 1
if depth == 0:
return i + 1
i += 1
# Never balanced, so this was not a whole link. A destination cannot hold unescaped
# whitespace, which bounds how much of the tail an unbalanced "(" may take: enough to
# swallow a URL the reply was cut off mid-way through, and no further. Consuming to the end
# instead deleted the rest of a reply whenever a "](" was never a link at all.
ws = _WHITESPACE_RE.search(text, start)
return ws.start() if ws else len(text)
def _strip_markdown(text: str) -> str:
"""Reduce the markdown the VOICE_SYSTEM_PROMPT forbids to plain spoken text."""
# Line endings first, so every line-anchored rule below can spell its line end "$" and
# mean it. A \r sits between a closing fence and its line end, which reads as "no closer"
# and hands the rest of the reply to the unterminated-fence rule.
text = text.replace("\r\n", "\n").replace("\r", "\n")
# A block at the margin is read BEFORE any marker comes off, and the containers below are
# read after, because each pass is wrong in the other's absence. Taking markers off first
# alone exposed a block's own interior: a body line like "> ```" is code, but with the
# quote marker gone it reads as this block's closer, so the real closer became an opener
# and the rest of the reply was deleted as an unterminated block -- while the code it was
# supposed to hide got spoken. Removing the whole margin block first means its interior is
# never on the page for the container pass to misread.
text = _FENCED_BLOCK_RE.sub(" ", text)
# Only quotes and list items can hold a block. A heading cannot, and stripping "#" here
# would promote its content to a line start, inventing a fence out of a heading that merely
# names one; it comes off further down, once fences have been decided. One pass: the
# pattern takes a line's whole marker run, so the inner marker of "> > ```" comes off with
# the outer one and re.sub never needs to rescan what it exposed. Whitespace is [ \t]
# rather than \s: \s matches a newline, so a bare marker could swallow its line ending and
# glue the next line onto a marker line -- promoting an indented backtick run to a fence
# that was never there.
text = _CONTAINER_RE.sub("", text)
text = _FENCED_BLOCK_RE.sub(" ", text) # blocks that sat behind a container
text = re.sub(
r"(?ms)^[ ]{0,3}(?:`{3,}|~{3,}).*", " ", text
) # unterminated trailing fence
text = re.sub(
r"(?m)^[ \t]{0,3}#{1,6}[ \t]+", "", text
) # heading markers, once fences are settled
# A code span's content is literal: `__init__` is the identifier, not bold "init", and
# `[label](url)` is the syntax being described, not a link to follow. Both rules
# therefore apply only BETWEEN the spans, and the ticks come off after them -- peeling
# them first would hand each rule content it has no way to tell from markup.
spoken: list[str] = []
pos = 0
for start, end, contents in _iter_code_spans(text):
spoken.append(_strip_emphasis(_strip_links(text[pos:start])))
spoken.append(contents)
pos = end
spoken.append(_strip_emphasis(_strip_links(text[pos:])))
return "".join(spoken).replace("`", "") # unpaired inline code ticks
def _truncate_spoken(text: str, max_chars: int) -> str:
"""Clip to max_chars on a sentence end when there is one, else a word break."""
clipped = text[:max_chars]
# The LAST sentence end of any kind, not the first kind that happens to match:
# checking ". " before "! " would cut at an earlier period and drop everything
# up to a later exclamation, throwing away speakable text we had room for.
idx = max(clipped.rfind(end) for end in (". ", "! ", "? "))
if idx >= max_chars // 2:
return clipped[: idx + 1].rstrip()
space = clipped.rfind(" ")
return (clipped[:space] if space > 0 else clipped).rstrip()
def sanitize_reply(
reply: str,
max_chars: int = MAX_SPOKEN_CHARS,
empty_fallback: str = EMPTY_REPLY_FALLBACK,
) -> str:
"""Make a brain reply safe to hand to speak().
Bounds length, drops markdown/emoji/control characters, and turns an empty
reply into a spoken sentence so a silent failure is audible rather than a
turn that just does nothing.
"""
if not reply or not reply.strip():
return empty_fallback
text = _ANSI_ESCAPE_RE.sub("", reply)
text = _strip_markdown(text)
text = _EMOJI_RE.sub("", text)
text = _CONTROL_CHARS_RE.sub("", text)
text = " ".join(text.split())
if not text:
return empty_fallback
if len(text) > max_chars:
text = _truncate_spoken(text, max_chars)
return text
def brain(text: str, model: str | None = None, timeout_s: int | None = None) -> str:
"""Answer transcribed text using the configured brain backend.
brain_backend == "bridge" routes to a persistent assistant session (so voice
and the user's text assistant are one session with shared memory); anything
else uses the local `claude` CLI fallback. Either way the reply is short
spoken text, and no network LLM API is called. Never raises for an expected
failure — the caller is a voice loop, so it returns a spoken error instead.
Both backends return through sanitize_reply(), so this is the single place a
reply becomes speakable and no raw model output can reach speak().
"""
cfg = load_config()
if cfg["brain_backend"] == "bridge":
reply = _brain_bridge(text, cfg)
else:
reply = _brain_cli(text, cfg, model=model, timeout_s=timeout_s)
spoken = sanitize_reply(reply, empty_fallback="")
if isinstance(reply, brain_bridge.BrainFailure):
return brain_bridge.BrainFailure(spoken or EMPTY_REPLY_FALLBACK, reply.reason)
if not spoken:
return brain_bridge.BrainFailure(EMPTY_REPLY_FALLBACK, "empty_reply")
return spoken
# One reply cursor per reply file, kept for the life of the process so the
# positional correlation survives across turns (a fresh cursor each turn would
# re-snapshot and lose the slot reservation that keeps timeouts aligned). A loop
# relaunch starts empty, so the first turn drains whatever backlog is on disk.
_bridge_cursors: dict[str, brain_bridge.ReplyCursor] = {}
def _brain_bridge(text: str, cfg: dict) -> str:
"""Route the transcript to a persistent assistant session over the bridge.
Transport is built from config by brain_bridge.build_brain(). A missing or
unknown setting degrades to a spoken error, never a crash.
"""
reply_path = cfg.get("brain_reply_path") or ""
cursor = (
_bridge_cursors.setdefault(reply_path, brain_bridge.ReplyCursor())
if reply_path
else None
)
return brain_bridge.build_brain(
cfg,
system_prompt=_bridge_voice_system_prompt(cfg.get("brain_persona")),
cursor=cursor,
)(text)
def _brain_cli(
text: str, cfg: dict, model: str | None = None, timeout_s: int | None = None
) -> str:
"""Dev fallback: send the transcript to the local `claude` CLI.
Uses the host Claude Code subscription via subprocess. Tools are disabled so
untrusted spoken input cannot drive local actions; session persistence is
off so audio transcripts are not written to disk. A CLI call, not a network
API request, and stateless — no shared memory with the text assistant.
"""
model = model or cfg["claude_model"]
timeout_s = cfg["claude_timeout_s"] if timeout_s is None else timeout_s
prompt = f"{VOICE_SYSTEM_PROMPT}\n\nUser: {text}\nAssistant:"
try:
result = subprocess.run(
[
CLAUDE_BIN,
"-p",
"--model",
model,
"--tools",
"",
"--no-session-persistence",
],
input=prompt,
capture_output=True,
text=True,
timeout=timeout_s,
cwd=tempfile.gettempdir(),
)
except subprocess.TimeoutExpired:
return brain_bridge.BrainFailure(
"Sorry, I timed out thinking about that.", "cli_timeout"
)
except (FileNotFoundError, OSError):
# The claude binary is not installed / not on PATH. Speak, do not crash.
return brain_bridge.BrainFailure(
"Sorry, the brain is not available right now.", "cli_unavailable"
)
if result.returncode != 0:
err = (result.stderr or "").strip().splitlines()
tail = err[-1] if err else f"exit {result.returncode}"
return brain_bridge.BrainFailure(
f"Sorry, the brain call failed: {tail}", "cli_exit"
)
return result.stdout.strip()
# --------------------------------------------------------------------------- #
# Stage 4: text-to-speech
# --------------------------------------------------------------------------- #
def _get_piper(model_path: str):
"""Load a Piper voice once and reuse it across speak() calls in one process.
Mirrors _get_oww_model/_get_whisper: the ONNX voice model is the largest
per-call cost, so loading it on the first speak and caching it turns every
later reply into synth-only work instead of a full model reload. Keyed by the
ONNX path so distinct voices cache independently. The sibling <model>.onnx.json
config is passed when present; otherwise Piper derives it.
"""
if model_path not in _piper_cache:
from piper import PiperVoice
config_path = model_path + ".json"
_piper_cache[model_path] = PiperVoice.load(
model_path,
config_path=config_path if os.path.exists(config_path) else None,
)
return _piper_cache[model_path]
def _speak_subprocess(text: str, onnx: Path, out_wav_path: str) -> None:
"""Fallback TTS: shell out to the piper CLI, reloading the voice each call."""
subprocess.run(
[sys.executable, "-m", "piper", "-m", str(onnx), "-f", out_wav_path],
input=text,
capture_output=True,
text=True,
check=True,
)
def speak(text: str, out_wav_path: str, voice_model: str | None = None) -> str:
"""Render text to a WAV with Piper. Returns the output path.
The voice loads once and is reused across calls (see _get_piper), so only the
first reply in a process pays the model-load cost. If the in-process load or
synth fails for any reason, fall back to the piper CLI so a bad install still
speaks rather than crashing the loop.
"""
cfg = load_config()
voice_model = voice_model or cfg["voice_model"]
onnx = VOICES_DIR / f"{voice_model}.onnx"
if not onnx.exists():
raise FileNotFoundError(
f"voice model not found: {onnx}. download with "
f"`python -m piper.download_voices {voice_model} --download-dir voices`"
)
out_wav_path = str(out_wav_path)
try:
voice = _get_piper(str(onnx))
with wave.open(out_wav_path, "wb") as wav_file:
voice.synthesize_wav(text, wav_file)
except Exception as exc:
# In-process load or synth failed (bad piper build, unexpected API change).
# Evict unconditionally: a voice that failed to load must not stick, and
# re-loading a good voice that merely hit a synth error next call is cheap,
# so either way the next call starts clean. Warn so the regression is
# visible, then fall back to the CLI so the reply still speaks. If the CLI
# also fails it raises, surfacing the real problem instead of going silent.
_piper_cache.pop(str(onnx), None)
warnings.warn(
f"piper in-process synth failed ({exc}); using CLI fallback",
RuntimeWarning,
stacklevel=2,
)
_speak_subprocess(text, onnx, out_wav_path)
return out_wav_path
# --------------------------------------------------------------------------- #
# End-to-end chain
# --------------------------------------------------------------------------- #
def run_pipeline(
wav_path: str, out_wav_path: str | None = None, wake_word: str | None = None
) -> dict:
"""Full chain on one input WAV. Returns a dict of stage results + timings."""
timings: dict[str, float] = {}
t0 = time.time()
fired, wname, score = detect_wake(wav_path, model_name=wake_word)
timings["detect_wake"] = time.time() - t0
result = {
"input": wav_path,
"wake_word": wname,
"wake_fired": fired,
"wake_score": round(score, 4),
"transcript": None,
"transcript_avg_logprob": None,
"transcript_no_speech_prob": None,
"reply": None,
"output_wav": None,
"timings_s": timings,
}
if not fired:
timings["total"] = time.time() - t0
return result
# The file path is an inspection tool, so it surfaces the confidence signals
# but does not gate on them: the live turn path (run_turn) is where the mishear
# guard stops a low-confidence command from reaching the brain.
t1 = time.time()
heard = transcribe_detailed(wav_path)
timings["transcribe"] = time.time() - t1
result["transcript"] = heard.text
result["transcript_avg_logprob"] = round(heard.avg_logprob, 3)
result["transcript_no_speech_prob"] = round(heard.no_speech_prob, 3)
t2 = time.time()
reply = brain(heard.text)
timings["brain"] = time.time() - t2
result["reply"] = reply
if isinstance(reply, brain_bridge.BrainFailure):
result["rejected"] = "brain_failure"
result["reject_reason"] = reply.reason
if out_wav_path is None:
out_wav_path = str(PROJECT_DIR / "test_audio" / "reply.wav")
t3 = time.time()
speak(reply, out_wav_path)
timings["speak"] = time.time() - t3
result["output_wav"] = out_wav_path
timings["total"] = time.time() - t0
return result
def run_turn(
frames,
model_name: str | None = None,
threshold: float | None = None,
out_wav_path: str | None = None,
on_capture=None,
on_wake=None,
) -> dict | None:
"""One live turn over a frame stream: wait for the wake word, capture the
request to its endpoint, then transcribe -> brain -> speak.
`frames` is any iterator of 80 ms int16 frames (iter_wav_frames for tests, a
mic source on hardware). Returns a result dict, or None when there is nothing
to act on: the stream ended before the wake word fired, the wake fired but no
speech followed it, or the captured audio transcribed to no words (noise). When
the mishear guard rejects a low-confidence transcript the brain is skipped, but
a result dict is still returned (reply set to a spoken re-prompt, "rejected" set
to "low_confidence") so the loop speaks the re-prompt. This is the unit an
always-on loop calls repeatedly; the loop wrapper and mic/speaker I/O land with
the microphone (issues #10, #11).
`on_wake`, if given, plays an acknowledgment chime to signal the mic is now
listening (issue #41). It is gated on a pause (issue #55): after a real wake,
peek_cue_gate reads the post-wake audio, and on_wake is called only when that window
is silent (the user paused, waiting for the cue). When a command is already being
spoken (no pause), the cue is skipped -- it would clip the command on a half-duplex
device -- and the peeked command frames are fed to capture ahead of the rest of the
stream, so nothing is lost. on_wake never fires when the stream ends with no
detection. Being half-duplex, the loop's hook keeps the cue out of the captured
request (e.g. by pausing the mic and flushing the cue frames). It returns True if it
established a fresh post-cue capture boundary (cue played, buffered audio flushed);
run_turn then drops the detection pre-roll too, so the pre-cue wake-word tail does
not prepend a command spoken after the cue. If the cue failed (returns falsy,
buffered audio kept), the pre-roll is kept so a no-pause command is still recovered
(issue #30) rather than clipped.
`on_capture`, if given, is called once the request has been captured (listening
for this turn is over). A half-duplex live loop uses it to stop the mic before
the slow transcribe/brain/speak stages, so they do not buffer input the loop
would only discard. It fires whether or not speech was found, so the caller can
pair every call with a resume.
"""
turn_started = time.monotonic()
cfg = load_config()
model_name = model_name or cfg["wake_word"]
threshold = cfg["wake_threshold"] if threshold is None else threshold
model = _get_oww_model(_resolve_wake_path(model_name))
frames = iter(frames)
# Keep the most recent frames during detection so the request's leading audio,
# consumed while the detector was crossing threshold, is recovered (issue #30).
wake_audio = WakeAudioBuffer()
stage_started = time.monotonic()
score = stream_detect_wake(frames, model, threshold, preroll=wake_audio)
if score is None:
return None
timings = {"detect_wake": time.monotonic() - stage_started}
WAKE_LOGGER.info("Turn started wake_word=%s", model_name)
WAKE_LOGGER.debug("Wake score=%.4f", score)
WAKE_LOGGER.info(
"Stage finished stage=detect_wake duration_s=%.3f", timings["detect_wake"]
)
# The wake fired. Acknowledge it now -- after detection, before capture -- so a
# live loop can chime "listening" (issue #41). on_wake runs only on a real wake,
# never on an ended stream, so silence never triggers the cue.
capture_frames = frames
if on_wake is not None:
# Gate the cue on a pause (issue #55). Peek the post-wake audio first: if a
# command is already being spoken (no pause), skip the cue so it cannot clip
# the command on this half-duplex device, and feed the peeked command frames
# back to capture ahead of the rest of the stream -- the pre-roll stays, so the
# leading audio detection ate is still recovered (issue #30). If the window is
# silent (the user waited for the cue), play it.
play_cue, peeked = peek_cue_gate(
frames, vad_threshold=cfg["capture_vad_threshold"]
)
if play_cue and on_wake():
# The window was silent (a pause) AND on_wake established a fresh post-cue
# capture boundary (cue played, the audio buffered up to now flushed). The
# peeked frames are pre-cue room tone, dropped with the cue bleed the hook
# flushes. Drop the detection pre-roll too, matching that flush: it exists
# only to recover a no-pause command's leading audio that detection latency
# ate (issue #30), and a user who waits for the cue speaks fresh after it, so
# keeping it would inject the stale pre-cue wake-word tail into the post-cue
# command (issue #41).
wake_audio.clear()
else:
# Either a no-pause command (play_cue False -- on_wake was not called, the
# cue is skipped so it cannot clip the command), or the cue failed to
# establish a boundary (on_wake returned falsy -- no flush, buffered audio
# kept). Both keep the pre-roll and feed the peeked frames back to capture
# ahead of the rest of the stream, so no leading audio is clipped
# (issues #30, #55).
capture_frames = itertools.chain(peeked, frames)
stage_started = time.monotonic()
request_pcm = capture_request(
capture_frames,
preroll=list(wake_audio.preroll),
vad_threshold=cfg["capture_vad_threshold"],
endpoint_silence_ms=cfg["endpoint_silence_ms"],
max_request_ms=cfg["max_request_ms"],
)
timings["capture"] = time.monotonic() - stage_started
WAKE_LOGGER.info("Stage finished stage=capture duration_s=%.3f", timings["capture"])
# Listening for this turn is done. Let a live loop stop the mic now, before the
# slow stages below, so it does not buffer (then throw away) input recorded
# while the assistant is thinking and speaking.
if on_capture is not None:
on_capture()
if request_pcm.size == 0:
empty_reason = getattr(request_pcm, "empty_reason", None)
if empty_reason != _EMPTY_NO_ONSET:
WAKE_LOGGER.info(
"Turn ignored reason=%s total_s=%.3f",
empty_reason or "empty_capture",
time.monotonic() - turn_started,
)
return None # VAD rejection or an unknown empty result stays fail-closed
stage_started = time.monotonic()
heard = recover_consumed_command(wake_audio.history, model_name)
if heard is None:
timings["transcribe"] = time.monotonic() - stage_started
STT_LOGGER.info(
"Stage finished stage=transcribe duration_s=%.3f",
timings["transcribe"],
)
WAKE_LOGGER.info(
"Turn ignored reason=no_command total_s=%.3f",
time.monotonic() - turn_started,
)
return None # bare/false wake, or no recoverable command — ignore the turn
else:
# faster-whisper accepts a normalized waveform, so transcribe captured PCM in
# memory instead of serializing each live request through a temporary WAV.
stage_started = time.monotonic()
heard = transcribe_detailed(request_pcm)
timings["transcribe"] = time.monotonic() - stage_started
STT_LOGGER.info(
"Stage finished stage=transcribe duration_s=%.3f", timings["transcribe"]
)
STT_LOGGER.debug("Transcript text=%r", heard.text)
if not heard.text.strip():
WAKE_LOGGER.info(
"Turn ignored reason=empty_transcript total_s=%.3f",
time.monotonic() - turn_started,
)
return None # captured audio whisper read as no words (noise) — ignore
if out_wav_path is None:
out_wav_path = str(PROJECT_DIR / "test_audio" / "reply.wav")
# Mishear guard: the brain acts on the command (files an issue, changes
# something), so a low-confidence transcript must never reach it. On a reject,
# speak a short re-prompt and return the turn marked rejected -- the loop still
# plays the re-prompt, since spoken feedback is the only channel, but the brain
# is never called, so a garbled word cannot trigger an action.
ok, reason = guard_transcript(heard, cfg)
if not ok:
STT_LOGGER.info("Transcript rejected reason=%s", reason)
stage_started = time.monotonic()
speak(STT_REPROMPT, out_wav_path)
timings["speak"] = time.monotonic() - stage_started
timings["total"] = time.monotonic() - turn_started
TTS_LOGGER.info("Stage finished stage=speak duration_s=%.3f", timings["speak"])
WAKE_LOGGER.info(
"Turn finished rejected=low_confidence total_s=%.3f", timings["total"]
)
return {
"wake_word": model_name,
"wake_score": round(score, 4),
"transcript": heard.text,
"reply": STT_REPROMPT,
"output_wav": out_wav_path,
"rejected": "low_confidence",
"reject_reason": reason,
"timings_s": {key: round(value, 3) for key, value in timings.items()},
}
stage_started = time.monotonic()
reply = brain(heard.text)
timings["brain"] = time.monotonic() - stage_started
BRAIN_LOGGER.info("Stage finished stage=brain duration_s=%.3f", timings["brain"])
BRAIN_LOGGER.debug("Reply text=%r", reply)
stage_started = time.monotonic()
speak(reply, out_wav_path)
timings["speak"] = time.monotonic() - stage_started
timings["total"] = time.monotonic() - turn_started
TTS_LOGGER.info("Stage finished stage=speak duration_s=%.3f", timings["speak"])
WAKE_LOGGER.info("Turn finished total_s=%.3f", timings["total"])
return {
"wake_word": model_name,
"wake_score": round(score, 4),
"transcript": heard.text,
"reply": reply,
"output_wav": out_wav_path,
"timings_s": {key: round(value, 3) for key, value in timings.items()},
}
def warm_models(
cfg: dict | None = None, wake_word: str | None = None
) -> dict[str, float]:
"""Load every model the first live turn would otherwise load lazily, up front.
A live turn touches four models, each loaded lazily on first use: the wake model
(_get_oww_model), the capture-time silero VAD (_get_vad, run over the post-wake
audio in capture_request), whisper (_get_whisper), and the Piper voice (_get_piper).
In a fresh process the first user turn pays all four loads while someone waits on the
reply, and only later turns are fast. The live loop calls this at startup, before it
listens, to move that one-time cost to a point where no one is waiting. It touches
only the models the live loop uses -- the wake word from config, the bundled VAD, the
configured whisper model and compute, and the configured Piper voice -- and the file
pipeline and the tests never call it, so they still load lazily and pay only for the
stages they run.
Warming is a cache fill, not a separate code path: the loaders populate the same
module caches the stages read, so the first real turn reuses these instances. A
stage that fails to warm (a missing voice file, say) is logged and skipped rather
than stopping the loop before it starts or blocking the other stages from warming;
the lazy path raises the same error later if that stage is actually used.
Returns the warm time in seconds per stage ("wake", "vad", "whisper", "piper"); a
stage that failed to warm is absent from the mapping.
"""
cfg = cfg or load_config()
wake_word = wake_word or cfg["wake_word"]
voice_onnx = VOICES_DIR / f"{cfg['voice_model']}.onnx"
# In first-turn order: wake fires, the VAD confirms the captured command, whisper
# transcribes it, and Piper speaks the reply.
loaders: list[tuple[str, logging.Logger, Callable[[], object]]] = [
("wake", WAKE_LOGGER, lambda: _get_oww_model(_resolve_wake_path(wake_word))),
("vad", WAKE_LOGGER, lambda: _get_vad()),
(
"whisper",
STT_LOGGER,
lambda: _get_whisper(
cfg["whisper_model"],
cfg["whisper_compute"],
cfg["whisper_cpu_threads"],
),
),
("piper", TTS_LOGGER, lambda: _get_piper(str(voice_onnx))),
]
warmed: dict[str, float] = {}
for label, logger, load in loaders:
t0 = time.time()
try:
load()
except Exception as exc: # noqa: BLE001 - one bad model must not stop the loop or the others
if label == "piper" and not voice_onnx.exists():
logger.warning(
"Model warm failed stage=%s reason=voice_model_missing path=%s "
"action='python -m piper.download_voices %s --download-dir voices'",
label,
voice_onnx,
cfg["voice_model"],
)
else:
logger.warning(
"Model warm failed stage=%s error=%s detail=%s",
label,
type(exc).__name__,
exc,
)
continue
warmed[label] = time.time() - t0
logger.info("Model warmed stage=%s duration_s=%.3f", label, warmed[label])
return warmed
def run_loop(
wake_word: str | None = None,
mic_name=None,
output_name=None,
threshold: float | None = None,
) -> None:
"""Always-on voice loop: listen, wake, capture, answer, speak -- repeat.
Drives run_turn over a live microphone (audio.Microphone) and plays each reply
through the configured output device. Half-duplex: run_turn's on_capture hook
pauses the mic the instant the request is captured, so the slow transcribe /
brain / speak stages and the reply playback never run while the mic is live.
After each turn the mic is resumed and flushed, dropping anything that leaked in
so the assistant never transcribes its own voice -- raw devices have no echo
cancellation, so the loop avoids capturing and playing at once rather than
relying on the hardware to cancel the speaker.
Device names come from config (mic_device / output_device) unless overridden.
`audio` is imported lazily so file-mode use and the tests never require the
PortAudio backend.
"""
import audio # lazy: only the live loop needs the PortAudio backend
import chime # the wake-acknowledgment cue (pure DSP; no audio backend)
cfg = load_config()
configure_logging(cfg.get("log_level", DEFAULTS["log_level"]))
wake_word = wake_word or cfg["wake_word"]
if mic_name is None:
mic_name = cfg.get("mic_device") or None
if output_name is None:
output_name = cfg.get("output_device") or None
out_wav = str(PROJECT_DIR / "test_audio" / "reply.wav")
Path(out_wav).parent.mkdir(parents=True, exist_ok=True)
# Render the acknowledgment cue once when enabled (issue #41); None disables it.
# Opt-in, default off (see DEFAULTS): the cue regresses the no-pause case on a
# half-duplex device, so it is off unless config turns it on. Rendering is a
# nicety -- if it fails (filesystem / soundfile), degrade to no chime rather than
# taking down the whole live loop before it starts listening.
cue_wav = None
if cfg.get("wake_chime", False):
try:
cue_wav = chime.wake_cue_wav()
except Exception as e: # noqa: BLE001 - the chime is optional, never fatal
WAKE_LOGGER.warning(
"Wake chime unavailable error=%s detail=%s", type(e).__name__, e
)
# Warm the models before listening so the first turn is as fast as the rest
# (issue #13). Each loader caches, so the first real turn reuses these instances.
t0 = time.time()
warmed = warm_models(cfg, wake_word)
elapsed = time.time() - t0
WAKE_LOGGER.info("Models warmed count=%d duration_s=%.3f", len(warmed), elapsed)
# Keep these two startup notices on stdout for existing operators and callers;
# the structured records alongside them feed the persistent service log.
print(f"computah listening -- wake word: {wake_word}. Ctrl-C to stop.")
WAKE_LOGGER.info("Listening wake_word=%s", wake_word)
with audio.Microphone(mic_name) as mic:
WAKE_LOGGER.info("Microphone opened device=%r", mic.device_label)
# Say it once, here, while someone is still watching the terminal: a bad
# capture device does not announce itself later, since only the transcript
# is wrong (issue #34). Warn and continue -- it still serves wake fine.
if mic.capture_risk is not None:
print(f" WARNING: {mic.capture_risk.message}")
WAKE_LOGGER.warning("Capture risk detail=%s", mic.capture_risk.message)
frames = mic.frames()
paused = False
def pause_for_turn():
nonlocal paused
mic.pause()
paused = True
def chime_for_turn():
nonlocal cue_wav
# The wake fired: pause the mic, play the cue, then flush so the cue --
# which bleeds straight back in on a shared mic/speaker device like the
# PowerConf -- is not captured as part of the request, then resume so
# capture starts from clean audio. A failed cue must never kill the loop.
# A failure does not prove whether playback emitted part of the cue; skip
# the flush because no clean post-cue boundary was established, and
# preserve all buffered post-wake audio that still exists for capture
# (matches live_driver).
# Return True only when the cue played and the buffer was flushed, so
# run_turn drops the detection pre-roll exactly when this boundary was
# established, and keeps it (no-pause recovery, issue #30) when the cue
# failed and the buffered audio was left intact.
mic.pause()
flushed = False
try:
audio.play_wav(cue_wav, output_name)
except Exception as e: # noqa: BLE001 - the chime is a nicety, not core
# A slow playback failure loses any speech made while the mic is
# stopped (#58). The current turn cannot recover those samples, but
# retrying the same broken output on every wake would repeat the loss.
# Disable it until restart so later turns use no-chime semantics.
cue_wav = None
WAKE_LOGGER.warning(
"Wake chime failed error=%s detail=%s action='disabled until restart'",
type(e).__name__,
e,
)
else:
mic.flush()
flushed = True
mic.resume()
return flushed
try:
while True:
paused = False
result = run_turn(
frames,
model_name=wake_word,
threshold=threshold,
out_wav_path=out_wav,
on_capture=pause_for_turn,
on_wake=chime_for_turn if cue_wav else None,
)
if result is not None:
# A dead or busy output device on the reply must degrade, never crash
# the always-on loop -- the same guarantee the wake chime above and
# live_driver.run_turn (issue #11) already hold. The reply is saved at
# out_wav, so on failure log it and keep listening rather than propagating.
try:
audio.play_wav(out_wav, output_name)
except Exception as e: # noqa: BLE001 - degrade to the saved WAV, never crash
TTS_LOGGER.warning(
"Reply playback failed error=%s detail=%s reply_wav=%s",
type(e).__name__,
e,
out_wav,
)
if paused:
# A turn was captured (mic stopped before the slow stages).
# Flush while the stream is still stopped: no fresh audio is
# arriving yet, so this drops only the stale audio buffered
# during thinking/speaking. Resume after, so the next utterance
# is captured from a clean buffer even if the user starts
# speaking the instant playback ends. Flushing after resume()
# would race the callback and discard the start of that speech.
mic.flush()
mic.resume()
elif not mic.active():
# run_turn returned without capturing a request and the stream
# is no longer delivering frames: the device ended. Stop.
WAKE_LOGGER.info("Microphone stream ended")
break
except KeyboardInterrupt:
WAKE_LOGGER.info("Stopped by operator")
def _cli() -> int:
p = argparse.ArgumentParser(
description="computah: process a WAV, or run the live mic loop with --listen"
)
p.add_argument("input_wav", nargs="?", help="input WAV to process")
p.add_argument("-o", "--output", help="output WAV path for the spoken reply")
p.add_argument("-w", "--wake-word", help="override the active wake word")
p.add_argument(
"--list-wake-words",
action="store_true",
help="list installed wake-word models and exit",
)
p.add_argument(
"--set-wake-word",
metavar="NAME",
help="persist a new active wake word and exit (config.json by default, "
"or config.local.json with --local)",
)
p.add_argument(
"--local",
action="store_true",
help="with --set-wake-word, write the override to the gitignored "
"config.local.json instead of the committed config.json",
)
p.add_argument(
"--listen",
action="store_true",
help="run the always-on live mic loop (needs sounddevice)",
)
p.add_argument(
"--mic",
metavar="NAME",
help="mic device name substring (overrides config mic_device)",
)
p.add_argument(
"--speaker",
metavar="NAME",
help="output device name substring (overrides config output_device)",
)
args = p.parse_args()
if args.local and not args.set_wake_word:
p.error("--local only applies to --set-wake-word")
if args.list_wake_words:
cfg = load_config()
for name in sorted(available_wake_models()):
mark = " (active)" if name == cfg["wake_word"] else ""
print(f"{name}{mark}")
return 0
if args.set_wake_word:
cfg = set_wake_word(args.set_wake_word, local=args.local)
target = "config.local.json" if args.local else "config.json"
print(f"wake word saved to {target}: {args.set_wake_word}")
if cfg["wake_word"] != args.set_wake_word:
print(
f"note: config.local.json still overrides the active wake word "
f"to {cfg['wake_word']!r}; pass --local to change that instead"
)
return 0
if args.listen:
run_loop(wake_word=args.wake_word, mic_name=args.mic, output_name=args.speaker)
return 0
if not args.input_wav:
p.error(
"input_wav is required unless using "
"--listen/--list-wake-words/--set-wake-word"
)
r = run_pipeline(args.input_wav, out_wav_path=args.output, wake_word=args.wake_word)
print(f"wake word : {r['wake_word']}")
print(f"wake fired : {r['wake_fired']} (score {r['wake_score']})")
if not r["wake_fired"]:
print("not woken — pipeline stopped (no transcription/brain/tts)")
else:
print(f"transcript : {r['transcript']}")
print(f"reply : {r['reply']}")
print(f"output wav : {r['output_wav']}")
print(
"timings (s) : " + ", ".join(f"{k}={v:.2f}" for k, v in r["timings_s"].items())
)
return 0
if __name__ == "__main__":
raise SystemExit(_cli())