my_voice_assistant_v2/app/providers/tts/piper.py

"""Lokales TTS über das piper-Binary (CPU-freundlich, niedrige Latenz, offline).

piper synthetisiert rohes s16le-Mono-PCM in der Sample-Rate des Stimmmodells. Das
Gateway erwartet 24000 Hz -> bei Abweichung wird mit ffmpeg resampelt. So bleibt eine
voll-lokale Konstellation (STT + LLM + TTS lokal) möglich (Datenschutz, keine API-Kosten).
"""

from __future__ import annotations

import asyncio
import io
import json
import shutil
import wave
from pathlib import Path

from app.providers.tts.base import TTSProvider


def _wrap_wav(pcm: bytes, sample_rate: int) -> bytes:
    buf = io.BytesIO()
    with wave.open(buf, "wb") as w:
        w.setnchannels(1)
        w.setsampwidth(2)
        w.setframerate(sample_rate)
        w.writeframes(pcm)
    return buf.getvalue()


class PiperTTSProvider(TTSProvider):
    def __init__(
        self,
        bin_path: str = "piper",
        voices_dir: str = "",
        voice: str = "",
        target_rate: int = 24000,
    ):
        self.bin_path = (bin_path or "piper").strip()
        self.voices_dir = Path(voices_dir).expanduser() if voices_dir else Path.cwd()
        self.voice = (voice or "").strip()
        self.target_rate = int(target_rate)

    def _model_paths(self, voice: str | None) -> tuple[Path, Path]:
        # Angeforderte Stimme zuerst; passt sie nicht (z. B. eine Cloud-Stimme wie
        # "Zephyr"/"alloy" aus der Route), auf die konfigurierte Default-Stimme zurueckfallen.
        if not (voice or self.voice or "").strip():
            raise ValueError("Piper-Stimme ist nicht gesetzt (PIPER_VOICE)")
        tried: list[str] = []
        for candidate in (voice, self.voice):
            name = (candidate or "").strip()
            if not name or name in tried:
                continue
            tried.append(name)
            model = Path(name).expanduser()
            if not model.suffix:  # nur ein Name -> im Voices-Verzeichnis suchen
                model = self.voices_dir / f"{name}.onnx"
            if model.exists():
                return model, Path(f"{model}.json")
        raise RuntimeError(
            f"Piper-Stimmmodell nicht gefunden (versucht: {', '.join(tried)}) in {self.voices_dir}"
        )

    def _native_rate(self, config: Path) -> int:
        try:
            with open(config) as fh:
                return int(json.load(fh)["audio"]["sample_rate"])
        except (OSError, KeyError, ValueError, TypeError):
            return self.target_rate  # Konfig unlesbar -> Resampling überspringen

    async def synthesize(
        self,
        text: str,
        voice: str | None = None,
        audio_format: str = "pcm",
    ) -> bytes:
        if not text or not text.strip():
            raise ValueError("TTS input text is empty")
        if not (shutil.which(self.bin_path) or Path(self.bin_path).exists()):
            raise RuntimeError(f"Piper-Binary nicht gefunden: {self.bin_path}")

        model, config = self._model_paths(voice)
        native_rate = self._native_rate(config)

        pcm = await self._run_piper(model, config, text.strip())
        if native_rate != self.target_rate:
            pcm = await self._resample(pcm, native_rate, self.target_rate)

        if audio_format == "wav":
            return _wrap_wav(pcm, self.target_rate)
        return pcm

    async def _run_piper(self, model: Path, config: Path, text: str) -> bytes:
        cmd = [self.bin_path, "-m", str(model), "-c", str(config), "--output-raw"]
        proc = await asyncio.create_subprocess_exec(
            *cmd,
            stdin=asyncio.subprocess.PIPE,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
        )
        out, err = await proc.communicate(text.encode("utf-8"))
        if proc.returncode != 0:
            detail = err.decode("utf-8", "replace").strip()[:300]
            raise RuntimeError(f"Piper-Fehler ({proc.returncode}): {detail}")
        if not out:
            raise RuntimeError("Piper lieferte kein Audio")
        return out

    async def _resample(self, pcm: bytes, src_rate: int, dst_rate: int) -> bytes:
        if not shutil.which("ffmpeg"):
            raise RuntimeError(
                f"ffmpeg fehlt: Piper-Stimme liefert {src_rate} Hz, benötigt werden {dst_rate} Hz"
            )
        cmd = [
            "ffmpeg", "-loglevel", "quiet",
            "-f", "s16le", "-ar", str(src_rate), "-ac", "1", "-i", "pipe:0",
            "-f", "s16le", "-ar", str(dst_rate), "-ac", "1", "pipe:1",
        ]
        proc = await asyncio.create_subprocess_exec(
            *cmd,
            stdin=asyncio.subprocess.PIPE,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
        )
        out, err = await proc.communicate(pcm)
        if proc.returncode != 0 or not out:
            detail = err.decode("utf-8", "replace").strip()[:300]
            raise RuntimeError(f"ffmpeg-Resampling fehlgeschlagen: {detail}")
        return out