From e2712be452ce3b5367a68541f026d2de35b07836 Mon Sep 17 00:00:00 2001 From: Assistant builder Date: Thu, 11 Jun 2026 14:18:44 +0000 Subject: [PATCH] Spec: voice assistant prototype phase --- .gitignore | 5 + CLAUDE.md | 26 + .../2026-06-11-voice-assistant-prototypes.md | 652 ++++++++++++++++++ ...06-11-voice-assistant-prototypes-design.md | 132 ++++ 4 files changed, 815 insertions(+) create mode 100644 .gitignore create mode 100644 CLAUDE.md create mode 100644 docs/superpowers/plans/2026-06-11-voice-assistant-prototypes.md create mode 100644 docs/superpowers/specs/2026-06-11-voice-assistant-prototypes-design.md diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..b324f13 --- /dev/null +++ b/.gitignore @@ -0,0 +1,5 @@ +__pycache__/ +*.pyc +.venv/ +*.wav +.openwakeword-cache/ diff --git a/CLAUDE.md b/CLAUDE.md new file mode 100644 index 0000000..3cae52c --- /dev/null +++ b/CLAUDE.md @@ -0,0 +1,26 @@ +# Assistant project + +Voice assistant on a Raspberry Pi, talking to OpenAI Realtime API for voice I/O. + +## Target device + +- Host: `assistant` (hostname), IP `192.168.50.115` +- mDNS name `assistant.local` is NOT resolvable from this sandbox — use the IP. +- OS: Debian (Raspberry Pi OS), kernel `6.12.75+rpt-rpi-v8`, aarch64 +- SSH user: `pi`, password: `assistant` + +## SSH access from this sandbox + +`ping` and `ssh`/`sshpass` are installed via apt (`iputils-ping`, `openssh-client`, `sshpass`). The host key is already in `~/.ssh/known_hosts`. + +Run a command on the Pi: + +```sh +sshpass -p 'assistant' ssh pi@192.168.50.115 '' +``` + +Copy a file to the Pi: + +```sh +sshpass -p 'assistant' scp pi@192.168.50.115: +``` diff --git a/docs/superpowers/plans/2026-06-11-voice-assistant-prototypes.md b/docs/superpowers/plans/2026-06-11-voice-assistant-prototypes.md new file mode 100644 index 0000000..cce40e0 --- /dev/null +++ b/docs/superpowers/plans/2026-06-11-voice-assistant-prototypes.md @@ -0,0 +1,652 @@ +# Voice Assistant Prototypes Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Build two throwaway test projects on a Raspberry Pi 4B — voice recording/playback and openwakeword detection — to confirm the hardware can do what the main OpenAI Realtime voice assistant will need. + +**Architecture:** Source of truth lives in this sandbox at `/app/data/home/Assistant`. A `bin/deploy ` shell helper rsyncs the project to the Pi (`pi@192.168.50.115`), ensures a shared venv exists, installs that test's `requirements.txt`, and SSHes in to run `main.py` with the user's TTY attached. Each test is one self-contained script. + +**Tech Stack:** Python 3.13, `python-sounddevice` (PortAudio), `soundfile`, `numpy`, `openwakeword` (with onnxruntime), `rsync` + `sshpass` for transport. PipeWire on the Pi exposes the USB Speaker Phone as a normal ALSA device, no PipeWire-specific code needed. + +**TDD note:** These are hardware-in-the-loop probes, not modules with unit tests. The "test" for each task is a manual run on the Pi against the spec's pass criteria. Every code-producing task ends with a deploy + run step that names the specific behaviour to look for. + +**Spec:** `docs/superpowers/specs/2026-06-11-voice-assistant-prototypes-design.md` + +--- + +## Task 1: Initialise git and commit the spec + +**Files:** +- Create: `.gitignore` +- Modify: (none — adds existing files to the index) + +- [ ] **Step 1: Init git and configure** + +```bash +cd /app/data/home/Assistant +git init -q -b main +git config user.email "meelstorm@gmail.com" +git config user.name "Assistant builder" +``` + +- [ ] **Step 2: Write `.gitignore`** + +``` +__pycache__/ +*.pyc +.venv/ +*.wav +.openwakeword-cache/ +``` + +- [ ] **Step 3: Stage and commit** + +```bash +git add .gitignore CLAUDE.md docs/ +git commit -q -m "Spec: voice assistant prototype phase" +git log --oneline +``` + +Expected output: one commit, e.g. ` Spec: voice assistant prototype phase`. + +--- + +## Task 2: Install local CLI tooling (rsync) + +**Files:** +- (none — system package install only) + +- [ ] **Step 1: Install rsync in the sandbox** + +```bash +apt-get install -y rsync 2>&1 | tail -3 +``` + +Expected: `rsync` reported as installed. + +- [ ] **Step 2: Verify rsync exists on the Pi** + +```bash +sshpass -p 'assistant' ssh pi@192.168.50.115 'which rsync && rsync --version | head -1' +``` + +Expected: `/usr/bin/rsync` and a version line. If missing, install on Pi: + +```bash +sshpass -p 'assistant' ssh -t pi@192.168.50.115 'sudo apt-get install -y rsync' +``` + +- [ ] **Step 3: Verify end-to-end sshpass+rsync** + +```bash +sshpass -p 'assistant' rsync -an --stats /app/data/home/Assistant/CLAUDE.md pi@192.168.50.115:/tmp/ | tail -5 +``` + +Expected: dry-run stats, exit 0. No commit (nothing changed). + +--- + +## Task 3: Bootstrap Pi system packages + +**Files:** +- Create: `bin/bootstrap` + +- [ ] **Step 1: Write `bin/bootstrap`** + +```bash +#!/usr/bin/env bash +# Install Pi-side system packages required by the test projects. +# Idempotent — safe to re-run. apt-get only installs what's missing. + +set -euo pipefail + +PI_USER="${PI_USER:-pi}" +PI_HOST="${PI_HOST:-192.168.50.115}" +PI_PASS="${PI_PASS:-assistant}" + +sshpass -p "$PI_PASS" ssh -t -o StrictHostKeyChecking=accept-new \ + "$PI_USER@$PI_HOST" \ + 'sudo apt-get update -qq && sudo apt-get install -y --no-install-recommends libportaudio2 libsndfile1 python3-venv' \ + 2>&1 | sed -E "s/${PI_PASS}/***/g" +``` + +- [ ] **Step 2: Make it executable and run it** + +```bash +chmod +x /app/data/home/Assistant/bin/bootstrap +/app/data/home/Assistant/bin/bootstrap +``` + +Expected: apt output, ending without errors. The packages may already be present (apt will say `is already the newest version`). + +- [ ] **Step 3: Commit** + +```bash +cd /app/data/home/Assistant +git add bin/bootstrap +git commit -q -m "Bootstrap Pi system packages (portaudio, sndfile, venv)" +``` + +--- + +## Task 4: Write `bin/deploy` + +**Files:** +- Create: `bin/deploy` + +- [ ] **Step 1: Write `bin/deploy`** + +```bash +#!/usr/bin/env bash +# Sync project to Pi, ensure venv + per-test requirements, run main.py with TTY. +# +# Usage: bin/deploy +# Example: bin/deploy tests/02-wakeword + +set -euo pipefail + +PI_USER="${PI_USER:-pi}" +PI_HOST="${PI_HOST:-192.168.50.115}" +PI_PASS="${PI_PASS:-assistant}" +PI_PATH="${PI_PATH:-/home/$PI_USER/assistant}" + +TEST_DIR="${1:?usage: bin/deploy }" +PROJECT_ROOT="$(cd "$(dirname "$0")/.." && pwd)" + +if [ ! -f "$PROJECT_ROOT/$TEST_DIR/main.py" ]; then + echo "no main.py at $PROJECT_ROOT/$TEST_DIR" >&2 + exit 1 +fi + +REDACT() { sed -E "s/${PI_PASS}/***/g"; } + +echo "==> sync $PROJECT_ROOT -> $PI_USER@$PI_HOST:$PI_PATH" +sshpass -p "$PI_PASS" rsync -az --delete \ + -e "ssh -o StrictHostKeyChecking=accept-new" \ + --exclude='.git/' --exclude='.venv/' --exclude='__pycache__/' \ + --exclude='*.wav' --exclude='.openwakeword-cache/' \ + "$PROJECT_ROOT/" "$PI_USER@$PI_HOST:$PI_PATH/" 2>&1 | REDACT + +echo "==> venv + pip install $TEST_DIR/requirements.txt" +sshpass -p "$PI_PASS" ssh -o StrictHostKeyChecking=accept-new \ + "$PI_USER@$PI_HOST" \ + "set -e + cd '$PI_PATH' + [ -d .venv ] || python3 -m venv .venv + .venv/bin/pip install --upgrade pip -q + .venv/bin/pip install -q -r '$TEST_DIR/requirements.txt'" 2>&1 | REDACT + +echo "==> run $TEST_DIR/main.py" +sshpass -p "$PI_PASS" ssh -t -o StrictHostKeyChecking=accept-new \ + "$PI_USER@$PI_HOST" \ + "cd '$PI_PATH/$TEST_DIR' && '$PI_PATH/.venv/bin/python' main.py" 2>&1 | REDACT +``` + +- [ ] **Step 2: Make executable** + +```bash +chmod +x /app/data/home/Assistant/bin/deploy +``` + +- [ ] **Step 3: Dry-check the usage error path** + +```bash +/app/data/home/Assistant/bin/deploy 2>&1 || true +``` + +Expected: `usage: bin/deploy ` on stderr, non-zero exit. + +- [ ] **Step 4: Commit** + +```bash +cd /app/data/home/Assistant +git add bin/deploy +git commit -q -m "Deploy helper: rsync, venv, run main.py over SSH" +``` + +--- + +## Task 5: Test 1 — recording with RMS meter and WAV save + +**Files:** +- Create: `tests/01-record-play/requirements.txt` +- Create: `tests/01-record-play/main.py` + +- [ ] **Step 1: Write `tests/01-record-play/requirements.txt`** + +``` +sounddevice==0.5.1 +soundfile==0.12.1 +numpy>=2.0 +``` + +- [ ] **Step 2: Write `tests/01-record-play/main.py` — record-only first cut** + +```python +#!/usr/bin/env python3 +"""Record 5s from the USB Speaker Phone with a live RMS meter, save out.wav. + +Pass criteria for the record stage: +- Level meter visibly responds to speech and to silence. +- out.wav is written and non-empty. +""" +import sys +import time + +import numpy as np +import sounddevice as sd +import soundfile as sf + +SAMPLE_RATE = 16_000 +DURATION_S = 5 +CHANNELS = 1 +DTYPE = "int16" +OUT_WAV = "out.wav" + + +def find_usb_device() -> int: + for idx, dev in enumerate(sd.query_devices()): + name = dev["name"].lower() + if "usb" in name and dev["max_input_channels"] >= 1: + return idx + print("USB audio device not found. Devices:", file=sys.stderr) + print(sd.query_devices(), file=sys.stderr) + sys.exit(1) + + +def rms(int16_buf: np.ndarray) -> float: + if int16_buf.size == 0: + return 0.0 + x = int16_buf.astype(np.float32) / 32768.0 + return float(np.sqrt((x * x).mean() + 1e-12)) + + +def record(device: int) -> np.ndarray: + print(f"Recording {DURATION_S}s from device {device} ({sd.query_devices(device)['name']!r})") + total_frames = SAMPLE_RATE * DURATION_S + buf = np.zeros((total_frames, CHANNELS), dtype=DTYPE) + frames_written = 0 + start = time.monotonic() + + def callback(indata, frames, time_info, status): + nonlocal frames_written + if status: + print(f"\n[status] {status}", file=sys.stderr) + end = min(frames_written + frames, total_frames) + buf[frames_written:end] = indata[: end - frames_written] + frames_written = end + + with sd.InputStream( + samplerate=SAMPLE_RATE, + channels=CHANNELS, + dtype=DTYPE, + device=device, + blocksize=1024, + callback=callback, + ): + while frames_written < total_frames: + elapsed = time.monotonic() - start + tail = buf[max(0, frames_written - 1600) : frames_written] + level = rms(tail) + bar = "#" * int(level * 60) + print(f"\r{elapsed:4.1f}s |{bar:<60}|", end="", flush=True) + time.sleep(0.05) + print() + return buf + + +def main() -> int: + device = find_usb_device() + audio = record(device) + sf.write(OUT_WAV, audio, SAMPLE_RATE) + print(f"Wrote {OUT_WAV} ({audio.shape[0]} frames @ {SAMPLE_RATE} Hz)") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) +``` + +- [ ] **Step 3: Deploy and run** + +```bash +/app/data/home/Assistant/bin/deploy tests/01-record-play +``` + +Talk into the mic during the 5 s window. + +Expected: time counter ticks up, bar grows while you talk and shrinks in silence. Final line `Wrote out.wav (80000 frames @ 16000 Hz)`. Exit 0. + +- [ ] **Step 4: Verify the WAV exists on the Pi** + +```bash +sshpass -p 'assistant' ssh pi@192.168.50.115 'ls -la ~/assistant/tests/01-record-play/out.wav && file ~/assistant/tests/01-record-play/out.wav' +``` + +Expected: a file ~160 KB (`80000 frames * 2 bytes`), `WAVE audio, Microsoft PCM, 16 bit, mono 16000 Hz`. + +- [ ] **Step 5: Commit** + +```bash +cd /app/data/home/Assistant +git add tests/01-record-play/ +git commit -q -m "Test 1: record 5s from USB mic with RMS meter" +``` + +--- + +## Task 6: Test 1 — add playback + +**Files:** +- Modify: `tests/01-record-play/main.py` + +- [ ] **Step 1: Add a `play` function and call it from `main`** + +Replace the existing `main` function and add `play` directly above it: + +```python +def play(device: int, audio: np.ndarray) -> None: + print(f"Playing back through device {device}") + sd.play(audio, samplerate=SAMPLE_RATE, device=device) + sd.wait() + + +def main() -> int: + device = find_usb_device() + audio = record(device) + sf.write(OUT_WAV, audio, SAMPLE_RATE) + print(f"Wrote {OUT_WAV} ({audio.shape[0]} frames @ {SAMPLE_RATE} Hz)") + play(device, audio) + return 0 +``` + +Update the module docstring's "Pass criteria" block to: + +```python +"""Record 5s from the USB Speaker Phone with a live RMS meter, save out.wav, play it back. + +Pass criteria (per spec): +- Playback is recognisable as what was said into the mic. +- Level meter visibly responds to speech and to silence. +- No clipping, dropouts, or sustained PortAudio under/overrun warnings. +""" +``` + +- [ ] **Step 2: Deploy and run** + +```bash +/app/data/home/Assistant/bin/deploy tests/01-record-play +``` + +Speak something memorable during the 5 s window (e.g. "the rain in Spain"). After the meter finishes you should hear it played back. + +Expected against the spec's pass criteria: +- You recognise your own utterance. +- No `[status]` lines printed during streaming. +- Exit 0. + +If the playback is silent or clipped, debug before continuing — that's a hardware-level failure of the prototype. + +- [ ] **Step 3: Commit** + +```bash +cd /app/data/home/Assistant +git add tests/01-record-play/main.py +git commit -q -m "Test 1: add playback step — record-play loop complete" +``` + +--- + +## Task 7: Test 2 — wakeword skeleton (load model, open stream, log scores) + +**Files:** +- Create: `tests/02-wakeword/requirements.txt` +- Create: `tests/02-wakeword/main.py` + +- [ ] **Step 1: Write `tests/02-wakeword/requirements.txt`** + +``` +sounddevice==0.5.1 +numpy>=2.0 +openwakeword==0.6.0 +onnxruntime>=1.18 +``` + +- [ ] **Step 2: Write `tests/02-wakeword/main.py` — skeleton, prints scores but no detection logic yet** + +```python +#!/usr/bin/env python3 +"""Skeleton: load openwakeword's 'alexa' model and print scores every 80 ms. + +This task validates that the model loads, the mic stream produces 1280-sample +frames, and predict() returns a non-empty score dict. Detection + beep come +in the next task. +""" +import sys +import time + +import numpy as np +import sounddevice as sd +import openwakeword.utils +from openwakeword.model import Model + +SAMPLE_RATE = 16_000 +FRAME_SAMPLES = 1280 # 80 ms @ 16 kHz, openwakeword's expected chunk +CHANNELS = 1 +DTYPE = "int16" +WAKEWORD = "alexa" + + +def find_usb_device() -> int: + for idx, dev in enumerate(sd.query_devices()): + name = dev["name"].lower() + if "usb" in name and dev["max_input_channels"] >= 1: + return idx + print("USB audio device not found. Devices:", file=sys.stderr) + print(sd.query_devices(), file=sys.stderr) + sys.exit(1) + + +def main() -> int: + print("Ensuring openwakeword onnx models are downloaded (idempotent)...") + openwakeword.utils.download_models(target_directory=None) + + print("Loading openwakeword model...") + t0 = time.monotonic() + model = Model(wakeword_models=[WAKEWORD], inference_framework="onnx") + print( + f"Model loaded in {time.monotonic() - t0:.2f}s. " + f"Score keys: {list(model.models.keys())}" + ) + + device = find_usb_device() + print( + f"Streaming from device {device} ({sd.query_devices(device)['name']!r}). " + f"Will print {WAKEWORD} score 5x/s for ~10s, then exit." + ) + + start = time.monotonic() + last_print = 0.0 + with sd.InputStream( + samplerate=SAMPLE_RATE, + channels=CHANNELS, + dtype=DTYPE, + device=device, + blocksize=FRAME_SAMPLES, + ) as stream: + while time.monotonic() - start < 10.0: + frame, overflowed = stream.read(FRAME_SAMPLES) + if overflowed: + print("[status] input overflow", file=sys.stderr) + mono = frame[:, 0] + scores = model.predict(mono) + now = time.monotonic() + if now - last_print >= 0.2: + score = float(scores.get(WAKEWORD, 0.0)) + print(f"t={now - start:4.1f}s {WAKEWORD}={score:.3f}") + last_print = now + return 0 + + +if __name__ == "__main__": + sys.exit(main()) +``` + +- [ ] **Step 3: Deploy and run** + +```bash +/app/data/home/Assistant/bin/deploy tests/02-wakeword +``` + +First run downloads model weights (~5–15 MB). Subsequent runs use the cache. + +Expected output, in order: +1. `Loading openwakeword model ...` +2. `Model loaded in s. Score keys: ['alexa']` +3. ~50 lines of `t=X.Xs alexa=0.XXX` over 10 s, mostly small numbers (< 0.2) when you stay quiet. Say "alexa" and one of those samples should jump (≥ 0.5). +4. Exit 0, no overflow status lines. + +If the model fails to download, check the Pi has internet and re-run. + +- [ ] **Step 4: Commit** + +```bash +cd /app/data/home/Assistant +git add tests/02-wakeword/ +git commit -q -m "Test 2 skeleton: openwakeword loads, scores print" +``` + +--- + +## Task 8: Test 2 — detection, beep, cooldown + +**Files:** +- Modify: `tests/02-wakeword/main.py` + +- [ ] **Step 1: Replace the file with the detection-loop version** + +```python +#!/usr/bin/env python3 +"""Listen for the 'alexa' wakeword via openwakeword. Beep on detection. + +Pass criteria (per spec): +- "alexa" at normal volume from ~1 m triggers within ~1 s (>= 8/10 attempts). +- During ~1 minute of unrelated speech, <= ~1 false positive. +- Python process < 50% of one core (check with htop). +""" +import sys +import time + +import numpy as np +import sounddevice as sd +import openwakeword.utils +from openwakeword.model import Model + +SAMPLE_RATE = 16_000 +FRAME_SAMPLES = 1280 +CHANNELS = 1 +DTYPE = "int16" +WAKEWORD = "alexa" +THRESHOLD = 0.5 +COOLDOWN_S = 1.0 +BEEP_FREQ_HZ = 880.0 +BEEP_DURATION_S = 0.2 + + +def find_usb_device() -> int: + for idx, dev in enumerate(sd.query_devices()): + name = dev["name"].lower() + if "usb" in name and dev["max_input_channels"] >= 1: + return idx + print("USB audio device not found. Devices:", file=sys.stderr) + print(sd.query_devices(), file=sys.stderr) + sys.exit(1) + + +def beep(device: int) -> None: + t = np.arange(int(SAMPLE_RATE * BEEP_DURATION_S)) / SAMPLE_RATE + tone = (0.3 * np.sin(2 * np.pi * BEEP_FREQ_HZ * t)).astype(np.float32) + sd.play(tone, samplerate=SAMPLE_RATE, device=device) + sd.wait() + + +def main() -> int: + print("Ensuring openwakeword onnx models are downloaded (idempotent)...") + openwakeword.utils.download_models(target_directory=None) + + print("Loading openwakeword model...") + t0 = time.monotonic() + model = Model(wakeword_models=[WAKEWORD], inference_framework="onnx") + print(f"Model loaded in {time.monotonic() - t0:.2f}s.") + + device = find_usb_device() + print( + f"Listening on device {device} ({sd.query_devices(device)['name']!r}). " + f"Say '{WAKEWORD}'. Ctrl-C to exit." + ) + + last_trigger = 0.0 + start = time.monotonic() + + with sd.InputStream( + samplerate=SAMPLE_RATE, + channels=CHANNELS, + dtype=DTYPE, + device=device, + blocksize=FRAME_SAMPLES, + ) as stream: + while True: + frame, overflowed = stream.read(FRAME_SAMPLES) + if overflowed: + print("[status] input overflow", file=sys.stderr) + mono = frame[:, 0] + scores = model.predict(mono) + score = float(scores.get(WAKEWORD, 0.0)) + now = time.monotonic() + if score >= THRESHOLD and (now - last_trigger) >= COOLDOWN_S: + print(f"DETECTED {WAKEWORD} score={score:.3f} t={now - start:.1f}s") + last_trigger = now + beep(device) + + +if __name__ == "__main__": + try: + sys.exit(main()) + except KeyboardInterrupt: + print("\nbye") + sys.exit(0) +``` + +- [ ] **Step 2: Deploy and run** + +```bash +/app/data/home/Assistant/bin/deploy tests/02-wakeword +``` + +- [ ] **Step 3: Validate pass criteria interactively** + +While the script is running: + +1. **Reliability:** say "alexa" ten times, with ~3 s gaps. Count detections. Expect ≥ 8. +2. **False positives:** for ~60 s, read aloud from any nearby text (not containing "alexa"). Expect ≤ 1 print of `DETECTED`. +3. **CPU:** in a second SSH session (`sshpass -p 'assistant' ssh pi@192.168.50.115`), run `top -bn1 | grep python` and confirm the python process is < 50% on a single core. + +If any criterion fails, decide whether it's a tuning issue (threshold, cooldown) or a hardware/library issue and fix before declaring the prototype phase done. + +- [ ] **Step 4: Stop with Ctrl-C** — expect `bye` and exit 0. + +- [ ] **Step 5: Commit** + +```bash +cd /app/data/home/Assistant +git add tests/02-wakeword/main.py +git commit -q -m "Test 2: detection threshold + cooldown + beep" +``` + +--- + +## Done criteria + +- Both `bin/deploy tests/01-record-play` and `bin/deploy tests/02-wakeword` produce runs that meet the spec's pass criteria. +- All eight tasks committed on `main`. +- No follow-up needed before opening the next design doc (main assistant + OpenAI Realtime). diff --git a/docs/superpowers/specs/2026-06-11-voice-assistant-prototypes-design.md b/docs/superpowers/specs/2026-06-11-voice-assistant-prototypes-design.md new file mode 100644 index 0000000..ae850c4 --- /dev/null +++ b/docs/superpowers/specs/2026-06-11-voice-assistant-prototypes-design.md @@ -0,0 +1,132 @@ +# Voice Assistant — prototype phase design + +**Date:** 2026-06-11 +**Status:** approved +**Scope:** two throwaway test projects that de-risk audio I/O and wakeword detection on the target device before main development of the OpenAI Realtime voice assistant begins. + +## Context + +The end goal is a voice assistant on a Raspberry Pi 4B that streams microphone audio to the OpenAI Realtime API and plays back the model's speech. Before building that, we need to confirm that the two foundational pieces work on this exact hardware: + +1. **Voice recording and playback** — can we capture clean mic audio and play it back through the speaker at the sample rate and format the Realtime API expects? +2. **Wakeword detection** — can we run a wakeword model locally on the Pi 4 in real time, accurately enough to gate the Realtime session? + +Both tests are scoped as standalone scripts. They are not meant to evolve into the final assistant — they exist to answer yes/no questions. The final assistant will be designed separately after both probes return "yes." + +## Target device + +- Raspberry Pi 4B, 4 GB RAM, aarch64 +- Debian 13 (trixie), kernel `6.12.75+rpt-rpi-v8` +- Python 3.13.5 (`/usr/bin/python3`) +- Audio server: PipeWire (active, user session) +- Audio device: ALSA card 3, `USB Speaker Phone` (Anhui LISTENAI, USB ID `f201:3220`) — combined microphone and speaker on a single USB device +- Reached over SSH from this sandbox at `pi@192.168.50.115` (see `CLAUDE.md`) + +The USB Speaker Phone is the only mic + speaker on the device. The HDMI outputs and bcm2835 headphone jack are present but unused for these tests. + +## Wakeword choice + +- Engine: **openwakeword** (Apache 2.0, runs on CPU, no API key) +- Model for the test: stock `alexa` — chosen because it ships with the library and lets us validate the engine end-to-end without training. The final assistant will use a custom-trained "assistant" (or similar) model; training that is a separate task that happens after this prototype phase passes. + +## Audio I/O library + +`python-sounddevice` (PortAudio bindings) for both tests: + +- One consistent stack across record, play, and wakeword feeding. +- Streaming callback API that matches openwakeword's chunked input requirement. +- Plays nicely with PipeWire's ALSA shim — no PipeWire-specific code needed. +- Same library will be reused in the main assistant for the OpenAI Realtime stream. + +`soundfile` is used in test 1 for WAV round-tripping. + +`pyaudio` and `arecord`/`aplay` subprocess were considered and rejected: `pyaudio` has no advantage and a worse install story on Debian 13; the subprocess approach is awkward for the wakeword feeding loop. + +## Repository layout + +``` +Assistant/ +├── CLAUDE.md +├── bin/ +│ └── deploy +├── tests/ +│ ├── 01-record-play/ +│ │ ├── main.py +│ │ └── requirements.txt +│ └── 02-wakeword/ +│ ├── main.py +│ └── requirements.txt +└── docs/superpowers/specs/ + └── 2026-06-11-voice-assistant-prototypes-design.md # this file +``` + +Source of truth lives in `/app/data/home/Assistant` (in this sandbox). Code is rsynced to the Pi to run. One shared venv on the Pi at `~/assistant/.venv` is created on first deploy; each test's `requirements.txt` is installed into it. + +### `bin/deploy` + +A small shell helper. Usage: `bin/deploy `, e.g. `bin/deploy tests/02-wakeword`. + +Responsibilities: +1. `rsync` the project tree to `pi@192.168.50.115:~/assistant/`. +2. Ensure `~/assistant/.venv` exists (create with `python3 -m venv` if not). +3. `pip install -r /requirements.txt` inside the venv. +4. SSH in and run `/main.py` with the venv's Python, attaching to the terminal so the user can interact and see output live. + +The script must not echo `$GIT_KEY` / `$GIT_USER`, and pipes ssh/rsync stderr through `sed` to redact them in case of error output. Auth uses `sshpass -p assistant`. + +## Test 1 — voice recording and playback + +`tests/01-record-play/main.py` does, in order: + +1. Open `sounddevice.InputStream` on the USB Speaker Phone, 16 kHz, mono, `int16`. +2. Record 5 seconds while printing a one-line, updating-in-place RMS meter so the operator can see the mic is producing signal. +3. Write the buffer to `out.wav` via `soundfile.write` (16 kHz mono PCM16). +4. Open `sounddevice.OutputStream` on the same device and play `out.wav` back, blocking until finished. + +### Pass criteria + +- Playback is recognisable as what was said into the mic. +- Level meter visibly responds to speech and to silence. +- No clipping, dropouts, or sustained underrun warnings from PortAudio. + +### Error handling + +- If the USB Speaker Phone is not found (no ALSA card whose name contains `USB`), print the output of `sounddevice.query_devices()` and exit with a non-zero status. Suggest the user check `arecord -l`. +- Any PortAudio error during streaming aborts with the raw error message — no retry. + +## Test 2 — wakeword + +`tests/02-wakeword/main.py` does: + +1. On startup, load openwakeword's stock `alexa` model. If the model files are not yet cached locally, openwakeword's downloader fetches them. Print the model load time. +2. Open `sounddevice.InputStream` on the USB Speaker Phone, 16 kHz, mono, `int16`, with a frame size of **1280 samples (80 ms)** — openwakeword's expected chunk size. +3. In the input callback, call `model.predict(frame)` and read the `alexa` score from the returned dict. +4. When the score crosses **0.5**, print `DETECTED alexa score= t=` and play a short (~200 ms) sine-wave beep through the same device. Keep listening afterwards. Apply a 1 s cooldown so a single trigger event doesn't fire repeatedly. +5. `Ctrl-C` exits cleanly (close the stream, exit 0). + +### Pass criteria + +- Saying "alexa" at normal speaking volume from ~1 m away triggers a detection within ~1 s, reliably (≥ 8 out of 10 attempts). +- During roughly a minute of unrelated speech (reading aloud, conversation), false positives stay under ~1 per minute. +- CPU usage stays well below saturating a single core (`htop` shows the Python process < 50% on one core). + +### Error handling + +- Audio device missing: same behaviour as test 1. +- Model download failure (no network on first run): print the underlying exception and exit non-zero. Suggest re-running once the Pi has internet. + +## Out of scope + +The following are explicitly **not** part of this phase: + +- OpenAI Realtime API wiring, session management, streaming audio to and from the model. +- Voice activity detection, barge-in, conversation turn-taking. +- A custom-trained wakeword model for the final wake phrase. +- A systemd service / autostart for the assistant. +- Any production-grade error handling, logging, or observability. + +These are intentionally deferred. Each will get its own spec after both tests above pass. + +## Success of this phase + +Both tests pass their criteria when run on the Pi. After that, we open a new design doc for the main assistant.