mission-control-v2/backend/services/system.py

"""
System/OS-Metriken für die Box (Bosgame / Strix Halo).

CPU/RAM/Disk via psutil (plattformübergreifend). GPU-Auslastung/VRAM/Temperatur
via sysfs (amdgpu) — nur Linux; auf anderen Plattformen None (amd-smi fehlt auf
der Box, daher sysfs). Verschachtelte Struktur wie v1 (cpu.percent, ram.used Bytes).
"""

import glob
import os
import subprocess
import threading
import time

import psutil
from config import MODELS_DIR


def _read_int(path: str) -> int | None:
    try:
        with open(path) as f:
            return int(f.read().strip())
    except Exception:
        return None


def _gpu_sysfs() -> dict | None:
    """AMD-GPU-Auslastung + Speicher via sysfs (Linux). Findet die Basis-Card
    dynamisch (Strix Halo ist oft card1, nicht card0) und überspringt die
    Connector-Verzeichnisse (card1-DP-1 …). Strix Halo nutzt Unified Memory →
    GTT ist der eigentliche große Pool; VRAM ist nur der kleine Carve-out."""
    for dev in sorted(glob.glob("/sys/class/drm/card*/device")):
        card = dev.split("/")[-2]            # z.B. "card1" oder "card1-DP-1"
        if "-" in card:                      # Connector-Dir → kein GPU-Device
            continue
        busy = _read_int(f"{dev}/gpu_busy_percent")
        if busy is None:
            continue
        return {
            "busy_percent": busy,
            "vram_used": _read_int(f"{dev}/mem_info_vram_used"),
            "vram_total": _read_int(f"{dev}/mem_info_vram_total"),
            "gtt_used": _read_int(f"{dev}/mem_info_gtt_used"),
            "gtt_total": _read_int(f"{dev}/mem_info_gtt_total"),
        }
    return None


def _temps() -> dict | None:
    """CPU/GPU-Temperatur via hwmon (Linux). None bei Fehlen."""
    out: dict = {}
    for hw in glob.glob("/sys/class/hwmon/hwmon*"):
        name = ""
        try:
            with open(f"{hw}/name") as f:
                name = f.read().strip()
        except Exception:
            continue
        t = _read_int(f"{hw}/temp1_input")
        if t is None:
            continue
        c = round(t / 1000.0, 1)
        if name in ("k10temp", "zenpower", "coretemp"):
            out["cpu"] = c
        elif name in ("amdgpu", "edge"):
            out["gpu"] = c
    return out or None


def get_git_info(path: str) -> dict | None:
    expanded = os.path.expanduser(path)
    if not os.path.isdir(expanded) or not os.path.exists(os.path.join(expanded, ".git")):
        return None
    try:
        res = subprocess.run(
            ["git", "log", "-1", "--format=%h|%cd|%s", "--date=short"],
            cwd=expanded, capture_output=True, text=True, timeout=3
        )
        if res.returncode != 0:
            return None
        parts = res.stdout.strip().split("|", 2)
        h = parts[0]
        d = parts[1]
        s = parts[2] if len(parts) > 2 else ""
        
        branch_res = subprocess.run(
            ["git", "rev-parse", "--abbrev-ref", "HEAD"],
            cwd=expanded, capture_output=True, text=True, timeout=2
        )
        branch = branch_res.stdout.strip() if branch_res.returncode == 0 else "unknown"

        status_res = subprocess.run(
            ["git", "status", "--porcelain"],
            cwd=expanded, capture_output=True, text=True, timeout=2
        )
        dirty = bool(status_res.stdout.strip()) if status_res.returncode == 0 else False

        return {
            "hash": h,
            "date": d,
            "subject": s,
            "branch": branch,
            "dirty": dirty,
            "path": expanded
        }
    except Exception:
        return None


def find_hermes_agent_git() -> dict | None:
    env_path = os.environ.get("MC_HERMES_AGENT_PATH")
    if env_path:
        info = get_git_info(env_path)
        if info:
            return info

    candidates = [
        "~/hermes-agent",
        "~/.hermes/hermes-agent",
        "~/.hermes"
    ]
    for c in candidates:
        info = get_git_info(c)
        if info:
            return info
    return None


def get_engine_version() -> dict:
    engine_path = os.environ.get("MC_ENGINE_PATH", "/opt/llamacpp")
    git_info = get_git_info(engine_path)
    if git_info:
        return {**git_info, "type": "git"}
    
    candidates = [
        os.path.join(engine_path, "llama-server"),
        os.path.join(engine_path, "bin", "llama-server"),
        "/usr/local/bin/llama-server",
        "/usr/bin/llama-server",
        "llama-server"
    ]
    
    for binary in candidates:
        if binary != "llama-server" and not os.path.exists(binary):
            continue
        try:
            res = subprocess.run([binary, "--version"], capture_output=True, text=True, timeout=2)
            output = (res.stdout or "").strip() or (res.stderr or "").strip()
            if output:
                lines = output.splitlines()
                ver = lines[0] if lines else "unknown"
                return {"version_text": ver, "type": "binary"}
        except Exception:
            pass
            
    return {"type": "unknown"}


_VERSION_CACHE = {"ts": 0.0, "data": {}}
_VERSION_LOCK = threading.Lock()


def check_versions_cached() -> dict:
    import time
    now = time.time()
    if now - _VERSION_CACHE["ts"] < 30.0:
        return _VERSION_CACHE["data"]

    # Lock gegen Scan-Stampede: FastAPI führt sync-Routen im Threadpool aus → ohne Lock würden
    # parallele Dashboard-/Agent-Aufrufe die git-/Versions-Scans mehrfach gleichzeitig anwerfen.
    # Doppelt geprüft, damit ein zweiter Thread den frisch gefüllten Cache nimmt statt neu zu scannen.
    with _VERSION_LOCK:
        now = time.time()
        if now - _VERSION_CACHE["ts"] < 30.0:
            return _VERSION_CACHE["data"]

        mc2_path = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))
        data = {
            "mc2": get_git_info(mc2_path),
            "engine": get_engine_version(),
            "hermes_agent": find_hermes_agent_git()
        }
        _VERSION_CACHE["ts"] = now
        _VERSION_CACHE["data"] = data
        return data


def system_status() -> dict:
    vm = psutil.virtual_memory()
    try:
        du = psutil.disk_usage(str(MODELS_DIR) if MODELS_DIR.exists() else os.getcwd())
        disk = {"total": du.total, "used": du.used, "percent": du.percent}
    except Exception:
        disk = None
    return {
        "cpu": {"percent": psutil.cpu_percent(interval=0.1), "cores": psutil.cpu_count()},
        "ram": {"total": vm.total, "used": vm.used, "percent": vm.percent},
        "gpu": _gpu_sysfs(),
        "temp": _temps(),
        "disk": disk,
        # Betriebszeit in Sekunden (v3-Umbau P3). Gehoert in die neue Statusleiste, weil
        # sich die Box woechentlich selbst neu startet, wenn das OS es verlangt — dann ist
        # "laeuft seit 20 Minuten" die Antwort auf eine ganze Klasse von Fragen.
        "uptime_s": _uptime_s(),
        "versions": check_versions_cached(),
    }


def _uptime_s() -> int | None:
    """Sekunden seit dem Systemstart. None statt einer Ausrede, wenn psutil hier nichts
    liefert — eine erfundene Zahl waere schlimmer als eine fehlende."""
    try:
        return int(time.time() - psutil.boot_time())
    except Exception:
        return None