145 lines
5.8 KiB
Python
145 lines
5.8 KiB
Python
"""Recorder — push-to-talk запись в RAM.
|
|
|
|
Схема работы в будущем ассистенте:
|
|
клавиша «Слушай» нажата → recorder.start()
|
|
клавиша отпущена → audio = recorder.stop() → STT
|
|
|
|
Особенности:
|
|
- пробует открыть вход на 16 кГц; если устройство не умеет — пишет в нативной
|
|
частоте и ресемплирует в 16 кГц при stop() (нужно STT);
|
|
- поток в фоновом потоке PortAudio, данные копируются в список блоков;
|
|
- слишком короткие записи (< min_seconds) возвращаются пустыми — защита от
|
|
случайных кликов;
|
|
- потоко-безопасность через threading.Lock.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import threading
|
|
import time
|
|
from typing import Callable, List, Optional
|
|
|
|
import numpy as np
|
|
import sounddevice as sd
|
|
|
|
from .resample import TARGET_RATE, resample_to_16k
|
|
|
|
|
|
class Recorder:
|
|
def __init__(
|
|
self,
|
|
samplerate: int = TARGET_RATE,
|
|
device: Optional[int] = None,
|
|
channels: int = 1,
|
|
blocksize: int = 800, # 50 мс при 16 кГц
|
|
max_seconds: float = 60.0, # защита от «забытой» клавиши
|
|
min_seconds: float = 0.3, # короче — считаем пустой записью
|
|
on_overrun: Optional[Callable[[str], None]] = None, # колбэк о проблемах
|
|
) -> None:
|
|
self._target_sr = samplerate
|
|
self._device = device
|
|
self._channels = channels
|
|
self._blocksize = blocksize
|
|
self._max_seconds = max_seconds
|
|
self._min_seconds = min_seconds
|
|
self._on_overrun = on_overrun
|
|
|
|
self._frames: List[np.ndarray] = []
|
|
self._stream: Optional[sd.InputStream] = None
|
|
self._native_sr: int = samplerate
|
|
self._started_at: float = 0.0
|
|
self._lock = threading.Lock()
|
|
|
|
# --- состояние -------------------------------------------------------
|
|
@property
|
|
def is_recording(self) -> bool:
|
|
return self._stream is not None and self._stream.active
|
|
|
|
@property
|
|
def duration(self) -> float:
|
|
"""Длительность текущей записи в секундах (0 если не пишем)."""
|
|
if not self.is_recording:
|
|
return 0.0
|
|
return time.perf_counter() - self._started_at
|
|
|
|
@property
|
|
def native_samplerate(self) -> int:
|
|
"""Частота, на которой реально шло устройство (до ресемплинга)."""
|
|
return self._native_sr
|
|
|
|
# --- управление ------------------------------------------------------
|
|
def _callback(self, indata, frames, time_info, status) -> None: # noqa: ANN001
|
|
if status and self._on_overrun:
|
|
self._on_overrun(str(status))
|
|
self._frames.append(indata.copy())
|
|
|
|
def start(self) -> None:
|
|
"""Начать запись. Бросает RuntimeError, если запись уже идёт."""
|
|
if self.is_recording:
|
|
raise RuntimeError("Запись уже идёт")
|
|
|
|
with self._lock:
|
|
self._frames = []
|
|
|
|
stream = self._try_open(self._target_sr)
|
|
if stream is None:
|
|
# Устройство не поддерживает 16 кГц — пишем в нативной и ресемплируем при stop()
|
|
self._native_sr = int(
|
|
sd.query_devices(self._device, "input")["default_samplerate"]
|
|
)
|
|
stream = self._try_open(self._native_sr, required=True)
|
|
else:
|
|
self._native_sr = self._target_sr
|
|
|
|
stream.start()
|
|
self._stream = stream
|
|
self._started_at = time.perf_counter()
|
|
|
|
def stop(self) -> np.ndarray:
|
|
"""Остановить запись и вернуть float32 mono, 16 кГц, [-1..1]."""
|
|
stream, self._stream = self._stream, None
|
|
if stream is None:
|
|
return np.zeros(0, dtype=np.float32)
|
|
|
|
stream.stop()
|
|
stream.close()
|
|
|
|
with self._lock:
|
|
blocks = self._frames
|
|
self._frames = []
|
|
|
|
audio = np.concatenate(blocks, axis=0) if blocks else np.zeros((0, self._channels))
|
|
mono = audio.mean(axis=1) if audio.ndim > 1 else audio
|
|
mono = np.clip(mono, -1.0, 1.0).astype(np.float32)
|
|
|
|
if self._native_sr != self._target_sr:
|
|
mono = resample_to_16k(mono, self._native_sr, self._target_sr)
|
|
|
|
min_len = int(self._min_seconds * self._target_sr)
|
|
if len(mono) < min_len:
|
|
return np.zeros(0, dtype=np.float32) # слишком коротко — пусто
|
|
return mono
|
|
|
|
def cancel(self) -> None:
|
|
"""Остановить запись, выбросив данные (например, нажали «Замолчи»)."""
|
|
stream, self._stream = self._stream, None
|
|
if stream is not None:
|
|
stream.stop()
|
|
stream.close()
|
|
with self._lock:
|
|
self._frames = []
|
|
|
|
# --- внутреннее ------------------------------------------------------
|
|
def _try_open(self, samplerate: int, required: bool = False) -> Optional[sd.InputStream]:
|
|
try:
|
|
return sd.InputStream(
|
|
samplerate=samplerate,
|
|
device=self._device,
|
|
channels=self._channels,
|
|
dtype="float32",
|
|
blocksize=self._blocksize,
|
|
callback=self._callback,
|
|
)
|
|
except (sd.PortAudioError, OSError) as exc:
|
|
if required:
|
|
raise RuntimeError(f"Не удалось открыть входное устройство: {exc}") from exc
|
|
return None |