Spec: voice assistant prototype phase

2026-06-11 14:18:44 +00:00
commit e2712be452
4 changed files with 815 additions and 0 deletions
@@ -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 <test-dir>` 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. `<sha> 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 <test-dir>
+# 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 <test-dir>}"
+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 <test-dir>` 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 <N>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).