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
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
|
"""System stats: CPU, memory, GPU (Intel xe), network, uptime.
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 time
from pathlib import Path
from ..module import Module
UNITS = {"": 1, "K": 1024, "M": 1024 ** 2, "G": 1024 ** 3, "T": 1024 ** 4}
def _read(path, default=""):
try:
return Path(path).read_text().strip()
except OSError:
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")
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):
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
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)
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():
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
# 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"]
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
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)
|
