Trixy7/trixy_core/satellite/connection_notification.py

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