Upload

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

tests/test_wake_word.py

@@ -0,0 +1,148 @@
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+import numpy as np
+import pytest
+
+from bolt_pet.audio import wake_word
+
+
+class _FakeStream:
+    """Yields a fixed sequence of frames, then silence forever."""
+
+    def __init__(self, frames, frame_len):
+        self._frames = list(frames)
+        self._frame_len = frame_len
+
+    def read(self, frames):
+        if self._frames:
+            frame = self._frames.pop(0)
+        else:
+            frame = np.zeros(self._frame_len, dtype=np.int16)
+        return frame.reshape(-1, 1), False
+
+
+def _frame(frame_len):
+    return np.zeros(frame_len, dtype=np.int16)
+
+
+class _FakeModel:
+    """Reports the given score sequence (one dict per predict() call, then
+    repeats the last entry) and records reset() calls."""
+
+    def __init__(self, score_sequence):
+        self._scores = list(score_sequence)
+        self.reset_calls = 0
+
+    def predict(self, frame):
+        if self._scores:
+            return self._scores.pop(0)
+        return {"thunderbolt": 0.0}
+
+    def reset(self):
+        self.reset_calls += 1
+
+
+def test_returns_true_and_resets_on_detection(monkeypatch):
+    frame_len = 1280
+    monkeypatch.setattr(wake_word.config, "FRAME_LEN", frame_len)
+    monkeypatch.setattr(wake_word.config, "SAMPLE_RATE", 16000)
+    monkeypatch.setattr(wake_word.config, "WAKE_CHECK_INTERVAL_SECONDS", 0.08)  # 1 frame
+    monkeypatch.setattr(wake_word.config, "WAKE_WORD_THRESHOLD", 0.5)
+
+    stream = _FakeStream([_frame(frame_len)] * 3, frame_len)
+    model = _FakeModel([{"thunderbolt": 0.1}, {"thunderbolt": 0.9}])
+
+    detected = wake_word.listen_for_wake_word(stream, model=model)
+
+    assert detected is True
+    assert model.reset_calls == 1
+
+
+def test_returns_false_when_should_continue_goes_false_first():
+    frame_len = 1280
+    stream = _FakeStream([_frame(frame_len)] * 5, frame_len)
+    model = _FakeModel([{"thunderbolt": 0.0}] * 5)
+    calls = {"n": 0}
+
+    def should_continue():
+        calls["n"] += 1
+        return calls["n"] <= 3
+
+    detected = wake_word.listen_for_wake_word(
+        stream, should_continue=should_continue, model=model,
+    )
+    assert detected is False
+    assert model.reset_calls == 0
+
+
+def test_custom_threshold_is_respected(monkeypatch):
+    frame_len = 1280
+    monkeypatch.setattr(wake_word.config, "FRAME_LEN", frame_len)
+    monkeypatch.setattr(wake_word.config, "SAMPLE_RATE", 16000)
+    monkeypatch.setattr(wake_word.config, "WAKE_CHECK_INTERVAL_SECONDS", 0.08)
+
+    stream = _FakeStream([_frame(frame_len)] * 3, frame_len)
+    model = _FakeModel([{"thunderbolt": 0.6}] * 3)
+    calls = {"n": 0}
+
+    def should_continue():
+        calls["n"] += 1
+        return calls["n"] <= 3
+
+    detected = wake_word.listen_for_wake_word(
+        stream, should_continue=should_continue, model=model, threshold=0.7,
+    )
+    assert detected is False
+
+
+def test_on_tick_fires_once_per_check_interval(monkeypatch):
+    frame_len = 1280
+    monkeypatch.setattr(wake_word.config, "FRAME_LEN", frame_len)
+    monkeypatch.setattr(wake_word.config, "SAMPLE_RATE", 16000)
+    monkeypatch.setattr(wake_word.config, "WAKE_CHECK_INTERVAL_SECONDS", 0.08)  # 1 frame/check
+
+    frames = [_frame(frame_len) for _ in range(5)]
+    stream = _FakeStream(frames, frame_len)
+    model = _FakeModel([{"thunderbolt": 0.0}] * 5)
+    ticks = {"n": 0}
+    state = {"i": 0}
+
+    def should_continue():
+        state["i"] += 1
+        return state["i"] <= 5
+
+    wake_word.listen_for_wake_word(
+        stream,
+        should_continue=should_continue,
+        model=model,
+        on_tick=lambda: ticks.__setitem__("n", ticks["n"] + 1),
+    )
+    assert ticks["n"] == 5  # one check-interval per frame at this config
+
+
+def test_on_tick_interval_can_span_multiple_frames(monkeypatch):
+    frame_len = 1280
+    monkeypatch.setattr(wake_word.config, "FRAME_LEN", frame_len)
+    monkeypatch.setattr(wake_word.config, "SAMPLE_RATE", 16000)
+    monkeypatch.setattr(wake_word.config, "WAKE_CHECK_INTERVAL_SECONDS", 0.24)  # 3 frames/check
+
+    frames = [_frame(frame_len) for _ in range(6)]
+    stream = _FakeStream(frames, frame_len)
+    model = _FakeModel([{"thunderbolt": 0.0}] * 6)
+    ticks = {"n": 0}
+    state = {"i": 0}
+
+    def should_continue():
+        state["i"] += 1
+        return state["i"] <= 6
+
+    wake_word.listen_for_wake_word(
+        stream,
+        should_continue=should_continue,
+        model=model,
+        on_tick=lambda: ticks.__setitem__("n", ticks["n"] + 1),
+    )
+    assert ticks["n"] == 2  # 6 frames / 3 frames-per-check