From 2afebcb041b12996d741cdf63c59c6db782fa9b3 Mon Sep 17 00:00:00 2001 From: KoyoNoguchi Date: Sun, 4 Oct 2026 23:30:37 +0900 Subject: [PATCH] =?UTF-8?q?=E4=B8=AD=E9=96=93=E7=99=BA=E8=A1=A8=E3=81=AE?= =?UTF-8?q?=E7=B2=BE=E5=BA=A6=E5=AE=9F=E9=A8=93=E3=82=92=E5=8A=B9=E7=8E=87?= =?UTF-8?q?=E5=8C=96=E3=81=99=E3=82=8B=E3=81=9F=E3=82=81=E3=81=AE=E3=82=B3?= =?UTF-8?q?=E3=83=BC=E3=83=89=E3=81=8A=E3=82=88=E3=81=B3=E3=83=86=E3=82=B9?= =?UTF-8?q?=E3=83=88=E3=82=92=E8=BF=BD=E5=8A=A0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/accuracy_report.py | 464 +++++++++++++++++++++++++++++++++ src/accuracy_session.py | 290 +++++++++++++++++++++ tests/test_accuracy_report.py | 139 ++++++++++ tests/test_accuracy_session.py | 99 +++++++ 4 files changed, 992 insertions(+) create mode 100644 src/accuracy_report.py create mode 100644 src/accuracy_session.py create mode 100644 tests/test_accuracy_report.py create mode 100644 tests/test_accuracy_session.py diff --git a/src/accuracy_report.py b/src/accuracy_report.py new file mode 100644 index 0000000..c6193d1 --- /dev/null +++ b/src/accuracy_report.py @@ -0,0 +1,464 @@ +"""Accuracy-experiment report: join trial markers with ``*_face_frames.csv`` by timestamp. + +Usage (after ``src.accuracy_session`` recorded the markers):: + + python -m src.accuracy_report --trials data/analysis/accuracy/trials_.csv ^ + --face-csv "data/logs/face/*_face_frames.csv" + +Per trial the baseline yaw (subject facing the camera) is averaged, the hold-phase yaw is +compared against it, and ``error = (hold - baseline) - true_yaw``. Outputs (``--out``): +``accuracy_trials.csv``, ``accuracy_conditions.csv``, ``accuracy_thresholds.csv``, +``accuracy_report.md`` and ``plots/*.png``. + +Threshold: ``theta_th = |bias| + k * SD`` of the frame errors for conditions with +``|true_yaw| <= --pool-max-angle`` (pitch 0), per distance. ``--basis worst`` (default) takes the +worst single condition, ``pooled`` pools all of them. k=3 and the pooling range are provisional +choices to be agreed on with the supervisor. +""" + +from __future__ import annotations + +import argparse +import bisect +import csv +import glob +import math +import statistics +import sys +from collections import Counter +from dataclasses import dataclass +from datetime import datetime +from pathlib import Path +from typing import Sequence + +NAN = float("nan") +TRIAL_COLUMNS = ["session_id", "seq", "condition_id", "distance_m", "true_yaw_deg", "pitch_deg", "repeat", "subject", + "status", "track_id", "baseline_n", "baseline_mean", "baseline_sd", "hold_n", "hold_mean", "hold_sd", + "delta_deg", "error_deg", "detection_rate"] +CONDITION_COLUMNS = ["distance_m", "pitch_deg", "true_yaw_deg", "n_trials", "n_frames", "bias_deg", "sd_trials_deg", + "sd_within_deg", "sd_total_deg", "rmse_frame_deg", "p95_abs_error_deg", "mean_detection_rate", + "exceed_rate"] +THRESHOLD_COLUMNS = ["distance_m", "n_frames", "bias_deg", "sd_deg", "theta_pooled_deg", "theta_worst_deg", + "theta_th_deg", "slope", "intercept_deg", "r2", "min_detectable_deg"] + + +@dataclass(frozen=True) +class Options: + signal: str = "yaw" + flip_sign: bool = False + settle: float = 1.5 + baseline_trim: float = 0.5 + min_frames: int = 10 + face_every: int = 1 + track_id: int | None = None + fps: float | None = None + k: float = 3.0 + pool_max_angle: float = 15.0 + basis: str = "worst" + detect_ratio: float = 0.95 + pitch_ref: float = 0.0 + + +@dataclass +class FaceSeries: + times: list[float] + track_ids: list[int] + pose: list[bool] + yaw: list[float | None] + raw_yaw: list[float | None] + fps: list[float] + + +def _ts(value: object) -> float | None: + try: + return datetime.fromisoformat(str(value).strip()).timestamp() + except ValueError: + return None + + +def _float(value: object) -> float | None: + try: + number = float(str(value)) + except ValueError: + return None + return number if math.isfinite(number) else None + + +def _mean(values: Sequence[float]) -> float: + return statistics.fmean(values) if values else NAN + + +def _sd(values: Sequence[float]) -> float: + return statistics.stdev(values) if len(values) >= 2 else NAN + + +def _wrap(angle: float) -> float: + return (angle + 180.0) % 360.0 - 180.0 + + +def percentile(sorted_values: Sequence[float], q: float) -> float: + if not sorted_values: + return NAN + position = (len(sorted_values) - 1) * q / 100.0 + lower, upper = math.floor(position), math.ceil(position) + return float(sorted_values[lower] + (sorted_values[upper] - sorted_values[lower]) * (position - lower)) + + +def expand_paths(patterns: Sequence[str]) -> list[Path]: + """Expand globs ourselves because Windows shells do not.""" + paths: list[Path] = [] + for pattern in patterns: + matches = sorted(glob.glob(pattern)) if any(c in pattern for c in "*?[") else [pattern] + paths.extend(Path(m) for m in matches) + return paths + + +def load_face_series(paths: Sequence[Path], default_fps: float = 30.0) -> FaceSeries: + rows: list[tuple[float, int, bool, float | None, float | None, float]] = [] + for path in paths: + file_rows: list[tuple[float, int, bool, float | None, float | None]] = [] + frames: list[tuple[float, float]] = [] + with path.open(encoding="utf-8-sig", newline="") as stream: + for row in csv.DictReader(stream): + t, track = _ts(row.get("timestamp")), _float(row.get("track_id")) + if t is None or track is None: + continue + file_rows.append((t, int(track), str(row.get("pose_estimated", "")).lower() == "true", + _float(row.get("yaw")), _float(row.get("raw_yaw")))) + frame = _float(row.get("frame_number")) + if frame is not None: + frames.append((t, frame)) + fps = default_fps + if len(frames) >= 2: + dt, df = frames[-1][0] - frames[0][0], frames[-1][1] - frames[0][1] + if dt > 0 and df > 0: + fps = df / dt + rows.extend((*r, fps) for r in file_rows) + rows.sort(key=lambda r: r[0]) + return FaceSeries([r[0] for r in rows], [r[1] for r in rows], [r[2] for r in rows], [r[3] for r in rows], + [r[4] for r in rows], [r[5] for r in rows]) + + +def load_trials(paths: Sequence[Path], exclude_seqs: Sequence[int] = ()) -> list[dict[str, str]]: + """Last log row per (session, seq); only status ``ok`` survives.""" + last: dict[tuple[str, int], dict[str, str]] = {} + for path in paths: + with path.open(encoding="utf-8-sig", newline="") as stream: + for row in csv.DictReader(stream): + last[(row.get("session_id", ""), int(row["seq"]))] = row + return [row for (_, seq), row in sorted(last.items()) if row["status"] == "ok" and seq not in set(exclude_seqs)] + + +def evaluate_trial(trial: dict[str, str], series: FaceSeries, opt: Options) -> tuple[dict[str, object], list[float]]: + """Return the trial record and the frame-level deltas (degrees, signed, hold phase).""" + record: dict[str, object] = {column: NAN for column in TRIAL_COLUMNS} + record.update({ + "session_id": trial.get("session_id", ""), "seq": int(trial["seq"]), "condition_id": trial["condition_id"], + "distance_m": float(trial["distance_m"]), "true_yaw_deg": float(trial["true_yaw_deg"]), + "pitch_deg": float(trial["pitch_deg"]), "repeat": trial.get("repeat", ""), "subject": trial.get("subject", ""), + "status": "ok", "track_id": "", + }) + t_base, t_cue, t_end = (_ts(trial.get(key)) for key in ("t_baseline_start", "t_turn_cue", "t_hold_end")) + if t_base is None or t_cue is None or t_end is None: + record["status"] = "bad_timestamps" + return record, [] + b0, b1, h0, h1 = t_base + opt.baseline_trim, t_cue, t_cue + opt.settle, t_end + values = series.raw_yaw if opt.signal == "raw_yaw" else series.yaw + + def span(a: float, b: float) -> tuple[int, int]: + return bisect.bisect_left(series.times, a), bisect.bisect_right(series.times, b) + + lo, hi = span(b0, h1) + counts = Counter(series.track_ids[i] for i in range(lo, hi) if series.pose[i] and values[i] is not None) + track = opt.track_id if opt.track_id is not None else (counts.most_common(1)[0][0] if counts else None) + if track is None or counts.get(track, 0) == 0: + record["status"] = "no_track" + return record, [] + record["track_id"] = track + + def pick(a: float, b: float) -> list[int]: + s, e = span(a, b) + return [i for i in range(s, e) if series.track_ids[i] == track and series.pose[i] and values[i] is not None] + + base, hold = pick(b0, b1), pick(h0, h1) + base_yaw = [float(values[i]) for i in base] # type: ignore[arg-type] + hold_yaw = [float(values[i]) for i in hold] # type: ignore[arg-type] + record.update({"baseline_n": len(base), "hold_n": len(hold), "baseline_mean": _mean(base_yaw), + "baseline_sd": _sd(base_yaw), "hold_mean": _mean(hold_yaw), "hold_sd": _sd(hold_yaw)}) + if len(base) < opt.min_frames: + record["status"] = "baseline_insufficient" + return record, [] + if len(hold) < opt.min_frames: + record["status"] = "hold_insufficient" + return record, [] + sign = -1.0 if opt.flip_sign else 1.0 + base_mean = float(record["baseline_mean"]) # type: ignore[arg-type] + deltas = [_wrap(sign * (y - base_mean)) for y in hold_yaw] + fps = opt.fps or series.fps[hold[0]] + expected = max(1.0, (h1 - h0) * fps / max(1, opt.face_every)) + record["detection_rate"] = min(1.0, len(hold) / expected) + record["delta_deg"] = _mean(deltas) + record["error_deg"] = float(record["delta_deg"]) - float(record["true_yaw_deg"]) # type: ignore[arg-type] + return record, deltas + + +Usable = list[tuple[dict[str, object], list[float]]] + + +def _errors(record: dict[str, object], deltas: Sequence[float]) -> list[float]: + return [d - float(record["true_yaw_deg"]) for d in deltas] # type: ignore[arg-type] + + +def summarize_conditions(usable: Usable) -> list[dict[str, object]]: + groups: dict[tuple[float, float, float], Usable] = {} + for record, deltas in usable: + key = (float(record["distance_m"]), float(record["pitch_deg"]), float(record["true_yaw_deg"])) # type: ignore[arg-type] + groups.setdefault(key, []).append((record, deltas)) + rows: list[dict[str, object]] = [] + for (distance, pitch, true), items in sorted(groups.items()): + frame_errors = [e for record, deltas in items for e in _errors(record, deltas)] + within = [float(r["hold_sd"]) ** 2 for r, _ in items if not math.isnan(float(r["hold_sd"]))] # type: ignore[arg-type] + rows.append({ + "distance_m": distance, "pitch_deg": pitch, "true_yaw_deg": true, "n_trials": len(items), + "n_frames": len(frame_errors), "bias_deg": _mean([float(r["error_deg"]) for r, _ in items]), # type: ignore[arg-type] + "sd_trials_deg": _sd([float(r["error_deg"]) for r, _ in items]), # type: ignore[arg-type] + "sd_within_deg": math.sqrt(_mean(within)) if within else NAN, "sd_total_deg": _sd(frame_errors), + "rmse_frame_deg": math.sqrt(_mean([e * e for e in frame_errors])), + "p95_abs_error_deg": percentile(sorted(abs(e) for e in frame_errors), 95), + "mean_detection_rate": _mean([float(r["detection_rate"]) for r, _ in items]), # type: ignore[arg-type] + "exceed_rate": NAN, + }) + return rows + + +def _line_fit(points: Sequence[tuple[float, float]]) -> tuple[float, float, float]: + xs, ys = [p[0] for p in points], [p[1] for p in points] + if len(points) < 2 or len(set(xs)) < 2: + return NAN, NAN, NAN + mx, my = statistics.fmean(xs), statistics.fmean(ys) + sxx = sum((x - mx) ** 2 for x in xs) + sxy = sum((x - mx) * (y - my) for x, y in points) + syy = sum((y - my) ** 2 for y in ys) + slope = sxy / sxx + return slope, my - slope * mx, (sxy * sxy / (sxx * syy) if syy > 0 else NAN) + + +def compute_thresholds(usable: Usable, opt: Options) -> dict[float, dict[str, float]]: + result: dict[float, dict[str, float]] = {} + for distance in sorted({float(r["distance_m"]) for r, _ in usable}): # type: ignore[arg-type] + ref = [(r, d) for r, d in usable if float(r["distance_m"]) == distance # type: ignore[arg-type] + and float(r["pitch_deg"]) == opt.pitch_ref] # type: ignore[arg-type] + pooled_sel = [(r, d) for r, d in ref if abs(float(r["true_yaw_deg"])) <= opt.pool_max_angle] # type: ignore[arg-type] + all_errors = [e for r, d in pooled_sel for e in _errors(r, d)] + if len(all_errors) < 2: + continue + pooled = abs(_mean(all_errors)) + opt.k * _sd(all_errors) + by_condition: dict[float, list[float]] = {} + for r, d in pooled_sel: + by_condition.setdefault(float(r["true_yaw_deg"]), []).extend(_errors(r, d)) # type: ignore[arg-type] + worst_values = [abs(_mean(e)) + opt.k * _sd(e) for e in by_condition.values() if len(e) >= 2] + worst = max(worst_values) if worst_values else pooled + slope, intercept, r2 = _line_fit([(float(r["true_yaw_deg"]), float(r["delta_deg"])) for r, _ in ref]) # type: ignore[arg-type] + result[distance] = { + "n_frames": float(len(all_errors)), "bias_deg": _mean(all_errors), "sd_deg": _sd(all_errors), + "theta_pooled_deg": pooled, "theta_worst_deg": worst, + "theta_th_deg": worst if opt.basis == "worst" else pooled, + "slope": slope, "intercept_deg": intercept, "r2": r2, "min_detectable_deg": NAN, + } + return result + + +def add_exceed_rates(conditions: list[dict[str, object]], usable: Usable, thresholds: dict[float, dict[str, float]], + opt: Options) -> None: + """exceed_rate = share of hold frames with |delta| >= theta_th (false-positive rate at 0 deg).""" + frames: dict[tuple[float, float, float], list[float]] = {} + for record, deltas in usable: + key = (float(record["distance_m"]), float(record["pitch_deg"]), float(record["true_yaw_deg"])) # type: ignore[arg-type] + frames.setdefault(key, []).extend(deltas) + for row in conditions: + theta = thresholds.get(float(row["distance_m"]), {}).get("theta_th_deg") # type: ignore[arg-type] + data = frames[(float(row["distance_m"]), float(row["pitch_deg"]), float(row["true_yaw_deg"]))] # type: ignore[arg-type] + row["exceed_rate"] = sum(abs(d) >= theta for d in data) / len(data) if theta is not None and data else NAN + for distance, info in thresholds.items(): + ref = [c for c in conditions if c["distance_m"] == distance and c["pitch_deg"] == opt.pitch_ref + and float(c["true_yaw_deg"]) != 0.0 and not math.isnan(float(c["exceed_rate"]))] # type: ignore[arg-type] + for angle in sorted({abs(float(c["true_yaw_deg"])) for c in ref}): # type: ignore[arg-type] + if all(float(c["exceed_rate"]) >= opt.detect_ratio for c in ref if abs(float(c["true_yaw_deg"])) >= angle): # type: ignore[arg-type] + info["min_detectable_deg"] = angle + break + + +def _fmt(value: object, digits: int = 2) -> str: + if isinstance(value, float): + return "-" if math.isnan(value) else f"{value:.{digits}f}" + return str(value) + + +def _write_csv(path: Path, columns: Sequence[str], rows: Sequence[dict[str, object]]) -> None: + with path.open("w", encoding="utf-8-sig", newline="") as stream: + writer = csv.DictWriter(stream, fieldnames=list(columns), extrasaction="ignore") + writer.writeheader() + for row in rows: + writer.writerow({k: (_fmt(v, 4) if isinstance(v, float) else v) for k, v in row.items()}) + + +def build_markdown(records: list[dict[str, object]], conditions: list[dict[str, object]], + thresholds: dict[float, dict[str, float]], opt: Options) -> str: + lines = ["# 振り向き角度推定の精度実験 結果", ""] + status = Counter(str(r["status"]) for r in records) + lines += [f"- 記録された有効試行: {len(records)}、解析に使用: {status['ok']}", "- 除外: " + ( + ", ".join(f"{k}={v}" for k, v in status.items() if k != "ok") or "なし")] + bad = [str(r["seq"]) for r in records if r["status"] != "ok"] + if bad: + lines.append(f"- 除外された seq: {', '.join(bad)}") + lines += [f"- 解析設定: signal={opt.signal}, flip_sign={opt.flip_sign}, settle={opt.settle:g}s, " + f"baseline_trim={opt.baseline_trim:g}s, k={opt.k:g}, pool_max_angle={opt.pool_max_angle:g}°, " + f"basis={opt.basis}", "", "誤差の定義: error = (保持区間の平均yaw − 基準yaw) − 真値角(度)。", ""] + lines += ["## しきい値 (距離別・pitch 基準条件)", "", + "| 距離[m] | frames | bias | SD | θ_pooled | θ_worst | **θ_th** | 傾き | 切片 | R² | 判定可能な最小角 |", + "|---:|---:|---:|---:|---:|---:|---:|---:|---:|---:|---:|"] + warn = [] + for distance, t in sorted(thresholds.items()): + lines.append(f"| {distance:g} | {int(t['n_frames'])} | {_fmt(t['bias_deg'])} | {_fmt(t['sd_deg'])} | " + f"{_fmt(t['theta_pooled_deg'])} | {_fmt(t['theta_worst_deg'])} | **{_fmt(t['theta_th_deg'])}** | " + f"{_fmt(t['slope'], 3)} | {_fmt(t['intercept_deg'])} | {_fmt(t['r2'], 3)} | " + f"{_fmt(t['min_detectable_deg'])} |") + if not math.isnan(t["slope"]) and t["slope"] < 0: + warn.append(f"{distance:g} m: 傾きが負です。角度の符号定義が逆の可能性があります (--flip-sign)。") + if thresholds: + lines += ["", f"全距離で共通に使う保守的な値: θ_th = {max(t['theta_th_deg'] for t in thresholds.values()):.2f}°"] + lines += [f"- {w}" for w in warn] + lines += ["", "## 条件別の結果", "", + "| 距離[m] | pitch | 真値[°] | 試行 | frames | bias | SD(試行間) | SD(試行内) | SD(全体) | RMSE | 95%|誤差| | 検出率 | θ超過率 |", + "|---:|---:|---:|---:|---:|---:|---:|---:|---:|---:|---:|---:|---:|"] + for c in conditions: + lines.append(f"| {float(c['distance_m']):g} | {float(c['pitch_deg']):+g} | {float(c['true_yaw_deg']):+g} | " # type: ignore[arg-type] + f"{c['n_trials']} | {c['n_frames']} | {_fmt(c['bias_deg'])} | {_fmt(c['sd_trials_deg'])} | " + f"{_fmt(c['sd_within_deg'])} | {_fmt(c['sd_total_deg'])} | {_fmt(c['rmse_frame_deg'])} | " + f"{_fmt(c['p95_abs_error_deg'])} | {_fmt(c['mean_detection_rate'], 3)} | {_fmt(c['exceed_rate'], 3)} |") + lines += ["", "注: θ超過率は保持区間のフレームのうち |Δyaw| ≥ θ_th の割合 (真値0°では誤検出率、それ以外では検出率)。", + "検出率は姿勢推定に成功したフレーム数 / 期待フレーム数 (フレームレートはログから推定)。"] + return "\n".join(lines) + "\n" + + +def make_plots(usable: Usable, thresholds: dict[float, dict[str, float]], out_dir: Path, opt: Options) -> list[Path]: + try: + import matplotlib + matplotlib.use("Agg", force=True) + import matplotlib.pyplot as plt + except ImportError: + return [] + out_dir.mkdir(parents=True, exist_ok=True) + ref = [(r, d) for r, d in usable if float(r["pitch_deg"]) == opt.pitch_ref] # type: ignore[arg-type] + distances = sorted({float(r["distance_m"]) for r, _ in ref}) # type: ignore[arg-type] + if not distances: + return [] + paths: list[Path] = [] + fig, ax = plt.subplots(figsize=(6, 6)) + for distance in distances: + pts = [(float(r["true_yaw_deg"]), float(r["delta_deg"])) for r, _ in ref if float(r["distance_m"]) == distance] # type: ignore[arg-type] + ax.scatter([p[0] for p in pts], [p[1] for p in pts], s=18, alpha=0.7, label=f"{distance:g} m") + lim = max(abs(float(r["true_yaw_deg"])) for r, _ in ref) * 1.15 + 1 # type: ignore[arg-type] + ax.plot([-lim, lim], [-lim, lim], "k--", linewidth=1, label="y = x") + ax.set(xlabel="true yaw [deg]", ylabel="estimated yaw change [deg]", title="Estimated vs true") + ax.legend() + ax.grid(alpha=0.3) + fig.tight_layout() + paths.append(out_dir / "accuracy_scatter.png") + fig.savefig(paths[-1]) + plt.close(fig) + fig, axes = plt.subplots(len(distances), 1, figsize=(8, 3.2 * len(distances)), squeeze=False) + for ax, distance in zip(axes[:, 0], distances): + by_angle: dict[float, list[float]] = {} + for r, d in ref: + if float(r["distance_m"]) == distance: # type: ignore[arg-type] + by_angle.setdefault(float(r["true_yaw_deg"]), []).extend(_errors(r, d)) # type: ignore[arg-type] + angles = sorted(by_angle) + ax.boxplot([by_angle[a] for a in angles], positions=list(range(len(angles))), showfliers=False) + ax.set_xticks(list(range(len(angles)))) + ax.set_xticklabels([f"{a:+g}" for a in angles]) + theta = thresholds.get(distance, {}).get("theta_th_deg") + ax.axhline(0, color="k", linewidth=0.8) + if theta is not None: + ax.axhline(theta, color="r", linestyle=":") + ax.axhline(-theta, color="r", linestyle=":") + ax.set(title=f"Frame error by true yaw ({distance:g} m)", xlabel="true yaw [deg]", ylabel="error [deg]") + fig.tight_layout() + paths.append(out_dir / "accuracy_error_by_condition.png") + fig.savefig(paths[-1]) + plt.close(fig) + fig, ax = plt.subplots(figsize=(7, 4)) + for distance in distances: + errors = [e for r, d in ref if float(r["distance_m"]) == distance # type: ignore[arg-type] + and abs(float(r["true_yaw_deg"])) <= opt.pool_max_angle for e in _errors(r, d)] # type: ignore[arg-type] + if errors: + ax.hist(errors, bins=40, histtype="step", density=True, label=f"{distance:g} m") + ax.set(xlabel="error [deg]", ylabel="density", title=f"Frame error distribution (|true| <= {opt.pool_max_angle:g} deg)") + ax.legend() + fig.tight_layout() + paths.append(out_dir / "accuracy_error_hist.png") + fig.savefig(paths[-1]) + plt.close(fig) + return paths + + +def run_report(trial_paths: Sequence[Path], face_paths: Sequence[Path], out_dir: Path, opt: Options, + exclude_seqs: Sequence[int] = (), plots: bool = True) -> dict[str, object]: + trials = load_trials(trial_paths, exclude_seqs) + series = load_face_series(face_paths) + evaluated = [evaluate_trial(t, series, opt) for t in trials] + records = [rec for rec, _ in evaluated] + usable: Usable = [(rec, deltas) for rec, deltas in evaluated if rec["status"] == "ok"] + conditions = summarize_conditions(usable) + thresholds = compute_thresholds(usable, opt) + add_exceed_rates(conditions, usable, thresholds, opt) + out_dir.mkdir(parents=True, exist_ok=True) + _write_csv(out_dir / "accuracy_trials.csv", TRIAL_COLUMNS, records) + _write_csv(out_dir / "accuracy_conditions.csv", CONDITION_COLUMNS, conditions) + _write_csv(out_dir / "accuracy_thresholds.csv", THRESHOLD_COLUMNS, + [{"distance_m": d, **t} for d, t in sorted(thresholds.items())]) + markdown = build_markdown(records, conditions, thresholds, opt) + (out_dir / "accuracy_report.md").write_text(markdown, encoding="utf-8") + figures = make_plots(usable, thresholds, out_dir / "plots", opt) if plots else [] + return {"records": records, "conditions": conditions, "thresholds": thresholds, "markdown": markdown, + "figures": figures} + + +def main(argv: Sequence[str] | None = None, out=print) -> int: # type: ignore[no-untyped-def] + parser = argparse.ArgumentParser(description="Accuracy report from trial markers and face_frames logs") + parser.add_argument("--trials", nargs="+", required=True, help="trials CSV file(s) from accuracy_session") + parser.add_argument("--face-csv", nargs="+", required=True, help="*_face_frames.csv file(s); globs allowed") + parser.add_argument("--out", type=Path, default=Path("data/analysis/accuracy/report")) + parser.add_argument("--signal", choices=("yaw", "raw_yaw"), default="yaw") + parser.add_argument("--flip-sign", action="store_true") + parser.add_argument("--settle", type=float, default=1.5, help="seconds after the cue ignored (turn in progress)") + parser.add_argument("--baseline-trim", type=float, default=0.5) + parser.add_argument("--min-frames", type=int, default=10) + parser.add_argument("--face-every", type=int, default=1, help="face.process_every_n_frames of the run") + parser.add_argument("--track-id", type=int, default=None) + parser.add_argument("--fps", type=float, default=None, help="override the frame rate used for detection rate") + parser.add_argument("--k", type=float, default=3.0) + parser.add_argument("--pool-max-angle", type=float, default=15.0) + parser.add_argument("--basis", choices=("worst", "pooled"), default="worst") + parser.add_argument("--detect-ratio", type=float, default=0.95) + parser.add_argument("--exclude-seq", type=int, nargs="*", default=[]) + parser.add_argument("--no-plots", action="store_true") + args = parser.parse_args(argv) + trial_paths, face_paths = expand_paths(args.trials), expand_paths(args.face_csv) + missing = [str(p) for p in trial_paths + face_paths if not p.is_file()] + if missing or not trial_paths or not face_paths: + print(f"file not found: {', '.join(missing) or 'no input files'}", file=sys.stderr) + return 2 + opt = Options(args.signal, args.flip_sign, args.settle, args.baseline_trim, args.min_frames, args.face_every, + args.track_id, args.fps, args.k, args.pool_max_angle, args.basis, args.detect_ratio) + result = run_report(trial_paths, face_paths, args.out, opt, args.exclude_seq, not args.no_plots) + if not result["records"]: + print("no usable trials (status ok) in the trials file", file=sys.stderr) + return 1 + out(str(result["markdown"])) + out(f"-> {args.out}") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/accuracy_session.py b/src/accuracy_session.py new file mode 100644 index 0000000..66b9039 --- /dev/null +++ b/src/accuracy_session.py @@ -0,0 +1,290 @@ +"""Trial-session runner for the head-yaw accuracy experiment. + +Run it in a second terminal while ``python main.py`` records ``*_face_frames.csv``. +The runner only writes *wall-clock markers* (``data/analysis/accuracy/trials_*.csv``); +``src.accuracy_report`` later joins those markers with the face log by timestamp, so +the live pipeline is not touched. + +1. Generate a randomized plan (blocked by distance, shuffled inside each block):: + + python -m src.accuracy_session plan --distances 1,2,3 --repeats 3 --seed 1 --subject S01 + +2. Start the system (voice prompt off), then run the session:: + + python main.py --disable-voice --camera-position-note accuracy + python -m src.accuracy_session run --plan data/analysis/accuracy/plan.csv + + Per trial: baseline (look straight ahead) -> cue -> turn the head to the target and + hold -> cue -> return. Prompt keys: Enter=start, r=redo previous, x=discard previous, + s=skip this condition, q=quit. ``--auto-interval 6`` removes the Enter key between + trials (single-person operation); a distance change always waits for Enter. + Re-run with ``--resume --trials `` to continue an interrupted session. + +The sign of ``true_yaw_deg`` is whatever you define on the floor sheet; the report +checks the regression slope and tells you when ``--flip-sign`` is needed. +""" + +from __future__ import annotations + +import argparse +import csv +import random +import sys +import time +from dataclasses import dataclass +from datetime import datetime +from pathlib import Path +from typing import Callable, Sequence + +DEFAULT_DIRECTORY = Path("data/analysis/accuracy") +DEFAULT_ANGLES = (0.0, 5.0, -5.0, 10.0, -10.0, 15.0, -15.0, 20.0, -20.0, 30.0, -30.0) +PLAN_FIELDS = ["seq", "block", "condition_id", "distance_m", "true_yaw_deg", "pitch_deg", "repeat", "subject"] +TRIAL_FIELDS = ["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", "baseline_sec", "hold_sec"] +FINAL_STATUSES = {"ok", "skipped"} + + +@dataclass(frozen=True) +class PlanRow: + seq: int + block: int + condition_id: str + distance_m: float + true_yaw_deg: float + pitch_deg: float + repeat: int + subject: str = "" + + +def condition_id(distance: float, yaw: float, pitch: float) -> str: + return f"d{distance:g}_y{yaw + 0.0:+g}_p{pitch + 0.0:+g}" + + +def parse_floats(text: str) -> list[float]: + return [float(part) for part in text.replace(" ", "").split(",") if part] + + +def build_plan( + distances: Sequence[float], angles: Sequence[float], repeats: int, *, seed: int | None = None, + subject: str = "", pitch_levels: Sequence[float] = (), pitch_distances: Sequence[float] = (), + pitch_angles: Sequence[float] = (0.0, 10.0, -10.0), +) -> list[PlanRow]: + """One block per distance; trials are shuffled inside a block (moving the subject is slow).""" + rng = random.Random(seed) + rows: list[PlanRow] = [] + seq = 0 + for block, distance in enumerate(distances, start=1): + combos = [(a, 0.0) for a in angles] + if distance in pitch_distances: + combos += [(a, p) for p in pitch_levels if p != 0 for a in pitch_angles] + items = [(a, p, r) for a, p in combos for r in range(1, repeats + 1)] + rng.shuffle(items) + for a, p, r in items: + seq += 1 + rows.append(PlanRow(seq, block, condition_id(distance, a, p), distance, a + 0.0, p + 0.0, r, subject)) + return rows + + +def write_plan(rows: Sequence[PlanRow], path: Path) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + with path.open("w", encoding="utf-8-sig", newline="") as stream: + writer = csv.DictWriter(stream, fieldnames=PLAN_FIELDS) + writer.writeheader() + for row in rows: + writer.writerow({field: getattr(row, field) for field in PLAN_FIELDS}) + + +def read_plan(path: Path) -> list[PlanRow]: + with path.open(encoding="utf-8-sig", newline="") as stream: + return [ + PlanRow(int(r["seq"]), int(r["block"]), r["condition_id"], float(r["distance_m"]), + float(r["true_yaw_deg"]), float(r["pitch_deg"]), int(r["repeat"]), r.get("subject", "")) + for r in csv.DictReader(stream) + ] + + +def load_done(path: Path) -> set[int]: + """Sequence numbers whose *last* log row is final (ok / skipped).""" + last: dict[int, str] = {} + with path.open(encoding="utf-8-sig", newline="") as stream: + for row in csv.DictReader(stream): + last[int(row["seq"])] = row["status"] + return {seq for seq, status in last.items() if status in FINAL_STATUSES} + + +def _now_iso() -> str: + return datetime.now().astimezone().isoformat(timespec="milliseconds") + + +def _beep(kind: str) -> None: + try: + import winsound + winsound.Beep({"start": 880, "turn": 1320, "end": 440}.get(kind, 660), 200) + except (ImportError, RuntimeError): + print("\a", end="", flush=True) + + +def _sign(value: float) -> str: + return f"{value + 0.0:+g}" + + +class SessionRunner: + """Interactive loop; all I/O and time sources are injectable for tests.""" + + def __init__( + self, plan: Sequence[PlanRow], trials_path: Path, session_id: str, *, baseline_sec: float = 3.0, + hold_sec: float = 5.0, auto_interval: float = 0.0, input_fn: Callable[[str], str] = input, + out: Callable[[str], object] = print, sleep: Callable[[float], object] = time.sleep, + clock: Callable[[], str] = _now_iso, beep: Callable[[str], object] = _beep, + ) -> None: + self.plan, self.trials_path, self.session_id = list(plan), trials_path, session_id + self.baseline_sec, self.hold_sec, self.auto_interval = baseline_sec, hold_sec, auto_interval + self.input_fn, self.out, self.sleep, self.clock, self.beep = input_fn, out, sleep, clock, beep + + def run(self, resume: bool = False) -> int: + path = self.trials_path + exists = path.is_file() and path.stat().st_size > 0 + if exists and not resume: + self.out(f"trials file already exists: {path} (use --resume or another --trials)") + return 2 + pending = [row for row in self.plan if row.seq not in (load_done(path) if exists else set())] + if not pending: + self.out("nothing to do: every planned condition is already finished") + return 0 + path.parent.mkdir(parents=True, exist_ok=True) + estimate = len(pending) * (self.baseline_sec + self.hold_sec + 4.0) / 60.0 + self.out(f"{len(pending)} trials pending (rough duration {estimate:.0f} min). session={self.session_id}") + with path.open("a", encoding="utf-8-sig", newline="") as stream: + writer = csv.DictWriter(stream, fieldnames=TRIAL_FIELDS) + if not exists: + writer.writeheader() + return self._loop(pending, writer, stream) + + def _record(self, row: PlanRow, status: str, t0: str = "", t1: str = "", t2: str = "") -> dict[str, object]: + return {"session_id": self.session_id, "seq": row.seq, "block": row.block, "condition_id": row.condition_id, + "distance_m": row.distance_m, "true_yaw_deg": row.true_yaw_deg, "pitch_deg": row.pitch_deg, + "repeat": row.repeat, "subject": row.subject, "status": status, "t_baseline_start": t0, + "t_turn_cue": t1, "t_hold_end": t2, "baseline_sec": self.baseline_sec, "hold_sec": self.hold_sec} + + def _loop(self, pending: list[PlanRow], writer: csv.DictWriter, stream) -> int: # type: ignore[type-arg] + def log(row: PlanRow, status: str, *stamps: str) -> None: + writer.writerow(self._record(row, status, *stamps)) + stream.flush() + + index, last_index, current_block, finished = 0, None, None, 0 + while index < len(pending): + row = pending[index] + new_block = row.block != current_block + if new_block: + current_block = row.block + self.out(f"\n=== block {row.block}: set the subject at {row.distance_m:g} m, then continue ===") + self.out(f"[{index + 1}/{len(pending)}] seq {row.seq} distance {row.distance_m:g} m " + f"yaw {_sign(row.true_yaw_deg)} deg pitch {_sign(row.pitch_deg)} deg (repeat {row.repeat})") + try: + if self.auto_interval > 0 and not new_block: + self._countdown() + command = "" + else: + command = self.input_fn("Enter=start r=redo-prev x=discard-prev s=skip q=quit > ").strip().lower() + except (KeyboardInterrupt, EOFError): + command = "q" + if command == "q": + break + if command in {"r", "x"}: + if last_index is None: + self.out("no previous trial in this run") + continue + log(pending[last_index], "redo" if command == "r" else "discarded") + finished -= 1 + if command == "r": + index, last_index, current_block = last_index, None, None + else: + last_index = None + continue + if command == "s": + log(row, "skipped") + index, last_index = index + 1, None + continue + try: + stamps = self._execute(row) + except KeyboardInterrupt: + log(row, "aborted") + self.out("aborted; the interrupted trial is marked and can be re-run with --resume") + return 130 + log(row, "ok", *stamps) + finished += 1 + last_index, index = index, index + 1 + self.out(f"session ended: {finished} trials recorded in this run -> {self.trials_path}") + return 0 + + def _countdown(self) -> None: + remaining = self.auto_interval + while remaining > 0: + self.out(f" starting in {remaining:.0f} s ...") + self.sleep(min(1.0, remaining)) + remaining -= 1.0 + + def _execute(self, row: PlanRow) -> tuple[str, str, str]: + t_base = self.clock() + self.beep("start") + self.out(f" baseline {self.baseline_sec:g} s: face the camera and stay still") + self.sleep(self.baseline_sec) + t_cue = self.clock() + self.beep("turn") + self.out(f" TURN to {_sign(row.true_yaw_deg)} deg (head only) and hold {self.hold_sec:g} s") + self.sleep(self.hold_sec) + t_end = self.clock() + self.beep("end") + self.out(" return to the front") + return t_base, t_cue, t_end + + +def _build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description="Accuracy experiment: plan generation and trial session") + sub = parser.add_subparsers(dest="command", required=True) + plan = sub.add_parser("plan", help="write a randomized condition plan CSV") + plan.add_argument("--distances", default="1,2,3", help="metres, comma separated (one block each)") + plan.add_argument("--angles", default=",".join(f"{a:g}" for a in DEFAULT_ANGLES), help="true yaw angles in degrees") + plan.add_argument("--repeats", type=int, default=3) + plan.add_argument("--seed", type=int, default=None) + plan.add_argument("--subject", default="") + plan.add_argument("--pitch-levels", default="", help="extra pitch conditions, e.g. -10,10 (optional)") + plan.add_argument("--pitch-distances", default="2", help="distances that also get the pitch conditions") + plan.add_argument("--pitch-angles", default="0,10,-10", help="yaw angles used with the extra pitch levels") + plan.add_argument("--out", type=Path, default=DEFAULT_DIRECTORY / "plan.csv") + run = sub.add_parser("run", help="run the interactive trial session") + run.add_argument("--plan", type=Path, default=DEFAULT_DIRECTORY / "plan.csv") + run.add_argument("--trials", type=Path, default=None, help="output markers CSV (default: trials_.csv)") + run.add_argument("--session-id", default=None) + run.add_argument("--baseline-sec", type=float, default=3.0) + run.add_argument("--hold-sec", type=float, default=5.0) + run.add_argument("--auto-interval", type=float, default=0.0, help="seconds between trials; 0 = wait for Enter") + run.add_argument("--resume", action="store_true") + return parser + + +def main(argv: Sequence[str] | None = None, out: Callable[[str], object] = print) -> int: + args = _build_parser().parse_args(argv) + if args.command == "plan": + rows = build_plan(parse_floats(args.distances), parse_floats(args.angles), args.repeats, seed=args.seed, + subject=args.subject, pitch_levels=parse_floats(args.pitch_levels), + pitch_distances=parse_floats(args.pitch_distances), + pitch_angles=parse_floats(args.pitch_angles)) + write_plan(rows, args.out) + out(f"{len(rows)} trials, {len({r.block for r in rows})} blocks -> {args.out}") + return 0 + if not args.plan.is_file(): + print(f"plan not found: {args.plan} (run the 'plan' command first)", file=sys.stderr) + return 2 + session_id = args.session_id or datetime.now().strftime("acc_%Y%m%d_%H%M%S") + trials = args.trials or DEFAULT_DIRECTORY / f"trials_{session_id}.csv" + if args.resume and args.trials is None: + print("--resume needs --trials ", file=sys.stderr) + return 2 + runner = SessionRunner(read_plan(args.plan), trials, session_id, baseline_sec=args.baseline_sec, + hold_sec=args.hold_sec, auto_interval=args.auto_interval) + return runner.run(resume=args.resume) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_accuracy_report.py b/tests/test_accuracy_report.py new file mode 100644 index 0000000..fc5887c --- /dev/null +++ b/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 diff --git a/tests/test_accuracy_session.py b/tests/test_accuracy_session.py new file mode 100644 index 0000000..7138f2d --- /dev/null +++ b/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