live_driver.py
Rama main · réplica de lectura del repositorio original.
#!/usr/bin/env python3
"""Always-on live voice loop for computah: real PowerConf mic -> wake -> STT ->
brain (bridge -> Syl) -> spoken reply, looping turn after turn.
Reads raw S16_LE 16 kHz mono frames from stdin (piped from `arecord -t raw`), so
the mic driver stays out of this process. Runs continuously: after each turn it
resets the wake model and resumes listening, so repeated "computah ..." turns work
without relaunching the process.
sudo arecord -D plughw:CARD=PowerConf,DEV=0 -f S16_LE -r 16000 -c 1 -t raw \
| .venv/bin/python live_driver.py --debug
-v/--debug sets the live logger to DEBUG and adds per-frame rms+score telemetry
to stderr (use it to see why a wake did or did not fire).
This is the real-hardware counterpart to the mic-free streaming proof in
test_stream_turn.py, which drives pipeline.stream_detect_wake and capture_request
with synthetic audio instead of a live mic. Keep them distinct: that one
proves the streaming/endpointing logic with no hardware; this one drives the actual
device against the real persistent brain.
"""
from __future__ import annotations
import argparse
import itertools
import os
import signal
import subprocess
import sys
import tempfile
import time
from collections.abc import Callable
import numpy as np
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import pipeline
import chime
from playback_abort import wait_or_abort
_DRAIN_FRAMES = 25 # ~2 s discarded after a turn: clears stale/echo audio from the
# pipe buffer so the spoken reply is not re-heard as the next wake
class _AplayFailed(subprocess.CalledProcessError):
"""An aplay child exited unsuccessfully, so the sudo retry is applicable."""
def _resolve_output_pcm(cli_output_device: str | None, cfg: dict) -> str | None:
"""Pick the ALSA PCM live_driver plays through, `-o` winning over config (issue #50).
Precedence: an explicit `-o/--output-device` first, else the config `live_output_pcm`,
else None (the ALSA default device). Only `live_output_pcm` is read here, never the
sounddevice `output_device`: that is a friendly-substring for the audio.py path and
would fail as an `aplay -D` argument, so the two naming conventions stay separate.
An empty-string value is treated as unset, matching the DEFAULTS sentinel.
`-o` present (including an explicit `-o ''`) always wins, so `None` (option omitted)
and `""` (explicit override to the ALSA default) stay distinct: a live_output_pcm in
config cannot override an `-o ''` on the command line."""
if cli_output_device is not None:
return cli_output_device or None
return cfg.get("live_output_pcm") or None
def _signal_playback_group(
process: subprocess.Popen, sig: signal.Signals, elevated: bool
) -> None:
"""Signal only the playback process group, including a root-owned fallback."""
if elevated:
result = subprocess.run(
["sudo", "-n", "kill", f"-{sig.name}", "--", f"-{process.pid}"],
check=False,
stdin=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)
# The process can finish after wait_or_abort's poll and before this kill.
# Its missing process group is then a successful cleanup, not a playback
# failure. A live process plus a failed privileged kill remains an error.
if result.returncode and process.poll() is None:
raise subprocess.CalledProcessError(result.returncode, result.args)
return
try:
os.killpg(process.pid, sig)
except ProcessLookupError:
pass
def _stop_playback_group(process: subprocess.Popen, elevated: bool) -> None:
"""Stop an aplay process group without leaving a child behind."""
_signal_playback_group(process, signal.SIGTERM, elevated)
try:
process.wait(timeout=1)
except subprocess.TimeoutExpired:
_signal_playback_group(process, signal.SIGKILL, elevated)
process.wait(timeout=1)
def _play_once(
cmd: list[str],
*,
should_stop: Callable[[], bool] | None,
poll_ms: int,
elevated: bool,
) -> bool:
"""Run one aplay attempt. Return False only when playback was interrupted."""
if should_stop is None:
result = subprocess.run(
cmd, check=False, stdin=subprocess.DEVNULL, stderr=subprocess.DEVNULL
)
if result.returncode:
raise _AplayFailed(result.returncode, cmd)
return True
process = subprocess.Popen(
cmd,
stdin=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
start_new_session=True,
)
try:
finished = wait_or_abort(
is_active=lambda: process.poll() is None,
should_stop=should_stop,
stop=lambda: _stop_playback_group(process, elevated),
sleep=lambda milliseconds: time.sleep(milliseconds / 1000),
poll_ms=poll_ms,
)
except BaseException:
# A detector error or Ctrl-C must not leave aplay speaking after the live
# loop has moved on or exited. BaseException is intentional for Ctrl-C.
if process.poll() is None:
_stop_playback_group(process, elevated)
raise
if not finished:
return False
return_code = process.wait()
if return_code:
raise _AplayFailed(return_code, cmd)
return True
def _play_wav(
path: str,
device: str | None,
should_stop: Callable[[], bool] | None = None,
poll_ms: int = 50,
) -> bool:
"""Play a WAV through ALSA's aplay — the Pi's native output, symmetric with the
arecord capture side and dependency-free. (The sounddevice-based audio.play_wav
is the cross-platform/dev path and is not installed in this venv.)
Tries unelevated first, then retries under sudo: on this Pi /dev/snd needs root in
a non-login launch context. Only the playback subprocess is elevated, never the
whole driver — the brain stage shells `ssh officejawn`, which must run as the
launching user so it uses that user's ssh config and keys, not root's.
Without `should_stop`, playback keeps its blocking behavior. With a callback,
aplay runs in its own process group and returns False after an interruption.
The root fallback is stopped through the same non-interactive sudo boundary
used to start it, so the unelevated driver never tries to signal a root process
directly. The live wake detector that supplies this callback is tracked in
issue #100."""
cmd = ["aplay", "-q"]
if device:
cmd += ["-D", device]
cmd.append(path)
try:
return _play_once(
cmd,
should_stop=should_stop,
poll_ms=poll_ms,
elevated=False,
)
except _AplayFailed:
# Retry once under non-interactive sudo: /dev/snd needs root in some launch
# contexts. -n so a missing or expired sudo timestamp fails fast instead of
# prompting, and stdin from /dev/null so a prompt can never consume the raw
# mic pipe that is this process's stdin. Playback then degrades to the logged
# error in run_turn rather than wedging the loop.
return _play_once(
["sudo", "-n", *cmd],
should_stop=should_stop,
poll_ms=poll_ms,
elevated=True,
)
class StdinMic:
"""The arecord stdin pipe as a flushable stream of 80 ms / 16 kHz / mono int16
frames — live_driver's counterpart to audio.Microphone, with the same
frames()/flush() contract on the raw-pipe path.
Reads the raw fd directly (os.read), not sys.stdin.buffer, so there is a single
place to flush: the OS pipe, plus a sub-frame byte remainder kept on the instance
(self._buf) so flush() can drop it too. Reading through a BufferedReader would add a
second, Python-level buffer that an fd-level flush cannot reach, leaving stale bytes
to prefix the next frame. flush() is the flush-to-now the post-cue drain needs
(issue #56); it matches run_loop's mic.flush() semantics.
"""
def __init__(self, fd: int, frame_size: int = pipeline.FRAME_SIZE):
self.fd = fd
self.frame_bytes = frame_size * 2 # int16: 1280 samples = 2560 bytes (80 ms)
self._buf = b"" # sub-frame remainder, on the instance so flush() reaches it
def frames(self):
"""Yield writable 80 ms int16 frames from the raw pipe. Ends (returns) on EOF,
i.e. when arecord exits. A short os.read is accumulated toward a full frame
rather than ending the stream, so a mid-stream partial read is not mistaken
for EOF (BufferedReader hid these; the raw fd surfaces them).
.copy() because np.frombuffer aliases a read-only buffer; the wake model's
predict path and capture_request both expect a writable array.
"""
while True:
chunk = os.read(self.fd, self.frame_bytes - len(self._buf))
if not chunk: # EOF: arecord exited
return
self._buf += chunk
if len(self._buf) >= self.frame_bytes:
frame = np.frombuffer(
self._buf[: self.frame_bytes], dtype=np.int16
).copy()
self._buf = self._buf[self.frame_bytes :]
yield frame
def flush(self) -> None:
"""Flush-to-now: drop everything buffered up to this instant — the sub-frame
remainder and the whole OS pipe backlog — without blocking, so the next frame
is built only from audio captured after the flush. Mirrors
audio.Microphone.flush.
Sets the fd non-blocking and reads until EAGAIN (no data buffered) or EOF, then
restores blocking mode. Safe to call only while frames() is suspended (between
detection and capture, or between turns), which is how run_turn uses it — there
is no concurrent os.read to race the non-blocking window.
"""
self._buf = b""
os.set_blocking(self.fd, False)
try:
while True:
try:
if not os.read(self.fd, 65536):
break # EOF: the writer (arecord) closed
except BlockingIOError:
break # pipe drained to now (EAGAIN)
finally:
os.set_blocking(self.fd, True)
def listen_for_wake(frames, model, threshold: float, debug: bool, preroll=None):
"""Feed frames through the wake model until it crosses threshold.
Resets the model ONCE at the start of each listen phase (the streaming
contract), then runs continuously. Returns the firing score, or None if the
input stream ended before any wake.
`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 them so a request spoken
with no pause after the wake word is not clipped (issue #30). Mirrors
pipeline.stream_detect_wake's pre-roll contract on the hardware path.
"""
pipeline._reset_oww(model)
i = 0
for frame in frames:
i += 1
if preroll is not None:
preroll.append(frame)
score = max(float(s) for s in model.predict(frame).values())
if debug and (score > 0.2 or i % 100 == 0):
rms = float(np.sqrt(np.mean(frame.astype(np.float32) ** 2)))
pipeline.WAKE_LOGGER.debug(
"Wake telemetry frame=%d rms=%.0f score=%.3f", i, rms, score
)
if score >= threshold:
return score
return None
def drain(frames, n: int) -> None:
"""Discard up to n frames — clears audio captured during the turn (TTS echo,
the tail of the user's speech) so it cannot retrigger the next wake."""
for _ in range(n):
if next(frames, None) is None:
return
def run_turn(
frames,
mic,
model,
threshold: float,
out_wav: str,
output_device,
cfg: dict,
debug: bool,
) -> bool:
"""Run one full turn off the live frame stream.
`mic` is the StdinMic that owns `frames`; its flush() is used to drop the cue's
mic bleed (issue #56). Returns True if a turn ran (or was correctly skipped as
noise), False only when the input stream ended and the loop should stop.
"""
turn_started = time.monotonic()
# Keep the most recent frames during detection so the request's leading audio,
# consumed while the detector crossed threshold, is recovered (issue #30).
wake_audio = pipeline.WakeAudioBuffer()
stage_started = time.monotonic()
score = listen_for_wake(frames, model, threshold, debug, preroll=wake_audio)
if score is None:
return False
timings = {"detect_wake": time.monotonic() - stage_started}
pipeline.WAKE_LOGGER.info("Turn started")
pipeline.WAKE_LOGGER.debug("Wake score=%.4f", score)
pipeline.WAKE_LOGGER.info(
"Stage finished stage=detect_wake duration_s=%.3f", timings["detect_wake"]
)
# Acknowledge the wake with a cue before capture (issue #41), but only when the
# user paused after the wake word (issue #55). arecord cannot be paused, so on a
# shared mic/speaker device (the PowerConf) the cue bleeds straight back into the
# mic, and a command spoken in one breath with the wake word would be lost to the
# cue window. peek_cue_gate reads the post-wake audio first: a no-pause command has
# speech there, a user waiting for the cue has room tone.
#
# No-pause command: skip the cue so it cannot clip the command, and prepend the
# peeked command frames to capture -- the pre-roll stays, recovering the leading
# audio detection ate (issue #30). Paused: play the cue, then drop the frames
# buffered while it played so it is not captured as part of the request. Flush-to-now
# does this (issue #56): mic.flush() drops, in one non-blocking shot, exactly what is
# buffered the instant capture starts -- cue bleed, the peeked pre-cue room tone, and
# ambient -- then capture reads only fresh audio.
#
# flush() replaces sizing the drop by _play_wav's wall-clock span, which conflated
# the cue's audio with subprocess overhead (a failed unprivileged aplay then a sudo
# retry, plus device-open and teardown latency). That frame-count guess could
# under-drain, leaking a cue tail into the transcript, or -- because drain() blocked
# per frame -- over-drain past the buffer and consume the start of a command spoken
# after the cue. flush() carries no time estimate and never blocks on or consumes
# fresh frames, so a command spoken once _play_wav returns is captured intact. A cue
# failure can happen before or after partial output. Skip the flush because no clean
# post-cue boundary exists; this preserves the available request audio but can also
# retain partial cue bleed. The transcript guard contains that degraded path, while
# issue #58 tracks the audio that buffering or backpressure can lose.
capture_frames = frames
if cfg.get("wake_chime", False): # opt-in, default off (issue #41) until the
# pause-gate (issue #55) is validated on the PowerConf hardware
play_cue, peeked = pipeline.peek_cue_gate(
frames, vad_threshold=cfg["capture_vad_threshold"]
)
cue_boundary = False
if play_cue:
try:
_play_wav(chime.wake_cue_wav(), output_device)
except Exception as e: # noqa: BLE001 - the chime is a nicety, not core
# cfg is the process-local mapping main() reuses for every turn. A
# slow playback failure can lose speech captured during that cue
# window (#58); disable the optional cue after its first failure so
# later turns use the no-chime path instead of repeating the loss.
cfg["wake_chime"] = False
pipeline.WAKE_LOGGER.warning(
"Wake chime failed error=%s detail=%s action='disabled until restart'",
type(e).__name__,
e,
)
else:
mic.flush()
# The flush established a clean post-cue capture boundary (it dropped the
# frames buffered while the cue played). The peeked pre-cue room tone was
# already pulled off the stream and is dropped here by not being chained
# back into capture. Drop the detection pre-roll too: it holds the pre-cue
# wake-word tail, kept only to recover a no-pause command's leading audio
# (issue #30) -- which does not apply once the user waits for the cue.
# Without this the stale tail would prepend the post-cue command and reach
# the brain (issue #41).
wake_audio.clear()
cue_boundary = True
pipeline.WAKE_LOGGER.debug(
"Wake cue played peeked_frames=%d capture_boundary=true",
len(peeked),
)
else:
pipeline.WAKE_LOGGER.debug(
"Wake cue skipped peeked_frames=%d reason=speech_detected", len(peeked)
)
if not cue_boundary:
# Either a no-pause command (cue skipped so it cannot clip the command) or a
# cue that failed to establish a boundary (nothing flushed, buffered audio
# kept). Both keep the pre-roll and feed the peeked frames to capture ahead of
# the rest of the stream, preserving all audio that is still available
# (issues #30, #55, #58).
capture_frames = itertools.chain(peeked, frames)
stage_started = time.monotonic()
request_pcm = pipeline.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
pipeline.WAKE_LOGGER.info(
"Stage finished stage=capture duration_s=%.3f", timings["capture"]
)
if request_pcm.size == 0:
empty_reason = getattr(request_pcm, "empty_reason", None)
if empty_reason != pipeline._EMPTY_NO_ONSET:
pipeline.WAKE_LOGGER.info(
"Turn ignored reason=%s total_s=%.3f",
empty_reason or "empty_capture",
time.monotonic() - turn_started,
)
return True
stage_started = time.monotonic()
heard = pipeline.recover_consumed_command(wake_audio.history, cfg["wake_word"])
if heard is None:
timings["transcribe"] = time.monotonic() - stage_started
pipeline.STT_LOGGER.info(
"Stage finished stage=transcribe duration_s=%.3f",
timings["transcribe"],
)
pipeline.WAKE_LOGGER.info(
"Turn ignored reason=no_command total_s=%.3f",
time.monotonic() - turn_started,
)
return True
else:
# The captured int16 PCM is normalized by transcribe_detailed and sent straight
# to faster-whisper; no request-side temporary WAV is needed.
stage_started = time.monotonic()
heard = pipeline.transcribe_detailed(request_pcm)
timings["transcribe"] = time.monotonic() - stage_started
pipeline.STT_LOGGER.info(
"Stage finished stage=transcribe duration_s=%.3f", timings["transcribe"]
)
pipeline.STT_LOGGER.debug("Transcript text=%r", heard.text)
if not heard.text.strip():
pipeline.WAKE_LOGGER.info(
"Turn ignored reason=empty_transcript total_s=%.3f",
time.monotonic() - turn_started,
)
return True
# Mishear guard: this is the real-hardware path to the action-capable brain, so
# a low-confidence transcript must not be dispatched. On a reject, speak the
# re-prompt and skip the brain, so a garbled command never triggers an action.
ok, reason = pipeline.guard_transcript(heard, cfg)
if not ok:
pipeline.STT_LOGGER.info("Transcript rejected reason=%s", reason)
reply = pipeline.STT_REPROMPT
else:
stage_started = time.monotonic()
reply = pipeline.brain(heard.text)
timings["brain"] = time.monotonic() - stage_started
pipeline.BRAIN_LOGGER.info(
"Stage finished stage=brain duration_s=%.3f", timings["brain"]
)
pipeline.BRAIN_LOGGER.debug("Reply text=%r", reply)
stage_started = time.monotonic()
pipeline.speak(reply, out_wav)
timings["speak"] = time.monotonic() - stage_started
pipeline.TTS_LOGGER.info(
"Stage finished stage=speak duration_s=%.3f", timings["speak"]
)
try:
_play_wav(out_wav, output_device)
except Exception as e: # noqa: BLE001 - degrade to a saved WAV, never crash
pipeline.TTS_LOGGER.warning(
"Reply playback failed error=%s detail=%s reply_wav=%s",
type(e).__name__,
e,
out_wav,
)
timings["total"] = time.monotonic() - turn_started
pipeline.WAKE_LOGGER.info("Turn finished total_s=%.3f", timings["total"])
return True
def main() -> int:
ap = argparse.ArgumentParser(description="computah always-on live voice loop")
ap.add_argument(
"out_wav",
nargs="?",
default=None,
help="reply WAV path (default: a temp file reused each turn)",
)
ap.add_argument(
"-v",
"--debug",
action="store_true",
help="set log level to DEBUG and add per-frame rms+score telemetry",
)
ap.add_argument(
"-o",
"--output-device",
default=None,
help="ALSA output PCM for aplay -D (e.g. "
"plughw:CARD=PowerConf,DEV=0); default: config live_output_pcm, "
"then the ALSA default device",
)
args = ap.parse_args()
out_wav = args.out_wav
auto_wav = out_wav is None
if auto_wav:
fd, out_wav = tempfile.mkstemp(prefix="computah-reply-", suffix=".wav")
os.close(fd)
cfg = pipeline.load_config()
pipeline.configure_logging(
"DEBUG" if args.debug else cfg.get("log_level", pipeline.DEFAULTS["log_level"])
)
output_pcm = _resolve_output_pcm(args.output_device, cfg)
name = cfg["wake_word"]
threshold = cfg["wake_threshold"]
model = pipeline._get_oww_model(pipeline._resolve_wake_path(name))
# Read the raw stdin fd directly (not sys.stdin.buffer) so the mic's flush() owns
# the only buffer between the pipe and a frame — see StdinMic (issue #56).
mic = StdinMic(sys.stdin.fileno())
frames = mic.frames()
pipeline.WAKE_LOGGER.info(
"Listening wake_word=%s threshold=%.3f action='Ctrl-C to stop'", name, threshold
)
turn = 0
try:
while True:
if not run_turn(
frames,
mic,
model,
threshold,
out_wav,
output_pcm,
cfg,
args.debug,
):
pipeline.WAKE_LOGGER.info("Input stream ended")
break
turn += 1
drain(frames, _DRAIN_FRAMES)
pipeline.WAKE_LOGGER.info("Listening again turn=%d", turn)
except KeyboardInterrupt:
pipeline.WAKE_LOGGER.info("Stopped by operator")
finally:
# Remove the reply WAV only when we created it; a user-supplied path is theirs.
if auto_wav:
try:
os.unlink(out_wav)
except OSError:
pass
return 0
if __name__ == "__main__":
sys.exit(main())