bambuddy/spoolbuddy/daemon/system_stats.py

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