From ca10876101a4901bcd2a912e1b769e2f3e888b79 Mon Sep 17 00:00:00 2001 From: KoyoNoguchi Date: Fri, 2 Oct 2026 16:45:21 +0900 Subject: [PATCH] =?UTF-8?q?=E5=86=8D=E6=8A=95=E5=BD=B1=E8=AA=A4=E5=B7=AE?= =?UTF-8?q?=E3=81=AB=E3=82=88=E3=82=8B=E7=A0=B4=E7=B6=BB=E3=83=95=E3=83=AC?= =?UTF-8?q?=E3=83=BC=E3=83=A0=E3=81=AE=E9=99=A4=E5=A4=96=E3=81=8A=E3=82=88?= =?UTF-8?q?=E3=81=B3=E7=AA=93=E3=81=AE=E6=99=82=E9=96=93=E6=8C=87=E5=AE=9A?= =?UTF-8?q?=E3=81=AE=E5=AE=9F=E8=A3=85=E3=81=A8=E5=AE=9F=E9=A8=93=E9=99=BD?= =?UTF-8?q?=E8=A8=AD=E5=AE=9A=E3=81=AE=E6=9B=B4=E6=96=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- config/default.yaml | 2 + config/experiment.yaml | 7 ++- docs/head_pose_smoothing.md | 35 +++++++++++- src/config_loader.py | 11 +++- src/face_processor.py | 1 + src/head_pose.py | 17 +++++- src/yaw_filters.py | 80 ++++++++++++++++++-------- src/yaw_noise_report.py | 4 ++ tests/test_head_pose_quality_gate.py | 84 ++++++++++++++++++++++++++++ tests/test_yaw_filters.py | 45 +++++++++++++++ 10 files changed, 256 insertions(+), 30 deletions(-) create mode 100644 tests/test_head_pose_quality_gate.py diff --git a/config/default.yaml b/config/default.yaml index 92ffb88..df2f235 100644 --- a/config/default.yaml +++ b/config/default.yaml @@ -98,6 +98,8 @@ head_pose: one_euro_beta: 0.0 # keep ~0 unless a median precedes it (e.g. 0.02) one_euro_d_cutoff: 1.0 filter_reset_gap_sec: 0.0 # reset filter state after a pose gap longer than this; 0 = never + smoothing_window_sec: 0.0 # >0: median-type windows span this many seconds (overrides smoothing_window) + max_reprojection_error_ratio: 0.0 # reject poses with reprojection error > ratio * eye distance [px]; 0 = off (normal frames ~0.07) face_logging: enabled: true log_frame_faces: true diff --git a/config/experiment.yaml b/config/experiment.yaml index e5f5e98..69a0634 100644 --- a/config/experiment.yaml +++ b/config/experiment.yaml @@ -38,8 +38,13 @@ head_pose: enabled: true smoothing_enabled: true smoothing_method: median_ema - ema_alpha: 0.4 + ema_alpha: 0.5 smoothing_window: 5 + # 3 m static data: median over 0.55 s + EMA 0.5 gave SD 1.85 deg (median5 + EMA 0.4: 3.50 deg) + # with about +0.04 s extra lag in a real turn. See docs/head_pose_smoothing.md. + smoothing_window_sec: 0.55 + # Poses with reprojection error > 0.3 x eye distance are degenerate; normal frames are ~0.07. + max_reprojection_error_ratio: 0.3 face_logging: enabled: true turn_detection: diff --git a/docs/head_pose_smoothing.md b/docs/head_pose_smoothing.md index a64163c..1040600 100644 --- a/docs/head_pose_smoothing.md +++ b/docs/head_pose_smoothing.md @@ -13,6 +13,8 @@ | `hampel_threshold` | 3.0 | Hampel判定のしきい値（頑健σの倍数） | | `one_euro_min_cutoff` / `one_euro_beta` / `one_euro_d_cutoff` | 1.0 / 0.0 / 1.0 | One Euro Filter のパラメータ | | `filter_reset_gap_sec` | 0.0 | 姿勢サンプルがこれより離れたらフィルタ状態をリセット（0で無効） | +| `smoothing_window_sec` | 0.0 | 0より大きいと、中央値系フィルタの窓を「直近N秒」で指定する（`smoothing_window` より優先）。更新レートが変わってもフィルタの強さが変わらない | +| `max_reprojection_error_ratio` | 0.0 | 再投影誤差が「目尻間距離×この値」を超えた姿勢を棄却（0で無効）。正常フレームの比率は約0.07、破綻フレームは0.4以上 | 既定値は従来動作（`ema`, α=0.4）と同じ。新キーはYAMLに書かなくてもよい（既定値が使われる）。 フィルタ本体は `src/yaw_filters.py`。 @@ -38,6 +40,31 @@ - 1か所・1距離・1人の静止データによる比較。実際の振り向き中の挙動は未検証。 - 現行（ema 0.4）の出力は、静止していても3σ=11.3°、最大変動幅28°あり、`subtle`(10°)・`weak`(20°)のしきい値に達しうる。 +## 追加計測（2026-10-02, 研究室, 正面・3m・静止＋移動, median5+EMA 0.4設定時のログ, 約23.1Hz） + +静止区間（フレーム250〜1390, 49.3秒）： + +| フィルタ | SD | 最大変動幅 | 平均から8°超の外れ | 階段90%遅れ | +|---|---:|---:|---:|---:| +| 生Yaw | 9.02° | 72.6° | - | 0 | +| ema 0.4（旧デフォルト） | 5.70° | 54.3° | 11.5% | 0.17秒 | +| median5 + ema 0.4 | 3.50° | 38.8° | 2.5% | 0.26秒 | +| median 0.45秒 + ema 0.5 | 2.27° | 31.0° | 0.5% | 0.35秒 | +| **median 0.55秒 + ema 0.5** | **1.85°** | **9.9°** | **0.0%** | 0.39秒 | + +- 実機の `yaw` 列は、生Yawにオフラインで同じフィルタをかけた結果と一致した（最大差 9e-7）。 +- 生Yawの+側スパイクは8.7%。連続長は1フレーム46回・2連続17回・3連続3回・4連続1回・6連続1回。 + 中央値は、窓n個で(n-1)/2連続までのスパイクを除去できる。median5（n=5）は2連続まで。 +- 移動しながら向く区間（フレーム218〜240）で、生Yawは-68.7°→-23.1°。90%到達は + median5+EMA0.4が0.76秒、旧デフォルトが0.67秒、median 0.55秒+EMA0.5が0.80秒（50%到達は0.31/0.09/0.40秒）。 + 窓を広げても、実際の振り向きでの追加の遅れは小さい。 +- 再投影誤差/目尻間距離の比は、正常フレームで p99 でも0.12〜0.19。0.4以上のフレームはすべて、 + 目尻間距離が20px以下の小さい顔（入退場の移動中）で、Yawも破綻していた（例: 比104.9でYaw -1.0°、比150でYaw -66.7°）。 + 比0.3で除外すると、静止区間のフレームは1つも除外されず、破綻フレームだけが落ちる。 + 静止時のスパイクとは相関しない（スパイク1.58px、その他1.57px）。 +- 静止中の中央値からの逸脱が10°を超えた状態の連続時間は、旧デフォルトで最長0.27秒、median5で0.22秒、 + median 0.55秒窓ではさらに短い。`min_turn_duration_sec`(0.3秒)との余裕は、窓を広げるほど大きくなる。 + ## 分かったこと - 生Yawの分散の大部分は、フレームごとの独立なノイズと、単発の+側スパイク（約5%）による。 @@ -58,6 +85,10 @@ `turn_detection.signage_yaw_direction` が実測に合わせて設定されていれば結果は一貫するが、 「正のYaw＝物理的にどちら向きか」を前提にした解釈・報告では注意が必要。 モデルを直す場合は符号が変わるので、`signage_yaw_direction` と過去ログの扱いも同時に見直すこと。 -- 正面を向いた静止時の平均が約-19.3°だった。立ち位置は画像中心から約31px（約1.4°）しか離れておらず、 - 位置では説明できない。カメラのレンズでなくプレビュー画面を見ていた可能性などを、レンズに目印を付けて再測定して確認する。 +- 正面を向いた静止時のYaw中央値が、2回の計測（2026-10-01, 10-02）とも約-20°だった（bbox中心はそれぞれ671px, 680px。 + 画像中心640pxから約30〜40px、約2°しか離れていない）。立ち位置では説明できない。 + レンズとプレビュー画面は物理的に離れているため、プレビューを見ていた可能性は除外されている。 + 原因は未解明（3Dモデル点と実際の顔の不一致、頭部の実際の向きなど）。 + 距離を変えて（例: 1.5m）測り、オフセットが距離で変わるか（小さい顔による見かけの偏り）、 + 変わらないか（系統的なずれ）を確認する。振り向き検出は基準Yawとの差を使うので、固定のオフセットは影響しない。 - 一部の設定値（`median_one_euro` の β など）は、この静止データのノイズ規模に合わせたもの。距離が変わったら再評価する。 diff --git a/src/config_loader.py b/src/config_loader.py index d3ca98a..d954cdc 100644 --- a/src/config_loader.py +++ b/src/config_loader.py @@ -212,6 +212,11 @@ class HeadPoseSettings: one_euro_d_cutoff: float = 1.0 # Reset filter state when pose samples are further apart than this [s]; 0 disables. filter_reset_gap_sec: float = 0.0 + # Median-type windows span this many seconds instead of smoothing_window samples; 0 = by count. + smoothing_window_sec: float = 0.0 + # Reject a pose when the solvePnP reprojection error exceeds this multiple of the + # outer-eye-corner distance (pixels); 0 disables. Normal frames are about 0.07. + max_reprojection_error_ratio: float = 0.0 @dataclass(frozen=True, slots=True) @@ -603,7 +608,8 @@ def _validate(config: AppConfig) -> None: if not 0.0 < pose.ema_alpha <= 1.0 or pose.max_abs_angle <= 0: raise ConfigurationError("head_pose EMA alpha and maximum angle are invalid") numeric = (pose.smoothing_window, pose.hampel_threshold, pose.one_euro_min_cutoff, pose.one_euro_beta, - pose.one_euro_d_cutoff, pose.filter_reset_gap_sec) + pose.one_euro_d_cutoff, pose.filter_reset_gap_sec, pose.smoothing_window_sec, + pose.max_reprojection_error_ratio) if any(isinstance(value, bool) or not isinstance(value, (int, float)) for value in numeric): raise ConfigurationError("head_pose filter settings must be numeric") if not isinstance(pose.smoothing_window, int) or pose.smoothing_window < 1: @@ -611,7 +617,8 @@ def _validate(config: AppConfig) -> None: if pose.smoothing_method == "hampel_ema" and pose.smoothing_window < 3: raise ConfigurationError("head_pose.smoothing_window must be >= 3 for hampel_ema") if (pose.hampel_threshold <= 0 or pose.one_euro_min_cutoff <= 0 or pose.one_euro_d_cutoff <= 0 - or pose.one_euro_beta < 0 or pose.filter_reset_gap_sec < 0): + or pose.one_euro_beta < 0 or pose.filter_reset_gap_sec < 0 or pose.smoothing_window_sec < 0 + or pose.max_reprojection_error_ratio < 0): raise ConfigurationError("head_pose filter settings are out of range") if config.face_logging.flush_interval_frames <= 0: raise ConfigurationError("face_logging.flush_interval_frames must be positive") diff --git a/src/face_processor.py b/src/face_processor.py index 94ccf78..d599d2a 100644 --- a/src/face_processor.py +++ b/src/face_processor.py @@ -144,6 +144,7 @@ class FaceProcessor: hampel_threshold=settings.hampel_threshold, one_euro_min_cutoff=settings.one_euro_min_cutoff, one_euro_beta=settings.one_euro_beta, one_euro_d_cutoff=settings.one_euro_d_cutoff, reset_gap_sec=settings.filter_reset_gap_sec, + window_sec=settings.smoothing_window_sec, ) for _ in range(3) ) diff --git a/src/head_pose.py b/src/head_pose.py index 343a85d..e150cf7 100644 --- a/src/head_pose.py +++ b/src/head_pose.py @@ -55,6 +55,8 @@ class HeadPoseEstimator: image_points = self.build_image_points(landmarks) if image_points is None: return HeadPoseEstimation(False, failure_reason="insufficient_landmarks") + # Outer eye corners (landmarks 33 and 263) are model points 2 and 3. + eye_distance = math.hypot(image_points[3][0] - image_points[2][0], image_points[3][1] - image_points[2][1]) try: import cv2 import numpy as np @@ -74,13 +76,24 @@ class HeadPoseEstimator: error = float(np.sqrt(np.mean(np.sum((projected.reshape(-1, 2) - image) ** 2, axis=1)))) except Exception: return HeadPoseEstimation(False, failure_reason="solvepnp_failed") - return self.result_from_angles(angles, error) + return self.result_from_angles(angles, error, eye_distance_px=eye_distance) def result_from_angles( self, angles: tuple[float, float, float], reprojection_error: float | None = None, + eye_distance_px: float | None = None, ) -> HeadPoseEstimation: - """Validate calculated angles and construct a normalized result.""" + """Validate calculated angles and construct a normalized result. + + When ``max_reprojection_error_ratio`` is set, a pose whose reprojection error + exceeds that multiple of ``eye_distance_px`` is rejected: it is a degenerate + solution (typically a very small or leaving face), not a noisy measurement. + """ normalized = self.normalize_angles(angles) + ratio = self.config.max_reprojection_error_ratio + if (ratio > 0.0 and reprojection_error is not None and eye_distance_px is not None + and eye_distance_px > 0.0 and reprojection_error > ratio * eye_distance_px): + return HeadPoseEstimation(False, *normalized, reprojection_error=reprojection_error, + failure_reason="reprojection_error_too_large") if any(not math.isfinite(value) or abs(value) > self.config.max_abs_angle for value in normalized): return HeadPoseEstimation(False, *normalized, reprojection_error=reprojection_error, failure_reason="angle_out_of_range") diff --git a/src/yaw_filters.py b/src/yaw_filters.py index e31e2af..b5f5add 100644 --- a/src/yaw_filters.py +++ b/src/yaw_filters.py @@ -25,8 +25,12 @@ Kinds put a median in front (``median_one_euro``). ``median_one_euro`` sliding median, then One Euro Filter. -Window sizes are counted in samples, so their duration depends on the pose update -rate (frame rate / ``face.process_every_n_frames``). +Median-type windows hold ``window`` samples by default, so their duration depends +on the pose update rate (frame rate / ``face.process_every_n_frames``). With +``window_sec > 0`` the window instead holds every sample from the last +``window_sec`` seconds (trimmed to an odd count), which keeps the filter strength +independent of the update rate. A median removes a spike run of at most +(n - 1) / 2 samples out of n. """ from __future__ import annotations @@ -44,6 +48,39 @@ HAMPEL_MIN_SCALE_DEG = 1.0 _MAD_TO_SIGMA = 1.4826 +class _SampleWindow: + """Recent samples bounded by count, or by age when ``duration_sec`` > 0.""" + + MAX_SAMPLES_IN_DURATION_MODE = 256 + + def __init__(self, size: int, duration_sec: float = 0.0) -> None: + if size < 1: + raise ValueError("window must be >= 1") + if duration_sec < 0.0: + raise ValueError("window_sec must be >= 0") + self._size = size + self._duration = duration_sec + self._items: deque[tuple[float, float]] = deque() + + def add(self, t: float, value: float) -> list[float]: + """Append a sample and return the values currently inside the window.""" + self._items.append((t, value)) + if self._duration > 0.0: + while len(self._items) > 1 and t - self._items[0][0] > self._duration: + self._items.popleft() + while len(self._items) > self.MAX_SAMPLES_IN_DURATION_MODE: + self._items.popleft() + values = [item[1] for item in self._items] + # An odd count makes the median an actual sample (and symmetric). + return values[1:] if len(values) % 2 == 0 else values + while len(self._items) > self._size: + self._items.popleft() + return [item[1] for item in self._items] + + def clear(self) -> None: + self._items.clear() + + class AngleFilter(Protocol): def update(self, value: float, t: float) -> float: ... @@ -76,15 +113,12 @@ class EmaFilter: class MedianEmaFilter: - def __init__(self, window: int, alpha: float) -> None: - if window < 1: - raise ValueError("window must be >= 1") - self._window: deque[float] = deque(maxlen=window) + def __init__(self, window: int, alpha: float, window_sec: float = 0.0) -> None: + self._window = _SampleWindow(window, window_sec) self._ema = EmaFilter(alpha) def update(self, value: float, t: float) -> float: - self._window.append(value) - return self._ema.update(statistics.median(self._window), t) + return self._ema.update(statistics.median(self._window.add(t, value)), t) def reset(self) -> None: self._window.clear() @@ -92,21 +126,21 @@ class MedianEmaFilter: class HampelEmaFilter: - def __init__(self, window: int, threshold: float, alpha: float) -> None: + def __init__(self, window: int, threshold: float, alpha: float, window_sec: float = 0.0) -> None: if window < 3: raise ValueError("hampel window must be >= 3") if threshold <= 0.0: raise ValueError("hampel threshold must be positive") self.threshold = threshold - self._window: deque[float] = deque(maxlen=window) + self._window = _SampleWindow(window, window_sec) self._ema = EmaFilter(alpha) def update(self, value: float, t: float) -> float: - self._window.append(value) + values = self._window.add(t, value) cleaned = value - if len(self._window) >= 3: - center = statistics.median(self._window) - mad = statistics.median(abs(item - center) for item in self._window) + if len(values) >= 3: + center = statistics.median(values) + mad = statistics.median(abs(item - center) for item in values) scale = max(_MAD_TO_SIGMA * mad, HAMPEL_MIN_SCALE_DEG) if abs(value - center) > self.threshold * scale: cleaned = center @@ -155,15 +189,14 @@ class OneEuroFilter: class MedianOneEuroFilter: - def __init__(self, window: int, min_cutoff: float, beta: float, d_cutoff: float = 1.0) -> None: - if window < 1: - raise ValueError("window must be >= 1") - self._window: deque[float] = deque(maxlen=window) + def __init__( + self, window: int, min_cutoff: float, beta: float, d_cutoff: float = 1.0, window_sec: float = 0.0, + ) -> None: + self._window = _SampleWindow(window, window_sec) self._euro = OneEuroFilter(min_cutoff, beta, d_cutoff) def update(self, value: float, t: float) -> float: - self._window.append(value) - return self._euro.update(statistics.median(self._window), t) + return self._euro.update(statistics.median(self._window.add(t, value)), t) def reset(self) -> None: self._window.clear() @@ -203,6 +236,7 @@ def create_angle_filter( one_euro_beta: float = 0.0, one_euro_d_cutoff: float = 1.0, reset_gap_sec: float = 0.0, + window_sec: float = 0.0, ) -> AngleFilter: """Build a filter by name; raises ``ValueError`` for unknown kinds.""" inner: AngleFilter @@ -211,13 +245,13 @@ def create_angle_filter( elif kind == "ema": inner = EmaFilter(alpha) elif kind == "median_ema": - inner = MedianEmaFilter(window, alpha) + inner = MedianEmaFilter(window, alpha, window_sec) elif kind == "hampel_ema": - inner = HampelEmaFilter(window, hampel_threshold, alpha) + inner = HampelEmaFilter(window, hampel_threshold, alpha, window_sec) elif kind == "one_euro": inner = OneEuroFilter(one_euro_min_cutoff, one_euro_beta, one_euro_d_cutoff) elif kind == "median_one_euro": - inner = MedianOneEuroFilter(window, one_euro_min_cutoff, one_euro_beta, one_euro_d_cutoff) + inner = MedianOneEuroFilter(window, one_euro_min_cutoff, one_euro_beta, one_euro_d_cutoff, window_sec) else: raise ValueError(f"unknown smoothing filter {kind!r}; expected one of {FILTER_KINDS}") return GapResettingFilter(inner, reset_gap_sec) if reset_gap_sec > 0.0 else inner diff --git a/src/yaw_noise_report.py b/src/yaw_noise_report.py index e27b0a4..9f4146c 100644 --- a/src/yaw_noise_report.py +++ b/src/yaw_noise_report.py @@ -175,6 +175,10 @@ def build_rows(entry: dict[str, list]) -> list[tuple[str, dict[str, float]]]: apply_filter("median_one_euro", raw, window=5, one_euro_min_cutoff=1.0, one_euro_beta=0.02)) add("raw + median5 + OneEuro mincut=1.5 beta=0.0", apply_filter("median_one_euro", raw, window=5, one_euro_min_cutoff=1.5, one_euro_beta=0.0)) + add("raw + median 0.55 s + EMA a=0.5", + apply_filter("median_ema", raw, window=5, alpha=0.5, window_sec=0.55)) + add("raw + median 0.45 s + EMA a=0.5", + apply_filter("median_ema", raw, window=5, alpha=0.5, window_sec=0.45)) return rows diff --git a/tests/test_head_pose_quality_gate.py b/tests/test_head_pose_quality_gate.py new file mode 100644 index 0000000..a358e8b --- /dev/null +++ b/tests/test_head_pose_quality_gate.py @@ -0,0 +1,84 @@ +from pathlib import Path +from typing import Any + +import pytest + +from src.config_loader import HeadPoseSettings, load_config +from src.exceptions import ConfigurationError +from src.face_processor import FaceProcessor +from src.head_pose import HeadPoseEstimator + +DEFAULT = Path(__file__).parents[1] / "config" / "default.yaml" + + +def build(head_pose: dict[str, Any] | None = None): + overrides: dict[str, Any] = {"face": {"enabled": False}} # no MediaPipe needed + if head_pose is not None: + overrides["head_pose"] = head_pose + return load_config(DEFAULT, None, overrides) + + +def estimator(ratio: float) -> HeadPoseEstimator: + return HeadPoseEstimator(HeadPoseSettings(True, "solvepnp", True, "ema", 0.4, 120.0, + max_reprojection_error_ratio=ratio)) + + +def test_new_options_default_to_the_previous_behaviour() -> None: + settings = build().head_pose + assert settings.smoothing_window_sec == 0.0 + assert settings.max_reprojection_error_ratio == 0.0 + + +@pytest.mark.parametrize("override", [{"smoothing_window_sec": -0.1}, {"max_reprojection_error_ratio": -1}, + {"smoothing_window_sec": True}, {"max_reprojection_error_ratio": "0.3"}]) +def test_invalid_new_options_are_rejected(override: dict[str, Any]) -> None: + with pytest.raises(ConfigurationError): + build(override) + + +def test_reprojection_gate_rejects_only_degenerate_solutions() -> None: + gate = estimator(0.3) + normal = gate.result_from_angles((-9.0, -20.0, 1.0), reprojection_error=1.6, eye_distance_px=24.0) + assert normal.success and normal.yaw == pytest.approx(-20.0) + degenerate = gate.result_from_angles((-9.0, -66.0, 1.0), reprojection_error=2261.0, eye_distance_px=15.0) + assert not degenerate.success + assert degenerate.failure_reason == "reprojection_error_too_large" + assert degenerate.reprojection_error == 2261.0 and degenerate.yaw == pytest.approx(-66.0) + # Scales with the face size: 12 px is fine for a 60 px eye distance, not for a 24 px one. + assert gate.result_from_angles((0.0, 0.0, 0.0), 12.0, 60.0).success + assert not gate.result_from_angles((0.0, 0.0, 0.0), 12.0, 24.0).success + # The limit itself is accepted (strictly greater is rejected); 0.5 * 24 is exact in floating point. + assert estimator(0.5).result_from_angles((0.0, 0.0, 0.0), 12.0, 24.0).success + assert not estimator(0.5).result_from_angles((0.0, 0.0, 0.0), 12.01, 24.0).success + + +def test_reprojection_gate_is_off_by_default_and_needs_both_inputs() -> None: + assert estimator(0.0).result_from_angles((0.0, 10.0, 0.0), 2261.0, 15.0).success + assert estimator(0.3).result_from_angles((0.0, 10.0, 0.0), 2261.0, None).success + assert estimator(0.3).result_from_angles((0.0, 10.0, 0.0), None, 24.0).success + assert estimator(0.3).result_from_angles((0.0, 10.0, 0.0)).success # original call form + + +def test_out_of_range_angles_are_still_reported_after_the_gate() -> None: + result = estimator(0.3).result_from_angles((0.0, 150.0, 0.0), 1.0, 24.0) + assert not result.success and result.failure_reason == "angle_out_of_range" + + +def _spike_series(run: int) -> list[float]: + return [-20.0] * 20 + [25.0] * run + [-20.0] * 20 + + +@pytest.mark.parametrize("dt", [0.043, 0.0565]) +def test_face_processor_time_window_removes_short_spike_runs(dt: float) -> None: + config = build({"smoothing_method": "median_ema", "ema_alpha": 0.5, "smoothing_window_sec": 0.55}) + proc = FaceProcessor(config.face, config.head_pose) + outputs = [proc._smooth(1, 0.0, value, 0.0, index * dt)[1] for index, value in enumerate(_spike_series(3))] + assert max(abs(value + 20.0) for value in outputs) < 1e-6 + count_based = processor_with({"smoothing_method": "median_ema", "ema_alpha": 0.5, "smoothing_window": 5}) + plain = [count_based._smooth(1, 0.0, value, 0.0, index * dt)[1] for index, value in enumerate(_spike_series(3))] + assert max(abs(value + 20.0) for value in plain) > 10.0 + + +def processor_with(head_pose: dict[str, Any]) -> FaceProcessor: + config = build(head_pose) + return FaceProcessor(config.face, config.head_pose) diff --git a/tests/test_yaw_filters.py b/tests/test_yaw_filters.py index e258aab..971a2cd 100644 --- a/tests/test_yaw_filters.py +++ b/tests/test_yaw_filters.py @@ -77,3 +77,48 @@ def test_reset_and_validation() -> None: create_angle_filter("hampel_ema", window=2) with pytest.raises(ValueError): create_angle_filter("one_euro", one_euro_min_cutoff=0.0) + + +def _spike_series(run: int) -> list[float]: + return [-20.0] * 15 + [25.0] * run + [-20.0] * 15 + + +def _max_dev(kind: str, values: list[float], dt: float, **params: float) -> float: + out = _run(kind, values, dt=dt, **params) + return max(abs(v + 20.0) for v in out) + + +@pytest.mark.parametrize("dt", [0.043, 0.0565]) +def test_time_window_removes_spike_runs_at_any_update_rate(dt: float) -> None: + # 0.45 s holds >= 7 samples at both ~23 Hz and ~17.7 Hz, enough for a 3-sample run. + assert _max_dev("median_ema", _spike_series(3), dt, alpha=0.5, window=5, window_sec=0.45) < 1e-6 + # A fixed 5-sample window cannot remove a 3-sample run. + assert _max_dev("median_ema", _spike_series(3), dt, alpha=0.5, window=5) > 10.0 + + +def test_time_window_adapts_its_sample_count_to_the_rate() -> None: + fast = create_angle_filter("median_ema", alpha=1.0, window=5, window_sec=0.45) + slow = create_angle_filter("median_ema", alpha=1.0, window=5, window_sec=0.45) + # 4-sample spike run: removed at 23 Hz (11-sample window) but not at ~17.7 Hz (7 samples). + fast_out = [fast.update(v, i * 0.043) for i, v in enumerate(_spike_series(4))] + slow_out = [slow.update(v, i * 0.0565) for i, v in enumerate(_spike_series(4))] + assert max(abs(v + 20.0) for v in fast_out) < 1e-6 + assert max(abs(v + 20.0) for v in slow_out) > 10.0 + + +@pytest.mark.parametrize("kind", ["median_ema", "hampel_ema", "median_one_euro"]) +def test_time_window_keeps_constant_signal_and_follows_steps(kind: str) -> None: + out = _run(kind, [0.0] * 20 + [30.0] * 60, dt=0.043, alpha=0.5, window=5, window_sec=0.45) + assert out[19] == pytest.approx(0.0, abs=1e-6) + assert out[-1] == pytest.approx(30.0, abs=0.5) + assert _run(kind, [7.0] * 20, dt=0.043, window_sec=0.45) == pytest.approx([7.0] * 20) + + +def test_time_window_validation_and_reset() -> None: + with pytest.raises(ValueError): + create_angle_filter("median_ema", window_sec=-0.1) + f = create_angle_filter("median_ema", alpha=1.0, window_sec=0.45, reset_gap_sec=1.0) + for i in range(10): + f.update(0.0, i * 0.043) + f.reset() + assert f.update(40.0, 5.0) == 40.0