243 lines
13 KiB
Python
243 lines
13 KiB
Python
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, kruskal_permutation, load_face_series, load_trials, main, percentile, run_report,
|
|
split_label, _signflip_p,
|
|
)
|
|
from src.accuracy_session import SessionRunner, build_plan
|
|
|
|
TZ = timezone(timedelta(hours=9))
|
|
FPS = 30.0
|
|
|
|
|
|
class FakeTime:
|
|
def __init__(self, hour: int = 12) -> None:
|
|
self.now = datetime(2026, 10, 4, hour, 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,
|
|
hour: int = 12, seed: int = 0):
|
|
"""Run a fake session and write a matching face_frames CSV (30 fps, yaw ramps after each cue).
|
|
|
|
``hour`` shifts the wall clock so that several simulated subjects never overlap in time.
|
|
"""
|
|
tmp_path.mkdir(parents=True, exist_ok=True)
|
|
plan = build_plan([1, 3], [0, 5, -5, 10, -10], 3, seed=3)
|
|
fake, trials = FakeTime(hour), 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(seed)
|
|
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 / f"20261004_{hour:02d}0000_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"]
|
|
# one subject: no multi-subject sections or files
|
|
assert result["multi"] is None and "被験者別" not in result["markdown"]
|
|
assert not (tmp_path / "out" / "accuracy_loso.csv").exists()
|
|
|
|
|
|
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
|
|
|
|
|
|
# ---------------------------------------------------------------------------------------------
|
|
# multiple subjects
|
|
# ---------------------------------------------------------------------------------------------
|
|
def _two_subjects(tmp_path: Path, scale_b: float = 0.80):
|
|
ta, fa = _simulate(tmp_path / "a", scale=0.95, offset=-19.0, hour=12)
|
|
tb, fb = _simulate(tmp_path / "b", scale=scale_b, offset=-5.0, hour=14, seed=1)
|
|
return ta, fa, tb, fb
|
|
|
|
|
|
def test_split_label() -> None:
|
|
assert split_label("A=trials_a.csv") == ("A", "trials_a.csv")
|
|
assert split_label(r"C:\data\x=1.csv") == ("", r"C:\data\x=1.csv")
|
|
assert split_label("trials.csv") == ("", "trials.csv")
|
|
|
|
|
|
def test_subject_labels_aliases_and_exclusions(tmp_path: Path) -> None:
|
|
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"
|
|
a, b = tmp_path / "a.csv", tmp_path / "b.csv"
|
|
a.write_text(header + "S,1,1,c,1,10,0,1,S01,ok,a,b,c\nS,2,1,c,1,10,0,2,S01,ok,a,b,c\n", encoding="utf-8-sig")
|
|
b.write_text(header + "T,1,1,c,1,10,0,1,,ok,a,b,c\nT,2,1,c,1,10,0,2,,ok,a,b,c\n", encoding="utf-8-sig")
|
|
assert {r["subject"] for r in load_trials([a, b])} == {"S01", "b"} # column, then file name
|
|
rows = load_trials([a, b], labels={b: "S02"}, aliases={"S01": "A"})
|
|
assert {r["subject"] for r in rows} == {"A", "S02"} and len(rows) == 4
|
|
kept = load_trials([a, b], exclude_seqs=["S02:2", 1], labels={b: "S02"})
|
|
assert [(r["subject"], r["seq"]) for r in kept] == [("S01", "2")]
|
|
only = load_trials([a, b], exclude_seqs=["S02:2"], labels={b: "S02"})
|
|
assert [(r["subject"], r["seq"]) for r in only] == [("S01", "1"), ("S01", "2"), ("S02", "1")]
|
|
|
|
|
|
def test_multi_subject_pools_frames_and_reports_each_subject(tmp_path: Path) -> None:
|
|
ta, fa, tb, fb = _two_subjects(tmp_path)
|
|
result = run_report([ta, tb], [fa, fb], tmp_path / "out", Options(), plots=False, labels={ta: "A", tb: "B"})
|
|
assert result["subjects"] == ["A", "B"] and len(result["records"]) == 60
|
|
assert all(r["status"] == "ok" for r in result["records"])
|
|
multi = result["multi"]
|
|
per = multi["per_subject"]
|
|
assert abs(per["A"]["thresholds"][1.0]["slope"] - 0.95) < 0.05
|
|
assert abs(per["B"]["thresholds"][1.0]["slope"] - 0.80) < 0.05
|
|
pooled = result["thresholds"][1.0]
|
|
assert pooled["n_subjects"] == 2 and 0.80 < pooled["slope"] < 0.95
|
|
cond = {(c["distance_m"], c["true_yaw_deg"]): c for c in result["conditions"]}
|
|
assert cond[(1.0, 10.0)]["n_subjects"] == 2 and cond[(1.0, 10.0)]["n_trials"] == 6
|
|
assert cond[(1.0, 10.0)]["sd_subjects_deg"] > 0.5 # +10 deg reads 9.5 (A) vs 8.0 (B)
|
|
# leave-one-subject-out: one row per held-out subject and distance
|
|
loso = multi["loso"]
|
|
assert {(r["held_out"], r["distance_m"]) for r in loso} == {("A", 1.0), ("A", 3.0), ("B", 1.0), ("B", 3.0)}
|
|
assert all(r["n_train_subjects"] == 1 for r in loso)
|
|
# the +/- gain is symmetric in this simulation; the per-subject gains differ
|
|
asym = {r["subject"]: r for r in multi["asymmetry"]}
|
|
assert set(asym) == {"A", "B", "ALL"} and abs(asym["ALL"]["gain_diff"]) < 0.05
|
|
assert asym["A"]["gain_pos"] > asym["B"]["gain_pos"] + 0.05
|
|
assert multi["differences"] and all(r["p_permutation"] < 0.05 for r in multi["differences"])
|
|
for name in ("accuracy_subject_conditions.csv", "accuracy_subject_thresholds.csv", "accuracy_loso.csv",
|
|
"accuracy_asymmetry.csv", "accuracy_asymmetry_cells.csv", "accuracy_subject_differences.csv"):
|
|
assert (tmp_path / "out" / name).is_file(), name
|
|
header = (tmp_path / "out" / "accuracy_conditions.csv").read_text(encoding="utf-8-sig").splitlines()[0]
|
|
assert header.endswith("exceed_rate,n_subjects,sd_subjects_deg")
|
|
for text in ("被験者別しきい値", "leave-one-subject-out", "正負方向の非対称", "A (30試行)", "B (30試行)"):
|
|
assert text in result["markdown"]
|
|
|
|
|
|
def test_flip_only_the_named_subject(tmp_path: Path) -> None:
|
|
ta, fa, tb, fb = _two_subjects(tmp_path, scale_b=-0.95)
|
|
labels = {ta: "A", tb: "B"}
|
|
raw = run_report([ta, tb], [fa, fb], tmp_path / "o1", Options(), plots=False, labels=labels)
|
|
assert raw["multi"]["per_subject"]["B"]["thresholds"][1.0]["slope"] < 0
|
|
assert "--flip-subject B" in raw["markdown"]
|
|
fixed = run_report([ta, tb], [fa, fb], tmp_path / "o2", Options(flip_subjects=("B",)), plots=False, labels=labels)
|
|
per = fixed["multi"]["per_subject"]
|
|
assert per["A"]["thresholds"][1.0]["slope"] > 0.9 and per["B"]["thresholds"][1.0]["slope"] > 0.9
|
|
assert "--flip-subject B" not in fixed["markdown"]
|
|
only_a = run_report([ta, tb], [fa, fb], tmp_path / "o3", Options(), plots=False, labels=labels,
|
|
exclude_subjects=["B"])
|
|
assert only_a["subjects"] == ["A"] and only_a["multi"] is None
|
|
|
|
|
|
def test_cli_multi_subject_with_labels_and_subject_plots(tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None:
|
|
ta, fa, tb, fb = _two_subjects(tmp_path)
|
|
out = tmp_path / "report"
|
|
assert main(["--trials", f"A={ta}", f"B={tb}", "--face-csv", str(fa), str(fb), "--out", str(out)]) == 0
|
|
assert "被験者別しきい値" in capsys.readouterr().out
|
|
for name in ("accuracy_scatter.png", "accuracy_scatter_by_subject.png", "accuracy_error_by_subject.png",
|
|
"accuracy_gain_by_side.png"):
|
|
assert (out / "plots" / name).is_file(), name
|
|
assert main(["--trials", f"A={ta}", "--face-csv", str(fa), "--alias", "broken"]) == 2
|
|
|
|
|
|
def test_permutation_helpers() -> None:
|
|
assert _signflip_p([1.0, 1.0, 1.0, 1.0]) == pytest.approx(2 / 16)
|
|
assert math.isnan(_signflip_p([0.0, 0.0, 0.0]))
|
|
h, p = kruskal_permutation([[1.0, 2.0, 3.0], [4.0, 5.0, 6.0]])
|
|
assert h == pytest.approx(3.857, abs=0.01) and 0.05 < p < 0.2
|