207 lines
8.9 KiB
Python
207 lines
8.9 KiB
Python
"""HandsFreeRecorder — свободный голосовой режим БЕЗ клавиш (М5.2).
|
|
|
|
Постоянно слушает микрофон; решение «речь/тишина» принимает Silero-VAD:
|
|
- началась речь → копим аудио (с pre-buffer 0.4 c, чтобы не терять первые слова);
|
|
- тишина silence_sec (по умолчанию 2.0 — «пауза 2 секунды» юзера) → фраза
|
|
закончена → вызов self.on_phrase(audio, samplerate);
|
|
- короче min_speech_sec — шум, отбраковка (событие on_state("discard"));
|
|
- длиннее max_seconds — принудительный срез;
|
|
- на время ответа ассистента прослушивание ставится на паузу (suppress),
|
|
чтобы микрофон не слышал её из динамиков.
|
|
|
|
Колбэки — публичные атрибуты: их можно переприсвоить и извне
|
|
(Assistant присваивает recorder.on_phrase = ...).
|
|
|
|
Фильтр шума: голосом считается окно, у которого prob VAD выше порога И
|
|
RMS выше noise_gate (Silero на белом шуме даёт ~0.85 — одной вероятности мало).
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import time
|
|
import warnings
|
|
from typing import Callable, List, Optional
|
|
|
|
import numpy as np
|
|
import sounddevice as sd
|
|
|
|
from .resample import TARGET_RATE, resample_to_16k
|
|
|
|
_VAD_WINDOW = 512 # 32 мс при 16 кГц — родной размер окна Silero
|
|
_PRE_BUFFER_SEC = 0.4 # буфер ДО начала речи (не терять начало фразы)
|
|
|
|
|
|
class HandsFreeRecorder:
|
|
def __init__(
|
|
self,
|
|
samplerate: int = TARGET_RATE,
|
|
device: Optional[int] = None,
|
|
silence_sec: float = 2.0,
|
|
speech_threshold: float = 0.5,
|
|
min_speech_sec: float = 0.3,
|
|
max_seconds: float = 30.0,
|
|
noise_gate_rms: float = 0.006, # ниже RMS — окно считается тишиной, как бы VAD ни хотел
|
|
on_phrase: Optional[Callable[[np.ndarray, int], None]] = None,
|
|
on_state: Optional[Callable[[str], None]] = None, # listening|speech_start|speech_end|too_short|paused
|
|
on_error: Optional[Callable[[str], None]] = None,
|
|
debug: bool = False,
|
|
) -> None:
|
|
self._sr = samplerate
|
|
self._device = device
|
|
self._silence_sec = silence_sec
|
|
self._threshold = speech_threshold
|
|
self._min_speech_sec = min_speech_sec
|
|
self._max_seconds = max_seconds
|
|
self._noise_gate = noise_gate_rms
|
|
|
|
# Публичные колбэки: Assistant переприсваивает on_phrase — так и задумано
|
|
self.on_phrase: Callable[[np.ndarray, int], None] = on_phrase or (lambda a, s: None)
|
|
self.on_state: Callable[[str], None] = on_state or (lambda s: None)
|
|
self.on_error: Callable[[str], None] = on_error or (lambda m: print(f" [VAD] {m}"))
|
|
|
|
self._vad = None
|
|
self._stream = None
|
|
self._running = False
|
|
self._suppress = False
|
|
|
|
self._recording = False
|
|
self._chunks: List[np.ndarray] = []
|
|
self._pre_buffer: List[np.ndarray] = []
|
|
self._speech_frames = 0
|
|
self._silence_frames = 0
|
|
self._native_sr = samplerate
|
|
self._carry: np.ndarray = np.zeros(0, dtype=np.float32) # хвост между колбэками
|
|
|
|
# ---------------------------------------------------------------- lifecycle
|
|
def start(self) -> None:
|
|
"""Открыть микрофон и слушать. Ошибки уходят в on_error (не роняют программу)."""
|
|
try:
|
|
warnings.filterwarnings("ignore", category=Warning)
|
|
from silero_vad import load_silero_vad
|
|
|
|
self._vad = load_silero_vad()
|
|
stream = self._open(self._sr)
|
|
if stream is None:
|
|
self._native_sr = int(
|
|
sd.query_devices(self._device, "input")["default_samplerate"]
|
|
)
|
|
stream = self._open(self._native_sr, required=True)
|
|
else:
|
|
self._native_sr = self._sr
|
|
stream.start()
|
|
self._stream = stream
|
|
self._running = True
|
|
self.on_state("listening")
|
|
except Exception as exc:
|
|
self.on_error(f"Свободный режим не запустился: {exc}")
|
|
|
|
def stop(self) -> None:
|
|
self._running = False
|
|
stream, self._stream = self._stream, None
|
|
if stream is not None:
|
|
try:
|
|
stream.stop()
|
|
stream.close()
|
|
except Exception:
|
|
pass
|
|
|
|
# ---------------------------------------------------------------- режим
|
|
@property
|
|
def is_listening(self) -> bool:
|
|
return self._running and not self._suppress
|
|
|
|
def suppress(self, on: bool) -> None:
|
|
"""Пауза прослушивания (пока сами говорим) + сброс недозаписи."""
|
|
if self._suppress == on:
|
|
return
|
|
self._suppress = on
|
|
self.on_state("paused" if on else "listening")
|
|
if on:
|
|
self._recording = False
|
|
self._chunks = []
|
|
self._pre_buffer = []
|
|
self._carry = np.zeros(0, dtype=np.float32)
|
|
|
|
# ---------------------------------------------------------------- внутреннее
|
|
def _open(self, samplerate: int, required: bool = False):
|
|
try:
|
|
return sd.InputStream(
|
|
samplerate=samplerate,
|
|
device=self._device,
|
|
channels=1,
|
|
dtype="float32",
|
|
blocksize=512 if samplerate == 16000 else 0,
|
|
callback=self._on_audio,
|
|
)
|
|
except Exception as exc:
|
|
if required:
|
|
raise RuntimeError(f"Не удалось открыть микрофон: {exc}") from exc
|
|
return None
|
|
|
|
def _vad_prob(self, window: np.ndarray) -> float:
|
|
"""Уверенность VAD для окна (0..1). JIT-модель требует torch.Tensor."""
|
|
import torch
|
|
with torch.no_grad():
|
|
return float(self._vad(torch.from_numpy(window), 16000))
|
|
|
|
def _on_audio(self, indata, frames, time_info, status) -> None: # noqa: ANN001
|
|
if status:
|
|
self.on_error(str(status))
|
|
if self._suppress or self._vad is None:
|
|
return
|
|
|
|
mono = indata[:, 0].copy()
|
|
if self._native_sr != 16000:
|
|
mono16 = resample_to_16k(mono, self._native_sr)
|
|
else:
|
|
mono16 = mono
|
|
|
|
# накопитель между колбэками: окно 512 всегда полное, хвост не теряется
|
|
buf = np.concatenate([self._carry, mono16])
|
|
n = len(buf) // _VAD_WINDOW
|
|
for i in range(n):
|
|
self._process_window(buf[i * _VAD_WINDOW:(i + 1) * _VAD_WINDOW])
|
|
self._carry = buf[n * _VAD_WINDOW:]
|
|
|
|
def _process_window(self, window: np.ndarray) -> None:
|
|
rms = float(np.sqrt(np.mean(np.square(window))))
|
|
voice = self._vad_prob(window) > self._threshold and rms > self._noise_gate
|
|
|
|
if voice:
|
|
if not self._recording:
|
|
self._recording = True
|
|
self._chunks = list(self._pre_buffer) if self._pre_buffer else []
|
|
self._pre_buffer = []
|
|
self._speech_frames = 0
|
|
self.on_state("speech_start")
|
|
self._chunks.append(window)
|
|
self._speech_frames += len(window)
|
|
self._silence_frames = 0
|
|
if self._speech_frames >= int(self._max_seconds * 16000):
|
|
self._finalize()
|
|
return
|
|
|
|
if self._recording:
|
|
self._chunks.append(window)
|
|
self._silence_frames += len(window)
|
|
if self._silence_frames >= int(self._silence_sec * 16000):
|
|
self._finalize()
|
|
else:
|
|
self._pre_buffer.append(window)
|
|
if len(self._pre_buffer) > int(_PRE_BUFFER_SEC * 16000 / _VAD_WINDOW):
|
|
self._pre_buffer.pop(0)
|
|
|
|
def _finalize(self) -> None:
|
|
chunks, self._chunks = self._chunks, []
|
|
self._recording = False
|
|
self._pre_buffer = []
|
|
speech_samples = self._speech_frames
|
|
self._speech_frames = 0
|
|
self._silence_frames = 0
|
|
|
|
total = sum(len(c) for c in chunks)
|
|
# Валидна именно ДОЛЯ голоса: суммарный буфер всегда ≥ паузе тишины
|
|
if speech_samples < int(self._min_speech_sec * 16000):
|
|
self.on_state("too_short") # видимая отбраковка, не тихая
|
|
return
|
|
self.on_state("speech_end")
|
|
self.on_phrase(np.concatenate(chunks).astype(np.float32), 16000) |