feat(edge): GhostIncidentDetector (collapse-then-occluded spatial incident)

edge/detectors/ghost.py

@@ -0,0 +1,48 @@
+from dataclasses import dataclass
+
+from edge.tracking import Track
+
+_DANGEROUS = {"FALLING", "FALLEN", "NO_MOTION"}
+
+
+@dataclass
+class GhostConfig:
+    reappear_s: float = 10.0
+    reappear_dist: float = 1.0  # bbox 높이 배수
+
+
+class GhostIncidentDetector:
+    def __init__(self, cfg: GhostConfig):
+        self.cfg = cfg
+        self._states: dict[int, str] = {}
+        # 후보: track_id -> (spatial(x,y), last_state, height, lost_at)
+        self._candidates: dict[int, tuple] = {}
+
+    def note_state(self, track_id: int, state: str) -> None:
+        self._states[track_id] = state
+
+    def on_lost(self, lost_tracks: list[Track], now: float) -> None:
+        for tr in lost_tracks:
+            state = self._states.get(tr.track_id, "STANDING")
+            if state in _DANGEROUS:
+                b = tr.latest.bbox
+                self._candidates[tr.track_id] = ((b.cx, b.cy), state, b.height, now)
+            self._states.pop(tr.track_id, None)
+
+    def update(self, active_tracks: list[Track], now: float) -> list[dict]:
+        # 인근 재출현이면 후보 취소
+        for tid, (spatial, _state, height, _lost_at) in list(self._candidates.items()):
+            for tr in active_tracks:
+                b = tr.latest.bbox
+                dist = ((b.cx - spatial[0]) ** 2 + (b.cy - spatial[1]) ** 2) ** 0.5
+                if dist <= self.cfg.reappear_dist * (height or 1.0):
+                    self._candidates.pop(tid, None)
+                    break
+        # 대기 초과 방출
+        out = []
+        for tid, (spatial, state, _height, lost_at) in list(self._candidates.items()):
+            if now - lost_at >= self.cfg.reappear_s:
+                out.append({"spatial": spatial, "last_state": state,
+                            "waited_s": round(now - lost_at, 1)})
+                self._candidates.pop(tid, None)
+        return out

tests/test_ghost.py

@@ -0,0 +1,32 @@
+from collections import deque
+from edge.models import Keypoint, BBox
+from edge.tracking import Track, TrackFrame
+from edge.detectors.ghost import GhostIncidentDetector, GhostConfig
+
+def _track(tid, x, y, t):
+    fr = TrackFrame(t, BBox(x, y, x + 20, y + 60), [Keypoint(x, y, 0.9) for _ in range(17)])
+    return Track(tid, deque([fr], maxlen=90))
+
+def test_lost_dangerous_track_becomes_ghost_after_wait():
+    d = GhostIncidentDetector(GhostConfig(reappear_s=1.0))
+    d.note_state(1, "FALLEN")
+    d.on_lost([_track(1, 100, 100, 0.0)], now=0.0)
+    assert d.update([], now=0.5) == []          # 아직 대기
+    ghosts = d.update([], now=1.5)               # 재출현 없음 > 1.0s
+    assert len(ghosts) == 1
+    assert ghosts[0]["last_state"] == "FALLEN"
+    assert ghosts[0]["spatial"] == (110.0, 130.0)  # bbox 중심
+
+def test_non_dangerous_lost_track_is_ignored():
+    d = GhostIncidentDetector(GhostConfig(reappear_s=1.0))
+    d.note_state(1, "STANDING")
+    d.on_lost([_track(1, 100, 100, 0.0)], now=0.0)
+    assert d.update([], now=2.0) == []
+
+def test_reappearance_near_location_cancels_ghost():
+    d = GhostIncidentDetector(GhostConfig(reappear_s=1.0, reappear_dist=2.0))
+    d.note_state(1, "FALLEN")
+    d.on_lost([_track(1, 100, 100, 0.0)], now=0.0)
+    # 인근에 새 트랙 재출현
+    assert d.update([_track(2, 105, 105, 0.5)], now=0.5) == []
+    assert d.update([], now=2.0) == []  # 취소되어 방출 없음