48 lines
1.8 KiB
Python
48 lines
1.8 KiB
Python
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
|