01c7f4636e
edge/app.py run_once used ONE shared KeypointSmoother across all persons BEFORE tracking assigned IDs. The smoother holds per-joint-index One-Euro filter state, so with >1 person in frame the filter state was shared and corrupted across different people, and same-timestamp calls within a frame would pass through unsmoothed while still clobbering the shared state. Multi-person scenes could get corrupted torso angles -> missed/false falls. Add TrackSmootherPool (edge/smoothing.py) that lazily allocates one KeypointSmoother per track_id. run_once now: tracks raw keypoints first, then smooths each track's latest frame in place using its own pooled smoother, and forgets a track's smoother when the track is lost (ties into FIX 5 lost-track cleanup). BEHAVIOR CHANGE: keypoints are now smoothed after tracking instead of before. Single-person pipelines are unaffected (still one smoother, one identity). Multi-person pipelines get correct, non-cross-contaminated smoothing. run_once signature: smoother param replaced with pool: TrackSmootherPool. build_pipeline now constructs "pool" instead of "smoother". Tests: tests/test_smoothing.py adds TrackSmootherPool identity/forget coverage; tests/test_app.py adds test_two_persons_get_independent_smoothing demonstrating each track's smoothed keypoints stay on its own side instead of being pulled toward the other person.
59 lines
2 KiB
Python
59 lines
2 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:
|
|
# 배경인물 머리영역 근사: 코 주변 작은 박스 (MVP; 실제는 얼굴 검출 포트로 교체)
|
|
boxes = []
|
|
for p in persons:
|
|
nose = p.keypoints[0]
|
|
boxes.append(BBox(nose.x - 20, nose.y - 25, nose.x + 20, nose.y + 15))
|
|
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)
|
|
events = engine.process(tracks, lost, now)
|
|
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()),
|
|
}
|