feat: stats module

2 files changed, 168 insertions, 0 deletions

diff --git a/deckit/modules/__init__.py b/deckit/modules/__init__.py
new file mode 100644
index 0000000..76fc9c5
--- /dev/null
+++ b/deckit/modules/__init__.py
@@ -0,0 +1,26 @@
+"""Module discovery.
+
+Every `deckit/modules/<name>.py` exporting `make(cfg) -> Module` is a module.
+Discovery is lazy and file-based, so a new module is a new file: nothing here
+lists the modules, and a not-yet-written one simply stays absent.
+"""
+
+import importlib
+from pathlib import Path
+
+from ..module import Registry
+
+
+def build(cfg: dict) -> Registry:
+    enabled = cfg.get("modules", {})
+    modules = []
+    for path in sorted(Path(__file__).parent.glob("*.py")):
+        name = path.stem
+        if name == "__init__" or not enabled.get(name, True):
+            continue
+        try:
+            mod = importlib.import_module(f"{__name__}.{name}")
+        except ImportError:
+            continue
+        modules.append(mod.make(cfg))
+    return Registry(modules)
diff --git a/deckit/modules/stats.py b/deckit/modules/stats.py
new file mode 100644
index 0000000..800524c
--- /dev/null
+++ b/deckit/modules/stats.py
@@ -0,0 +1,142 @@
+"""System stats: CPU, memory, temperature, GPU, network, uptime."""
+
+import shutil
+import socket
+import subprocess
+import time
+from pathlib import Path
+
+from ..module import Module
+
+
+def _run(cmd, timeout=2):
+    try:
+        return subprocess.run(cmd, capture_output=True, text=True, timeout=timeout).stdout.strip()
+    except (OSError, subprocess.TimeoutExpired):
+        return ""
+
+
+def _read(path, default=""):
+    try:
+        return Path(path).read_text().strip()
+    except OSError:
+        return default
+
+
+def _cpu_sensor(hw):
+    names = {_read(h / "name"): h for h in Path("/sys/class/hwmon").glob("hwmon*")}
+    want = hw.get("cpu_sensor", "auto")
+    for name in ([want] if want != "auto" else ["k10temp", "zenpower", "coretemp", "cpu_thermal", "acpitz"]):
+        if name in names:
+            return names[name] / "temp1_input"
+    return None
+
+
+def _battery(hw):
+    want = hw.get("battery", "auto")
+    if want != "auto":
+        return Path("/sys/class/power_supply") / want
+    for p in sorted(Path("/sys/class/power_supply").glob("*")):
+        if _read(p / "type") == "Battery":
+            return p
+    return None
+
+
+def _gpu(hw):
+    want = hw.get("gpu", "auto")
+    if want in ("auto", "nvidia") and shutil.which("nvidia-smi"):
+        for dev in Path("/sys/bus/pci/drivers/nvidia").glob("0000:*"):
+            info = _read(next(Path("/proc/driver/nvidia/gpus").glob(f"{dev.name}*/information"), "/dev/null"))
+            model = next((l.split(":", 1)[1].strip() for l in info.splitlines() if l.startswith("Model")), "NVIDIA GPU")
+            return {"kind": "nvidia", "dev": dev, "name": model.removeprefix("NVIDIA ").removeprefix("GeForce ")}
+    if want in ("auto", "amd"):
+        for busy in sorted(Path("/sys/class/drm").glob("card*/device/gpu_busy_percent")):
+            return {"kind": "amd", "dev": busy.parent, "name": "AMD GPU"}
+    return None
+
+
+def _net_iface(hw):
+    if hw.get("net", "auto") != "auto":
+        return hw["net"]
+    for line in _read("/proc/net/route").splitlines()[1:]:
+        f = line.split()
+        if f[1] == "00000000":
+            return f[0]
+    return None
+
+
+def make(cfg: dict) -> Module:
+    hw = cfg.get("hardware", {})
+    cpu_sensor = _cpu_sensor(hw)
+    battery = _battery(hw)
+    gpu = _gpu(hw)
+    prev = {"cpu": None, "net": None, "t": None, "gpu_t": 0, "gpu_v": None}
+
+    def cpu_percent():
+        fields = list(map(int, _read("/proc/stat").splitlines()[0].split()[1:]))
+        idle, total = fields[3] + fields[4], sum(fields)
+        last, prev["cpu"] = prev["cpu"], (idle, total)
+        if not last or total == last[1]:
+            return 0
+        return round(100 * (1 - (idle - last[0]) / (total - last[1])))
+
+    def net_rates(now):
+        iface = _net_iface(hw)
+        for line in _read("/proc/net/dev").splitlines():
+            if iface and line.strip().startswith(iface + ":"):
+                v = line.split(":")[1].split()
+                cur = (iface, int(v[0]), int(v[8]))
+                break
+        else:
+            return 0, 0
+        last, lt = prev["net"], prev["t"]
+        prev["net"], prev["t"] = cur, now
+        if not last or last[0] != iface or now == lt:
+            return 0, 0
+        return round((cur[1] - last[1]) / (now - lt)), round((cur[2] - last[2]) / (now - lt))
+
+    def read_gpu():
+        if not gpu:
+            return None
+        # Querying a suspended dGPU wakes it, so only ask when it is already awake.
+        if _read(gpu["dev"] / "power/runtime_status") == "suspended":
+            return {"asleep": True, "name": gpu["name"]}
+        if time.time() - prev["gpu_t"] < 5:
+            return prev["gpu_v"]
+        v = None
+        try:
+            if gpu["kind"] == "nvidia":
+                out = _run(["nvidia-smi", "--query-gpu=utilization.gpu,temperature.gpu,memory.used,memory.total",
+                            "--format=csv,noheader,nounits"])
+                u, t, mu, mt = map(int, out.splitlines()[0].split(", "))
+            else:
+                d = gpu["dev"]
+                u = int(_read(d / "gpu_busy_percent"))
+                temp = next(d.glob("hwmon/hwmon*/temp1_input"), None)
+                t = round(int(_read(temp)) / 1000) if temp else None
+                mu, mt = (int(_read(d / f"mem_info_vram_{x}", "0")) // 1048576 for x in ("used", "total"))
+            v = {"asleep": False, "name": gpu["name"], "util": u, "temp": t, "memUsed": mu, "memTotal": mt}
+        except (ValueError, IndexError):
+            pass
+        prev.update(gpu_t=time.time(), gpu_v=v)
+        return v
+
+    def state():
+        now = time.monotonic()
+        mem = {line.split(":")[0]: int(line.split()[1]) for line in _read("/proc/meminfo").splitlines()}
+        down, up = net_rates(now)
+        t = _read(cpu_sensor) if cpu_sensor else ""
+        return {
+            "host": socket.gethostname(),
+            "uptime": int(float(_read("/proc/uptime", "0").split()[0])),
+            "cpu": cpu_percent(),
+            "cpuTemp": round(int(t) / 1000) if t else None,
+            "memUsed": mem["MemTotal"] - mem["MemAvailable"],
+            "memTotal": mem["MemTotal"],
+            "gpu": read_gpu(),
+            "battery": {"level": int(_read(battery / "capacity", "0")), "status": _read(battery / "status")}
+            if battery else None,
+            "net": {"down": down, "up": up},
+        }
+
+    return Module(name="stats", title="System", state=state)