From 717bc8fe193352d5387c0631e0642195bb2272e4 Mon Sep 17 00:00:00 2001 From: KoyoNoguchi Date: Mon, 5 Oct 2026 15:18:16 +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=E7=94=A8=E3=82=B9=E3=82=AF?= =?UTF-8?q?=E3=83=AA=E3=83=97=E3=83=88=E3=81=AB=EF=BC=8C=E3=83=96=E3=83=AD?= =?UTF-8?q?=E3=83=83=E3=82=AF=E9=96=93=E3=82=AA=E3=83=BC=E3=83=88=E3=82=A4?= =?UTF-8?q?=E3=83=B3=E3=82=BF=E3=83=BC=E3=83=90=E3=83=AB=E3=81=A8=E3=82=A4?= =?UTF-8?q?=E3=83=B3=E3=82=BF=E3=83=BC=E3=83=90=E3=83=AB=E9=96=8B=E5=A7=8B?= =?UTF-8?q?=E6=99=82=E3=81=AE=E9=9F=B3=E5=A3=B0=E5=86=8D=E7=94=9F=E6=A9=9F?= =?UTF-8?q?=E8=83=BD=E3=82=92=E5=AE=9F=E8=A3=85?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/accuracy_session.py | 116 +++++++++++++++++++++++++++++++++++----- 1 file changed, 102 insertions(+), 14 deletions(-) diff --git a/src/accuracy_session.py b/src/accuracy_session.py index 66b9039..27a4a76 100644 --- a/src/accuracy_session.py +++ b/src/accuracy_session.py @@ -17,7 +17,8 @@ the live pipeline is not touched. 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. + trials (single-person operation); a distance change waits for Enter, then optionally + uses ``--block-interval`` to give the subject time to move before the next trial. 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 @@ -28,6 +29,8 @@ from __future__ import annotations import argparse import csv +import importlib +import logging import random import sys import time @@ -37,6 +40,11 @@ from pathlib import Path from typing import Callable, Sequence DEFAULT_DIRECTORY = Path("data/analysis/accuracy") +INTERVAL_AUDIO = { + "block": Path("assets/audio/block.wav"), + "auto": Path("assets/audio/auto_interval.wav"), +} +LOGGER = logging.getLogger(__name__) 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", @@ -128,37 +136,109 @@ def _sign(value: float) -> str: return f"{value + 0.0:+g}" +class IntervalAudioPlayer: + """Minimal pygame-backed audio player for accuracy-session interval cues.""" + + def __init__(self, backend: object | None = None) -> None: + self._backend = backend + self._sounds: dict[str, object] = {} + self._initialized = False + self._closed = False + + if self._backend is None: + try: + self._backend = importlib.import_module("pygame") + except ImportError: + LOGGER.warning("pygame is not available; interval audio cues are disabled") + return + + self._load_sounds() + + def _load_sounds(self) -> None: + if self._backend is None or self._closed: + return + try: + mixer = self._backend.mixer # type: ignore[attr-defined] + if not mixer.get_init(): + mixer.init() + self._initialized = True + for kind, path in INTERVAL_AUDIO.items(): + if not path.is_file(): + LOGGER.warning("Interval audio file does not exist: %s", path) + continue + try: + self._sounds[kind] = mixer.Sound(str(path)) + except Exception as exc: + LOGGER.error("Failed to load interval audio %s: %s", path, exc) + except Exception as exc: + LOGGER.error("Failed to initialize pygame mixer for interval audio: %s", exc) + self._initialized = False + self._sounds.clear() + + def play(self, kind: str) -> bool: + if self._closed or kind not in self._sounds: + return False + try: + sound = self._sounds[kind] + sound.play() # type: ignore[attr-defined] + return True + except Exception as exc: + LOGGER.error("Failed to play interval audio %s: %s", kind, exc) + return False + + def close(self) -> None: + if self._closed: + return + if self._backend is not None and self._initialized: + try: + mixer = self._backend.mixer # type: ignore[attr-defined] + if mixer.get_init(): + mixer.quit() + except Exception as exc: + LOGGER.warning("Failed to close pygame mixer for interval audio: %s", exc) + self._closed = True + self._sounds.clear() + + 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, + hold_sec: float = 5.0, auto_interval: float = 0.0, block_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, + interval_audio: IntervalAudioPlayer | None = None, ) -> 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.baseline_sec, self.hold_sec = baseline_sec, hold_sec + self.auto_interval, self.block_interval = auto_interval, block_interval self.input_fn, self.out, self.sleep, self.clock, self.beep = input_fn, out, sleep, clock, beep + self.interval_audio = interval_audio or IntervalAudioPlayer() 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)") + self.interval_audio.close() 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") + self.interval_audio.close() 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) + try: + 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) + finally: + self.interval_audio.close() 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, @@ -175,6 +255,7 @@ class SessionRunner: while index < len(pending): row = pending[index] new_block = row.block != current_block + block_change = current_block is not None and new_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 ===") @@ -182,7 +263,8 @@ class SessionRunner: 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() + self.interval_audio.play("auto") + self._countdown(self.auto_interval) command = "" else: command = self.input_fn("Enter=start r=redo-prev x=discard-prev s=skip q=quit > ").strip().lower() @@ -205,6 +287,9 @@ class SessionRunner: log(row, "skipped") index, last_index = index + 1, None continue + if block_change and self.block_interval > 0: + self.interval_audio.play("block") + self._countdown(self.block_interval, label="moving to next block") try: stamps = self._execute(row) except KeyboardInterrupt: @@ -217,10 +302,10 @@ class SessionRunner: self.out(f"session ended: {finished} trials recorded in this run -> {self.trials_path}") return 0 - def _countdown(self) -> None: - remaining = self.auto_interval + def _countdown(self, seconds: float, *, label: str = "starting") -> None: + remaining = seconds while remaining > 0: - self.out(f" starting in {remaining:.0f} s ...") + self.out(f" {label}: {remaining:.0f} s ...") self.sleep(min(1.0, remaining)) remaining -= 1.0 @@ -258,7 +343,8 @@ def _build_parser() -> argparse.ArgumentParser: 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("--auto-interval", type=float, default=0.0, help="seconds between trials within a block; 0 = wait for Enter") + run.add_argument("--block-interval", type=float, default=0.0, help="seconds to allow moving after Enter at a block change; 0 = no extra wait") run.add_argument("--resume", action="store_true") return parser @@ -282,9 +368,11 @@ def main(argv: Sequence[str] | None = None, out: Callable[[str], object] = print 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) + hold_sec=args.hold_sec, auto_interval=args.auto_interval, + block_interval=args.block_interval) return runner.run(resume=args.resume) if __name__ == "__main__": raise SystemExit(main()) +