Trixy7/trixy_core/client.py

# -*- coding: utf-8 -*-
"""
Client/Satellite-Anwendung für Trixy.

Der Client ist ein leichtgewichtiger Satellite, der:
- Wakeword-Erkennung durchführt
- Audio zum Server streamt
- Audio-Ausgaben abspielt
"""

import asyncio
import wave
import time
from pathlib import Path

from trixy_core.application import IApplication
from trixy_core.config.datasets.client import ClientConfig
from trixy_core.events.event_data.basic import SystemShutdown
from trixy_core.network.client_connection import ClientConnection, ConnectionConfig, get_mac_address
from trixy_core.network.protocol import ProtocolMessage
from trixy_core.audio.player import AudioOutputManager
from trixy_core.input.keyboard_input import create_keyboard_input, KeyboardInputService, SimpleKeyboardInput
from trixy_core.wakeword.service import WakewordService, WakewordServiceConfig
from trixy_core.wakeword.detector import DetectorConfig
from trixy_core.wakeword.audio_buffer import BufferConfig
from trixy_core.wakeword.vad import VADConfig
from trixy_core.audio.microphone import MicrophoneCapture
from trixy_core.network.config_listener import ConfigListener
from trixy_core.utils.debug import pinfo, pdebug, perror, pwarn


class ClientApplication(IApplication):
    """
    Client/Satellite-Anwendung für Trixy.

    Verbindet sich mit einem Server und führt
    lokale Wakeword-Erkennung durch.
    """

    def __init__(
        self,
        host: str | None = None,
        port: int | None = None,
        room: str | None = None,
        alias: str | None = None,
        config_path: str | Path = "config/client_config.json",
        debug: bool = False,
        keyboard_input: bool = False,
    ) -> None:
        """
        Initialisiert den Client.

        Args:
            host: Server-Host
            port: Server-Port
            room: Raum-Kennung
            alias: Anzeigename
            config_path: Pfad zur Konfigurationsdatei
            debug: Debug-Modus aktivieren
            keyboard_input: Keyboard-Eingabe aktivieren (Ctrl+I oder Enter)
        """
        super().__init__(debug)

        self._config_path = Path(config_path)
        self._client_config: ClientConfig | None = None

        # Überschreibbare CLI-Parameter (None = aus Config lesen)
        self._cli_host: str | None = host
        self._cli_port: int | None = port
        self._cli_room: str | None = room
        self._cli_alias: str | None = alias
        # Im Debug-Modus immer Keyboard-Shortcuts aktivieren (Ctrl+P/N/S)
        self._keyboard_input_enabled = keyboard_input or debug

        # Verbindungsstatus
        self._connected = False
        self._reconnect_task: asyncio.Task | None = None
        self._connection: ClientConnection | None = None

        # Audio-Ausgabe Manager
        self._audio_manager: AudioOutputManager | None = None

        # Keyboard-Input Service
        self._keyboard_service: KeyboardInputService | SimpleKeyboardInput | None = None

        # Wakeword-Service
        self._wakeword_service: WakewordService | None = None
        self._microphone: MicrophoneCapture | None = None

        # ConfigListener
        self._config_listener: ConfigListener | None = None

        # Server-gesteuerte Aufnahme (Fallback ohne WakewordService)
        self._direct_streaming_active: bool = False

        # DEBUG: Audio-Chunks für WAV-Export sammeln
        self._debug_tts_chunks: list[bytes] = []
        self._debug_music_chunks: list[bytes] = []

    @property
    def client_config(self) -> ClientConfig:
        """Client-Konfiguration."""
        if self._client_config is None:
            raise RuntimeError("Client nicht initialisiert")
        return self._client_config

    @property
    def is_connected(self) -> bool:
        """Ist der Client mit dem Server verbunden?"""
        return self._connected and self._connection is not None and self._connection.is_connected

    @property
    def connection(self) -> ClientConnection | None:
        """Die aktive Verbindung zum Server."""
        return self._connection

    @property
    def audio_manager(self) -> AudioOutputManager | None:
        """Audio-Ausgabe Manager."""
        return self._audio_manager

    @property
    def server_host(self) -> str:
        """Server-Host."""
        if self._cli_host is not None:
            return self._cli_host
        if self._client_config:
            return self._client_config.server.host
        return "localhost"

    @property
    def server_port(self) -> int:
        """Server-Port."""
        if self._cli_port is not None:
            return self._cli_port
        if self._client_config:
            return self._client_config.server.port
        return 2101

    @property
    def room(self) -> str:
        """Raum-Kennung."""
        if self._cli_room is not None:
            return self._cli_room
        if self._client_config:
            return self._client_config.identity.room
        return ""

    @property
    def alias(self) -> str:
        """Anzeigename."""
        if self._cli_alias is not None:
            return self._cli_alias
        if self._client_config:
            return self._client_config.identity.alias
        return ""

    async def initialize(self) -> None:
        """Initialisiert den Client."""
        pinfo("Initialisiere Client...")

        # Konfiguration laden (optional, kann fehlen)
        if self._config_path.exists():
            self._client_config = self.config_manager.load(
                self._config_path,
                ClientConfig,
                name="client",
                auto_reload=True
            )
        else:
            self._client_config = ClientConfig()
            pwarn(f"Keine Konfigurationsdatei gefunden: {self._config_path}")

        # CLI-Parameter in Konfiguration überschreiben
        if self._cli_host is not None:
            self._client_config.server.host = self._cli_host
        if self._cli_port is not None:
            self._client_config.server.port = self._cli_port
        if self._cli_room is not None:
            self._client_config.identity.room = self._cli_room
        if self._cli_alias is not None:
            self._client_config.identity.alias = self._cli_alias

        # Audio-Ausgabe Manager initialisieren (Sample-Rates aus Config)
        audio_config = self._client_config.audio
        music_sample_rate = getattr(audio_config, 'music_sample_rate', 48000)
        music_channels = getattr(audio_config, 'music_channels', 2)

        self._audio_manager = AudioOutputManager(
            tts_sample_rate=22050,  # Piper Standard (wird vom Server überschrieben)
            music_sample_rate=music_sample_rate,
        )
        audio_ok = await self._audio_manager.initialize()
        if audio_ok:
            pinfo(f"Audio-Manager initialisiert: Musik={music_sample_rate}Hz, {music_channels}ch")
        else:
            pwarn("Audio-Manager konnte nicht initialisiert werden (keine Audio-Wiedergabe)")

        # Wakeword-Service erstellen
        ww_config = self._client_config.wakeword
        if ww_config.enabled:
            try:
                service_config = WakewordServiceConfig(
                    detector=DetectorConfig(
                        model_directory=ww_config.model_directory,
                        models=ww_config.models,
                        threshold=ww_config.threshold,
                        use_onnx=ww_config.use_onnx,
                    ),
                    buffer=BufferConfig(
                        pre_buffer_seconds=ww_config.pre_buffer_seconds,
                        max_duration_seconds=ww_config.max_recording_seconds,
                    ),
                    vad=VADConfig(
                        silence_duration_ms=int(ww_config.silence_timeout_seconds * 1000),
                        max_duration_ms=int(ww_config.max_recording_seconds * 1000),
                        no_speech_timeout_ms=int(ww_config.no_speech_timeout_seconds * 1000),
                    ),
                    standalone_mode=False,
                    wakeword_cooldown_seconds=ww_config.wakeword_cooldown_seconds,
                )
                self._wakeword_service = WakewordService(self, service_config)
                self._microphone = MicrophoneCapture(
                    device=self._client_config.audio.input_device,
                )
                pinfo("Wakeword-Service erstellt")
            except Exception as e:
                pwarn(f"Wakeword-Service nicht verfügbar: {e}")

        # HID Media Key Service (Konferenzmikrofone, Headsets)
        try:
            from trixy_core.hid import HIDService, HIDServiceConfig
            self._hid_service = HIDService(self, HIDServiceConfig())
            self.services.register_instance(self._hid_service)
            pdebug("HID-Service registriert")
        except Exception as e:
            pdebug(f"HID-Service nicht verfuegbar: {e}")

        # ALSA Volume Handler (Hardware-Lautstaerke via amixer)
        try:
            from trixy_core.audio.volume import VolumeHandler
            self._volume_handler = VolumeHandler(self)
            pdebug("Volume Handler erstellt")
        except Exception as e:
            pdebug(f"Volume Handler nicht verfuegbar: {e}")

        # ConfigListener erstellen und registrieren (Remote-Config-Tool Zugriff)
        if self._client_config.config_listener_enabled:
            self._config_listener = ConfigListener(
                self,
                port=self._client_config.config_port,
                bind_address="0.0.0.0",
                encryption_key_path="certs/encryption.key",
                instance_type="client",
            )
            self.services.register_instance(self._config_listener)
        else:
            pdebug("ConfigListener deaktiviert")

        pdebug(f"Client initialisiert: {self.alias} in {self.room}")

    async def start(self) -> None:
        """Startet den Client."""
        pinfo(f"Starte Client als '{self.alias}' in '{self.room}'...")

        from trixy_core.utils.version import VERSION_STRING
        await self.events.emit("application_starting", {
            "mode": "client",
            "version": VERSION_STRING,
        })

        # Services starten
        await self.services.start_all()

        # Keyboard-Input vorbereiten (falls aktiviert)
        # Wird nach Verbindungsaufbau gestartet um Connection zu haben
        if self._keyboard_input_enabled:
            self._keyboard_service = create_keyboard_input(
                self,
                connection=None,  # Wird nach Connect gesetzt
                prompt=f"{self.alias}> ",
            )

        # Verbindung zum Server herstellen
        await self._connect()

        # Volume Handler starten (nach Services, braucht Events)
        if hasattr(self, "_volume_handler"):
            await self._volume_handler.start()

        pinfo(f"Client gestartet, verbinde zu {self.server_host}:{self.server_port}")

        from trixy_core.utils.version import VERSION_STRING
        await self.events.emit("application_ready", {
            "mode": "client",
            "version": VERSION_STRING,
            "services_count": len(self.services.services),
        })

    async def stop(self) -> None:
        """Stoppt den Client."""
        pinfo("Stoppe Client...")
        await self.events.emit("application_stopping", {"reason": "shutdown"})

        # DEBUG: Gesammelte Audio-Chunks als WAV speichern
        if self._debug_tts_chunks:
            self._debug_save_wav(self._debug_tts_chunks, "tts")
            self._debug_tts_chunks = []
        if self._debug_music_chunks:
            self._debug_save_wav(self._debug_music_chunks, "music")
            self._debug_music_chunks = []

        # Wakeword-Service stoppen
        if self._wakeword_service:
            await self._wakeword_service.stop_listening()
            await self._wakeword_service.on_stop()
        if self._microphone:
            self._microphone.stop()

        # Keyboard-Input stoppen
        if self._keyboard_service:
            await self._keyboard_service.stop()
            self._keyboard_service = None

        # Reconnect-Task abbrechen
        if self._reconnect_task and not self._reconnect_task.done():
            self._reconnect_task.cancel()
            try:
                await self._reconnect_task
            except asyncio.CancelledError:
                pass

        # Shutdown-Event auslösen
        await self.events.trigger(
            "system_shutdown",
            SystemShutdown(reason="Client shutdown", source="client")
        )

        # Verbindung trennen
        await self._disconnect()

        # Audio-Manager beenden
        if self._audio_manager:
            await self._audio_manager.shutdown()
            self._audio_manager = None

        # Services stoppen
        await self.services.stop_all()

        # ConfigManager aufräumen
        self.config_manager.stop_watching()

        pinfo("Client gestoppt")

    async def _connect(self) -> bool:
        """
        Verbindet sich mit dem Server.

        Returns:
            True bei erfolgreicher Verbindung
        """
        try:
            pdebug(f"Verbinde zu {self.server_host}:{self.server_port}...")

            # MAC-Adresse ermitteln
            mac_address = self._client_config.identity.mac_address if self._client_config else ""
            if not mac_address:
                mac_address = get_mac_address()
                pdebug(f"MAC-Adresse ermittelt: {mac_address}")

            # Verbindungskonfiguration
            connection_config = ConnectionConfig(
                connect_timeout=self._client_config.server.connection_timeout_seconds if self._client_config else 10.0,
                reconnect_interval=self._client_config.server.reconnect_interval_seconds if self._client_config else 5.0,
            )

            # ClientConnection erstellen
            self._connection = ClientConnection(
                host=self.server_host,
                port=self.server_port,
                room=self.room,
                alias=self.alias,
                mac_address=mac_address,
                config=connection_config,
            )

            # Callbacks registrieren
            self._connection.on_connected(self._on_connected)
            self._connection.on_disconnected(self._on_disconnected)
            self._connection.on_message(self._on_message)
            self._connection.on_audio_out(self._on_audio_out)
            self._connection.on_music(self._on_music)

            # Verbinden
            success = await self._connection.connect()

            if success:
                self._connected = True
                pinfo(f"Verbunden mit Server {self.server_host}:{self.server_port}")

                # Audio-Format vom Server anwenden
                if self._audio_manager:
                    tts_format = self._connection.tts_format
                    music_format = self._connection.music_format

                    await self._audio_manager.update_tts_format(
                        sample_rate=tts_format["sample_rate"],
                        channels=tts_format["channels"],
                        sample_width=tts_format["sample_width"]
                    )
                    await self._audio_manager.update_music_format(
                        sample_rate=music_format["sample_rate"],
                        channels=music_format["channels"],
                        sample_width=music_format["sample_width"]
                    )
                    pinfo(f"Audio-Format: TTS={tts_format['sample_rate']}Hz, Musik={music_format['sample_rate']}Hz")

                # Event auslösen
                await self.events.emit("client_connected", {
                    "host": self.server_host,
                    "port": self.server_port,
                    "satellite_id": self._connection.satellite_id,
                })
                return True
            else:
                self._connected = False
                self._schedule_reconnect()
                return False

        except Exception as e:
            perror(f"Verbindungsfehler: {e}")
            self._connected = False
            self._schedule_reconnect()
            return False

    async def _on_connected(self) -> None:
        """Callback wenn Verbindung hergestellt."""
        pdebug("Verbindungs-Callback: connected")

        # Keyboard-Input mit Verbindung starten
        if self._keyboard_service and self._connection:
            self._keyboard_service.set_connection(self._connection)
            await self._keyboard_service.start()

        # Wakeword-Service konfigurieren und starten
        if self._wakeword_service and self._connection and self._microphone:
            self._wakeword_service.setup_for_client(self._connection, self._microphone)

            # Callback für Wakeword-getriggertes Recording-Stop registrieren
            connection = self._connection

            async def on_recording_stopped_by_wakeword(recording_id: str, duration: float) -> None:
                if connection:
                    from trixy_core.network.cmd import SatelliteRecordStopped
                    cmd = SatelliteRecordStopped(
                        recording_id=recording_id,
                        reason="wakeword",
                        duration_seconds=duration,
                    )
                    await connection.send_command(cmd)

            self._wakeword_service.set_server_recording_stopped_callback(
                on_recording_stopped_by_wakeword
            )

            try:
                await self._wakeword_service.on_start()
                await self._wakeword_service.start_listening()
                pinfo("Wakeword-Erkennung aktiv")
            except Exception as e:
                pwarn(f"Wakeword-Start fehlgeschlagen: {e}")

    async def _on_disconnected(self, reason: str) -> None:
        """Callback wenn Verbindung getrennt."""
        pwarn(f"Verbindung getrennt: {reason}")
        self._connected = False

        # Event auslösen
        await self.events.emit("client_disconnected", {
            "reason": reason,
        })

        # Wiederverbindung planen
        if self._running:
            self._schedule_reconnect()

    async def _on_message(self, message: ProtocolMessage) -> None:
        """Callback für eingehende Nachrichten vom Server."""
        pdebug(f"Nachricht empfangen: {message.class_name}")

        # Spezielle Behandlung für bestimmte Commands
        if message.class_name == "StreamStop":
            await self._handle_stream_stop(message.data)
        elif message.class_name == "TTSStop":
            await self._handle_tts_stop()
        elif message.class_name == "AudioConfigUpdate":
            await self._handle_audio_config_update(message.data)
        elif message.class_name == "FollowUpRequest":
            await self._handle_follow_up_request(message.data)
        elif message.class_name == "WakewordSelected":
            if self._wakeword_service:
                await self._wakeword_service.handle_wakeword_selected(message.data)
        elif message.class_name == "WakewordAbort":
            if self._wakeword_service:
                await self._wakeword_service.handle_wakeword_abort(message.data)
        elif message.class_name == "RecordingStop":
            if self._wakeword_service:
                await self._wakeword_service.handle_conversation_end()
        elif message.class_name == "ConversationEnd":
            await self._handle_conversation_end(message.data)
            if self._wakeword_service:
                await self._wakeword_service.handle_conversation_end()
        elif message.class_name == "WakewordPause":
            if self._wakeword_service:
                self._wakeword_service.external_pause()
        elif message.class_name == "WakewordResume":
            if self._wakeword_service:
                self._wakeword_service.external_resume()
        elif message.class_name == "WakewordControlStop":
            if self._wakeword_service:
                await self._wakeword_service.external_stop()
        elif message.class_name == "WakewordControlStart":
            if self._wakeword_service:
                await self._wakeword_service.external_start()
        elif message.class_name == "TextInputAck":
            await self._handle_text_input_ack(message.data)
        elif message.class_name == "ServerShutdown":
            await self._handle_server_shutdown(message.data)
        # Server-gesteuerte Aufnahme
        elif message.class_name == "SatelliteRecordStart":
            await self._handle_satellite_record_start(message.data)
        elif message.class_name == "SatelliteRecordStop":
            await self._handle_satellite_record_stop(message.data)
        # Media-Commands
        elif message.class_name == "MusicPlayPause":
            await self._handle_music_play_pause()
        elif message.class_name == "MusicNext":
            await self._handle_music_next(message.data)
        elif message.class_name == "MusicStop":
            await self._handle_music_stop()
        elif message.class_name == "MediaStopAll":
            await self._handle_media_stop_all()
        # Config-Proxy-Anfragen vom Server (fuer Config-Tool)
        elif message.class_name == "ConfigReadRequest":
            await self._handle_config_read_proxy(message.data)
        elif message.class_name == "ConfigWriteRequest":
            await self._handle_config_write_proxy(message.data)

        # Event auslösen
        await self.events.emit("network_message_received", {
            "class_name": message.class_name,
            "data": message.data,
            "flags": int(message.flags),
        })

    async def _handle_stream_stop(self, data: dict | object) -> None:
        """Behandelt Stream-Stop-Befehle vom Server."""
        if not self._audio_manager:
            return

        # Stream-Type und immediate Flag ermitteln
        stream_type = getattr(data, "stream_type", None) or "all"
        immediate = getattr(data, "immediate", False)
        if isinstance(data, dict):
            stream_type = data.get("stream_type", "all")
            immediate = data.get("immediate", False)

        pdebug(f"Stream-Stop empfangen: {stream_type}, immediate={immediate}")

        # DEBUG: WAV speichern vor Stop
        if stream_type in ("all", "tts") and self._debug_tts_chunks:
            self._debug_save_wav(self._debug_tts_chunks, "tts")
            self._debug_tts_chunks = []
        if stream_type in ("all", "music") and self._debug_music_chunks:
            self._debug_save_wav(self._debug_music_chunks, "music")
            self._debug_music_chunks = []

        # Bei immediate=True sofort stoppen, sonst Buffer abspielen (finish)
        if immediate:
            if stream_type == "all":
                self._audio_manager.stop_all()
            elif stream_type == "tts":
                self._audio_manager.stop_tts()
            elif stream_type == "music":
                self._audio_manager.stop_music()
        else:
            if stream_type == "all":
                self._audio_manager.finish_tts()
                self._audio_manager.finish_music()
            elif stream_type == "tts":
                self._audio_manager.finish_tts()
            elif stream_type == "music":
                self._audio_manager.finish_music()

        # Flags zurücksetzen
        if stream_type in ("all", "tts"):
            self._tts_playing = False
        if stream_type in ("all", "music"):
            self._music_playing = False

    async def _handle_tts_stop(self) -> None:
        """Behandelt TTS-Stop-Befehle vom Server."""
        # DEBUG: WAV speichern vor Stop
        if self._debug_tts_chunks:
            self._debug_save_wav(self._debug_tts_chunks, "tts")
            self._debug_tts_chunks = []

        # Flag zurücksetzen für nächsten TTS-Stream
        self._tts_playing = False

        if self._audio_manager:
            # finish_tts() signalisiert Ende und spielt Rest ab
            # stop_tts() würde sofort stoppen und Buffer leeren
            self._audio_manager.finish_tts()
            pdebug("TTS-Stream beendet (finish)")

    async def _handle_audio_config_update(self, data: dict | object) -> None:
        """
        Behandelt Audio-Konfigurationsänderungen vom Server.

        Wird aufgerufen wenn sich das TTS-Plugin ändert und
        der Client seine Audio-Ausgabe anpassen muss.
        """
        if not self._audio_manager:
            return

        # TTS-Format aktualisieren
        tts_sample_rate = getattr(data, "tts_sample_rate", 0)
        tts_channels = getattr(data, "tts_channels", 0)
        tts_sample_width = getattr(data, "tts_sample_width", 0)

        if isinstance(data, dict):
            tts_sample_rate = data.get("tts_sample_rate", 0)
            tts_channels = data.get("tts_channels", 0)
            tts_sample_width = data.get("tts_sample_width", 0)

        if tts_sample_rate > 0:
            await self._audio_manager.update_tts_format(
                sample_rate=tts_sample_rate,
                channels=tts_channels or 1,
                sample_width=tts_sample_width or 2
            )
            pinfo(f"TTS-Format aktualisiert: {tts_sample_rate}Hz, {tts_channels}ch")

        # Musik-Format aktualisieren (falls angegeben)
        music_sample_rate = getattr(data, "music_sample_rate", 0)
        music_channels = getattr(data, "music_channels", 0)
        music_sample_width = getattr(data, "music_sample_width", 0)

        if isinstance(data, dict):
            music_sample_rate = data.get("music_sample_rate", 0)
            music_channels = data.get("music_channels", 0)
            music_sample_width = data.get("music_sample_width", 0)

        if music_sample_rate > 0:
            await self._audio_manager.update_music_format(
                sample_rate=music_sample_rate,
                channels=music_channels or 2,
                sample_width=music_sample_width or 2
            )
            pinfo(f"Musik-Format aktualisiert: {music_sample_rate}Hz, {music_channels}ch")

    async def _handle_follow_up_request(self, data: dict | object) -> None:
        """
        Behandelt Follow-Up-Anfragen vom Server.

        Bei Keyboard-Input wird automatisch ein neuer Input-Prompt geöffnet.
        """
        from trixy_core.network.cmd import FollowUpRequest

        # Daten extrahieren
        if isinstance(data, FollowUpRequest):
            request = data
        else:
            session_id = getattr(data, "session_id", "") or data.get("session_id", "")
            question = getattr(data, "question", "") or data.get("question", "")
            request = FollowUpRequest(session_id=session_id, question=question)

        pdebug(f"Follow-Up-Request empfangen: {request.question[:50]}...")

        # Keyboard-Input für Antwort öffnen (falls aktiv)
        if self._keyboard_service:
            self._keyboard_service.handle_follow_up_request(request)

    async def _handle_conversation_end(self, data: dict | object) -> None:
        """
        Behandelt das Ende einer Conversation.

        Setzt den Keyboard-Input-Status zurück.
        """
        session_id = getattr(data, "session_id", "") or ""
        if isinstance(data, dict):
            session_id = data.get("session_id", "")

        pdebug(f"Conversation beendet: {session_id}")

        # Keyboard-Input zurücksetzen
        if self._keyboard_service:
            self._keyboard_service.handle_conversation_end(session_id)

    async def _handle_text_input_ack(self, data: dict | object) -> None:
        """
        Behandelt Bestätigung für Text-Eingabe.
        """
        session_id = getattr(data, "session_id", "") or ""
        received = getattr(data, "received", True)
        error = getattr(data, "error", "") or ""

        if isinstance(data, dict):
            session_id = data.get("session_id", "")
            received = data.get("received", True)
            error = data.get("error", "")

        if received:
            pdebug(f"TextInput bestätigt: session={session_id}")
        else:
            perror(f"TextInput fehlgeschlagen: {error}")

    async def _handle_server_shutdown(self, data: dict | object) -> None:
        """
        Behandelt Server-Shutdown-Benachrichtigung.

        Der Server fährt herunter, keine Wiederverbindung versuchen.
        """
        reason = getattr(data, "reason", "") or ""
        reconnect_after = getattr(data, "reconnect_after_seconds", 0) or 0

        if isinstance(data, dict):
            reason = data.get("reason", "")
            reconnect_after = data.get("reconnect_after_seconds", 0)

        pinfo(f"Server-Shutdown empfangen: {reason}")

        # Keine Wiederverbindung wenn reconnect_after_seconds = 0
        if reconnect_after == 0:
            # Anwendung beenden
            pinfo("Server fährt herunter, beende Client...")
            self.shutdown()
        else:
            # Wiederverbindung nach Wartezeit
            pinfo(f"Server-Neustart erwartet, Wiederverbindung in {reconnect_after}s...")
            await asyncio.sleep(reconnect_after)
            if self._running:
                self._schedule_reconnect()

    async def _handle_music_play_pause(self) -> None:
        """Toggle Musik Play/Pause."""
        if self._audio_manager:
            is_paused = self._audio_manager.toggle_music_pause()
            status = "pausiert" if is_paused else "fortgesetzt"
            pinfo(f"Musik {status}")
            await self.events.emit("music_paused" if is_paused else "music_resumed", {})

    async def _handle_music_next(self, data: dict | object) -> None:
        """Nächstes Lied - Signal an Musik-Plugin weitergeben."""
        use_crossfade = getattr(data, "use_crossfade", True)
        if isinstance(data, dict):
            use_crossfade = data.get("use_crossfade", True)

        pdebug(f"MusicNext empfangen, crossfade={use_crossfade}")
        await self.events.emit("music_next", {
            "use_crossfade": use_crossfade,
        })

    async def _handle_music_stop(self) -> None:
        """Musik stoppen."""
        if self._audio_manager:
            self._audio_manager.stop_music()
            pinfo("Musik gestoppt")
            await self.events.emit("music_stopped", {})

    async def _handle_media_stop_all(self) -> None:
        """Alle Medien stoppen (Musik + TTS)."""
        if self._audio_manager:
            self._audio_manager.stop_all()
            pinfo("Alle Wiedergabe gestoppt")
            await self.events.emit("media_stopped", {})

    async def _handle_satellite_record_start(self, data: dict | object) -> None:
        """
        Behandelt SatelliteRecordStart vom Server.

        Aktiviert Audio-Streaming zum Server — entweder über den
        WakewordService oder als Fallback direkt über das Mikrofon.
        """
        recording_id = getattr(data, "recording_id", "") or ""
        timeout_seconds = getattr(data, "timeout_seconds", 300.0)
        if isinstance(data, dict):
            recording_id = data.get("recording_id", "")
            timeout_seconds = data.get("timeout_seconds", 300.0)

        pinfo(f"Server-Recording empfangen: {recording_id} (timeout={timeout_seconds}s)")

        if self._wakeword_service:
            # Mit WakewordService: Audio-Streaming im LISTENING-Zustand aktivieren
            self._wakeword_service.start_server_recording(recording_id)
        elif self._microphone and self._connection:
            # Fallback ohne WakewordService: Mikrofon direkt streamen
            connection = self._connection
            loop = asyncio.get_event_loop()

            self._microphone.set_callback(
                lambda audio: loop.call_soon_threadsafe(
                    asyncio.ensure_future, connection.send_audio(audio)
                )
            )
            self._microphone.start()
            self._direct_streaming_active = True
            pinfo("Direktes Mikrofon-Streaming gestartet (ohne WakewordService)")

    async def _handle_satellite_record_stop(self, data: dict | object) -> None:
        """
        Behandelt SatelliteRecordStop vom Server.

        Beendet das Audio-Streaming und sendet Bestätigung.
        """
        recording_id = getattr(data, "recording_id", "") or ""
        reason = getattr(data, "reason", "manual") or "manual"
        if isinstance(data, dict):
            recording_id = data.get("recording_id", "")
            reason = data.get("reason", "manual")

        pinfo(f"Server-Recording-Stop empfangen: {recording_id} (reason={reason})")

        duration = 0.0

        if self._wakeword_service:
            # WakewordService: Recording stoppen
            duration = self._wakeword_service.stop_server_recording()
        elif self._direct_streaming_active:
            # Fallback: Direktes Streaming stoppen
            if self._microphone:
                self._microphone.stop()
            self._direct_streaming_active = False
            pinfo("Direktes Mikrofon-Streaming gestoppt")

        # Bestätigung an Server senden
        if self._connection:
            from trixy_core.network.cmd import SatelliteRecordStopped
            cmd = SatelliteRecordStopped(
                recording_id=recording_id,
                reason=reason,
                duration_seconds=duration,
            )
            await self._connection.send_command(cmd)

    async def _on_audio_out(self, audio_data: bytes) -> None:
        """
        Callback für Audio-Ausgabe (TTS) vom Server.

        Die Chunks werden direkt in den Streaming-Player eingespeist
        und sofort abgespielt.
        """
        # DEBUG: Chunks sammeln
        self._debug_tts_chunks.append(audio_data)

        # In Audio-Manager einspeisen
        if self._audio_manager:
            # Debug: Ersten Chunk und Playback-Start loggen
            if not hasattr(self, "_tts_playing"):
                self._tts_playing = False

            was_playing = self._tts_playing
            success = self._audio_manager.play_tts_chunk(audio_data)

            if not success:
                pwarn(f"[CLIENT] TTS-Chunk abgelehnt! ({len(audio_data)} bytes)")

            if success and not was_playing:
                pdebug(f"[CLIENT] TTS-Wiedergabe gestartet ({len(audio_data)} bytes erster Chunk)")
                self._tts_playing = True

            # Event für zusätzliche Verarbeitung auslösen (fire-and-forget, blockiert nicht)
            tts_format = self._audio_manager.tts_format
            asyncio.create_task(self.events.emit("audio_output_received", {
                "audio_data": audio_data,
                "sample_rate": tts_format["sample_rate"],
                "channels": tts_format["channels"],
            }))

    async def _on_music(self, audio_data: bytes) -> None:
        """
        Callback für Musik-Stream vom Server.

        Die Chunks werden direkt in den Streaming-Player eingespeist
        und sofort abgespielt.
        """
        # DEBUG: Chunks sammeln
        self._debug_music_chunks.append(audio_data)

        # In Audio-Manager einspeisen
        if self._audio_manager:
            # Debug: Ersten Chunk und Playback-Start loggen
            if not hasattr(self, "_music_playing"):
                self._music_playing = False

            was_playing = self._music_playing
            success = self._audio_manager.play_music_chunk(audio_data)

            if success and not was_playing:
                pdebug(f"[CLIENT] Music-Wiedergabe gestartet ({len(audio_data)} bytes erster Chunk)")
                self._music_playing = True

            # Event für zusätzliche Verarbeitung auslösen (fire-and-forget, blockiert nicht)
            music_format = self._audio_manager.music_format
            asyncio.create_task(self.events.emit("music_output_received", {
                "audio_data": audio_data,
                "sample_rate": music_format["sample_rate"],
                "channels": music_format["channels"],
            }))

    def _debug_save_wav(self, chunks: list[bytes], stream_type: str) -> None:
        """DEBUG: Speichert gesammelte Audio-Chunks als WAV-Datei."""
        if not chunks:
            pdebug(f"[DEBUG] Keine {stream_type}-Chunks zum Speichern")
            return

        if not self._audio_manager:
            return

        # Format ermitteln
        if stream_type == "tts":
            fmt = self._audio_manager.tts_format
        else:
            fmt = self._audio_manager.music_format

        sample_rate = fmt["sample_rate"]
        channels = fmt["channels"]
        sample_width = fmt.get("sample_width", 2)

        # Alle Chunks zusammenfügen
        all_data = b"".join(chunks)

        # Dateiname mit Timestamp
        timestamp = int(time.time())
        filename = f"/tmp/debug_{stream_type}_{timestamp}.wav"

        try:
            with wave.open(filename, "wb") as wf:
                wf.setnchannels(channels)
                wf.setsampwidth(sample_width)
                wf.setframerate(sample_rate)
                wf.writeframes(all_data)

            pinfo(f"[DEBUG] WAV gespeichert: {filename} ({len(all_data)} bytes, {sample_rate}Hz, {channels}ch)")
        except Exception as e:
            perror(f"[DEBUG] WAV speichern fehlgeschlagen: {e}")

    async def _handle_config_read_proxy(self, data: object) -> None:
        """Beantwortet eine Config-Lese-Anfrage vom Server (Config-Tool Proxy)."""
        import dataclasses
        from trixy_core.network.cmd.config_cmd import ConfigReadResponse

        message_id = getattr(data, "message_id", "") or ""
        config_name = getattr(data, "config_name", "") or ""
        section = getattr(data, "section", "") or ""

        response = ConfigReadResponse(
            message_id=message_id,
            config_name=config_name,
            section=section,
        )

        try:
            config_obj = self._client_config
            if config_obj is None:
                response.data = {"error": "Keine Client-Konfiguration verfuegbar"}
            else:
                config_dict = dataclasses.asdict(config_obj)
                if section and section in config_dict:
                    response.data = {section: config_dict[section]}
                else:
                    response.data = config_dict
        except Exception as e:
            response.data = {"error": str(e)}

        if self._connection:
            await self._connection.send_command(response)

    async def _handle_config_write_proxy(self, data: object) -> None:
        """Beantwortet eine Config-Schreib-Anfrage vom Server (Config-Tool Proxy)."""
        from trixy_core.network.cmd.config_cmd import ConfigWriteResponse

        message_id = getattr(data, "message_id", "") or ""
        key_path = getattr(data, "key_path", "") or ""
        value = getattr(data, "value", None)

        response = ConfigWriteResponse(message_id=message_id)

        try:
            config_obj = self._client_config
            if config_obj is None:
                response.error = "Keine Client-Konfiguration verfuegbar"
            else:
                # Punkt-Pfad navigieren und Wert setzen
                parts = key_path.split(".")
                target = config_obj
                for part in parts[:-1]:
                    if hasattr(target, part):
                        target = getattr(target, part)
                    else:
                        response.error = f"Pfad '{key_path}' nicht gefunden"
                        break

                if not response.error:
                    field_name = parts[-1]
                    if not hasattr(target, field_name):
                        response.error = f"Feld '{field_name}' nicht gefunden"
                    else:
                        # Typ-Konvertierung
                        current_value = getattr(target, field_name)
                        if isinstance(current_value, bool):
                            if isinstance(value, str):
                                value = value.lower() in ("true", "1", "ja", "yes")
                            else:
                                value = bool(value)
                        elif isinstance(current_value, int):
                            value = int(value)
                        elif isinstance(current_value, float):
                            value = float(value)
                        elif isinstance(current_value, list):
                            if isinstance(value, str):
                                value = [v.strip() for v in value.split(",") if v.strip()]
                        elif isinstance(current_value, str):
                            value = str(value)

                        setattr(target, field_name, value)

                        # Konfiguration speichern
                        self.config_manager.save("client")

                        response.success = True
                        if any(p in key_path for p in ("port", "bind_address", "host")):
                            response.requires_restart = True

        except Exception as e:
            response.error = str(e)

        if self._connection:
            await self._connection.send_command(response)

    async def _disconnect(self) -> None:
        """Trennt die Verbindung zum Server."""
        if self._connection:
            await self._connection.disconnect("Client shutdown")
            self._connection = None
        self._connected = False
        pdebug("Verbindung getrennt")

    def _schedule_reconnect(self) -> None:
        """Plant einen Wiederverbindungsversuch."""
        if self._reconnect_task and not self._reconnect_task.done():
            return

        async def reconnect() -> None:
            interval = self._client_config.server.reconnect_interval_seconds if self._client_config else 5.0
            while self._running and not self._connected:
                pwarn(f"Wiederverbindung in {interval} Sekunden...")
                await asyncio.sleep(interval)
                if self._running:
                    await self._connect()

        self._reconnect_task = asyncio.create_task(reconnect())