feat(tts): integrate autonomous Piper neural TTS with zero-latency pre-warming for words and context

app/tts.py

@@ -0,0 +1,174 @@
+import os
+import re
+import hashlib
+import asyncio
+import logging
+from pathlib import Path
+from typing import Optional, Dict, Any
+
+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)
+
+VOICE_MODELS = {
+    "en": {
+        "model": MODELS_DIR / "en_US-lessac-medium.onnx",
+        "config": MODELS_DIR / "en_US-lessac-medium.onnx.json",
+        "name": "en_US-lessac",
+    },
+    "ru": {
+        "model": MODELS_DIR / "ru_RU-dmitri-medium.onnx",
+        "config": MODELS_DIR / "ru_RU-dmitri-medium.onnx.json",
+        "name": "ru_RU-dmitri",
+    },
+}
+
+
+def clean_tts_text(text: str) -> str:
+    """Cleans input text from markdown, HTML tags, and abnormal whitespace."""
+    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)
+    # 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()
+    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")
+    return hashlib.md5(key).hexdigest()
+
+
+def get_cached_audio_path(text: str, lang: str = "en", speed: float = 1.0) -> Optional[Path]:
+    """Returns Path if already cached and valid, else None."""
+    h = get_audio_hash(text, lang, speed)
+    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 en 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()
+
+
+async def synthesize_speech(
+    text: str,
+    lang: str = "en",
+    speed: float = 1.0,
+) -> 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)
+    if not clean:
+        return None
+
+    # Check cache first
+    cached_path = get_cached_audio_path(clean, lang, speed)
+    if cached_path:
+        return cached_path
+
+    if not is_piper_available():
+        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"]
+
+    if not model_path.is_file() or not config_path.is_file():
+        logger.warning(f"Voice model files missing for {norm_lang}: {model_path}")
+        return None
+
+    h = get_audio_hash(clean, norm_lang, speed)
+    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))
+
+    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,
+        "--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 [{norm_lang}, 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):
+    """
+    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)
+        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)
+    except Exception as e:
+        logger.warning(f"Background prewarm error: {e}")