89 lines
2.8 KiB
Python
Executable File
89 lines
2.8 KiB
Python
Executable File
"""Voice activity detection."""
|
|
from typing import Optional
|
|
import math
|
|
import array
|
|
|
|
|
|
class SileroVad:
|
|
"""Voice activity detection with silero VAD."""
|
|
|
|
def __init__(self, threshold: float, trigger_level: int) -> None:
|
|
from pysilero_vad import SileroVoiceActivityDetector
|
|
|
|
self.detector = SileroVoiceActivityDetector()
|
|
self.threshold = threshold
|
|
self.trigger_level = trigger_level
|
|
self._activation = 0
|
|
|
|
def __call__(self, audio_bytes: Optional[bytes]) -> bool:
|
|
if audio_bytes is None:
|
|
# Reset
|
|
self._activation = 0
|
|
self.detector.reset()
|
|
return False
|
|
|
|
if self.detector(audio_bytes) >= self.threshold:
|
|
# Speech detected
|
|
self._activation += 1
|
|
if self._activation >= self.trigger_level:
|
|
self._activation = 0
|
|
return True
|
|
else:
|
|
# Silence detected
|
|
self._activation = max(0, self._activation - 1)
|
|
|
|
return False
|
|
|
|
|
|
class WebRtcVadEnergy:
|
|
"""Voice activity detection using webrtcvad + RMS energy threshold.
|
|
|
|
Expects 16-bit mono PCM at 16Khz. Chunks are internally processed in 20ms windows.
|
|
"""
|
|
|
|
def __init__(self, trigger_level: int, energy_threshold: Optional[float] = None) -> None:
|
|
import webrtcvad # type: ignore
|
|
|
|
self.vad = webrtcvad.Vad(3)
|
|
self.trigger_level = trigger_level
|
|
self.energy_threshold = energy_threshold if energy_threshold is not None else 200.0
|
|
self._activation = 0
|
|
|
|
# 20ms @16kHz, 16-bit mono
|
|
self._samples_per_window = 320
|
|
self._bytes_per_window = self._samples_per_window * 2
|
|
|
|
def __call__(self, audio_bytes: Optional[bytes]) -> bool:
|
|
if audio_bytes is None:
|
|
# Reset
|
|
self._activation = 0
|
|
return False
|
|
|
|
# Process in 20ms windows as required by webrtcvad
|
|
for i in range(0, len(audio_bytes) - (len(audio_bytes) % self._bytes_per_window), self._bytes_per_window):
|
|
window = audio_bytes[i : i + self._bytes_per_window]
|
|
|
|
# RMS energy
|
|
pcm = array.array("h")
|
|
pcm.frombytes(window)
|
|
if len(pcm) == 0:
|
|
continue
|
|
rms = math.sqrt(sum(int(s) * int(s) for s in pcm) / len(pcm))
|
|
|
|
is_speech = False
|
|
try:
|
|
is_speech = self.vad.is_speech(window, 16000) and (rms >= float(self.energy_threshold))
|
|
except Exception:
|
|
# If VAD errors on malformed window, skip
|
|
is_speech = False
|
|
|
|
if is_speech:
|
|
self._activation += 1
|
|
if self._activation >= self.trigger_level:
|
|
self._activation = 0
|
|
return True
|
|
else:
|
|
self._activation = max(0, self._activation - 1)
|
|
|
|
return False
|