move_detect/edge/engine.py

from dataclasses import dataclass

from edge.tracking import Track
from edge.events import Event, EventType, severity_for
from edge.detectors.fall import FallDetector
from edge.detectors.immobility import ImmobilityWatchdog
from edge.detectors.ghost import GhostIncidentDetector
from edge.detectors.zone import ZoneIntrusionDetector


@dataclass
class EngineConfig:
    camera_id: str = "cam"
    cooldown_s: float = 30.0


class EventEngine:
    def __init__(self, fall: FallDetector, immobility: ImmobilityWatchdog,
                 ghost: GhostIncidentDetector, zone: ZoneIntrusionDetector,
                 cfg: EngineConfig):
        self.fall = fall
        self.immobility = immobility
        self.ghost = ghost
        self.zone = zone
        self.cfg = cfg
        self._last_emit: dict[tuple, float] = {}

    def _allow(self, key: tuple, now: float) -> bool:
        last = self._last_emit.get(key)
        if last is not None and now - last < self.cfg.cooldown_s:
            return False
        self._last_emit[key] = now
        return True

    def forget(self, track_id: int) -> None:
        """트랙 소실 시 하위 감지기 + dedup 캐시의 트랙별 상태를 정리 (24/7 장치 메모리 누수 방지)."""
        self.fall.forget(track_id)
        self.immobility.forget(track_id)
        self.zone.forget(track_id)
        self._last_emit = {k: v for k, v in self._last_emit.items()
                           if not (len(k) >= 2 and k[1] == track_id)}

    def _mk(self, etype: EventType, now: float, track_id=None, zone_id=None,
            metrics=None, spatial=None) -> Event:
        m = metrics or {}
        duration_s = m.get("immobility_s") or m.get("waited_s") or 0.0
        return Event(etype, severity_for(etype), self.cfg.camera_id, track_id,
                     zone_id, now, duration_s, m, spatial)

    def process(self, active_tracks: list[Track], lost_tracks: list[Track],
                now: float, frame_wh=None) -> list[Event]:
        events: list[Event] = []
        self.ghost.on_lost(lost_tracks, now)

        for tr in active_tracks:
            # frame_wh lets the zone detector normalize the foot to the dashboard's 0..1 zone
            # space (see ZoneIntrusionDetector._foot). run_once supplies the live frame size.
            zone_hits = self.zone.update(tr, now, frame_wh)
            in_zone = bool(self.zone.active_zone_ids(tr.track_id))

            sig, meta = self.fall.update(tr, now)
            state = self.fall._states.get(tr.track_id)
            self.ghost.note_state(tr.track_id, state.name if state else "STANDING")
            if sig == "fall" and self._allow((EventType.FALL, tr.track_id), now):
                events.append(self._mk(EventType.FALL, now, tr.track_id, metrics=meta))
            elif sig == "no_motion" and self._allow((EventType.NO_MOTION, tr.track_id), now):
                events.append(self._mk(EventType.NO_MOTION, now, tr.track_id, metrics=meta))

            fired, imeta = self.immobility.update(tr, now, in_danger_zone=in_zone)
            if fired and self._allow((EventType.IMMOBILITY, tr.track_id), now):
                events.append(self._mk(EventType.IMMOBILITY, now, tr.track_id, metrics=imeta))

            for hit in zone_hits:
                # 협착(pinch) is the highest-danger zone (machinery entrapment) and must emit
                # PINCH/HIGH — the old `restricted else WARNING` silently downgraded it. The
                # dashboard offers exactly these three levels (see zones.php).
                etype = {"pinch": EventType.PINCH,
                         "restricted": EventType.ZONE_RESTRICTED}.get(
                             hit["level"], EventType.ZONE_WARNING)
                if self._allow((etype, tr.track_id, hit["zone_id"]), now):
                    events.append(self._mk(etype, now, tr.track_id, zone_id=hit["zone_id"]))

        for g in self.ghost.update(active_tracks, now):
            key = (EventType.GHOST, g["track_id"])
            if self._allow(key, now):
                events.append(self._mk(EventType.GHOST, now, track_id=g["track_id"],
                                       spatial=g["spatial"],
                                       metrics={"last_state": g["last_state"],
                                                "waited_s": g["waited_s"]}))

        # ghost.on_lost already consumed lost tracks' state above, so forgetting here is safe.
        for tr in lost_tracks:
            self.forget(tr.track_id)
        return events