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