diff --git a/tests/02-wakeword/main.py b/tests/02-wakeword/main.py index f3404b7..3dbe8ef 100644 --- a/tests/02-wakeword/main.py +++ b/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 -frames, and predict() returns a non-empty score dict. Detection + beep come -in the next task. +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 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( - f"Model loaded in {time.monotonic() - t0:.2f}s. " - f"Score keys: {list(model.models.keys())}" - ) + print(f"Model loaded in {time.monotonic() - t0:.2f}s.") 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." + 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() - 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) + score = float(scores.get(WAKEWORD, 0.0)) 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 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__": - sys.exit(main()) + try: + sys.exit(main()) + except KeyboardInterrupt: + print("\nbye") + sys.exit(0)