346 lines
12 KiB
Python
346 lines
12 KiB
Python
"""PCM adaptation and worker ownership, with a controlled Silero iterator."""
|
|
|
|
import asyncio
|
|
import struct
|
|
import threading
|
|
|
|
import pytest
|
|
|
|
from voice_assistant.vad import silero
|
|
from voice_assistant.vad.models import SpeechEnded, SpeechStarted
|
|
from voice_assistant.vad.silero import SileroVADAdapter
|
|
|
|
|
|
@pytest.fixture
|
|
def anyio_backend():
|
|
return "asyncio"
|
|
|
|
|
|
class ScriptedIterator:
|
|
def __init__(self, boundaries=()):
|
|
self.boundaries = boundaries
|
|
self.windows = []
|
|
self.position = 0
|
|
self.resets = 0
|
|
|
|
def __call__(self, samples, *, return_seconds):
|
|
assert return_seconds is False
|
|
self.windows.append(samples)
|
|
index = self.position
|
|
self.position += 1
|
|
return self.boundaries[index] if index < len(self.boundaries) else None
|
|
|
|
def reset_states(self):
|
|
self.resets += 1
|
|
self.position = 0
|
|
self.windows.clear()
|
|
|
|
|
|
def install_iterator(monkeypatch, iterator):
|
|
monkeypatch.setattr(silero, "_load_iterator", lambda *args: iterator)
|
|
|
|
|
|
@pytest.mark.parametrize("options", [
|
|
{"threshold": 0}, {"threshold": 1}, {"threshold": float("nan")},
|
|
{"threshold": True}, {"min_silence_duration_ms": 0},
|
|
{"min_silence_duration_ms": True}, {"speech_pad_ms": -1},
|
|
{"speech_pad_ms": 1.5},
|
|
])
|
|
def test_invalid_settings_are_rejected(options):
|
|
with pytest.raises(ValueError):
|
|
SileroVADAdapter(**options)
|
|
|
|
|
|
@pytest.mark.anyio
|
|
@pytest.mark.parametrize("sizes", [[4096], [2, 66, 1024, 1600, 1404], [512] * 8])
|
|
async def test_boundaries_are_independent_of_pcm_chunk_sizes(monkeypatch, sizes):
|
|
iterator = ScriptedIterator([
|
|
{"start": 0}, {"end": 700}, {"start": 1100}, {"end": 1700},
|
|
])
|
|
install_iterator(monkeypatch, iterator)
|
|
detector = SileroVADAdapter()
|
|
audio = b"\x00\x00" * 2048
|
|
events = []
|
|
offset = 0
|
|
try:
|
|
for size in sizes:
|
|
events.extend(await detector.process(audio[offset:offset + size]))
|
|
offset += size
|
|
assert offset == len(audio)
|
|
assert events == [
|
|
SpeechStarted(0), SpeechEnded(700, "silence"),
|
|
SpeechStarted(1100), SpeechEnded(1700, "silence"),
|
|
]
|
|
assert len(iterator.windows) == 4
|
|
assert all(len(window) == 512 for window in iterator.windows)
|
|
finally:
|
|
await detector.aclose()
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_pcm_decoding_is_little_endian_and_normalized(monkeypatch):
|
|
iterator = ScriptedIterator()
|
|
install_iterator(monkeypatch, iterator)
|
|
detector = SileroVADAdapter()
|
|
audio = struct.pack("<5h", -32768, -16384, 0, 16384, 32767) + bytes(1014)
|
|
try:
|
|
await detector.process(audio)
|
|
assert iterator.windows[0][:5] == [-1.0, -0.5, 0.0, 0.5, 32767 / 32768]
|
|
finally:
|
|
await detector.aclose()
|
|
|
|
|
|
@pytest.mark.anyio
|
|
@pytest.mark.parametrize("audio", [b"", b"\x00", b"\x00" * 3])
|
|
async def test_invalid_pcm_preserves_detection_timeline(monkeypatch, audio):
|
|
iterator = ScriptedIterator([{"start": 0}])
|
|
install_iterator(monkeypatch, iterator)
|
|
detector = SileroVADAdapter()
|
|
try:
|
|
with pytest.raises(ValueError):
|
|
await detector.process(audio)
|
|
assert iterator.windows == []
|
|
assert await detector.process(bytes(1024)) == (SpeechStarted(0),)
|
|
finally:
|
|
await detector.aclose()
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_flush_processes_tail_and_keeps_real_audio_positions(monkeypatch):
|
|
iterator = ScriptedIterator([{"start": 0}, {"end": 1000}])
|
|
install_iterator(monkeypatch, iterator)
|
|
detector = SileroVADAdapter()
|
|
try:
|
|
assert await detector.process(bytes(1026)) == (SpeechStarted(0),)
|
|
assert await detector.flush() == (SpeechEnded(513, "end_of_input"),)
|
|
assert await detector.flush() == ()
|
|
assert len(iterator.windows) == 2
|
|
with pytest.raises(RuntimeError, match="finished"):
|
|
await detector.process(bytes(1024))
|
|
finally:
|
|
await detector.aclose()
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_short_utterance_is_closed_at_real_end_of_input(monkeypatch):
|
|
iterator = ScriptedIterator([{"start": 0}])
|
|
install_iterator(monkeypatch, iterator)
|
|
detector = SileroVADAdapter()
|
|
try:
|
|
assert await detector.process(bytes(200)) == ()
|
|
assert await detector.flush() == (
|
|
SpeechStarted(0), SpeechEnded(100, "end_of_input"),
|
|
)
|
|
finally:
|
|
await detector.aclose()
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_flush_does_not_emit_speech_inside_synthetic_padding(monkeypatch):
|
|
iterator = ScriptedIterator([{"start": 400}])
|
|
install_iterator(monkeypatch, iterator)
|
|
detector = SileroVADAdapter()
|
|
try:
|
|
await detector.process(bytes(200))
|
|
assert await detector.flush() == ()
|
|
finally:
|
|
await detector.aclose()
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_reset_discards_partial_audio_and_restarts_timeline(monkeypatch):
|
|
iterator = ScriptedIterator([{"start": 0}])
|
|
install_iterator(monkeypatch, iterator)
|
|
detector = SileroVADAdapter()
|
|
try:
|
|
await detector.process(bytes(1124))
|
|
await detector.reset()
|
|
assert iterator.resets == 1
|
|
assert await detector.process(bytes(1024)) == (SpeechStarted(0),)
|
|
assert len(iterator.windows) == 1
|
|
assert await detector.flush() == (SpeechEnded(512, "end_of_input"),)
|
|
finally:
|
|
await detector.aclose()
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_close_discards_audio_without_loading_or_flushing_model(monkeypatch):
|
|
def fail_loading(*args):
|
|
raise AssertionError("Close must not load the model")
|
|
|
|
monkeypatch.setattr(silero, "_load_iterator", fail_loading)
|
|
detector = SileroVADAdapter()
|
|
await detector.process(bytes(100))
|
|
await detector.aclose()
|
|
await detector.aclose()
|
|
for operation in [lambda: detector.process(bytes(1024)), detector.flush, detector.reset]:
|
|
with pytest.raises(RuntimeError):
|
|
await operation()
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_detector_state_is_isolated_between_input_streams(monkeypatch):
|
|
iterators = []
|
|
|
|
def create(*args):
|
|
iterator = ScriptedIterator([{"start": 0}, {"end": 900}])
|
|
iterators.append(iterator)
|
|
return iterator
|
|
|
|
monkeypatch.setattr(silero, "_load_iterator", create)
|
|
first, second = SileroVADAdapter(), SileroVADAdapter()
|
|
try:
|
|
assert await first.process(bytes(1024)) == (SpeechStarted(0),)
|
|
assert await second.process(bytes(1024)) == (SpeechStarted(0),)
|
|
await first.reset()
|
|
assert await second.process(bytes(1024)) == (SpeechEnded(900, "silence"),)
|
|
assert iterators[0] is not iterators[1]
|
|
assert iterators[1].resets == 0
|
|
finally:
|
|
await first.aclose()
|
|
await second.aclose()
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_model_failure_requires_reset_before_accepting_more_audio(monkeypatch):
|
|
class Iterator(ScriptedIterator):
|
|
failed = False
|
|
|
|
def __call__(self, samples, *, return_seconds):
|
|
if not self.failed:
|
|
self.failed = True
|
|
raise RuntimeError("Model failed")
|
|
return super().__call__(samples, return_seconds=return_seconds)
|
|
|
|
install_iterator(monkeypatch, Iterator([{"start": 0}]))
|
|
detector = SileroVADAdapter()
|
|
try:
|
|
with pytest.raises(RuntimeError, match="Model failed"):
|
|
await detector.process(bytes(1024))
|
|
with pytest.raises(RuntimeError, match="requires reset"):
|
|
await detector.process(bytes(1024))
|
|
await detector.reset()
|
|
assert await detector.process(bytes(1024)) == (SpeechStarted(0),)
|
|
finally:
|
|
await detector.aclose()
|
|
|
|
|
|
@pytest.mark.anyio
|
|
@pytest.mark.parametrize("recovery", ["reset", "aclose"])
|
|
async def test_cancelled_inference_is_joined_before_reset_or_close(monkeypatch, recovery):
|
|
started = threading.Event()
|
|
release = threading.Event()
|
|
busy = threading.Event()
|
|
|
|
class Iterator(ScriptedIterator):
|
|
def __call__(self, samples, *, return_seconds):
|
|
busy.set()
|
|
started.set()
|
|
try:
|
|
assert release.wait(timeout=2.0)
|
|
return super().__call__(samples, return_seconds=return_seconds)
|
|
finally:
|
|
busy.clear()
|
|
|
|
def reset_states(self):
|
|
assert not busy.is_set(), "Reset raced with active inference"
|
|
super().reset_states()
|
|
|
|
iterator = Iterator([{"start": 0}])
|
|
install_iterator(monkeypatch, iterator)
|
|
detector = SileroVADAdapter()
|
|
task = asyncio.create_task(detector.process(bytes(1024)))
|
|
recovery_task = None
|
|
try:
|
|
assert await asyncio.to_thread(started.wait, 1.0)
|
|
task.cancel()
|
|
with pytest.raises(asyncio.CancelledError):
|
|
await task
|
|
recovery_task = asyncio.create_task(getattr(detector, recovery)())
|
|
await asyncio.sleep(0)
|
|
assert not recovery_task.done()
|
|
assert busy.is_set()
|
|
release.set()
|
|
await asyncio.wait_for(recovery_task, timeout=1.0)
|
|
assert not busy.is_set()
|
|
if recovery == "reset":
|
|
assert await detector.process(bytes(1024)) == (SpeechStarted(0),)
|
|
else:
|
|
with pytest.raises(RuntimeError):
|
|
await detector.process(bytes(1024))
|
|
finally:
|
|
release.set()
|
|
await asyncio.gather(task, return_exceptions=True)
|
|
if recovery_task is not None:
|
|
await asyncio.gather(recovery_task, return_exceptions=True)
|
|
await detector.aclose()
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_cancelled_waiting_call_does_not_cancel_running_inference(monkeypatch):
|
|
started = threading.Event()
|
|
release = threading.Event()
|
|
|
|
class Iterator(ScriptedIterator):
|
|
def __call__(self, samples, *, return_seconds):
|
|
started.set()
|
|
assert release.wait(timeout=2.0)
|
|
return super().__call__(samples, return_seconds=return_seconds)
|
|
|
|
install_iterator(monkeypatch, Iterator([{"start": 0}, {"end": 900}]))
|
|
detector = SileroVADAdapter()
|
|
first = asyncio.create_task(detector.process(bytes(1024)))
|
|
second = None
|
|
try:
|
|
assert await asyncio.to_thread(started.wait, 1.0)
|
|
second = asyncio.create_task(detector.process(bytes(1024)))
|
|
await asyncio.sleep(0)
|
|
second.cancel()
|
|
with pytest.raises(asyncio.CancelledError):
|
|
await second
|
|
release.set()
|
|
assert await asyncio.wait_for(first, timeout=1.0) == (SpeechStarted(0),)
|
|
assert await detector.process(bytes(1024)) == (SpeechEnded(900, "silence"),)
|
|
finally:
|
|
release.set()
|
|
await asyncio.gather(first, return_exceptions=True)
|
|
if second is not None:
|
|
await asyncio.gather(second, return_exceptions=True)
|
|
await detector.aclose()
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_worker_timeout_preserves_ownership_until_safe_cleanup(monkeypatch):
|
|
started = threading.Event()
|
|
release = threading.Event()
|
|
finished = threading.Event()
|
|
|
|
class Iterator(ScriptedIterator):
|
|
def __call__(self, samples, *, return_seconds):
|
|
started.set()
|
|
try:
|
|
assert release.wait(timeout=2.0)
|
|
return super().__call__(samples, return_seconds=return_seconds)
|
|
finally:
|
|
finished.set()
|
|
|
|
install_iterator(monkeypatch, Iterator([{"start": 0}]))
|
|
detector = SileroVADAdapter(worker_timeout_seconds=0.05)
|
|
try:
|
|
with pytest.raises(TimeoutError):
|
|
await detector.process(bytes(1024))
|
|
assert started.is_set()
|
|
assert not finished.is_set()
|
|
with pytest.raises(TimeoutError):
|
|
await detector.aclose()
|
|
assert not finished.is_set()
|
|
release.set()
|
|
assert await asyncio.to_thread(finished.wait, 1.0)
|
|
await detector.aclose()
|
|
with pytest.raises(RuntimeError):
|
|
await detector.process(bytes(1024))
|
|
finally:
|
|
release.set()
|
|
assert await asyncio.to_thread(finished.wait, 1.0)
|
|
await detector.aclose()
|