from edge.config import Config from edge.tracking import IouTracker from edge.smoothing import TrackSmootherPool from edge.detectors.fall import FallDetector from edge.detectors.immobility import ImmobilityWatchdog from edge.detectors.ghost import GhostIncidentDetector from edge.detectors.zone import ZoneIntrusionDetector from edge.engine import EventEngine from edge.snapshot import RingBuffer, FaceBlurrer, Snapshotter from edge.models import BBox from edge.events import EventType # Evidence timeline offsets (seconds before the event) — 3 frames tell how it developed. # The server accepts at most 3 snapshots per event (EventValidator), so this is the cap. SNAPSHOT_OFFSETS_S = (2.5, 1.0, 0.0) def _head_boxes(persons: list) -> list: # Head region for face anonymization (PIPA). Sized from the PERSON bbox, not a fixed # ~40x40 patch — a close worker fills far more than 40px, so a constant box left the face # fully identifiable. Centered on the nose when it is confident, else on the bbox's top # (the head is the upper part of the person) so a bad keypoint can't move the blur off it. boxes = [] for p in persons: bb = p.bbox nose = p.keypoints[0] half_w = max(8.0, 0.25 * bb.width) if nose.conf >= 0.3: cx, cy = nose.x, nose.y else: cx, cy = bb.cx, bb.y1 + 0.12 * bb.height boxes.append(BBox(cx - half_w, cy - 0.18 * bb.height, cx + half_w, cy + 0.08 * bb.height)) return boxes def run_once(source, pose, tracker, pool, engine, ring, snapshotter, sink, now_fn, wall_fn=None): ok, frame = source.read() if not ok: # No frame this tick. Return None (not []) so the daemon can tell "camera produced no # frame" apart from "frame processed, no events" — a dead/unplugged USB camera returns # (False, None) forever, and counting it as a healthy frame made a monitoring outage # look perfectly green (fps~=target, cameras_erroring=0). return None now = now_fn() persons = pose.infer(frame) tracks = tracker.update(persons, now) # RAW keypoints; smoothing happens per-track below lost = tracker.lost() # Head boxes are computed EVERY tick and stored with the frame (not only when an event # fires): a past frame pulled from the ring later must be blurred with the head positions # of its own moment, not the current tick's. head_boxes = _head_boxes(persons) # A contended (crossing) match may be an id swap — FALLEN state, frozen baseline, or an # immobility timer would ride along to the wrong person. Reset that track's safety state; # a genuinely fallen worker still prone re-confirms within prone_confirm_s. for tid in tracker.ambiguous(): engine.forget(tid) pool.forget(tid) # A track not re-detected this tick still sits in the tracker (< max_age) with a frozen # latest frame. Feeding it downstream would re-smooth the same frame and let time-based # detector timers (prone confirm / no-motion / immobility) accrue on stale data — treat # it as no-data until it is re-detected or ages into lost(). fresh = [tr for tr in tracks if tr.latest.t == now] for tr in fresh: f = tr.latest f.keypoints = pool.smoother_for(tr.track_id).smooth(f.keypoints, now) ring.push(now, frame, head_boxes) h, w = frame.shape[:2] # `now` (monotonic) drives detector timers; wall time stamps the event (occurred_at). Off the # daemon (direct callers / tests) wall_fn is None → engine falls back to now, as before. wall_now = wall_fn() if wall_fn is not None else None events = engine.process(fresh, lost, now, frame_wh=(w, h), wall_now=wall_now) for tr in lost: pool.forget(tr.track_id) if events: # ERGO_REPORT is an anonymous aggregate (posture-exposure sums) — attaching a frame # would turn a statistic into surveillance imagery. Capture only when a real incident # needs evidence, and never attach snapshots to the ergo report itself. evidential = [e for e in events if e.event_type != EventType.ERGO_REPORT] snaps = [] if evidential: # Evidence timeline, oldest→now: how the incident DEVELOPED, not just its end # state (a fall's value is in the seconds before the person hit the floor). # Right after startup the ring is short and fewer frames go out — never zero, # since the current frame was pushed above. Each frame blurs with its own boxes. frames, boxes = ring.timeline(now, SNAPSHOT_OFFSETS_S) snaps = snapshotter.capture(frames, boxes) for e in events: sink.try_send(e, [] if e.event_type == EventType.ERGO_REPORT else snaps) return events def build_pipeline(config: Config, source, pose) -> dict: return { "source": source, "pose": pose, "tracker": IouTracker(), "pool": TrackSmootherPool(), "engine": EventEngine( FallDetector(config.fall), ImmobilityWatchdog(config.immobility), GhostIncidentDetector(config.ghost), ZoneIntrusionDetector([], config.zone), config.engine, ), "ring": RingBuffer(), "snapshotter": Snapshotter(FaceBlurrer()), }