# -*- coding: utf-8 -*- """ Trainer SubView: Dataset (F3). Zeigt Dataset-Informationen mit Aufschluesselung, Konfiguration der Verzeichnisse und Dataset-Optimierung. """ from __future__ import annotations from textual.app import ComposeResult from textual.containers import Vertical from textual.widgets import Static, Button, Input from trixy_core.tui.views.base import MainView from trixy_core.tui.widgets.status_table import StatusTable from trixy_core.tui.widgets.popup import ConfirmDialog class TrainerDatasetView(MainView): """F3 im Trainer-Kontext: Dataset-Informationen und Konfiguration.""" TITLE = "Dataset" VIEW_ID = "trainer_dataset" def __init__(self, **kwargs) -> None: super().__init__(**kwargs) self._trainer_data: dict = {} self._detail_data: object | None = None self._field_ids: list[str] = [] self._action_buttons: list[tuple[str, str]] = [] def set_trainer(self, trainer_data: dict) -> None: """Setzt die Trainer-Basisdaten.""" self._trainer_data = trainer_data self._detail_data = None self._field_ids = [] self._action_buttons = [] def compose(self) -> ComposeResult: with Vertical(classes="trainer-section"): yield Static("[bold]Dataset-Info[/bold]", classes="trainer-section-header") yield StatusTable(id="trainer-dataset-info") with Vertical(classes="trainer-section"): yield Static("[bold]Dataset-Konfiguration[/bold]", classes="trainer-section-header") yield Vertical(id="trainer-dataset-form", classes="trainer-form-container") yield Static("", id="trainer-dataset-status") async def on_activate(self) -> None: """Laedt Dataset-Info und Schema.""" await self._load_detail() async def refresh_data(self) -> None: """Aktualisiert Dataset-Zaehler.""" await self._load_detail() async def _load_detail(self) -> None: """Laedt Detail-Daten vom Server.""" 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 self._populate_info(detail) # Formular nur einmal aufbauen if not self._field_ids and not self._action_buttons: await self._build_form(detail) def _populate_info(self, detail: object) -> None: """Befuellt die Dataset-Info-Tabelle mit Aufschluesselung.""" info = getattr(detail, "dataset_info", {}) table = self.query_one("#trainer-dataset-info", StatusTable) rows = [] # Positive Aufschluesselung pos_bd = info.get("positive_breakdown", {}) if pos_bd: rows.append(("Positiv Gesamt", str(pos_bd.get("total", 0)))) rows.append((" Basis (Raw/TTS)", str(pos_bd.get("basis", 0)))) if pos_bd.get("morphs", 0): rows.append((" Morphs", str(pos_bd["morphs"]))) if pos_bd.get("bgmix", 0): rows.append((" Background-Mix", str(pos_bd["bgmix"]))) if pos_bd.get("augmented", 0): rows.append((" Augmentiert", str(pos_bd["augmented"]))) else: rows.append(("Positive Samples", str(info.get("positive_count", 0)))) rows.append(("", "")) # Negative Aufschluesselung neg_bd = info.get("negative_breakdown", {}) if neg_bd: rows.append(("Negativ Gesamt", str(neg_bd.get("total", 0)))) rows.append((" Basis (Assets/TTS/BG)", str(neg_bd.get("basis", 0)))) if neg_bd.get("morphs", 0): rows.append((" Morphs", str(neg_bd["morphs"]))) if neg_bd.get("bgmix", 0): rows.append((" Background-Mix", str(neg_bd["bgmix"]))) if neg_bd.get("augmented", 0): rows.append((" Augmentiert", str(neg_bd["augmented"]))) else: rows.append(("Negative Samples", str(info.get("negative_count", 0)))) # Verhaeltnis pos_total = pos_bd.get("total", info.get("positive_count", 0)) neg_total = neg_bd.get("total", info.get("negative_count", 0)) if pos_total > 0 and neg_total > 0: ratio = neg_total / pos_total rows.append(("", "")) rows.append(("Verhaeltnis Neg:Pos", f"1:{ratio:.1f}")) if ratio < 1.0: rows.append(("Empfehlung", "[yellow]Mehr Negative noetig![/yellow]")) elif pos_total > 600: rows.append(("Empfehlung", "[yellow]Dataset zu gross — Overfitting-Gefahr[/yellow]")) else: rows.append(("Empfehlung", "[green]OK[/green]")) # Verzeichnisse rows.append(("", "")) rows.append(("Positiv-Verzeichnis", info.get("positive_dir", ""))) rows.append(("Negativ-Verzeichnis", info.get("negative_dir", ""))) table.update_data(rows) async def _build_form(self, detail: object) -> None: """Baut das Dataset-Formular dynamisch auf.""" container = self.query_one("#trainer-dataset-form", Vertical) for child in list(container.children): await child.remove() self._field_ids = [] self._action_buttons = [] schema = getattr(detail, "dataset_schema", {}) fields = schema.get("fields", []) if isinstance(schema, dict) else [] dataset_info = getattr(detail, "dataset_info", {}) for field_def in fields: name = field_def.get("name", "") or field_def.get("key", "") label = field_def.get("label", name) help_text = field_def.get("help_text", "") or field_def.get("help", "") default = field_def.get("default") field_type = field_def.get("field_type", "") or field_def.get("type", "text") current_value = dataset_info.get(name, default) if field_type == "action": # Aktions-Button action_name = field_def.get("action_name", name) btn_id = f"trainer-ds-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: await container.mount( Static(f"[dim]{help_text}[/dim]", classes="trainer-field-help") ) else: # Input-Feld field_id = f"trainer-ds-{name}" self._field_ids.append(field_id) await container.mount(Static(label, classes="trainer-field-label")) is_float = field_def.get("is_float", False) if field_type == "number": input_type = "number" if is_float else "integer" widget = Input( id=field_id, value=str(current_value) if current_value is not None else "", placeholder=str(default) if default is not None else "", type=input_type, ) 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="trainer-field-help") ) def _collect_params(self) -> dict: """Sammelt alle Formularwerte.""" params = {} for field_id in self._field_ids: try: widget = self.query_one(f"#{field_id}", Input) name = field_id.replace("trainer-ds-", "", 1) params[name] = widget.value except Exception: pass return params async def on_button_pressed(self, event: Button.Pressed) -> None: """Behandelt Button-Klicks.""" btn_id = event.button.id or "" if not btn_id.startswith("trainer-ds-action-"): return action_name = btn_id.replace("trainer-ds-action-", "", 1) trainer_id = self._trainer_data.get("trainer_id", "") if not trainer_id or not self.connection: return if action_name == "optimize_dataset": await self._confirm_optimize(trainer_id) else: await self._execute_action(trainer_id, action_name) async def _confirm_optimize(self, trainer_id: str) -> None: """Zeigt Bestaetigungs-Dialog vor der Dataset-Optimierung.""" params = self._collect_params() target_pos = params.get("target_positive", "400") target_neg = params.get("target_negative", "600") max_var = params.get("max_variants", "2") # Aktuelle Zahlen aus der Info info = getattr(self._detail_data, "dataset_info", {}) if self._detail_data else {} pos_bd = info.get("positive_breakdown", {}) neg_bd = info.get("negative_breakdown", {}) pos_total = pos_bd.get("total", 0) neg_total = neg_bd.get("total", 0) pos_delete = max(0, pos_total - int(target_pos)) neg_delete = max(0, neg_total - int(target_neg)) if pos_delete == 0 and neg_delete == 0: status = self.query_one("#trainer-dataset-status", Static) status.update("[green]Dataset ist bereits im Zielbereich — nichts zu tun.[/green]") return message = ( f"[bold]Dataset-Optimierung[/bold]\n\n" f"Positiv: {pos_total} → ~{target_pos} " f"([red]~{pos_delete} loeschen[/red])\n" f"Negativ: {neg_total} → ~{target_neg} " f"([red]~{neg_delete} loeschen[/red])\n" f"Max Varianten pro Basis: {max_var}\n\n" f"Basis-Samples (Raw/TTS) bleiben erhalten.\n" f"Ueberfluessige Morphs/BG-Mix werden geloescht." ) dialog = ConfirmDialog(title="Dataset optimieren?", message=message) def on_confirm(confirmed: bool) -> None: if confirmed: self.call_later(self._execute_action, trainer_id, "optimize_dataset") self.app.push_screen(dialog, on_confirm) async def _execute_action(self, trainer_id: str, action_name: str) -> None: """Fuehrt eine Aktion aus.""" status = self.query_one("#trainer-dataset-status", Static) status.update(f"[yellow]Aktion '{action_name}' wird ausgefuehrt...[/yellow]") params = self._collect_params() 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]") # Schema und Info neu laden self._field_ids = [] 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]")