85 lines
2.6 KiB
Python
85 lines
2.6 KiB
Python
from __future__ import annotations
|
|
|
|
import logging
|
|
import os
|
|
from dataclasses import dataclass
|
|
from typing import Any
|
|
|
|
import numpy as np
|
|
|
|
from speech_to_speech.pipeline.messages import TTSInput
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
_TRUE_VALUES = {"1", "true", "yes", "on"}
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class DelayFillerConfig:
|
|
enabled: bool
|
|
threshold_s: float
|
|
short_phrases: list[str]
|
|
full_phrases: list[str]
|
|
|
|
|
|
def _split_env_list(name: str) -> list[str]:
|
|
raw = os.environ.get(name, "").strip()
|
|
return [part.strip() for part in raw.split("|") if part.strip()]
|
|
|
|
|
|
def _env_flag(name: str, *, default: bool) -> bool:
|
|
raw = os.environ.get(name)
|
|
if raw is None:
|
|
return default
|
|
return raw.strip().lower() in _TRUE_VALUES
|
|
|
|
|
|
def load_delay_filler_config() -> DelayFillerConfig:
|
|
short_phrases = _split_env_list("TTS_DELAY_FILLERS_SHORT_ES")
|
|
full_phrases = _split_env_list("TTS_DELAY_FILLERS_FULL_ES")
|
|
|
|
# Backward compatibility with the old single-list variable.
|
|
if not short_phrases and not full_phrases:
|
|
full_phrases = _split_env_list("TTS_DELAY_FILLERS_ES")
|
|
|
|
enabled = _env_flag("TTS_DELAY_FILLERS_ENABLED", default=bool(short_phrases or full_phrases))
|
|
threshold_ms_raw = os.environ.get("TTS_DELAY_FILLERS_THRESHOLD_MS", "1200").strip() or "1200"
|
|
try:
|
|
threshold_ms = max(0, int(threshold_ms_raw))
|
|
except ValueError:
|
|
logger.warning("Invalid TTS_DELAY_FILLERS_THRESHOLD_MS=%r; using 1200", threshold_ms_raw)
|
|
threshold_ms = 1200
|
|
return DelayFillerConfig(
|
|
enabled=enabled and bool(short_phrases or full_phrases),
|
|
threshold_s=threshold_ms / 1000.0,
|
|
short_phrases=short_phrases,
|
|
full_phrases=full_phrases,
|
|
)
|
|
|
|
|
|
def _chunk_to_bytes(chunk: Any) -> bytes:
|
|
audio = getattr(chunk, "audio", chunk)
|
|
if isinstance(audio, np.ndarray) or hasattr(audio, "tobytes"):
|
|
return audio.tobytes()
|
|
if isinstance(audio, bytes):
|
|
return audio
|
|
raise TypeError(f"Unsupported delay filler audio chunk type: {type(audio)!r}")
|
|
|
|
|
|
def pregenerate_delay_filler_clips(tts_handler: Any, phrases: list[str]) -> list[bytes]:
|
|
clips: list[bytes] = []
|
|
for phrase in phrases:
|
|
try:
|
|
clip = b"".join(
|
|
_chunk_to_bytes(chunk)
|
|
for chunk in tts_handler.process(TTSInput(text=phrase, language_code="es"))
|
|
)
|
|
except Exception:
|
|
logger.exception("Failed to pregenerate TTS delay filler for phrase %r", phrase)
|
|
continue
|
|
if not clip:
|
|
logger.warning("Discarding empty TTS delay filler for phrase %r", phrase)
|
|
continue
|
|
clips.append(clip)
|
|
return clips
|