feat: VAD ingestion buffer — Silero VAD, per-room accumulation
TDD: 3 tests covering silence passthrough, speech→silence segment emission, and short-speech discard. Uses _speech_ms tracking (not total buffer length) for accurate min_speech_ms enforcement. Silero VAD import is try/except'd so tests run without torch via mocker.patch on vad_prob. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,55 @@
|
||||
import numpy as np
|
||||
import pytest
|
||||
from datetime import datetime, timezone
|
||||
from unittest.mock import patch, MagicMock
|
||||
|
||||
|
||||
def _make_chunk(samples: int = 4000, amplitude: float = 0.0) -> bytes:
|
||||
"""Create a fake PCM int16 chunk."""
|
||||
audio = (np.ones(samples) * amplitude * 32767).astype(np.int16)
|
||||
return audio.tobytes()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def buf(mocker):
|
||||
"""RoomVADBuffer with mocked VAD model."""
|
||||
mocker.patch(
|
||||
"app.pipeline.ingestion.vad_prob",
|
||||
side_effect=lambda chunk: 0.9 if np.frombuffer(chunk, np.int16).max() > 100 else 0.1,
|
||||
)
|
||||
from app.pipeline.ingestion import RoomVADBuffer
|
||||
return RoomVADBuffer(silence_threshold_ms=600, min_speech_ms=500, chunk_ms=250)
|
||||
|
||||
|
||||
def test_silence_produces_no_segment(buf):
|
||||
ts = datetime(2026, 1, 1, tzinfo=timezone.utc)
|
||||
for _ in range(10):
|
||||
result = buf.process_chunk(_make_chunk(amplitude=0.0), ts)
|
||||
assert result is None
|
||||
|
||||
|
||||
def test_speech_then_silence_produces_segment(buf):
|
||||
ts = datetime(2026, 1, 1, tzinfo=timezone.utc)
|
||||
# 4 speech chunks = 1 second of speech
|
||||
for _ in range(4):
|
||||
result = buf.process_chunk(_make_chunk(amplitude=0.5), ts)
|
||||
assert result is None # still accumulating
|
||||
# 3 silence chunks = 750ms silence (> 600ms threshold)
|
||||
segment = None
|
||||
for _ in range(3):
|
||||
segment = buf.process_chunk(_make_chunk(amplitude=0.0), ts)
|
||||
assert segment is not None
|
||||
assert isinstance(segment, tuple)
|
||||
audio_bytes, start_time = segment
|
||||
assert len(audio_bytes) > 0
|
||||
|
||||
|
||||
def test_short_speech_discarded(buf):
|
||||
ts = datetime(2026, 1, 1, tzinfo=timezone.utc)
|
||||
# 1 speech chunk = 250ms (below 500ms min)
|
||||
buf.process_chunk(_make_chunk(amplitude=0.5), ts)
|
||||
# Now silence to trigger emission
|
||||
result = None
|
||||
for _ in range(3):
|
||||
result = buf.process_chunk(_make_chunk(amplitude=0.0), ts)
|
||||
assert result is None # discarded as too short
|
||||
Reference in New Issue
Block a user