# -*- coding: utf-8 -*- """ Connection Notification Handler. Spielt Benachrichtigungen ab, wenn ein Satellite sich verbindet. """ import asyncio from pathlib import Path from typing import TYPE_CHECKING from trixy_core.config.datasets.server import ConnectionNotification from trixy_core.utils.debug import pdebug, pinfo, perror if TYPE_CHECKING: from trixy_core.application import IApplication class ConnectionNotificationHandler: """ Handler für Verbindungsbenachrichtigungen. Reagiert auf satellite_connected Events und spielt: - TTS "Verbunden" auf dem Client - Wave-Datei "connected.wav" auf dem Music-Stream - Beides - Nichts Konfiguration über server_config.satellite_connection. """ def __init__(self, application: "IApplication") -> None: """ Initialisiert den Handler. Args: application: Referenz zur Hauptanwendung """ self._application = application self._registered = False @property def _config(self): """Holt die Notification-Konfiguration.""" return self._application.server_config.satellite_connection @property def _notification_flags(self) -> ConnectionNotification: """Holt die Notification-Flags.""" return ConnectionNotification(self._config.notification) def register(self) -> None: """Registriert den Event-Handler.""" if self._registered: pdebug("ConnectionNotificationHandler: Bereits registriert") return # EventManager ist als 'events' Property verfügbar events = getattr(self._application, "events", None) if events: events.register("satellite_connected", self._on_satellite_connected) self._registered = True flags = self._notification_flags pdebug(f"ConnectionNotificationHandler registriert (Flags: {flags}, TTS: {bool(flags & ConnectionNotification.AUDIO_TTS)}, Wave: {bool(flags & ConnectionNotification.AUDIO_WAVE)})") else: perror("ConnectionNotificationHandler: Kein EventManager verfügbar") def unregister(self) -> None: """Deregistriert den Event-Handler.""" if not self._registered: return events = getattr(self._application, "events", None) if events: events.unregister("satellite_connected", self._on_satellite_connected) self._registered = False async def _on_satellite_connected(self, event_name: str, event_data) -> None: """ Handler für satellite_connected Event. Args: event_name: Name des Events event_data: Event-Daten mit satellite_id (dict oder Objekt) """ pdebug(f"ConnectionNotificationHandler: Event empfangen") flags = self._notification_flags pdebug(f"ConnectionNotificationHandler: Flags = {flags} ({int(flags)})") if flags == ConnectionNotification.NONE: pdebug("ConnectionNotificationHandler: Notification ist NONE, überspringe") return # Event-Data kann dict oder Objekt sein if isinstance(event_data, dict): satellite_id = event_data.get("satellite_id") alias = event_data.get("alias", "") else: satellite_id = getattr(event_data, "satellite_id", None) alias = getattr(event_data, "alias", "") if not satellite_id: perror("ConnectionNotificationHandler: Keine satellite_id in Event-Daten") return pdebug(f"Verbindungsbenachrichtigung für: {alias} ({satellite_id})") # Warte bis alle benötigten Sockets verbunden sind sockets_ready = await self._wait_for_sockets(satellite_id, flags) if not sockets_ready: perror(f"Sockets nicht bereit für Benachrichtigung: {alias}") return # TTS-Benachrichtigung abspielen falls aktiviert if flags & ConnectionNotification.AUDIO_TTS: asyncio.create_task(self._play_tts(satellite_id)) # Musik-Wiedergabe läuft über MusicPlayerService (reagiert auf music_socket_connected) # Der MusicPlayerService erstellt automatisch eine Playlist aus dem Assets-Ordner pdebug("Musik-Wiedergabe wird von MusicPlayerService gesteuert") async def _wait_for_sockets( self, satellite_id: str, flags: ConnectionNotification, timeout: float = 10.0 ) -> bool: """ Wartet bis die benötigten Audio-Sockets verbunden sind. Die Audio-Sockets (audio_out, music_out) verbinden sich separat nach dem Command-Socket auf eigenen Ports. Args: satellite_id: ID des Satellites flags: Benachrichtigungs-Flags timeout: Maximale Wartezeit in Sekunden Returns: True wenn alle benötigten Sockets bereit sind """ satellite = self._application.satellites.get(satellite_id) if not satellite: return False start_time = asyncio.get_event_loop().time() check_interval = 0.2 # Alle 200ms prüfen while asyncio.get_event_loop().time() - start_time < timeout: # Prüfe ob Satellite noch verbunden ist if not satellite.is_connected: return False # Prüfe benötigte Sockets tts_ready = True wave_ready = True if flags & ConnectionNotification.AUDIO_TTS: # TTS braucht audio_out Socket tts_ready = satellite.sockets.audio_out is not None if flags & ConnectionNotification.AUDIO_WAVE: # Wave braucht music_out Socket wave_ready = satellite.sockets.music_out is not None if tts_ready and wave_ready: pdebug(f"Alle Sockets bereit für: {satellite.alias}") return True await asyncio.sleep(check_interval) pdebug(f"Timeout beim Warten auf Sockets für: {satellite.alias}") return False async def _play_tts(self, satellite_id: str) -> bool: """ Spielt TTS-Benachrichtigung auf dem Satellite. Synthetisiert den Text direkt und sendet das Audio zum audio_out Socket des Satellites. Args: satellite_id: ID des Satellites Returns: True bei Erfolg """ try: from trixy_core.utils.template_formatter import format_template text = format_template(self._config.tts_text, application=self._application) satellite = self._application.satellites.get(satellite_id) if not satellite or not satellite.is_connected: perror(f"Satellite nicht verbunden für TTS: {satellite_id}") return False if not satellite.sockets.audio_out: perror(f"Audio-out Socket nicht verfügbar: {satellite_id}") return False pdebug(f"TTS-Benachrichtigung: '{text}' an {satellite.alias}") # Versuche TTS-Provider aus Plugin-System zu bekommen audio_data = await self._synthesize_tts(text) if not audio_data: perror("TTS-Synthese fehlgeschlagen") return False # Audio direkt zum Satellite senden pdebug(f"[SERVER] Sende TTS-Audio ({len(audio_data)} bytes) an {satellite.alias} via audio_out") success = await satellite.say(audio_data) if success: pinfo(f"[SERVER] TTS-Benachrichtigung gesendet: '{text}' an {satellite.alias} ({len(audio_data)} bytes)") # TTSStop senden damit Client weiß dass Stream beendet ist from trixy_core.network.cmd import TTSStop network = getattr(self._application.services, "network", None) if network and hasattr(network, "send_to_satellite"): await network.send_to_satellite(satellite_id, TTSStop()) pdebug(f"[SERVER] TTSStop gesendet an {satellite.alias}") else: perror(f"[SERVER] TTS-Audio senden fehlgeschlagen für {satellite.alias}") return success except Exception as e: perror(f"TTS-Benachrichtigung fehlgeschlagen: {e}") return False async def _synthesize_tts(self, text: str) -> bytes | None: """ Synthetisiert Text zu Audio. Versucht verschiedene Methoden: 1. TTS-Provider aus Plugin-System 2. Event-basierte TTS-Synthese Args: text: Zu synthetisierender Text Returns: Audio-Bytes oder None bei Fehler """ # Versuche über Event-System mit Callback audio_result: bytes | None = None synthesis_done = asyncio.Event() async def on_tts_completed(event_name: str, data) -> None: nonlocal audio_result # EventData und dict haben beide .get() Methode audio_data_hex = data.get("audio_data") if hasattr(data, "get") else None if audio_data_hex: try: audio_result = bytes.fromhex(audio_data_hex) pdebug(f"TTS-Audio empfangen: {len(audio_result)} bytes") except (ValueError, AttributeError) as e: perror(f"TTS-Audio Dekodierung fehlgeschlagen: {e}") synthesis_done.set() events = getattr(self._application, "events", None) if not events: perror("ConnectionNotificationHandler: Kein EventManager für TTS-Synthese") return None # Temporären Handler registrieren events.register("tts_completed", on_tts_completed) try: # TTS-Request auslösen (als dict, wie vom Piper Plugin erwartet) import uuid request = { "request_id": str(uuid.uuid4()), "satellite_id": "", # Nicht wichtig für Synthese "text": text, "voice_id": None, "source": "connection_notification" } pdebug(f"ConnectionNotificationHandler: Triggere tts_request für '{text}'") await events.emit("tts_request", request) # Auf Synthese warten (max 10 Sekunden) try: await asyncio.wait_for(synthesis_done.wait(), timeout=10.0) except asyncio.TimeoutError: pdebug("TTS-Synthese Timeout") return None return audio_result finally: events.unregister("tts_completed", on_tts_completed) async def _play_crossfade_mix(self, satellite_id: str) -> bool: """ Spielt einen Mix mehrerer Songs mit Crossfade. Lädt mehrere MP3-Dateien aus dem Assets-Ordner und spielt sie mit Crossfade-Übergängen ab. Args: satellite_id: ID des Satellites Returns: True bei Erfolg """ try: assets_dir = self._application.server_config.assets_directory profile = self._application.server_config.profile # Liste der Songs für den Mix (in Reihenfolge) song_files = [ "06 - Drumsound & Bassline Smith - Cold Turkey (G Dub, Beat City VIP).mp3", "Alex Clare - Too Close.mp3", "Alice Cooper - Poison.mp3", "Apache 207 - Roller.mp3", ] # Songs finden song_paths = [] for song_file in song_files: path = self._find_music_file(song_file, assets_dir, profile) if path: song_paths.append(path) pdebug(f"Song gefunden: {song_file}") else: pdebug(f"Song nicht gefunden: {song_file}") if len(song_paths) < 2: perror("Nicht genug Songs für Crossfade-Mix gefunden") return False pinfo(f"[SERVER] Starte Crossfade-Mix mit {len(song_paths)} Songs an {satellite_id}") return await self._stream_crossfade_mix(satellite_id, song_paths) except Exception as e: perror(f"Crossfade-Mix fehlgeschlagen: {e}") import traceback traceback.print_exc() return False async def _stream_crossfade_mix( self, satellite_id: str, song_paths: list[Path], crossfade_duration_ms: int = 5000, song_duration_seconds: float = 30.0, ) -> bool: """ Streamt einen Crossfade-Mix mehrerer Songs. Args: satellite_id: ID des Satellites song_paths: Liste der Song-Pfade crossfade_duration_ms: Crossfade-Dauer in Millisekunden song_duration_seconds: Wie lange jeder Song gespielt wird (0 = komplett) Returns: True bei Erfolg """ from trixy_core.music.formats.mp3 import MP3Decoder from plugins.crossfade.processor import crossfade_audio try: # Satellite und Socket holen satellite = self._application.satellites.get(satellite_id) if not satellite or not satellite.is_connected: perror(f"Satellite nicht verbunden: {satellite_id}") return False music_socket = satellite.sockets.music_out if not music_socket: perror(f"Music-Socket nicht verfügbar für: {satellite_id}") return False # Audio-Format aus Config audio_config = self._application.server_config.audio target_sample_rate = audio_config.music_sample_rate target_channels = audio_config.music_channels bytes_per_second = target_sample_rate * target_channels * 2 # 16-bit # MP3-Decoder decoder = MP3Decoder() if not decoder._check_dependencies(): perror("MP3-Decoder nicht verfügbar") return False # Songs dekodieren (nur Ausschnitte wenn song_duration_seconds > 0) decoded_songs: list[bytes] = [] for i, song_path in enumerate(song_paths): pinfo(f"[SERVER] Dekodiere Song {i+1}/{len(song_paths)}: {song_path.name}") decoded = await decoder.decode( song_path, target_sample_rate=target_sample_rate, target_channels=target_channels ) audio_data = decoded.pcm_data # Nur Ausschnitt nehmen wenn gewünscht if song_duration_seconds > 0: max_bytes = int(song_duration_seconds * bytes_per_second) audio_data = audio_data[:max_bytes] decoded_songs.append(audio_data) pdebug(f" Dekodiert: {len(audio_data)} bytes ({len(audio_data) / bytes_per_second:.1f}s)") # Crossfade-Mix erstellen pinfo(f"[SERVER] Erstelle Crossfade-Mix ({crossfade_duration_ms}ms Übergang)") mixed_audio = decoded_songs[0] for i in range(1, len(decoded_songs)): pinfo(f"[SERVER] Crossfade {i}: {song_paths[i-1].stem[:30]}... -> {song_paths[i].stem[:30]}...") mixed_audio = crossfade_audio( mixed_audio, decoded_songs[i], crossfade_ms=crossfade_duration_ms, sample_rate=target_sample_rate, channels=target_channels, fade_curve="cosine", ) total_duration = len(mixed_audio) / bytes_per_second pinfo(f"[SERVER] Mix erstellt: {len(mixed_audio)} bytes ({total_duration:.1f}s)") # Mix streamen return await self._stream_pcm_data( satellite_id, mixed_audio, target_sample_rate, target_channels, f"Crossfade-Mix ({len(song_paths)} Songs)", ) except Exception as e: perror(f"Fehler beim Crossfade-Mix: {e}") import traceback traceback.print_exc() return False async def _stream_pcm_data( self, satellite_id: str, audio_data: bytes, sample_rate: int, channels: int, description: str = "Audio", ) -> bool: """ Streamt rohe PCM-Daten zum Satellite. Args: satellite_id: ID des Satellites audio_data: PCM-Audio-Daten (16-bit) sample_rate: Sample-Rate channels: Anzahl Kanäle description: Beschreibung für Logs Returns: True bei Erfolg """ import time try: satellite = self._application.satellites.get(satellite_id) if not satellite or not satellite.is_connected: return False music_socket = satellite.sockets.music_out if not music_socket: return False sample_width = 2 # 16-bit bytes_per_second = sample_rate * channels * sample_width # Chunk-Größe: ~100ms Audio chunk_duration_ms = 100 chunk_size = (bytes_per_second * chunk_duration_ms) // 1000 frame_size = channels * sample_width chunk_size = (chunk_size // frame_size) * frame_size # Prebuffer: 5 Sekunden prebuffer_bytes = int(bytes_per_second * 5.0) total_chunks = (len(audio_data) + chunk_size - 1) // chunk_size duration_seconds = len(audio_data) / bytes_per_second pinfo(f"[SERVER] Starte Streaming: {description}") pinfo(f"[SERVER] {len(audio_data)} bytes, {duration_seconds:.1f}s, {total_chunks} chunks") stream_start_time = time.monotonic() bytes_sent = 0 chunk_count = 0 offset = 0 while offset < len(audio_data): chunk = audio_data[offset:offset + chunk_size] try: music_socket.write(chunk) await music_socket.drain() bytes_sent += len(chunk) chunk_count += 1 except Exception as e: perror(f"[SERVER] Fehler beim Senden: {e}") return False offset += chunk_size # Rate-Limiting nach Prebuffer if bytes_sent <= prebuffer_bytes: if chunk_count % 50 == 0: await asyncio.sleep(0) else: expected_time = (bytes_sent - prebuffer_bytes) / (bytes_per_second * 1.05) elapsed_time = time.monotonic() - stream_start_time if elapsed_time < expected_time: await asyncio.sleep(expected_time - elapsed_time) elapsed_total = time.monotonic() - stream_start_time pinfo(f"[SERVER] Streaming beendet: {description} in {elapsed_total:.1f}s") # StreamStop senden from trixy_core.network.cmd import StreamStop network = getattr(self._application.services, "network", None) if network and hasattr(network, "send_to_satellite"): await network.send_to_satellite(satellite_id, StreamStop(stream_type="music")) return True except Exception as e: perror(f"Fehler beim PCM-Streaming: {e}") return False async def _play_wave(self, satellite_id: str) -> bool: """ Spielt Audio-Datei auf dem Music-Stream des Satellites. Unterstützt Wave und MP3-Dateien. Args: satellite_id: ID des Satellites Returns: True bei Erfolg """ try: assets_dir = self._application.server_config.assets_directory profile = self._application.server_config.profile # TEST: MP3-Datei statt Wave für Musik-Stream-Test music_file = "Alex Clare - Too Close.mp3" music_path = self._find_music_file(music_file, assets_dir, profile) if not music_path: perror(f"Musik-Datei nicht gefunden: {music_file}") return False pdebug(f"Musik-Benachrichtigung: '{music_path}' an {satellite_id}") return await self._stream_music_to_satellite(satellite_id, music_path) except Exception as e: perror(f"Musik-Benachrichtigung fehlgeschlagen: {e}") return False def _find_asset_file( self, filename: str, assets_dir: str, profile: str ) -> Path | None: """ Sucht eine Asset-Datei im Profil oder Default-Verzeichnis. Suchpfade (in dieser Reihenfolge): 1. assets/{profile}/audio/{filename} 2. assets/{profile}/{filename} 3. assets/default/audio/{filename} 4. assets/default/{filename} 5. assets/{filename} Args: filename: Dateiname assets_dir: Assets-Verzeichnis profile: Profilname Returns: Pfad zur Datei oder None """ assets_path = Path(assets_dir) # Suchpfade in Prioritätsreihenfolge search_paths = [ assets_path / profile / "audio" / filename, assets_path / profile / filename, assets_path / "default" / "audio" / filename, assets_path / "default" / filename, assets_path / filename, ] for path in search_paths: if path.exists(): return path return None def _find_music_file( self, filename: str, assets_dir: str, profile: str ) -> Path | None: """ Sucht eine Musik-Datei im Profil oder Default-Verzeichnis. Suchpfade (in dieser Reihenfolge): 1. assets/{profile}/music/{filename} 2. assets/default/music/{filename} Args: filename: Dateiname assets_dir: Assets-Verzeichnis profile: Profilname Returns: Pfad zur Datei oder None """ assets_path = Path(assets_dir) # Suchpfade in Prioritätsreihenfolge search_paths = [ assets_path / profile / "music" / filename, assets_path / "default" / "music" / filename, ] for path in search_paths: if path.exists(): return path return None async def _stream_wave_to_satellite( self, satellite_id: str, wave_path: Path ) -> bool: """ Streamt eine Wave-Datei zum Music-Socket des Satellites. Args: satellite_id: ID des Satellites wave_path: Pfad zur Wave-Datei Returns: True bei Erfolg """ import wave import struct try: # Satellite holen satellite = self._application.satellites.get(satellite_id) if not satellite or not satellite.is_connected: perror(f"Satellite nicht verbunden: {satellite_id}") return False music_socket = satellite.sockets.music_out if not music_socket: perror(f"Music-Socket nicht verfügbar für: {satellite_id}") return False # Wave-Datei lesen und konvertieren with wave.open(str(wave_path), "rb") as wf: sample_rate = wf.getframerate() channels = wf.getnchannels() sample_width = wf.getsampwidth() n_frames = wf.getnframes() pdebug(f"Wave: {sample_rate}Hz, {channels}ch, {sample_width*8}bit, {n_frames} frames") # Audio-Daten lesen audio_data = wf.readframes(n_frames) # Konvertierung zu 44100Hz Stereo falls nötig (Standard für Musik-Stream) target_rate = 44100 target_channels = 2 if sample_rate != target_rate or channels != target_channels: audio_data = self._convert_audio( audio_data, sample_rate, channels, sample_width, target_rate, target_channels, ) # In Chunks aufteilen und senden (passend zur Client Buffer-Größe) chunk_size = 40960 # Bytes pro Chunk offset = 0 total_chunks = (len(audio_data) + chunk_size - 1) // chunk_size pdebug(f"[SERVER] Starte Wave-Streaming: {wave_path.name} ({len(audio_data)} bytes, {total_chunks} chunks) an {satellite.alias} via music_out") chunk_count = 0 while offset < len(audio_data): chunk = audio_data[offset:offset + chunk_size] try: music_socket.write(chunk) await music_socket.drain() chunk_count += 1 except Exception as e: perror(f"[SERVER] Fehler beim Senden von Audio-Chunk {chunk_count}/{total_chunks}: {e}") return False offset += chunk_size # Minimale Pause für Flow-Control if offset < len(audio_data): await asyncio.sleep(0.001) pinfo(f"[SERVER] Wave-Datei gestreamt: {wave_path.name} ({chunk_count} chunks, {len(audio_data)} bytes) an {satellite.alias}") # StreamStop senden damit Client weiß dass Stream beendet ist from trixy_core.network.cmd import StreamStop network = getattr(self._application.services, "network", None) if network and hasattr(network, "send_to_satellite"): await network.send_to_satellite(satellite_id, StreamStop(stream_type="music")) pdebug(f"[SERVER] StreamStop (music) gesendet an {satellite.alias}") return True except Exception as e: perror(f"Fehler beim Streaming der Wave-Datei: {e}") return False async def _stream_music_to_satellite( self, satellite_id: str, music_path: Path ) -> bool: """ Streamt eine Musik-Datei (MP3/WAV) zum Music-Socket des Satellites. Verwendet den MP3Decoder für MP3-Dateien. Args: satellite_id: ID des Satellites music_path: Pfad zur Musik-Datei Returns: True bei Erfolg """ try: # Satellite holen satellite = self._application.satellites.get(satellite_id) if not satellite or not satellite.is_connected: perror(f"Satellite nicht verbunden: {satellite_id}") return False music_socket = satellite.sockets.music_out if not music_socket: perror(f"Music-Socket nicht verfügbar für: {satellite_id}") return False # Datei-Extension prüfen suffix = music_path.suffix.lower() if suffix == ".mp3": # MP3 mit Decoder dekodieren from trixy_core.music.formats.mp3 import MP3Decoder decoder = MP3Decoder() if not decoder._check_dependencies(): perror("MP3-Decoder nicht verfügbar (pydub oder ffmpeg erforderlich)") return False # Audio-Format aus Server-Config holen audio_config = self._application.server_config.audio target_sample_rate = audio_config.music_sample_rate target_channels = audio_config.music_channels target_bit_depth = getattr(audio_config, 'music_bit_depth', 16) pinfo(f"[SERVER] Dekodiere MP3: {music_path.name} -> {target_sample_rate}Hz, {target_channels}ch") decoded = await decoder.decode( music_path, target_sample_rate=target_sample_rate, target_channels=target_channels ) audio_data = decoded.pcm_data pdebug(f"[SERVER] MP3 dekodiert: {len(audio_data)} bytes, {decoded.sample_rate}Hz, {decoded.channels}ch") elif suffix == ".wav": # Audio-Format aus Server-Config holen audio_config = self._application.server_config.audio target_sample_rate = audio_config.music_sample_rate target_channels = audio_config.music_channels target_bit_depth = getattr(audio_config, 'music_bit_depth', 16) target_sample_width = target_bit_depth // 8 # Wave-Datei direkt lesen import wave with wave.open(str(music_path), "rb") as wf: src_sample_rate = wf.getframerate() src_channels = wf.getnchannels() src_sample_width = wf.getsampwidth() n_frames = wf.getnframes() audio_data = wf.readframes(n_frames) # Konvertierung zum Zielformat falls nötig if src_sample_rate != target_sample_rate or src_channels != target_channels: audio_data = self._convert_audio( audio_data, src_sample_rate, src_channels, src_sample_width, target_sample_rate, target_channels ) else: perror(f"Nicht unterstütztes Audio-Format: {suffix}") return False # Audio-Format aus Config für Rate-Limited Streaming audio_config = self._application.server_config.audio sample_rate = audio_config.music_sample_rate channels = audio_config.music_channels sample_width = getattr(audio_config, 'music_bit_depth', 16) // 8 bytes_per_second = sample_rate * channels * sample_width # Chunk-Größe: ~100ms Audio pro Chunk chunk_duration_ms = 100 chunk_size = (bytes_per_second * chunk_duration_ms) // 1000 # Auf Frame-Grenze ausrichten frame_size = channels * sample_width chunk_size = (chunk_size // frame_size) * frame_size # Prebuffer: Sende zuerst 5 Sekunden sofort (genug für stabilen Start) prebuffer_bytes = int(bytes_per_second * 5.0) total_chunks = (len(audio_data) + chunk_size - 1) // chunk_size duration_seconds = len(audio_data) / bytes_per_second pinfo(f"[SERVER] Starte Rate-Limited Musik-Streaming: {music_path.name}") pinfo(f"[SERVER] {len(audio_data)} bytes, {duration_seconds:.1f}s, {total_chunks} chunks à {chunk_duration_ms}ms") pinfo(f"[SERVER] Prebuffer: {prebuffer_bytes} bytes ({prebuffer_bytes * 1000 // bytes_per_second}ms)") pinfo(f"[SERVER] Format: {sample_rate}Hz, {channels}ch, {sample_width*8}-bit") import time stream_start_time = time.monotonic() bytes_sent = 0 chunk_count = 0 offset = 0 while offset < len(audio_data): chunk = audio_data[offset:offset + chunk_size] try: music_socket.write(chunk) await music_socket.drain() bytes_sent += len(chunk) chunk_count += 1 except Exception as e: perror(f"[SERVER] Fehler beim Senden von Musik-Chunk {chunk_count}/{total_chunks}: {e}") return False offset += chunk_size # Da wir als Background-Task laufen, brauchen wir minimale Pausen if bytes_sent <= prebuffer_bytes: # Prebuffer: So schnell wie möglich senden, nur yield für Event-Loop if chunk_count % 50 == 0: # Alle 50 Chunks (5 Sekunden) kurz yielden await asyncio.sleep(0) # Nur yield, keine echte Pause else: # Nach Prebuffer: Rate-Limiting, aber leicht schneller als Echtzeit # Sende mit 105% der Echtzeit-Rate, damit Buffer sich füllt expected_time = (bytes_sent - prebuffer_bytes) / (bytes_per_second * 1.05) elapsed_time = time.monotonic() - stream_start_time if elapsed_time < expected_time: wait_time = expected_time - elapsed_time await asyncio.sleep(wait_time) # Kein else - wenn wir zu langsam sind, nicht pausieren! elapsed_total = time.monotonic() - stream_start_time pinfo(f"[SERVER] Musik gestreamt: {music_path.name} ({chunk_count} chunks, {bytes_sent} bytes) in {elapsed_total:.1f}s an {satellite.alias}") # StreamStop senden damit Client weiß dass Stream beendet ist from trixy_core.network.cmd import StreamStop network = getattr(self._application.services, "network", None) if network and hasattr(network, "send_to_satellite"): await network.send_to_satellite(satellite_id, StreamStop(stream_type="music")) pdebug(f"[SERVER] StreamStop (music) gesendet an {satellite.alias}") return True except Exception as e: perror(f"Fehler beim Musik-Streaming: {e}") import traceback traceback.print_exc() return False def _convert_audio( self, audio_data: bytes, src_rate: int, src_channels: int, sample_width: int, dst_rate: int, dst_channels: int, ) -> bytes: """ Konvertiert Audio-Daten (einfache Implementierung). Args: audio_data: Quell-Audio-Daten src_rate: Quell-Samplerate src_channels: Quell-Kanäle sample_width: Sample-Breite in Bytes dst_rate: Ziel-Samplerate dst_channels: Ziel-Kanäle Returns: Konvertierte Audio-Daten """ import array # Samples extrahieren if sample_width == 2: samples = array.array("h", audio_data) elif sample_width == 1: # 8-bit zu 16-bit konvertieren samples = array.array("h", [(s - 128) * 256 for s in audio_data]) else: # Unsupportet, zurückgeben wie es ist return audio_data # Mono zu Stereo if src_channels == 1 and dst_channels == 2: stereo_samples = [] for sample in samples: stereo_samples.extend([sample, sample]) samples = array.array("h", stereo_samples) # Stereo zu Mono (falls benötigt) elif src_channels == 2 and dst_channels == 1: mono_samples = [] for i in range(0, len(samples), 2): mono_samples.append((samples[i] + samples[i + 1]) // 2) samples = array.array("h", mono_samples) # Resampling (einfache lineare Interpolation) if src_rate != dst_rate: ratio = dst_rate / src_rate new_length = int(len(samples) * ratio) resampled = [] for i in range(new_length): src_idx = i / ratio idx = int(src_idx) frac = src_idx - idx if idx + 1 < len(samples): sample = int(samples[idx] * (1 - frac) + samples[idx + 1] * frac) else: sample = samples[idx] if idx < len(samples) else 0 resampled.append(sample) samples = array.array("h", resampled) return samples.tobytes() def setup_connection_notification(application: "IApplication") -> ConnectionNotificationHandler: """ Richtet den Connection Notification Handler ein. Args: application: Referenz zur Hauptanwendung Returns: Der registrierte Handler """ handler = ConnectionNotificationHandler(application) handler.register() return handler