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, parse_host_list
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 (erster Eintrag der Host-Liste fuer Backward-Compat)."""
        hosts = self.server_hosts
        return hosts[0] if hosts else "localhost"

    @property
    def server_hosts(self) -> list[str]:
        """Liste aller Server-Hosts die nacheinander probiert werden.

        Quellen (in Reihenfolge):
            1. CLI-Argument (--host kommagetrennt oder mehrfach)
            2. Client-Config (server.host als String oder Liste)
            3. Default ["localhost"]
        """
        # CLI-Wert hat Vorrang
        if self._cli_host is not None:
            return parse_host_list(self._cli_host)

        # Aus Client-Config
        if self._client_config:
            return parse_host_list(self._client_config.server.host) or ["localhost"]

        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,
                output_control=self._client_config.audio.output_mixer_control,
                input_control=self._client_config.audio.input_mixer_control,
                config_name="client",
            )
            pdebug("Volume Handler erstellt")
        except Exception as e:
            pdebug(f"Volume Handler nicht verfuegbar: {e}")

        # PulseAudio Combined-Sink (Multi-Speaker, Phase 2)
        try:
            from trixy_core.audio.pulse_combined import PulseCombinedSink
            self._combined_sink = PulseCombinedSink(
                sink_name=self._client_config.audio.output_combine_sink_name,
            )
            pdebug("Combined-Sink Wrapper erstellt")
        except Exception as e:
            pdebug(f"Combined-Sink nicht verfuegbar: {e}")

        # PulseAudio Mic-Mix (Multi-Mic, Phase 3)
        try:
            from trixy_core.audio.pulse_mic_mix import PulseMicrophoneMix
            self._mic_mix = PulseMicrophoneMix(
                sink_name=self._client_config.audio.input_mix_sink_name,
            )
            pdebug("Mic-Mix Wrapper erstellt")
        except Exception as e:
            pdebug(f"Mic-Mix nicht verfuegbar: {e}")

        # LED-Service (WS2812/NeoPixel fuer visuelle Status-Anzeige)
        if self._client_config.led.enabled:
            try:
                from trixy_core.led.service import LEDService
                led_config = {
                    "gpio_pin": self._client_config.led.gpio_pin,
                    "led_count": self._client_config.led.led_count,
                    "type": self._client_config.led.type,
                    "brightness": self._client_config.led.brightness,
                    "states": self._client_config.led.states,
                }
                self._led_service = LEDService(self, led_config)
                self.services.register_instance(self._led_service)
                pdebug("LED-Service erstellt")
            except Exception as e:
                pdebug(f"LED-Service 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()

        # Combined-Sink anwenden falls Multi-Speaker konfiguriert
        if hasattr(self, "_combined_sink"):
            await self._apply_combined_sink()

        # Mic-Mix anwenden falls Multi-Mic konfiguriert
        if hasattr(self, "_mic_mix"):
            await self._apply_mic_mix()

        hosts = self.server_hosts
        if len(hosts) > 1:
            pinfo(f"Client gestartet, verbinde zu einem von {len(hosts)} Hosts: {', '.join(hosts)} (Port {self.server_port})")
        else:
            pinfo(f"Client gestartet, verbinde zu {hosts[0]}:{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 _apply_combined_sink(self) -> None:
        """Erstellt/aktualisiert den Combined-Sink anhand der Config.

        Wird beim Start aufgerufen und kann zur Laufzeit (z.B. ueber
        F6 Audio View) erneut getriggert werden um eine neue Speaker-
        Auswahl scharfzuschalten.
        """
        if not hasattr(self, "_combined_sink"):
            return
        try:
            speakers = self._client_config.audio.output_speakers or []
            enabled = [
                sp for sp in speakers
                if getattr(sp, "enabled", False) and getattr(sp, "device", "")
            ]

            if len(enabled) < 2:
                # Keine bzw. nur ein Sprecher → Combined-Sink abbauen,
                # klassischer Single-Output-Pfad bleibt aktiv.
                await self._combined_sink.teardown()
                return

            if not await self._combined_sink.available():
                pwarn(
                    "Multi-Speaker konfiguriert aber PulseAudio/PipeWire "
                    "nicht verfuegbar — bleibe auf Single-Output."
                )
                return

            slaves = [sp.device for sp in enabled]
            ok = await self._combined_sink.setup(slaves)
            if not ok:
                return

            # Per-Slave-Lautstaerken anwenden
            for sp in enabled:
                await self._combined_sink.set_sink_volume(sp.device, int(sp.volume))

            # Combined-Sink zum System-Default machen damit der
            # sounddevice-Player (oeffnet ohne explizites device=)
            # automatisch im Verbund landet. Vorheriger Default wird
            # gemerkt und bei stop()/teardown() wiederhergestellt.
            await self._combined_sink.make_default()

            pinfo(
                f"Multi-Speaker aktiv: {len(slaves)} Lautsprecher → "
                f"Combined-Sink '{self._combined_sink.sink_name}'"
            )
        except Exception as e:
            perror(f"Combined-Sink konnte nicht angewendet werden: {e}")

    async def _apply_mic_mix(self) -> None:
        """Erstellt/aktualisiert den Multi-Mic-Mix anhand der Config.

        Bei ≥2 enabled Mikrofonen wird ein Pulse null-sink + loopback
        pro Mic erstellt und der System-Default-Source auf den Monitor
        des Null-Sinks gesetzt. Dadurch sieht der bestehende
        MicrophoneCapture-Code weiterhin nur EINE Quelle.
        """
        if not hasattr(self, "_mic_mix"):
            return
        try:
            mics = self._client_config.audio.input_microphones or []
            enabled = [
                m for m in mics
                if getattr(m, "enabled", False) and getattr(m, "device", "")
            ]

            if len(enabled) < 2:
                await self._mic_mix.teardown()
                return

            if not await self._mic_mix.available():
                pwarn(
                    "Multi-Mic konfiguriert aber PulseAudio/PipeWire "
                    "nicht verfuegbar — bleibe auf Single-Input."
                )
                return

            sources = [m.device for m in enabled]
            ok = await self._mic_mix.setup(sources)
            if not ok:
                return

            # Monitor-Source zum System-Default machen damit der
            # Wakeword-Service (oeffnet ohne explizites device=)
            # automatisch aus dem Mix liest. Vorheriger Default wird
            # gemerkt und bei stop()/teardown() wiederhergestellt.
            await self._mic_mix.make_default()

            pinfo(
                f"Multi-Mic aktiv: {len(sources)} Mikrofone → "
                f"'{self._mic_mix.monitor_source_name}'"
            )
        except Exception as e:
            perror(f"Mic-Mix konnte nicht angewendet werden: {e}")

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

        # Mic-Mix entladen (vor Combined-Sink, weil beide den Default
        # zuruecksetzen koennen — Reihenfolge macht's deterministischer)
        if hasattr(self, "_mic_mix"):
            try:
                await self._mic_mix.teardown()
            except Exception as e:
                pdebug(f"Mic-Mix teardown Fehler: {e}")

        # Combined-Sink Modul entladen falls vorhanden
        if hasattr(self, "_combined_sink"):
            try:
                await self._combined_sink.teardown()
            except Exception as e:
                pdebug(f"Combined-Sink teardown Fehler: {e}")

        # 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 mit allen konfigurierten Hosts
            # ClientConnection probiert die Hosts der Reihe nach durch
            self._connection = ClientConnection(
                host=self.server_hosts,
                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}")

        # Der Client bleibt laufend und versucht periodisch neu zu verbinden.
        # reconnect_after_seconds ist nur ein Hinweis wie lange gewartet werden soll,
        # bevor der erste Neuverbindungsversuch startet.
        if reconnect_after > 0:
            pinfo(f"Warte {reconnect_after}s bevor Wiederverbindung startet...")
            await asyncio.sleep(reconnect_after)

        if self._running:
            pinfo("Starte Wiederverbindungsversuche...")
            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.

        Laeuft in einer Endlosschleife solange der Client laeuft und nicht
        verbunden ist. Alle Fehler werden abgefangen, damit die Schleife
        nicht durch Exceptions abbricht.

        Wenn der Server via SatelliteDenied eine Wartezeit vorgibt
        (z.B. "Server wird gestartet"), wird diese respektiert — aber
        mindestens das konfigurierte reconnect_interval gewartet.
        """
        if self._reconnect_task and not self._reconnect_task.done():
            return

        async def reconnect() -> None:
            default_interval = (
                self._client_config.server.reconnect_interval_seconds
                if self._client_config else 10.0
            )
            attempt = 0
            while self._running and not self._connected:
                attempt += 1

                # Wartezeit bestimmen: wenn der Server eine Wartezeit vorgegeben
                # hat (via SatelliteDenied), diese nutzen — sonst Default
                wait = default_interval
                if self._connection and self._connection.last_denied_retry_after > 0:
                    wait = max(default_interval, self._connection.last_denied_retry_after)

                pwarn(f"Wiederverbindungsversuch #{attempt} in {wait:.0f} Sekunden...")
                try:
                    await asyncio.sleep(wait)
                except asyncio.CancelledError:
                    raise

                if not self._running:
                    break

                try:
                    await self._connect()
                except asyncio.CancelledError:
                    raise
                except Exception as e:
                    perror(f"Reconnect-Versuch #{attempt} fehlgeschlagen: {e}")
                    # Weitermachen — naechster Versuch in naechster Iteration

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