Trixy7/trixy_core/plugins/extensions/points/audio.py

# -*- 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 []