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.
44 lines
1.4 KiB
Python
44 lines
1.4 KiB
Python
from edge.models import Keypoint
|
|
from edge.smoothing import OneEuroFilter, KeypointSmoother, TrackSmootherPool
|
|
|
|
|
|
def test_one_euro_first_sample_returns_input():
|
|
f = OneEuroFilter(min_cutoff=1.0, beta=0.0)
|
|
assert f.filter(5.0, t=0.0) == 5.0
|
|
|
|
|
|
def test_one_euro_attenuates_jitter():
|
|
f = OneEuroFilter(min_cutoff=0.5, beta=0.0)
|
|
f.filter(0.0, t=0.0)
|
|
out = f.filter(10.0, t=0.1) # big jump at low cutoff -> damped below input
|
|
assert 0.0 < out < 10.0
|
|
|
|
|
|
def test_keypoint_smoother_preserves_conf_and_length():
|
|
s = KeypointSmoother()
|
|
kps = [Keypoint(float(i), float(i), 0.8) for i in range(17)]
|
|
out = s.smooth(kps, t=0.0)
|
|
assert len(out) == 17
|
|
assert out[3].conf == 0.8
|
|
|
|
|
|
def test_pool_returns_same_smoother_instance_for_same_track():
|
|
pool = TrackSmootherPool()
|
|
assert pool.smoother_for(1) is pool.smoother_for(1)
|
|
|
|
|
|
def test_pool_gives_independent_smoothers_per_track():
|
|
pool = TrackSmootherPool()
|
|
assert pool.smoother_for(1) is not pool.smoother_for(2)
|
|
|
|
|
|
def test_pool_forget_drops_state_so_next_call_is_fresh():
|
|
pool = TrackSmootherPool()
|
|
s1 = pool.smoother_for(1)
|
|
s1.smooth([Keypoint(0.0, 0.0, 0.9) for _ in range(17)], t=0.0)
|
|
pool.forget(1)
|
|
s2 = pool.smoother_for(1)
|
|
assert s2 is not s1
|
|
# fresh filter: first sample passes through unfiltered regardless of prior state
|
|
out = s2.smooth([Keypoint(50.0, 50.0, 0.9) for _ in range(17)], t=1.0)
|
|
assert out[0].x == 50.0
|