752afc3f00
- Split clocks: detector timers/cooldowns run on a MONOTONIC clock (immune to an NTP step on an RTC-less Pi); wall time is used only for the HMAC X-Timestamp and Event.occurred_at. Fixes false IMMOBILITY / dropped tracks after a clock jump. - Heartbeat is time-based (30s, predictable regardless of load) and now carries temp/throttle/queue_pending/config_version telemetry. - A dead (no-frame) camera counts as erroring instead of a healthy frame, so a silent monitoring outage no longer looks green. - Offline queue: isolate one undecodable row instead of wedging every event behind it; classify 401 (stale clock) as transient, never a poison-drop. - New edge/health.py: vcgencmd temp/throttle parsing, None off-Pi (no fake 0).
47 lines
1.9 KiB
Python
47 lines
1.9 KiB
Python
"""Pi thermal telemetry for the heartbeat.
|
|
|
|
Reports CPU temperature (C) and whether the SoC is currently throttling — the #1 cause of
|
|
Hailo inference falling off a cliff in a hot summer factory. Off-Pi (no `vcgencmd`) or on any
|
|
failure every field is None so the caller OMITS it: the dashboard must never show a fabricated
|
|
temp=0 / throttle=false that hides an overheating device.
|
|
"""
|
|
import shutil
|
|
import subprocess
|
|
|
|
|
|
def parse_temp(out: str):
|
|
"""`vcgencmd measure_temp` -> "temp=54.3'C" -> 54.3. None if unparseable."""
|
|
try:
|
|
return float(out.split("=", 1)[1].split("'", 1)[0])
|
|
except (IndexError, ValueError):
|
|
return None
|
|
|
|
|
|
def parse_throttle(out: str):
|
|
"""`vcgencmd get_throttled` -> "throttled=0x50005". True when any CURRENTLY-active bit is
|
|
set: bit0 under-voltage now, bit1 ARM freq capped now, bit2 currently throttled, bit3 soft
|
|
temp limit now. The 0x1_0000+ bits are "happened since boot" latches, deliberately ignored
|
|
so a past blip does not show as a permanent warning. None if unparseable."""
|
|
try:
|
|
val = int(out.strip().split("=", 1)[1], 16)
|
|
return bool(val & 0xF)
|
|
except (IndexError, ValueError):
|
|
return None
|
|
|
|
|
|
def read_health(run=None, which=None) -> dict:
|
|
"""{'temp': float|None, 'throttle': bool|None}. `run`/`which` are injectable for tests;
|
|
default to subprocess/shutil. Returns Nones immediately when vcgencmd is absent (off-Pi)."""
|
|
which = which or shutil.which
|
|
run = run or _run
|
|
if which("vcgencmd") is None:
|
|
return {"temp": None, "throttle": None}
|
|
return {"temp": parse_temp(run(["vcgencmd", "measure_temp"])),
|
|
"throttle": parse_throttle(run(["vcgencmd", "get_throttled"]))}
|
|
|
|
|
|
def _run(cmd: list) -> str: # pragma: no cover (thin subprocess wrapper, exercised on device)
|
|
try:
|
|
return subprocess.run(cmd, capture_output=True, text=True, timeout=2).stdout
|
|
except Exception:
|
|
return ""
|