Trixy7/plugins/crossfade/processor.py

# -*- coding: utf-8 -*-
"""
Crossfade Audio-Prozessor.

Mischt das Ende eines Tracks mit dem Anfang des nächsten Tracks
für einen nahtlosen Übergang.
"""

import asyncio
import struct
import threading
from collections import deque
from dataclasses import dataclass, field
from pathlib import Path
from typing import TYPE_CHECKING, Any

from trixy_core.audio.processing import (
    AudioProcessor,
    AudioProcessingContext,
    ProcessorPriority,
    AudioType,
)
from trixy_core.audio.processing.processor import mix_chunks, apply_volume
from trixy_core.audio.config import get_audio_settings

if TYPE_CHECKING:
    from trixy_core.music.formats import DecoderRegistry
    from trixy_core.music.track import Track


@dataclass
class CrossfadeConfig:
    """Konfiguration für Crossfade."""
    # Crossfade-Dauer in Millisekunden
    duration_ms: int = 5000

    # Fade-Kurve: "linear", "exponential", "cosine"
    fade_curve: str = "linear"

    # Minimale Track-Dauer für Crossfade (kürzere Tracks werden nicht gefaded)
    min_track_duration_ms: int = 30000

    # Prebuffer-Dauer (wie viel vom nächsten Track vorgeladen wird)
    prebuffer_duration_ms: int = 6000

    # Aktiviert/Deaktiviert
    enabled: bool = True


class CrossfadeProcessor(AudioProcessor):
    """
    Audio-Prozessor für Crossfade zwischen Tracks.

    Funktionsweise:
    1. Wenn Track sich dem Ende nähert und ein nächster Track vorhanden ist,
       wird der Anfang des nächsten Tracks vorgeladen.
    2. Während der Crossfade-Region wird der aktuelle Track ausgeblendet
       und der nächste Track eingeblendet.
    3. Die Chunks werden gemischt und zurückgegeben.

    WICHTIG: Der nächste Track muss vorab dekodiert und gepuffert werden,
    da process() synchron läuft.
    """

    def __init__(
        self,
        processor_id: str = "crossfade",
        name: str = "Crossfade",
        priority: int = ProcessorPriority.CROSSFADE,
        enabled: bool = True,
        config: CrossfadeConfig | None = None,
        decoder_registry: "DecoderRegistry | None" = None,
    ) -> None:
        """
        Initialisiert den Crossfade-Prozessor.

        Args:
            processor_id: Eindeutige ID
            name: Anzeigename
            priority: Priorität in der Pipeline
            enabled: Aktiviert
            config: Crossfade-Konfiguration
            decoder_registry: Decoder-Registry für Audio-Dekodierung
        """
        # Nur MUSIC verarbeiten - kein Crossfade für TTS oder Assets
        super().__init__(processor_id, name, priority, enabled, audio_types=[AudioType.MUSIC])

        self.config = config or CrossfadeConfig()
        self._decoder_registry = decoder_registry

        # Puffer für nächsten Track
        self._next_track_buffer: deque[bytes] = deque()
        self._next_track_loaded = False
        self._next_track_position = 0  # Position im Puffer (in Bytes)

        # Aktueller Track
        self._current_track: "Track | None" = None
        self._next_track: "Track | None" = None

        # Crossfade-Status
        self._is_crossfading = False
        self._crossfade_started_at_ms = 0

        # Synchronisation
        self._buffer_lock = threading.Lock()
        self._loading_task: asyncio.Task | None = None

        # Audio-Format aus globalen Einstellungen holen (wird bei Track-Start überschrieben)
        audio_settings = get_audio_settings()
        self._sample_rate = audio_settings.music_sample_rate
        self._channels = audio_settings.music_channels
        self._bytes_per_ms = 0  # Wird berechnet

        # Statistiken
        self._crossfades_performed = 0
        self._chunks_mixed = 0

    def set_decoder_registry(self, registry: "DecoderRegistry") -> None:
        """Setzt die Decoder-Registry."""
        self._decoder_registry = registry

    def update_config(self, **kwargs) -> None:
        """Aktualisiert Konfiguration."""
        for key, value in kwargs.items():
            if hasattr(self.config, key):
                setattr(self.config, key, value)

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

        Wenn wir in der Crossfade-Region sind und der nächste Track
        gepuffert ist, wird gemischt.
        """
        if not self.enabled or not self.config.enabled:
            return chunk

        # Audio-Format aktualisieren
        self._sample_rate = context.sample_rate
        self._channels = context.channels
        self._bytes_per_ms = (context.sample_rate * context.channels * 2) // 1000  # 16-bit

        # Prüfen ob Crossfade möglich
        if not self._should_crossfade(context):
            return chunk

        # Crossfade starten wenn noch nicht aktiv
        if not self._is_crossfading:
            self._start_crossfade(context)

        # Wenn keine gepufferten Daten, nur Fade-Out
        with self._buffer_lock:
            if not self._next_track_loaded or not self._next_track_buffer:
                return self._apply_fade_out(chunk, context)

            # Nächsten Chunk aus Puffer holen
            next_chunk = self._get_next_chunk(len(chunk))

        if not next_chunk or len(next_chunk) != len(chunk):
            # Nicht genug Daten, nur Fade-Out
            return self._apply_fade_out(chunk, context)

        # Mix-Faktor berechnen (0.0 = nur aktueller Track, 1.0 = nur nächster Track)
        mix_factor = self._calculate_mix_factor(context)

        # Chunks mischen
        mixed = mix_chunks(chunk, next_chunk, mix_factor)
        self._chunks_mixed += 1

        return mixed

    def on_track_start(self, context: AudioProcessingContext) -> None:
        """Wird bei Track-Start aufgerufen."""
        self._current_track = context.track
        self._next_track = context.next_track
        self._is_crossfading = False
        self._crossfade_started_at_ms = 0

        # Audio-Format speichern
        self._sample_rate = context.sample_rate
        self._channels = context.channels
        self._bytes_per_ms = (context.sample_rate * context.channels * 2) // 1000

        # Prüfen ob Crossfade überhaupt möglich
        if not self._can_crossfade(context):
            return

        # Nächsten Track vorladen
        if context.next_track and context.next_track.path:
            self._start_preloading(context.next_track)

    def on_track_end(self, context: AudioProcessingContext) -> None:
        """Wird bei Track-Ende aufgerufen."""
        if self._is_crossfading:
            self._crossfades_performed += 1

        self._is_crossfading = False

        # Puffer NICHT leeren - wird für den nächsten Track-Start benötigt
        # Der nächste Track verwendet diesen Puffer als seinen Anfang

    def on_state_change(self, state_name: str, value: bool) -> None:
        """Wird bei Zustandsänderungen aufgerufen."""
        # Bei Wakeword/Conversation/TTS: Crossfade abbrechen
        if value and state_name in ("wakeword_active", "conversation_active", "tts_playing"):
            self._abort_crossfade()

    def reset(self) -> None:
        """Setzt den Prozessor zurück."""
        self._abort_crossfade()
        with self._buffer_lock:
            self._next_track_buffer.clear()
            self._next_track_loaded = False
            self._next_track_position = 0
        self._current_track = None
        self._next_track = None

    def get_config(self) -> dict[str, Any]:
        """Liefert Konfiguration."""
        base = super().get_config()
        base.update({
            "crossfade_duration_ms": self.config.duration_ms,
            "fade_curve": self.config.fade_curve,
            "min_track_duration_ms": self.config.min_track_duration_ms,
            "crossfades_performed": self._crossfades_performed,
            "chunks_mixed": self._chunks_mixed,
            "is_crossfading": self._is_crossfading,
            "next_track_loaded": self._next_track_loaded,
        })
        return base

    # ==========================================================================
    # Interne Methoden
    # ==========================================================================

    def _can_crossfade(self, context: AudioProcessingContext) -> bool:
        """Prüft ob Crossfade für diesen Track möglich ist."""
        if not self.enabled or not self.config.enabled:
            return False

        # Kein nächster Track
        if not context.has_next_track:
            return False

        # Track zu kurz
        if context.duration_ms < self.config.min_track_duration_ms:
            return False

        # Crossfade länger als halber Track
        if self.config.duration_ms > context.duration_ms // 2:
            return False

        return True

    def _should_crossfade(self, context: AudioProcessingContext) -> bool:
        """Prüft ob wir gerade crossfaden sollten."""
        if not self._can_crossfade(context):
            return False

        # Noch nicht nahe genug am Ende?
        remaining_ms = context.remaining_ms
        if remaining_ms > self.config.duration_ms:
            return False

        return True

    def _start_crossfade(self, context: AudioProcessingContext) -> None:
        """Startet Crossfade."""
        self._is_crossfading = True
        self._crossfade_started_at_ms = context.position_ms

    def _abort_crossfade(self) -> None:
        """Bricht Crossfade ab."""
        if self._loading_task and not self._loading_task.done():
            self._loading_task.cancel()
        self._is_crossfading = False

    def _calculate_mix_factor(self, context: AudioProcessingContext) -> float:
        """
        Berechnet Mix-Faktor basierend auf Position.

        0.0 = 100% aktueller Track
        1.0 = 100% nächster Track
        """
        remaining_ms = context.remaining_ms
        duration_ms = self.config.duration_ms

        if remaining_ms >= duration_ms:
            return 0.0
        if remaining_ms <= 0:
            return 1.0

        # Fortschritt im Crossfade (0.0 = Start, 1.0 = Ende)
        progress = 1.0 - (remaining_ms / duration_ms)

        # Kurve anwenden
        if self.config.fade_curve == "exponential":
            # Exponentiell: schneller Anstieg am Anfang
            return progress ** 0.5
        elif self.config.fade_curve == "cosine":
            # Kosinus: S-Kurve
            import math
            return (1.0 - math.cos(progress * math.pi)) / 2.0
        else:
            # Linear (Standard)
            return progress

    def _apply_fade_out(self, chunk: bytes, context: AudioProcessingContext) -> bytes:
        """Wendet nur Fade-Out auf aktuellen Track an."""
        remaining_ms = context.remaining_ms
        duration_ms = self.config.duration_ms

        if remaining_ms >= duration_ms:
            return chunk

        # Lautstärke basierend auf verbleibender Zeit
        volume = max(0.0, remaining_ms / duration_ms)

        return apply_volume(chunk, volume)

    def _get_next_chunk(self, size: int) -> bytes | None:
        """Holt Chunk aus Puffer für nächsten Track."""
        if not self._next_track_buffer:
            return None

        result = bytearray()
        remaining = size

        while remaining > 0 and self._next_track_buffer:
            chunk = self._next_track_buffer[0]

            if len(chunk) <= remaining:
                # Ganzen Chunk verwenden
                result.extend(chunk)
                remaining -= len(chunk)
                self._next_track_buffer.popleft()
            else:
                # Teil des Chunks verwenden
                result.extend(chunk[:remaining])
                self._next_track_buffer[0] = chunk[remaining:]
                remaining = 0

        self._next_track_position += len(result)

        return bytes(result) if result else None

    def _start_preloading(self, track: "Track") -> None:
        """Startet Vorladen des nächsten Tracks im Hintergrund."""
        if not self._decoder_registry:
            return

        # Alten Puffer leeren
        with self._buffer_lock:
            self._next_track_buffer.clear()
            self._next_track_loaded = False
            self._next_track_position = 0

        # Vorladen in separatem Thread (async in sync integrieren)
        try:
            loop = asyncio.get_running_loop()
            self._loading_task = loop.create_task(self._preload_track(track))
        except RuntimeError:
            # Kein Event-Loop - synchron laden nicht möglich
            pass

    async def _preload_track(self, track: "Track") -> None:
        """Lädt Anfang des nächsten Tracks."""
        if not track.path or not self._decoder_registry:
            return

        try:
            decoder = self._decoder_registry.get_best_decoder(
                path=track.path,
                extension=track.metadata.format if track.metadata else None,
            )

            if not decoder:
                return

            # Chunk-Größe: 100ms
            chunk_duration_ms = 100
            target_duration_ms = self.config.prebuffer_duration_ms
            loaded_ms = 0

            async for chunk in decoder.decode_stream(track.path, chunk_duration_ms):
                if loaded_ms >= target_duration_ms:
                    break

                with self._buffer_lock:
                    self._next_track_buffer.append(chunk.pcm_data)

                loaded_ms += chunk.duration_ms

            with self._buffer_lock:
                self._next_track_loaded = True

        except Exception:
            # Fehler beim Laden - Crossfade wird ohne nächsten Track durchgeführt
            pass

    # ==========================================================================
    # Synchrone Hilfsfunktionen für Tests
    # ==========================================================================

    def preload_from_pcm(self, pcm_data: bytes, chunk_size: int = 19200) -> None:
        """
        Lädt PCM-Daten direkt in den Puffer (für Tests).

        Args:
            pcm_data: Rohe PCM-Daten (16-bit stereo)
            chunk_size: Größe der Chunks
        """
        with self._buffer_lock:
            self._next_track_buffer.clear()

            # In Chunks aufteilen
            for i in range(0, len(pcm_data), chunk_size):
                chunk = pcm_data[i:i + chunk_size]
                self._next_track_buffer.append(chunk)

            self._next_track_loaded = True
            self._next_track_position = 0


def crossfade_audio(
    audio1: bytes,
    audio2: bytes,
    crossfade_ms: int,
    sample_rate: int = 44100,
    channels: int = 2,
    fade_curve: str = "linear",
) -> bytes:
    """
    Erstellt einen Crossfade zwischen zwei Audio-Daten.

    Diese Standalone-Funktion ist nützlich für Tests und
    Batch-Verarbeitung.

    Args:
        audio1: PCM-Daten des ersten Tracks (16-bit)
        audio2: PCM-Daten des zweiten Tracks (16-bit)
        crossfade_ms: Crossfade-Dauer in Millisekunden
        sample_rate: Sample-Rate
        channels: Anzahl Kanäle
        fade_curve: "linear", "exponential", oder "cosine"

    Returns:
        Gemischte PCM-Daten
    """
    import math

    bytes_per_ms = (sample_rate * channels * 2) // 1000  # 16-bit
    crossfade_bytes = crossfade_ms * bytes_per_ms

    # Sicherstellen dass genug Daten vorhanden
    if len(audio1) < crossfade_bytes or len(audio2) < crossfade_bytes:
        # Nicht genug Daten für Crossfade
        return audio1 + audio2

    # Audio1 bis auf Crossfade-Region
    result = bytearray(audio1[:-crossfade_bytes])

    # Crossfade-Region
    fade_out_region = audio1[-crossfade_bytes:]
    fade_in_region = audio2[:crossfade_bytes]

    # Samples extrahieren
    sample_count = crossfade_bytes // 2  # 16-bit = 2 bytes pro sample
    fade_out_samples = struct.unpack(f"<{sample_count}h", fade_out_region)
    fade_in_samples = struct.unpack(f"<{sample_count}h", fade_in_region)

    # Mischen
    mixed_samples = []
    for i in range(sample_count):
        progress = i / sample_count

        # Kurve anwenden
        if fade_curve == "exponential":
            mix = progress ** 0.5
        elif fade_curve == "cosine":
            mix = (1.0 - math.cos(progress * math.pi)) / 2.0
        else:
            mix = progress

        # Mischen
        sample1 = fade_out_samples[i] * (1.0 - mix)
        sample2 = fade_in_samples[i] * mix
        mixed = int(sample1 + sample2)

        # Clipping vermeiden
        mixed = max(-32768, min(32767, mixed))
        mixed_samples.append(mixed)

    # Gemischte Region hinzufügen
    result.extend(struct.pack(f"<{sample_count}h", *mixed_samples))

    # Rest von Audio2 (nach Crossfade-Region)
    result.extend(audio2[crossfade_bytes:])

    return bytes(result)


def save_wav(
    pcm_data: bytes,
    path: Path | str,
    sample_rate: int = 44100,
    channels: int = 2,
    bit_depth: int = 16,
) -> None:
    """
    Speichert PCM-Daten als WAV-Datei.

    Args:
        pcm_data: Rohe PCM-Daten
        path: Ziel-Pfad
        sample_rate: Sample-Rate in Hz
        channels: Anzahl Kanäle
        bit_depth: Bits pro Sample
    """
    import wave

    path = Path(path)
    path.parent.mkdir(parents=True, exist_ok=True)

    with wave.open(str(path), "wb") as wav:
        wav.setnchannels(channels)
        wav.setsampwidth(bit_depth // 8)
        wav.setframerate(sample_rate)
        wav.writeframes(pcm_data)


def load_wav(path: Path | str) -> tuple[bytes, int, int]:
    """
    Lädt WAV-Datei.

    Args:
        path: Pfad zur WAV-Datei

    Returns:
        Tuple (pcm_data, sample_rate, channels)
    """
    import wave

    with wave.open(str(path), "rb") as wav:
        return (
            wav.readframes(wav.getnframes()),
            wav.getframerate(),
            wav.getnchannels(),
        )