# -*- 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(), )