move_detect/edge/detectors/ghost.py

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
                # Snapshot the OTHER tracks active at loss time. Only a track that did NOT
                # exist then can be the victim re-appearing (a re-acquired track gets a new
                # id); a track already present is a bystander/colleague and must not cancel
                # the incident by walking over the loss point.
                bystanders = set(self._states) - {tr.track_id}
                self._candidates[tr.track_id] = ((b.cx, b.cy), state, b.height, now, bystanders)
            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, bystanders) in list(self._candidates.items()):
            for tr in active_tracks:
                if tr.track_id in bystanders:
                    continue  # a colleague who was already here — not the reappearing victim
                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, _by) in list(self._candidates.items()):
            if now - lost_at >= self.cfg.reappear_s:
                out.append({"track_id": tid, "spatial": spatial, "last_state": state,
                            "waited_s": round(now - lost_at, 1)})
                self._candidates.pop(tid, None)
        return out