Trixy7/trixy_core/audio/processing/context.py

# -*- coding: utf-8 -*-
"""
Audio Processing Context.

Stellt Kontextinformationen für Audio-Prozessoren bereit.
"""

from dataclasses import dataclass, field
from enum import Enum, auto
from typing import TYPE_CHECKING, Any

if TYPE_CHECKING:
    from trixy_core.music.track import Track


class AudioType(Enum):
    """
    Typ der Audio-Quelle.

    Ermöglicht Plugins, nur bestimmte Audio-Typen zu verarbeiten.
    """
    # Text-to-Speech Ausgabe
    TTS = auto()

    # Sound-Effekte aus dem Assets-Ordner
    ASSET = auto()

    # Musik-Streaming
    MUSIC = auto()

    # Unbekannt/Generic
    UNKNOWN = auto()


@dataclass
class AudioFormatInfo:
    """
    Audio-Format-Informationen für einen bestimmten Audio-Typ.

    Wird von Plugins genutzt um das korrekte Format zu kennen.
    """
    sample_rate: int = 44100
    channels: int = 2
    bit_depth: int = 16  # Bits pro Sample

    @property
    def sample_width(self) -> int:
        """Bytes pro Sample."""
        return self.bit_depth // 8

    @property
    def bytes_per_second(self) -> int:
        """Bytes pro Sekunde."""
        return self.sample_rate * self.channels * self.sample_width

    @property
    def bytes_per_ms(self) -> int:
        """Bytes pro Millisekunde."""
        return self.bytes_per_second // 1000


@dataclass
class AudioProcessingContext:
    """
    Kontext für Audio-Verarbeitung.

    Enthält alle Informationen die ein Prozessor benötigt,
    um Entscheidungen zu treffen (z.B. Ducking bei Wakeword,
    Crossfade am Track-Ende).

    Attributes:
        position_ms: Aktuelle Position im Track (Millisekunden)
        duration_ms: Gesamtdauer des Tracks (Millisekunden)
        chunk_duration_ms: Dauer des aktuellen Chunks (Millisekunden)
        track: Aktueller Track
        next_track: Nächster Track (falls vorhanden)
        sample_rate: Sample-Rate in Hz
        channels: Anzahl Kanäle (1=mono, 2=stereo)
        sample_width: Bytes pro Sample (2=16-bit)
        is_wakeword_active: Wakeword wurde erkannt
        is_conversation_active: Conversation läuft
        is_tts_playing: TTS wird gerade abgespielt
        volume: Aktuelle Master-Lautstärke (0.0 - 1.0)
        is_muted: Ist stummgeschaltet
        satellite_ids: Ziel-Satellites für diesen Chunk
        metadata: Zusätzliche Metadaten
    """

    # Audio-Typ (TTS, Asset, Musik)
    audio_type: AudioType = AudioType.UNKNOWN

    # Track-Position
    position_ms: int = 0
    duration_ms: int = 0
    chunk_duration_ms: int = 0

    # Track-Informationen
    track: "Track | None" = None
    next_track: "Track | None" = None

    # Audio-Format
    sample_rate: int = 44100
    channels: int = 2
    sample_width: int = 2  # 16-bit
    bit_depth: int = 16  # Bits pro Sample (für Kompatibilität)

    # Systemzustand
    is_wakeword_active: bool = False
    is_conversation_active: bool = False
    is_tts_playing: bool = False

    # Lautstärke
    volume: float = 1.0
    is_muted: bool = False

    # Ziel
    satellite_ids: list[str] = field(default_factory=list)

    # Erweiterbar
    metadata: dict[str, Any] = field(default_factory=dict)

    @property
    def remaining_ms(self) -> int:
        """Verbleibende Zeit bis Track-Ende in Millisekunden."""
        if self.duration_ms > 0:
            return max(0, self.duration_ms - self.position_ms)
        return 0

    @property
    def progress(self) -> float:
        """Fortschritt im Track (0.0 - 1.0)."""
        if self.duration_ms > 0:
            return min(1.0, self.position_ms / self.duration_ms)
        return 0.0

    @property
    def is_near_end(self) -> bool:
        """Prüft ob Track fast zu Ende ist (< 5 Sekunden)."""
        return self.remaining_ms < 5000

    @property
    def has_next_track(self) -> bool:
        """Prüft ob ein nächster Track vorhanden ist."""
        return self.next_track is not None

    @property
    def should_duck(self) -> bool:
        """Prüft ob Audio gedämpft werden sollte (Wakeword/Conversation/TTS)."""
        return self.is_wakeword_active or self.is_conversation_active or self.is_tts_playing

    @property
    def is_tts(self) -> bool:
        """Prüft ob Audio TTS ist."""
        return self.audio_type == AudioType.TTS

    @property
    def is_asset(self) -> bool:
        """Prüft ob Audio ein Asset-Sound ist."""
        return self.audio_type == AudioType.ASSET

    @property
    def is_music(self) -> bool:
        """Prüft ob Audio Musik ist."""
        return self.audio_type == AudioType.MUSIC

    @property
    def format_info(self) -> AudioFormatInfo:
        """Gibt Audio-Format-Informationen zurück."""
        return AudioFormatInfo(
            sample_rate=self.sample_rate,
            channels=self.channels,
            bit_depth=self.bit_depth,
        )

    @property
    def bytes_per_sample(self) -> int:
        """Bytes pro Sample (channels * sample_width)."""
        return self.channels * self.sample_width

    @property
    def samples_per_ms(self) -> float:
        """Samples pro Millisekunde."""
        return self.sample_rate / 1000.0

    def copy(self, **updates) -> "AudioProcessingContext":
        """
        Erstellt eine Kopie mit optionalen Updates.

        Args:
            **updates: Felder die überschrieben werden sollen

        Returns:
            Neue Kontext-Instanz
        """
        from dataclasses import asdict
        data = asdict(self)
        data.update(updates)
        return AudioProcessingContext(**data)