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 _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) -> list[Event]: events: list[Event] = [] self.ghost.on_lost(lost_tracks, now) for tr in active_tracks: zone_hits = self.zone.update(tr, now) 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: etype = (EventType.ZONE_RESTRICTED if hit["level"] == "restricted" else 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"]})) return events