Test 2: detection threshold + cooldown + beep

tests/02-wakeword/main.py

@@ -1,9 +1,10 @@
 #!/usr/bin/env python3
-"""Skeleton: load openwakeword's 'alexa' model and print scores every 80 ms.
+"""Listen for the 'alexa' wakeword via openwakeword. Beep on detection.
 
-This task validates that the model loads, the mic stream produces 1280-sample
+Pass criteria (per spec):
-frames, and predict() returns a non-empty score dict. Detection + beep come
+- "alexa" at normal volume from ~1 m triggers within ~1 s (>= 8/10 attempts).
-in the next task.
+- During ~1 minute of unrelated speech, <= ~1 false positive.
+- Python process < 50% of one core (check with htop).
 """
 import os
 os.environ.setdefault("PA_ALSA_PLUGHW", "1")
@@ -17,10 +18,14 @@ import openwakeword.utils
 from openwakeword.model import Model
 
 SAMPLE_RATE = 16_000
-FRAME_SAMPLES = 1280  # 80 ms @ 16 kHz, openwakeword's expected chunk
+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:
@@ -33,6 +38,13 @@ def find_usb_device() -> int:
     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()
@@ -40,19 +52,17 @@ def main() -> int:
     print("Loading openwakeword model...")
     t0 = time.monotonic()
     model = Model(wakeword_models=[WAKEWORD], inference_framework="onnx")
-    print(
+    print(f"Model loaded in {time.monotonic() - t0:.2f}s.")
-        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"Listening on device {device} ({sd.query_devices(device)['name']!r}). "
-        f"Will print {WAKEWORD} score 5x/s for ~10s, then exit."
+        f"Say '{WAKEWORD}'. Ctrl-C to exit."
     )
 
+    last_trigger = 0.0
     start = time.monotonic()
-    last_print = 0.0
+
     with sd.InputStream(
         samplerate=SAMPLE_RATE,
         channels=CHANNELS,
@@ -60,19 +70,23 @@ def main() -> int:
         device=device,
         blocksize=FRAME_SAMPLES,
     ) as stream:
-        while time.monotonic() - start < 10.0:
+        while True:
             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}")
+            now = time.monotonic()
-                last_print = now
+            if score >= THRESHOLD and (now - last_trigger) >= COOLDOWN_S:
-    return 0
+                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)