Trixy7/plugins/voice_morph/main.py

# -*- coding: utf-8 -*-
"""
VoiceMorph Plugin.

Morpht die TTS-Ausgabe in Echtzeit mit konfigurierbaren Presets.
Klinkt sich als HIGH-Prioritaet Event-Handler in tts_completed ein
und veraendert die audio_data bevor der IntentDispatcher (NORMAL) sie
an den Satellite sendet.

Presets koennen in der config.json definiert oder aus dem globalen
Preset-Verzeichnis (assets/default/audio_morpher/) geladen werden.
"""

from __future__ import annotations

import numpy as np

from trixy_core.plugins.trixy_plugin import TrixyPlugin
from trixy_core.events.enums import EventPriority
from trixy_core.audio_morpher.morpher import AudioMorpher
from trixy_core.audio_morpher.presets import VoiceMorphPreset
from trixy_core.nlp.decorators import intent, admin_only
from trixy_core.nlp.handler import IntentResult, IntentReceivedData
from trixy_core.utils.debug import pinfo, pdebug, perror


class VoiceMorphPlugin(TrixyPlugin):
    """TTS-Stimmveraenderung mit konfigurierbaren Presets."""

    NAME = "voice_morph"
    VERSION = "1.0.0"
    DESCRIPTION = "Morpht die TTS-Ausgabe (Pitch, Speed, Formanten, Effekte)"
    AUTHOR = "Trixy"

    def __init__(self, **kwargs) -> None:
        super().__init__(**kwargs)
        self._morpher = AudioMorpher()
        self._active_preset: VoiceMorphPreset | None = None
        self._intensity: float = 1.0
        self._handler_ref = None

    async def on_load(self) -> None:
        """Plugin laden — Presets initialisieren und Event-Handler registrieren."""
        # Eigene Presets aus config.json laden und in den Morpher einfuegen
        custom_presets = self.config.get("presets", {})
        for preset_id, preset_data in custom_presets.items():
            full_data = {"preset_id": preset_id, "label": preset_data.get("label", preset_id)}
            full_data.update(preset_data)
            preset = VoiceMorphPreset.from_dict(full_data)
            self._morpher._presets.append(preset)
            self._morpher._preset_map[preset_id] = preset

        # Intensitaet laden
        self._intensity = self.config.get("intensity", 1.0)

        # Aktives Preset setzen
        active_id = self.config.get("active_preset", "")
        if active_id:
            self._active_preset = self._morpher.get_preset(active_id)
            if self._active_preset:
                pinfo(f"[VoiceMorph] Aktives Preset: {self._active_preset.label} (ID: {active_id})")
            else:
                pdebug(f"[VoiceMorph] Preset '{active_id}' nicht gefunden — deaktiviert")

        # Event-Handler mit HIGH-Prioritaet registrieren
        # HIGH = vor IntentDispatcher (NORMAL), damit audio_data gemorph wird
        # bevor sie an den Satellite gesendet wird
        self._handler_ref = self._on_tts_completed
        self.application.events.register(
            "tts_completed",
            self._handler_ref,
            priority=EventPriority.HIGH,
        )

        preset_count = len(self._morpher.presets)
        status = self._active_preset.label if self._active_preset else "deaktiviert"
        pinfo(f"[VoiceMorph] Plugin geladen — {preset_count} Presets, Status: {status}")

    async def on_unload(self) -> None:
        """Plugin entladen — Event-Handler entfernen."""
        if self._handler_ref:
            try:
                self.application.events.unregister("tts_completed", self._handler_ref)
            except Exception:
                pass
        pdebug("[VoiceMorph] Plugin entladen")

    async def _on_tts_completed(self, event_name: str, event_data) -> None:
        """Morpht TTS-Audio in-place bevor der IntentDispatcher es sendet."""
        if not self._active_preset:
            return

        audio_hex = event_data.get("audio_data", "")
        if not audio_hex:
            return

        sample_rate = event_data.get("sample_rate", 22050)

        try:
            # Hex → Bytes → int16 numpy array
            audio_bytes = bytes.fromhex(audio_hex)
            audio = np.frombuffer(audio_bytes, dtype=np.int16).astype(np.float64)
            audio /= 32768.0  # Normalisieren auf [-1.0, 1.0]

            # Morph anwenden
            morphed = self._morpher.morph(
                audio, sample_rate, self._active_preset, self._intensity
            )

            # Zurueck in int16
            morphed = np.clip(morphed, -1.0, 1.0)
            morphed_int16 = (morphed * 32767).astype(np.int16)

            # In-place ersetzen im Event-Dict
            event_data["audio_data"] = morphed_int16.tobytes().hex()

            pdebug(f"[VoiceMorph] Audio gemorph: {len(audio)} → {len(morphed)} samples "
                   f"(Preset: {self._active_preset.preset_id})")

        except Exception as e:
            perror(f"[VoiceMorph] Morph-Fehler: {e}")

    # =========================================================================
    # Intent-Handler
    # =========================================================================

    @intent("voice_morph_enable", description="Voice Morph aktivieren")
    @admin_only
    async def handle_enable(self, data: IntentReceivedData) -> IntentResult:
        """Aktiviert den Voice Morph mit dem konfigurierten Preset."""
        preset_id = self.config.get("active_preset", "")
        if not preset_id:
            return IntentResult.failure("Kein Preset konfiguriert.",
                                        "Es ist kein Voice Morph Preset konfiguriert.")

        preset = self._morpher.get_preset(preset_id)
        if not preset:
            return IntentResult.failure("Preset nicht gefunden.",
                                        f"Preset '{preset_id}' wurde nicht gefunden.")

        self._active_preset = preset
        return IntentResult.success_with_response(
            f"Voice Morph aktiviert mit Preset {preset.label}."
        )

    @intent("voice_morph_disable", description="Voice Morph deaktivieren")
    @admin_only
    async def handle_disable(self, data: IntentReceivedData) -> IntentResult:
        """Deaktiviert den Voice Morph."""
        self._active_preset = None
        return IntentResult.success_with_response("Voice Morph deaktiviert. Normale Stimme aktiv.")

    @intent("voice_morph_set_preset", description="Voice Morph Preset wechseln")
    @admin_only
    async def handle_set_preset(self, data: IntentReceivedData, preset_name: str = "") -> IntentResult:
        """Wechselt das aktive Preset."""
        if not preset_name:
            return IntentResult.failure("Kein Preset angegeben.", "Bitte einen Preset-Namen angeben.")

        # Suche nach ID oder Label (case-insensitive)
        found = None
        for preset in self._morpher.presets:
            if preset.preset_id.lower() == preset_name.lower():
                found = preset
                break
            if preset.label.lower() == preset_name.lower():
                found = preset
                break

        if not found:
            available = ", ".join(p.label for p in self._morpher.presets)
            return IntentResult.failure(
                "Preset nicht gefunden.",
                f"Preset '{preset_name}' nicht gefunden. Verfuegbar: {available}."
            )

        self._active_preset = found
        self.set_config_value("active_preset", found.preset_id)
        self.save_config()
        return IntentResult.success_with_response(f"Voice Morph gewechselt auf {found.label}.")

    @intent("voice_morph_status", description="Voice Morph Status abfragen")
    @admin_only
    async def handle_status(self, data: IntentReceivedData) -> IntentResult:
        """Gibt den aktuellen Voice Morph Status zurueck."""
        if self._active_preset:
            intensity_pct = int(self._intensity * 100)
            return IntentResult.success_with_response(
                f"Voice Morph ist aktiv mit Preset {self._active_preset.label}, "
                f"Intensitaet {intensity_pct} Prozent."
            )
        return IntentResult.success_with_response("Voice Morph ist deaktiviert.")

    @intent("voice_morph_list_presets", description="Verfuegbare Presets auflisten")
    @admin_only
    async def handle_list_presets(self, data: IntentReceivedData) -> IntentResult:
        """Listet alle verfuegbaren Presets auf."""
        if not self._morpher.presets:
            return IntentResult.success_with_response("Keine Voice Morph Presets verfuegbar.")

        names = [p.label for p in self._morpher.presets]
        return IntentResult.success_with_response(
            f"Verfuegbare Presets: {', '.join(names)}."
        )

    @intent("voice_morph_set_intensity", description="Voice Morph Intensitaet setzen")
    @admin_only
    async def handle_set_intensity(self, data: IntentReceivedData, value: str = "") -> IntentResult:
        """Setzt die Intensitaet des Voice Morph."""
        if not value:
            return IntentResult.failure("Kein Wert angegeben.", "Bitte einen Wert angeben.")

        try:
            # "80 Prozent" → 80, "0.8" → 0.8
            clean = value.lower().replace("prozent", "").replace("%", "").strip()
            num = float(clean)
            if num > 1.0:
                num /= 100.0  # 80 → 0.8
            num = max(0.0, min(1.0, num))
        except ValueError:
            return IntentResult.failure("Ungueltiger Wert.", f"'{value}' ist kein gueltiger Prozentwert.")

        self._intensity = num
        self.set_config_value("intensity", num)
        self.save_config()
        return IntentResult.success_with_response(
            f"Voice Morph Intensitaet auf {int(num * 100)} Prozent gesetzt."
        )