45 lines
1.6 KiB
Python
45 lines
1.6 KiB
Python
from dataclasses import dataclass
|
|
|
|
from edge.tracking import Track
|
|
from edge.geometry import normalize
|
|
|
|
|
|
@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 _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:
|
|
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, {}
|