42 lines
1.2 KiB
Python
42 lines
1.2 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Optional
|
|
|
|
|
|
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
|
|
try:
|
|
from silero_vad import VADIterator, load_silero_vad
|
|
|
|
self._model = load_silero_vad()
|
|
self._iterator = VADIterator(
|
|
self._model,
|
|
threshold=0.5,
|
|
sampling_rate=16000,
|
|
min_silence_duration_ms=300,
|
|
speech_pad_ms=30,
|
|
)
|
|
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}
|