From 752afc3f006d52a02a4c90fbb709e7708a76d070 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=EC=9C=A0=EC=B0=BD=EC=9A=B1?= Date: Sat, 25 Jul 2026 19:20:23 +0900 Subject: [PATCH] feat(edge): heartbeat telemetry + monotonic timers + queue/camera robustness - 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). --- edge/alert.py | 9 +++++- edge/app.py | 13 ++++++-- edge/daemon.py | 76 ++++++++++++++++++++++++++++++++++++-------- edge/engine.py | 22 ++++++++----- edge/health.py | 47 +++++++++++++++++++++++++++ edge/queue.py | 16 ++++++++-- edge/runtime.py | 7 ++-- tests/test_alert.py | 12 ++++++- tests/test_app.py | 24 ++++++++++++++ tests/test_daemon.py | 63 +++++++++++++++++++++++++++++++++++- tests/test_engine.py | 69 ++++++++++++++++++++++++++++++++++++++++ tests/test_health.py | 41 ++++++++++++++++++++++++ tests/test_queue.py | 24 ++++++++++++++ 13 files changed, 391 insertions(+), 32 deletions(-) create mode 100644 edge/health.py create mode 100644 tests/test_health.py diff --git a/edge/alert.py b/edge/alert.py index fb3953e..b1848df 100644 --- a/edge/alert.py +++ b/edge/alert.py @@ -74,6 +74,13 @@ class PostClient: # 4xx codes that explicitly mean try-again (408 Request Timeout, 425 Too Early, 429 Too # Many Requests). Counting these toward the poison drop cap would delete real events during # a proxy/server hiccup. Only a definite "bad request" 4xx is a permanent rejection. - if code in (408, 425, 429) or code >= 500: + # + # 401 is transient too: the server returns it for a STALE timestamp (abs(now-ts) > window), + # which is exactly what an RTC-less Pi hits after a power cut until NTP steps the clock — + # the event is valid and will be accepted verbatim once time syncs. The only permanent 401 + # cause is a wrong/rotated secret, a provisioning error to fix; retrying (events stay queued, + # bounded by max_rows) is far safer than poison-dropping a real fall alert. Poison-drop is + # reserved for a definite CONTENT rejection (400/422), never an auth/clock condition. + if code in (401, 408, 425, 429) or code >= 500: return SendResult.TRANSIENT return SendResult.REJECTED diff --git a/edge/app.py b/edge/app.py index ffcf651..aeaa9d3 100644 --- a/edge/app.py +++ b/edge/app.py @@ -29,10 +29,14 @@ def _head_boxes(persons: list) -> list: return boxes -def run_once(source, pose, tracker, pool, engine, ring, snapshotter, sink, now_fn): +def run_once(source, pose, tracker, pool, engine, ring, snapshotter, sink, now_fn, wall_fn=None): ok, frame = source.read() if not ok: - return [] + # No frame this tick. Return None (not []) so the daemon can tell "camera produced no + # frame" apart from "frame processed, no events" — a dead/unplugged USB camera returns + # (False, None) forever, and counting it as a healthy frame made a monitoring outage + # look perfectly green (fps~=target, cameras_erroring=0). + return None now = now_fn() persons = pose.infer(frame) tracks = tracker.update(persons, now) # RAW keypoints; smoothing happens per-track below @@ -42,7 +46,10 @@ def run_once(source, pose, tracker, pool, engine, ring, snapshotter, sink, now_f f.keypoints = pool.smoother_for(tr.track_id).smooth(f.keypoints, now) ring.push(now, frame) h, w = frame.shape[:2] - events = engine.process(tracks, lost, now, frame_wh=(w, h)) + # `now` (monotonic) drives detector timers; wall time stamps the event (occurred_at). Off the + # daemon (direct callers / tests) wall_fn is None → engine falls back to now, as before. + wall_now = wall_fn() if wall_fn is not None else None + events = engine.process(tracks, lost, now, frame_wh=(w, h), wall_now=wall_now) for tr in lost: pool.forget(tr.track_id) if events: diff --git a/edge/daemon.py b/edge/daemon.py index bea7708..5b928cb 100644 --- a/edge/daemon.py +++ b/edge/daemon.py @@ -17,7 +17,8 @@ def _default_log(msg, exc=None): # pragma: no cover (thin stderr wrapper) def run_loop(runtime, client, queue, *, clock, sleep, iterations=None, sync_every=150, heartbeat_every=900, flush_every=300, - frame_interval=0.0, zones_for=None, log=None): + frame_interval=0.0, zones_for=None, log=None, + heartbeat_interval_s=None, health_fn=None): # `runtime` may be a single PipelineRuntime or a list of them (multi-camera). All # cameras share this loop, one shared clock read per tick, and the periodic # sync/heartbeat/flush. A per-camera step exception is counted and rate-limit-logged, @@ -29,39 +30,72 @@ def run_loop(runtime, client, queue, *, clock, sleep, iterations=None, zones_for = lambda cfg, rt: zones_from_config(cfg) if log is None: log = _default_log + if health_fn is None: + from edge.health import read_health + health_fn = read_health counters = {"frames": 0, "syncs": 0, "beats": 0, "flushes": 0, "errors": 0} streak = {} # id(rt)/"periodic" -> consecutive-error count, for rate-limited logging - def note(key, label, exc): + def note(key, label, exc=None): counters["errors"] += 1 n = streak.get(key, 0) + 1 streak[key] = n if n == 1 or n % _LOG_EVERY == 0: - log(f"{label}: {type(exc).__name__}: {exc} (consecutive #{n})", exc) + if exc is not None: + log(f"{label}: {type(exc).__name__}: {exc} (consecutive #{n})", exc) + else: + log(f"{label} (consecutive #{n})", None) beat_anchor = None # (wall_time, frames) at the previous heartbeat, for measured fps + last_beat = None # wall time of the last heartbeat, for the time-based cadence + reported_version = None # last config version the server told us; echoed in heartbeats i = 0 while iterations is None or i < iterations: i += 1 now = clock() # one timestamp per tick, shared across cameras if beat_anchor is None: beat_anchor = (now, 0) + if last_beat is None: + last_beat = now for rt in runtimes: try: - rt.step(now) - counters["frames"] += 1 - streak.pop(id(rt), None) # recovered — reset the streak + produced = rt.step(now) except Exception as e: cam = getattr(rt, "camera_id", "?") note(id(rt), f"camera {cam} step failed", e) + continue + if produced is None: + # step() returned no frame (unplugged/browned-out camera). Count it toward the + # erroring streak instead of as a healthy frame — otherwise a dead camera ticks + # frames++ forever and reports fps~=target with cameras_erroring=0 (a silent + # monitoring outage that looks green). A single dropped frame self-clears on the + # next good frame, so only SUSTAINED starvation shows as erroring. + cam = getattr(rt, "camera_id", "?") + note(id(rt), f"camera {cam} produced no frame") + else: + counters["frames"] += 1 + streak.pop(id(rt), None) # recovered — reset the streak try: if sync_every and i % sync_every == 0: cfg = client.fetch_config() if cfg is not None: for rt in runtimes: rt.apply_zones(zones_for(cfg, rt)) + # Remember the config version so heartbeats can echo it back — the server's + # config-drift indicator is meaningless until the device reports what it runs. + if isinstance(cfg, dict) and cfg.get("version") is not None: + reported_version = int(cfg["version"]) counters["syncs"] += 1 - if heartbeat_every and i % heartbeat_every == 0: + # Heartbeat cadence: time-based when heartbeat_interval_s is set (production), so the + # interval is predictable regardless of per-tick cost (CPU MoveNet, multi-cam, thermal + # throttle can push a tick past 100ms) — an iteration count cannot guarantee a wall-clock + # cadence, and the server marks a device offline on ELAPSED time. Falls back to the + # iteration count when unset (used by the deterministic unit tests). + if heartbeat_interval_s: + do_beat = (now - last_beat) >= heartbeat_interval_s + else: + do_beat = bool(heartbeat_every) and (i % heartbeat_every == 0) + if do_beat: # Measured fps over the window since the last beat, and per-camera health, so # the server can tell a live device from a degraded/dead one (was hardcoded 0.0 # + a frames counter that ticked even when no frame was read). @@ -71,10 +105,19 @@ def run_loop(runtime, client, queue, *, clock, sleep, iterations=None, erroring = sum(1 for rt in runtimes if id(rt) in streak) metrics = {"fps": round(fps, 2), "frames": counters["frames"], "cameras_total": len(runtimes), "cameras_erroring": erroring, - "errors": counters["errors"]} + "errors": counters["errors"], "queue_pending": queue.pending()} + # Pi thermal telemetry (None off-Pi or on failure → field omitted, never a fake 0). + health = health_fn() + if health.get("temp") is not None: + metrics["temp"] = health["temp"] + if health.get("throttle") is not None: + metrics["throttle"] = health["throttle"] + if reported_version is not None: + metrics["config_version"] = reported_version if client.send_heartbeat(metrics): counters["beats"] += 1 beat_anchor = (now, counters["frames"]) + last_beat = now if flush_every and i % flush_every == 0: queue.flush() counters["flushes"] += 1 @@ -159,10 +202,16 @@ def main() -> None: # pragma: no cover (integration entrypoint, exercised on de cfg = load_config(dict(os.environ)) require_upload_config(cfg) # fail loud now, not silently upload nothing forever nonce = lambda: os.urandom(8).hex() - now = time.time + # Two clocks, deliberately: the loop/detector timers run on a MONOTONIC clock so an NTP step + # (a Pi 5 has no RTC battery; it boots on stale fake-hwclock time until chrony corrects it) + # cannot instantly satisfy immobile_s, expire every track, or reset an immobility countdown. + # Wall time is used only where a real timestamp is required: the HMAC X-Timestamp headers, + # which the server validates against its own wall clock, and Event.occurred_at. + mono = time.monotonic + wall = time.time base = cfg.cafe24_url.rstrip("/") - client = ServerClient(base, cfg.device_id, cfg.secret, nonce, now) - post = PostClient(base + "/api/events", cfg.device_id, cfg.secret, nonce, now) + client = ServerClient(base, cfg.device_id, cfg.secret, nonce, wall) + post = PostClient(base + "/api/events", cfg.device_id, cfg.secret, nonce, wall) queue = OfflineQueue(queue_db_path(dict(os.environ)), post) pose = build_pose() # shared across cameras @@ -170,12 +219,13 @@ def main() -> None: # pragma: no cover (integration entrypoint, exercised on de def make_runtime(cam, source): # Per-camera config so events are stamped with that camera's id. cam_cfg = replace(cfg, camera_id=cam, engine=replace(cfg.engine, camera_id=cam)) - return PipelineRuntime(cam_cfg, source, pose, queue) + return PipelineRuntime(cam_cfg, source, pose, queue, wall_clock=wall) # A missing/unseated camera is skipped (logged), not fatal — the others still run. runtimes = build_runtimes(parse_cameras(dict(os.environ), cfg.camera_id), build_source, make_runtime) - run_loop(runtimes, client, queue, clock=now, sleep=time.sleep, iterations=None, + run_loop(runtimes, client, queue, clock=mono, sleep=time.sleep, iterations=None, frame_interval=frame_interval_from_env(dict(os.environ)), + heartbeat_interval_s=30, # predictable cadence; server marks offline after 120s (4x) zones_for=lambda c, rt: zones_from_config(c, rt.camera_id)) diff --git a/edge/engine.py b/edge/engine.py index 5ad24be..62109bc 100644 --- a/edge/engine.py +++ b/edge/engine.py @@ -40,15 +40,21 @@ class EventEngine: self._last_emit = {k: v for k, v in self._last_emit.items() if not (len(k) >= 2 and k[1] == track_id)} - def _mk(self, etype: EventType, now: float, track_id=None, zone_id=None, + def _mk(self, etype: EventType, occurred_at: float, track_id=None, zone_id=None, metrics=None, spatial=None) -> Event: m = metrics or {} duration_s = m.get("immobility_s") or m.get("waited_s") or 0.0 return Event(etype, severity_for(etype), self.cfg.camera_id, track_id, - zone_id, now, duration_s, m, spatial) + zone_id, occurred_at, duration_s, m, spatial) def process(self, active_tracks: list[Track], lost_tracks: list[Track], - now: float, frame_wh=None) -> list[Event]: + now: float, frame_wh=None, wall_now: float | None = None) -> list[Event]: + # `now` is a MONOTONIC clock (never jumps): it drives every detector timer, hysteresis, + # and cooldown, so an NTP step cannot instantly satisfy immobile_s / expire all tracks / + # reset a genuine immobility countdown. `wall_now` is real wall-clock time, used ONLY to + # stamp Event.occurred_at (the server correlates and retains by wall time). Defaults to + # `now` so single-clock callers (unit tests) keep their existing occurred_at behavior. + occurred_at = wall_now if wall_now is not None else now events: list[Event] = [] self.ghost.on_lost(lost_tracks, now) @@ -62,13 +68,13 @@ class EventEngine: state = self.fall._states.get(tr.track_id) self.ghost.note_state(tr.track_id, state.name if state else "STANDING") if sig == "fall" and self._allow((EventType.FALL, tr.track_id), now): - events.append(self._mk(EventType.FALL, now, tr.track_id, metrics=meta)) + events.append(self._mk(EventType.FALL, occurred_at, tr.track_id, metrics=meta)) elif sig == "no_motion" and self._allow((EventType.NO_MOTION, tr.track_id), now): - events.append(self._mk(EventType.NO_MOTION, now, tr.track_id, metrics=meta)) + events.append(self._mk(EventType.NO_MOTION, occurred_at, tr.track_id, metrics=meta)) fired, imeta = self.immobility.update(tr, now, in_danger_zone=in_zone) if fired and self._allow((EventType.IMMOBILITY, tr.track_id), now): - events.append(self._mk(EventType.IMMOBILITY, now, tr.track_id, metrics=imeta)) + events.append(self._mk(EventType.IMMOBILITY, occurred_at, tr.track_id, metrics=imeta)) for hit in zone_hits: # 협착(pinch) is the highest-danger zone (machinery entrapment) and must emit @@ -78,12 +84,12 @@ class EventEngine: "restricted": EventType.ZONE_RESTRICTED}.get( hit["level"], EventType.ZONE_WARNING) if self._allow((etype, tr.track_id, hit["zone_id"]), now): - events.append(self._mk(etype, now, tr.track_id, zone_id=hit["zone_id"])) + events.append(self._mk(etype, occurred_at, tr.track_id, zone_id=hit["zone_id"])) for g in self.ghost.update(active_tracks, now): key = (EventType.GHOST, g["track_id"]) if self._allow(key, now): - events.append(self._mk(EventType.GHOST, now, track_id=g["track_id"], + events.append(self._mk(EventType.GHOST, occurred_at, track_id=g["track_id"], spatial=g["spatial"], metrics={"last_state": g["last_state"], "waited_s": g["waited_s"]})) diff --git a/edge/health.py b/edge/health.py new file mode 100644 index 0000000..2107131 --- /dev/null +++ b/edge/health.py @@ -0,0 +1,47 @@ +"""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 "" diff --git a/edge/queue.py b/edge/queue.py index a8709a4..271011d 100644 --- a/edge/queue.py +++ b/edge/queue.py @@ -64,9 +64,19 @@ class OfflineQueue: "SELECT id, payload, snaps, attempts FROM queue ORDER BY id").fetchall() sent = 0 for rid, payload_json, snaps_json, attempts in rows: - payload = json.loads(payload_json) - snaps = [base64.b64decode(s) for s in json.loads(snaps_json)] - event, snapshots = self._rehydrate(payload, snaps) + try: + payload = json.loads(payload_json) + snaps = [base64.b64decode(s) for s in json.loads(snaps_json)] + event, snapshots = self._rehydrate(payload, snaps) + except Exception: + # An undecodable row — a torn SD-card write, or schema drift from an older edge + # build whose payload lacks a key _rehydrate indexes — can NEVER be delivered. + # flush() breaks on the first undelivered row to preserve order, so leaving this + # row in place would wedge every queued safety event BEHIND it forever. Drop the + # single unrecoverable row and continue; losing it beats freezing the whole queue. + self.db.execute("DELETE FROM queue WHERE id = ?", (rid,)) + self.db.commit() + continue # A row queued before event_uuid existed has none stored; _rehydrate minted a fresh # one just now. Persist it so every retry of this row reuses the SAME key — otherwise # each flush sends a different uuid and a timeout-retried delivery duplicates on the diff --git a/edge/runtime.py b/edge/runtime.py index fc0d1bd..70d9a57 100644 --- a/edge/runtime.py +++ b/edge/runtime.py @@ -10,12 +10,15 @@ from edge.snapshot import RingBuffer, FaceBlurrer, Snapshotter class PipelineRuntime: - def __init__(self, config, source, pose, sink): + def __init__(self, config, source, pose, sink, wall_clock=None): self.config = config self.camera_id = config.camera_id # for per-camera zone routing (multi-cam) self.source = source self.pose = pose self.sink = sink + # Wall clock for Event.occurred_at only; the loop clock (step's `now`) is monotonic. None + # off-daemon -> run_once stamps occurred_at with the monotonic `now`, preserving old tests. + self._wall = wall_clock self.tracker = IouTracker() self.pool = TrackSmootherPool() self.zone_detector = ZoneIntrusionDetector([], config.zone) @@ -32,7 +35,7 @@ class PipelineRuntime: def step(self, now: float) -> list: return run_once(self.source, self.pose, self.tracker, self.pool, self.engine, self.ring, self.snapshotter, self.sink, - now_fn=lambda: now) + now_fn=lambda: now, wall_fn=self._wall) def apply_zones(self, zones: list) -> None: self.zone_detector.zones = list(zones) diff --git a/tests/test_alert.py b/tests/test_alert.py index df623dd..75aa34f 100644 --- a/tests/test_alert.py +++ b/tests/test_alert.py @@ -49,11 +49,21 @@ def test_send_classifies_5xx_as_transient(): client = PostClient("https://x", "pi-01", "s", lambda: "n", lambda: 1.0, http_post=lambda *a, **k: _FakeResp(500)) assert client.send(_event(), []) is SendResult.TRANSIENT - for code in (408, 425, 429): # try-again 4xx codes are transient, not permanent rejections + for code in (401, 408, 425, 429): # try-again codes + 401 (stale clock) are transient c = PostClient("https://x", "pi-01", "s", lambda: "n", lambda: 1.0, http_post=lambda *a, _c=code, **k: _FakeResp(_c)) assert c.send(_event(), []) is SendResult.TRANSIENT + +def test_send_treats_401_stale_timestamp_as_transient_not_poison(): + # The server returns 401 for a STALE timestamp (RTC-less Pi after a power cut, before NTP). + # That is transient — the event is valid and accepted once the clock syncs — so it must NOT + # count toward the poison drop cap, or a real fall alert captured during clock skew is deleted + # after ~20 retries instead of delivered. + client = PostClient("https://x", "pi-01", "s", lambda: "n", lambda: 1.0, + http_post=lambda *a, **k: _FakeResp(401)) + assert client.send(_event(), []) is SendResult.TRANSIENT + def test_send_classifies_4xx_as_rejected(): # A definite 4xx (not 429) means the server will never accept this event — the queue counts it # toward the poison drop cap so it can eventually be discarded and unblock the queue. diff --git a/tests/test_app.py b/tests/test_app.py index 8f9d855..9ac8e85 100644 --- a/tests/test_app.py +++ b/tests/test_app.py @@ -48,6 +48,30 @@ def test_run_once_pipeline_emits_fall_after_prone(): assert any(e.event_type == EventType.FALL for e in sink.events) +def test_run_once_stamps_events_with_wall_clock_not_monotonic(): + # run_once must thread a separate wall clock through to occurred_at: the daemon runs the loop on + # a monotonic clock (NTP-step-immune timers) but events have to carry real wall time. + src, pose = _Source(), _Pose() + tracker = IouTracker() + pool = TrackSmootherPool() + engine = EventEngine( + FallDetector(FallConfig(prone_confirm_s=0.5)), + ImmobilityWatchdog(ImmobilityConfig(immobile_s=999)), + GhostIncidentDetector(GhostConfig()), + ZoneIntrusionDetector([], ZoneConfig()), + EngineConfig(camera_id="cam-01"), + ) + ring = RingBuffer(); snap = Snapshotter(FaceBlurrer()); sink = _Sink() + mono = iter([0.0, 0.6]) # monotonic loop clock (drives prone_confirm_s) + wallt = iter([1_700_000_000.0, 1_700_000_000.6]) # wall clock (stamps occurred_at) + run_once(src, pose, tracker, pool, engine, ring, snap, sink, + now_fn=lambda: next(mono), wall_fn=lambda: next(wallt)) + run_once(src, pose, tracker, pool, engine, ring, snap, sink, + now_fn=lambda: next(mono), wall_fn=lambda: next(wallt)) + falls = [e for e in sink.events if e.event_type == EventType.FALL] + assert falls and falls[0].occurred_at == 1_700_000_000.6 + + def _person_at(cx): # standing person centered at cx; nose at (cx, 60) kps = [Keypoint(cx, 150, 0.9) for _ in range(17)] diff --git a/tests/test_daemon.py b/tests/test_daemon.py index c89dc99..2a18f57 100644 --- a/tests/test_daemon.py +++ b/tests/test_daemon.py @@ -16,8 +16,9 @@ class _Client: class _Queue: - def __init__(self): self.flushes = 0 + def __init__(self, pending=0): self.flushes = 0; self._pending = pending def flush(self): self.flushes += 1; return 0 + def pending(self): return self._pending def test_loop_steps_and_periodics(): @@ -98,6 +99,66 @@ def test_heartbeat_flags_a_failing_camera(): assert m["cameras_erroring"] == 1 # the server can now see the camera is down +class _NoFrameRuntime: + # step() returns None on a tick with no frame (dead/unplugged camera), a list otherwise. + camera_id = "cam-dead" + def __init__(self): self.steps = 0 + def step(self, now): self.steps += 1; return None + + +def test_no_frame_counts_as_erroring_not_a_healthy_frame(): + # A camera that returns no frame every tick (unplugged USB) must NOT tick frames++ and must + # surface as erroring — otherwise a monitoring outage looks perfectly green (the whole point + # of run_once returning None instead of []). + rt, cl = _NoFrameRuntime(), _Client() + ticks = iter(float(t) for t in range(1000)) + counters = run_loop(rt, cl, _Queue(), clock=lambda: next(ticks), sleep=lambda s: None, + iterations=10, sync_every=99, heartbeat_every=10, flush_every=99, + log=lambda *a, **k: None) + assert rt.steps == 10 + assert counters["frames"] == 0 # no frame was ever counted as healthy + m = cl.metrics[0] + assert m["fps"] == 0.0 + assert m["cameras_erroring"] == 1 # the dead camera is visible to the server + + +def test_heartbeat_reports_queue_depth_and_echoes_config_version(): + class _VersionClient(_Client): + def fetch_config(self): + self.configs += 1 + return {"version": 7, "zones": [], "thresholds": {}} + rt, cl, q = _Runtime(), _VersionClient(), _Queue(pending=4) + ticks = iter(float(t) for t in range(1000)) + # sync at frame 5 (captures version 7) BEFORE the heartbeat at frame 10. + run_loop(rt, cl, q, clock=lambda: next(ticks), sleep=lambda s: None, + iterations=10, sync_every=5, heartbeat_every=10, flush_every=99) + m = cl.metrics[0] + assert m["queue_pending"] == 4 # transmitted so the dashboard can show backlog + assert m["config_version"] == 7 # echoed so config-drift detection actually works + + +def test_heartbeat_omits_config_version_until_a_config_is_fetched(): + # Before any sync, the device does not know its version — it must OMIT config_version rather + # than send a fake 0 (which made the server's drift indicator fire on every device forever). + rt, cl, q = _Runtime(), _Client(), _Queue() + ticks = iter(float(t) for t in range(1000)) + run_loop(rt, cl, q, clock=lambda: next(ticks), sleep=lambda s: None, + iterations=10, sync_every=99, heartbeat_every=10, flush_every=99) + assert "config_version" not in cl.metrics[0] + + +def test_time_based_heartbeat_fires_on_elapsed_wall_time_not_iteration_count(): + # Production uses heartbeat_interval_s so the cadence is predictable no matter how slow a tick + # is. With the clock advancing 10s/tick and a 30s interval, a beat fires every ~3 ticks. + rt, cl, q = _Runtime(), _Client(), _Queue() + ticks = iter(float(t * 10) for t in range(1000)) # t = 0,10,20,30,... + run_loop(rt, cl, q, clock=lambda: next(ticks), sleep=lambda s: None, + iterations=10, sync_every=99, heartbeat_every=None, flush_every=99, + heartbeat_interval_s=30) + # ticks at 0,10,20,30,...,90. Beats when now-last_beat >= 30: at 30, 60, 90 -> 3 beats. + assert cl.beats == 3 + + def test_loop_survives_flush_exception(): class BoomQueue(_Queue): def flush(self): diff --git a/tests/test_engine.py b/tests/test_engine.py index 6ac1b7d..6e0593a 100644 --- a/tests/test_engine.py +++ b/tests/test_engine.py @@ -19,6 +19,17 @@ def _lying_track(track, t): track.frames.append(TrackFrame(t, BBox(90, 290, 210, 320), _lying_kps())) return track +def _standing_kps(cx=150): + kps = [Keypoint(cx, 150, 0.9) for _ in range(17)] + kps[5] = Keypoint(cx - 10, 100, 0.9); kps[6] = Keypoint(cx + 10, 100, 0.9) # shoulders above + kps[11] = Keypoint(cx - 10, 200, 0.9); kps[12] = Keypoint(cx + 10, 200, 0.9) # hips below + kps[0] = Keypoint(cx, 60, 0.9) # head above hips + return kps + +def _standing_track(track, t, cx=150): + track.frames.append(TrackFrame(t, BBox(cx - 20, 60, cx + 20, 210), _standing_kps(cx))) + return track + def _engine(): return EventEngine( FallDetector(FallConfig(prone_confirm_s=0.5)), @@ -106,6 +117,64 @@ def test_pinch_zone_emits_pinch_event_not_warning(): assert not any(e.event_type == EventType.ZONE_WARNING for e in events) +def test_two_simultaneous_tracks_have_independent_fall_state(): + # Per-track detector state must not leak between two people monitored at once: a fall on track A + # must arm and emit for A only, never for a standing track B. All other engine/detector tests + # exercise a single person, so this pins the multi-person isolation the fall/immobility/zone + # timers rely on (they key every bit of state off track_id). + eng = _engine() # prone_confirm_s=0.5 + a = Track(1, deque(maxlen=90)); b = Track(2, deque(maxlen=90)) + _lying_track(a, 0.0); _standing_track(b, 0.0, cx=500) + eng.process([a, b], [], now=0.0) + _lying_track(a, 0.6); _standing_track(b, 0.6, cx=500) + events = eng.process([a, b], [], now=0.6) + falls = [e for e in events if e.event_type == EventType.FALL] + assert len(falls) == 1 # exactly one fall... + assert falls[0].track_id == 1 # ...for the lying track A, not the standing track B + + +def test_fall_fires_at_exactly_prone_confirm_s_inclusive_boundary(): + # The confirm is `now - prone_since >= prone_confirm_s`, so a fall must fire at EXACTLY the + # threshold, not one frame later. Pins the inclusive boundary against a >= -> > regression that + # would silently delay every fall confirmation by a frame — up to +0.5s at a thermally throttled + # 2 fps, on the single most safety-critical event. Existing tests only feed comfortably-past + # values (0.6 vs 0.5), so none of them would catch that shift. + eng = _engine() # prone_confirm_s=0.5 + tr = Track(1, deque(maxlen=90)) + _lying_track(tr, 0.0); eng.process([tr], [], now=0.0) # prone_since = 0.0 + _lying_track(tr, 0.5) # exactly prone_confirm_s elapsed + events = eng.process([tr], [], now=0.5) + assert any(e.event_type == EventType.FALL for e in events) + + +def test_occurred_at_uses_wall_clock_while_timers_use_monotonic(): + # A fall is CONFIRMED by the monotonic timer (prone_confirm_s over the injected `now`), but the + # event must be STAMPED with wall time so an NTP step does not mis-timestamp evidence (and the + # server's wall-time retention purge does not delete it). Timer `now` and `wall_now` are + # independent axes here to prove occurred_at follows wall time, not the monotonic clock. + eng = _engine() # prone_confirm_s=0.5 + tr = Track(1, deque(maxlen=90)) + _lying_track(tr, 0.0) + eng.process([tr], [], now=0.0, wall_now=1_700_000_000.0) + _lying_track(tr, 0.6) + events = eng.process([tr], [], now=0.6, wall_now=1_700_000_000.6) + fall = [e for e in events if e.event_type == EventType.FALL] + assert len(fall) == 1 + assert fall[0].occurred_at == 1_700_000_000.6 # wall time, NOT the monotonic 0.6 + + +def test_process_defaults_occurred_at_to_now_when_no_wall_clock(): + # Single-clock callers (unit tests, direct callers) omit wall_now — occurred_at must fall back + # to the timer `now` so existing behavior is unchanged. + eng = _engine() + tr = Track(1, deque(maxlen=90)) + _lying_track(tr, 0.0); eng.process([tr], [], now=0.0) + _lying_track(tr, 0.6) + events = eng.process([tr], [], now=0.6) + fall = [e for e in events if e.event_type == EventType.FALL] + assert len(fall) == 1 and fall[0].occurred_at == 0.6 + + def test_allow_cooldown_primitive(): eng = _engine() # EngineConfig(camera_id="cam-01", cooldown_s=30.0) key = ("FALL", 1) diff --git a/tests/test_health.py b/tests/test_health.py new file mode 100644 index 0000000..842e836 --- /dev/null +++ b/tests/test_health.py @@ -0,0 +1,41 @@ +from edge.health import parse_temp, parse_throttle, read_health + + +def test_parse_temp_reads_vcgencmd_output(): + assert parse_temp("temp=54.3'C\n") == 54.3 + + +def test_parse_temp_returns_none_on_garbage(): + assert parse_temp("") is None + assert parse_temp("no equals sign") is None + + +def test_parse_throttle_true_when_a_currently_active_bit_is_set(): + assert parse_throttle("throttled=0x4") is True # bit2: currently throttled + assert parse_throttle("throttled=0x1") is True # bit0: under-voltage now + + +def test_parse_throttle_false_for_zero_and_for_past_only_latches(): + assert parse_throttle("throttled=0x0") is False + # 0x50000 = bits 16,18 (under-voltage/throttle HAPPENED since boot) but nothing active now. + assert parse_throttle("throttled=0x50000") is False + + +def test_parse_throttle_none_on_garbage(): + assert parse_throttle("throttled=notahexnum") is None + + +def test_read_health_returns_nones_off_pi_without_touching_subprocess(): + # No vcgencmd on the build machine / server: read_health must return Nones (never a fake 0) + # and must NOT invoke the runner at all. + calls = [] + out = read_health(run=lambda cmd: calls.append(cmd) or "", which=lambda name: None) + assert out == {"temp": None, "throttle": None} + assert calls == [] + + +def test_read_health_parses_injected_vcgencmd_on_pi(): + def fake_run(cmd): + return "temp=61.7'C\n" if "measure_temp" in cmd else "throttled=0x0\n" + out = read_health(run=fake_run, which=lambda name: "/usr/bin/vcgencmd") + assert out == {"temp": 61.7, "throttle": False} diff --git a/tests/test_queue.py b/tests/test_queue.py index fc375ba..86db1ed 100644 --- a/tests/test_queue.py +++ b/tests/test_queue.py @@ -120,6 +120,30 @@ def test_sustained_outage_never_drops_events(tmp_path): assert q.pending() == 0 and _Outage.sent == [1] # drains once connectivity returns +def test_undecodable_row_is_dropped_and_unblocks_the_queue(tmp_path): + # A corrupt row (torn SD-card write / schema drift) can never be decoded, so it must not wedge + # every queued safety event behind it forever. flush() drops the single bad row and drains the + # rest — losing one unrecoverable event beats freezing the whole queue. + db = str(tmp_path / "q.db") + con = sqlite3.connect(db) + con.execute("CREATE TABLE queue (id INTEGER PRIMARY KEY AUTOINCREMENT, payload TEXT, " + "snaps TEXT, attempts INTEGER DEFAULT 0)") + con.execute("INSERT INTO queue (payload, snaps) VALUES (?, ?)", + ("{not valid json", json.dumps([]))) # poison row at the head + good = {"event_type": "fall", "severity": "high", "camera_id": "cam-01", "track_id": 7, + "zone_id": None, "occurred_at": 100.0, "duration_s": 3.0, "metrics": {}, + "spatial": None, "event_uuid": "u7"} + con.execute("INSERT INTO queue (payload, snaps) VALUES (?, ?)", + (json.dumps(good), json.dumps([]))) # good row behind it + con.commit(); con.close() + + client = _Client() + q = OfflineQueue(db, client) + assert q.flush() == 1 # the good row drained past the dropped poison row + assert client.sent == [7] + assert q.pending() == 0 # both rows gone: poison dropped, good delivered + + def test_legacy_row_without_uuid_reuses_minted_uuid_across_retries(tmp_path): # A row queued before event_uuid existed has none stored. On first flush a fresh uuid is # minted; it MUST be persisted so every retry reuses the SAME key — else each failed-then-