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 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-webui/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": {}}


def check_versions_cached() -> dict:
    import time
    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_ui": get_git_info("~/hermes-webui"),
        "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,
        "versions": check_versions_cached(),
    }