149 lines
4.5 KiB
Python
149 lines
4.5 KiB
Python
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
|