# -*- coding: utf-8 -*- """ Health-View: Systemmetriken und Instanz-Status. Zeigt OS-Info, RAM, CPU, Disk, Netzwerk, Uptime und Trixy-spezifische Informationen. """ from __future__ import annotations from textual.app import ComposeResult from textual.containers import Vertical from textual.widgets import Static from trixy_core.tui.views.base import MainView from trixy_core.tui.widgets.status_table import StatusTable def _make_bar(percent: float, width: int = 30) -> str: """Erstellt einen farbigen ASCII-Balken.""" filled = int(width * percent / 100) empty = width - filled bar = "\u2593" * filled + "\u2591" * empty if percent > 80: color = "red" elif percent > 60: color = "yellow" else: color = "green" return f"[{color}]{bar}[/{color}] {percent:.1f}%" def _iface_icon(iface_type: str) -> str: """Gibt ein kurzes Label fuer den Interface-Typ zurueck.""" mapping = { "wifi": "[cyan]W-LAN[/cyan]", "ethernet": "[green]LAN[/green]", "loopback": "Loopback", "other": "Andere", } return mapping.get(iface_type, "Andere") def _format_bytes(num_bytes: int) -> str: """Formatiert Bytes in lesbare Groesse.""" for unit in ("B", "KB", "MB", "GB", "TB"): if abs(num_bytes) < 1024.0: return f"{num_bytes:.1f} {unit}" num_bytes /= 1024.0 return f"{num_bytes:.1f} PB" def _format_uptime(seconds: float) -> str: """Formatiert Sekunden in lesbare Uptime.""" days = int(seconds // 86400) hours = int((seconds % 86400) // 3600) minutes = int((seconds % 3600) // 60) secs = int(seconds % 60) parts = [] if days > 0: parts.append(f"{days}d") if hours > 0: parts.append(f"{hours}h") if minutes > 0: parts.append(f"{minutes}m") parts.append(f"{secs}s") return " ".join(parts) class HealthView(MainView): """ F1: System-Health Ansicht. Zeigt System-Metriken und Trixy-Status, periodisch aktualisiert. """ TITLE = "Health" VIEW_ID = "health" def compose(self) -> ComposeResult: with Vertical(classes="metric-section"): yield Static("[bold]System[/bold]", classes="metric-title") yield StatusTable(id="health-system") with Vertical(classes="metric-section"): yield Static("[bold]RAM[/bold]", classes="metric-title") yield StatusTable(id="health-ram") with Vertical(classes="metric-section"): yield Static("[bold]CPU[/bold]", classes="metric-title") yield StatusTable(id="health-cpu") with Vertical(classes="metric-section"): yield Static("[bold]Disk[/bold]", classes="metric-title") yield StatusTable(id="health-disk") with Vertical(classes="metric-section"): yield Static("[bold]Netzwerk[/bold]", classes="metric-title") yield StatusTable(id="health-network") with Vertical(classes="metric-section"): yield Static("[bold]Trixy[/bold]", classes="metric-title") yield StatusTable(id="health-trixy") async def on_activate(self) -> None: """Laedt initiale Daten.""" await self.refresh_data() async def refresh_data(self) -> None: """Aktualisiert alle Metriken.""" if not self.connection: return try: status = await self.connection.request_status() except Exception: return if status is None: return # System-Tabelle system_table = self.query_one("#health-system", StatusTable) system_table.update_data([ ("Betriebssystem", f"{status.os_name} {status.os_version}"), ("Hostname", status.hostname), ("Python", status.python_version), ("Uptime", _format_uptime(status.uptime_seconds)), ]) # RAM-Tabelle ram_table = self.query_one("#health-ram", StatusTable) ram_table.update_data([ ("Gesamt", f"{status.ram_total_mb:.0f} MB"), ("Belegt", f"{status.ram_used_mb:.0f} MB"), ("Auslastung", _make_bar(status.ram_percent)), ]) # CPU-Tabelle cpu_table = self.query_one("#health-cpu", StatusTable) cpu_rows = [ ("Modell", status.cpu_model or "Unbekannt"), ("Kerne", str(status.cpu_count)), ("Auslastung", _make_bar(status.cpu_percent)), ] if status.cpu_freq_mhz > 0: cpu_rows.append(("Frequenz", f"{status.cpu_freq_mhz:.0f} MHz")) cpu_table.update_data(cpu_rows) # Disk-Tabelle disk_table = self.query_one("#health-disk", StatusTable) disk_table.update_data([ ("Gesamt", f"{status.disk_total_gb:.1f} GB"), ("Belegt", f"{status.disk_used_gb:.1f} GB"), ("Auslastung", _make_bar(status.disk_percent)), ]) # Netzwerk-Tabelle net_table = self.query_one("#health-network", StatusTable) net_rows = [ ("Gesendet gesamt", _format_bytes(status.net_bytes_sent)), ("Empfangen gesamt", _format_bytes(status.net_bytes_recv)), ("Verbindungen", str(status.net_connections)), ] # Pro-Interface Stats (wenn vom Server geliefert) interfaces = getattr(status, "net_interfaces", None) or [] # Loopback und down-Interfaces rausfiltern relevant = [ i for i in interfaces if i.get("type") != "loopback" and i.get("is_up", False) ] if relevant: # Trennzeile net_rows.append(("", "")) for iface in relevant: name = iface.get("name", "?") iface_type = iface.get("type", "other") bytes_sent = iface.get("bytes_sent", 0) bytes_recv = iface.get("bytes_recv", 0) sat_count = iface.get("satellites_count", 0) icon = _iface_icon(iface_type) label = f"{icon} {name}" # Header-Zeile fuer das Interface (fett) net_rows.append((f"[bold]{label}[/bold]", "")) net_rows.append((" Gesendet", _format_bytes(bytes_sent))) net_rows.append((" Empfangen", _format_bytes(bytes_recv))) if sat_count > 0: net_rows.append((" Satellites", str(sat_count))) net_table.update_data(net_rows) # Trixy-Tabelle trixy_table = self.query_one("#health-trixy", StatusTable) trixy_rows = [ ("Instanz-Typ", status.instance_type), ("Version", status.trixy_version), ] if status.satellites_total > 0 or status.instance_type == "server": trixy_rows.append(( "Satellites", f"{status.satellites_connected}/{status.satellites_total} verbunden" )) if status.plugins_loaded: trixy_rows.append(("Plugins", ", ".join(status.plugins_loaded))) else: trixy_rows.append(("Plugins", "Keine")) if status.services_running: trixy_rows.append(("Services", ", ".join(status.services_running))) trixy_table.update_data(trixy_rows)