feat(edge): ImmobilityWatchdog (posture-independent prolonged immobility)

edge/detectors/immobility.py

@@ -0,0 +1,45 @@
+from dataclasses import dataclass
+
+from edge.tracking import Track
+from edge.geometry import normalize
+
+
+@dataclass
+class ImmobilityConfig:
+    motion_thr: float = 0.02
+    immobile_s: float = 120.0
+    zone_immobile_s: float = 30.0
+
+
+class ImmobilityWatchdog:
+    def __init__(self, cfg: ImmobilityConfig):
+        self.cfg = cfg
+        self._still_since: dict[int, float] = {}
+        self._fired: set[int] = set()
+
+    def _displacement(self, track: Track) -> float:
+        if len(track.frames) < 2:
+            return 1.0
+        a, b = track.frames[-2], track.frames[-1]
+        h = b.bbox.height or 1.0
+        d = sum(((p.x - q.x) ** 2 + (p.y - q.y) ** 2) ** 0.5
+                for p, q in zip(a.keypoints, b.keypoints)) / len(b.keypoints)
+        return normalize(d, h)
+
+    def update(self, track: Track, now: float, in_danger_zone: bool = False) -> tuple[bool, dict]:
+        tid = track.track_id
+        window = self.cfg.zone_immobile_s if in_danger_zone else self.cfg.immobile_s
+        if self._displacement(track) >= self.cfg.motion_thr:
+            self._still_since.pop(tid, None)
+            self._fired.discard(tid)
+            return False, {}
+        if tid not in self._still_since:
+            if len(track.frames) >= 2:
+                self._still_since[tid] = track.frames[-2].t
+            else:
+                self._still_since[tid] = now
+        start = self._still_since[tid]
+        if now - start >= window and tid not in self._fired:
+            self._fired.add(tid)
+            return True, {"immobility_s": round(now - start, 1)}
+        return False, {}

tests/test_immobility.py

@@ -0,0 +1,47 @@
+from collections import deque
+from edge.models import Keypoint, BBox
+from edge.tracking import Track, TrackFrame
+from edge.detectors.immobility import ImmobilityWatchdog, ImmobilityConfig
+
+def _kps(x, y):
+    return [Keypoint(x, y, 0.9) for _ in range(17)]
+
+def _push(track, x, y, t):
+    track.frames.append(TrackFrame(t, BBox(x, y, x + 20, y + 60), _kps(x, y)))
+
+def test_still_person_triggers_after_threshold():
+    w = ImmobilityWatchdog(ImmobilityConfig(motion_thr=0.05, immobile_s=1.0))
+    tr = Track(1, deque(maxlen=90))
+    _push(tr, 100, 100, 0.0)
+    assert w.update(tr, now=0.0)[0] is False
+    _push(tr, 100, 100, 0.5)
+    assert w.update(tr, now=0.5)[0] is False
+    _push(tr, 100, 100, 1.5)  # 1.5s 정지
+    fired, meta = w.update(tr, now=1.5)
+    assert fired is True
+    assert meta["immobility_s"] >= 1.0
+
+def test_fires_only_once_until_movement():
+    w = ImmobilityWatchdog(ImmobilityConfig(motion_thr=0.05, immobile_s=1.0))
+    tr = Track(1, deque(maxlen=90))
+    _push(tr, 100, 100, 0.0)
+    _push(tr, 100, 100, 1.5); w.update(tr, now=1.5)
+    _push(tr, 100, 100, 2.0)
+    assert w.update(tr, now=2.0)[0] is False  # 재발화 안 함
+
+def test_movement_resets_timer():
+    w = ImmobilityWatchdog(ImmobilityConfig(motion_thr=0.05, immobile_s=1.0))
+    tr = Track(1, deque(maxlen=90))
+    _push(tr, 100, 100, 0.0)
+    _push(tr, 100, 100, 0.9); w.update(tr, now=0.9)
+    _push(tr, 300, 300, 1.0)  # 크게 이동
+    fired, _ = w.update(tr, now=1.0)
+    assert fired is False
+
+def test_danger_zone_uses_shorter_window():
+    w = ImmobilityWatchdog(ImmobilityConfig(motion_thr=0.05, immobile_s=100.0, zone_immobile_s=1.0))
+    tr = Track(1, deque(maxlen=90))
+    _push(tr, 100, 100, 0.0)
+    _push(tr, 100, 100, 1.5)
+    fired, _ = w.update(tr, now=1.5, in_danger_zone=True)
+    assert fired is True