138 lines
3.7 KiB
Python
138 lines
3.7 KiB
Python
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
|