From 991178a896b85229d814012f05e7ec5df9ff9ab4 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?=EC=9C=A0=EC=B0=BD=EC=9A=B1?= <changukyu@gmail.com>
Date: Sat, 11 Jul 2026 20:23:21 +0900
Subject: [PATCH] feat(edge): One-Euro keypoint smoothing (INT8 jitter)

---
 edge/smoothing.py       | 49 +++++++++++++++++++++++++++++++++++++++++
 tests/test_smoothing.py | 22 ++++++++++++++++++
 2 files changed, 71 insertions(+)
 create mode 100644 edge/smoothing.py
 create mode 100644 tests/test_smoothing.py

diff --git a/edge/smoothing.py b/edge/smoothing.py
new file mode 100644
index 0000000..e9c2ff3
--- /dev/null
+++ b/edge/smoothing.py
@@ -0,0 +1,49 @@
+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
diff --git a/tests/test_smoothing.py b/tests/test_smoothing.py
new file mode 100644
index 0000000..f883420
--- /dev/null
+++ b/tests/test_smoothing.py
@@ -0,0 +1,22 @@
+from edge.models import Keypoint
+from edge.smoothing import OneEuroFilter, KeypointSmoother
+
+
+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