中間発表の精度実験を効率化するためのコードおよびテストを追加

tests/test_accuracy_report.py

@@ -0,0 +1,139 @@
+import csv
+import math
+import random
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+
+import pytest
+
+from src.accuracy_report import (
+    Options, evaluate_trial, load_face_series, load_trials, main, percentile, run_report,
+)
+from src.accuracy_session import SessionRunner, build_plan
+
+TZ = timezone(timedelta(hours=9))
+FPS = 30.0
+
+
+class FakeTime:
+    def __init__(self) -> None:
+        self.now = datetime(2026, 10, 4, 12, 0, 0, tzinfo=TZ)
+
+    def sleep(self, seconds: float) -> None:
+        self.now += timedelta(seconds=seconds)
+
+    def clock(self) -> str:
+        return self.now.isoformat(timespec="milliseconds")
+
+
+def _simulate(tmp_path: Path, scale: float = 0.95, offset: float = -19.0, noise: float = 0.5, face_dropout: float = 0.0):
+    """Run a fake session and write a matching face_frames CSV (30 fps, yaw ramps after each cue)."""
+    plan = build_plan([1, 3], [0, 5, -5, 10, -10], 3, seed=3)
+    fake, trials = FakeTime(), tmp_path / "trials.csv"
+    start = fake.now
+
+    def operator_pause(_prompt: str) -> str:
+        fake.sleep(4.0)  # time the subject needs to return and the operator needs to press Enter
+        return ""
+
+    SessionRunner(plan, trials, "S", input_fn=operator_pause, out=lambda _m: None, sleep=fake.sleep, clock=fake.clock,
+                  beep=lambda _k: None).run()
+    end = fake.now + timedelta(seconds=2)
+    rng = random.Random(0)
+    marks = [(datetime.fromisoformat(r["t_turn_cue"]), float(r["true_yaw_deg"]), datetime.fromisoformat(r["t_hold_end"]))
+             for r in csv.DictReader(trials.open(encoding="utf-8-sig"))]
+    path = tmp_path / "20261004_120000_exp_face_frames.csv"
+    with path.open("w", encoding="utf-8-sig", newline="") as stream:
+        writer = csv.writer(stream)
+        writer.writerow(["frame_number", "timestamp", "track_id", "pose_estimated", "yaw", "raw_yaw"])
+        t, frame = start - timedelta(seconds=1), 0
+        while t < end:
+            frame += 1
+            true = 0.0
+            for cue, angle, stop in marks:
+                if cue <= t <= stop:
+                    true = angle * min(1.0, (t - cue).total_seconds() / 0.8)
+            yaw = offset + scale * true + rng.gauss(0, noise)
+            ok = rng.random() >= face_dropout
+            writer.writerow([frame, t.isoformat(timespec="milliseconds"), 1, ok, f"{yaw:.4f}" if ok else "", f"{yaw:.4f}" if ok else ""])
+            t += timedelta(seconds=1 / FPS)
+    return trials, path
+
+
+def test_percentile() -> None:
+    assert percentile([0.0, 10.0], 50) == 5.0 and math.isnan(percentile([], 50))
+
+
+def test_full_pipeline_recovers_scale_bias_and_threshold(tmp_path: Path) -> None:
+    trials, face = _simulate(tmp_path)
+    result = run_report([trials], [face], tmp_path / "out", Options(), plots=False)
+    records = result["records"]
+    assert len(records) == 30 and all(r["status"] == "ok" for r in records)
+    # a constant -19 deg offset must cancel; 0.95 scale makes +10 read about 9.5 -> bias about -0.5
+    cond = {(c["distance_m"], c["true_yaw_deg"]): c for c in result["conditions"]}
+    assert abs(cond[(1.0, 0.0)]["bias_deg"]) < 0.3
+    assert abs(cond[(3.0, 10.0)]["bias_deg"] + 0.5) < 0.3
+    th = result["thresholds"][1.0]
+    assert 0.95 - 0.05 < th["slope"] < 0.95 + 0.05 and th["r2"] > 0.99
+    assert 1.0 < th["theta_th_deg"] < 3.5
+    assert th["theta_worst_deg"] >= th["theta_pooled_deg"] - 1e-9
+    assert th["min_detectable_deg"] == 5.0           # 5 deg turns exceed theta_th in >= 95 % of frames
+    assert cond[(1.0, 0.0)]["exceed_rate"] < 0.05    # false-positive rate at 0 deg
+    assert cond[(1.0, 10.0)]["mean_detection_rate"] > 0.95
+    for name in ("accuracy_trials.csv", "accuracy_conditions.csv", "accuracy_thresholds.csv", "accuracy_report.md"):
+        assert (tmp_path / "out" / name).is_file()
+    assert "θ_th" in result["markdown"]
+
+
+def test_flip_sign_is_flagged_by_negative_slope(tmp_path: Path) -> None:
+    trials, face = _simulate(tmp_path, scale=-1.0)
+    result = run_report([trials], [face], tmp_path / "out", Options(), plots=False)
+    assert result["thresholds"][1.0]["slope"] < 0 and "--flip-sign" in result["markdown"]
+    fixed = run_report([trials], [face], tmp_path / "out2", Options(flip_sign=True), plots=False)
+    assert fixed["thresholds"][1.0]["slope"] > 0.9
+
+
+def test_dropout_lowers_detection_rate_and_missing_data_is_excluded(tmp_path: Path) -> None:
+    trials, face = _simulate(tmp_path, face_dropout=0.3)
+    result = run_report([trials], [face], tmp_path / "out", Options(), plots=False)
+    rates = [r["detection_rate"] for r in result["records"] if r["status"] == "ok"]
+    assert 0.55 < sum(rates) / len(rates) < 0.85
+    # a face log that ends early leaves later trials without any track
+    series = load_face_series([face])
+    cut = series.times[len(series.times) // 2]
+    short = tmp_path / "short_face_frames.csv"
+    with face.open(encoding="utf-8-sig", newline="") as src, short.open("w", encoding="utf-8-sig", newline="") as dst:
+        reader, writer = csv.reader(src), csv.writer(dst)
+        for index, row in enumerate(reader):
+            if index == 0 or datetime.fromisoformat(row[1]).timestamp() <= cut:
+                writer.writerow(row)
+    partial = run_report([trials], [short], tmp_path / "out3", Options(), plots=False)
+    statuses = {r["status"] for r in partial["records"]}
+    assert "no_track" in statuses and "ok" in statuses
+
+
+def test_last_row_wins_and_exclude_seq(tmp_path: Path) -> None:
+    path = tmp_path / "t.csv"
+    header = "session_id,seq,block,condition_id,distance_m,true_yaw_deg,pitch_deg,repeat,subject,status,t_baseline_start,t_turn_cue,t_hold_end\n"
+    path.write_text(header + "S,1,1,c,1,10,0,1,,ok,a,b,c\nS,1,1,c,1,10,0,1,,discarded,,,\nS,2,1,c,1,10,0,2,,ok,a,b,c\n"
+                    "S,3,1,c,1,10,0,3,,ok,a,b,c\n", encoding="utf-8-sig")
+    assert [r["seq"] for r in load_trials([path])] == ["2", "3"]
+    assert [r["seq"] for r in load_trials([path], exclude_seqs=[3])] == ["2"]
+
+
+def test_bad_timestamps_do_not_crash(tmp_path: Path) -> None:
+    trials, face = _simulate(tmp_path)
+    series = load_face_series([face])
+    record, deltas = evaluate_trial({"seq": "1", "condition_id": "x", "distance_m": "1", "true_yaw_deg": "5",
+                                     "pitch_deg": "0", "t_baseline_start": "", "t_turn_cue": "", "t_hold_end": ""},
+                                    series, Options())
+    assert record["status"] == "bad_timestamps" and deltas == []
+
+
+def test_cli_end_to_end_with_glob_and_plots(tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None:
+    trials, face = _simulate(tmp_path)
+    out = tmp_path / "report"
+    assert main(["--trials", str(trials), "--face-csv", str(tmp_path / "*_face_frames.csv"), "--out", str(out)]) == 0
+    assert "しきい値" in capsys.readouterr().out
+    assert (out / "plots" / "accuracy_scatter.png").is_file()  # matplotlib is in requirements.txt
+    assert main(["--trials", str(trials), "--face-csv", str(tmp_path / "missing.csv")]) == 2

tests/test_accuracy_session.py

@@ -0,0 +1,99 @@
+import csv
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+
+from src.accuracy_session import (
+    SessionRunner, build_plan, condition_id, load_done, main, read_plan, write_plan,
+)
+
+
+class FakeTime:
+    def __init__(self) -> None:
+        self.now = datetime(2026, 10, 4, 12, 0, 0, tzinfo=timezone(timedelta(hours=9)))
+
+    def sleep(self, seconds: float) -> None:
+        self.now += timedelta(seconds=seconds)
+
+    def clock(self) -> str:
+        return self.now.isoformat(timespec="milliseconds")
+
+
+def _runner(plan, path, answers, **kwargs):
+    fake, queue, log = FakeTime(), list(answers), []
+    runner = SessionRunner(plan, path, "S", input_fn=lambda _p: queue.pop(0) if queue else "q", out=log.append,
+                           sleep=fake.sleep, clock=fake.clock, beep=lambda _k: None, **kwargs)
+    return runner, log
+
+
+def _rows(path: Path):
+    with path.open(encoding="utf-8-sig", newline="") as stream:
+        return list(csv.DictReader(stream))
+
+
+def test_condition_id_has_no_negative_zero() -> None:
+    assert condition_id(2.0, -0.0, 0.0) == "d2_y+0_p+0"
+    assert condition_id(1.5, -10.0, 10.0) == "d1.5_y-10_p+10"
+
+
+def test_plan_is_blocked_by_distance_and_reproducible() -> None:
+    a = build_plan([1, 2, 3], [0, 10, -10], 2, seed=7, subject="S01")
+    b = build_plan([1, 2, 3], [0, 10, -10], 2, seed=7, subject="S01")
+    assert a == b and len(a) == 3 * 3 * 2
+    assert [r.seq for r in a] == list(range(1, 19))
+    assert [r.block for r in a] == sorted(r.block for r in a)
+    assert {r.distance_m for r in a if r.block == 2} == {2.0}
+    assert build_plan([1, 2], [0, 10], 2, seed=8) != build_plan([1, 2], [0, 10], 2, seed=7)
+
+
+def test_pitch_conditions_only_at_selected_distances(tmp_path: Path) -> None:
+    plan = build_plan([1, 2], [0], 1, pitch_levels=[-10, 10], pitch_distances=[2], pitch_angles=[0, 10])
+    assert [r for r in plan if r.distance_m == 1 and r.pitch_deg != 0] == []
+    assert len([r for r in plan if r.distance_m == 2 and r.pitch_deg != 0]) == 4
+    write_plan(plan, tmp_path / "p.csv")
+    assert read_plan(tmp_path / "p.csv") == plan
+
+
+def test_runner_logs_ordered_timestamps_and_resume(tmp_path: Path) -> None:
+    plan = build_plan([1], [0, 10, -10], 1, seed=1)
+    path = tmp_path / "trials.csv"
+    runner, _ = _runner(plan, path, ["", "", "q"], baseline_sec=3.0, hold_sec=5.0)
+    assert runner.run() == 0
+    rows = _rows(path)
+    assert [r["status"] for r in rows] == ["ok", "ok"]
+    for row in rows:
+        t0, t1, t2 = (datetime.fromisoformat(row[k]) for k in ("t_baseline_start", "t_turn_cue", "t_hold_end"))
+        assert (t1 - t0).total_seconds() == 3.0 and (t2 - t1).total_seconds() == 5.0
+    assert load_done(path) == {plan[0].seq, plan[1].seq}
+    again, log = _runner(plan, path, [""])
+    assert again.run() == 2 and "already exists" in log[0]
+    resumed, _ = _runner(plan, path, [""])
+    assert resumed.run(resume=True) == 0
+    assert [r["seq"] for r in _rows(path)][-1] == str(plan[2].seq)
+    assert load_done(path) == {p.seq for p in plan}
+
+
+def test_redo_discard_skip_use_last_row_semantics(tmp_path: Path) -> None:
+    plan = build_plan([1], [0, 10, -10], 1, seed=1)
+    path = tmp_path / "t.csv"
+    # run trial 1, redo it, run it again, run trial 2, discard trial 2, skip trial 3
+    runner, _ = _runner(plan, path, ["", "r", "", "", "x", "s"])
+    runner.run()
+    statuses = [(r["seq"], r["status"]) for r in _rows(path)]
+    s1, s2, s3 = (str(p.seq) for p in plan)
+    assert statuses == [(s1, "ok"), (s1, "redo"), (s1, "ok"), (s2, "ok"), (s2, "discarded"), (s3, "skipped")]
+    assert load_done(path) == {plan[0].seq, plan[2].seq}
+
+
+def test_auto_interval_counts_down_without_enter(tmp_path: Path) -> None:
+    plan = build_plan([1], [0, 10], 1, seed=1)
+    runner, log = _runner(plan, tmp_path / "t.csv", [""], auto_interval=2.0)  # only the block-start Enter
+    assert runner.run() == 0
+    assert [r["status"] for r in _rows(tmp_path / "t.csv")] == ["ok", "ok"]
+    assert any("starting in" in line for line in log)
+
+
+def test_cli_plan_and_missing_plan(tmp_path: Path) -> None:
+    out = tmp_path / "plan.csv"
+    assert main(["plan", "--distances", "1,2", "--angles", "0,10", "--repeats", "2", "--seed", "1", "--out", str(out)]) == 0
+    assert len(read_plan(out)) == 8
+    assert main(["run", "--plan", str(tmp_path / "none.csv")]) == 2