# -*- coding: utf-8 -*- """ Trixy-TUI Hauptanwendung. Textual-basierte Benutzeroberflaeche fuer das Config-Tool. """ from __future__ import annotations from pathlib import Path from typing import TYPE_CHECKING from textual.app import App, ComposeResult from textual.binding import Binding from textual.containers import Container from trixy_core.tui.menu_context import MenuContext, MenuBinding from trixy_core.tui.widgets.header import TrixyHeader from trixy_core.tui.widgets.footer import TrixyFooter from trixy_core.tui.views.base import MainView if TYPE_CHECKING: from trixy_core.config_app import ConfigConnection # View-IDs der Satellite-SubViews SAT_SUBVIEW_IDS = ("sat_info", "sat_connection", "sat_conversation", "sat_config", "sat_updates") # View-IDs der Plugin-SubViews (Basis) PLUGIN_SUBVIEW_IDS = ("plugin_info", "plugin_config") # View-IDs der Schedule-SubViews SCHED_SUBVIEW_IDS = ("sched_info", "sched_trigger", "sched_conditions", "sched_actions") # View-IDs der Trainer-SubViews TRAINER_SUBVIEW_IDS = ( "trainer_info", "trainer_settings", "trainer_dataset", "trainer_optional", "trainer_validate", "trainer_training", ) class TrixyTUI(App): """ Textual-App fuer die Trixy Remote-Verwaltung. Verwaltet Views, F-Tasten-Bindungen und die Verbindung zur laufenden Trixy-Instanz. Unterstuetzt zwei Modi: - Haupt-Navigation (F1=Health, F2=Config, F3=Satellites, F9=Log) - Satellite-Kontext (F1=Info, F2=Connection, F3=Conversation, F4=Config, F5=Updates) """ CSS_PATH = [ Path(__file__).parent / "css" / "base.tcss", Path(__file__).parent / "css" / "health.tcss", Path(__file__).parent / "css" / "config.tcss", Path(__file__).parent / "css" / "satellites.tcss", Path(__file__).parent / "css" / "plugins.tcss", Path(__file__).parent / "css" / "popup.tcss", Path(__file__).parent / "css" / "schedule.tcss", Path(__file__).parent / "css" / "trainer.tcss", ] BINDINGS = [ Binding("f1", "fkey('f1')", "F1", show=False), Binding("f2", "fkey('f2')", "F2", show=False), Binding("f3", "fkey('f3')", "F3", show=False), Binding("f4", "fkey('f4')", "F4", show=False), Binding("f5", "fkey('f5')", "F5", show=False), Binding("f6", "fkey('f6')", "F6", show=False), Binding("f7", "fkey('f7')", "F7", show=False), Binding("f9", "fkey('f9')", "F9", show=False), Binding("f10", "back", "Zurueck", show=False), Binding("pageup", "scroll_up_page", "PageUp", show=False), Binding("pagedown", "scroll_down_page", "PageDown", show=False), Binding("q", "quit", "Beenden", show=False), ] # Default-Refresh-Intervall fuer den aktiven View (in Sekunden) # Kann via Config oder Laufzeit-Parameter ueberschrieben werden DEFAULT_REFRESH_INTERVAL = 3.0 def __init__( self, views: list[MainView] | None = None, title: str = "Trixy Config", connection_info: str = "", refresh_interval: float | None = None, **kwargs, ) -> None: super().__init__(**kwargs) self.scroll_sensitivity_y = 1.0 self._views = views or [] self._view_map: dict[str, MainView] = {} self._active_view: MainView | None = None self._connection: ConfigConnection | None = None self._connection_info = connection_info self._app_title = title self._refresh_interval = refresh_interval or self.DEFAULT_REFRESH_INTERVAL # Satellite-Kontext self._in_satellite_context: bool = False self._satellite_data: dict = {} self._parent_view_id: str = "" # Plugin-Kontext self._in_plugin_context: bool = False self._plugin_data: dict = {} self._plugin_custom_views: list[MainView] = [] # Schedule-Kontext self._in_schedule_context: bool = False self._schedule_data: dict = {} # Trainer-Kontext self._in_trainer_context: bool = False self._trainer_data: dict = {} # MenuContexte: Haupt, Satellite, Plugin, Schedule und Trainer self._main_menu = MenuContext() self._sat_menu = MenuContext() self._plugin_menu = MenuContext() self._sched_menu = MenuContext() self._trainer_menu = MenuContext() # Views registrieren for view in self._views: self._view_map[view.VIEW_ID] = view # Haupt-MenuContext aufbauen main_key_map = { "health": "f1", "config": "f2", "satellites": "f3", "plugins": "f4", "schedule": "f5", "trainer": "f7", "log": "f9", } for view in self._views: key = main_key_map.get(view.VIEW_ID, "") if key: self._main_menu.add(MenuBinding( key=key, label=view.TITLE, view_id=view.VIEW_ID, )) # Satellite-MenuContext aufbauen sat_key_map = { "sat_info": "f1", "sat_connection": "f2", "sat_conversation": "f3", "sat_config": "f4", "sat_updates": "f5", } for view in self._views: key = sat_key_map.get(view.VIEW_ID, "") if key: self._sat_menu.add(MenuBinding( key=key, label=view.TITLE, view_id=view.VIEW_ID, )) # F10 Zurueck im Satellite-Kontext self._sat_menu.add(MenuBinding(key="f10", label="Zurueck", view_id="")) # Plugin-MenuContext aufbauen plugin_key_map = { "plugin_info": "f1", "plugin_config": "f2", } for view in self._views: key = plugin_key_map.get(view.VIEW_ID, "") if key: self._plugin_menu.add(MenuBinding( key=key, label=view.TITLE, view_id=view.VIEW_ID, )) self._plugin_fkey_map: dict[str, str] = {} for vid, key in plugin_key_map.items(): if vid in self._view_map: self._plugin_fkey_map[key] = vid # F10 Zurueck im Plugin-Kontext self._plugin_menu.add(MenuBinding(key="f10", label="Zurueck", view_id="")) # Schedule-MenuContext aufbauen sched_key_map = { "sched_info": "f1", "sched_trigger": "f2", "sched_conditions": "f3", "sched_actions": "f4", } for view in self._views: key = sched_key_map.get(view.VIEW_ID, "") if key: self._sched_menu.add(MenuBinding( key=key, label=view.TITLE, view_id=view.VIEW_ID, )) self._sched_fkey_map: dict[str, str] = {} for vid, key in sched_key_map.items(): if vid in self._view_map: self._sched_fkey_map[key] = vid # F10 Zurueck im Schedule-Kontext self._sched_menu.add(MenuBinding(key="f10", label="Zurueck", view_id="")) # Trainer-MenuContext aufbauen trainer_key_map = { "trainer_info": "f1", "trainer_settings": "f2", "trainer_dataset": "f3", "trainer_optional": "f4", "trainer_validate": "f5", "trainer_training": "f6", } for view in self._views: key = trainer_key_map.get(view.VIEW_ID, "") if key: self._trainer_menu.add(MenuBinding( key=key, label=view.TITLE, view_id=view.VIEW_ID, )) self._trainer_fkey_map: dict[str, str] = {} for vid, key in trainer_key_map.items(): if vid in self._view_map: self._trainer_fkey_map[key] = vid # F10 Zurueck im Trainer-Kontext self._trainer_menu.add(MenuBinding(key="f10", label="Zurueck", view_id="")) # F10 Zurueck im Haupt-Menue (nur anzeigen, Aktion in action_back) self._main_menu.add(MenuBinding(key="f10", label="Zurueck", view_id="")) # F-Tasten-Zuordnungen self._main_fkey_map: dict[str, str] = {} for vid, key in main_key_map.items(): if vid in self._view_map: self._main_fkey_map[key] = vid self._sat_fkey_map: dict[str, str] = {} for vid, key in sat_key_map.items(): if vid in self._view_map: self._sat_fkey_map[key] = vid # Client-Modus: Satellite SubViews direkt als Haupt-Navigation self._client_satellite_mode = self._detect_client_satellite_mode() if self._client_satellite_mode: # Im Client-Modus sind die Sat-SubViews die Haupt-Navigation self._main_fkey_map = dict(self._sat_fkey_map) self._main_menu = MenuContext() for view in self._views: key = sat_key_map.get(view.VIEW_ID, "") if key: self._main_menu.add(MenuBinding( key=key, label=view.TITLE, view_id=view.VIEW_ID, )) # Auch Health/Config/Log behalten wenn vorhanden if "health" in self._view_map: self._main_fkey_map["f9"] = "health" self._main_menu.add(MenuBinding(key="f9", label="Health", view_id="health")) def _detect_client_satellite_mode(self) -> bool: """Erkennt ob wir im Client-Modus sind (SubViews direkt anzeigen).""" has_sat_subviews = any(vid in self._view_map for vid in SAT_SUBVIEW_IDS) has_satellites_list = "satellites" in self._view_map return has_sat_subviews and not has_satellites_list @property def connection(self) -> "ConfigConnection | None": """Aktive Verbindung.""" return self._connection def set_connection(self, connection: "ConfigConnection") -> None: """Setzt die Netzwerkverbindung und propagiert an alle Views.""" self._connection = connection for view in self._views: view.connection = connection def compose(self) -> ComposeResult: """Baut das UI-Layout auf.""" yield TrixyHeader(id="header") with Container(id="content-area"): for view in self._views: view.display = False yield view yield TrixyFooter(menu_context=self._main_menu, id="footer") async def on_mount(self) -> None: """Initialisiert die TUI nach dem Mounten.""" header = self.query_one("#header", TrixyHeader) header.connection_info = self._connection_info # Ersten View aktivieren if self._client_satellite_mode: # Client-Modus: Ersten Sat-SubView aktivieren for vid in SAT_SUBVIEW_IDS: if vid in self._view_map: await self._activate_view(vid) break elif self._views: await self._activate_view(self._views[0].VIEW_ID) # Periodische Aktualisierung starten self.set_interval(self._refresh_interval, self._periodic_refresh) async def action_fkey(self, key: str) -> None: """Verarbeitet F-Tasten-Druecke kontextabhaengig.""" if self._in_satellite_context: # Satellite-Kontext: Sat-SubViews wechseln view_id = self._sat_fkey_map.get(key) if view_id and view_id in self._view_map: await self._activate_view(view_id) self._update_header_satellite() elif self._in_plugin_context: # Plugin-Kontext: Plugin-SubViews wechseln view_id = self._plugin_fkey_map.get(key) if view_id and view_id in self._view_map: await self._activate_view(view_id) self._update_header_plugin() elif self._in_schedule_context: # Schedule-Kontext: Schedule-SubViews wechseln view_id = self._sched_fkey_map.get(key) if view_id and view_id in self._view_map: await self._activate_view(view_id) self._update_header_schedule() elif self._in_trainer_context: # Trainer-Kontext: Trainer-SubViews wechseln view_id = self._trainer_fkey_map.get(key) if view_id and view_id in self._view_map: await self._activate_view(view_id) self._update_header_trainer() else: # Haupt-Navigation view_id = self._main_fkey_map.get(key) if view_id and view_id in self._view_map: self._parent_view_id = "" await self._activate_view(view_id) async def action_back(self) -> None: """F10: Zurueck — verlaesst Satellite/Plugin-Kontext oder navigiert zurueck.""" if self._in_satellite_context: await self._leave_satellite_context() elif self._in_plugin_context: await self._leave_plugin_context() elif self._in_schedule_context: await self._leave_schedule_context() elif self._in_trainer_context: await self._leave_trainer_context() elif self._parent_view_id and self._parent_view_id in self._view_map: parent_id = self._parent_view_id self._parent_view_id = "" await self._activate_view(parent_id) async def open_satellite_detail(self, sat_data: dict) -> None: """Oeffnet die Satellite-SubViews fuer einen ausgewaehlten Satellite.""" await self._enter_satellite_context(sat_data) async def _enter_satellite_context(self, sat_data: dict) -> None: """Wechselt in den Satellite-Kontext mit Sub-Navigation.""" self._in_satellite_context = True self._satellite_data = sat_data self._parent_view_id = "satellites" # Satellite-Daten an alle SubViews propagieren for vid in SAT_SUBVIEW_IDS: view = self._view_map.get(vid) if view and hasattr(view, "set_satellite"): view.set_satellite(sat_data) # Footer auf Satellite-Menue umschalten footer = self.query_one("#footer", TrixyFooter) footer.update_menu(self._sat_menu) # Ersten Satellite-SubView aktivieren for vid in SAT_SUBVIEW_IDS: if vid in self._view_map: await self._activate_view(vid) break self._update_header_satellite() async def _leave_satellite_context(self) -> None: """Verlaesst den Satellite-Kontext und kehrt zur Satellite-Liste zurueck.""" self._in_satellite_context = False self._satellite_data = {} # Footer auf Haupt-Menue zurueckschalten footer = self.query_one("#footer", TrixyFooter) footer.update_menu(self._main_menu) # Zurueck zur Satellite-Liste if "satellites" in self._view_map: await self._activate_view("satellites") elif self._parent_view_id and self._parent_view_id in self._view_map: await self._activate_view(self._parent_view_id) self._parent_view_id = "" # ── Plugin-Kontext ── async def open_plugin_detail(self, plugin_data: dict) -> None: """Oeffnet die Plugin-SubViews fuer ein ausgewaehltes Plugin.""" await self._enter_plugin_context(plugin_data) async def _enter_plugin_context(self, plugin_data: dict) -> None: """Wechselt in den Plugin-Kontext mit Sub-Navigation.""" self._in_plugin_context = True self._plugin_data = plugin_data self._parent_view_id = "plugins" # Plugin-Daten an alle SubViews propagieren for vid in PLUGIN_SUBVIEW_IDS: view = self._view_map.get(vid) if view and hasattr(view, "set_plugin"): view.set_plugin(plugin_data) # Benutzerdefinierte Tabs hinzufuegen (aus custom_tabs der Detail-Daten) await self._setup_plugin_custom_tabs(plugin_data) # Footer auf Plugin-Menue umschalten footer = self.query_one("#footer", TrixyFooter) footer.update_menu(self._plugin_menu) # Ersten Plugin-SubView aktivieren for vid in PLUGIN_SUBVIEW_IDS: if vid in self._view_map: await self._activate_view(vid) break self._update_header_plugin() async def _setup_plugin_custom_tabs(self, plugin_data: dict) -> None: """Richtet benutzerdefinierte Plugin-Tabs ein.""" # Alte Custom-Views entfernen for view in self._plugin_custom_views: vid = view.VIEW_ID self._view_map.pop(vid, None) self._plugin_menu.remove(vid) self._plugin_fkey_map = { k: v for k, v in self._plugin_fkey_map.items() if v != vid } self._plugin_custom_views.clear() # Wenn custom_tabs vorhanden, Detail anfordern if not self.connection or not plugin_data.get("name"): return try: detail = await self.connection.request_plugin_detail(plugin_data["name"]) except Exception: return if not detail or not detail.custom_tabs: return # Detail-Daten (inkl. Config) in plugin_data einfuegen plugin_data["config"] = detail.config or {} # Custom-Tabs erstellen (F3, F4, F5, ...) from trixy_core.tui.views.plugin_custom_view import PluginCustomView fkey_start = 3 # F3 aufwaerts content_area = self.query_one("#content-area") for i, tab in enumerate(detail.custom_tabs): fkey_num = fkey_start + i if fkey_num > 8: # Max F8 (F9=Log, F10=Zurueck) break tab_title = tab.get("title", f"Tab {fkey_num}") tab_id = f"plugin_custom_{fkey_num}" fkey = f"f{fkey_num}" custom_view = PluginCustomView( tab_title=tab_title, tab_id=tab_id, tab_data=tab, connection=self._connection, ) custom_view.set_plugin(plugin_data) custom_view.display = False # Dynamisch in den Content-Bereich mounten await content_area.mount(custom_view) self._view_map[tab_id] = custom_view self._plugin_custom_views.append(custom_view) self._plugin_menu.add(MenuBinding( key=fkey, label=tab_title, view_id=tab_id, )) self._plugin_fkey_map[fkey] = tab_id async def _leave_plugin_context(self) -> None: """Verlaesst den Plugin-Kontext und kehrt zur Plugin-Liste zurueck.""" self._in_plugin_context = False self._plugin_data = {} # Custom-Views entfernen und unmounten for view in self._plugin_custom_views: vid = view.VIEW_ID self._view_map.pop(vid, None) self._plugin_menu.remove(vid) self._plugin_fkey_map = { k: v for k, v in self._plugin_fkey_map.items() if v != vid } try: await view.remove() except Exception: pass self._plugin_custom_views.clear() # Footer auf Haupt-Menue zurueckschalten footer = self.query_one("#footer", TrixyFooter) footer.update_menu(self._main_menu) # Zurueck zur Plugin-Liste if "plugins" in self._view_map: await self._activate_view("plugins") self._parent_view_id = "" def _update_header_plugin(self) -> None: """Aktualisiert den Header im Plugin-Kontext.""" header = self.query_one("#header", TrixyHeader) name = self._plugin_data.get("name", "Plugin") if self._active_view: header.subtitle = f"Plugins > {name} > {self._active_view.TITLE}" else: header.subtitle = f"Plugins > {name}" # ── Schedule-Kontext ── async def open_schedule_detail(self, job_data: dict) -> None: """Oeffnet die Schedule-SubViews fuer einen ausgewaehlten Job.""" await self._enter_schedule_context(job_data) async def _enter_schedule_context(self, job_data: dict) -> None: """Wechselt in den Schedule-Kontext mit Sub-Navigation.""" self._in_schedule_context = True self._schedule_data = job_data self._parent_view_id = "schedule" # FormMeta laden (Trigger/Condition/Action-Typen mit Formular-Feldern) form_meta = None if self._connection: try: form_meta = await self._connection.request_schedule_form_meta() except Exception: pass # Job-Daten und FormMeta an alle SubViews propagieren for vid in SCHED_SUBVIEW_IDS: view = self._view_map.get(vid) if view and hasattr(view, "set_job"): view.set_job(job_data) if form_meta and view and hasattr(view, "set_form_meta"): if vid == "sched_trigger": view.set_form_meta(form_meta.triggers if hasattr(form_meta, "triggers") else {}) elif vid == "sched_conditions": view.set_form_meta(form_meta.conditions if hasattr(form_meta, "conditions") else {}) elif vid == "sched_actions": view.set_form_meta(form_meta.actions if hasattr(form_meta, "actions") else {}) # Footer auf Schedule-Menue umschalten footer = self.query_one("#footer", TrixyFooter) footer.update_menu(self._sched_menu) # Ersten Schedule-SubView aktivieren for vid in SCHED_SUBVIEW_IDS: if vid in self._view_map: await self._activate_view(vid) break self._update_header_schedule() async def _leave_schedule_context(self) -> None: """Verlaesst den Schedule-Kontext und kehrt zur Schedule-Liste zurueck.""" self._in_schedule_context = False self._schedule_data = {} # Footer auf Haupt-Menue zurueckschalten footer = self.query_one("#footer", TrixyFooter) footer.update_menu(self._main_menu) # Zurueck zur Schedule-Liste if "schedule" in self._view_map: await self._activate_view("schedule") self._parent_view_id = "" async def save_schedule_job(self) -> tuple[bool, str]: """Sammelt Daten aus allen Schedule-SubViews und speichert den Job.""" if not self._connection: return False, "Keine Verbindung" # Daten aus allen Sub-Views sammeln info_view = self._view_map.get("sched_info") trigger_view = self._view_map.get("sched_trigger") cond_view = self._view_map.get("sched_conditions") action_view = self._view_map.get("sched_actions") info_data = info_view.collect_data() if info_view and hasattr(info_view, "collect_data") else {} trigger_data = trigger_view.collect_data() if trigger_view and hasattr(trigger_view, "collect_data") else {} conditions_data = cond_view.collect_data() if cond_view and hasattr(cond_view, "collect_data") else [] actions_data = action_view.collect_data() if action_view and hasattr(action_view, "collect_data") else [] job_id = self._schedule_data.get("job_id", "") name = info_data.get("name", "") description = info_data.get("description", "") config = info_data.get("config", {}) config["enabled"] = info_data.get("enabled", True) if not name: return False, "Name ist erforderlich" try: result = await self._connection.save_schedule_job( job_id=job_id, name=name, description=description, trigger=trigger_data, conditions=conditions_data, actions=actions_data, config=config, ) if result and result.success: # Job-ID aktualisieren (bei neuem Job) if result.job_id: self._schedule_data["job_id"] = result.job_id self._schedule_data["name"] = name return True, f"Job gespeichert (ID: {result.job_id})" error = result.error if result else "Unbekannter Fehler" return False, f"Fehler: {error}" except Exception as e: return False, f"Fehler: {e}" def _update_header_schedule(self) -> None: """Aktualisiert den Header im Schedule-Kontext.""" header = self.query_one("#header", TrixyHeader) name = self._schedule_data.get("name", "Job") if self._active_view: header.subtitle = f"Schedule > {name} > {self._active_view.TITLE}" else: header.subtitle = f"Schedule > {name}" # ── Trainer-Kontext ── async def open_trainer_detail(self, trainer_data: dict) -> None: """Oeffnet die Trainer-SubViews fuer einen ausgewaehlten Trainer.""" await self._enter_trainer_context(trainer_data) async def _enter_trainer_context(self, trainer_data: dict) -> None: """Wechselt in den Trainer-Kontext mit Sub-Navigation.""" self._in_trainer_context = True self._trainer_data = trainer_data self._parent_view_id = "trainer" # Trainer-Daten an alle SubViews propagieren for vid in TRAINER_SUBVIEW_IDS: view = self._view_map.get(vid) if view and hasattr(view, "set_trainer"): view.set_trainer(trainer_data) # Optional-Tab ausblenden falls kein Schema # (wird beim Laden in on_activate geprueft) # Footer auf Trainer-Menue umschalten footer = self.query_one("#footer", TrixyFooter) footer.update_menu(self._trainer_menu) # Ersten Trainer-SubView aktivieren for vid in TRAINER_SUBVIEW_IDS: if vid in self._view_map: await self._activate_view(vid) break self._update_header_trainer() async def _leave_trainer_context(self) -> None: """Verlaesst den Trainer-Kontext und kehrt zur Trainer-Liste zurueck.""" self._in_trainer_context = False self._trainer_data = {} # Footer auf Haupt-Menue zurueckschalten footer = self.query_one("#footer", TrixyFooter) footer.update_menu(self._main_menu) # Zurueck zur Trainer-Liste if "trainer" in self._view_map: await self._activate_view("trainer") self._parent_view_id = "" def _update_header_trainer(self) -> None: """Aktualisiert den Header im Trainer-Kontext.""" header = self.query_one("#header", TrixyHeader) name = self._trainer_data.get("name", "Trainer") if self._active_view: header.subtitle = f"Trainer > {name} > {self._active_view.TITLE}" else: header.subtitle = f"Trainer > {name}" def _update_header_satellite(self) -> None: """Aktualisiert den Header im Satellite-Kontext.""" header = self.query_one("#header", TrixyHeader) alias = self._satellite_data.get("alias", "Satellite") if self._active_view: header.subtitle = f"Satellites > {alias} > {self._active_view.TITLE}" else: header.subtitle = f"Satellites > {alias}" async def _activate_view(self, view_id: str) -> None: """Aktiviert einen View und deaktiviert den aktuellen.""" new_view = self._view_map.get(view_id) if not new_view or new_view is self._active_view: return # Alten View deaktivieren if self._active_view: await self._active_view.on_deactivate() self._active_view.display = False # Neuen View aktivieren new_view.display = True self._active_view = new_view await new_view.on_activate() # Header aktualisieren (nur wenn nicht in Sub-Kontext) if not self._in_satellite_context and not self._in_plugin_context and not self._in_schedule_context and not self._in_trainer_context: header = self.query_one("#header", TrixyHeader) header.subtitle = new_view.TITLE def action_scroll_up_page(self) -> None: """PageUp: Scrollt 5 Zeilen nach oben.""" if self._active_view: self._active_view.scroll_relative(y=-5, animate=False) def action_scroll_down_page(self) -> None: """PageDown: Scrollt 5 Zeilen nach unten.""" if self._active_view: self._active_view.scroll_relative(y=5, animate=False) async def _periodic_refresh(self) -> None: """Aktualisiert den aktiven View periodisch.""" if self._active_view and self._connection: try: await self._active_view.refresh_data() except Exception: pass