100 lines
4.3 KiB
Python
100 lines
4.3 KiB
Python
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
|