diff --git a/edge/detectors/fall.py b/edge/detectors/fall.py index 8256a52..6e27c4a 100644 --- a/edge/detectors/fall.py +++ b/edge/detectors/fall.py @@ -2,7 +2,8 @@ from dataclasses import dataclass from edge.tracking import Track from edge.geometry import ( - torso_angle_deg, head_below_hips, core_visible, visible_count, normalize, + torso_angle_deg, head_below_hips, core_visible, visible_count, + track_displacement, ) @@ -55,15 +56,6 @@ class FallDetector: "aspect_ratio": round(ar, 2), "height_remaining": round(remaining, 2)} - 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) -> tuple[str | None, dict]: st = self._states.setdefault(track.track_id, _State()) f = track.latest @@ -106,7 +98,7 @@ class FallDetector: # FALLEN 이후 미동 감시 if st.fall_emitted: - if self._displacement(track) < self.cfg.motion_thr: + if track_displacement(track) < self.cfg.motion_thr: if st.still_since is None: st.still_since = now elif now - st.still_since >= self.cfg.no_motion_s and not st.no_motion_emitted: diff --git a/edge/detectors/immobility.py b/edge/detectors/immobility.py index 3a0dd6d..5ca8721 100644 --- a/edge/detectors/immobility.py +++ b/edge/detectors/immobility.py @@ -1,7 +1,7 @@ from dataclasses import dataclass from edge.tracking import Track -from edge.geometry import normalize +from edge.geometry import track_displacement @dataclass @@ -21,19 +21,10 @@ class ImmobilityWatchdog: self._still_since.pop(track_id, None) self._fired.discard(track_id) - 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: + if track_displacement(track) >= self.cfg.motion_thr: self._still_since.pop(tid, None) self._fired.discard(tid) return False, {} diff --git a/edge/geometry.py b/edge/geometry.py index 9e4e86a..a2626c4 100644 --- a/edge/geometry.py +++ b/edge/geometry.py @@ -53,3 +53,14 @@ def height_of(bbox: BBox) -> float: def normalize(value: float, bbox_height: float) -> float: return value / bbox_height if bbox_height else 0.0 + + +def track_displacement(track) -> float: + """Mean per-keypoint displacement between the last two frames, normalized by bbox height. 1.0 if <2 frames.""" + 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) diff --git a/tests/test_geometry.py b/tests/test_geometry.py index 565b445..5593718 100644 --- a/tests/test_geometry.py +++ b/tests/test_geometry.py @@ -1,8 +1,10 @@ import math +from collections import deque from edge.models import Keypoint, BBox +from edge.tracking import Track, TrackFrame from edge.geometry import ( com, torso_angle_deg, head_below_hips, visible_count, - core_visible, normalize, + core_visible, normalize, track_displacement, ) def _kps(overrides): @@ -37,3 +39,18 @@ def test_visible_count_and_core_visible(): def test_normalize_divides_by_bbox_height(): assert normalize(30, 60) == 0.5 assert normalize(30, 0) == 0.0 + +def test_track_displacement_defaults_to_one_with_fewer_than_two_frames(): + tr = Track(1, deque(maxlen=90)) + tr.frames.append(TrackFrame(0.0, BBox(0, 0, 20, 60), + [Keypoint(0, 0, 0.9) for _ in range(17)])) + assert track_displacement(tr) == 1.0 + +def test_track_displacement_normalizes_mean_delta_by_bbox_height(): + tr = Track(1, deque(maxlen=90)) + tr.frames.append(TrackFrame(0.0, BBox(0, 0, 20, 60), + [Keypoint(0, 0, 0.9) for _ in range(17)])) + tr.frames.append(TrackFrame(0.1, BBox(0, 0, 20, 60), + [Keypoint(6, 8, 0.9) for _ in range(17)])) + # per-point delta = sqrt(6^2+8^2) = 10; bbox height = 60 -> 10/60 + assert math.isclose(track_displacement(tr), 10 / 60)