my_voice_assistant_v3/app/providers/tts/chatterbox.py

"""TTS über den lokalen Chatterbox-HTTP-Service (Resemble AI, hohe Qualitaet + Voice-Cloning).

Architektur wie beim llama.cpp-LLM: ein separater Dienst (eigene Conda-Env, GPU) haelt
das Modell geladen; dieser Provider ruft ihn per HTTP auf. Der Dienst ist job-basiert:
POST /speak -> /status pollen -> GET /audio/{id} (WAV).

Wichtig: `no_playback=true` -> der Dienst spielt NICHT lokal ab, sondern liefert nur die
WAV (fuer Remote-/Gateway-Nutzung). Chatterbox ist neural und ~Echtzeit langsam -> als
QUALITAETS-Provider gedacht (piper bleibt der schnelle Default).
"""

from __future__ import annotations

import asyncio
import io
import time
import wave
from pathlib import Path

import httpx

from app.providers.tts.base import TTSProvider
from app.providers.tts.piper import _resample, _wrap_wav


class ChatterboxTTSProvider(TTSProvider):
    def __init__(
        self,
        base_url: str = "http://127.0.0.1:9999",
        voice: str = "",
        lang: str = "de",
        speed: float = 1.0,
        target_rate: int = 24000,
        timeout: int = 180,
        voices_dir: str = "",
    ):
        self.base_url = base_url.rstrip("/")
        self.voice = (voice or "").strip()  # Pfad zu Referenz-WAV (Voice-Cloning) oder leer
        self.lang = lang
        self.speed = speed
        self.target_rate = int(target_rate)
        self.timeout = timeout
        # Absolut auflösen: der Chatterbox-Dienst liest den Pfad mit eigenem CWD.
        self.voices_dir = Path(voices_dir).expanduser().resolve() if voices_dir else None

    def _lang_ref(self, lang: str) -> str | None:
        """Native Referenz-WAV der Sprache (Konvention <voices_dir>/<lang>.wav), falls vorhanden."""
        if not self.voices_dir:
            return None
        path = self.voices_dir / f"{lang}.wav"
        return str(path) if path.exists() else None

    def _ref_voice(self, voice: str | None, lang: str) -> str | None:
        # 1. Explizit angefragte Stimme nur, wenn sie wie ein WAV-Pfad aussieht
        #    (Cloud-Stimmennamen wie "Zephyr"/"alloy" ignorieren).
        cand = (voice or "").strip()
        if cand.endswith(".wav"):
            return cand
        # 2. Native Stimme der jeweiligen Sprache, sonst die konfigurierte Default-Stimme.
        return self._lang_ref(lang) or self.voice or None

    # Chatterbox ist mehrsprachig und klont die Stimme cross-lingual: die Referenz-WAV
    # liefert das Timbre, `lang` die Aussprache. So spricht die Klon-Stimme jede Sprache.
    _LANG_ALIASES = {"cmn": "zh"}

    async def synthesize(
        self,
        text: str,
        voice: str | None = None,
        audio_format: str = "pcm",
        language: str | None = None,
    ) -> bytes:
        if not text or not text.strip():
            raise ValueError("TTS input text is empty")

        # Gesprächssprache bevorzugen; ohne Angabe der konfigurierte Default.
        lang = (language or self.lang or "de").strip()
        lang = self._LANG_ALIASES.get(lang, lang)

        payload = {
            "text": text.strip(),
            "lang": lang,
            "speed": self.speed,
            "keep_audio": True,
            "no_playback": True,
        }
        ref = self._ref_voice(voice, lang)
        if ref:
            payload["voice"] = ref

        async with httpx.AsyncClient(timeout=self.timeout) as client:
            resp = await client.post(f"{self.base_url}/speak", json=payload)
            resp.raise_for_status()
            job_id = resp.json()["job_id"]
            wav_bytes = await self._await_audio(client, job_id)

        pcm, rate = self._wav_to_pcm(wav_bytes)
        if rate != self.target_rate:
            pcm = await asyncio.to_thread(_resample, pcm, rate, self.target_rate)
        if audio_format == "wav":
            return _wrap_wav(pcm, self.target_rate)
        return pcm

    async def _await_audio(self, client: httpx.AsyncClient, job_id: str) -> bytes:
        """Wartet (via /status) bis der Job fertig ist und laedt dann die WAV."""
        deadline = time.monotonic() + self.timeout
        while True:
            status = await client.get(f"{self.base_url}/status")
            status.raise_for_status()
            match = next(
                (j for j in status.json().get("recent_jobs", []) if j["id"] == job_id),
                None,
            )
            if match:
                if match["status"] == "done":
                    break
                raise RuntimeError(
                    f"Chatterbox-Job {match['status']}: {match.get('error') or ''}"
                )
            if time.monotonic() > deadline:
                raise RuntimeError("Chatterbox-Timeout (Job nicht rechtzeitig fertig)")
            await asyncio.sleep(0.3)

        audio = await client.get(f"{self.base_url}/audio/{job_id}")
        if audio.status_code != 200 or not audio.content:
            raise RuntimeError(f"Chatterbox-Audio {audio.status_code}: {audio.text[:200]}")
        return audio.content

    @staticmethod
    def _wav_to_pcm(wav_bytes: bytes) -> tuple[bytes, int]:
        with wave.open(io.BytesIO(wav_bytes)) as w:
            rate = w.getframerate()
            frames = w.readframes(w.getnframes())
        return frames, rate