# -*- coding: utf-8 -*- """ Trainer SubView: Optional (F4). Zeigt optionale Konfigurationsfelder und Aktions-Buttons. Wird ausgeblendet wenn der Trainer kein optionales Schema liefert. """ from __future__ import annotations import asyncio from textual.app import ComposeResult from textual.containers import Vertical, Horizontal from textual.widgets import Static, Input, Button, Checkbox from textual.timer import Timer from trixy_core.tui.views.base import MainView from trixy_core.tui.widgets.progress_dialog import ProgressDialog class TrainerOptionalView(MainView): """F4 im Trainer-Kontext: Optionale Daten + Aktionen.""" TITLE = "Optional" VIEW_ID = "trainer_optional" def __init__(self, **kwargs) -> None: super().__init__(**kwargs) self._trainer_data: dict = {} self._detail_data: object | None = None self._has_schema: bool = False self._field_ids: list[str] = [] self._multi_select_fields: dict[str, list[str]] = {} # name -> list of checkbox IDs self._multi_select_values: dict[str, list[tuple[str, str]]] = {} # name -> [(cb_id, value)] self._action_buttons: list[tuple[str, str]] = [] # (action_name, field_name) def set_trainer(self, trainer_data: dict) -> None: """Setzt die Trainer-Basisdaten.""" self._trainer_data = trainer_data self._detail_data = None self._has_schema = False self._field_ids = [] self._multi_select_fields = {} self._multi_select_values = {} self._action_buttons = [] @property def has_schema(self) -> bool: """Ob ein optionales Schema vorhanden ist.""" return self._has_schema def compose(self) -> ComposeResult: yield Static("", id="trainer-optional-header") yield Vertical(id="trainer-optional-form", classes="trainer-form-container") yield Static("", id="trainer-optional-status") async def on_activate(self) -> None: """Laedt optionales Schema.""" await self._load_detail() async def _load_detail(self) -> None: """Laedt Detail-Daten und baut optional ein Formular auf.""" if not self.connection or not self._trainer_data: return trainer_id = self._trainer_data.get("trainer_id", "") if not trainer_id: return try: detail = await self.connection.request_trainer_detail(trainer_id) except Exception: return if detail is None or detail.error: return self._detail_data = detail optional_schema = getattr(detail, "optional_schema", None) header = self.query_one("#trainer-optional-header", Static) if optional_schema is None: self._has_schema = False header.update("[dim]Keine optionalen Einstellungen fuer diesen Trainer[/dim]") return self._has_schema = True header.update("[bold]Optionale Einstellungen / Datengenerierung[/bold]") if not self._field_ids and not self._action_buttons and not self._multi_select_fields: await self._build_form(optional_schema) async def _build_form(self, schema: dict) -> None: """Baut das Formular aus dem optionalen Schema.""" container = self.query_one("#trainer-optional-form", Vertical) for child in list(container.children): await child.remove() self._field_ids = [] self._multi_select_fields = {} self._multi_select_values = {} self._action_buttons = [] fields = schema.get("fields", []) if isinstance(schema, dict) else [] for field_def in fields: name = field_def.get("name", "") label = field_def.get("label", name) help_text = field_def.get("help_text", "") default = field_def.get("default") field_type = field_def.get("field_type", "text") if field_type == "action": # Aktions-Button action_name = field_def.get("action_name", name) btn_id = f"trainer-action-{action_name}" self._action_buttons.append((action_name, name)) variant = field_def.get("variant", "primary") btn = Button(label, id=btn_id, variant=variant) await container.mount(btn) if help_text: help_widget = Static( f"[dim]{help_text}[/dim]", classes="trainer-field-help" ) await container.mount(help_widget) elif field_type == "multi_select": # Checkbox-Liste fuer Mehrfachauswahl options = field_def.get("options", []) defaults = default if isinstance(default, list) else [] label_widget = Static(label, classes="trainer-field-label") await container.mount(label_widget) cb_ids = [] cb_values = [] for opt_idx, opt in enumerate(options): opt_value = opt.get("value", "") if isinstance(opt, dict) else str(opt) opt_label = opt.get("label", opt_value) if isinstance(opt, dict) else str(opt) cb_id = f"trainer-ms-{name}-{opt_idx}" is_checked = opt_value in defaults cb = Checkbox(opt_label, id=cb_id, value=is_checked) await container.mount(cb) cb_ids.append(cb_id) cb_values.append((cb_id, opt_value)) self._multi_select_fields[name] = cb_ids self._multi_select_values[name] = cb_values if help_text: help_widget = Static( f"[dim]{help_text}[/dim]", classes="trainer-field-help" ) await container.mount(help_widget) else: # Normales Eingabefeld field_id = f"trainer-opt-{name}" self._field_ids.append(field_id) label_widget = Static(label, classes="trainer-field-label") await container.mount(label_widget) widget = Input( id=field_id, value=str(default) if default is not None else "", placeholder=field_def.get("placeholder", ""), ) await container.mount(widget) if help_text: help_widget = Static( f"[dim]{help_text}[/dim]", classes="trainer-field-help" ) await container.mount(help_widget) def _collect_params(self) -> dict: """Sammelt alle Formularwerte als Parameter-Dict.""" params = {} # Normale Input-Felder for field_id in self._field_ids: try: widget = self.query_one(f"#{field_id}", Input) name = field_id.replace("trainer-opt-", "", 1) params[name] = widget.value except Exception: pass # Multi-Select-Felder (Checkbox-Listen) for name, cb_values in self._multi_select_values.items(): selected = [] for cb_id, value in cb_values: try: cb = self.query_one(f"#{cb_id}", Checkbox) if cb.value: selected.append(value) except Exception: pass params[name] = selected return params # Aktionen die einen Progress-Dialog anzeigen sollen _PROGRESS_ACTIONS = {"generate_tts", "generate_negative", "generate_background_mix", "generate_from_raw"} async def on_button_pressed(self, event: Button.Pressed) -> None: """Behandelt Klicks auf Aktions-Buttons.""" btn_id = event.button.id or "" if not btn_id.startswith("trainer-action-"): return action_name = btn_id.replace("trainer-action-", "", 1) trainer_id = self._trainer_data.get("trainer_id", "") if not trainer_id or not self.connection: return # Aktuelle Formularwerte als Parameter sammeln params = self._collect_params() if action_name in self._PROGRESS_ACTIONS: await self._run_with_progress(trainer_id, action_name, params) else: await self._run_simple(trainer_id, action_name, params) async def _run_with_progress(self, trainer_id: str, action_name: str, params: dict) -> None: """Fuehrt eine Aktion mit Progress-Dialog aus.""" dialog = ProgressDialog(title="Generierung laeuft ...") self.app.push_screen(dialog) dialog.update_progress(0, 100, "Starte ...") # Aktion als Background-Task starten action_task = asyncio.create_task( self.connection.execute_trainer_action(trainer_id, action_name, params) ) # Progress-Polling per Timer (alle 0.5s) poll_timer: Timer | None = None async def _poll_progress() -> None: if action_task.done() or dialog.cancelled: if poll_timer is not None: poll_timer.stop() return try: progress = await self.connection.execute_trainer_action( trainer_id, "get_generation_progress", {}, ) if progress and progress.data: current = progress.data.get("current", 0) total = progress.data.get("total", 0) msg = progress.data.get("message", "") if total > 0: dialog.update_progress(current, total, msg) except Exception: pass poll_timer = self.set_interval(0.5, _poll_progress) # Auf Ergebnis warten try: result = await action_task except Exception as e: result = None if not dialog.cancelled: dialog.show_error(f"Fehler: {e}") # Polling stoppen if poll_timer is not None: poll_timer.stop() if dialog.cancelled: return status = self.query_one("#trainer-optional-status", Static) if result and result.success: dialog.finish(result.message) status.update(f"[green]{result.message}[/green]") elif result: dialog.show_error(result.message) status.update(f"[red]{result.message}[/red]") else: if not dialog.cancelled: dialog.show_error("Keine Antwort vom Server") status.update("[red]Keine Antwort vom Server[/red]") async def _run_simple(self, trainer_id: str, action_name: str, params: dict) -> None: """Fuehrt eine einfache Aktion ohne Progress-Dialog aus.""" status = self.query_one("#trainer-optional-status", Static) status.update(f"[yellow]Aktion '{action_name}' wird ausgefuehrt...[/yellow]") try: result = await self.connection.execute_trainer_action( trainer_id, action_name, params ) if result and result.success: status.update(f"[green]{result.message}[/green]") # Nach install_*-Aktionen: Schema neu laden (Stimmen erscheinen) if action_name.startswith("install_"): refresh = getattr(result, "refresh_schema", False) if refresh or action_name.startswith("install_"): self._field_ids = [] self._multi_select_fields = {} self._multi_select_values = {} self._action_buttons = [] await self._load_detail() elif result: status.update(f"[red]{result.message}[/red]") else: status.update("[red]Keine Antwort vom Server[/red]") except Exception as e: status.update(f"[red]Fehler: {e}[/red]")