49 lines
1.4 KiB
Python
49 lines
1.4 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Optional
|
|
|
|
from config import vad_bundle_path
|
|
|
|
|
|
class VADService:
|
|
def __init__(self) -> None:
|
|
self._model = None
|
|
self._iterator = None
|
|
self._ready = False
|
|
self._attempted = False
|
|
self._error: Optional[str] = None
|
|
|
|
def _ensure_loaded(self) -> None:
|
|
if self._ready or self._attempted:
|
|
return
|
|
self._attempted = True
|
|
path = vad_bundle_path().resolve()
|
|
if not path.is_file():
|
|
self._error = f"missing Silero VAD weights: {path}"
|
|
return
|
|
try:
|
|
from silero_vad import VADIterator
|
|
from silero_vad.utils_vad import init_jit_model
|
|
|
|
self._model = init_jit_model(str(path))
|
|
self._iterator = VADIterator(
|
|
self._model,
|
|
threshold=0.65,
|
|
sampling_rate=16000,
|
|
min_silence_duration_ms=450,
|
|
speech_pad_ms=80,
|
|
)
|
|
self._ready = True
|
|
except Exception as exc: # pragma: no cover
|
|
self._error = str(exc)
|
|
|
|
def push_chunk(self, pcm_chunk):
|
|
self._ensure_loaded()
|
|
if self._ready and self._iterator is not None:
|
|
return self._iterator(pcm_chunk)
|
|
return None
|
|
|
|
@property
|
|
def health(self) -> dict:
|
|
return {"ready": self._ready, "attempted": self._attempted, "error": self._error}
|