Antigravity/notes-qpid-pro
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(tts): multi-voice switcher for EN & RU, Irina default voice, punctuation pause normalization and speed calibration

Browse Source
This commit is contained in:
Antigravity
2026-10-07 12:07:01 +03:00
parent 7893b353aa
commit 31e105b2c2
11 changed files with 486 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/tts.py
+137 -34
View File
@@ -4,7 +4,7 @@ import hashlib
import asyncio
import logging
from pathlib import Path
from typing import Optional, Dict, Any
from typing import Optional, Dict, Any, List
from app import config
@@ -18,45 +18,127 @@ AUDIO_CACHE_DIR = config.BASE_DIR / "data" / "audio_cache"
# Ensure audio cache directory exists
AUDIO_CACHE_DIR.mkdir(parents=True, exist_ok=True)
VOICE_MODELS = {
"en": {
# 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",
"name": "en_US-lessac",
"base_length_scale": 1.0,
},
"ru": {
"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",
"name": "ru_RU-dmitri",
"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) -> str:
"""Cleans input text from markdown, HTML tags, and abnormal whitespace."""
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 get_audio_hash(text: str, lang: str = "en", speed: float = 1.0) -> str:
"""Computes a deterministic MD5 hash for audio caching."""
clean = clean_tts_text(text).lower()
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"
key = f"{norm_lang}:{norm_speed:.2f}:{clean}".encode("utf-8")
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) -> Optional[Path]:
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)
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
@@ -64,28 +146,52 @@ def get_cached_audio_path(text: str, lang: str = "en", speed: float = 1.0) -> Op
def is_piper_available() -> bool:
"""Checks if piper binary and at least en model exist and are executable."""
"""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
en_model = VOICE_MODELS["en"]["model"]
return en_model.is_file()
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)
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, lang, speed)
cached_path = get_cached_audio_path(clean, norm_lang, speed, v_id)
if cached_path:
return cached_path
@@ -93,25 +199,22 @@ async def synthesize_speech(
logger.warning("Piper TTS binary or models not available")
return None
norm_lang = "ru" if lang.lower().startswith("ru") else "en"
voice_info = VOICE_MODELS.get(norm_lang, VOICE_MODELS["en"])
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 {norm_lang}: {model_path}")
logger.warning(f"Voice model files missing for {v_id}: {model_path}")
return None
h = get_audio_hash(clean, norm_lang, speed)
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"
# Piper length_scale: > 1.0 is slower, < 1.0 is faster
# 0.75x speed -> length_scale ~ 1.33
# 1.0x speed -> length_scale = 1.0
# 1.25x speed -> length_scale ~ 0.8
clamped_speed = max(0.5, min(2.0, float(speed)))
length_scale = str(round(1.0 / clamped_speed, 2))
# 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)
@@ -125,7 +228,8 @@ async def synthesize_speech(
"--model", str(model_path),
"--config", str(config_path),
"--espeak_data", str(PIPER_DIR / "espeak-ng-data"),
"--length_scale", length_scale,
"--length_scale", length_scale_str,
"--sentence_silence", "0.35",
"--output_file", str(temp_wav),
]
@@ -141,7 +245,7 @@ async def synthesize_speech(
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 [{norm_lang}, speed={speed}]: {clean[:40]}... -> {out_wav.name}")
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()
@@ -156,19 +260,18 @@ async def synthesize_speech(
return None
async def prewarm_card_audio(term: str, context: Optional[str] = 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)
await synthesize_speech(term.strip(), lang="en", speed=1.0, voice=voice)
if context and context.strip():
clean_ctx = context.strip().replace('"', '').strip()
# If context is very long, cap to first sentence or 300 chars
if len(clean_ctx) > 300:
clean_ctx = clean_ctx[:300]
await synthesize_speech(clean_ctx, lang="en", speed=1.0)
await synthesize_speech(clean_ctx, lang="en", speed=1.0, voice=voice)
except Exception as e:
logger.warning(f"Background prewarm error: {e}")