219 lines
7.7 KiB
Python
219 lines
7.7 KiB
Python
"""Barge-in detection, driven by a fake mic stream and a fake wake model
|
|
(no audio hardware, no ONNX runtime)."""
|
|
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import numpy as np
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
|
|
|
from bolt_pet.audio.barge_in import BargeInDetector, WakeWordBargeIn, make_detector
|
|
|
|
|
|
class FakeStream:
|
|
"""Yields frames of a given amplitude, mimicking sounddevice's
|
|
(data, overflowed) 2-D int16 return shape."""
|
|
|
|
def __init__(self, amplitudes):
|
|
self._amplitudes = list(amplitudes)
|
|
|
|
def read(self, frames):
|
|
amplitude = self._amplitudes.pop(0) if self._amplitudes else 0
|
|
data = np.full((frames, 1), amplitude, dtype=np.int16)
|
|
return data, False
|
|
|
|
|
|
def test_silence_never_interrupts():
|
|
detector = BargeInDetector(FakeStream([0] * 20), threshold=1000, required_frames=3)
|
|
assert not any(detector.check() for _ in range(20))
|
|
|
|
|
|
def test_sustained_speech_interrupts_after_the_required_frames():
|
|
detector = BargeInDetector(FakeStream([2000] * 5), threshold=1000, required_frames=3)
|
|
assert detector.check() is False
|
|
assert detector.check() is False
|
|
assert detector.check() is True
|
|
|
|
|
|
def test_a_single_thump_does_not_interrupt():
|
|
# loud, quiet, loud, quiet ... never three in a row
|
|
detector = BargeInDetector(FakeStream([2000, 0, 2000, 0, 2000, 0]), threshold=1000, required_frames=3)
|
|
assert not any(detector.check() for _ in range(6))
|
|
|
|
|
|
def test_counter_resets_after_a_quiet_frame():
|
|
detector = BargeInDetector(FakeStream([2000, 2000, 0, 2000, 2000, 2000]), threshold=1000, required_frames=3)
|
|
results = [detector.check() for _ in range(6)]
|
|
assert results == [False, False, False, False, False, True]
|
|
|
|
|
|
def test_reset_clears_progress():
|
|
detector = BargeInDetector(FakeStream([2000] * 6), threshold=1000, required_frames=3)
|
|
detector.check()
|
|
detector.check()
|
|
detector.reset()
|
|
assert detector.check() is False
|
|
assert detector.loud_frames == 1
|
|
|
|
|
|
def test_a_mic_error_mid_playback_is_not_fatal():
|
|
class BrokenStream:
|
|
def read(self, frames):
|
|
raise OSError("device disappeared")
|
|
|
|
detector = BargeInDetector(BrokenStream(), threshold=1000, required_frames=1)
|
|
assert detector.check() is False
|
|
|
|
|
|
# ── wake-word mode ───────────────────────────────────────────────────────────
|
|
|
|
|
|
class FakeModel:
|
|
"""Scores frames from a canned list, mimicking openWakeWord's
|
|
{class_name: score} return. Records reset() calls."""
|
|
|
|
def __init__(self, scores):
|
|
self._scores = list(scores)
|
|
self.resets = 0
|
|
|
|
def predict(self, frame):
|
|
score = self._scores.pop(0) if self._scores else 0.0
|
|
return {"thunderbolt": score}
|
|
|
|
def reset(self):
|
|
self.resets += 1
|
|
|
|
|
|
def _wake_detector(scores, threshold=0.5, amplitudes=None):
|
|
stream = FakeStream(amplitudes if amplitudes is not None else [500] * len(scores))
|
|
return WakeWordBargeIn(stream, model=FakeModel(scores), threshold=threshold), stream
|
|
|
|
|
|
def test_loud_noise_alone_does_not_interrupt_in_wake_mode():
|
|
"""The whole point of wake mode: a slammed door is deafening and scores
|
|
nothing, so the pet keeps talking."""
|
|
detector, _ = _wake_detector([0.01] * 6, amplitudes=[30000] * 6)
|
|
assert not any(detector.check() for _ in range(6))
|
|
|
|
|
|
def test_the_wake_word_interrupts():
|
|
detector, _ = _wake_detector([0.1, 0.2, 0.9])
|
|
assert [detector.check() for _ in range(3)] == [False, False, True]
|
|
|
|
|
|
def test_a_single_frame_is_enough_when_it_clears_the_threshold():
|
|
detector, _ = _wake_detector([0.55])
|
|
assert detector.check() is True
|
|
|
|
|
|
def test_scores_just_under_the_threshold_do_not_fire():
|
|
detector, _ = _wake_detector([0.49, 0.499], threshold=0.5)
|
|
assert not any(detector.check() for _ in range(2))
|
|
|
|
|
|
def test_detecting_resets_the_model_so_the_tail_is_not_reused():
|
|
model = FakeModel([0.9])
|
|
detector = WakeWordBargeIn(FakeStream([500]), model=model, threshold=0.5)
|
|
assert detector.check() is True
|
|
assert model.resets == 1
|
|
|
|
|
|
def test_reset_clears_the_models_audio_window_not_just_predictions():
|
|
"""The regression that made the pet interrupt itself a word into every
|
|
reply: openwakeword's reset() clears only the prediction buffer, so the
|
|
"thunderbolt" that started the turn was still in the preprocessor's
|
|
rolling window when playback began, and the first frame fed to the model
|
|
re-fired on it."""
|
|
|
|
class FakePreprocessor:
|
|
def __init__(self):
|
|
self.raw_data_buffer = [1, 2, 3]
|
|
self.feature_buffer = np.ones((120, 96))
|
|
self.melspectrogram_buffer = np.zeros((76, 32))
|
|
self.accumulated_samples = 4096
|
|
|
|
def _get_embeddings(self, audio):
|
|
return np.zeros((120, 96))
|
|
|
|
model = FakeModel([0.9])
|
|
model.preprocessor = FakePreprocessor()
|
|
|
|
WakeWordBargeIn(FakeStream([500]), model=model, threshold=0.5).reset()
|
|
|
|
assert model.preprocessor.raw_data_buffer == []
|
|
assert model.preprocessor.accumulated_samples == 0
|
|
assert not model.preprocessor.feature_buffer.any() # blank, not the old audio
|
|
assert model.preprocessor.melspectrogram_buffer.all() # restored to ones
|
|
|
|
|
|
def test_the_lazy_wrapper_exposes_its_preprocessor():
|
|
"""The pet holds _default_model — a lazy *wrapper* around openwakeword's
|
|
Model. If the wrapper stops proxying .preprocessor, hard_reset() finds
|
|
nothing to clear and silently degrades to the shallow reset that leaves
|
|
the last detection in the audio window. That failure is invisible: no
|
|
exception, no log, the pet just interrupts itself again."""
|
|
from bolt_pet.audio.wake_word import _OpenWakeWordModel
|
|
|
|
wrapper = _OpenWakeWordModel()
|
|
assert hasattr(wrapper, "preprocessor")
|
|
assert wrapper.preprocessor is None # not loaded yet: a no-op, not a load
|
|
|
|
class FakeInner:
|
|
preprocessor = object()
|
|
|
|
def reset(self):
|
|
pass
|
|
|
|
wrapper._model = FakeInner()
|
|
assert wrapper.preprocessor is FakeInner.preprocessor
|
|
|
|
|
|
def test_a_model_without_a_preprocessor_still_resets():
|
|
"""Fakes in tests, and any future openwakeword whose internals moved."""
|
|
model = FakeModel([0.0])
|
|
WakeWordBargeIn(FakeStream([500]), model=model, threshold=0.5).reset()
|
|
assert model.resets == 1
|
|
|
|
|
|
def test_a_callable_threshold_is_read_every_frame():
|
|
"""The tray tuner's slider has to apply mid-playback, not just mid-idle."""
|
|
threshold = {"value": 0.9}
|
|
detector = WakeWordBargeIn(
|
|
FakeStream([500] * 2), model=FakeModel([0.6, 0.6]),
|
|
threshold=lambda: threshold["value"],
|
|
)
|
|
assert detector.check() is False
|
|
threshold["value"] = 0.5
|
|
assert detector.check() is True
|
|
|
|
|
|
def test_a_model_that_blows_up_mid_playback_is_not_fatal():
|
|
class BrokenModel:
|
|
def predict(self, frame):
|
|
raise RuntimeError("onnx session died")
|
|
|
|
def reset(self):
|
|
raise RuntimeError("still dead")
|
|
|
|
detector = WakeWordBargeIn(FakeStream([500]), model=BrokenModel(), threshold=0.5)
|
|
assert detector.check() is False
|
|
detector.reset() # must not raise either
|
|
|
|
|
|
def test_a_mic_error_is_not_fatal_in_wake_mode():
|
|
class BrokenStream:
|
|
def read(self, frames):
|
|
raise OSError("device disappeared")
|
|
|
|
detector = WakeWordBargeIn(BrokenStream(), model=FakeModel([0.9]), threshold=0.5)
|
|
assert detector.check() is False
|
|
|
|
|
|
def test_make_detector_picks_the_mode():
|
|
stream = FakeStream([0])
|
|
assert isinstance(make_detector(stream, mode="wake"), WakeWordBargeIn)
|
|
assert isinstance(make_detector(stream, mode="energy"), BargeInDetector)
|
|
# A typo in .env shouldn't stop the pet from starting.
|
|
assert isinstance(make_detector(stream, mode="waek"), BargeInDetector)
|