# -*- coding: utf-8 -*- """ Server-Anwendung für Trixy. Der Server ist die zentrale Komponente, die: - Satellites verwaltet - Plugins lädt und ausführt - ML-Inferenz durchführt - Events koordiniert """ import asyncio from pathlib import Path from trixy_core.application import IApplication from trixy_core.utils.version import VERSION_STRING from trixy_core.config.datasets.server import ServerConfig from trixy_core.satellite.satellite_manager import SatelliteManager from trixy_core.satellite.registration_manager import RegistrationManager from trixy_core.satellite.connection_notification import ( ConnectionNotificationHandler, setup_connection_notification, ) from trixy_core.plugins.plugin_manager import PluginManager from trixy_core.plugins.extensions import get_global_registry from trixy_core.plugins.extensions.points import ( setup_music_extension_points, setup_network_extension_points, setup_conversation_extension_points, ) from trixy_core.events.event_data.basic import SystemShutdown from trixy_core.network.service import NetworkService from trixy_core.network.message_handler import setup_default_event_bridge from trixy_core.audio.config import init_audio_settings_from_config from trixy_core.input.text_input_handler import setup_text_input_handler, TextInputHandler 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.arbitration.arbitrator import WakewordArbitrator, ArbitrationConfig from trixy_core.audio.accumulator import AudioAccumulatorService from trixy_core.network.config_listener import ConfigListener from trixy_core.email.service import EmailService from trixy_core.scheduler.service import SchedulerService from trixy_core.trainer.service import TrainerService from trixy_core.activity.tracker import ActivityTracker from trixy_core.utils.debug import pinfo, pdebug, perror class ServerApplication(IApplication): """ Server-Anwendung für Trixy. Verwaltet alle zentralen Funktionen und koordiniert die Kommunikation mit verbundenen Satellites. """ def __init__( self, config_path: str | Path = "config/server_config.json", debug: bool = False, auto_regist: bool = False ) -> None: """ Initialisiert die Server-Anwendung. Args: config_path: Pfad zur Konfigurationsdatei debug: Debug-Modus aktivieren auto_regist: Dauerhafter Auto-Registrierungsmodus """ super().__init__(debug) self._config_path = Path(config_path) self._auto_regist = auto_regist self._server_config: ServerConfig | None = None # Server-spezifische Manager self._satellite_manager: SatelliteManager | None = None self._registration_manager: RegistrationManager | None = None self._plugin_manager: PluginManager | None = None self._network_service: NetworkService | None = None self._connection_notification: ConnectionNotificationHandler | None = None self._text_input_handler: TextInputHandler | None = None self._wakeword_arbitrator: WakewordArbitrator | None = None self._audio_accumulator: AudioAccumulatorService | None = None self._config_listener: ConfigListener | None = None @property def server_config(self) -> ServerConfig: """Server-Konfiguration.""" if self._server_config is None: raise RuntimeError("Server nicht initialisiert") return self._server_config @property def satellites(self) -> SatelliteManager: """Satellite-Manager.""" if self._satellite_manager is None: raise RuntimeError("Server nicht initialisiert") return self._satellite_manager @property def registration(self) -> RegistrationManager: """Registrierungs-Manager.""" if self._registration_manager is None: raise RuntimeError("Server nicht initialisiert") return self._registration_manager @property def plugins(self) -> PluginManager: """Plugin-Manager.""" if self._plugin_manager is None: raise RuntimeError("Server nicht initialisiert") return self._plugin_manager @property def network(self) -> NetworkService: """Network-Service.""" if self._network_service is None: raise RuntimeError("Server nicht initialisiert") return self._network_service @property def extensions(self): """Extension-Registry.""" return get_global_registry() def _setup_extension_points(self) -> None: """ Richtet alle Extension Points ein. Wird vor dem Plugin-Laden aufgerufen, damit Plugins ihre Extensions registrieren können. """ pdebug("Initialisiere Extension Points...") # Music Extension Points setup_music_extension_points() # Network Extension Points setup_network_extension_points() # Conversation Extension Points setup_conversation_extension_points() registry = get_global_registry() pdebug(f"Extension Points initialisiert: {len(registry)} Points") async def initialize(self) -> None: """Initialisiert den Server.""" pinfo("Initialisiere Server...") # Konfiguration laden self._server_config = self.config_manager.load( self._config_path, ServerConfig, name="server", auto_reload=True ) # File-Logging initialisieren self._init_file_logging(self._server_config.logging) # Audio-Einstellungen initialisieren (für Plugin-Zugriff) audio_settings = init_audio_settings_from_config(self._server_config) pdebug(f"Audio-Einstellungen: TTS={audio_settings.tts_sample_rate}Hz, Musik={audio_settings.music_sample_rate}Hz") # Manager initialisieren self._satellite_manager = SatelliteManager(self) self._registration_manager = RegistrationManager( self, directory=self._server_config.satellites_directory ) self._plugin_manager = PluginManager( self, directory=self._server_config.plugins.plugin_directory, timeout_seconds=self._server_config.logging.plugin_timeout_seconds, max_failures=self._server_config.logging.plugin_max_failures, ) # Extension Points initialisieren (vor Plugin-Laden) self._setup_extension_points() # Registrierte Satellites laden registered = self._registration_manager.load_all() for satellite in registered: self._satellite_manager.add(satellite) # Auto-Registrierungsmodus aktivieren falls gewünscht if self._auto_regist: self._registration_manager.enable_permanent_registration_mode() else: # File-basierte Registrierung aktivieren self._registration_manager.enable_file_trigger() # NetworkService erstellen und registrieren self._network_service = NetworkService(self) self.services.register_instance(self._network_service) # Event-Bridge für Netzwerk-Nachrichten einrichten event_bridge = setup_default_event_bridge(self.events) self._network_service.register_message_handler(event_bridge.on_message) # ConfigListener erstellen und registrieren (Remote-Config-Tool Zugriff) self._config_listener = ConfigListener( self, port=self._server_config.network.config_port, bind_address=self._server_config.network.bind_address, encryption_key_path=self._server_config.security.encryption_key_path, instance_type="server", ) self.services.register_instance(self._config_listener) # Wakeword-Arbitrator erstellen (koordiniert Multi-Satellite-Erkennung) ww_config = self._server_config.wakeword self._wakeword_arbitrator = WakewordArbitrator( config=ArbitrationConfig( window_duration_ms=int(ww_config.arbitration_timeout_seconds * 1000), min_audio_level=ww_config.min_audio_level, min_confidence=ww_config.min_confidence, ), application=self, ) # Audio-Akkumulator erstellen (sammelt gestreamte Frames pro Conversation) ww_trim = self._server_config.wakeword.wakeword_trim_seconds self._audio_accumulator = AudioAccumulatorService(self, wakeword_trim_seconds=ww_trim) # SyncStore und SyncHandler erstellen from trixy_core.sync.store import SyncStore from trixy_core.sync.handler import SyncHandler self._sync_store = SyncStore(base_directory="./sync_data") self._sync_handler = SyncHandler(self) self._network_service.register_message_handler(self._sync_handler.on_message) pdebug("SyncStore und SyncHandler initialisiert") # SchedulerService erstellen und registrieren (Job-Scheduling) self._scheduler_service = SchedulerService(self) self.services.register_instance(self._scheduler_service) # TrainerService erstellen und registrieren (ML-Trainer-Framework) self._trainer_service = TrainerService(self) self.services.register_instance(self._trainer_service) # ActivityTracker erstellen und registrieren (zentrales Activity-Logging) self._activity_tracker = ActivityTracker(self) self.services.register_instance(self._activity_tracker) # EmailService erstellen und registrieren (SMTP-Versand fuer Plugins) from dataclasses import asdict email_cfg = asdict(self._server_config.email) self._email_service = EmailService(self, email_cfg) self.services.register_instance(self._email_service) # MusicPlayerService erstellen und registrieren from trixy_core.music import MusicPlayerService self._music_player_service = MusicPlayerService(self) self.services.register_instance(self._music_player_service) # Connection Notification Handler einrichten (für TTS-Notifications) # Musik-Wiedergabe läuft über MusicPlayerService self._connection_notification = setup_connection_notification(self) # 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) # Text-Input-Handler für Keyboard-Eingabe einrichten self._text_input_handler = setup_text_input_handler(self) pdebug(f"Server initialisiert mit {len(registered)} registrierten Satellites") async def start(self) -> None: """Startet den Server.""" pinfo("Starte Server...") await self.events.emit("application_starting", { "mode": "server", "version": VERSION_STRING, }) # Services starten await self.services.start_all() # Wakeword-Arbitrator starten (Event-Handler registrieren) if self._wakeword_arbitrator: await self._wakeword_arbitrator.start() # Audio-Akkumulator starten if self._audio_accumulator: await self._audio_accumulator.start() # Plugins laden if self._server_config and self._server_config.plugins.enabled: # Debug: Extension Points Status registry = get_global_registry() pdebug(f"Vor Plugin-Laden: {len(registry)} Extension Points, IDs: {[p.id for p in registry.points]}") await self._plugin_manager.load_all() pinfo(f"Server läuft auf Ports {self._server_config.network.command_port}-{self._server_config.network.music_port}") # Server ist jetzt vollstaendig bereit — Satellites duerfen sich verbinden self.services.set_ready(True) await self.events.emit("application_ready", { "mode": "server", "version": VERSION_STRING, "services_count": len(self.services.services), }) async def stop(self) -> None: """Stoppt den Server.""" pinfo("Stoppe Server...") # Markiere Server als nicht mehr bereit — neue HELO werden abgelehnt self.services.set_ready(False) await self.events.emit("application_stopping", {"reason": "shutdown"}) # Shutdown-Event auslösen await self.events.trigger( "system_shutdown", SystemShutdown(reason="Server shutdown", source="server") ) # Audio-Akkumulator stoppen if self._audio_accumulator: await self._audio_accumulator.stop() # Wakeword-Arbitrator stoppen if self._wakeword_arbitrator: await self._wakeword_arbitrator.stop() # Satellites trennen if self._satellite_manager: await self._satellite_manager.disconnect_all("Server shutdown") # 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("Server gestoppt")