diff options
Diffstat (limited to 'modules/desktop/eww/system-stats.py')
| -rw-r--r-- | modules/desktop/eww/system-stats.py | 217 |
1 files changed, 197 insertions, 20 deletions
diff --git a/modules/desktop/eww/system-stats.py b/modules/desktop/eww/system-stats.py index 380dc25..1f0bf9a 100644 --- a/modules/desktop/eww/system-stats.py +++ b/modules/desktop/eww/system-stats.py @@ -1,13 +1,81 @@ #!/usr/bin/env python3 -"""system-stats: emit {cpu, ram, net} as JSON. CPU/RAM sample on an interval -inside one long-lived process (no per-tick spawn)""" +"""system-stats: emit {cpu, ram, net} as JSON. + +net is now an object: { + "status": "up"|"limited"|"down", + "type": "wifi"|"ethernet"|None, + "iface": "wlan0"|None, + "speed": 866|None, # link speed in Mbps + "rx": "12.5M"|None, # formatted download throughput rate (B/K/M/G) + "tx": "540K"|None # formatted upload throughput rate (B/K/M/G) +} + +Requires `iw` (for wifi link speed) and `iproute2` (for route lookup) on PATH. +""" import json +import os import subprocess import threading import time STATE_LOCK = threading.Lock() -STATE = {"cpu": 0, "ram": 0, "net": "down"} +STATE = { + "cpu": 0, + "ram": 0, + "net": { + "status": "down", + "type": None, + "iface": None, + "speed": None, + "rx": None, + "tx": None, + }, +} + +# Module-level counters for network byte delta calculations +NET_LAST_TIME = None +NET_LAST_BYTES = {} # {iface: (rx_bytes, tx_bytes)} + + +def format_rate(bytes_per_sec): + """Format bytes/sec into a ultra-compact 3-5 character string (e.g., '500B', '1.2M').""" + if bytes_per_sec is None or bytes_per_sec < 0: + return "0B" + + units = ["B", "K", "M", "G", "T"] + val = float(bytes_per_sec) + unit_idx = 0 + + while val >= 1024 and unit_idx < len(units) - 1: + val /= 1024.0 + unit_idx += 1 + + unit = units[unit_idx] + + if val >= 100 or unit == "B": + return f"{int(val)}{unit}" + elif val >= 10: + return f"{val:.1f}{unit}" + else: + return f"{val:.1f}{unit}" + + +def get_iface_io_bytes(iface): + """Read cumulative rx/tx bytes for a specific interface from /proc/net/dev.""" + if not iface: + return None, None + try: + with open("/proc/net/dev") as f: + for line in f: + if ":" not in line: + continue + dev, data = line.split(":", 1) + if dev.strip() == iface: + fields = data.split() + return int(fields[0]), int(fields[8]) + except (OSError, ValueError, IndexError): + pass + return None, None def read_cpu_times(): @@ -23,12 +91,15 @@ def cpu_ram_loop(interval=2.0): prev_idle, prev_total = read_cpu_times() while True: time.sleep(interval) + + # Calculate CPU usage idle, total = read_cpu_times() d_idle = idle - prev_idle d_total = total - prev_total prev_idle, prev_total = idle, total cpu_pct = round((1 - d_idle / d_total) * 100) if d_total else 0 + # Calculate RAM usage with open("/proc/meminfo") as f: meminfo = {} for line in f: @@ -38,34 +109,135 @@ def cpu_ram_loop(interval=2.0): avail_kb = meminfo["MemAvailable"] ram_pct = round((1 - avail_kb / total_kb) * 100) if total_kb else 0 + # Refresh network metrics on the main ticker (for live throughput) + net_status = check_net_status() + with STATE_LOCK: STATE["cpu"] = cpu_pct STATE["ram"] = ram_pct + STATE["net"] = net_status emit() -def check_net_status() -> str: +def get_default_iface(): + """Interface carrying the default route, or None.""" try: - with open("/run/systemd/netif/state") as f: - state = dict( - line.strip().split("=", 1) for line in f if "=" in line - ) - except FileNotFoundError: + out = subprocess.check_output( + ["ip", "-o", "route", "show", "default"], + text=True, stderr=subprocess.DEVNULL, + ) + except (subprocess.CalledProcessError, FileNotFoundError): + return None + for line in out.splitlines(): + parts = line.split() + if "dev" in parts: + return parts[parts.index("dev") + 1] + return None + + +def get_operstate(iface): + try: + with open(f"/sys/class/net/{iface}/operstate") as f: + return f.read().strip() + except OSError: return "down" - oper = state.get("OPER_STATE", "off") - if oper == "routable": - return "up" - if oper in ("degraded", "carrier", "degraded-carrier"): - return "limited" - return "down" +def get_iface_type(iface): + if os.path.exists(f"/sys/class/net/{iface}/wireless"): + return "wifi" + if os.path.isdir(f"/sys/class/net/{iface}/bridge"): + return "bridge" + if os.path.exists(f"/sys/class/net/{iface}/tun_flags"): + return "tun" + return "ethernet" -def net_loop(): - with STATE_LOCK: - STATE["net"] = check_net_status() - emit() +def get_speed_mbps(iface, iftype): + if iftype == "ethernet": + try: + with open(f"/sys/class/net/{iface}/speed") as f: + return int(f.read().strip()) + except (OSError, ValueError): + return None + if iftype == "wifi": + try: + out = subprocess.check_output( + ["iw", "dev", iface, "link"], + text=True, stderr=subprocess.DEVNULL, + ) + except (subprocess.CalledProcessError, FileNotFoundError): + return None + for line in out.splitlines(): + line = line.strip() + if line.startswith("tx bitrate:"): + try: + return round(float(line.split(":", 1)[1].split()[0])) + except (IndexError, ValueError): + return None + return None + + +def check_net_status(): + global NET_LAST_TIME, NET_LAST_BYTES + + iface = get_default_iface() + now = time.monotonic() + + if not iface: + NET_LAST_TIME = now + NET_LAST_BYTES = {} + return { + "status": "down", + "type": None, + "iface": None, + "speed": None, + "rx": None, + "tx": None, + } + + oper = get_operstate(iface) + if oper == "up": + status = "up" + elif oper == "dormant": + status = "limited" + else: + status = "down" + + iftype = get_iface_type(iface) + speed = get_speed_mbps(iface, iftype) if status != "down" else None + + # Calculate throughput rate (bytes/sec) + rx_bytes, tx_bytes = get_iface_io_bytes(iface) + rx_str, tx_str = "0B", "0B" + + if status != "down" and rx_bytes is not None and tx_bytes is not None: + if NET_LAST_TIME is not None and iface in NET_LAST_BYTES: + dt = now - NET_LAST_TIME + if dt > 0: + prev_rx, prev_tx = NET_LAST_BYTES[iface] + rx_rate = max(0, (rx_bytes - prev_rx) / dt) + tx_rate = max(0, (tx_bytes - prev_tx) / dt) + rx_str = format_rate(rx_rate) + tx_str = format_rate(tx_rate) + NET_LAST_BYTES[iface] = (rx_bytes, tx_bytes) + else: + rx_str, tx_str = None, None + + NET_LAST_TIME = now + + return { + "status": status, + "type": iftype, + "iface": iface, + "speed": speed, + "rx": rx_str, + "tx": tx_str, + } + + +def net_loop(): + """Event-driven listener for link state transitions.""" proc = subprocess.Popen( ["inotifywait", "-m", "-e", "modify", "/run/systemd/netif/state"], stdout=subprocess.PIPE, text=True, bufsize=1, @@ -82,9 +254,14 @@ def emit(): def main(): + # Initial state calculation before loop + with STATE_LOCK: + STATE["net"] = check_net_status() + emit() + threading.Thread(target=cpu_ram_loop, daemon=True).start() threading.Thread(target=net_loop, daemon=True).start() - threading.Event().wait() # keep main thread alive + threading.Event().wait() if __name__ == "__main__": |