diff --git a/docs/delivery-audit.md b/docs/delivery-audit.md index 4ccbb9c..00e4c30 100644 --- a/docs/delivery-audit.md +++ b/docs/delivery-audit.md @@ -99,14 +99,22 @@ misconfig/degradation fail LOUD. arms a CRITICAL ghost if the person then leaves. **Fix:** require sustained prone. - [ ] **P2-8 [M]** `edge/app.py:30` — detection dropouts keep a track 'active' with a frozen frame; time-based timers accrue on stale data. **Fix:** treat a stale track as no-data. -- [ ] **P2-9 [M]** `edge/detectors/zone.py:45` — foot point averages both ankles with no +- [x] **P2-9 [M]** `edge/detectors/zone.py:45` — foot point averages both ankles with no confidence check; a hallucinated ankle at (0,0) drags the foot into/out of a zone. **Fix:** confidence-gate the ankles, fall back to bbox-bottom-center. -- [ ] **P2-10 [M]** `edge/adapters/movenet_pose.py:34` — the CPU MoveNet fallback is + DONE — `_foot` averages only ankles with conf ≥ MIN_ANKLE_CONF (0.2, the existing 'visible' + gate); with neither trustworthy it falls back to bbox bottom-centre. No tuning knob — reuses + the codebase's keypoint-visible convention. test_zone.py locks the fallback. +- [x] **P2-10 [M]** `edge/adapters/movenet_pose.py:34` — the CPU MoveNet fallback is single-person; with 2+ workers all but one are invisible to every detector, no warning. **Fix:** warn loudly that MoveNet fallback is single-person (Hailo is multi-person). -- [ ] **P2-11 [L]** `edge/geometry.py:39` — `head_below_hips` uses the NOSE keypoint + DONE — MoveNetPoseAdapter.__init__ logs a startup WARNING ("SINGLE-PERSON … use POSE_BACKEND= + hailo at multi-person sites"). test_movenet_pose.py asserts the warning via caplog. +- [x] **P2-11 [L]** `edge/geometry.py:39` — `head_below_hips` uses the NOSE keypoint regardless of confidence. **Fix:** confidence-gate. + DONE — `head_below_hips` returns False when nose conf < nose_thr (0.2), so an unconfident/ + hallucinated nose no longer fires the fall cue. test_geometry.py locks it; test_fall.py stays + green (the cue's confident-nose path is unchanged). ## Phase 3 — Privacy / PIPA (delivery compliance) diff --git a/edge/adapters/movenet_pose.py b/edge/adapters/movenet_pose.py index 5e50f1e..a8bc659 100644 --- a/edge/adapters/movenet_pose.py +++ b/edge/adapters/movenet_pose.py @@ -1,3 +1,5 @@ +import logging + import cv2 import numpy as np @@ -17,6 +19,13 @@ class MoveNetPoseAdapter: session = onnxruntime.InferenceSession(onnx_path, providers=["CPUExecutionProvider"]) self.session = session self._input_name = session.get_inputs()[0].name + # MoveNet SinglePose returns ONE person per frame. At a multi-worker site every worker + # but one is invisible to fall/zone/ghost detection with no other symptom — so say it + # loudly at startup. The Hailo backend (yolov8s_pose) is multi-person. + logging.getLogger("edge.pose").warning( + "MoveNet CPU fallback is SINGLE-PERSON: only one worker is tracked per frame; any " + "additional workers are invisible to fall/zone/ghost detection. Use the Hailo " + "backend (POSE_BACKEND=hailo) at multi-person sites.") def infer(self, frame) -> list: h, w = frame.shape[:2] diff --git a/edge/detectors/zone.py b/edge/detectors/zone.py index 22d6e8d..64a1b5b 100644 --- a/edge/detectors/zone.py +++ b/edge/detectors/zone.py @@ -3,6 +3,9 @@ from dataclasses import dataclass from edge.tracking import Track from edge.models import L_ANKLE, R_ANKLE +# Below this an ankle keypoint is treated as unusable (typically hallucinated at conf~0). +MIN_ANKLE_CONF = 0.2 + def point_in_polygon(x: float, y: float, poly: list[tuple[float, float]]) -> bool: inside = False @@ -41,9 +44,18 @@ class ZoneIntrusionDetector: self._fired = {k for k in self._fired if k[0] != track_id} def _foot(self, track: Track, frame_wh=None) -> tuple[float, float]: - kps = track.latest.keypoints - a, b = kps[L_ANKLE], kps[R_ANKLE] - fx, fy = (a.x + b.x) / 2, (a.y + b.y) / 2 + p = track.latest + a, b = p.keypoints[L_ANKLE], p.keypoints[R_ANKLE] + # Average only the ankles the model is confident about. A hallucinated ankle (conf~0, + # often pinned to (0,0) or a frame corner) would otherwise drag the foot point into or + # out of a zone. With neither ankle trustworthy, fall back to the bbox bottom-centre — a + # robust standing-foot proxy (same bbox-fallback discipline as fall.py's cues). + good = [k for k in (a, b) if k.conf >= MIN_ANKLE_CONF] + if good: + fx = sum(k.x for k in good) / len(good) + fy = sum(k.y for k in good) / len(good) + else: + fx, fy = p.bbox.cx, p.bbox.y2 # Zones are stored NORMALIZED (0..1 of the image) by the dashboard, which stretches the # camera snapshot to a fixed canvas. The foot is camera-native pixels, so normalize it # by the frame size to compare in the same space — otherwise every zone is misplaced diff --git a/edge/geometry.py b/edge/geometry.py index 3452b52..ce8397c 100644 --- a/edge/geometry.py +++ b/edge/geometry.py @@ -34,9 +34,15 @@ def torso_angle_deg(keypoints: list[Keypoint]) -> float: return math.degrees(math.atan2(abs(vx), abs(vy))) if (vx or vy) else 0.0 -def head_below_hips(keypoints: list[Keypoint]) -> bool: +def head_below_hips(keypoints: list[Keypoint], nose_thr: float = 0.2) -> bool: + # Only trust the nose if the model is confident about it. An unconfident nose (commonly + # hallucinated at (0,0) / a frame corner) would otherwise assert — or silently suppress — + # this fall cue on data we don't trust. Same 'visible' gate as visible_count(). + nose = keypoints[NOSE] + if nose.conf < nose_thr: + return False _, hy = _hip_center(keypoints) - return keypoints[NOSE].y > hy + return nose.y > hy def visible_count(keypoints: list[Keypoint], thr: float = 0.2) -> int: diff --git a/tests/test_geometry.py b/tests/test_geometry.py index 0f1a251..e704864 100644 --- a/tests/test_geometry.py +++ b/tests/test_geometry.py @@ -31,6 +31,15 @@ def test_head_below_hips_true_when_nose_lower_on_screen(): kps = _kps({0: (5, 120, 1), 11: (0, 100, 1), 12: (10, 100, 1)}) assert head_below_hips(kps) is True +def test_head_below_hips_false_when_nose_low_confidence(): + # A hallucinated low-confidence nose must not fire the head-below-hips fall cue, even when + # its coordinate happens to sit below the hips. + low = _kps({0: (5, 120, 0.1), 11: (0, 100, 1), 12: (10, 100, 1)}) + assert head_below_hips(low) is False + # ...but a confident nose in the same position still fires. + ok = _kps({0: (5, 120, 0.9), 11: (0, 100, 1), 12: (10, 100, 1)}) + assert head_below_hips(ok) is True + def test_visible_count_and_core_visible(): kps = _kps({5: (0, 0, 0.9), 6: (1, 0, 0.9), 11: (0, 1, 0.9), 12: (1, 1, 0.9)}) assert visible_count(kps, thr=0.2) == 4 diff --git a/tests/test_movenet_pose.py b/tests/test_movenet_pose.py index 0db0a8a..fa024df 100644 --- a/tests/test_movenet_pose.py +++ b/tests/test_movenet_pose.py @@ -1,3 +1,4 @@ +import logging import numpy as np from edge.adapters.movenet_pose import MoveNetPoseAdapter @@ -30,3 +31,10 @@ def test_low_score_person_dropped(): return [np.zeros((1, 1, 17, 3), dtype=np.float32)] # all score 0 a = MoveNetPoseAdapter("unused.onnx", session=Low()) assert a.infer(np.zeros((10, 10, 3), np.uint8)) == [] + +def test_warns_that_movenet_is_single_person(caplog): + # MoveNet tracks only ONE person; at a multi-worker site the rest are silently invisible to + # every detector. Construction must warn loudly so the operator knows to use the Hailo backend. + with caplog.at_level(logging.WARNING, logger="edge.pose"): + MoveNetPoseAdapter("unused.onnx", session=_FakeSession()) + assert any("SINGLE-PERSON" in r.getMessage() for r in caplog.records) diff --git a/tests/test_zone.py b/tests/test_zone.py index bfebb5c..481497c 100644 --- a/tests/test_zone.py +++ b/tests/test_zone.py @@ -33,6 +33,21 @@ def test_leaving_zone_resets(): assert out == [] +def _track_lowconf_ankles(bbox, t): + kps = [Keypoint(0, 0, 0.0) for _ in range(17)] # every keypoint hallucinated at the origin + return Track(1, deque([TrackFrame(t, bbox, kps)], maxlen=90)) + +def test_foot_falls_back_to_bbox_when_ankles_low_confidence(): + # Both ankles are hallucinated at (0,0) with conf 0. Averaging them would pin the foot to the + # image origin — outside any centred zone. The detector must instead use the bbox + # bottom-centre, where the worker actually stands, so the intrusion still fires. + z = Zone("z1", "restricted", SQUARE) + det = ZoneIntrusionDetector([z], ZoneConfig(debounce_s=0.0)) + bbox = BBox(40, 10, 60, 50) # bottom-centre = (cx=50, y2=50), inside SQUARE + out = det.update(_track_lowconf_ankles(bbox, 0.0), now=0.0) + assert out == [{"zone_id": "z1", "level": "restricted"}] + + def test_zone_uses_normalized_coordinates_when_frame_size_given(): # The dashboard stores zones as 0..1 of the image (it stretches the snapshot to a fixed # 960x540 canvas). The edge foot is camera-native pixels. Normalizing the foot by the