# -*- coding: utf-8 -*- """ Plugin SubView: Benutzerdefinierter Tab. Generischer View fuer plugin-spezifische Inhalte, die vom Plugin ueber get_config_tool_tabs() definiert werden. Unterstuetzte View-Typen: 1. Schema-basiert: Formular aus tab_data["schema"]["fields"] 2. Interactive List: Liste + Detail-Panel (tab_data["view_type"] = "interactive_list") 3. Info-Modus: Statischer Text (tab_data["description"]) """ from __future__ import annotations from typing import Any from textual.app import ComposeResult from textual.containers import Vertical, Horizontal from textual.widgets import Static, Button, Input, Switch, Select, ListView, ListItem, Label, DataTable from trixy_core.tui.views.base import MainView class PluginCustomView(MainView): """Dynamisch erstellter Tab fuer plugin-spezifische Inhalte.""" def __init__( self, tab_title: str, tab_id: str, tab_data: dict | None = None, **kwargs, ) -> None: super().__init__(**kwargs) self.TITLE = tab_title self.VIEW_ID = tab_id self._plugin_data: dict = {} self._tab_data: dict = tab_data or {} self._field_ids: list[str] = [] self._view_type = self._tab_data.get("view_type", "") self._has_schema: bool = bool( self._tab_data.get("schema", {}).get("fields") ) # Interactive List State self._list_items: list[dict] = [] self._list_extra: dict = {} # z.B. satellites self._selected_item: dict | None = None def set_plugin(self, plugin_data: dict) -> None: """Setzt die Plugin-Basisdaten.""" self._plugin_data = plugin_data def compose(self) -> ComposeResult: with Vertical(classes="section-box"): yield Static( f"[bold]{self.TITLE}[/bold]", classes="section-header" ) if self._view_type == "table": yield DataTable(id=f"{self.VIEW_ID}-table", zebra_stripes=True) yield Static("", id=f"{self.VIEW_ID}-status") elif self._view_type == "interactive_list": yield ListView(id=f"{self.VIEW_ID}-list") yield Static("", id=f"{self.VIEW_ID}-separator") yield Vertical(id=f"{self.VIEW_ID}-detail", classes="plugin-detail-panel") yield Button("Speichern", id=f"{self.VIEW_ID}-save", variant="success") yield Static("", id=f"{self.VIEW_ID}-status") elif self._has_schema: yield Vertical(id=f"{self.VIEW_ID}-form", classes="plugin-form-container") # Action-Buttons actions = self._tab_data.get("schema", {}).get("actions", []) for action in actions: yield Button( action.get("label", action.get("id", "Aktion")), id=f"{self.VIEW_ID}-action-{action['id']}", variant=action.get("variant", "default"), ) yield Button("Speichern", id=f"{self.VIEW_ID}-save", variant="success") yield Static("", id=f"{self.VIEW_ID}-status") else: desc = self._tab_data.get("description", "") yield Static( f"[dim]{desc}[/dim]" if desc else "[dim]Plugin-spezifische Inhalte[/dim]", id=f"{self.VIEW_ID}-content", ) async def on_activate(self) -> None: """Aktiviert den View.""" if self._view_type == "table": await self._load_table_data() elif self._view_type == "interactive_list": await self._load_list_data() elif self._has_schema: await self._refresh_config() await self._build_form() # ======================================================================== # Table View # ======================================================================== async def _load_table_data(self) -> None: """Laedt Daten und befuellt die DataTable.""" from trixy_core.utils.debug import pdebug if not self.connection or not self._plugin_data: self._set_status("[red]Keine Verbindung[/red]") return plugin_name = self._plugin_data.get("name", "") load_action = self._tab_data.get("load_action", "") if not plugin_name or not load_action: return self._set_status("[yellow]Lade Daten...[/yellow]") try: result = await self.connection.plugin_action( action=load_action, plugin_name=plugin_name, ) if not result or not result.success: self._set_status(f"[red]{result.error if result else 'Keine Antwort'}[/red]") return rows = result.result_data.get("rows", []) columns = self._tab_data.get("columns", []) table = self.query_one(f"#{self.VIEW_ID}-table", DataTable) table.clear(columns=True) # Spalten hinzufuegen for col in columns: table.add_column( col.get("label", col.get("key", "")), key=col.get("key", ""), width=col.get("width"), ) # Zeilen hinzufuegen (mit Rich Markup) from rich.text import Text for row in rows: cells = [] for col in columns: val = row.get(col.get("key", ""), "") cells.append(Text.from_markup(str(val))) table.add_row(*cells) self._set_status(f"[green]{len(rows)} Geraete[/green]") except Exception as e: self._set_status(f"[red]Fehler: {e}[/red]") # ======================================================================== # Interactive List # ======================================================================== async def _load_list_data(self) -> None: """Laedt Daten fuer die interaktive Liste vom Plugin.""" from trixy_core.utils.debug import pdebug if not self.connection: pdebug(f"[PluginCustomView] {self.VIEW_ID}: Keine Connection") self._set_status("[red]Keine Verbindung[/red]") return if not self._plugin_data: pdebug(f"[PluginCustomView] {self.VIEW_ID}: Keine plugin_data") self._set_status("[red]Keine Plugin-Daten[/red]") return plugin_name = self._plugin_data.get("name", "") load_action = self._tab_data.get("load_action", "") pdebug(f"[PluginCustomView] {self.VIEW_ID}: Lade {load_action} fuer {plugin_name}") if not plugin_name or not load_action: self._set_status("[red]Kein Plugin oder Aktion[/red]") return self._set_status("[yellow]Lade Daten...[/yellow]") try: result = await self.connection.plugin_action( action=load_action, plugin_name=plugin_name, ) pdebug(f"[PluginCustomView] Result type={type(result).__name__}, " f"success={getattr(result, 'success', '?')}, " f"result_data type={type(getattr(result, 'result_data', None)).__name__}, " f"result_data keys={list(getattr(result, 'result_data', {}).keys()) if result else 'None'}, " f"all attrs={[a for a in dir(result) if not a.startswith('_')]}") if not result or not result.success: error = result.error if result else "Keine Antwort" pdebug(f"[PluginCustomView] Fehler: {error}") self._set_status(f"[red]{error}[/red]") return list_key = self._tab_data.get("list_key", "items") rd = result.result_data self._list_items = rd.get(list_key, []) # Zusaetzliche Daten (z.B. Satellite-Liste) for key, value in result.result_data.items(): if key != list_key: self._list_extra[key] = value await self._populate_list() self._set_status(f"[green]{len(self._list_items)} Eintraege[/green]") except Exception as e: pdebug(f"[PluginCustomView] Exception: {e}") self._set_status(f"[red]Fehler: {e}[/red]") async def _populate_list(self) -> None: """Befuellt die ListView mit Eintraegen.""" try: list_view = self.query_one(f"#{self.VIEW_ID}-list", ListView) except Exception: return await list_view.clear() label_field = self._tab_data.get("list_label_field", "name") subtitle_field = self._tab_data.get("list_subtitle_field", "") subtitle_suffix = self._tab_data.get("list_subtitle_suffix", "") for item in self._list_items: label = str(item.get(label_field, "")) if subtitle_field and item.get(subtitle_field) is not None: label += f" [dim]({item[subtitle_field]}{subtitle_suffix})[/dim]" list_item = ListItem(Label(label)) await list_view.append(list_item) async def on_list_view_selected(self, event: ListView.Selected) -> None: """Wird aufgerufen wenn ein Listen-Eintrag ausgewaehlt wird.""" if self._view_type != "interactive_list": return index = event.list_view.index if index is None or index >= len(self._list_items): return self._selected_item = self._list_items[index] await self._build_detail_panel() async def _build_detail_panel(self) -> None: """Baut das Detail-Panel fuer den ausgewaehlten Eintrag.""" try: detail = self.query_one(f"#{self.VIEW_ID}-detail", Vertical) except Exception: return # Alte Widgets entfernen for child in list(detail.children): await child.remove() self._field_ids = [] if not self._selected_item: return item = self._selected_item detail_fields = self._tab_data.get("detail_fields", []) # Raumname als Header name = item.get("name", "") await detail.mount(Static(f"[bold]{name}[/bold]", classes="plugin-detail-header")) for field_def in detail_fields: fname = field_def.get("name", "") flabel = field_def.get("label", fname) ftype = field_def.get("type", "text") fhelp = field_def.get("help", "") value_from = field_def.get("value_from", fname) field_id = f"{self.VIEW_ID}-detail-{fname}" self._field_ids.append(field_id) await detail.mount(Static(flabel, classes="plugin-field-label")) if ftype == "readonly": raw_value = item.get(value_from, item.get(fname, "")) if isinstance(raw_value, list): raw_value = ", ".join(str(v) for v in raw_value) await detail.mount(Static(f" {raw_value or '—'}")) continue elif ftype == "text": raw_value = item.get(value_from, item.get(fname, "")) if isinstance(raw_value, list): raw_value = ", ".join(str(v) for v in raw_value) widget = Input( id=field_id, value=str(raw_value) if raw_value else "", placeholder=field_def.get("placeholder", ""), ) await detail.mount(widget) elif ftype == "select": choices_from = field_def.get("choices_from", "") choices_data = self._list_extra.get(choices_from, []) current = item.get(fname, "") # Optionen aufbauen: [("Label", "value"), ...] options: list[tuple[str, str]] = [("— Kein Satellite —", "")] valid_values: set[str] = {""} for choice in choices_data: if isinstance(choice, dict): c_id = choice.get("id", "") c_alias = choice.get("alias", "") c_room = choice.get("room_id", "") label = c_alias or c_id if c_room: label += f" ({c_room})" val = c_alias or c_room or c_id options.append((label, val)) valid_values.add(val) else: options.append((str(choice), str(choice))) valid_values.add(str(choice)) # Fallback: Wenn current nicht in Options, auf leer setzen if current not in valid_values: current = "" widget = Select( id=field_id, options=options, value=current if current else "", ) await detail.mount(widget) if fhelp: await detail.mount( Static(f"[dim]{fhelp}[/dim]", classes="plugin-field-help") ) # ======================================================================== # Schema-basiertes Formular # ======================================================================== async def _refresh_config(self) -> None: """Laedt die aktuelle Plugin-Config vom Server.""" if not self.connection or not self._plugin_data: return name = self._plugin_data.get("name", "") if not name: return try: detail = await self.connection.request_plugin_detail(name) if detail and not detail.error and detail.config: self._plugin_data["config"] = detail.config except Exception: pass async def _build_form(self) -> None: """Baut das Formular aus dem Schema dynamisch auf.""" form_id = f"{self.VIEW_ID}-form" try: container = self.query_one(f"#{form_id}", Vertical) except Exception: return for child in list(container.children): await child.remove() self._field_ids = [] schema = self._tab_data.get("schema", {}) fields = schema.get("fields", []) config_section = schema.get("config_section", "") config = self._plugin_data.get("config", {}) if config_section and isinstance(config.get(config_section), dict): section_config = config[config_section] else: section_config = config for field_def in fields: name = field_def.get("name", "") label = field_def.get("label", name) field_type = field_def.get("type", "text") help_text = field_def.get("help", "") default = field_def.get("default") current_value = section_config.get(name, default) field_id = f"{self.VIEW_ID}-field-{name}" self._field_ids.append(field_id) await container.mount(Static(label, classes="plugin-field-label")) if field_type == "checkbox": widget = Switch(id=field_id, value=bool(current_value)) elif field_type == "number": is_float = field_def.get("is_float", False) input_type = "number" if is_float else "integer" val_str = str(current_value) if current_value is not None else "" widget = Input(id=field_id, value=val_str, placeholder=str(default) if default is not None else "", type=input_type) elif field_type == "password": widget = Input(id=field_id, value=str(current_value) if current_value else "", password=True, placeholder=field_def.get("placeholder", "")) elif field_type == "select": choices = field_def.get("choices", []) options = [(label_text, value) for value, label_text in choices] widget = Select(id=field_id, options=options, value=current_value if current_value is not None else Select.BLANK) elif field_type == "readonly": widget = Static(f"[dim]{current_value or ''}[/dim]", id=field_id) else: widget = Input(id=field_id, value=str(current_value) if current_value is not None else "", placeholder=field_def.get("placeholder", "")) await container.mount(widget) if help_text: await container.mount( Static(f"[dim]{help_text}[/dim]", classes="plugin-field-help")) def _collect_values(self) -> dict[str, Any]: """Sammelt die Formularwerte.""" schema = self._tab_data.get("schema", {}) fields = schema.get("fields", []) result: dict[str, Any] = {} for field_def in fields: name = field_def.get("name", "") field_type = field_def.get("type", "text") field_id = f"{self.VIEW_ID}-field-{name}" if field_type == "readonly": continue try: if field_type == "checkbox": widget = self.query_one(f"#{field_id}", Switch) result[name] = widget.value elif field_type == "number": widget = self.query_one(f"#{field_id}", Input) is_float = field_def.get("is_float", False) if is_float: result[name] = float(widget.value) if widget.value else 0.0 else: result[name] = int(widget.value) if widget.value else 0 elif field_type == "select": widget = self.query_one(f"#{field_id}", Select) result[name] = widget.value if widget.value != Select.BLANK else "" else: widget = self.query_one(f"#{field_id}", Input) result[name] = widget.value except Exception: pass return result def _collect_detail_values(self) -> dict[str, Any]: """Sammelt die Detail-Panel Werte.""" detail_fields = self._tab_data.get("detail_fields", []) result: dict[str, Any] = {} for field_def in detail_fields: fname = field_def.get("name", "") ftype = field_def.get("type", "text") field_id = f"{self.VIEW_ID}-detail-{fname}" try: if ftype == "select": widget = self.query_one(f"#{field_id}", Select) result[fname] = widget.value if widget.value != Select.BLANK else "" else: widget = self.query_one(f"#{field_id}", Input) result[fname] = widget.value except Exception: pass return result # ======================================================================== # Button-Handler # ======================================================================== async def on_button_pressed(self, event: Button.Pressed) -> None: """Button Handler.""" btn_id = event.button.id or "" action_prefix = f"{self.VIEW_ID}-action-" if btn_id == f"{self.VIEW_ID}-save": if self._view_type == "interactive_list": await self._handle_save_detail() else: await self._handle_save() elif btn_id.startswith(action_prefix): action_id = btn_id[len(action_prefix):] await self._handle_action(action_id) async def _handle_save_detail(self) -> None: """Speichert Detail-Panel Werte ueber Plugin-Action.""" if not self._selected_item: self._set_status("[red]Kein Eintrag ausgewaehlt[/red]") return plugin_name = self._plugin_data.get("name", "") save_action = self._tab_data.get("save_action", "") if not plugin_name or not save_action: return values = self._collect_detail_values() values["room_name"] = self._selected_item.get("name", "") try: self._set_status("[yellow]Speichern...[/yellow]") result = await self.connection.plugin_action( action=save_action, plugin_name=plugin_name, params=values, ) if result and result.success: self._set_status(f"[green]{result.message}[/green]") # Liste neu laden await self._load_list_data() elif result: self._set_status(f"[red]{result.error}[/red]") except Exception as e: self._set_status(f"[red]Fehler: {e}[/red]") async def _handle_action(self, action_id: str) -> None: """Fuehrt eine plugin-spezifische Aktion aus.""" if not self.connection or not self._plugin_data: self._set_status("[red]Keine Verbindung[/red]") return plugin_name = self._plugin_data.get("name", "") if not plugin_name: return schema = self._tab_data.get("schema", {}) actions = schema.get("actions", []) action_def = next((a for a in actions if a.get("id") == action_id), None) params: dict[str, Any] = {} if action_def and action_def.get("collect_fields"): params = self._collect_values() try: self._set_status("[yellow]Wird geprueft...[/yellow]") result = await self.connection.plugin_action( action=action_id, plugin_name=plugin_name, params=params, ) if result and result.success: self._set_status(f"[green]{result.message}[/green]") elif result and result.error: self._set_status(f"[red]{result.error}[/red]") else: self._set_status("[red]Keine Antwort vom Server[/red]") except Exception as e: self._set_status(f"[red]Fehler: {e}[/red]") async def _handle_save(self) -> None: """Speichert Schema-Formular Werte.""" if not self.connection or not self._plugin_data: self._set_status("[red]Keine Verbindung[/red]") return plugin_name = self._plugin_data.get("name", "") if not plugin_name: return values = self._collect_values() schema = self._tab_data.get("schema", {}) config_section = schema.get("config_section", "") try: self._set_status("[yellow]Speichern...[/yellow]") for key, value in values.items(): path = f"{config_section}.{key}" if config_section else key result = await self.connection.write_plugin_config( plugin_name=plugin_name, key_path=path, value=value, ) if not result or not result.success: error = result.error if result else "Unbekannt" self._set_status(f"[red]Fehler bei {key}: {error}[/red]") return self._set_status("[green]Einstellungen gespeichert[/green]") except Exception as e: self._set_status(f"[red]Fehler: {e}[/red]") # ======================================================================== # Hilfsmethoden # ======================================================================== def _set_status(self, text: str) -> None: """Setzt die Statuszeile.""" status_id = f"{self.VIEW_ID}-status" try: status = self.query_one(f"#{status_id}", Static) status.update(text) except Exception: pass