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
    # 높이 무장의 유효 기간. 무장에 시간 제한이 없으면 발판에 잠깐 올라갔던 사람이 몇 분 뒤
    # 그냥 걸어 나가도 최상위 추락 경보가 된다.
    height_arm_ttl_s: float = 20.0
    reappear_dist: float = 1.0  # bbox 높이 배수


class GhostIncidentDetector:
    def __init__(self, cfg: GhostConfig):
        self.cfg = cfg
        self._states: dict[int, str] = {}
        # 높이 기반 무장: 위험 자세가 아니어도 '떨어지는 중'이거나 '고소 구역 안'이면
        # 소실 시 추락 후보로 잡는다. 발판에 서 있다가 화각 밖으로 떨어지는 사람은
        # 소실 순간 STANDING 이라 _DANGEROUS 조건만으로는 영원히 안 걸린다.
        self._height: dict[int, dict] = {}
        # 후보: 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 forget(self, track_id: int) -> None:
        """트랙 상태를 버린다. 엔진이 트랙 교체(ambiguous)에서도 부른다 — 무장이 남으면
        교차한 뒤 살아남은 다른 사람이 그 무장을 물려받아 추락 경보를 낸다."""
        self._states.pop(track_id, None)
        self._height.pop(track_id, None)

    def disarm_height(self, track_id: int) -> None:
        """프레임 안에서 이미 추락으로 발보했으면 소실 경로가 같은 사고를 또 내지 않게 한다."""
        self._height.pop(track_id, None)

    def arm_height(self, track_id: int, descent: dict | None, elevated: bool,
                   now: float) -> None:
        """이 트랙을 추락 후보로 표시한다. 소실되면 fall_from_height 로 발보한다."""
        info = self._height.setdefault(track_id, {})
        info["armed_at"] = now
        if descent:
            info.update(descent)
            info["armed_by"] = "descent"
        elif elevated and "armed_by" not in info:
            info["armed_by"] = "zone"
        if elevated:
            info["elevated_zone"] = True

    def on_lost(self, lost_tracks: list[Track], now: float) -> None:
        for tr in lost_tracks:
            state = self._states.get(tr.track_id, "STANDING")
            height = self._height.get(tr.track_id)
            # 오래된 무장은 무시한다 (H2). 고소 구역 안에 있는 동안은 매 틱 갱신되므로
            # 구역 기반 무장은 이 검사에 걸리지 않는다.
            if height is not None and now - float(height.get("armed_at", now)) > self.cfg.height_arm_ttl_s:
                height = None
                self._height.pop(tr.track_id, None)
            if state in _DANGEROUS or height:
                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, height)
            self._states.pop(tr.track_id, None)
            self._height.pop(tr.track_id, None)

    def update(self, active_tracks: list[Track], now: float) -> list[dict]:
        # 인근 재출현이면 후보 취소 — 단, 손실 시점에 없던 새 트랙만 (진짜 재출현).
        for tid, (spatial, _state, height, _lost_at, bystanders, _h) 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, hinfo) in list(self._candidates.items()):
            if now - lost_at >= self.cfg.reappear_s:
                rec = {"track_id": tid, "spatial": spatial, "last_state": state,
                       "waited_s": round(now - lost_at, 1)}
                if hinfo:
                    # 추락 후보였다 — 엔진이 ghost 가 아니라 fall_from_height 로 올린다.
                    rec["height_fall"] = True
                    rec.update(hinfo)
                out.append(rec)
                self._candidates.pop(tid, None)
        return out