from dataclasses import dataclass from edge.tracking import Track from edge.geometry import ( torso_angle_deg, head_below_hips, core_visible, visible_count, track_displacement, ) @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 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 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: 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, {}