Refactor: Store samples in a Sample Dataclass instead of loose dicts

AGENTS.md

@@ -58,7 +58,7 @@ agent's own shell command line and kills the session.
 ## Conventions
 
 - No code comments (the codebase has none).
-- basedpyright is configured as linter 
+- basedpyright is configured as linter, use with `uvx`.
 - Match surrounding style; keep functions small and typed where the codebase already is.
 - Keep polling endpoints cheap: collectors may cache lookups (unit names,
   enabled-state maps, SSID, temperature paths) with short TTLs.

app/collect/cpu.py

@@ -2,6 +2,8 @@ import glob
 
 import psutil
 
+from app.sample import Sample
+
 _temp_path: str | None = None
 _temp_checked = False
 
@@ -52,13 +54,10 @@ def core_count() -> int:
     return psutil.cpu_count(logical=True) or 1
 
 
-def sample() -> dict[str, float]:
+def fill(s: Sample) -> None:
-    out: dict[str, float] = {"cpu": psutil.cpu_percent(None)}
+    s.cpu = psutil.cpu_percent(None)
-    t = temp()
+    s.cpu_temp = temp()
-    if t is not None:
-        out["cpu_temp"] = t
     l1, l5, l15 = psutil.getloadavg()
-    out["load1"] = l1
+    s.load1 = l1
-    out["load5"] = l5
+    s.load5 = l5
-    out["load15"] = l15
+    s.load15 = l15
-    return out

app/collect/disks.py

@@ -8,7 +8,7 @@ def counters() -> dict[str, sdiskio]:
     return psutil.disk_io_counters(perdisk=True) or {}
 
 
-def rates(prev: dict[str, sdiskio], dt: float) -> dict[str, float]:
+def rates(prev: dict[str, sdiskio], dt: float) -> tuple[float, float]:
     cur = counters()
     r = 0
     w = 0
@@ -17,7 +17,7 @@ def rates(prev: dict[str, sdiskio], dt: float) -> dict[str, float]:
         if p is not None and dt > 0:
             r += max(0, int(c.read_bytes) - int(p.read_bytes))
             w += max(0, int(c.write_bytes) - int(p.write_bytes))
-    return {"io_read": r / dt if dt > 0 else 0.0, "io_write": w / dt if dt > 0 else 0.0}
+    return (r / dt if dt > 0 else 0.0, w / dt if dt > 0 else 0.0)
 
 
 def partitions() -> list[dict[str, Any]]:

app/collect/gpu.py

@@ -2,7 +2,8 @@ import glob
 import re
 import shutil
 import subprocess
-from typing import Any
+
+from app.sample import Sample
 
 _name_cache: str | None = None
 
@@ -44,10 +45,10 @@ def _gpu_name() -> str:
     return _name_cache
 
 
-def _amd_sample() -> dict[str, Any] | None:
+def _amd(s: Sample) -> bool:
     devices = sorted(glob.glob("/sys/class/drm/card[0-9]*/device/gpu_busy_percent"))
     if not devices:
-        return None
+        return False
     busy_sum = 0
     count = 0
     vram_used = 0
@@ -70,20 +71,19 @@ def _amd_sample() -> dict[str, Any] | None:
                 except ValueError:
                     pass
     if count == 0:
-        return None
+        return False
-    return {
+    s.gpu = round(busy_sum / count, 1)
-        "gpu": round(busy_sum / count, 1),
+    s.vram_used = vram_used
-        "vram_used": vram_used,
+    s.vram_total = vram_total
-        "vram_total": vram_total,
+    s.vram_pct = round(vram_used / vram_total * 100, 1) if vram_total else None
-        "vram_pct": round(vram_used / vram_total * 100, 1) if vram_total else None,
+    s.gpu_temp = max(temps) if temps else None
-        "gpu_temp": max(temps) if temps else None,
+    s.gpu_name = _gpu_name()
-        "gpu_name": _gpu_name(),
+    return True
-    }
 
 
-def _nvidia_sample() -> dict[str, Any] | None:
+def _nvidia(s: Sample) -> bool:
     if not shutil.which("nvidia-smi"):
-        return None
+        return False
     try:
         out = subprocess.run(
             [
@@ -97,10 +97,10 @@ def _nvidia_sample() -> dict[str, Any] | None:
             check=True,
         ).stdout
     except (OSError, subprocess.SubprocessError):
-        return None
+        return False
     lines = [l for l in out.splitlines() if l.strip()]
     if not lines:
-        return None
+        return False
     busy = used = total = 0
     temp = 0
     for line in lines:
@@ -115,26 +115,14 @@ def _nvidia_sample() -> dict[str, Any] | None:
     name = lines[0].split(",")[-1].strip()
     vram_used = used * 1024 * 1024
     vram_total = total * 1024 * 1024
-    return {
+    s.gpu = round(busy / len(lines), 1)
-        "gpu": round(busy / len(lines), 1),
+    s.vram_used = vram_used
-        "vram_used": vram_used,
+    s.vram_total = vram_total
-        "vram_total": vram_total,
+    s.vram_pct = round(vram_used / vram_total * 100, 1) if vram_total else None
-        "vram_pct": round(vram_used / vram_total * 100, 1) if vram_total else None,
+    s.gpu_temp = float(temp)
-        "gpu_temp": float(temp),
+    s.gpu_name = name
-        "gpu_name": name,
+    return True
-    }
 
 
-def sample() -> dict[str, Any]:
+def fill(s: Sample) -> None:
-    return (
+    _ = _amd(s) or _nvidia(s)
-        _amd_sample()
-        or _nvidia_sample()
-        or {
-            "gpu": None,
-            "vram_used": None,
-            "vram_total": None,
-            "vram_pct": None,
-            "gpu_temp": None,
-            "gpu_name": "no GPU detected",
-        }
-    )

app/collect/mem.py

@@ -1,14 +1,14 @@
 import psutil
 
+from app.sample import Sample
 
-def sample() -> dict[str, int | float]:
+
+def fill(s: Sample) -> None:
     v = psutil.virtual_memory()
-    s = psutil.swap_memory()
+    s.mem_used = v.used
-    return {
+    s.mem_total = v.total
-        "mem_used": v.used,
+    s.mem_pct = v.percent
-        "mem_total": v.total,
+    sw = psutil.swap_memory()
-        "mem_pct": v.percent,
+    s.swap_used = sw.used
-        "swap_used": s.used,
+    s.swap_total = sw.total
-        "swap_total": s.total,
+    s.swap_pct = sw.percent
-        "swap_pct": s.percent,
-    }

app/collect/power.py

@@ -1,5 +1,6 @@
 import glob
-from typing import Any
+
+from app.sample import Sample
 
 _PS = "/sys/class/power_supply"
 
@@ -21,8 +22,7 @@ def _supplies() -> list[tuple[str, str]]:
     return out
 
 
-def sample() -> dict[str, Any | None]:
+def fill(s: Sample) -> None:
-    out: dict[str, Any | None] = {"battery": None, "battery_status": None, "ac_online": None}
     try:
         supplies = _supplies()
         for t, p in supplies:
@@ -30,20 +30,19 @@ def sample() -> dict[str, Any | None]:
                 cap = _read(f"{p}/capacity")
                 if cap is not None:
                     try:
-                        out["battery"] = int(cap)
+                        s.battery = int(cap)
                     except ValueError:
                         pass
-                    out["battery_status"] = _read(f"{p}/status")
+                    s.battery_status = _read(f"{p}/status")
                 break
         for t, p in supplies:
             if t == "mains" and _read(f"{p}/online") == "1":
-                out["ac_online"] = True
+                s.ac_online = True
                 break
-        if out["ac_online"] is None:
+        if s.ac_online is None:
             for t, p in supplies:
                 if t == "usb" and _read(f"{p}/online") == "1":
-                    out["ac_online"] = True
+                    s.ac_online = True
                     break
     except OSError:
         pass
-    return out

app/routers/overview.py

@@ -2,6 +2,7 @@ import asyncio
 import math
 import socket
 import time
+from dataclasses import fields
 from typing import Any
 
 import psutil
@@ -11,25 +12,27 @@ from fastapi.responses import HTMLResponse, JSONResponse
 from app.collect import net as net_col
 from app.config import get_settings
 from app.render import render, uptime_str
+from app.sample import Sample
 
 router = APIRouter(prefix="/api", tags=["overview"])
 
 RowAgg = dict[str, float | int | None]
 
 
-def _window(
+def _window(snap: list[Sample], max_points: int) -> list[tuple[float, dict[str, RowAgg]]]:
-    snap: list[tuple[float, dict[str, float | int | None]]], max_points: int
-) -> list[tuple[float, dict[str, RowAgg]]]:
     n = len(snap)
     w = max(1, math.ceil(n / max_points))
     out: list[tuple[float, dict[str, RowAgg]]] = []
     for start in range(0, n, w):
         chunk = snap[start : start + w]
         vals: dict[str, list[int | float]] = {}
-        for _, sample in chunk:
+        for sample in chunk:
-            for k, v in sample.items():
+            for f in fields(sample):
+                if f.name == "ts":
+                    continue
+                v = getattr(sample, f.name)
                 if isinstance(v, (int, float)) and not isinstance(v, bool):
-                    vals.setdefault(k, []).append(v)
+                    vals.setdefault(f.name, []).append(v)
         row: dict[str, RowAgg] = {}
         for k, lst in vals.items():
             ints = all(isinstance(v, int) for v in lst)
@@ -39,39 +42,39 @@ def _window(
                 "min": min(lst) if ints else round(min(lst), 1),
                 "max": max(lst) if ints else round(max(lst), 1),
             }
-        out.append((chunk[-1][0], row))
+        out.append((chunk[-1].ts, row))
     return out
 
 
 @router.get("/overview")
 async def overview(request: Request):
     store = request.app.state.store
-    s: dict[str, Any] = store.latest() or {}
+    s = store.latest() or Sample()
-    mem_total = s.get("mem_total") or 0
+    mem_total = s.mem_total or 0
-    vram_total = s.get("vram_total") or 0
+    vram_total = s.vram_total or 0
-    vram_used = s.get("vram_used")
+    vram_used = s.vram_used
     c = {
-        "cpu": s.get("cpu"),
+        "cpu": s.cpu,
-        "cpu_temp": s.get("cpu_temp"),
+        "cpu_temp": s.cpu_temp,
-        "load1": s.get("load1"),
+        "load1": s.load1,
-        "load5": s.get("load5"),
+        "load5": s.load5,
-        "load15": s.get("load15"),
+        "load15": s.load15,
-        "mem_used": s.get("mem_used"),
+        "mem_used": s.mem_used,
         "mem_total": mem_total,
-        "mem_pct": s.get("mem_pct"),
+        "mem_pct": s.mem_pct,
-        "swap_used": s.get("swap_used"),
+        "swap_used": s.swap_used,
-        "swap_total": s.get("swap_total") or 0,
+        "swap_total": s.swap_total or 0,
-        "swap_pct": s.get("swap_pct"),
+        "swap_pct": s.swap_pct,
-        "gpu": s.get("gpu"),
+        "gpu": s.gpu,
-        "gpu_name": s.get("gpu_name"),
+        "gpu_name": s.gpu_name,
-        "gpu_temp": s.get("gpu_temp"),
+        "gpu_temp": s.gpu_temp,
         "vram_used": vram_used,
         "vram_total": vram_total,
-        "vram_pct": s.get("vram_pct")
+        "vram_pct": s.vram_pct
         or ((vram_used / vram_total * 100) if (vram_total and vram_used is not None) else None),
-        "battery": s.get("battery"),
+        "battery": s.battery,
-        "battery_status": s.get("battery_status"),
+        "battery_status": s.battery_status,
-        "ac_online": s.get("ac_online"),
+        "ac_online": s.ac_online,
         "uptime": uptime_str(time.time() - psutil.boot_time()),
         "hostname": socket.gethostname(),
         "cores": psutil.cpu_count(logical=True) or 1,

app/sample.py

@@ -0,0 +1,28 @@
+from dataclasses import dataclass
+
+
+@dataclass
+class Sample:
+    ts: float = 0.0
+    cpu: float = 0.0
+    cpu_temp: float | None = None
+    load1: float = 0.0
+    load5: float = 0.0
+    load15: float = 0.0
+    mem_used: int = 0
+    mem_total: int = 0
+    mem_pct: float = 0.0
+    swap_used: int = 0
+    swap_total: int = 0
+    swap_pct: float = 0.0
+    gpu: float | None = None
+    vram_used: int | None = None
+    vram_total: int | None = None
+    vram_pct: float | None = None
+    gpu_temp: float | None = None
+    gpu_name: str = "no GPU detected"
+    battery: int | None = None
+    battery_status: str | None = None
+    ac_online: bool | None = None
+    io_read: float = 0.0
+    io_write: float = 0.0

app/sampling.py

@@ -2,15 +2,16 @@ import asyncio
 import time
 
 from app.collect import cpu, disks, gpu, mem, power
+from app.sample import Sample
 from app.state import HistoryStore
 
 
-def _collect() -> dict[str, float | int | None]:
+def _collect() -> Sample:
-    sample: dict[str, float | int | None] = {}
+    sample = Sample()
-    sample.update(cpu.sample())
+    cpu.fill(sample)
-    sample.update(mem.sample())
+    mem.fill(sample)
-    sample.update(gpu.sample())
+    gpu.fill(sample)
-    sample.update(power.sample())
+    power.fill(sample)
     return sample
 
 
@@ -23,7 +24,7 @@ async def sampler_loop(store: HistoryStore, sample_interval: float) -> None:
         sample = await asyncio.to_thread(_collect)
         now = time.monotonic()
         dt = now - prev_t
-        sample.update(disks.rates(prev_disk, dt))
+        sample.io_read, sample.io_write = disks.rates(prev_disk, dt)
         prev_disk = disks.counters()
         prev_t = now
         store.record(sample)

app/state.py

@@ -1,19 +1,22 @@
 import time
 from collections import deque
 
+from app.sample import Sample
+
 
 class HistoryStore:
     def __init__(self, maxlen: int) -> None:
-        self._buf: deque[tuple[float, dict[str, float | int | None]]] = deque(maxlen=maxlen)
+        self._buf: deque[Sample] = deque(maxlen=maxlen)
 
-    def record(self, sample: dict[str, float | int | None]) -> None:
+    def record(self, sample: Sample) -> None:
-        self._buf.append((time.time(), sample))
+        sample.ts = time.time()
+        self._buf.append(sample)
 
-    def snapshot(self) -> list[tuple[float, dict[str, float | int | None]]]:
+    def snapshot(self) -> list[Sample]:
         return list(self._buf)
 
-    def latest(self) -> dict[str, float | int | None] | None:
+    def latest(self) -> Sample | None:
-        return self._buf[-1][1] if self._buf else None
+        return self._buf[-1] if self._buf else None
 
     def __len__(self) -> int:
         return len(self._buf)