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

@@ -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(persons, now)  # RAW keypoints; smoothing happens per-track below
-    tracks = tracker.update(smoothed, now)
     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),

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)

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, pool, 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)
     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

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