Test 2 skeleton: openwakeword loads via --no-deps split, scores print

tests/02-wakeword/main.py

@@ -0,0 +1,78 @@
+#!/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 os
+os.environ.setdefault("PA_ALSA_PLUGHW", "1")
+
+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()
+
+    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())