From e81ea8961c812ef5bb3616ebddee30ff8b5c03d4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=EC=9C=A0=EC=B0=BD=EC=9A=B1?= Date: Sat, 11 Jul 2026 22:21:27 +0900 Subject: [PATCH] refactor(edge): DRY displacement helper into geometry.track_displacement (FIX 6) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit _displacement (mean per-keypoint delta between the last two track frames, normalized by bbox height) was duplicated verbatim in edge/detectors/fall.py and edge/detectors/immobility.py. Move it to edge/geometry.py as track_displacement(track) and have both detectors call the shared helper instead of their own private copy. Pure refactor — behavior is identical (byte-identical formula), no config or public API changes. Tests: tests/test_geometry.py adds direct coverage for the new public helper (fewer-than-two-frames default, and the normalized-by-height calculation). Existing fall/immobility tests are unchanged and stay green, confirming behavior parity. --- edge/detectors/fall.py | 14 +++----------- edge/detectors/immobility.py | 13 ++----------- edge/geometry.py | 11 +++++++++++ tests/test_geometry.py | 19 ++++++++++++++++++- 4 files changed, 34 insertions(+), 23 deletions(-) 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)