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.
96 lines
3.8 KiB
Python
96 lines
3.8 KiB
Python
import numpy as np
|
|
from edge.models import Person, BBox, Keypoint
|
|
from edge.tracking import IouTracker
|
|
from edge.smoothing import TrackSmootherPool
|
|
from edge.detectors.fall import FallDetector, FallConfig
|
|
from edge.detectors.immobility import ImmobilityWatchdog, ImmobilityConfig
|
|
from edge.detectors.ghost import GhostIncidentDetector, GhostConfig
|
|
from edge.detectors.zone import ZoneIntrusionDetector, ZoneConfig
|
|
from edge.engine import EventEngine, EngineConfig
|
|
from edge.snapshot import RingBuffer, FaceBlurrer, Snapshotter
|
|
from edge.events import EventType
|
|
from edge.app import run_once
|
|
|
|
def _lying_person():
|
|
kps = [Keypoint(150, 300, 0.9) for _ in range(17)]
|
|
kps[5] = Keypoint(100, 300, 0.9); kps[6] = Keypoint(100, 310, 0.9)
|
|
kps[11] = Keypoint(200, 300, 0.9); kps[12] = Keypoint(200, 310, 0.9)
|
|
kps[0] = Keypoint(90, 320, 0.9)
|
|
return Person(BBox(90, 290, 210, 320), kps)
|
|
|
|
class _Source:
|
|
def read(self): return True, np.full((360, 640, 3), 128, np.uint8)
|
|
|
|
class _Pose:
|
|
def infer(self, frame): return [_lying_person()]
|
|
|
|
class _Sink:
|
|
def __init__(self): self.events = []
|
|
def try_send(self, event, snapshots):
|
|
self.events.append(event); return True
|
|
|
|
def test_run_once_pipeline_emits_fall_after_prone():
|
|
src, pose = _Source(), _Pose()
|
|
tracker = IouTracker()
|
|
pool = TrackSmootherPool()
|
|
engine = EventEngine(
|
|
FallDetector(FallConfig(prone_confirm_s=0.5)),
|
|
ImmobilityWatchdog(ImmobilityConfig(immobile_s=999)),
|
|
GhostIncidentDetector(GhostConfig()),
|
|
ZoneIntrusionDetector([], ZoneConfig()),
|
|
EngineConfig(camera_id="cam-01"),
|
|
)
|
|
ring = RingBuffer(); snap = Snapshotter(FaceBlurrer()); sink = _Sink()
|
|
times = iter([0.0, 0.6])
|
|
def now_fn(): return next(times)
|
|
run_once(src, pose, tracker, pool, engine, ring, snap, sink, now_fn)
|
|
run_once(src, pose, tracker, pool, engine, ring, snap, sink, now_fn)
|
|
assert any(e.event_type == EventType.FALL for e in sink.events)
|
|
|
|
|
|
def _person_at(cx):
|
|
# standing person centered at cx; nose at (cx, 60)
|
|
kps = [Keypoint(cx, 150, 0.9) for _ in range(17)]
|
|
kps[5] = Keypoint(cx - 10, 100, 0.9); kps[6] = Keypoint(cx + 10, 100, 0.9)
|
|
kps[11] = Keypoint(cx - 10, 200, 0.9); kps[12] = Keypoint(cx + 10, 200, 0.9)
|
|
kps[0] = Keypoint(cx, 60, 0.9)
|
|
return Person(BBox(cx - 20, 60, cx + 20, 210), kps)
|
|
|
|
|
|
class _TwoPersonPose:
|
|
def __init__(self, cx_a, cx_b):
|
|
self.cx_a, self.cx_b = cx_a, cx_b
|
|
|
|
def infer(self, frame):
|
|
return [_person_at(self.cx_a), _person_at(self.cx_b)]
|
|
|
|
|
|
def test_two_persons_get_independent_smoothing():
|
|
# person A far on the left, person B far on the right; small per-frame
|
|
# jitter so IoU tracking stays continuous across frames.
|
|
src = _Source()
|
|
pose = _TwoPersonPose(cx_a=100, cx_b=900)
|
|
tracker = IouTracker()
|
|
pool = TrackSmootherPool()
|
|
engine = EventEngine(
|
|
FallDetector(FallConfig(prone_confirm_s=999)),
|
|
ImmobilityWatchdog(ImmobilityConfig(immobile_s=999)),
|
|
GhostIncidentDetector(GhostConfig()),
|
|
ZoneIntrusionDetector([], ZoneConfig()),
|
|
EngineConfig(camera_id="cam-01"),
|
|
)
|
|
ring = RingBuffer(); snap = Snapshotter(FaceBlurrer()); sink = _Sink()
|
|
times = iter([0.0, 0.1])
|
|
def now_fn(): return next(times)
|
|
run_once(src, pose, tracker, pool, engine, ring, snap, sink, now_fn)
|
|
run_once(src, pose, tracker, pool, engine, ring, snap, sink, now_fn)
|
|
|
|
tracks = {tr.track_id: tr for tr in tracker._tracks.values()}
|
|
assert len(tracks) == 2
|
|
xs = sorted(tr.latest.keypoints[0].x for tr in tracks.values())
|
|
left_nose_x, right_nose_x = xs
|
|
# each track's smoothed nose must stay on its own side, nowhere near
|
|
# the midpoint between the two persons (500) — proves no shared filter
|
|
# state leaked between tracks.
|
|
assert left_nose_x < 300
|
|
assert right_nose_x > 700
|