# -*- coding: utf-8 -*- """ Audio Extension Points. Extension Points für Audio-Verarbeitung und -Effekte. """ from abc import abstractmethod from typing import Any, TYPE_CHECKING from trixy_core.plugins.extensions.extension import Extension from trixy_core.plugins.extensions.extension_point import ExtensionPoint, get_global_registry if TYPE_CHECKING: from trixy_core.audio.processing.processor import AudioProcessor # Extension Point IDs AUDIO_PROCESSOR_POINT = "audio.processor" class AudioProcessorExtension(Extension["AudioProcessor"]): """ Extension für Audio-Prozessoren. Ermöglicht Plugins, Audio-Streams zu verarbeiten: - Ducking (Lautstärke bei Wakeword/Conversation reduzieren) - Crossfade (Übergänge zwischen Tracks) - Equalizer (Frequenz-Anpassungen) - Effekte (Hall, Echo, etc.) - Limiter/Compressor - Analyse (Visualisierung, BPM-Erkennung) Beispiel: class DuckingExtension(AudioProcessorExtension): def __init__(self, plugin_name: str): super().__init__( "ducking_processor", "Ducking", plugin_name, priority=300, description="Reduziert Musik bei Sprachausgabe", ) async def create_instance(self, config): from my_plugin.processors import DuckingProcessor return DuckingProcessor( duck_factor=config.get("duck_factor", 0.3), fade_duration_ms=config.get("fade_duration_ms", 500), ) """ def __init__( self, extension_id: str, name: str, plugin_name: str = "", priority: int = 500, description: str = "", category: str = "effect", ) -> None: """ Initialisiert die AudioProcessor Extension. Args: extension_id: Eindeutige ID name: Anzeigename plugin_name: Plugin-Name priority: Priorität in der Pipeline (niedriger = früher) description: Beschreibung category: Kategorie (effect, analysis, utility) """ super().__init__(extension_id, name, plugin_name) self._info.description = description self._info.extension_point = AUDIO_PROCESSOR_POINT self._info.metadata["priority"] = priority self._info.metadata["category"] = category @property def priority(self) -> int: """Priorität in der Processing-Pipeline.""" return self._info.metadata.get("priority", 500) @property def category(self) -> str: """Kategorie des Prozessors.""" return self._info.metadata.get("category", "effect") @abstractmethod async def create_instance(self, config: dict[str, Any]) -> "AudioProcessor": """ Erstellt die AudioProcessor-Instanz. Args: config: Konfiguration für den Prozessor Returns: Konfigurierte AudioProcessor-Instanz """ pass def setup_audio_extension_points() -> dict[str, ExtensionPoint]: """ Richtet alle Audio Extension Points ein. Returns: Dict mit allen erstellten Extension Points """ registry = get_global_registry() points = {} # Audio-Prozessoren points[AUDIO_PROCESSOR_POINT] = registry.create_point( id=AUDIO_PROCESSOR_POINT, name="Audio-Prozessoren", description=( "Erweiterung um Audio-Verarbeitung: " "Ducking, Crossfade, Equalizer, Effekte, etc." ), category="audio", allows_multiple=True, ) return points def get_audio_processors() -> list["AudioProcessor"]: """ Holt alle registrierten Audio-Prozessoren. Returns: Liste aller AudioProcessor-Instanzen """ registry = get_global_registry() point = registry.get_point(AUDIO_PROCESSOR_POINT) if point: return point.get_all_instances() return [] def get_audio_processor_extensions() -> list[AudioProcessorExtension]: """ Holt alle registrierten AudioProcessor-Extensions. Returns: Liste aller Extensions """ registry = get_global_registry() point = registry.get_point(AUDIO_PROCESSOR_POINT) if point: return [e for e in point.extensions if isinstance(e, AudioProcessorExtension)] return []