e81ea8961c
_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.
40 lines
1.3 KiB
Python
40 lines
1.3 KiB
Python
from dataclasses import dataclass
|
|
|
|
from edge.tracking import Track
|
|
from edge.geometry import track_displacement
|
|
|
|
|
|
@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 forget(self, track_id: int) -> None:
|
|
self._still_since.pop(track_id, None)
|
|
self._fired.discard(track_id)
|
|
|
|
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 track_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, {}
|