# -*- coding: utf-8 -*- """ Audio Processing Pipeline. Verwaltet eine Kette von Audio-Prozessoren. """ import logging import threading from typing import TYPE_CHECKING from trixy_core.audio.processing.context import AudioProcessingContext from trixy_core.audio.processing.processor import AudioProcessor if TYPE_CHECKING: from trixy_core.music.track import Track class AudioProcessingPipeline: """ Pipeline für Audio-Verarbeitung. Verwaltet eine sortierte Liste von Audio-Prozessoren, die nacheinander auf Audio-Chunks angewendet werden. Features: - Prozessoren nach Priorität sortiert - Thread-safe - Kontext-Management - Fehlerbehandlung (Chunk wird unverändert weitergegeben) Usage: pipeline = AudioProcessingPipeline() # Prozessoren hinzufügen pipeline.add(DuckingProcessor()) pipeline.add(CrossfadeProcessor()) # Audio verarbeiten processed = pipeline.process(chunk, context) """ def __init__(self) -> None: """Initialisiert die Pipeline.""" self._processors: list[AudioProcessor] = [] self._lock = threading.RLock() self._logger = logging.getLogger(__name__) # Kontext-Zustand (wird von außen gesetzt) self._is_wakeword_active = False self._is_conversation_active = False self._is_tts_playing = False self._current_track: "Track | None" = None self._next_track: "Track | None" = None # ========================================================================== # Processor Management # ========================================================================== def add(self, processor: AudioProcessor) -> None: """ Fügt einen Prozessor zur Pipeline hinzu. Prozessoren werden nach Priorität sortiert (niedriger = früher). Args: processor: Der hinzuzufügende Prozessor """ with self._lock: # Prüfen ob bereits vorhanden if any(p.id == processor.id for p in self._processors): self._logger.warning( f"Prozessor {processor.id} bereits vorhanden, wird ersetzt" ) self._processors = [ p for p in self._processors if p.id != processor.id ] self._processors.append(processor) self._processors.sort(key=lambda p: p.priority) self._logger.debug( f"Prozessor hinzugefügt: {processor.name} (Priorität: {processor.priority})" ) def remove(self, processor_id: str) -> bool: """ Entfernt einen Prozessor aus der Pipeline. Args: processor_id: ID des zu entfernenden Prozessors Returns: True wenn entfernt wurde """ with self._lock: before = len(self._processors) self._processors = [ p for p in self._processors if p.id != processor_id ] removed = len(self._processors) < before if removed: self._logger.debug(f"Prozessor entfernt: {processor_id}") return removed def get(self, processor_id: str) -> AudioProcessor | None: """ Holt einen Prozessor nach ID. Args: processor_id: ID des Prozessors Returns: Prozessor oder None """ with self._lock: for p in self._processors: if p.id == processor_id: return p return None def enable(self, processor_id: str) -> bool: """ Aktiviert einen Prozessor. Args: processor_id: ID des Prozessors Returns: True wenn gefunden """ processor = self.get(processor_id) if processor: processor.enabled = True return True return False def disable(self, processor_id: str) -> bool: """ Deaktiviert einen Prozessor. Args: processor_id: ID des Prozessors Returns: True wenn gefunden """ processor = self.get(processor_id) if processor: processor.enabled = False return True return False def clear(self) -> None: """Entfernt alle Prozessoren.""" with self._lock: self._processors.clear() self._logger.debug("Alle Prozessoren entfernt") @property def processors(self) -> list[AudioProcessor]: """Liste aller Prozessoren (sortiert nach Priorität).""" with self._lock: return list(self._processors) @property def count(self) -> int: """Anzahl der Prozessoren.""" return len(self._processors) # ========================================================================== # State Management # ========================================================================== def set_wakeword_active(self, active: bool) -> None: """Setzt Wakeword-Status.""" if self._is_wakeword_active != active: self._is_wakeword_active = active self._notify_state_change("wakeword_active", active) def set_conversation_active(self, active: bool) -> None: """Setzt Conversation-Status.""" if self._is_conversation_active != active: self._is_conversation_active = active self._notify_state_change("conversation_active", active) def set_tts_playing(self, playing: bool) -> None: """Setzt TTS-Wiedergabe-Status.""" if self._is_tts_playing != playing: self._is_tts_playing = playing self._notify_state_change("tts_playing", playing) def set_current_track(self, track: "Track | None") -> None: """Setzt den aktuellen Track.""" self._current_track = track def set_next_track(self, track: "Track | None") -> None: """Setzt den nächsten Track.""" self._next_track = track def _notify_state_change(self, state_name: str, value: bool) -> None: """Benachrichtigt alle Prozessoren über Zustandsänderung.""" with self._lock: for processor in self._processors: if processor.enabled: try: processor.on_state_change(state_name, value) except Exception as e: self._logger.error( f"Fehler in on_state_change von {processor.id}: {e}" ) # ========================================================================== # Audio Processing # ========================================================================== def process( self, chunk: bytes, context: AudioProcessingContext | None = None, ) -> bytes: """ Verarbeitet einen Audio-Chunk durch alle aktiven Prozessoren. Args: chunk: PCM-Audio-Daten context: Optionaler Kontext (wird automatisch erstellt wenn None) Returns: Verarbeiteter Chunk """ if not chunk: return chunk # Kontext erstellen falls nicht vorhanden if context is None: context = self._create_default_context() # Pipeline-Status in Kontext übernehmen context.is_wakeword_active = self._is_wakeword_active context.is_conversation_active = self._is_conversation_active context.is_tts_playing = self._is_tts_playing context.track = self._current_track context.next_track = self._next_track # Durch alle Prozessoren (gefiltert nach enabled + audio_type) with self._lock: processors = [p for p in self._processors if p.should_process(context)] for processor in processors: try: result = processor.process(chunk, context) # Validierung: Chunk-Länge darf sich nicht ändern if len(result) != len(chunk): self._logger.warning( f"Prozessor {processor.id} hat Chunk-Länge geändert " f"({len(chunk)} -> {len(result)}), ignoriert" ) continue chunk = result except Exception as e: self._logger.error( f"Fehler in Prozessor {processor.id}: {e}" ) # Bei Fehler wird der unveränderte Chunk weitergegeben return chunk def _create_default_context(self) -> AudioProcessingContext: """Erstellt einen Standard-Kontext.""" return AudioProcessingContext( is_wakeword_active=self._is_wakeword_active, is_conversation_active=self._is_conversation_active, is_tts_playing=self._is_tts_playing, track=self._current_track, next_track=self._next_track, ) # ========================================================================== # Track Events # ========================================================================== def notify_track_start(self, context: AudioProcessingContext) -> None: """Benachrichtigt alle Prozessoren über Track-Start.""" with self._lock: for processor in self._processors: if processor.enabled: try: processor.on_track_start(context) except Exception as e: self._logger.error( f"Fehler in on_track_start von {processor.id}: {e}" ) def notify_track_end(self, context: AudioProcessingContext) -> None: """Benachrichtigt alle Prozessoren über Track-Ende.""" with self._lock: for processor in self._processors: if processor.enabled: try: processor.on_track_end(context) except Exception as e: self._logger.error( f"Fehler in on_track_end von {processor.id}: {e}" ) def reset_all(self) -> None: """Setzt alle Prozessoren zurück.""" with self._lock: for processor in self._processors: try: processor.reset() except Exception as e: self._logger.error( f"Fehler in reset von {processor.id}: {e}" ) # Zustand zurücksetzen self._is_wakeword_active = False self._is_conversation_active = False self._is_tts_playing = False self._current_track = None self._next_track = None # ========================================================================== # Status # ========================================================================== def get_status(self) -> dict: """Liefert Pipeline-Status.""" with self._lock: return { "processor_count": len(self._processors), "active_count": sum(1 for p in self._processors if p.enabled), "is_wakeword_active": self._is_wakeword_active, "is_conversation_active": self._is_conversation_active, "is_tts_playing": self._is_tts_playing, "processors": [ { "id": p.id, "name": p.name, "priority": p.priority, "enabled": p.enabled, } for p in self._processors ], } def __len__(self) -> int: return len(self._processors) def __contains__(self, processor_id: str) -> bool: return any(p.id == processor_id for p in self._processors) def __repr__(self) -> str: return f"AudioProcessingPipeline(processors={len(self._processors)})"