refactor(edge): DRY displacement helper into geometry.track_displacement (FIX 6)

_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.
This commit is contained in:
유창욱 2026-07-11 22:21:27 +09:00
parent 752aabaff1
commit e81ea8961c
4 changed files with 34 additions and 23 deletions

View file

@ -2,7 +2,8 @@ from dataclasses import dataclass
from edge.tracking import Track from edge.tracking import Track
from edge.geometry import ( 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), "aspect_ratio": round(ar, 2),
"height_remaining": round(remaining, 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]: def update(self, track: Track, now: float) -> tuple[str | None, dict]:
st = self._states.setdefault(track.track_id, _State()) st = self._states.setdefault(track.track_id, _State())
f = track.latest f = track.latest
@ -106,7 +98,7 @@ class FallDetector:
# FALLEN 이후 미동 감시 # FALLEN 이후 미동 감시
if st.fall_emitted: 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: if st.still_since is None:
st.still_since = now st.still_since = now
elif now - st.still_since >= self.cfg.no_motion_s and not st.no_motion_emitted: elif now - st.still_since >= self.cfg.no_motion_s and not st.no_motion_emitted:

View file

@ -1,7 +1,7 @@
from dataclasses import dataclass from dataclasses import dataclass
from edge.tracking import Track from edge.tracking import Track
from edge.geometry import normalize from edge.geometry import track_displacement
@dataclass @dataclass
@ -21,19 +21,10 @@ class ImmobilityWatchdog:
self._still_since.pop(track_id, None) self._still_since.pop(track_id, None)
self._fired.discard(track_id) 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]: def update(self, track: Track, now: float, in_danger_zone: bool = False) -> tuple[bool, dict]:
tid = track.track_id tid = track.track_id
window = self.cfg.zone_immobile_s if in_danger_zone else self.cfg.immobile_s 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._still_since.pop(tid, None)
self._fired.discard(tid) self._fired.discard(tid)
return False, {} return False, {}

View file

@ -53,3 +53,14 @@ def height_of(bbox: BBox) -> float:
def normalize(value: float, bbox_height: float) -> float: def normalize(value: float, bbox_height: float) -> float:
return value / bbox_height if bbox_height else 0.0 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)

View file

@ -1,8 +1,10 @@
import math import math
from collections import deque
from edge.models import Keypoint, BBox from edge.models import Keypoint, BBox
from edge.tracking import Track, TrackFrame
from edge.geometry import ( from edge.geometry import (
com, torso_angle_deg, head_below_hips, visible_count, com, torso_angle_deg, head_below_hips, visible_count,
core_visible, normalize, core_visible, normalize, track_displacement,
) )
def _kps(overrides): def _kps(overrides):
@ -37,3 +39,18 @@ def test_visible_count_and_core_visible():
def test_normalize_divides_by_bbox_height(): def test_normalize_divides_by_bbox_height():
assert normalize(30, 60) == 0.5 assert normalize(30, 60) == 0.5
assert normalize(30, 0) == 0.0 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)