Add SpoolBuddy System tab with live OS stats from Raspberry Pi

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.
This commit is contained in:
maziggy
2026-03-26 09:57:09 +01:00
parent 1df5130df2
commit 77cb7158d2
14 changed files with 974 additions and 427 deletions
+3
View File
@@ -81,6 +81,7 @@ def _device_to_response(device: SpoolBuddyDevice) -> DeviceResponse:
uptime_s=device.uptime_s,
update_status=device.update_status,
update_message=device.update_message,
system_stats=json.loads(device.system_stats) if device.system_stats else None,
online=_is_online(device),
created_at=device.created_at,
updated_at=device.updated_at,
@@ -216,6 +217,8 @@ async def device_heartbeat(
device.nfc_connection = req.nfc_connection
if req.backend_url:
device.backend_url = req.backend_url
if req.system_stats is not None:
device.system_stats = json.dumps(req.system_stats)
# Return and clear pending command
pending = device.pending_command
+6
View File
@@ -1424,6 +1424,12 @@ async def run_migrations(conn):
except OperationalError:
pass # Already applied
# Migration: Add system_stats JSON blob column to spoolbuddy_devices
try:
await conn.execute(text("ALTER TABLE spoolbuddy_devices ADD COLUMN system_stats TEXT"))
except OperationalError:
pass # Already applied
# Migration: Convert ams_labels table from (printer_id, ams_id) key to ams_serial_number key
# Labels are now keyed by AMS serial number so they persist when the AMS is moved to another printer.
try:
+1
View File
@@ -36,5 +36,6 @@ class SpoolBuddyDevice(Base):
nfc_ok: Mapped[bool] = mapped_column(Boolean, default=False)
scale_ok: Mapped[bool] = mapped_column(Boolean, default=False)
uptime_s: Mapped[int] = mapped_column(Integer, default=0)
system_stats: Mapped[str | None] = mapped_column(Text, nullable=True)
created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())
updated_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now(), onupdate=func.now())
+2
View File
@@ -44,6 +44,7 @@ class DeviceResponse(BaseModel):
uptime_s: int
update_status: str | None = None
update_message: str | None = None
system_stats: dict | None = None
online: bool = False
ssh_public_key: str | None = None
created_at: datetime
@@ -62,6 +63,7 @@ class HeartbeatRequest(BaseModel):
nfc_reader_type: str | None = None
nfc_connection: str | None = None
backend_url: str | None = None
system_stats: dict | None = None
class HeartbeatResponse(BaseModel):
@@ -0,0 +1,118 @@
"""Tests for SpoolBuddy daemon system_stats collector."""
from unittest.mock import patch
from spoolbuddy.daemon.system_stats import (
_cpu_count,
_cpu_temp,
_disk_info,
_load_avg,
_memory_info,
_os_info,
_system_uptime,
collect,
)
class TestCpuTemp:
def test_reads_thermal_zone(self):
with patch("spoolbuddy.daemon.system_stats._read_file", return_value="52100"):
assert _cpu_temp() == 52.1
def test_returns_none_on_missing_file(self):
with patch("spoolbuddy.daemon.system_stats._read_file", return_value=None):
assert _cpu_temp() is None
def test_returns_none_on_bad_value(self):
with patch("spoolbuddy.daemon.system_stats._read_file", return_value="not_a_number"):
assert _cpu_temp() is None
class TestMemoryInfo:
SAMPLE_MEMINFO = (
"MemTotal: 1024000 kB\n"
"MemFree: 200000 kB\n"
"MemAvailable: 512000 kB\n"
"Buffers: 50000 kB\n"
)
def test_parses_meminfo(self):
with patch("spoolbuddy.daemon.system_stats._read_file", return_value=self.SAMPLE_MEMINFO):
result = _memory_info()
assert result is not None
assert result["total_mb"] == 1000
assert result["available_mb"] == 500
assert result["used_mb"] == 500
assert result["percent"] == 50.0
def test_returns_none_on_missing(self):
with patch("spoolbuddy.daemon.system_stats._read_file", return_value=None):
assert _memory_info() is None
class TestDiskInfo:
def test_returns_disk_stats(self):
result = _disk_info()
# Should always work on Linux
assert result is not None
assert "total_gb" in result
assert "used_gb" in result
assert "free_gb" in result
assert "percent" in result
assert 0 <= result["percent"] <= 100
class TestLoadAvg:
def test_returns_three_values(self):
result = _load_avg()
assert result is not None
assert len(result) == 3
for val in result:
assert isinstance(val, float)
class TestCpuCount:
def test_returns_positive_int(self):
result = _cpu_count()
assert result is not None
assert result > 0
class TestOsInfo:
def test_returns_required_keys(self):
result = _os_info()
assert "os" in result
assert "kernel" in result
assert "arch" in result
assert "python" in result
def test_parses_pretty_name(self):
fake_release = 'PRETTY_NAME="Raspbian GNU/Linux 12 (bookworm)"\nID=raspbian\n'
with patch("spoolbuddy.daemon.system_stats._read_file", return_value=fake_release):
result = _os_info()
assert result["os"] == "Raspbian GNU/Linux 12 (bookworm)"
class TestSystemUptime:
def test_parses_uptime(self):
with patch("spoolbuddy.daemon.system_stats._read_file", return_value="86400.55 172000.10"):
assert _system_uptime() == 86400
def test_returns_none_on_missing(self):
with patch("spoolbuddy.daemon.system_stats._read_file", return_value=None):
assert _system_uptime() is None
class TestCollect:
def test_returns_dict_with_expected_keys(self):
result = collect()
assert isinstance(result, dict)
assert "os" in result
# These may or may not be present depending on platform, but os is always present
def test_all_values_are_json_serializable(self):
import json
result = collect()
# Should not raise
json.dumps(result)