静止時のyaw値変動に対して平滑化を行うための計測基盤を実装

tests/test_yaw_noise_report.py

@@ -0,0 +1,80 @@
+import csv
+from pathlib import Path
+
+import pytest
+
+from src.config_loader import FaceLoggingSettings
+from src.face_logger import FaceLogger
+from src.models import BoundingBox, FaceROI, HeadPoseResult
+from src.yaw_noise_report import (
+    ema, load_tracks, main, median_then_ema, one_euro, percentile, summarize,
+)
+
+
+def test_percentile_interpolates() -> None:
+    assert percentile([0.0, 10.0], 50) == pytest.approx(5.0)
+    assert percentile([1.0, 2.0, 3.0], 0) == 1.0
+    assert percentile([1.0, 2.0, 3.0], 100) == 3.0
+
+
+def test_summarize_constant_and_alternating_series() -> None:
+    constant = summarize([2.0] * 5)
+    assert constant is not None and constant["sd"] == 0.0 and constant["p2p"] == 0.0
+    alternating = summarize([0.0, 2.0, 0.0, 2.0])
+    assert alternating is not None
+    assert alternating["p2p"] == 2.0 and alternating["mean"] == 1.0
+    assert summarize([1.0]) is None
+
+
+def test_filters_preserve_length_and_constant_signal() -> None:
+    values = [3.0] * 6
+    samples = [(i * 0.05, 3.0) for i in range(6)]
+    assert ema(values, 0.4) == pytest.approx(values)
+    assert median_then_ema(values, 0.4) == pytest.approx(values)
+    assert one_euro(samples, 1.0, 0.0) == pytest.approx(values)
+    assert ema([1.0, 5.0], 1.0) == [1.0, 5.0]
+
+
+def test_ema_reduces_alternating_noise() -> None:
+    noisy = [0.0, 4.0] * 20
+    raw_sd = summarize(noisy)["sd"]  # type: ignore[index]
+    assert summarize(ema(noisy, 0.2))["sd"] < raw_sd  # type: ignore[index]
+
+
+def _result(frame: int, t: float, yaw: float, raw: float | None) -> HeadPoseResult:
+    roi = FaceROI(1, 10, 10, 110, 110, BoundingBox(0, 0, 200, 400), .9)
+    return HeadPoseResult(1, frame, "now", t, True, True, 0.0, yaw, 0.0, 468, roi, .9,
+                          raw_yaw=raw, reprojection_error=1.5, interocular_px=60.0)
+
+
+def test_logger_writes_diagnostic_columns_and_report_reads_them(tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None:
+    logger = FaceLogger(FaceLoggingSettings(True, True, True, 1, str(tmp_path)), "exp", "20260101_000000")
+    logger.open()
+    raws = [0.0, 2.0, -2.0, 1.0, -1.0, 0.5]
+    logger.write_results([_result(i + 1, i * 0.1, r * 0.5, r) for i, r in enumerate(raws)])
+    logger.close()
+    path = next(tmp_path.glob("*face_frames.csv"))
+    with path.open(encoding="utf-8-sig", newline="") as stream:
+        rows = list(csv.DictReader(stream))
+    assert rows[1]["raw_yaw"] == "2.000000" and rows[1]["interocular_px"] == "60.000000"
+    assert rows[0]["bbox_center_x"] == "100" and rows[0]["reprojection_error"] == "1.500000"
+
+    tracks = load_tracks(path)
+    assert len(tracks[1]["raw"]) == 6
+    trimmed = load_tracks(path, track_id=1, t_start=0.2)
+    assert len(trimmed[1]["raw"]) == 4
+    assert main([str(path), "--track-id", "1", "--csv", str(tmp_path / "out.csv")]) == 0
+    assert "raw + OneEuro" in capsys.readouterr().out
+    assert (tmp_path / "out.csv").is_file()
+
+
+def test_report_handles_old_logs_without_raw_columns(tmp_path: Path) -> None:
+    path = tmp_path / "old_face_frames.csv"
+    path.write_text(
+        "track_id,pose_estimated,elapsed_time_sec,yaw\n1,True,0.0,1.0\n1,True,0.1,2.0\n1,False,0.2,\n",
+        encoding="utf-8-sig",
+    )
+    tracks = load_tracks(path)
+    assert len(tracks[1]["logged"]) == 2 and tracks[1]["raw"] == []
+    assert main([str(path)]) == 0
+    assert main([str(tmp_path / "missing.csv")]) == 2