"""CROWD-01: bounded periodic P/H estimates, owned by one existing result worker. Media PTS determines samples, sliding windows and report cadence. Monotonic time is only an idle/freshness watchdog; it never creates samples or advances media coverage. No threads, pixel references, tracking, I/O or Event production. """ from collections import Counter, deque from dataclasses import asdict, dataclass import math import time from app.contracts.frames import DetectionObservation, FrameEnvelope from app.crowd_config import CROWD_MODEL, FRAME_OBJECT_LIMIT, SAMPLE_LIMIT, CrowdConfig, duration_ns from app.detection_nms import detection_nms_indices from app.intrusion import _inside_edges @dataclass(frozen=True, slots=True) class CrowdEstimate: person_count: int head_count: int person_mean: float head_mean: float sample_count: int report_pts_ns: int first_sample_pts_ns: int last_sample_pts_ns: int @dataclass(frozen=True, slots=True) class CrowdEvaluation: accepted: bool indices: tuple[int, ...] = () person_count: int = 0 head_count: int = 0 estimate: CrowdEstimate | None = None reported: bool = False reason: str | None = None class Crowd: def __init__(self, config: CrowdConfig, *, clock=time.monotonic): if not isinstance(config, CrowdConfig): raise ValueError('CROWD requires CrowdConfig.') self.config, self.clock = config, clock self._window_ns = duration_ns(config.window_seconds) self._report_ns = duration_ns(config.report_seconds) self._stale_ns = duration_ns(config.stale_seconds) self._edges = tuple(zip(config.roi, config.roi[1:] + config.roi[:1])) self.metrics = Counter() self.clear() def clear(self): """Call only from the owner, or after its worker has joined.""" self._samples = deque() self._identity = None self._last_seq = -1 self._last_pts = None self._next_report_ns = None self._estimate = None self._last_sample_mono = None self._last_report_mono = None self._last_clock = None self.metrics.clear() def _now(self): value = self.clock() if (type(value) not in (int, float) or not math.isfinite(value) or (self._last_clock is not None and value < self._last_clock)): raise ValueError('Invalid CROWD monotonic clock.') self._last_clock = value return value def _expire_idle(self, now): changed = False if (self._estimate is not None and self._last_report_mono is not None and now - self._last_report_mono >= self.config.stale_seconds): self._estimate = None self._last_report_mono = None self.metrics['idle_estimate_expired'] += 1 changed = True if (self._last_sample_mono is not None and now - self._last_sample_mono >= self.config.stale_seconds): # A frozen source cannot carry old samples through an idle interval. self._samples.clear() self._next_report_ns = None self._last_sample_mono = None self.metrics['idle_history_cleared'] += 1 return changed def expire_idle(self): """Housekeeping only; does not fabricate or republish a FrameResult. A browser retaining its last packet needs its existing overlay-expiry policy. New source-frame results will no longer contain the old label. """ return self._expire_idle(self._now()) @staticmethod def _frame_identity(frame): if not isinstance(frame, FrameEnvelope): raise ValueError('CROWD requires a canonical FrameEnvelope.') for value in (frame.frame_seq, frame.run_generation, frame.frame_width, frame.frame_height): if type(value) is not int: raise ValueError('Invalid CROWD Frame identity.') return (frame.server_id, frame.source_id, frame.stream_id, frame.run_id, frame.run_generation, frame.frame_width, frame.frame_height, frame.config_version) def evaluate(self, frame, observations, *, usable=True): """Only successful, current inference is usable; an empty tuple is valid. `usable=False` is a missing/rejected result, NOT a zero-count sample. Late/queue-rejected packets must normally be discarded by the caller before this stateful method. Raw slots must already be canonical 0/1. """ identity = self._frame_identity(frame) if type(usable) is not bool: raise ValueError('Invalid CROWD observation status.') if self._identity is not None and identity != self._identity: raise ValueError('CROWD run/config/geometry changed; clear or replace the engine.') if not self.config.enabled: return CrowdEvaluation(False, reason='DISABLED') if not usable: self.metrics['unusable'] += 1 return CrowdEvaluation(False, reason='UNUSABLE') if not isinstance(observations, (tuple, list)) or len(observations) > FRAME_OBJECT_LIMIT: raise ValueError('Invalid CROWD observation bound.') if any(not isinstance(o, DetectionObservation) or o.frame != frame or o.model_key != CROWD_MODEL or type(o.class_id) is not int or o.class_id not in (0, 1) for o in observations): raise ValueError('CROWD Frame/model/canonical class mismatch.') if frame.frame_seq <= self._last_seq: self.metrics['duplicate_or_out_of_order'] += 1 return CrowdEvaluation(False, reason='FRAME_ORDER') pts = frame.pts_ns if (type(pts) is not int or not 0 <= pts < 2**63 or (self._last_pts is not None and pts <= self._last_pts)): self.metrics['invalid_or_nonincreasing_pts'] += 1 return CrowdEvaluation(False, reason='PTS_INVALID') # Perform input/NMS validation before advancing temporal state. selected = tuple(i for i, o in enumerate(observations) if o.score >= self.config.confidence_for(o.class_id) and o.bbox_xyxy[0] < o.bbox_xyxy[2] and o.bbox_xyxy[1] < o.bbox_xyxy[3]) pre_nms = len(selected) if self.config.nms_enabled: subset = tuple(observations[i] for i in selected) selected = tuple(selected[i] for i in detection_nms_indices( subset, self.config.nms_iou_threshold)) after_nms = len(selected) kept = [] for i in selected: o = observations[i] x1, y1, x2, y2 = o.bbox_xyxy point = ((x1 + x2) / 2, y2 if o.class_id == 0 else (y1 + y2) / 2) if _inside_edges(point, self._edges): kept.append(i) person = sum(observations[i].class_id == 0 for i in kept) head = len(kept) - person now = self._now() self._expire_idle(now) self._identity, self._last_seq, self._last_pts = identity, frame.frame_seq, pts self._last_sample_mono = now self.metrics['valid_samples'] += 1 self.metrics['nms_suppressed'] += pre_nms - after_nms self.metrics['roi_excluded'] += after_nms - len(kept) self.metrics['object_limit_hits'] += int(len(observations) == FRAME_OBJECT_LIMIT) # Exact interval: (report_pts - window, report_pts]. No double-counted edge. while self._samples and self._samples[0][0] <= pts - self._window_ns: self._samples.popleft() if len(self._samples) >= SAMPLE_LIMIT: # Never silently turn a full-window mean into a truncated tail mean. # Discard only scalar monitoring state, then warm up again; no fatal. self._samples.clear() self._estimate = None self._last_report_mono = None self._next_report_ns = None self.metrics['history_overflow_reset'] += 1 self._samples.append((pts, person, head)) if self._estimate is not None and pts - self._estimate.report_pts_ns >= self._stale_ns: self._estimate = None self._last_report_mono = None self.metrics['media_estimate_expired'] += 1 if self._next_report_ns is None: self._next_report_ns = pts + self._report_ns reported = False if pts >= self._next_report_ns: # Skip elapsed schedule slots; never create catch-up reports. self._next_report_ns += ((pts - self._next_report_ns) // self._report_ns + 1) * self._report_ns n = len(self._samples) if n >= self.config.minimum_samples: p_mean = sum(s[1] for s in self._samples) / n h_mean = sum(s[2] for s in self._samples) / n self._estimate = CrowdEstimate(round(p_mean), round(h_mean), p_mean, h_mean, n, pts, self._samples[0][0], self._samples[-1][0]) self._last_report_mono = now self.metrics['reports'] += 1 reported = True else: self.metrics['insufficient_samples'] += 1 return CrowdEvaluation(True, tuple(kept), person, head, self._estimate, reported) def diagnostics(self): """Bounded private status. Never attach it to Redis Frame Metadata.""" return dict(metrics=dict(self.metrics), samples=len(self._samples), next_report_pts_ns=self._next_report_ns, estimate=asdict(self._estimate) if self._estimate is not None else None)