Trixy7/trixy_core/tui/views/health_view.py

# -*- 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)