Trixy7/plugins/tts_piper/main.py

# -*- coding: utf-8 -*-
"""
Piper TTS Plugin - Text-to-Speech mit Piper.

Piper ist ein schnelles, lokales TTS-System basierend auf VITS.

Unterstützt:
- Lokale Synthese (offline)
- Sehr schnell (Echtzeit auf CPU)
- Hochwertige Stimmen
- Mehrere Sprachen
- Auto-Download von Hugging Face
"""

import asyncio
import io
import time
import wave
from pathlib import Path
from typing import Any, AsyncIterator

from trixy_core.plugins import TrixyPlugin
from trixy_core.audio.tts import TTSProvider, TTSConfig, TTSResult, TTSState, Voice


# Piper Voice Katalog mit Hugging Face Pfaden
PIPER_VOICES = {
    # Deutsche Stimmen
    "de_DE-thorsten-high": {
        "name": "Thorsten (High)",
        "language": "de-DE",
        "gender": "male",
        "description": "Deutsche männliche Stimme, hohe Qualität",
        "sample_rate": 22050,
        "hf_path": "de/de_DE/thorsten/high/de_DE-thorsten-high.onnx",
    },
    "de_DE-thorsten-medium": {
        "name": "Thorsten (Medium)",
        "language": "de-DE",
        "gender": "male",
        "description": "Deutsche männliche Stimme, mittlere Qualität",
        "sample_rate": 22050,
        "hf_path": "de/de_DE/thorsten/medium/de_DE-thorsten-medium.onnx",
    },
    "de_DE-thorsten-low": {
        "name": "Thorsten (Low)",
        "language": "de-DE",
        "gender": "male",
        "description": "Deutsche männliche Stimme, niedrige Qualität (schnell)",
        "sample_rate": 22050,
        "hf_path": "de/de_DE/thorsten/low/de_DE-thorsten-low.onnx",
    },
    "de_DE-eva_k-x_low": {
        "name": "Eva",
        "language": "de-DE",
        "gender": "female",
        "description": "Deutsche weibliche Stimme",
        "sample_rate": 22050,
        "hf_path": "de/de_DE/eva_k/x_low/de_DE-eva_k-x_low.onnx",
    },
    # Englische Stimmen
    "en_US-lessac-medium": {
        "name": "Lessac (Medium)",
        "language": "en-US",
        "gender": "female",
        "description": "Englische weibliche Stimme (US)",
        "sample_rate": 22050,
        "hf_path": "en/en_US/lessac/medium/en_US-lessac-medium.onnx",
    },
    "en_US-amy-medium": {
        "name": "Amy (Medium)",
        "language": "en-US",
        "gender": "female",
        "description": "Englische weibliche Stimme (US)",
        "sample_rate": 22050,
        "hf_path": "en/en_US/amy/medium/en_US-amy-medium.onnx",
    },
    "en_GB-alan-medium": {
        "name": "Alan (Medium)",
        "language": "en-GB",
        "gender": "male",
        "description": "Englische männliche Stimme (UK)",
        "sample_rate": 22050,
        "hf_path": "en/en_GB/alan/medium/en_GB-alan-medium.onnx",
    },
}

HUGGINGFACE_REPO = "rhasspy/piper-voices"


class PiperTTSProvider(TTSProvider):
    """Piper-basierter TTS-Provider."""

    def __init__(
        self,
        config: TTSConfig | None = None,
        models_dir: Path | None = None,
        voice_id: str = "de_DE-thorsten-medium",
        auto_download: bool = True,
        event_manager: Any = None,
    ):
        super().__init__(config)
        self._models_dir = models_dir
        self._voice_id = voice_id
        self._auto_download = auto_download
        self._event_manager = event_manager
        self._piper = None
        self._model_path: Path | None = None

    @property
    def name(self) -> str:
        return "piper"

    @property
    def supported_languages(self) -> list[str]:
        return ["de-DE", "en-US", "en-GB", "fr-FR", "es-ES", "it-IT", "nl-NL", "pl-PL"]

    @property
    def supports_streaming(self) -> bool:
        return True

    async def initialize(self) -> None:
        """Lädt das Piper-Modell."""
        try:
            from piper import PiperVoice

            self._state = TTSState.SYNTHESIZING

            # Modell-Pfad bestimmen
            model_file = f"{self._voice_id}.onnx"
            json_file = f"{self._voice_id}.onnx.json"
            self._model_path = self._models_dir / model_file
            json_path = self._models_dir / json_file

            # Auto-Download wenn nicht vorhanden
            if not self._model_path.exists():
                if self._auto_download:
                    await self._download_model()
                else:
                    raise RuntimeError(
                        f"Piper-Modell nicht gefunden: {self._model_path}. "
                        "Auto-Download ist deaktiviert."
                    )

            # Piper laden (in Thread)
            loop = asyncio.get_event_loop()
            self._piper = await loop.run_in_executor(
                None,
                lambda: PiperVoice.load(str(self._model_path))
            )

            # Aktuelle Stimme setzen
            if self._voice_id in PIPER_VOICES:
                voice_info = PIPER_VOICES[self._voice_id]
                self._current_voice = Voice(
                    id=self._voice_id,
                    name=voice_info["name"],
                    language=voice_info["language"],
                    gender=voice_info["gender"],
                    description=voice_info["description"],
                    sample_rate=voice_info["sample_rate"],
                )

            self._model_loaded = True
            self._state = TTSState.READY

        except ImportError:
            raise RuntimeError(
                "Piper nicht installiert. "
                "Installieren mit: pip install piper-tts"
            )
        except Exception as e:
            self._state = TTSState.ERROR
            raise RuntimeError(f"Fehler beim Laden des Piper-Modells: {e}")

    async def _download_model(self) -> None:
        """Lädt das Modell von Hugging Face herunter."""
        from trixy_core.utils.debug import pinfo
        from trixy_core.utils.download import download_from_huggingface

        if self._voice_id not in PIPER_VOICES:
            raise RuntimeError(
                f"Unbekannte Stimme: {self._voice_id}. "
                f"Verfügbar: {', '.join(PIPER_VOICES.keys())}"
            )

        voice_info = PIPER_VOICES[self._voice_id]
        hf_path = voice_info["hf_path"]
        hf_json_path = hf_path + ".json"

        self._models_dir.mkdir(parents=True, exist_ok=True)

        # ONNX-Modell herunterladen
        pinfo(f"Lade Piper-Modell: {self._voice_id}...")

        model_dest = self._models_dir / f"{self._voice_id}.onnx"
        success = await download_from_huggingface(
            repo_id=HUGGINGFACE_REPO,
            filename=hf_path,
            dest_path=model_dest,
            event_manager=self._event_manager,
            download_id=f"piper-{self._voice_id}",
        )

        if not success:
            raise RuntimeError(f"Download des Modells fehlgeschlagen: {self._voice_id}")

        # JSON-Config herunterladen
        json_dest = self._models_dir / f"{self._voice_id}.onnx.json"
        await download_from_huggingface(
            repo_id=HUGGINGFACE_REPO,
            filename=hf_json_path,
            dest_path=json_dest,
            event_manager=self._event_manager,
            download_id=f"piper-{self._voice_id}-json",
        )

        pinfo(f"Piper-Modell heruntergeladen: {self._voice_id}")

    async def shutdown(self) -> None:
        """Gibt Ressourcen frei."""
        self._piper = None
        self._model_loaded = False
        self._state = TTSState.UNINITIALIZED

    async def get_voices(self, language: str | None = None) -> list[Voice]:
        """Gibt verfügbare Stimmen zurück."""
        voices = []
        for voice_id, info in PIPER_VOICES.items():
            voices.append(Voice(
                id=voice_id,
                name=info["name"],
                language=info["language"],
                gender=info["gender"],
                description=info["description"],
                sample_rate=info["sample_rate"],
            ))

        if language:
            lang_prefix = language.replace("-", "_").split("_")[0]
            voices = [v for v in voices if v.language.replace("-", "_").startswith(lang_prefix)]

        return voices

    async def synthesize(
        self,
        text: str,
        voice_id: str | None = None,
        language: str | None = None,
    ) -> TTSResult:
        """Synthetisiert Text mit Piper."""
        if not self._piper:
            raise RuntimeError("Piper nicht geladen")

        self._state = TTSState.SYNTHESIZING
        start_time = time.time()

        try:
            # Audio-Buffer
            audio_buffer = io.BytesIO()

            # WAV-Writer
            with wave.open(audio_buffer, "wb") as wav_file:
                wav_file.setnchannels(1)
                wav_file.setsampwidth(2)  # 16-bit
                wav_file.setframerate(self._piper.config.sample_rate)

                # Synthese in Thread
                loop = asyncio.get_event_loop()

                def do_synthesize():
                    for audio_chunk in self._piper.synthesize(text):
                        wav_file.writeframes(audio_chunk.audio_int16_bytes)

                await loop.run_in_executor(None, do_synthesize)

            processing_time = (time.time() - start_time) * 1000
            audio_data = audio_buffer.getvalue()

            self._state = TTSState.READY

            # Dauer berechnen
            sample_rate = self._piper.config.sample_rate
            # WAV hat 44 bytes Header
            pcm_data = audio_data[44:]
            num_samples = len(pcm_data) // 2  # 16-bit = 2 bytes
            duration = num_samples / sample_rate

            return TTSResult(
                audio_data=audio_data,
                sample_rate=sample_rate,
                channels=1,
                text=text,
                duration_seconds=duration,
                processing_time_ms=processing_time,
                voice_id=voice_id or self._voice_id,
                voice_name=self._current_voice.name if self._current_voice else "",
                language=language or self._config.language,
                provider="piper",
                model=self._voice_id,
            )

        except Exception as e:
            self._state = TTSState.ERROR
            raise RuntimeError(f"Piper-Synthese fehlgeschlagen: {e}")

    async def synthesize_stream(
        self,
        text: str,
        voice_id: str | None = None,
        language: str | None = None,
        chunk_size: int = 4096,
    ) -> AsyncIterator[bytes]:
        """Synthetisiert Text als Stream."""
        if not self._piper:
            raise RuntimeError("Piper nicht geladen")

        self._state = TTSState.SYNTHESIZING

        try:
            loop = asyncio.get_event_loop()

            # Generator für Chunks
            def generate_chunks():
                for audio_chunk in self._piper.synthesize(text):
                    audio_bytes = audio_chunk.audio_int16_bytes
                    # In Chunks aufteilen
                    for i in range(0, len(audio_bytes), chunk_size):
                        yield audio_bytes[i:i + chunk_size]

            # Chunks yielden
            chunks = await loop.run_in_executor(None, lambda: list(generate_chunks()))

            for chunk in chunks:
                yield chunk

            self._state = TTSState.READY

        except Exception as e:
            self._state = TTSState.ERROR
            raise RuntimeError(f"Piper-Stream fehlgeschlagen: {e}")


class PiperTTSPlugin(TrixyPlugin):
    """Piper TTS Plugin für Trixy."""

    def __init__(self, application, plugin_path, config: dict | None = None):
        super().__init__(application, plugin_path, config)
        self._provider: PiperTTSProvider | None = None

    async def on_load(self) -> None:
        """Plugin wird geladen."""
        from trixy_core.utils.debug import pinfo
        pinfo("Piper TTS Plugin: Lade...")

        # Konfiguration
        voice_id = self.config.get("voice", "de_DE-thorsten-medium")
        language = self.config.get("language", "de-DE")
        auto_download = self.config.get("auto_download", True)

        # Models-Verzeichnis im Plugin-Ordner
        models_dir = self.plugin_path / "models"
        models_dir.mkdir(parents=True, exist_ok=True)

        tts_config = TTSConfig(
            language=language,
            voice_id=voice_id,
        )

        # Provider erstellen (mit EventManager für Download-Events)
        self._provider = PiperTTSProvider(
            tts_config,
            models_dir=models_dir,
            voice_id=voice_id,
            auto_download=auto_download,
            event_manager=self.application.events,
        )

        # Modell laden (und ggf. downloaden)
        await self._provider.initialize()

        # Extension registrieren
        if hasattr(self.application, "extension_points"):
            ext_point = self.application.extension_points.get("conversation.tts")
            if ext_point:
                ext_point.register(self._provider)
                pinfo("Piper TTS Plugin: Extension registriert")

        # Event-Handler registrieren
        self._register_event_handlers()

        pinfo(f"Piper TTS Plugin: Geladen (Stimme: {voice_id})")

    def _register_event_handlers(self) -> None:
        """Registriert Event-Handler."""
        em = self.application.events

        @em.on("tts_request")
        async def on_tts_request(event_name: str, data: dict) -> None:
            """Verarbeitet TTS-Anfragen."""
            if not self._provider or not self._provider.is_ready:
                return

            text = data.get("text")
            satellite_id = data.get("satellite_id")
            session_id = data.get("session_id")
            voice_id = data.get("voice_id")

            if not text:
                return

            from trixy_core.utils.debug import pinfo
            pinfo(f"Piper TTS: Synthetisiere '{text[:50]}...'")

            try:
                result = await self._provider.synthesize(text, voice_id=voice_id)

                await em.emit("tts_completed", {
                    "audio_data": result.audio_data.hex(),
                    "sample_rate": result.sample_rate,
                    "duration_seconds": result.duration_seconds,
                    "provider": "piper",
                    "text": text,
                    "satellite_id": satellite_id,
                    "session_id": session_id,
                    "processing_time_ms": result.processing_time_ms,
                })

                pinfo(f"Piper TTS: Fertig ({result.duration_seconds:.1f}s Audio, "
                      f"{result.processing_time_ms:.0f}ms)")

            except Exception as e:
                from trixy_core.utils.debug import perror
                perror(f"Piper TTS Fehler: {e}")

                await em.emit("tts_error", {
                    "error": str(e),
                    "provider": "piper",
                    "text": text,
                    "satellite_id": satellite_id,
                    "session_id": session_id,
                })

    async def on_unload(self) -> None:
        """Plugin wird entladen."""
        if self._provider:
            await self._provider.shutdown()
            self._provider = None

        from trixy_core.utils.debug import pinfo
        pinfo("Piper TTS Plugin: Entladen")

    @property
    def provider(self) -> PiperTTSProvider | None:
        """TTS-Provider."""
        return self._provider


# Plugin-Export
Plugin = PiperTTSPlugin