7a86545367
- P3-1 [H] _head_boxes sizes the anonymization region from the person bbox (scales with distance) + bbox-top fallback when the nose keypoint is low-confidence; a fixed ~40px patch had left close faces identifiable. test_faceblur.py. - P3-3 [H] fetch_movenet.py rewritten to stage-validate (reject HTML/LFS-pointer, optional pinned sha256+size, fail loud) per the SHA256 spec requirement. Discovery: the old PINTO raw URL is DEAD (GitHub 404 → it was silently saving an HTML page as the .onnx). Non-blocking for the Hailo delivery path; the CPU MoveNet fallback needs a valid MOVENET_ONNX_URL pinned. python + phpunit suites green. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_019Xmn8KZWqiQUguEMRtn6rW
70 lines
2.5 KiB
Python
70 lines
2.5 KiB
Python
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
|
|
|
|
|
|
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):
|
|
ok, frame = source.read()
|
|
if not ok:
|
|
return []
|
|
now = now_fn()
|
|
persons = pose.infer(frame)
|
|
tracks = tracker.update(persons, now) # RAW keypoints; smoothing happens per-track below
|
|
lost = tracker.lost()
|
|
for tr in tracks:
|
|
f = tr.latest
|
|
f.keypoints = pool.smoother_for(tr.track_id).smooth(f.keypoints, now)
|
|
ring.push(now, frame)
|
|
h, w = frame.shape[:2]
|
|
events = engine.process(tracks, lost, now, frame_wh=(w, h))
|
|
for tr in lost:
|
|
pool.forget(tr.track_id)
|
|
if events:
|
|
snaps = snapshotter.capture([frame], [_head_boxes(persons)])
|
|
for e in events:
|
|
sink.try_send(e, 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()),
|
|
}
|