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