"""Utterance endpointing while the device plays audio through its own speakers. Silero rates music/TV as speech, so without a loudness gate a recording never sees silence and runs until the hard length cap. These tests drive synthetic frames through the real ``record_utterance`` loop. """ from __future__ import annotations import asyncio import unittest import numpy as np from javis.audio_io.vad import SpeechEndpointer, VADResult, make_vad CHUNK_MS = 32 SR = 16000 SAMPLES = SR * CHUNK_MS // 1000 MUSIC_RMS = 0.02 VOICE_RMS = 0.15 QUIET_RMS = 0.0005 def frame(rms: float) -> np.ndarray: """A constant-amplitude frame whose RMS is exactly ``rms``.""" return np.full(SAMPLES, rms, dtype=np.float32) class AlwaysSpeech: """Stand-in for Silero: calls everything voice-like, just as it does for music, singing and TV dialogue.""" sample_rate = SR threshold = 0.5 def __init__(self) -> None: self.resets = 0 def reset(self) -> None: self.resets += 1 def process(self, chunk: np.ndarray) -> VADResult: return VADResult(is_speech=True, probability=1.0) class FakeMic: chunk_ms = CHUNK_MS def __init__(self, frames: list[np.ndarray]) -> None: self._frames = frames def drain(self) -> None: pass async def chunks(self, stale_timeout=None): for f in self._frames: yield f await asyncio.sleep(0) def endpointer(ratio: float = 3.0) -> SpeechEndpointer: return SpeechEndpointer(AlwaysSpeech(), onset_ratio=ratio, chunk_ms=CHUNK_MS) def record(frames, vad, *, silence_ms=800, grace_ms=1500, max_seconds=20.0): from javis.audio_io.capture import record_utterance from javis.core.config import Config cfg = Config() cfg.wake_word.silence_ms = silence_ms return asyncio.run(record_utterance( cfg, FakeMic(frames), vad, max_seconds=max_seconds, initial_grace_ms=grace_ms, )) class SpeechEndpointerTests(unittest.TestCase): def test_music_alone_is_not_speech_once_background_is_known(self) -> None: ep = endpointer() for _ in range(50): ep.observe(frame(MUSIC_RMS)) ep.reset() self.assertFalse(ep.process(frame(MUSIC_RMS)).is_speech) def test_voice_over_music_is_speech(self) -> None: ep = endpointer() for _ in range(50): ep.observe(frame(MUSIC_RMS)) ep.reset() self.assertTrue(ep.process(frame(VOICE_RMS)).is_speech) def test_quiet_room_still_detects_normal_speech(self) -> None: ep = endpointer() for _ in range(50): ep.observe(frame(QUIET_RMS)) ep.reset() self.assertTrue(ep.process(frame(0.05)).is_speech) # ...but not the room tone itself. ep.reset() self.assertFalse(ep.process(frame(QUIET_RMS)).is_speech) def test_long_utterance_does_not_raise_the_gate_onto_itself(self) -> None: ep = endpointer() for _ in range(50): ep.observe(frame(QUIET_RMS)) ep.reset() for _ in range(200): self.assertTrue(ep.process(frame(0.05)).is_speech) def test_release_is_more_forgiving_than_onset(self) -> None: """A trailing quiet syllable must not be clipped off mid-word.""" ep = endpointer() for _ in range(50): ep.observe(frame(MUSIC_RMS)) ep.reset() trailing = frame(MUSIC_RMS * 2.0) # below onset (3x), above release self.assertFalse(ep.process(trailing).is_speech) ep.process(frame(VOICE_RMS)) # user starts talking self.assertTrue(ep.process(trailing).is_speech) def test_reset_keeps_the_background_but_clears_the_latch(self) -> None: ep = endpointer() for _ in range(50): ep.observe(frame(MUSIC_RMS)) ep.process(frame(VOICE_RMS)) before = ep.background ep.reset() self.assertAlmostEqual(ep.background, before) self.assertFalse(ep.process(frame(MUSIC_RMS * 2.0)).is_speech) def test_a_higher_ratio_is_stricter(self) -> None: """Pins the documented tuning direction. Raising the ratio rejects more audio (so it can clip quiet commands); lowering it admits more (so playback can keep the recorder running). """ def accepts(ratio: float) -> bool: ep = endpointer(ratio=ratio) for _ in range(50): ep.observe(frame(MUSIC_RMS)) ep.reset() return ep.process(frame(MUSIC_RMS * 2.5)).is_speech self.assertTrue(accepts(2.0)) # permissive — the frame counts as the user self.assertFalse(accepts(4.0)) # strict — the same frame is rejected def test_ratio_zero_disables_the_gate(self) -> None: ep = endpointer(ratio=0.0) self.assertFalse(ep.enabled) for _ in range(50): ep.observe(frame(MUSIC_RMS)) ep.reset() self.assertTrue(ep.process(frame(MUSIC_RMS)).is_speech) def test_reset_resets_the_inner_vad(self) -> None: inner = AlwaysSpeech() ep = SpeechEndpointer(inner, chunk_ms=CHUNK_MS) ep.reset() self.assertEqual(inner.resets, 1) def test_threshold_proxies_to_the_inner_vad(self) -> None: inner = AlwaysSpeech() ep = SpeechEndpointer(inner, chunk_ms=CHUNK_MS) ep.threshold = 0.9 self.assertEqual(inner.threshold, 0.9) self.assertEqual(ep.threshold, 0.9) def test_empty_frame_is_harmless(self) -> None: ep = endpointer() self.assertEqual(ep.observe(np.zeros(0, dtype=np.float32)), 0.0) self.assertFalse(ep.process(np.zeros(0, dtype=np.float32)).is_speech) class RecordUtteranceEndpointingTests(unittest.TestCase): def test_recording_ends_when_the_user_stops_not_the_music(self) -> None: ep = endpointer() for _ in range(50): ep.observe(frame(MUSIC_RMS)) # wake-word loop heard the music speech = [frame(VOICE_RMS)] * 30 # ~1.0 s of command tail = [frame(MUSIC_RMS)] * 200 # music keeps playing audio = record(speech + tail, ep) self.assertIsNotNone(audio) # 30 speech frames + 25 frames (800 ms) of "silence" = ~1.76 s, far # short of the 20 s cap the old endpointer ran into. self.assertLess(len(audio) / SR, 3.0) def test_music_only_returns_none_instead_of_20s_of_audio(self) -> None: ep = endpointer() for _ in range(50): ep.observe(frame(MUSIC_RMS)) audio = record([frame(MUSIC_RMS)] * 400, ep) self.assertIsNone(audio) def test_without_the_gate_music_runs_to_the_cap(self) -> None: """The behaviour this change fixes, pinned so it cannot creep back.""" ep = endpointer(ratio=0.0) audio = record([frame(MUSIC_RMS)] * 400, ep, max_seconds=2.0) self.assertIsNotNone(audio) self.assertGreaterEqual(len(audio) / SR, 1.9) def test_quiet_room_recording_is_unaffected(self) -> None: ep = endpointer() for _ in range(50): ep.observe(frame(QUIET_RMS)) speech = [frame(0.05)] * 30 tail = [frame(QUIET_RMS)] * 200 audio = record(speech + tail, ep) self.assertIsNotNone(audio) self.assertLess(len(audio) / SR, 3.0) class MakeVadWiringTests(unittest.TestCase): def test_make_vad_returns_an_endpointer(self) -> None: vad = make_vad(voice_gate_ratio=2.5) self.assertIsInstance(vad, SpeechEndpointer) self.assertEqual(vad.onset_ratio, 2.5) self.assertTrue(hasattr(vad, "observe")) def test_config_clamps_the_ratio(self) -> None: from javis.core.config import Config, validate cfg = Config() self.assertEqual(cfg.wake_word.voice_gate_ratio, 3.0) cfg.wake_word.voice_gate_ratio = -1.0 validate(cfg) self.assertEqual(cfg.wake_word.voice_gate_ratio, 3.0) def test_master_pushes_the_ratio_to_slaves(self) -> None: from javis.api.routes import _wake_word_block from javis.core.config import Config cfg = Config() cfg.wake_word.voice_gate_ratio = 4.0 self.assertEqual(_wake_word_block(cfg).voice_gate_ratio, 4.0) if __name__ == "__main__": unittest.main()