# -*- coding: utf-8 -*- """ Standalone-Anwendung für Trixy. Kombiniert Server- und Client-Funktionalität für den Betrieb ohne separate Server-Instanz. """ from __future__ import annotations from pathlib import Path from typing import TYPE_CHECKING if TYPE_CHECKING: from trixy_core.sync.service import SyncService from trixy_core.application import IApplication from trixy_core.config.datasets.standalone import StandaloneConfig from trixy_core.plugins.plugin_manager import PluginManager from trixy_core.events.event_data.basic import SystemShutdown from trixy_core.audio.config import init_audio_settings_from_config from trixy_core.audio.player import AudioOutputManager from trixy_core.audio.microphone import MicrophoneCapture from trixy_core.input.keyboard_input import create_keyboard_input, KeyboardInputService, SimpleKeyboardInput from trixy_core.nlp.intent_dispatcher import IntentDispatcherService from trixy_core.stt.corrector import STTCorrectorService from trixy_core.conversation.profiler import ConversationProfiler 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.network.config_listener import ConfigListener from trixy_core.email.service import EmailService from trixy_core.activity.tracker import ActivityTracker from trixy_core.utils.debug import pinfo, pdebug, perror, pwarn class StandaloneApplication(IApplication): """ Standalone-Anwendung für Trixy. Bietet alle Funktionen von Server und Client in einer einzelnen Anwendung. """ def __init__( self, config_path: str | Path = "config/standalone_config.json", debug: bool = False, keyboard_input: bool = False, ) -> None: """ Initialisiert die Standalone-Anwendung. Args: config_path: Pfad zur Konfigurationsdatei debug: Debug-Modus aktivieren keyboard_input: Keyboard-Eingabe aktivieren """ super().__init__(debug) self._config_path = Path(config_path) self._standalone_config: StandaloneConfig | None = None self._plugin_manager: PluginManager | None = None # Im Debug-Modus immer Keyboard-Shortcuts aktivieren (Ctrl+P/N/S) self._keyboard_input_enabled = keyboard_input or debug self._keyboard_service: KeyboardInputService | SimpleKeyboardInput | None = None # ConfigListener self._config_listener: ConfigListener | None = None # Audio- und Wakeword-Komponenten self._wakeword_service: WakewordService | None = None self._microphone: MicrophoneCapture | None = None self._audio_manager: AudioOutputManager | None = None # Sync self._sync_service: "SyncService | None" = None @property def standalone_config(self) -> StandaloneConfig: """Standalone-Konfiguration.""" if self._standalone_config is None: raise RuntimeError("Standalone nicht initialisiert") return self._standalone_config @property def plugins(self) -> PluginManager: """Plugin-Manager.""" if self._plugin_manager is None: raise RuntimeError("Standalone nicht initialisiert") return self._plugin_manager @property def room(self) -> str: """Raum-Kennung.""" if self._standalone_config: return self._standalone_config.room return "standalone" @property def alias(self) -> str: """Anzeigename.""" if self._standalone_config: return self._standalone_config.alias return "Trixy" async def initialize(self) -> None: """Initialisiert die Standalone-Anwendung.""" pinfo("Initialisiere Standalone...") # Konfiguration laden self._standalone_config = self.config_manager.load( self._config_path, StandaloneConfig, name="standalone", auto_reload=True ) # File-Logging initialisieren self._init_file_logging(self._standalone_config.logging) # Audio-Einstellungen initialisieren (für Plugin-Zugriff) audio_settings = init_audio_settings_from_config(self._standalone_config) pdebug(f"Audio-Einstellungen: TTS={audio_settings.tts_sample_rate}Hz, Musik={audio_settings.music_sample_rate}Hz") # Plugin-Manager initialisieren self._plugin_manager = PluginManager( self, directory=self._standalone_config.plugins.plugin_directory, timeout_seconds=self._standalone_config.logging.plugin_timeout_seconds, max_failures=self._standalone_config.logging.plugin_max_failures, ) # EmailService erstellen und registrieren (SMTP-Versand fuer Plugins) from dataclasses import asdict email_cfg = asdict(self._standalone_config.email) self._email_service = EmailService(self, email_cfg) self.services.register_instance(self._email_service) # 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 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) self._config_listener = ConfigListener( self, port=self._standalone_config.config_port, bind_address="0.0.0.0", encryption_key_path=self._standalone_config.security.encryption_key_path, instance_type="standalone", ) self.services.register_instance(self._config_listener) # ActivityTracker erstellen und registrieren (zentrales Activity-Logging) self._activity_tracker = ActivityTracker(self) self.services.register_instance(self._activity_tracker) # IntentDispatcher-Service registrieren (verarbeitet intent_received Events) self._intent_dispatcher = IntentDispatcherService(self) self.services.register_instance(self._intent_dispatcher) self.events.register_object(self._intent_dispatcher) # STT-Korrektur-Service registrieren (korrigiert Text vor NLP mit HIGH-Prioritaet) self._stt_corrector = STTCorrectorService(self) self.services.register_instance(self._stt_corrector) self.events.register_object(self._stt_corrector) # Conversation-Profiler registrieren (Debug-Modus: misst Phasen-Dauern) self._conversation_profiler = ConversationProfiler(self) self.services.register_instance(self._conversation_profiler) self.events.register_object(self._conversation_profiler) # SyncStore erstellen (immer, auch ohne Server-Verbindung fuer lokale Persistenz) from trixy_core.sync.store import SyncStore sync_dir = self._standalone_config.sync.sync_data_directory self._sync_store = SyncStore(base_directory=sync_dir) pdebug(f"SyncStore erstellt: {sync_dir}") # SyncService erstellen (falls konfiguriert) if self._standalone_config.sync.enabled and self._standalone_config.sync.server_host: from trixy_core.sync.service import SyncService as _SyncService self._sync_service = _SyncService(self, self._standalone_config.sync) pdebug("SyncService erstellt") # AudioOutputManager erstellen (für lokale TTS-Wiedergabe) self._audio_manager = AudioOutputManager( tts_sample_rate=22050, # Piper-Default, wird ggf. von TTS-Plugin aktualisiert music_sample_rate=self._standalone_config.audio.music_sample_rate, ) await self._audio_manager.initialize() # Wakeword-Service erstellen (falls aktiviert) ww_config = self._standalone_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=True, wakeword_cooldown_seconds=ww_config.wakeword_cooldown_seconds, ) self._wakeword_service = WakewordService(self, service_config) # Mikrofon erstellen self._microphone = MicrophoneCapture( device=self._standalone_config.audio.input_device, ) pinfo("Wakeword-Service erstellt") except Exception as e: pwarn(f"Wakeword-Service nicht verfügbar: {e}") pdebug("Standalone initialisiert") async def start(self) -> None: """Startet die Standalone-Anwendung.""" pinfo(f"Starte Standalone als '{self.alias}'...") from trixy_core.utils.version import VERSION_STRING await self.events.emit("application_starting", { "mode": "standalone", "version": VERSION_STRING, }) # Services starten await self.services.start_all() # Plugins laden if self._standalone_config and self._standalone_config.plugins.enabled: await self._plugin_manager.load_all() # Wakeword-Service starten if self._wakeword_service and self._microphone: try: await self._wakeword_service.on_start() # Mikrofon direkt mit WakewordService verbinden (ohne Netzwerk) self._wakeword_service._microphone = self._microphone await self._wakeword_service.start_listening() except Exception as e: pwarn(f"Wakeword-Service konnte nicht gestartet werden: {e}") # SyncService starten if self._sync_service: await self._sync_service.start() # TTS-Wiedergabe-Handler registrieren self._setup_standalone_tts_playback() # Follow-Up-Bridge: followup_expected → WakewordService FOLLOW_UP State self._setup_standalone_followup() # Keyboard-Input starten (falls aktiviert) # Im Standalone-Modus gibt es keine Server-Verbindung, # daher wird speech_recognized Event direkt emittiert if self._keyboard_input_enabled: self._keyboard_service = create_keyboard_input( self, connection=None, # Standalone hat keine Server-Verbindung prompt=f"{self.alias}> ", ) await self._keyboard_service.start() # Volume Handler starten if hasattr(self, "_volume_handler"): await self._volume_handler.start() pinfo("Standalone gestartet") from trixy_core.utils.version import VERSION_STRING await self.events.emit("application_ready", { "mode": "standalone", "version": VERSION_STRING, "services_count": len(self.services.services), }) def _setup_standalone_tts_playback(self) -> None: """Registriert Event-Handler für lokale TTS-Wiedergabe.""" if not self._audio_manager: return audio_manager = self._audio_manager async def on_tts_completed(event_name: str, data) -> None: """Spielt TTS-Audio lokal ab (data ist EventData, Felder in metadata).""" audio_data_hex = data.get("audio_data") if not audio_data_hex: return # audio_data ist hex-kodiert (str) — dekodieren zu bytes audio_bytes = bytes.fromhex(audio_data_hex) # TTS-Format aktualisieren falls mitgeliefert sample_rate = data.get("sample_rate") if sample_rate and sample_rate != audio_manager.tts_format.get("sample_rate"): channels = data.get("channels") or 1 sample_width = data.get("sample_width") or 2 await audio_manager.update_tts_format(sample_rate, channels, sample_width) audio_manager.play_tts_chunk(audio_bytes) audio_manager.finish_tts() self.events.register("tts_completed", on_tts_completed) def _setup_standalone_followup(self) -> None: """Bruecke: followup_expected → WakewordService FOLLOW_UP State. Im Client-Modus sendet der Server FollowUpRequest als Command. Im Standalone-Modus muss dieses Event direkt an den WakewordService weitergeleitet werden, damit dieser in den FOLLOW_UP-State wechselt und ohne Wakeword-Erkennung aufnimmt. """ if not self._wakeword_service: return wakeword_service = self._wakeword_service async def on_followup_expected(event_name: str, data) -> None: """Startet Aufnahme fuer Rueckfrage ohne Wakeword.""" session_id = getattr(data, "session_id", "") or "" timeout = getattr(data, "timeout_seconds", 30.0) pdebug(f"Standalone Follow-Up Bridge: session={session_id}, timeout={timeout}") await wakeword_service._handle_follow_up({ "session_id": session_id, "timeout_seconds": timeout, }) self.events.register("followup_expected", on_followup_expected) async def stop(self) -> None: """Stoppt die Standalone-Anwendung.""" pinfo("Stoppe Standalone...") await self.events.emit("application_stopping", {"reason": "shutdown"}) # SyncService stoppen if self._sync_service: await self._sync_service.stop() # Wakeword-Service stoppen if self._wakeword_service: try: await self._wakeword_service.on_stop() except Exception as e: perror(f"Fehler beim Stoppen des Wakeword-Service: {e}") # Mikrofon stoppen if self._microphone and self._microphone.is_running: self._microphone.stop() # AudioOutputManager herunterfahren if self._audio_manager: await self._audio_manager.shutdown() # Keyboard-Input stoppen if self._keyboard_service: await self._keyboard_service.stop() self._keyboard_service = None # Shutdown-Event auslösen await self.events.trigger( "system_shutdown", SystemShutdown(reason="Standalone shutdown", source="standalone") ) # Plugins entladen if self._plugin_manager: await self._plugin_manager.unload_all() # Services stoppen await self.services.stop_all() # ConfigManager aufräumen self.config_manager.stop_watching() # File-Logging herunterfahren self._shutdown_file_logging() pinfo("Standalone gestoppt")