aboutsummaryrefslogtreecommitdiffstats
path: root/deckit/modules/stats.py
blob: c190ec826f3112ddd660bc73a84ef8a4098ca522 (plain)
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
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
# deck-it - stats module
# Copyright (C) 2026 Danilo M. <danix@danix.xyz>
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License version 2 as
# published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.

"""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)