# -*- 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." )