Test 1: record 5s from USB mic with RMS meter (PA_ALSA_PLUGHW for resampling)
This commit is contained in:
@@ -10,6 +10,8 @@
|
|||||||
|
|
||||||
**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.
|
**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.
|
||||||
|
|
||||||
|
**Audio note:** The Pi's USB Speaker Phone only supports 48 kHz natively. We set `PA_ALSA_PLUGHW=1` at the top of every test's `main.py` (before importing sounddevice) so PortAudio opens via ALSA's `plughw:` device, which auto-resamples to the requested 16 kHz.
|
||||||
|
|
||||||
**Spec:** `docs/superpowers/specs/2026-06-11-voice-assistant-prototypes-design.md`
|
**Spec:** `docs/superpowers/specs/2026-06-11-voice-assistant-prototypes-design.md`
|
||||||
|
|
||||||
---
|
---
|
||||||
@@ -230,6 +232,9 @@ Pass criteria for the record stage:
|
|||||||
- Level meter visibly responds to speech and to silence.
|
- Level meter visibly responds to speech and to silence.
|
||||||
- out.wav is written and non-empty.
|
- out.wav is written and non-empty.
|
||||||
"""
|
"""
|
||||||
|
import os
|
||||||
|
os.environ.setdefault("PA_ALSA_PLUGHW", "1")
|
||||||
|
|
||||||
import sys
|
import sys
|
||||||
import time
|
import time
|
||||||
|
|
||||||
@@ -422,6 +427,9 @@ 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
|
frames, and predict() returns a non-empty score dict. Detection + beep come
|
||||||
in the next task.
|
in the next task.
|
||||||
"""
|
"""
|
||||||
|
import os
|
||||||
|
os.environ.setdefault("PA_ALSA_PLUGHW", "1")
|
||||||
|
|
||||||
import sys
|
import sys
|
||||||
import time
|
import time
|
||||||
|
|
||||||
@@ -534,6 +542,9 @@ Pass criteria (per spec):
|
|||||||
- During ~1 minute of unrelated speech, <= ~1 false positive.
|
- During ~1 minute of unrelated speech, <= ~1 false positive.
|
||||||
- Python process < 50% of one core (check with htop).
|
- Python process < 50% of one core (check with htop).
|
||||||
"""
|
"""
|
||||||
|
import os
|
||||||
|
os.environ.setdefault("PA_ALSA_PLUGHW", "1")
|
||||||
|
|
||||||
import sys
|
import sys
|
||||||
import time
|
import time
|
||||||
|
|
||||||
|
|||||||
@@ -0,0 +1,85 @@
|
|||||||
|
#!/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 os
|
||||||
|
os.environ.setdefault("PA_ALSA_PLUGHW", "1")
|
||||||
|
|
||||||
|
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())
|
||||||
@@ -0,0 +1,3 @@
|
|||||||
|
sounddevice==0.5.1
|
||||||
|
soundfile==0.12.1
|
||||||
|
numpy>=2.0
|
||||||
Reference in New Issue
Block a user