1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
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)
|
