Trixy7/trixy_core/audio/processing/processor.py

# -*- coding: utf-8 -*-
"""
Audio Processor Interface.

Basis-Interface für alle Audio-Prozessoren.
"""

from abc import ABC, abstractmethod
from enum import IntEnum
from typing import TYPE_CHECKING

if TYPE_CHECKING:
    from trixy_core.audio.processing.context import AudioProcessingContext, AudioType


class ProcessorPriority(IntEnum):
    """
    Prioritäten für Audio-Prozessoren.

    Niedrigere Werte werden zuerst ausgeführt.
    """
    # Früheste Verarbeitung
    FIRST = 0

    # Analyse (nur lesen, nicht modifizieren)
    ANALYSIS = 100

    # Effekte die früh angewendet werden
    PRE_EFFECTS = 200

    # Ducking (Lautstärke bei Wakeword/Conversation)
    DUCKING = 300

    # Crossfade und Übergänge
    CROSSFADE = 400

    # Equalizer und Frequenz-Anpassungen
    EQUALIZER = 500

    # Späte Effekte
    POST_EFFECTS = 600

    # Lautstärke-Anpassung
    VOLUME = 700

    # Limiter/Compressor (Clipping verhindern)
    LIMITER = 800

    # Letzte Verarbeitung
    LAST = 900

    # Standard für Plugins ohne spezifische Priorität
    DEFAULT = 500


class AudioProcessor(ABC):
    """
    Basis-Klasse für Audio-Prozessoren.

    Audio-Prozessoren verarbeiten PCM-Audio-Chunks in einer Pipeline.
    Jeder Prozessor erhält den Chunk und einen Kontext mit Informationen
    über den aktuellen Zustand (Position, Wakeword, etc.).

    Beispiel:
        class DuckingProcessor(AudioProcessor):
            def __init__(self):
                super().__init__(
                    processor_id="ducking",
                    name="Ducking",
                    priority=ProcessorPriority.DUCKING,
                )
                self.duck_factor = 0.3

            def process(self, chunk: bytes, context: AudioProcessingContext) -> bytes:
                if context.should_duck:
                    return self._apply_volume(chunk, self.duck_factor)
                return chunk
    """

    def __init__(
        self,
        processor_id: str,
        name: str,
        priority: int = ProcessorPriority.DEFAULT,
        enabled: bool = True,
        audio_types: list["AudioType"] | None = None,
    ) -> None:
        """
        Initialisiert den Prozessor.

        Args:
            processor_id: Eindeutige ID
            name: Anzeigename
            priority: Reihenfolge in der Pipeline (niedriger = früher)
            enabled: Ist der Prozessor aktiv?
            audio_types: Liste der unterstützten Audio-Typen (None = alle)
        """
        self._id = processor_id
        self._name = name
        self._priority = priority
        self._enabled = enabled
        self._audio_types = audio_types  # None = alle Typen

    @property
    def id(self) -> str:
        """Eindeutige ID des Prozessors."""
        return self._id

    @property
    def name(self) -> str:
        """Anzeigename."""
        return self._name

    @property
    def priority(self) -> int:
        """Priorität in der Pipeline (niedriger = früher)."""
        return self._priority

    @property
    def enabled(self) -> bool:
        """Ist der Prozessor aktiv?"""
        return self._enabled

    @enabled.setter
    def enabled(self, value: bool) -> None:
        """Aktiviert/Deaktiviert den Prozessor."""
        self._enabled = value

    @property
    def audio_types(self) -> list["AudioType"] | None:
        """
        Unterstützte Audio-Typen.

        Returns:
            Liste der Typen oder None für alle Typen.
        """
        return self._audio_types

    def supports_audio_type(self, audio_type: "AudioType") -> bool:
        """
        Prüft ob dieser Prozessor einen Audio-Typ unterstützt.

        Args:
            audio_type: Der zu prüfende Audio-Typ

        Returns:
            True wenn unterstützt (oder wenn alle Typen unterstützt werden)
        """
        if self._audio_types is None:
            return True  # Alle Typen unterstützt
        return audio_type in self._audio_types

    def should_process(self, context: "AudioProcessingContext") -> bool:
        """
        Prüft ob dieser Prozessor für den Kontext aktiv sein sollte.

        Berücksichtigt: enabled, audio_type

        Args:
            context: Der Verarbeitungskontext

        Returns:
            True wenn der Prozessor aktiv sein sollte
        """
        if not self._enabled:
            return False
        return self.supports_audio_type(context.audio_type)

    @abstractmethod
    def process(
        self,
        chunk: bytes,
        context: "AudioProcessingContext",
    ) -> bytes:
        """
        Verarbeitet einen Audio-Chunk.

        Args:
            chunk: PCM-Audio-Daten (16-bit, little-endian)
            context: Verarbeitungskontext mit Zustandsinformationen

        Returns:
            Verarbeiteter Audio-Chunk (gleiche Länge wie Eingabe)

        Note:
            - Der zurückgegebene Chunk MUSS die gleiche Länge haben
            - Für Crossfade: Zweiten Track separat laden und mixen
            - Bei Fehlern: Unveränderten Chunk zurückgeben
        """
        pass

    def on_track_start(self, context: "AudioProcessingContext") -> None:
        """
        Callback wenn ein neuer Track startet.

        Args:
            context: Verarbeitungskontext
        """
        pass

    def on_track_end(self, context: "AudioProcessingContext") -> None:
        """
        Callback wenn ein Track endet.

        Args:
            context: Verarbeitungskontext
        """
        pass

    def on_state_change(self, state_name: str, value: bool) -> None:
        """
        Callback bei Zustandsänderungen.

        Args:
            state_name: Name des Zustands (z.B. "wakeword_active")
            value: Neuer Wert
        """
        pass

    def reset(self) -> None:
        """
        Setzt den Prozessor zurück.

        Wird aufgerufen wenn die Wiedergabe gestoppt wird.
        """
        pass

    def get_config(self) -> dict:
        """
        Liefert die aktuelle Konfiguration.

        Returns:
            Konfigurations-Dictionary
        """
        return {
            "id": self._id,
            "name": self._name,
            "priority": self._priority,
            "enabled": self._enabled,
        }

    def __repr__(self) -> str:
        return (
            f"{self.__class__.__name__}("
            f"id={self._id!r}, "
            f"priority={self._priority}, "
            f"enabled={self._enabled})"
        )


# =============================================================================
# Hilfsfunktionen für Audio-Verarbeitung
# =============================================================================

def apply_volume(chunk: bytes, volume: float) -> bytes:
    """
    Wendet Lautstärke auf PCM-Daten an.

    Args:
        chunk: PCM-Audio-Daten (16-bit, little-endian)
        volume: Lautstärke-Faktor (0.0 - 1.0+)

    Returns:
        Modifizierter Chunk
    """
    import struct

    if volume == 1.0:
        return chunk

    if volume == 0.0:
        return b'\x00' * len(chunk)

    # 16-bit Samples entpacken
    num_samples = len(chunk) // 2
    samples = struct.unpack(f"<{num_samples}h", chunk)

    # Lautstärke anwenden mit Clipping
    adjusted = [
        max(-32768, min(32767, int(s * volume)))
        for s in samples
    ]

    return struct.pack(f"<{num_samples}h", *adjusted)


def mix_chunks(chunk1: bytes, chunk2: bytes, mix: float = 0.5) -> bytes:
    """
    Mischt zwei Audio-Chunks.

    Args:
        chunk1: Erster Chunk
        chunk2: Zweiter Chunk
        mix: Mix-Verhältnis (0.0 = nur chunk1, 1.0 = nur chunk2)

    Returns:
        Gemischter Chunk
    """
    import struct

    if len(chunk1) != len(chunk2):
        # Längen angleichen (kürzeren mit Stille auffüllen)
        max_len = max(len(chunk1), len(chunk2))
        chunk1 = chunk1.ljust(max_len, b'\x00')
        chunk2 = chunk2.ljust(max_len, b'\x00')

    num_samples = len(chunk1) // 2
    samples1 = struct.unpack(f"<{num_samples}h", chunk1)
    samples2 = struct.unpack(f"<{num_samples}h", chunk2)

    # Mixen
    factor1 = 1.0 - mix
    factor2 = mix

    mixed = [
        max(-32768, min(32767, int(s1 * factor1 + s2 * factor2)))
        for s1, s2 in zip(samples1, samples2)
    ]

    return struct.pack(f"<{num_samples}h", *mixed)


def fade_chunk(
    chunk: bytes,
    fade_in: bool = False,
    fade_out: bool = False,
) -> bytes:
    """
    Wendet Fade-In oder Fade-Out auf einen Chunk an.

    Args:
        chunk: PCM-Audio-Daten
        fade_in: Fade von 0 auf 1
        fade_out: Fade von 1 auf 0

    Returns:
        Chunk mit Fade
    """
    import struct

    if not fade_in and not fade_out:
        return chunk

    num_samples = len(chunk) // 2
    samples = struct.unpack(f"<{num_samples}h", chunk)

    result = []
    for i, sample in enumerate(samples):
        progress = i / max(1, num_samples - 1)

        if fade_in:
            factor = progress
        elif fade_out:
            factor = 1.0 - progress
        else:
            factor = 1.0

        result.append(max(-32768, min(32767, int(sample * factor))))

    return struct.pack(f"<{num_samples}h", *result)