from __future__ import annotations import asyncio import audioop import io import logging import os import random import wave from collections.abc import AsyncIterable from collections.abc import Iterable from realtime_voice_service.providers.base import BaseTTS LOGGER = logging.getLogger("uvicorn.error") DEFAULT_FILLER_TEXTS: dict[str, tuple[str, ...]] = { "serper": ( "Проверяю информацию.", "Сейчас подскажу.", ), "generic": ( "Сейчас отвечу.", "Уже отвечаю.", ), } class FillerAudioLibrary: def __init__( self, *, sample_rate_hz: int = 16000, config_env: str = "REALTIME_VOICE_FILLER_AUDIO_FILES", enabled_env: str = "REALTIME_VOICE_FILLER_AUDIO_ENABLED", ) -> None: self._sample_rate_hz = max(int(sample_rate_hz), 1) self._config_env = config_env self._enabled = self._read_bool_env(enabled_env, True) self._clips: dict[str, list[bytes]] = {} self._preload_lock = asyncio.Lock() self._preloaded = False self._random = random.Random() async def preload(self, *, tts: BaseTTS | None = None) -> None: if not self._enabled or self._preloaded: LOGGER.info( "filler audio preload skipped: enabled=%s preloaded=%s clips=%s", self._enabled, self._preloaded, {key: len(value) for key, value in self._clips.items()}, ) return async with self._preload_lock: if self._preloaded: return LOGGER.info("filler audio preload start: sample_rate=%s", self._sample_rate_hz) self._load_from_files() if tts is not None: await self._synthesize_missing_fillers(tts) self._preloaded = True LOGGER.info( "filler audio preload done: clips=%s", {key: len(value) for key, value in self._clips.items()}, ) def pick(self, key: str | None = None, *, allow_fallback: bool = True) -> bytes | None: if not self._enabled: return None normalized_key = (key or "").strip().lower() or "generic" candidates = self._clips.get(normalized_key) or [] if not candidates and allow_fallback: candidates = self._clips.get("generic") or [] if not candidates: LOGGER.info("filler audio pick missed: key=%s available=%s", normalized_key, list(self._clips)) return None clip = self._random.choice(candidates) LOGGER.info( "filler audio picked: key=%s candidates=%s bytes=%s", normalized_key, len(candidates), len(clip), ) return clip def add_clip(self, key: str, clip: bytes) -> None: if not clip: return normalized_key = (key or "").strip().lower() or "generic" self._clips.setdefault(normalized_key, []).append(clip) LOGGER.info( "filler audio clip added: key=%s total=%s bytes=%s", normalized_key, len(self._clips[normalized_key]), len(clip), ) async def synthesize_text(self, tts: BaseTTS, text: str) -> bytes: return await self._synthesize_clip(tts, text) async def synthesize_segments( self, tts: BaseTTS, segments: Iterable[tuple[str, str | None]], ) -> bytes: clip = bytearray() for i, (text, language_code) in enumerate(segments): LOGGER.info( "synthesize_segments: starting segment %s language=%s text=%r", i, language_code, text[:60], ) segment = await self._synthesize_clip(tts, text, language_code=language_code) if segment: clip.extend(segment) LOGGER.info( "synthesize_segments: completed segment %s bytes=%s total_bytes=%s", i, len(segment), len(clip), ) else: LOGGER.warning( "synthesize_segments: segment %s produced NO audio; language=%s", i, language_code, ) return bytes(clip) def _load_from_files(self) -> None: raw_config = str(os.getenv(self._config_env, "")).strip() if not raw_config: return for raw_entry in raw_config.split(","): entry = raw_entry.strip() if not entry: continue clip_key = "generic" clip_path = entry if "=" in entry: maybe_key, maybe_path = entry.split("=", 1) clip_key = maybe_key.strip().lower() or "generic" clip_path = maybe_path.strip() if not clip_path: continue try: clip = self._load_clip_from_path(clip_path) except Exception: LOGGER.exception("failed to preload filler audio clip path=%s key=%s", clip_path, clip_key) continue self._clips.setdefault(clip_key, []).append(clip) LOGGER.info( "filler audio file loaded: key=%s path=%s bytes=%s", clip_key, clip_path, len(clip), ) async def _synthesize_missing_fillers(self, tts: BaseTTS) -> None: for key, texts in DEFAULT_FILLER_TEXTS.items(): if self._clips.get(key): continue for text in texts: try: LOGGER.info("filler audio synth start: key=%s text=%r", key, text) clip = await self._synthesize_clip(tts, text) except Exception: LOGGER.exception("failed to synthesize filler clip key=%s", key) break if clip: self._clips.setdefault(key, []).append(clip) LOGGER.info( "filler audio synth done: key=%s text=%r bytes=%s", key, text, len(clip), ) if not self._clips.get(key): LOGGER.warning("no filler clips available for key=%s", key) async def _synthesize_clip(self, tts: BaseTTS, text: str, *, language_code: str | None = None) -> bytes: async def one_shot_text_stream() -> AsyncIterable[str]: yield text clip = bytearray() async for audio_chunk in tts.synthesize_stream(one_shot_text_stream(), language_code=language_code): if audio_chunk: clip.extend(audio_chunk) return bytes(clip) def _load_clip_from_path(self, clip_path: str) -> bytes: with open(clip_path, "rb") as handle: payload = handle.read() if clip_path.lower().endswith(".wav"): return self._decode_wav(payload) return payload def _decode_wav(self, payload: bytes) -> bytes: with wave.open(io.BytesIO(payload), "rb") as wav_file: pcm_bytes = wav_file.readframes(wav_file.getnframes()) sample_width = wav_file.getsampwidth() channels = wav_file.getnchannels() sample_rate_hz = int(wav_file.getframerate() or self._sample_rate_hz) if sample_width != 2: raise ValueError("filler audio WAV must be PCM16") if channels == 2: pcm_bytes = audioop.tomono(pcm_bytes, sample_width, 0.5, 0.5) if sample_rate_hz == self._sample_rate_hz: return pcm_bytes converted, _ = audioop.ratecv( pcm_bytes, 2, 1, sample_rate_hz, self._sample_rate_hz, None, ) return converted @staticmethod def _read_bool_env(name: str, default: bool) -> bool: raw = os.getenv(name) if raw is None: return default return str(raw).strip().lower() in {"1", "true", "yes", "on"}