diff --git a/edge/detectors/immobility.py b/edge/detectors/immobility.py new file mode 100644 index 0000000..e3a35bc --- /dev/null +++ b/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, {} diff --git a/tests/test_immobility.py b/tests/test_immobility.py new file mode 100644 index 0000000..50a11cd --- /dev/null +++ b/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