278 lines
9.9 KiB
Python
278 lines
9.9 KiB
Python
import os
|
||
import re
|
||
import hashlib
|
||
import asyncio
|
||
import logging
|
||
from pathlib import Path
|
||
from typing import Optional, Dict, Any, List
|
||
|
||
from app import config
|
||
|
||
logger = logging.getLogger("notes_app.tts")
|
||
|
||
PIPER_DIR = config.BASE_DIR / "data" / "piper"
|
||
PIPER_BIN = PIPER_DIR / "piper"
|
||
MODELS_DIR = PIPER_DIR / "models"
|
||
AUDIO_CACHE_DIR = config.BASE_DIR / "data" / "audio_cache"
|
||
|
||
# Ensure audio cache directory exists
|
||
AUDIO_CACHE_DIR.mkdir(parents=True, exist_ok=True)
|
||
|
||
# Curated high-fidelity Piper ONNX models for English and Russian
|
||
VOICE_MODELS: Dict[str, Dict[str, Any]] = {
|
||
# --- English Voices ---
|
||
"en_US-lessac": {
|
||
"id": "en_US-lessac",
|
||
"lang": "en",
|
||
"gender": "female",
|
||
"title": "Paula (US • Женский, спокойный)",
|
||
"model": MODELS_DIR / "en_US-lessac-medium.onnx",
|
||
"config": MODELS_DIR / "en_US-lessac-medium.onnx.json",
|
||
"base_length_scale": 1.0,
|
||
},
|
||
"en_US-ryan": {
|
||
"id": "en_US-ryan",
|
||
"lang": "en",
|
||
"gender": "male",
|
||
"title": "Ryan (US • Мужской, естественный)",
|
||
"model": MODELS_DIR / "en_US-ryan-medium.onnx",
|
||
"config": MODELS_DIR / "en_US-ryan-medium.onnx.json",
|
||
"base_length_scale": 1.0,
|
||
},
|
||
"en_GB-alan": {
|
||
"id": "en_GB-alan",
|
||
"lang": "en",
|
||
"gender": "male",
|
||
"title": "Alan (UK • Британский акцент)",
|
||
"model": MODELS_DIR / "en_GB-alan-medium.onnx",
|
||
"config": MODELS_DIR / "en_GB-alan-medium.onnx.json",
|
||
"base_length_scale": 1.0,
|
||
},
|
||
|
||
# --- Russian Voices ---
|
||
"ru_RU-irina": {
|
||
"id": "ru_RU-irina",
|
||
"lang": "ru",
|
||
"gender": "female",
|
||
"title": "Ирина (Женский, студийный, плавный)",
|
||
"model": MODELS_DIR / "ru_RU-irina-medium.onnx",
|
||
"config": MODELS_DIR / "ru_RU-irina-medium.onnx.json",
|
||
"base_length_scale": 1.0,
|
||
},
|
||
"ru_RU-denis": {
|
||
"id": "ru_RU-denis",
|
||
"lang": "ru",
|
||
"gender": "male",
|
||
"title": "Денис (Мужской, умеренный)",
|
||
"model": MODELS_DIR / "ru_RU-denis-medium.onnx",
|
||
"config": MODELS_DIR / "ru_RU-denis-medium.onnx.json",
|
||
"base_length_scale": 1.08,
|
||
},
|
||
"ru_RU-dmitri": {
|
||
"id": "ru_RU-dmitri",
|
||
"lang": "ru",
|
||
"gender": "male",
|
||
"title": "Дмитрий (Мужской, энергичный)",
|
||
"model": MODELS_DIR / "ru_RU-dmitri-medium.onnx",
|
||
"config": MODELS_DIR / "ru_RU-dmitri-medium.onnx.json",
|
||
"base_length_scale": 1.18, # Calibrated to eliminate rushing
|
||
},
|
||
}
|
||
|
||
DEFAULT_VOICES = {
|
||
"en": "en_US-lessac",
|
||
"ru": "ru_RU-irina", # Irina provides natural pacing and clear punctuation pauses
|
||
}
|
||
|
||
|
||
def clean_tts_text(text: str, lang: str = "en") -> str:
|
||
"""
|
||
Cleans input text from markdown, HTML tags, abnormal whitespace and
|
||
normalizes punctuation pauses so the engine doesn't swallow commas or dashes.
|
||
"""
|
||
if not text:
|
||
return ""
|
||
# Remove HTML tags if any
|
||
t = re.sub(r"<[^>]+>", " ", text)
|
||
# Remove markdown link formatting [text](url) -> text
|
||
t = re.sub(r"\[([^\]]+)\]\([^)]+\)", r"\1", t)
|
||
|
||
# Normalize quotes
|
||
t = t.replace("«", '"').replace("»", '"').replace("“", '"').replace("”", '"')
|
||
|
||
# Normalize em-dashes and en-dashes for natural acoustic pauses:
|
||
# Em-dash (—) and en-dash (–) are not in Piper's phoneme map, so eSpeak discards them without pause.
|
||
# Replacing surrounded dashes with a comma-pause ensures an intonational breath pause.
|
||
t = re.sub(r"\s*[—–]\s*", ", — ", t)
|
||
t = re.sub(r"\s*--\s*", ", — ", t)
|
||
|
||
# Fix punctuation without trailing space: "слово,слово" -> "слово, слово"
|
||
t = re.sub(r"([,;:!?])([^\s0-9])", r"\1 \2", t)
|
||
|
||
# Collapse multiple whitespaces
|
||
t = " ".join(t.strip().split())
|
||
return t
|
||
|
||
|
||
def resolve_voice(lang: str = "en", voice: Optional[str] = None) -> Dict[str, Any]:
|
||
"""Resolves requested voice ID or falls back to default voice for language."""
|
||
norm_lang = "ru" if lang.lower().startswith("ru") else "en"
|
||
if voice and voice in VOICE_MODELS:
|
||
candidate = VOICE_MODELS[voice]
|
||
if candidate["lang"] == norm_lang:
|
||
return candidate
|
||
def_id = DEFAULT_VOICES.get(norm_lang, "en_US-lessac")
|
||
return VOICE_MODELS.get(def_id, list(VOICE_MODELS.values())[0])
|
||
|
||
|
||
def get_audio_hash(text: str, lang: str = "en", speed: float = 1.0, voice: Optional[str] = None) -> str:
|
||
"""Computes a deterministic MD5 hash for audio caching, accounting for voice, speed and text."""
|
||
clean = clean_tts_text(text, lang).lower()
|
||
norm_speed = round(float(speed), 2)
|
||
norm_lang = "ru" if lang.lower().startswith("ru") else "en"
|
||
v_info = resolve_voice(norm_lang, voice)
|
||
v_id = v_info["id"]
|
||
key = f"{norm_lang}:{v_id}:{norm_speed:.2f}:{clean}".encode("utf-8")
|
||
return hashlib.md5(key).hexdigest()
|
||
|
||
|
||
def get_cached_audio_path(text: str, lang: str = "en", speed: float = 1.0, voice: Optional[str] = None) -> Optional[Path]:
|
||
"""Returns Path if already cached and valid, else None."""
|
||
h = get_audio_hash(text, lang, speed, voice)
|
||
file_path = AUDIO_CACHE_DIR / f"{h}.wav"
|
||
if file_path.is_file() and file_path.stat().st_size > 100:
|
||
return file_path
|
||
return None
|
||
|
||
|
||
def is_piper_available() -> bool:
|
||
"""Checks if piper binary and at least one model exist and are executable."""
|
||
if not PIPER_BIN.is_file() or not os.access(PIPER_BIN, os.X_OK):
|
||
return False
|
||
return any(v["model"].is_file() for v in VOICE_MODELS.values())
|
||
|
||
|
||
def get_available_voices() -> Dict[str, Any]:
|
||
"""Returns metadata for all available models and active default voices."""
|
||
available = []
|
||
for k, v in VOICE_MODELS.items():
|
||
if v["model"].is_file():
|
||
available.append({
|
||
"id": v["id"],
|
||
"lang": v["lang"],
|
||
"gender": v["gender"],
|
||
"title": v["title"],
|
||
"is_default": (v["id"] == DEFAULT_VOICES.get(v["lang"])),
|
||
})
|
||
return {
|
||
"available": bool(available),
|
||
"engine": "piper-vits",
|
||
"default_voices": DEFAULT_VOICES,
|
||
"voices": available,
|
||
}
|
||
|
||
|
||
async def synthesize_speech(
|
||
text: str,
|
||
lang: str = "en",
|
||
speed: float = 1.0,
|
||
voice: Optional[str] = None,
|
||
) -> Optional[Path]:
|
||
"""
|
||
Synthesizes speech using local Piper TTS and returns Path to the generated WAV file.
|
||
Fast disk-cached: if already generated, returns immediately without running synthesis.
|
||
"""
|
||
clean = clean_tts_text(text, lang)
|
||
if not clean:
|
||
return None
|
||
|
||
norm_lang = "ru" if lang.lower().startswith("ru") else "en"
|
||
voice_info = resolve_voice(norm_lang, voice)
|
||
v_id = voice_info["id"]
|
||
|
||
# Check cache first
|
||
cached_path = get_cached_audio_path(clean, norm_lang, speed, v_id)
|
||
if cached_path:
|
||
return cached_path
|
||
|
||
if not is_piper_available():
|
||
logger.warning("Piper TTS binary or models not available")
|
||
return None
|
||
|
||
model_path = voice_info["model"]
|
||
config_path = voice_info["config"]
|
||
base_length_scale = voice_info.get("base_length_scale", 1.0)
|
||
|
||
if not model_path.is_file() or not config_path.is_file():
|
||
logger.warning(f"Voice model files missing for {v_id}: {model_path}")
|
||
return None
|
||
|
||
h = get_audio_hash(clean, norm_lang, speed, v_id)
|
||
out_wav = AUDIO_CACHE_DIR / f"{h}.wav"
|
||
temp_wav = AUDIO_CACHE_DIR / f"{h}.tmp.wav"
|
||
|
||
clamped_speed = max(0.5, min(2.0, float(speed)))
|
||
# Apply voice-specific base length scale calibration
|
||
eff_length_scale = base_length_scale * (1.0 / clamped_speed)
|
||
length_scale_str = str(round(eff_length_scale, 2))
|
||
|
||
env = os.environ.copy()
|
||
ld_path = str(PIPER_DIR)
|
||
if "LD_LIBRARY_PATH" in env and env["LD_LIBRARY_PATH"]:
|
||
env["LD_LIBRARY_PATH"] = f"{ld_path}:{env['LD_LIBRARY_PATH']}"
|
||
else:
|
||
env["LD_LIBRARY_PATH"] = ld_path
|
||
|
||
cmd = [
|
||
str(PIPER_BIN),
|
||
"--model", str(model_path),
|
||
"--config", str(config_path),
|
||
"--espeak_data", str(PIPER_DIR / "espeak-ng-data"),
|
||
"--length_scale", length_scale_str,
|
||
"--sentence_silence", "0.35",
|
||
"--output_file", str(temp_wav),
|
||
]
|
||
|
||
try:
|
||
proc = await asyncio.create_subprocess_exec(
|
||
*cmd,
|
||
stdin=asyncio.subprocess.PIPE,
|
||
stdout=asyncio.subprocess.PIPE,
|
||
stderr=asyncio.subprocess.PIPE,
|
||
env=env,
|
||
)
|
||
stdout, stderr = await proc.communicate(input=clean.encode("utf-8"))
|
||
|
||
if proc.returncode == 0 and temp_wav.is_file() and temp_wav.stat().st_size > 100:
|
||
temp_wav.replace(out_wav)
|
||
logger.info(f"Synthesized speech [{v_id}, speed={speed}]: {clean[:40]}... -> {out_wav.name}")
|
||
return out_wav
|
||
else:
|
||
err_text = stderr.decode("utf-8", errors="replace").strip()
|
||
logger.warning(f"Piper exited code {proc.returncode}: {err_text}")
|
||
if temp_wav.exists():
|
||
temp_wav.unlink(missing_ok=True)
|
||
except Exception as e:
|
||
logger.warning(f"Piper execution error: {e}")
|
||
if temp_wav.exists():
|
||
temp_wav.unlink(missing_ok=True)
|
||
|
||
return None
|
||
|
||
|
||
async def prewarm_card_audio(term: str, context: Optional[str] = None, voice: Optional[str] = None):
|
||
"""
|
||
Background pre-synthesis of word and context sentence.
|
||
Invoked when a user saves a card to ensure zero-latency audio playback in SRS.
|
||
"""
|
||
try:
|
||
if term and term.strip():
|
||
await synthesize_speech(term.strip(), lang="en", speed=1.0, voice=voice)
|
||
if context and context.strip():
|
||
clean_ctx = context.strip().replace('"', '').strip()
|
||
if len(clean_ctx) > 300:
|
||
clean_ctx = clean_ctx[:300]
|
||
await synthesize_speech(clean_ctx, lang="en", speed=1.0, voice=voice)
|
||
except Exception as e:
|
||
logger.warning(f"Background prewarm error: {e}")
|