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.
67 lines
2.4 KiB
Python
67 lines
2.4 KiB
Python
import math
|
|
|
|
from edge.models import Keypoint
|
|
|
|
|
|
class OneEuroFilter:
|
|
"""1€ filter (Casiez et al.). 저지연 지터 감쇠."""
|
|
|
|
def __init__(self, min_cutoff: float = 1.0, beta: float = 0.0, d_cutoff: float = 1.0):
|
|
self.min_cutoff = min_cutoff
|
|
self.beta = beta
|
|
self.d_cutoff = d_cutoff
|
|
self._x_prev: float | None = None
|
|
self._dx_prev: float = 0.0
|
|
self._t_prev: float | None = None
|
|
|
|
@staticmethod
|
|
def _alpha(cutoff: float, dt: float) -> float:
|
|
tau = 1.0 / (2 * math.pi * cutoff)
|
|
return 1.0 / (1.0 + tau / dt)
|
|
|
|
def filter(self, value: float, t: float) -> float:
|
|
if self._x_prev is None or self._t_prev is None or t <= self._t_prev:
|
|
self._x_prev, self._t_prev, self._dx_prev = value, t, 0.0
|
|
return value
|
|
dt = t - self._t_prev
|
|
dx = (value - self._x_prev) / dt
|
|
a_d = self._alpha(self.d_cutoff, dt)
|
|
dx_hat = a_d * dx + (1 - a_d) * self._dx_prev
|
|
cutoff = self.min_cutoff + self.beta * abs(dx_hat)
|
|
a = self._alpha(cutoff, dt)
|
|
x_hat = a * value + (1 - a) * self._x_prev
|
|
self._x_prev, self._dx_prev, self._t_prev = x_hat, dx_hat, t
|
|
return x_hat
|
|
|
|
|
|
class KeypointSmoother:
|
|
"""관절 17개 각각 x,y에 독립 1€ 필터."""
|
|
|
|
def __init__(self, min_cutoff: float = 1.0, beta: float = 0.007):
|
|
self._fx = [OneEuroFilter(min_cutoff, beta) for _ in range(17)]
|
|
self._fy = [OneEuroFilter(min_cutoff, beta) for _ in range(17)]
|
|
|
|
def smooth(self, keypoints: list[Keypoint], t: float) -> list[Keypoint]:
|
|
out = []
|
|
for i, k in enumerate(keypoints):
|
|
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)
|