752aabaff1
fall._states, immobility._still_since/_fired, zone._inside_since/_fired, and engine._last_emit were never pruned when a track was lost (ghost already pruned via on_lost). On a 24/7 edge device this is unbounded growth as track IDs keep incrementing. Add forget(track_id) to FallDetector, ImmobilityWatchdog, and ZoneIntrusionDetector (drops that track's per-track keys/sets). EventEngine.forget(track_id) calls each sub-detector's forget and removes _last_emit cooldown keys whose tuple's second element is that track_id (covers FALL/NO_MOTION/IMMOBILITY/zone/GHOST — GHOST keys are (GHOST, track_id) after FIX 4). EventEngine.process now calls self.forget(tr.track_id) for every track in lost_tracks after emitting, since ghost.on_lost already consumed that track's ghost-relevant state at the top of process(). app.py run_once already calls pool.forget(tr.track_id) for lost tracks as part of FIX 1, completing the cleanup chain (smoother pool + detector state + dedup cache). Tests: tests/test_engine.py adds test_forget_prunes_state_on_lost — fires a FALL event, then passes the track as lost, and asserts engine._last_emit no longer has that track's keys and fall._states no longer has the track id.
118 lines
4.4 KiB
Python
118 lines
4.4 KiB
Python
from dataclasses import dataclass
|
|
|
|
from edge.tracking import Track
|
|
from edge.geometry import (
|
|
torso_angle_deg, head_below_hips, core_visible, visible_count, normalize,
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class FallConfig:
|
|
falling_angle: float = 45.0
|
|
fallen_angle: float = 70.0
|
|
aspect_ratio: float = 1.25
|
|
height_collapse: float = 0.35 # baseline 대비 남은 높이 비율(이 값 미만이면 급붕괴)
|
|
prone_confirm_s: float = 3.0
|
|
no_motion_s: float = 10.0
|
|
motion_thr: float = 0.02 # 정규화 관절변위 임계
|
|
min_conf: float = 0.5
|
|
min_visible: int = 10
|
|
recover_grace_s: float = 1.0 # 사고 상태에서 회복 판정까지 요구되는 지속 비-prone 시간
|
|
|
|
|
|
@dataclass
|
|
class _State:
|
|
name: str = "STANDING"
|
|
baseline_height: float | None = None
|
|
prone_since: float | None = None
|
|
still_since: float | None = None
|
|
fall_emitted: bool = False
|
|
no_motion_emitted: bool = False
|
|
not_prone_since: float | None = None
|
|
|
|
|
|
class FallDetector:
|
|
def __init__(self, cfg: FallConfig):
|
|
self.cfg = cfg
|
|
self._states: dict[int, _State] = {}
|
|
|
|
def forget(self, track_id: int) -> None:
|
|
self._states.pop(track_id, None)
|
|
|
|
def _prone(self, kps, bbox, st: _State) -> tuple[bool, dict]:
|
|
angle = torso_angle_deg(kps)
|
|
ar = bbox.aspect
|
|
remaining = 1.0
|
|
if st.baseline_height:
|
|
remaining = bbox.height / st.baseline_height
|
|
prone = (
|
|
angle > self.cfg.fallen_angle
|
|
or ar > self.cfg.aspect_ratio
|
|
or remaining < self.cfg.height_collapse # 높이 급붕괴(정면/후면 낙상)
|
|
or head_below_hips(kps)
|
|
)
|
|
return prone, {"torso_angle": round(angle, 1),
|
|
"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
|
|
kps, bbox = f.keypoints, f.bbox
|
|
|
|
# 신뢰도 게이트
|
|
if not core_visible(kps, self.cfg.min_conf) or visible_count(kps) < self.cfg.min_visible:
|
|
return None, {}
|
|
|
|
prone, meta = self._prone(kps, bbox, st)
|
|
|
|
# incident 비활성일 때만 baseline(정상 키높이) 갱신 = incident 중 freeze
|
|
if st.name == "STANDING" and not prone:
|
|
st.baseline_height = bbox.height
|
|
|
|
if not prone:
|
|
if st.name in ("FALLING", "FALLEN", "NO_MOTION"):
|
|
# 단일 튐(noisy) 프레임만으로 CRITICAL 사고를 즉시 취소하지 않도록
|
|
# recover_grace_s만큼 지속된 비-prone을 요구한다.
|
|
if st.not_prone_since is None:
|
|
st.not_prone_since = now
|
|
return None, {}
|
|
if now - st.not_prone_since >= self.cfg.recover_grace_s:
|
|
self._states[track.track_id] = _State(baseline_height=bbox.height)
|
|
return "recover", {}
|
|
return None, {}
|
|
st.prone_since = None
|
|
return None, {}
|
|
|
|
# prone 상태 (재-prone 프레임은 대기 중인 recover 유예를 취소)
|
|
st.not_prone_since = None
|
|
if st.prone_since is None:
|
|
st.prone_since = now
|
|
st.name = "FALLING"
|
|
if now - st.prone_since >= self.cfg.prone_confirm_s and not st.fall_emitted:
|
|
st.name = "FALLEN"
|
|
st.fall_emitted = True
|
|
st.still_since = now
|
|
return "fall", meta
|
|
|
|
# FALLEN 이후 미동 감시
|
|
if st.fall_emitted:
|
|
if self._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:
|
|
st.name = "NO_MOTION"
|
|
st.no_motion_emitted = True
|
|
return "no_motion", {**meta, "immobility_s": round(now - st.still_since, 1)}
|
|
else:
|
|
st.still_since = None
|
|
return None, {}
|