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