diff options
Diffstat (limited to 'deckit/modules/stats.py')
| -rw-r--r-- | deckit/modules/stats.py | 95 |
1 files changed, 58 insertions, 37 deletions
diff --git a/deckit/modules/stats.py b/deckit/modules/stats.py index 800524c..255efc6 100644 --- a/deckit/modules/stats.py +++ b/deckit/modules/stats.py @@ -1,19 +1,17 @@ -"""System stats: CPU, memory, temperature, GPU, network, uptime.""" +"""System stats: CPU, memory, GPU (Intel xe), network, uptime. -import shutil +The Arc driver exposes temperature and fan through hwmon and no VRAM capacity; +VRAM in use comes from the DRM fdinfo scan, as quickshell's perf panel does. +""" + +import re 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 "" +UNITS = {"": 1, "K": 1024, "M": 1024 ** 2, "G": 1024 ** 3, "T": 1024 ** 4} def _read(path, default=""): @@ -23,6 +21,39 @@ def _read(path, default=""): return default +def _parse_bytes(value): + if isinstance(value, int): + return value + m = re.fullmatch(r"(\d+(?:\.\d+)?)\s*([KMGT]?)(?:i?B)?", str(value).strip(), re.I) + if not m: + return None + return int(float(m[1]) * UNITS[m[2].upper()]) + + +def _intel_vram_used(root=Path("/proc")): + """Sum drm-resident-vram0 over xe clients, counting each client id once.""" + seen, total = set(), 0 + for fdinfo in root.glob("[0-9]*/fdinfo/*"): + try: + text = fdinfo.read_text() + except OSError: + continue + fields = {} + for line in text.splitlines(): + key, sep, value = line.partition(":") + if sep and key.startswith("drm-"): + fields[key] = value.strip() + client, resident = fields.get("drm-client-id"), fields.get("drm-resident-vram0") + if fields.get("drm-driver") != "xe" or not client or client in seen or not resident: + continue + m = re.match(r"(\d+)\s*([KMGT]?)(?:i?B)?", resident) + if not m: + continue + seen.add(client) + total += int(m[1]) * UNITS[m[2].upper()] + return total + + def _cpu_sensor(hw): names = {_read(h / "name"): h for h in Path("/sys/class/hwmon").glob("hwmon*")} want = hw.get("cpu_sensor", "auto") @@ -43,15 +74,11 @@ def _battery(hw): 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"} + if hw.get("gpu", "auto") == "none": + return None + for name in Path("/sys/class/hwmon").glob("hwmon*/name"): + if _read(name) == "xe": + return {"hwmon": name.parent, "name": "Intel Arc"} return None @@ -70,6 +97,7 @@ def make(cfg: dict) -> Module: cpu_sensor = _cpu_sensor(hw) battery = _battery(hw) gpu = _gpu(hw) + gpu_total = _parse_bytes(hw.get("gpu_mem_total", "")) if gpu else None prev = {"cpu": None, "net": None, "t": None, "gpu_t": 0, "gpu_v": None} def cpu_percent(): @@ -98,26 +126,19 @@ def make(cfg: dict) -> Module: 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: + # The fdinfo scan touches every process, so keep it to once every 2 s. + if time.time() - prev["gpu_t"] < 2: 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 + temp = _read(gpu["hwmon"] / "temp2_input") + fan = _read(gpu["hwmon"] / "fan1_input") + v = { + "name": gpu["name"], + "util": None, + "temp": round(int(temp) / 1000) if temp else None, + "fan": int(fan) if fan else None, + "memUsed": _intel_vram_used(), + "memTotal": gpu_total, + } prev.update(gpu_t=time.time(), gpu_v=v) return v |