From 01c7f4636eec5557e9e84dca9182fe4de9f85964 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=EC=9C=A0=EC=B0=BD=EC=9A=B1?= Date: Sat, 11 Jul 2026 22:14:03 +0900 Subject: [PATCH] fix(edge): smooth keypoints per-track after tracking, not before (FIX 1) 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. --- edge/app.py | 16 +++++++----- edge/smoothing.py | 18 +++++++++++++ tests/test_app.py | 56 ++++++++++++++++++++++++++++++++++++++--- tests/test_smoothing.py | 24 +++++++++++++++++- 4 files changed, 103 insertions(+), 11 deletions(-) diff --git a/edge/app.py b/edge/app.py index aad12b4..15ed491 100644 --- a/edge/app.py +++ b/edge/app.py @@ -1,13 +1,13 @@ from edge.config import Config from edge.tracking import IouTracker -from edge.smoothing import KeypointSmoother +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 Person, BBox +from edge.models import BBox def _head_boxes(persons: list) -> list: @@ -19,17 +19,21 @@ def _head_boxes(persons: list) -> list: return boxes -def run_once(source, pose, tracker, smoother, engine, ring, snapshotter, sink, now_fn): +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) - smoothed = [Person(p.bbox, smoother.smooth(p.keypoints, now)) for p in persons] - tracks = tracker.update(smoothed, now) + 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: @@ -42,7 +46,7 @@ def build_pipeline(config: Config, source, pose) -> dict: "source": source, "pose": pose, "tracker": IouTracker(), - "smoother": KeypointSmoother(), + "pool": TrackSmootherPool(), "engine": EventEngine( FallDetector(config.fall), ImmobilityWatchdog(config.immobility), diff --git a/edge/smoothing.py b/edge/smoothing.py index e9c2ff3..053caa5 100644 --- a/edge/smoothing.py +++ b/edge/smoothing.py @@ -47,3 +47,21 @@ class KeypointSmoother: out.append(Keypoint(self._fx[i].filter(k.x, t), self._fy[i].filter(k.y, t), k.conf)) return out + + +class TrackSmootherPool: + """트랙별 독립 KeypointSmoother 풀. 다인 장면에서 필터 상태가 섞이지 않도록 + 트랙 ID마다 전용 1€ 필터 세트를 유지한다.""" + + def __init__(self, min_cutoff: float = 1.0, beta: float = 0.007): + self._min_cutoff = min_cutoff + self._beta = beta + self._pool: dict[int, KeypointSmoother] = {} + + def smoother_for(self, track_id: int) -> KeypointSmoother: + if track_id not in self._pool: + self._pool[track_id] = KeypointSmoother(self._min_cutoff, self._beta) + return self._pool[track_id] + + def forget(self, track_id: int) -> None: + self._pool.pop(track_id, None) diff --git a/tests/test_app.py b/tests/test_app.py index 76f650d..8f9d855 100644 --- a/tests/test_app.py +++ b/tests/test_app.py @@ -1,7 +1,7 @@ import numpy as np from edge.models import Person, BBox, Keypoint from edge.tracking import IouTracker -from edge.smoothing import KeypointSmoother +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 @@ -32,7 +32,7 @@ class _Sink: def test_run_once_pipeline_emits_fall_after_prone(): src, pose = _Source(), _Pose() tracker = IouTracker() - smoother = KeypointSmoother() + pool = TrackSmootherPool() engine = EventEngine( FallDetector(FallConfig(prone_confirm_s=0.5)), ImmobilityWatchdog(ImmobilityConfig(immobile_s=999)), @@ -43,6 +43,54 @@ def test_run_once_pipeline_emits_fall_after_prone(): ring = RingBuffer(); snap = Snapshotter(FaceBlurrer()); sink = _Sink() times = iter([0.0, 0.6]) def now_fn(): return next(times) - run_once(src, pose, tracker, smoother, engine, ring, snap, sink, now_fn) - run_once(src, pose, tracker, smoother, engine, ring, snap, sink, now_fn) + 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 diff --git a/tests/test_smoothing.py b/tests/test_smoothing.py index f883420..b3fd9c6 100644 --- a/tests/test_smoothing.py +++ b/tests/test_smoothing.py @@ -1,5 +1,5 @@ from edge.models import Keypoint -from edge.smoothing import OneEuroFilter, KeypointSmoother +from edge.smoothing import OneEuroFilter, KeypointSmoother, TrackSmootherPool def test_one_euro_first_sample_returns_input(): @@ -20,3 +20,25 @@ def test_keypoint_smoother_preserves_conf_and_length(): 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