b8950b38a4
Xiaozhi WebSocket endpoint with handshake/auth, mock STT/TTS/Hermes backends, Thai chunker, barge-in queue, latency logger, GPU planner, voice-profile guard (reasoning=none, session_search only). CON-002: no GPU/audio libs loaded at import. MUST-NOT-001: all model names/tokens from config.
72 lines
2.2 KiB
Python
72 lines
2.2 KiB
Python
"""Voice-activity / end-of-speech detection (REQ-002).
|
|
|
|
Two implementations behind one interface:
|
|
|
|
* :class:`EnergyVAD` — frame-level RMS hysteresis. No model, runs anywhere.
|
|
Good enough for a PoC and for tests.
|
|
* :class:`SileroVAD` — lazy adapter for a real VAD model (webrtcvad/silero)
|
|
loaded only if selected.
|
|
|
|
The interface is binary per frame: ``active`` means "speech in this frame".
|
|
End-of-speech is derived by the session: a gap of silence >= ``min_silence_ms``
|
|
after speech ends an utterance.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import struct
|
|
from abc import ABC, abstractmethod
|
|
|
|
|
|
class VAD(ABC):
|
|
@abstractmethod
|
|
def is_speech(self, pcm: bytes) -> bool:
|
|
"""Return True if ``pcm`` (one frame) contains speech."""
|
|
|
|
|
|
class EnergyVAD(VAD):
|
|
"""RMS hysteresis VAD over 16-bit mono PCM.
|
|
|
|
``threshold`` is the RMS floor. ``rise_frames`` frames must exceed it to
|
|
trigger speech; ``fall_frames`` below it to stop — simple anti-flap.
|
|
"""
|
|
|
|
def __init__(self, threshold: float = 300.0) -> None:
|
|
self.threshold = threshold
|
|
|
|
def is_speech(self, pcm: bytes) -> bool:
|
|
samples = max(1, len(pcm) // 2)
|
|
# Unpack as many int16 samples as available (pad-safe).
|
|
n = (len(pcm) // 2) * 2
|
|
if n == 0:
|
|
return False
|
|
try:
|
|
samples = struct.unpack(f"<{n // 2}h", pcm[:n])
|
|
except struct.error:
|
|
return False
|
|
if not samples:
|
|
return False
|
|
rms = (sum(s * s for s in samples) / len(samples)) ** 0.5
|
|
return rms >= self.threshold
|
|
|
|
|
|
class SileroVAD(VAD):
|
|
"""Lazy adapter. Import fails on machines without the model installed —
|
|
that is expected and fine (CON-002); select EnergyVAD (default) there."""
|
|
|
|
def __init__(self) -> None:
|
|
# Imported at construction, not at module load.
|
|
import webrtcvad # type: ignore # noqa: F401
|
|
|
|
self._vad = webrtcvad.Vad(2)
|
|
|
|
def is_speech(self, pcm: bytes) -> bool:
|
|
return self._vad.is_speech(pcm, 16000)
|
|
|
|
|
|
def make_vad(kind: str = "energy", **kw) -> VAD:
|
|
if kind == "energy":
|
|
return EnergyVAD(**kw)
|
|
if kind == "silero":
|
|
return SileroVAD()
|
|
raise ValueError(f"unknown vad kind: {kind}")
|