feat(edge): FallDetector state machine (OR-logic, baseline freeze, post-fall no-motion)

edge/detectors/fall.py

@@ -0,0 +1,105 @@
+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
+
+
+@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
+
+
+class FallDetector:
+    def __init__(self, cfg: FallConfig):
+        self.cfg = cfg
+        self._states: dict[int, _State] = {}
+
+    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"):
+                self._states[track.track_id] = _State(baseline_height=bbox.height)
+                return "recover", {}
+            st.prone_since = None
+            return None, {}
+
+        # prone 상태
+        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, {}

tests/test_fall.py

@@ -0,0 +1,77 @@
+from collections import deque
+from edge.models import Keypoint, BBox
+from edge.tracking import Track, TrackFrame
+from edge.detectors.fall import FallDetector, FallConfig
+
+def _standing_kps():
+    kps = [Keypoint(50, 150, 0.9) for _ in range(17)]
+    kps[5] = Keypoint(40, 100, 0.9); kps[6] = Keypoint(60, 100, 0.9)
+    kps[11] = Keypoint(40, 200, 0.9); kps[12] = Keypoint(60, 200, 0.9)
+    kps[0] = Keypoint(50, 60, 0.9)   # nose above hips
+    return kps
+
+def _lying_kps():
+    kps = [Keypoint(150, 300, 0.9) for _ in range(17)]
+    kps[5] = Keypoint(100, 300, 0.9); kps[6] = Keypoint(100, 310, 0.9)
+    kps[11] = Keypoint(200, 300, 0.9); kps[12] = Keypoint(200, 310, 0.9)
+    kps[0] = Keypoint(90, 320, 0.9)  # nose below hips
+    return kps
+
+def _track():
+    return Track(1, deque(maxlen=90))
+
+def _feed(det, track, kps, bbox, t):
+    track.frames.append(TrackFrame(t, bbox, kps))
+    return det.update(track, now=t)
+
+def test_standing_produces_no_event():
+    det = FallDetector(FallConfig())
+    tr = _track()
+    sig, _ = _feed(det, tr, _standing_kps(), BBox(40, 60, 60, 210), t=0.0)
+    assert sig is None
+
+def test_lying_confirmed_after_prone_seconds_emits_fall():
+    det = FallDetector(FallConfig(prone_confirm_s=1.0))
+    tr = _track()
+    lying_box = BBox(90, 290, 210, 320)  # 넓고 낮음 (aspect>1.25)
+    _feed(det, tr, _lying_kps(), lying_box, t=0.0)
+    sig, meta = _feed(det, tr, _lying_kps(), lying_box, t=1.5)  # prone 1.5s > 1.0
+    assert sig == "fall"
+    assert meta["torso_angle"] > 80
+
+def test_front_fall_height_collapse_triggers_without_angle():
+    # 정면 낙상: 각도/종횡비 안 변해도 bbox 높이 급붕괴
+    det = FallDetector(FallConfig(prone_confirm_s=0.5))
+    tr = _track()
+    tall = BBox(40, 60, 60, 210)   # height 150
+    _feed(det, tr, _standing_kps(), tall, t=0.0)
+    short = BBox(40, 180, 60, 210)  # height 30 (붕괴 0.8)
+    _feed(det, tr, _standing_kps(), short, t=0.2)
+    sig, _ = _feed(det, tr, _standing_kps(), short, t=0.9)
+    assert sig == "fall"
+
+def test_no_motion_after_fall_escalates():
+    det = FallDetector(FallConfig(prone_confirm_s=0.5, no_motion_s=1.0, motion_thr=0.05))
+    tr = _track()
+    box = BBox(90, 290, 210, 320)
+    _feed(det, tr, _lying_kps(), box, t=0.0)
+    _feed(det, tr, _lying_kps(), box, t=0.6)  # -> fall
+    _feed(det, tr, _lying_kps(), box, t=1.0)  # 미동 지속
+    sig, _ = _feed(det, tr, _lying_kps(), box, t=2.0)  # 부동 >1.0s
+    assert sig == "no_motion"
+
+def test_recovery_clears_state():
+    det = FallDetector(FallConfig(prone_confirm_s=0.5))
+    tr = _track()
+    box = BBox(90, 290, 210, 320)
+    _feed(det, tr, _lying_kps(), box, t=0.0)
+    _feed(det, tr, _lying_kps(), box, t=0.6)  # fall
+    sig, _ = _feed(det, tr, _standing_kps(), BBox(40, 60, 60, 210), t=1.0)
+    assert sig == "recover"
+
+def test_low_confidence_is_ignored():
+    det = FallDetector(FallConfig())
+    tr = _track()
+    kps = [Keypoint(0, 0, 0.1) for _ in range(17)]  # 전부 저신뢰
+    sig, _ = _feed(det, tr, kps, BBox(0, 0, 100, 30), t=0.0)
+    assert sig is None