Add Raspberry Pi health monitoring

Co-authored-by: Copilot <[email protected]>

modules/app_events.py

@@ -1,44 +1,10 @@
 from flask_socketio import SocketIO
-from threading import Thread
 from modules import app_obd
-import time, threading
 
 socketio = SocketIO()
-uptime_lock = threading.Lock()
-thread_uptime = None
 
 @socketio.on('connect')
 def connect_event():
     print('Client connected')
     print('Getting current OBD status...')
     app_obd.obd_connection()
-    uptime()
-
-def uptime():
-    global thread_uptime 
-
-    if thread_uptime is None or not thread_uptime.is_alive():
-        thread_uptime = Thread(target=loop_uptime)
-        thread_uptime.daemon = True
-        thread_uptime.start()
-        print(f"Uptime thread started, alive: {thread_uptime.is_alive()}")
-    else:
-        print(f"Uptime thread already running, alive: {thread_uptime.is_alive()}")
-
-def loop_uptime():
-    uptime_lock.acquire()
-    try:
-        loop = True
-        while loop == True:
-            boottime = time.clock_gettime(time.CLOCK_BOOTTIME)
-            # Convert to hours, minutes, and seconds
-            seconds = int(boottime)
-            minutes, seconds = divmod(seconds, 60)
-            hours, minutes = divmod(minutes, 60)
-            time.sleep(1)
-            uptime = f"{hours:02d}:{minutes:02d}:{seconds:02d}"   
-            socketio.emit("rpi_uptime", uptime)
-    finally:
-        uptime_lock.release()
-
-    

modules/app_health.py

@@ -0,0 +1,137 @@
+import os
+import shutil
+import subprocess
+
+
+_BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+THERMAL_PATH = "/sys/class/thermal/thermal_zone0/temp"
+MEMINFO_PATH = "/proc/meminfo"
+UPTIME_PATH = "/proc/uptime"
+
+THROTTLE_FLAGS = {
+    0: "under_voltage",
+    1: "frequency_capped",
+    2: "throttled",
+    3: "soft_temperature_limit",
+}
+
+
+def _read_text(path):
+    with open(path, encoding="ascii") as source:
+        return source.read().strip()
+
+
+def parse_meminfo(text):
+    values = {}
+    for line in text.splitlines():
+        name, separator, remainder = line.partition(":")
+        if separator and name in {"MemTotal", "MemAvailable"}:
+            values[name] = int(remainder.strip().split()[0]) * 1024
+    if set(values) != {"MemTotal", "MemAvailable"}:
+        raise ValueError("MemTotal or MemAvailable missing from /proc/meminfo")
+    total = values["MemTotal"]
+    available = values["MemAvailable"]
+    return {
+        "total_bytes": total,
+        "used_bytes": total - available,
+        "available_bytes": available,
+        "used_percent": round((total - available) / total * 100, 1),
+    }
+
+
+def parse_throttled(value):
+    prefix, separator, raw_value = value.strip().partition("=")
+    if prefix != "throttled" or not separator:
+        raise ValueError(f"Unexpected vcgencmd response: {value.strip()}")
+    bits = int(raw_value, 16)
+    current = [
+        label for bit, label in THROTTLE_FLAGS.items() if bits & (1 << bit)
+    ]
+    occurred = [
+        label for bit, label in THROTTLE_FLAGS.items() if bits & (1 << (bit + 16))
+    ]
+    return {
+        "raw": f"0x{bits:x}",
+        "current": current,
+        "occurred": occurred,
+    }
+
+
+def _temperature():
+    return round(int(_read_text(THERMAL_PATH)) / 1000, 1)
+
+
+def _memory():
+    return parse_meminfo(_read_text(MEMINFO_PATH))
+
+
+def _uptime():
+    return int(float(_read_text(UPTIME_PATH).split()[0]))
+
+
+def _disk():
+    usage = shutil.disk_usage(_BASE_DIR)
+    return {
+        "total_bytes": usage.total,
+        "used_bytes": usage.used,
+        "free_bytes": usage.free,
+        "used_percent": round(usage.used / (usage.used + usage.free) * 100, 1),
+    }
+
+
+def _throttling():
+    result = subprocess.run(
+        ["vcgencmd", "get_throttled"],
+        check=True,
+        capture_output=True,
+        text=True,
+        timeout=2,
+    )
+    return parse_throttled(result.stdout)
+
+
+def collect_health():
+    readers = {
+        "temperature_c": _temperature,
+        "memory": _memory,
+        "disk": _disk,
+        "uptime_seconds": _uptime,
+        "throttling": _throttling,
+    }
+    health = {}
+    errors = {}
+    for name, reader in readers.items():
+        try:
+            health[name] = reader()
+        except (OSError, ValueError, subprocess.SubprocessError) as error:
+            health[name] = None
+            errors[name] = str(error)
+
+    warnings = []
+    historical_warnings = []
+    throttling = health.get("throttling")
+    if throttling:
+        warnings.extend(throttling["current"])
+        historical_warnings.extend(
+            warning
+            for warning in throttling["occurred"]
+            if warning not in throttling["current"]
+        )
+    if health.get("temperature_c") is not None and health["temperature_c"] >= 80:
+        warnings.append("high_temperature")
+    if health.get("disk") and health["disk"]["used_percent"] >= 90:
+        warnings.append("disk_nearly_full")
+    if health.get("memory") and health["memory"]["used_percent"] >= 90:
+        warnings.append("memory_pressure")
+
+    health["status"] = (
+        "warning"
+        if warnings
+        else "degraded"
+        if errors
+        else "ok"
+    )
+    health["warnings"] = warnings
+    health["historical_warnings"] = historical_warnings
+    health["errors"] = errors
+    return health