mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-08-11 00:30:12 -04:00
77cb7158d2
The daemon now collects CPU temp, core count, load average, memory/disk usage, OS info, and system uptime every heartbeat using stdlib-only reads from /proc and /sys. Stats are sent as a JSON blob in the heartbeat payload, stored in a new system_stats TEXT column, and displayed in a new "System" tab in SpoolBuddy Settings with color-coded usage bars.
137 lines
3.3 KiB
Python
137 lines
3.3 KiB
Python
"""Collect OS-level system stats from the Raspberry Pi using stdlib only."""
|
|
|
|
import os
|
|
import platform
|
|
|
|
|
|
def _read_file(path: str) -> str | None:
|
|
try:
|
|
with open(path) as f:
|
|
return f.read().strip()
|
|
except OSError:
|
|
return None
|
|
|
|
|
|
def _cpu_temp() -> float | None:
|
|
raw = _read_file("/sys/class/thermal/thermal_zone0/temp")
|
|
if raw is None:
|
|
return None
|
|
try:
|
|
return round(int(raw) / 1000, 1)
|
|
except (ValueError, TypeError):
|
|
return None
|
|
|
|
|
|
def _memory_info() -> dict | None:
|
|
raw = _read_file("/proc/meminfo")
|
|
if raw is None:
|
|
return None
|
|
info: dict[str, int] = {}
|
|
for line in raw.splitlines():
|
|
parts = line.split()
|
|
if len(parts) >= 2 and parts[0].endswith(":"):
|
|
key = parts[0][:-1]
|
|
try:
|
|
info[key] = int(parts[1]) # kB
|
|
except ValueError:
|
|
continue
|
|
total = info.get("MemTotal", 0)
|
|
available = info.get("MemAvailable", 0)
|
|
if total == 0:
|
|
return None
|
|
return {
|
|
"total_mb": round(total / 1024),
|
|
"available_mb": round(available / 1024),
|
|
"used_mb": round((total - available) / 1024),
|
|
"percent": round((total - available) / total * 100, 1),
|
|
}
|
|
|
|
|
|
def _disk_info() -> dict | None:
|
|
try:
|
|
st = os.statvfs("/")
|
|
except OSError:
|
|
return None
|
|
total = st.f_frsize * st.f_blocks
|
|
free = st.f_frsize * st.f_bavail
|
|
used = total - free
|
|
if total == 0:
|
|
return None
|
|
return {
|
|
"total_gb": round(total / (1024**3), 1),
|
|
"used_gb": round(used / (1024**3), 1),
|
|
"free_gb": round(free / (1024**3), 1),
|
|
"percent": round(used / total * 100, 1),
|
|
}
|
|
|
|
|
|
def _load_avg() -> list[float] | None:
|
|
try:
|
|
load = os.getloadavg()
|
|
return [round(x, 2) for x in load]
|
|
except OSError:
|
|
return None
|
|
|
|
|
|
def _cpu_count() -> int | None:
|
|
return os.cpu_count()
|
|
|
|
|
|
def _os_info() -> dict:
|
|
uname = platform.uname()
|
|
os_release = _read_file("/etc/os-release")
|
|
pretty_name = None
|
|
if os_release:
|
|
for line in os_release.splitlines():
|
|
if line.startswith("PRETTY_NAME="):
|
|
pretty_name = line.split("=", 1)[1].strip().strip('"')
|
|
break
|
|
return {
|
|
"os": pretty_name or f"{uname.system} {uname.release}",
|
|
"kernel": uname.release,
|
|
"arch": uname.machine,
|
|
"python": platform.python_version(),
|
|
}
|
|
|
|
|
|
def _system_uptime() -> int | None:
|
|
raw = _read_file("/proc/uptime")
|
|
if raw is None:
|
|
return None
|
|
try:
|
|
return int(float(raw.split()[0]))
|
|
except (ValueError, IndexError):
|
|
return None
|
|
|
|
|
|
def collect() -> dict:
|
|
"""Collect all system stats. Returns a flat dict safe for JSON serialization."""
|
|
stats: dict = {}
|
|
|
|
stats["os"] = _os_info()
|
|
|
|
temp = _cpu_temp()
|
|
if temp is not None:
|
|
stats["cpu_temp_c"] = temp
|
|
|
|
cpu_count = _cpu_count()
|
|
if cpu_count is not None:
|
|
stats["cpu_count"] = cpu_count
|
|
|
|
load = _load_avg()
|
|
if load is not None:
|
|
stats["load_avg"] = load
|
|
|
|
mem = _memory_info()
|
|
if mem is not None:
|
|
stats["memory"] = mem
|
|
|
|
disk = _disk_info()
|
|
if disk is not None:
|
|
stats["disk"] = disk
|
|
|
|
uptime = _system_uptime()
|
|
if uptime is not None:
|
|
stats["system_uptime_s"] = uptime
|
|
|
|
return stats
|