"""LOITERING-01: bounded, result-worker-owned ROI dwell using media nanoseconds. No I/O, pixels, tracker matching or background threads. A missing observation breaks the integration interval, not the visit's event latch. Only observed outside time rearms that latch (or explicit clear/idle retention expiry). """ from dataclasses import dataclass import time import uuid from app.intrusion import IntrusionConfig, IntrusionEvent, ScenarioDecision, _inside_edges from app.pose_config import PERSON_MODELS @dataclass(frozen=True) class LoiteringConfig(IntrusionConfig): """Same ROI/settings shape as INTRUSION; distinct temporal semantics.""" roi_name: str = 'Loitering ROI' dwell_seconds: float = 5.0 grace_seconds: float = 1.0 rearm_seconds: float = 1.0 retention_seconds: float = 60.0 def __post_init__(self): super().__post_init__() # Reject unrepresentable timers before settings persistence, not at Start. if min(round(v * 1_000_000_000) for v in (self.dwell_seconds, self.grace_seconds, self.rearm_seconds)) < 1: raise ValueError('Scenario timers must be at least one nanosecond.') @classmethod def parse(cls, value): if type(value) is not dict or set(value) - set(cls.__dataclass_fields__): raise ValueError('Invalid LOITERING settings.') return cls(**value) @dataclass(frozen=True) class LoiteringDecision(ScenarioDecision): def payload(self): return dict(super().payload(), scenario='LOITERING') @dataclass(frozen=True) class LoiteringEvent(IntrusionEvent): def payload(self): return dict(super().payload(), scenario='LOITERING') @dataclass class _Visit: last_seen: float last_valid_ns: int previous_index: int inside: bool dwell_ns: int = 0 outside_ns: int = 0 event_id: str | None = None class Loitering: """One instance per stream/run; clear only after the result worker joins. Samples at the two ends of an interval must both qualify, be consecutive evaluations, and lie inside the ROI. Normal sampling gaps <= grace count; explicit empty/low-score/bad-time evaluations do not. Source frame_seq gaps alone are not detector misses (frame selection legitimately skips them). """ def __init__(self, config: LoiteringConfig, *, clock=time.monotonic, wall_ns=time.time_ns, event_id_factory=lambda: str(uuid.uuid4())): if not isinstance(config, LoiteringConfig): raise ValueError('LOITERING requires LoiteringConfig.') self.config, self.clock = config, clock self.wall_ns, self.event_id_factory = wall_ns, event_id_factory self._edges = tuple(zip(config.roi, config.roi[1:] + config.roi[:1])) self._dwell_ns = round(config.dwell_seconds * 1_000_000_000) self._grace_ns = round(config.grace_seconds * 1_000_000_000) self._rearm_ns = round(config.rearm_seconds * 1_000_000_000) if min(self._dwell_ns, self._grace_ns, self._rearm_ns) < 1: raise ValueError('Scenario timers must be at least one nanosecond.') self.clear() def clear(self): self.states = {} self.identity = None self.last_seq = -1 self.last_pts = None self.index = 0 def expire(self): now = self.clock() expired = [tid for tid, visit in self.states.items() if now - visit.last_seen >= self.config.retention_seconds] for tid in expired: del self.states[tid] def _validate(self, frame, observations, ids): identity = (frame.server_id, frame.source_id, frame.stream_id, frame.run_id, frame.run_generation) if ((self.identity is not None and self.identity != identity) or type(frame.frame_seq) is not int or frame.frame_seq <= self.last_seq): raise ValueError('Stale scenario Frame.') if len(observations) != len(ids) or len(observations) > 300: raise ValueError('Scenario candidate association differs.') seen = set() for obs, tid in zip(observations, ids): if obs.frame != frame: raise ValueError('Scenario Frame association differs.') if obs.model_key not in PERSON_MODELS or obs.class_id != 0: continue if tid is not None: if (type(tid) not in (str, int) or (type(tid) is str and (not tid or len(tid) > 256)) or tid in seen): raise ValueError('Invalid or duplicate scenario track.') seen.add(tid) return identity def evaluate(self, frame, observations, ids, *, gap_before=False): identity = self._validate(frame, observations, ids) if not self.config.enabled: return (None,) * len(observations), () self.expire() pts = frame.pts_ns valid_time = (type(pts) is int and 0 <= pts < 2**63 and (self.last_pts is None or pts >= self.last_pts)) now = self.clock() # Preflight capacity before advancing any observation's temporal state. new_ids = {tid for obs, tid in zip(observations, ids) if valid_time and tid is not None and obs.model_key in PERSON_MODELS and obs.class_id == 0 and obs.score >= self.config.confidence and tid not in self.states} if len(self.states) + len(new_ids) > self.config.max_states: raise RuntimeError('LOITERING state capacity exceeded.') self.identity, self.last_seq = identity, frame.frame_seq self.index += 1 if valid_time: self.last_pts = pts decisions, events = [], [] for index, (obs, tid) in enumerate(zip(observations, ids)): if obs.model_key not in PERSON_MODELS or obs.class_id != 0: decisions.append(None) continue x1, y1, x2, y2 = obs.bbox_xyxy point = (x1 + (x2-x1) * self.config.point_x, y1 + (y2-y1) * self.config.point_y) within = _inside_edges(point, self._edges) visit = self.states.get(tid) if tid is not None else None if visit is not None: visit.last_seen = now event_id = None if not valid_time: state = 'TIME_UNAVAILABLE' elif obs.score < self.config.confidence: state = 'LOW_CONFIDENCE' elif tid is None: state = 'PENDING_ID' else: if visit is None: visit = _Visit(now, pts, self.index, within) self.states[tid] = visit gap = pts - visit.last_valid_ns continuous = not gap_before and visit.previous_index == self.index-1 and gap <= self._grace_ns delta = gap if continuous else 0 if gap > self._grace_ns: visit.dwell_ns = 0 # Unobserved absence is not proof of departure: retain event_id. if not continuous: visit.outside_ns = 0 if within: visit.outside_ns = 0 if visit.inside: visit.dwell_ns += delta # Returning after a short observed outside period keeps prior dwell, # but neither the outside interval nor the return edge adds time. if visit.dwell_ns >= self._dwell_ns: state = 'LOITERING' from app.event_cooldown import allow_event if visit.event_id is None and allow_event(self, 'LOITERING'): visit.event_id = self.event_id_factory() events.append(LoiteringEvent( visit.event_id, frame, tid, self.config.roi_id, str(index), obs.score, obs.bbox_xyxy, obs.model_key, self.wall_ns() // 1_000_000)) event_id = visit.event_id else: state = 'CONFIRMING' else: visit.outside_ns = visit.outside_ns + delta if not visit.inside else 0 if visit.outside_ns >= self._rearm_ns: visit.dwell_ns = 0 visit.event_id = None state = 'OUTSIDE' visit.last_valid_ns, visit.previous_index = pts, self.index visit.inside = within decisions.append(LoiteringDecision(point, within, state, self.config.confidence, event_id)) return tuple(decisions), tuple(events)