added progress from previous repo host
This commit is contained in:
@@ -0,0 +1,359 @@
|
||||
"""Guided screen-based gaze exercises for Pupil Capture."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import csv
|
||||
from datetime import datetime
|
||||
import math
|
||||
from pathlib import Path
|
||||
import queue
|
||||
import random
|
||||
import statistics
|
||||
import threading
|
||||
import time
|
||||
import tkinter as tk
|
||||
|
||||
from show_apriltag_surface import SurfaceTagWindow, enable_windows_dpi_awareness, validate_args
|
||||
from visualize_surface_gaze import GazeSample, SurfaceReceiver
|
||||
|
||||
|
||||
SCREEN_WIDTH = 2560
|
||||
SCREEN_HEIGHT = 1440
|
||||
SCREEN_DIAGONAL_IN = 31.5
|
||||
DEFAULT_VIEWING_DISTANCE_CM = 60.0
|
||||
TARGET_RADIUS = 80
|
||||
HIT_RADIUS = 140
|
||||
SAFE_LEFT, SAFE_RIGHT = 260, SCREEN_WIDTH - 260
|
||||
SAFE_TOP, SAFE_BOTTOM = 180, SCREEN_HEIGHT - 180
|
||||
|
||||
|
||||
def parse_args() -> argparse.Namespace:
|
||||
parser = argparse.ArgumentParser(description="Run four guided gaze exercises.")
|
||||
parser.add_argument("--host", default="127.0.0.1")
|
||||
parser.add_argument("--port", type=int, default=50020)
|
||||
parser.add_argument("--surface", default="screen")
|
||||
parser.add_argument("--confidence", type=float, default=0.9)
|
||||
parser.add_argument("--screen-diagonal", type=float, default=SCREEN_DIAGONAL_IN,
|
||||
help="physical screen diagonal in inches (default: 31.5)")
|
||||
parser.add_argument("--viewing-distance", type=float, default=DEFAULT_VIEWING_DISTANCE_CM,
|
||||
help="eye-to-screen distance in cm (default: 60)")
|
||||
parser.add_argument("--rounds", type=int, default=3,
|
||||
help="number of complete 1-2-3-4 exercise rounds (default: 3)")
|
||||
parser.add_argument("--animation-fps", type=int, default=120,
|
||||
help="target animation refresh rate (default: 120)")
|
||||
parser.add_argument("--cursor", action="store_true",
|
||||
help="show the live gaze cursor (hidden by default)")
|
||||
parser.add_argument("--tag-ids", nargs=4, type=int, default=(0, 1, 2, 3))
|
||||
parser.add_argument("--tag-size", type=int, default=200)
|
||||
parser.add_argument("--margin", type=int, default=0)
|
||||
parser.add_argument("--background", default="#10141c")
|
||||
parser.add_argument("--windowed", action="store_true")
|
||||
return parser.parse_args()
|
||||
|
||||
|
||||
class ExerciseWindow(SurfaceTagWindow):
|
||||
BLOCKS = ("fixation", "saccade", "colors", "pursuit")
|
||||
|
||||
def __init__(self, root: tk.Tk, args: argparse.Namespace) -> None:
|
||||
super().__init__(root, args)
|
||||
self.args = args
|
||||
self.frame_interval_ms = max(1, round(1000 / args.animation_fps))
|
||||
self.messages: queue.Queue = queue.Queue(maxsize=256)
|
||||
self.stop_event = threading.Event()
|
||||
self.receiver = SurfaceReceiver(
|
||||
self.messages, self.stop_event, args.host, args.port,
|
||||
args.surface, args.confidence
|
||||
)
|
||||
self.receiver.start()
|
||||
self.gaze: tuple[float, float] | None = None
|
||||
self.rng = random.Random()
|
||||
|
||||
self.block_index = -1
|
||||
self.round = 1
|
||||
self.block = "welcome"
|
||||
self.trial = 0
|
||||
self.trial_started = 0.0
|
||||
self.block_started = 0.0
|
||||
self.target = (SCREEN_WIDTH / 2, SCREEN_HEIGHT / 2)
|
||||
self.distractor: tuple[float, float] | None = None
|
||||
self.correct_color = "#39e681"
|
||||
self.hit_started: float | None = None
|
||||
self.errors: list[float] = []
|
||||
self.hits = 0
|
||||
self.samples = 0
|
||||
self.summary_lines: list[str] = []
|
||||
diagonal_px = math.hypot(SCREEN_WIDTH, SCREEN_HEIGHT)
|
||||
self.cm_per_px = args.screen_diagonal * 2.54 / diagonal_px
|
||||
|
||||
results_dir = Path(__file__).resolve().parent / "results"
|
||||
results_dir.mkdir(exist_ok=True)
|
||||
stamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
||||
self.result_path = results_dir / f"gaze_exercises_{stamp}.csv"
|
||||
self.result_file = self.result_path.open("w", newline="", encoding="utf-8")
|
||||
self.writer = csv.writer(self.result_file)
|
||||
self.writer.writerow([
|
||||
"pupil_timestamp", "round", "block", "trial", "trial_elapsed_s",
|
||||
"target_x", "target_y", "gaze_x", "gaze_y", "error_px",
|
||||
"error_cm", "error_deg", "hit"
|
||||
])
|
||||
|
||||
self.target_item = self.canvas.create_oval(0, 0, 0, 0, state="hidden", tags="exercise")
|
||||
self.target_center = self.canvas.create_oval(0, 0, 0, 0, state="hidden", tags="exercise")
|
||||
self.distractor_item = self.canvas.create_oval(0, 0, 0, 0, state="hidden", tags="exercise")
|
||||
self.gaze_item = self.canvas.create_oval(
|
||||
0, 0, 0, 0, fill="#ffd43b", outline="white", width=2,
|
||||
state="hidden", tags="exercise"
|
||||
)
|
||||
self.heading = self.canvas.create_text(
|
||||
0, 70, fill="white", font=("Segoe UI", 25, "bold"), tags="exercise"
|
||||
)
|
||||
self.instructions = self.canvas.create_text(
|
||||
0, 125, fill="#d7deea", font=("Segoe UI", 16),
|
||||
justify="center", tags="exercise"
|
||||
)
|
||||
self.network_status = self.canvas.create_text(
|
||||
0, 20, fill="#ffca58", font=("Segoe UI", 11), tags="exercise"
|
||||
)
|
||||
root.bind("<space>", self.on_space)
|
||||
root.protocol("WM_DELETE_WINDOW", self.close)
|
||||
self.show_welcome()
|
||||
root.after(self.frame_interval_ms, self.tick)
|
||||
|
||||
def sx(self, x: float) -> float:
|
||||
return x / SCREEN_WIDTH * self.canvas.winfo_width()
|
||||
|
||||
def sy(self, y: float) -> float:
|
||||
return y / SCREEN_HEIGHT * self.canvas.winfo_height()
|
||||
|
||||
def draw(self, event: tk.Event) -> None:
|
||||
super().draw(event)
|
||||
self.canvas.coords(self.heading, event.width / 2, 65)
|
||||
self.canvas.coords(self.instructions, event.width / 2, 115)
|
||||
self.canvas.coords(self.network_status, event.width / 2, 18)
|
||||
self.position_shapes()
|
||||
self.canvas.tag_raise("exercise")
|
||||
|
||||
def close(self, _event: tk.Event | None = None) -> None:
|
||||
self.stop_event.set()
|
||||
if not self.result_file.closed:
|
||||
self.result_file.close()
|
||||
super().close(_event)
|
||||
|
||||
def show_welcome(self) -> None:
|
||||
self.block = "welcome"
|
||||
self.hide_targets()
|
||||
self.canvas.itemconfigure(self.heading, text="Gaze exercise sequence")
|
||||
self.canvas.itemconfigure(
|
||||
self.instructions,
|
||||
text=(f"The following sequence repeats {self.args.rounds} times in 1-2-3-4 order:\n\n"
|
||||
"1. Hold gaze on 10 targets for 3 seconds each\n"
|
||||
"2. Move quickly to 10 appearing targets\n"
|
||||
"3. Choose the instructed color in 20 trials\n"
|
||||
"4. Follow a moving target for 10 seconds × 3\n\n"
|
||||
"Sit comfortably, keep your head natural, then press SPACE."),
|
||||
)
|
||||
|
||||
def on_space(self, _event: tk.Event | None = None) -> None:
|
||||
if self.block in {"welcome", "between"}:
|
||||
self.start_next_block()
|
||||
|
||||
def start_next_block(self) -> None:
|
||||
self.block_index += 1
|
||||
if self.block_index >= len(self.BLOCKS):
|
||||
if self.round >= self.args.rounds:
|
||||
self.finish()
|
||||
return
|
||||
self.round += 1
|
||||
self.block_index = 0
|
||||
self.block = self.BLOCKS[self.block_index]
|
||||
self.trial = 0
|
||||
self.errors.clear()
|
||||
self.hits = self.samples = 0
|
||||
self.block_started = self.trial_started = time.monotonic()
|
||||
self.new_trial()
|
||||
|
||||
def random_target(self) -> tuple[float, float]:
|
||||
return self.rng.uniform(SAFE_LEFT, SAFE_RIGHT), self.rng.uniform(SAFE_TOP, SAFE_BOTTOM)
|
||||
|
||||
def new_trial(self) -> None:
|
||||
self.trial_started = time.monotonic()
|
||||
self.hit_started = None
|
||||
self.distractor = None
|
||||
if self.block == "fixation":
|
||||
self.target = self.random_target()
|
||||
self.set_text(f"Round {self.round}/{self.args.rounds} — Steady fixation — {self.trial + 1}/10", "Keep looking at the center dot")
|
||||
elif self.block == "saccade":
|
||||
self.target = self.random_target()
|
||||
self.set_text(f"Round {self.round}/{self.args.rounds} — Quick movement — {self.trial + 1}/10", "Look at the new target as quickly as possible")
|
||||
elif self.block == "colors":
|
||||
self.target = self.random_target()
|
||||
while True:
|
||||
self.distractor = self.random_target()
|
||||
if math.dist(self.target, self.distractor) > 500:
|
||||
break
|
||||
self.correct_color = "#39e681"
|
||||
self.set_text(
|
||||
f"Round {self.round}/{self.args.rounds} — Always look at GREEN",
|
||||
f"Ignore red • Trial {self.trial + 1}/20",
|
||||
)
|
||||
elif self.block == "pursuit":
|
||||
self.set_text(f"Round {self.round}/{self.args.rounds} — Smooth pursuit — {self.trial + 1}/3", "Follow the moving target with your eyes")
|
||||
self.position_shapes()
|
||||
|
||||
def set_text(self, heading: str, instructions: str) -> None:
|
||||
self.canvas.itemconfigure(self.heading, text=heading)
|
||||
self.canvas.itemconfigure(self.instructions, text=instructions)
|
||||
|
||||
def hide_targets(self) -> None:
|
||||
for item in (self.target_item, self.target_center, self.distractor_item):
|
||||
self.canvas.itemconfigure(item, state="hidden")
|
||||
|
||||
def position_circle(self, item: int, point: tuple[float, float], radius: int) -> None:
|
||||
x, y = self.sx(point[0]), self.sy(point[1])
|
||||
rx = radius / SCREEN_WIDTH * self.canvas.winfo_width()
|
||||
ry = radius / SCREEN_HEIGHT * self.canvas.winfo_height()
|
||||
self.canvas.coords(item, x - rx, y - ry, x + rx, y + ry)
|
||||
|
||||
def position_shapes(self) -> None:
|
||||
if self.block not in self.BLOCKS:
|
||||
return
|
||||
color = self.correct_color if self.block == "colors" else "#55aaff"
|
||||
self.position_circle(self.target_item, self.target, TARGET_RADIUS)
|
||||
self.position_circle(self.target_center, self.target, 7)
|
||||
self.canvas.itemconfigure(self.target_item, fill=color, outline="white", width=3, state="normal")
|
||||
self.canvas.itemconfigure(self.target_center, fill="white", outline="", state="normal")
|
||||
if self.block == "colors" and self.distractor is not None:
|
||||
other = "#ff5263" if self.correct_color == "#39e681" else "#39e681"
|
||||
self.position_circle(self.distractor_item, self.distractor, TARGET_RADIUS)
|
||||
self.canvas.itemconfigure(self.distractor_item, fill=other, outline="white", width=3, state="normal")
|
||||
else:
|
||||
self.canvas.itemconfigure(self.distractor_item, state="hidden")
|
||||
|
||||
def process_network(self) -> list[tuple[GazeSample, tuple[float, float]]]:
|
||||
samples: list[tuple[GazeSample, tuple[float, float]]] = []
|
||||
try:
|
||||
while True:
|
||||
kind, value = self.messages.get_nowait()
|
||||
if kind == "gaze":
|
||||
sample: GazeSample = value # type: ignore[assignment]
|
||||
gaze = (
|
||||
sample.x_norm * SCREEN_WIDTH,
|
||||
(1.0 - sample.y_norm) * SCREEN_HEIGHT,
|
||||
)
|
||||
samples.append((sample, gaze))
|
||||
else:
|
||||
self.canvas.itemconfigure(self.network_status, text=str(value))
|
||||
except queue.Empty:
|
||||
pass
|
||||
if samples:
|
||||
self.gaze = samples[-1][1]
|
||||
self.position_circle(self.gaze_item, self.gaze, 14)
|
||||
if self.args.cursor:
|
||||
self.canvas.itemconfigure(self.gaze_item, state="normal")
|
||||
return samples
|
||||
|
||||
def record_sample(
|
||||
self,
|
||||
sample: GazeSample,
|
||||
gaze: tuple[float, float],
|
||||
now: float,
|
||||
) -> None:
|
||||
error = math.dist(gaze, self.target)
|
||||
error_cm = error * self.cm_per_px
|
||||
error_deg = math.degrees(2.0 * math.atan2(error_cm / 2.0, self.args.viewing_distance))
|
||||
hit = error <= HIT_RADIUS
|
||||
self.errors.append(error)
|
||||
self.samples += 1
|
||||
self.hits += int(hit)
|
||||
self.writer.writerow([
|
||||
f"{sample.timestamp:.6f}", self.round, self.block, self.trial + 1,
|
||||
f"{now - self.trial_started:.3f}", f"{self.target[0]:.1f}",
|
||||
f"{self.target[1]:.1f}", f"{gaze[0]:.1f}", f"{gaze[1]:.1f}",
|
||||
f"{error:.1f}", f"{error_cm:.3f}", f"{error_deg:.3f}", int(hit)
|
||||
])
|
||||
|
||||
def tick(self) -> None:
|
||||
gaze_samples = self.process_network()
|
||||
now = time.monotonic()
|
||||
if self.block in self.BLOCKS:
|
||||
elapsed = now - self.trial_started
|
||||
if self.block == "pursuit":
|
||||
phase = elapsed / 10.0 * math.tau
|
||||
self.target = (
|
||||
SCREEN_WIDTH / 2 + 760 * math.sin(phase),
|
||||
SCREEN_HEIGHT / 2 + 390 * math.sin(phase * 2),
|
||||
)
|
||||
self.position_shapes()
|
||||
for sample, gaze in gaze_samples:
|
||||
self.record_sample(sample, gaze, now)
|
||||
|
||||
if self.block == "fixation" and elapsed >= 3.0:
|
||||
self.advance_trial(10)
|
||||
elif self.block == "saccade":
|
||||
on_target = self.gaze is not None and math.dist(self.gaze, self.target) <= HIT_RADIUS
|
||||
if on_target:
|
||||
self.hit_started = self.hit_started or now
|
||||
else:
|
||||
self.hit_started = None
|
||||
if (self.hit_started and now - self.hit_started >= 0.15) or elapsed >= 3.0:
|
||||
self.advance_trial(10)
|
||||
elif self.block == "colors" and elapsed >= 1.75:
|
||||
self.advance_trial(20)
|
||||
elif self.block == "pursuit" and elapsed >= 10.0:
|
||||
self.advance_trial(3)
|
||||
|
||||
if self.root.winfo_exists():
|
||||
self.root.after(self.frame_interval_ms, self.tick)
|
||||
|
||||
def advance_trial(self, total: int) -> None:
|
||||
self.trial += 1
|
||||
if self.trial >= total:
|
||||
self.end_block()
|
||||
else:
|
||||
self.new_trial()
|
||||
|
||||
def end_block(self) -> None:
|
||||
mean_error = statistics.fmean(self.errors) if self.errors else float("nan")
|
||||
hit_rate = self.hits / self.samples * 100 if self.samples else 0.0
|
||||
mean_cm = mean_error * self.cm_per_px
|
||||
mean_deg = math.degrees(2.0 * math.atan2(mean_cm / 2.0, self.args.viewing_distance))
|
||||
summary = (f"round {self.round} {self.block}: mean error {mean_error:.0f} px / {mean_cm:.2f} cm / "
|
||||
f"{mean_deg:.2f} deg, on-target {hit_rate:.0f}%")
|
||||
self.summary_lines.append(summary)
|
||||
self.block = "between"
|
||||
self.hide_targets()
|
||||
self.result_file.flush()
|
||||
self.set_text("Block complete", f"{summary}\n\nPress SPACE for the next exercise")
|
||||
|
||||
def finish(self) -> None:
|
||||
self.block = "finished"
|
||||
self.hide_targets()
|
||||
self.result_file.flush()
|
||||
self.set_text(
|
||||
"All exercises complete",
|
||||
"\n".join(self.summary_lines) + f"\n\nResults saved to:\n{self.result_path}\n\nPress Esc to close",
|
||||
)
|
||||
|
||||
|
||||
def main() -> None:
|
||||
args = parse_args()
|
||||
validate_args(args)
|
||||
if not 0 <= args.confidence <= 1:
|
||||
raise SystemExit("error: --confidence must be between 0 and 1")
|
||||
if args.screen_diagonal <= 0:
|
||||
raise SystemExit("error: --screen-diagonal must be greater than 0")
|
||||
if args.viewing_distance <= 0:
|
||||
raise SystemExit("error: --viewing-distance must be greater than 0")
|
||||
if args.rounds < 1:
|
||||
raise SystemExit("error: --rounds must be at least 1")
|
||||
if args.animation_fps < 1:
|
||||
raise SystemExit("error: --animation-fps must be at least 1")
|
||||
enable_windows_dpi_awareness()
|
||||
root = tk.Tk()
|
||||
ExerciseWindow(root, args)
|
||||
root.mainloop()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user