steam-frame/DeeJanuz--frametop
1
0 Fork 0
mirror of https://github.com/DeeJanuz/frametop.git synced 2026-10-06 05:00:08 +02:00
Code Issues Packages Projects Releases 4 Wiki Activity
Files
d995b815f8a81c699350637b4e6b437b98883f3e
DeeJanuz--frametop/gaze/probe/ft-gazeprobe
T
DeeJanuzandClaude Opus 5.5 ed75614f62 Bring our own eye tracker into the gaze folder, run by the gaze service 
frame-eyes, a separate project until now, becomes gaze/tracker:
- ft-eyegrab (was fe-bufprobe) copies the eye-camera frames, read-only, out of
  SteamVR's eyetracking process. It runs as the system service
  frametop-eyegrab.service, which gaze/tracker/install.sh installs to
  /etc/frametop with sudo. It keeps only CAP_SYS_PTRACE, CAP_DAC_READ_SEARCH, and
  CAP_CHOWN, and copies frames only while /dev/shm/frametop-eyes-want is fresh,
  holding none of the tracker's buffers otherwise.
- ft-eyes (was fe-trackd) runs under ft-gazed in the dev container, with
  build/venv's pinned numpy and OpenCV: while Eye tracker is Own tracker, or on
  the probe's lease ("eyes SECONDS"). No sudo password or fe-live script at
  run time any more.
- lab/ holds the research tools (ft-eyes-score, -e2e, -record, -replay,
  -session) and findings.md. Recordings live outside the repo, in
  ~/.local/share/frametop/eyes/captures; .gitignore catches stray frame dumps.

Its socket is now @ft_eyes, its output /dev/shm/frametop-eyes-gaze, and its state
~/.local/state/frametop/gaze/eyes. On practice1 -> practice2 the whole live path
(ft-eyes-e2e) gives 1.30 deg median and 3.18 for the worst tenth, as before the
move (1.30, 3.21).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-09-30 10:00:07 -06:00

2891 lines
135 KiB
Python
Executable File
Raw Blame History

#!/usr/bin/python3
"""ft-gazeprobe: a playground for eye tracking as pointer input on the Frametop desktop.
Opens fullscreen on one Frametop screen and shows where the headset's eye tracker says
you're looking, from the sources ft-gaze reads (SteamVR's eye tracking action, the two
gaze sets in eye-server.mmap, and each eye alone). The modes:
Free look the gaze dot; the trigger calibrates wherever you are looking.
Accuracy test look at each target and tap the trigger; measures every source's error
(degrees and pixels) and jitter across the screen.
Click practice the white dot is a gaze pointer. Look at the target and press; if the
dot isn't on it, keep holding and move the mouse to drag it there. The
drag is learned as the tracker's error.
Snap practice desktop-like elements; the gaze snaps to the nearest one. Tap to click
it; if it's the wrong one, hold, then glance toward the right one or
move the mouse, and let go on it. Each click teaches the click
corrections (LiveCorrection).
Headset fit how well the tracker sees each eye (fitcheck.py): live per eye, whether
it's tracked, how open it is, and the tracker's confidence, and maps of
where you looked and where each eye got lost, with hints. Enter runs a
guided check (dots around the screen, then down, up, left and right),
R starts over. Adjust the headset while you watch it.
Run calibration: the initial calibration, after Apple Vision Pro's eye setup. One dot,
then six in a circle, in three rounds that go from a dark to a bright screen (pupil size
changes with brightness, and the tracker's error with it). Look at each highlighted dot
and press the trigger. At the end, each source's calibration is fitted from all 21 dots;
freeze and look refines it on demand after that.
Trigger: Enter, Space, or a mouse button. Tab shows and hides the panel, C collapses it to
its title bar (or its arrow button does), F11 toggles fullscreen, Esc quits.
Freeze and look (the default trigger): the press freezes the dot where the tracker says
you're looking. Look at the frozen dot. After a moment to settle, the probe averages where
the tracker now says you're looking; the difference is the tracker's error at that spot,
in degrees, and the calibration learns it, stored against where in your view (relative to
the head) you were looking. The live dot is hidden while frozen so it can't pull your eye.
Tap for a fixed capture, or hold to keep capturing until you let go.
The older nudge triggers are still there: hold, and move the frozen dot with your head or
eyes (gain times the movement); the move is what's learned.
Smoothing: Fixation lock (the default) holds the dot on the running mean of the current
fixation and jumps when the gaze leaves `fixation radius`; One Euro follows smoothly, with
beta per degree a second. Raw gaze jitters by about 0.25-0.3 degrees while you hold still.
Calibration models (per source, saved in ~/.local/state/frametop/gaze/calibration.json):
none raw gaze
offset one offset for the whole view
affine the offset plus a straight-line change across the view
affine+grid plus a correction per 10-degree cell of the view (bilinear), for error
that bends across the field of view
Attempts and test results are logged to ~/.local/state/frametop/gaze/ as JSON.
"""
import argparse
import json
import math
import os
import random
import signal
import socket
import statistics
import subprocess
import sys
import time
from collections import deque
from pathlib import Path
import gi
gi.require_version("Gtk", "4.0")
gi.require_version("Gdk", "4.0")
gi.require_version("Adw", "1")
from gi.repository import Adw, Gdk, Gio, GLib, Gtk # noqa: E402
REPO = Path(__file__).resolve().parents[2]
HELPER = REPO / "gaze" / "build" / "ft-gaze"
STATE = Path.home() / ".local" / "state" / "frametop" / "gaze"
# left and right: each eye alone (set 2's own reading of that eye), calibrated and tested
# like the rest, to see what one eye is worth against both. own: our own tracker
# (gaze/tracker/ft-eyes, which the gaze service runs); it has its own calibration.
SOURCES = ["action", "mmap1", "mmap2", "left", "right", "own"]
SOURCE_NAMES = {"action": "SteamVR action", "mmap1": "mmap set 1", "mmap2": "mmap set 2", "left": "Left eye",
"right": "Right eye", "own": "Own tracker"}
SOURCE_COLORS = {"action": (0.2, 0.8, 1.0), "mmap1": (1.0, 0.6, 0.1), "mmap2": (0.9, 0.3, 0.9),
"left": (0.4, 1.0, 0.6), "right": (1.0, 1.0, 0.4), "own": (1.0, 0.35, 0.35)}
TRIGGER_KEYS = {Gdk.KEY_Return, Gdk.KEY_KP_Enter, Gdk.KEY_space}
class OwnTracker:
"""Talks to our own tracker (gaze/tracker/ft-eyes) over its control socket (@ft_eyes).
It keeps its own calibration: the probe sends it calibration dots and clicks, each with
when you looked and where (head-relative degrees), and it learns from its own frames.
The gaze service (ft-gazed) runs it: `lease` asks it to keep it running a while longer,
whatever the Eye tracker setting says."""
def __init__(self):
self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
self.sock.bind(f"\0ft_gazeprobe.{os.getpid()}")
self.sock.settimeout(1.0)
def ask(self, command):
"""The reply line, or "fail ..." if the tracker isn't running or doesn't answer."""
try:
while True: # drop a late reply to an earlier command
self.sock.setblocking(False)
self.sock.recv(4096)
except (BlockingIOError, OSError):
pass
self.sock.settimeout(1.0)
try:
self.sock.sendto(command.encode(), "\0ft_eyes")
return self.sock.recv(4096).decode()
except (ConnectionRefusedError, FileNotFoundError):
return "fail the Own tracker isn't running yet (the gaze service starts it)"
except (socket.timeout, OSError) as e:
return f"fail no answer from the Own tracker ({e})"
def lease(self, seconds=30):
"""Ask the gaze service to keep the Own tracker running for `seconds` more. False if
the gaze service isn't running."""
try:
self.sock.sendto(f"eyes {seconds}".encode(), "\0ft_gazed")
return True
except OSError:
return False
# The math, filters, correction models, and SteamVR log reader are shared with ft-gazed.
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from gazecal import (DEFAULT_MODEL, MODELS, Correction, Fixation, LiveCorrection, OneEuro, # noqa: E402
SteamEyeLog, cross_validate, deg_from_px, px_from_deg, steady_samples)
from fitcheck import FitCheck # noqa: E402
# --- Gaze from ft-gaze ----------------------------------------------------------------
class GazeReader:
"""Runs ft-gaze in the dev container and hands each sample (a dict) to `on_sample`.
If ft-gaze exits (SteamVR restarting, or something killed it), it's started again."""
RETRY = 3 # seconds before starting ft-gaze again
def __init__(self, on_sample, on_status):
self.on_sample, self.on_status = on_sample, on_status
self.proc = None
self.stream = None
self.cancel = Gio.Cancellable()
self.stopping = False
self.last_err = ""
def start(self):
if self.stopping:
return False
self.cancel = Gio.Cancellable()
if not HELPER.exists():
self.on_status(f"ft-gaze isn't built: run {REPO}/gaze/build.sh")
return
env = dict(os.environ)
# The Frametop desktop has its own runtime dir; podman needs the real one.
env["XDG_RUNTIME_DIR"] = f"/run/user/{os.getuid()}"
subprocess.run([str(REPO / "scripts" / "container-up.sh")], env=env, check=False)
distrobox = Path.home() / ".local" / "bin" / "distrobox"
# ft-gaze quits when its stdin closes, which is the one thing distrobox passes on
# when we go away (even if we're killed).
self.proc = subprocess.Popen([str(distrobox), "enter", "dev", "--", str(HELPER), "--watch-stdin"], env=env,
stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
start_new_session=True)
out = Gio.UnixInputStream.new(self.proc.stdout.fileno(), False)
self.stream = Gio.DataInputStream.new(out)
err = Gio.DataInputStream.new(Gio.UnixInputStream.new(self.proc.stderr.fileno(), False))
self._read()
self._read_err(err)
self.on_status("starting ft-gaze...")
return False # (also a one-shot GLib timeout)
def _read(self):
self.stream.read_line_async(GLib.PRIORITY_HIGH, self.cancel, self._got_line)
def _got_line(self, stream, result):
try:
line, _ = stream.read_line_finish_utf8(result)
except GLib.Error:
return
if line is None:
self.restart()
return
try:
self.on_sample(json.loads(line))
except (ValueError, KeyError, TypeError) as e:
print(f"bad sample: {e}: {line[:200]}", file=sys.stderr)
self._read()
def _read_err(self, err):
def got(stream, result):
try:
line, _ = stream.read_line_finish_utf8(result)
except GLib.Error:
return
if line is None:
return
print(line, file=sys.stderr)
text = line.removeprefix("ft-gaze: ")
if not text.startswith(("action manifest", "eye-server.mmap open")):
self.last_err = text # worth repeating if ft-gaze stops (not the startup lines)
self.on_status(text)
stream.read_line_async(GLib.PRIORITY_DEFAULT, self.cancel, got)
err.read_line_async(GLib.PRIORITY_DEFAULT, self.cancel, got)
def restart(self):
"""ft-gaze's output ended: say why, and start it again shortly."""
if self.stopping:
return
self.cancel.cancel()
code = None
if self.proc:
if self.proc.stdin and not self.proc.stdin.closed:
self.proc.stdin.close()
try:
code = self.proc.wait(timeout=2)
except subprocess.TimeoutExpired:
pass
why = f"exit {code}" if code is not None else "no exit code"
if self.last_err:
why += f": {self.last_err}"
self.on_status(f"ft-gaze stopped ({why}); restarting in {self.RETRY} s")
print(f"ft-gaze stopped ({why})", file=sys.stderr)
self.last_err = ""
GLib.timeout_add_seconds(self.RETRY, self.start)
def stop(self):
self.stopping = True
self.cancel.cancel()
if not self.proc:
return
# Closing stdin ends ft-gaze (--watch-stdin), and distrobox with it. Killing distrobox
# first would cut that path and leave ft-gaze running in the container.
if self.proc.stdin and not self.proc.stdin.closed:
self.proc.stdin.close()
try:
self.proc.wait(timeout=2)
except subprocess.TimeoutExpired:
try:
os.killpg(self.proc.pid, signal.SIGTERM)
except ProcessLookupError:
pass
# --- Where KWin has the window --------------------------------------------------------
def frametop_bus():
"""The Frametop session's D-Bus address (KWin's), from its plasmashell, or None."""
for pid in os.listdir("/proc"):
if not pid.isdigit():
continue
try:
with open(f"/proc/{pid}/comm") as f:
if f.read().strip() != "plasmashell":
continue
with open(f"/proc/{pid}/environ", "rb") as f:
env = dict(kv.split(b"=", 1) for kv in f.read().split(b"\0") if b"=" in kv)
except OSError:
continue
if env.get(b"XDG_RUNTIME_DIR", b"").endswith(b"/frametop") and b"DBUS_SESSION_BUS_ADDRESS" in env:
return env[b"DBUS_SESSION_BUS_ADDRESS"].decode()
return None
class KWinWindow:
"""Asks KWin where our window is, since a Wayland client can't know, and moves it.
A small KWin script reports the window's rectangle and the outputs (KWin's logical
desktop coordinates) whenever they change, so windowed mode can still put the dot in
the right place. Leaving fullscreen, KWin puts a window back where it first placed it,
which for this one was far off every screen, so `center_on` moves it onto the screen
it was on.
"""
IFACE = "dev.frametop.GazeProbe.Window"
XML = f"""<node><interface name="{IFACE}">
<method name="Report"><arg type="s" direction="in"/></method>
</interface></node>"""
REPORT_JS = r"""
function report() {
var lines = [];
workspace.screens.forEach(function (s) {
var g = s.geometry;
lines.push(["output", s.name, g.x, g.y, g.width, g.height].join(" "));
});
workspace.windowList().forEach(function (w) {
if (w.pid != @PID@ || !w.normalWindow) return;
var g = w.frameGeometry;
lines.push(["window", g.x, g.y, g.width, g.height, w.fullScreen ? 1 : 0, w.output ? w.output.name : "-"].join(" "));
});
callDBus("@NAME@", "/", "@IFACE@", "Report", lines.join("\n"));
}
function hook(w) {
if (w.pid != @PID@) return;
["frameGeometryChanged", "fullScreenChanged", "outputChanged"].forEach(function (sig) {
try { w[sig].connect(report); } catch (e) {}
});
}
workspace.windowList().forEach(hook);
workspace.windowAdded.connect(function (w) { hook(w); report(); });
workspace.screensChanged.connect(report);
var timer = new QTimer();
timer.interval = 1000;
timer.timeout.connect(report);
timer.start();
report();
"""
CENTER_JS = r"""
workspace.windowList().forEach(function (w) {
if (w.pid != @PID@ || !w.normalWindow || w.fullScreen) return;
var out = null;
workspace.screens.forEach(function (s) { if (s.name == "@OUTPUT@") out = s; });
if (!out) return;
// A copy of the rectangle, changed and assigned back: an object literal is ignored.
var o = out.geometry, r = w.frameGeometry;
r.x = o.x + Math.max(0, (o.width - r.width) / 2);
r.y = o.y + Math.max(0, (o.height - r.height) / 2);
w.frameGeometry = r;
workspace.activeWindow = w;
});
"""
def __init__(self, on_report):
self.on_report = on_report
self.conn = None
self.pid = os.getpid()
self.name = f"dev.frametop.GazeProbeWindow.p{self.pid}"
self.plugin = f"ftgazeprobe{self.pid}"
self.dir = Path(GLib.get_user_runtime_dir())
self.moves = 0
def start(self):
addr = frametop_bus()
if not addr:
print("KWin: the Frametop desktop's bus wasn't found; windowed mode can't place the dot", file=sys.stderr)
return
try:
self.conn = Gio.DBusConnection.new_for_address_sync(
addr, Gio.DBusConnectionFlags.AUTHENTICATION_CLIENT | Gio.DBusConnectionFlags.MESSAGE_BUS_CONNECTION,
None, None)
self.conn.register_object("/", Gio.DBusNodeInfo.new_for_xml(self.XML).interfaces[0], self.on_call,
None, None)
Gio.bus_own_name_on_connection(self.conn, self.name, Gio.BusNameOwnerFlags.NONE, None, None)
self.run(self.REPORT_JS, self.plugin)
except GLib.Error as e:
print(f"KWin: {e.message}", file=sys.stderr)
self.conn = None
def on_call(self, conn, sender, path, iface, method, params, invocation):
if method != "Report":
invocation.return_dbus_error("org.freedesktop.DBus.Error.UnknownMethod", method)
return
outputs, window = {}, None
for line in params.unpack()[0].splitlines():
f = line.split()
if f[0] == "output" and len(f) == 6:
outputs[f[1]] = tuple(float(v) for v in f[2:6])
elif f[0] == "window" and len(f) == 7:
window = (*(float(v) for v in f[1:5]), f[5] == "1", f[6])
invocation.return_value(None)
self.on_report(outputs, window)
def kwin(self, method, args, sig):
return self.conn.call_sync("org.kde.KWin", "/Scripting", "org.kde.kwin.Scripting", method,
GLib.Variant(sig, args), None, Gio.DBusCallFlags.NONE, 2000, None).unpack()
def run(self, js, plugin):
path = self.dir / f"{plugin}.js"
path.write_text(js.replace("@PID@", str(self.pid)).replace("@NAME@", self.name)
.replace("@IFACE@", self.IFACE))
if self.kwin("isScriptLoaded", (plugin,), "(s)")[0]:
self.kwin("unloadScript", (plugin,), "(s)")
sid = self.kwin("loadScript", (str(path), plugin), "(ss)")[0]
self.conn.call_sync("org.kde.KWin", f"/Scripting/Script{sid}", "org.kde.kwin.Script", "run",
None, None, Gio.DBusCallFlags.NONE, 2000, None)
def center_on(self, output):
"""Move the (windowed) window to the middle of `output` (WL-0, ...)."""
if not self.conn:
return
self.moves += 1
plugin = f"{self.plugin}c{self.moves}"
try:
self.run(self.CENTER_JS.replace("@OUTPUT@", output), plugin)
except GLib.Error as e:
print(f"KWin: {e.message}", file=sys.stderr)
return
GLib.timeout_add(1500, lambda: (self.unload(plugin), False)[1])
def unload(self, plugin):
try:
if self.conn and not self.conn.is_closed():
self.kwin("unloadScript", (plugin,), "(s)")
except GLib.Error:
pass
(self.dir / f"{plugin}.js").unlink(missing_ok=True)
def stop(self):
self.unload(self.plugin)
# --- The window -----------------------------------------------------------------------
class Probe(Adw.ApplicationWindow):
def __init__(self, app, screen):
super().__init__(application=app, title="Frametop gaze probe")
STATE.mkdir(parents=True, exist_ok=True)
self.want_screen = screen
self.screen = screen or 1
self.monitors = {} # frametop screen number -> Gdk.Monitor
# Where the window's top left corner is on its screen (0, 0 when fullscreen), from
# KWin; None when KWin hasn't said (then windowed positions are off).
self.origin = None
self.set_default_size(1400, 1000)
self.kwin = KWinWindow(self.on_window_report)
# Gaze state
self.sample = None
self.last_arrival = 0.0
self.rate = 0.0
self.rate_count, self.rate_since = 0, time.monotonic()
self.filters = {}
self.fix = {}
self.cursor = None # (x, y): the dot, after correction and filtering
self.gaze = None # (x, y): the filtered gaze, before any hold adjustment
self.head = None # (x, y): where the head points on this screen
self.raw = {} # source -> (x, y, j, hy, hp, dpp) for this screen, or None
self.status = ""
# Hold to adjust
self.held = False
self.press = None # dict: cursor, gaze, head, j, hy, hp, time, target
self.clicks = [] # recent clicks to draw: (x, y, time, hit)
# Freeze and look (see on_press)
self.capture = None # dict while the dot is frozen and we measure where you look
self.result = None # the last capture, drawn for a moment: F, L, error, verdict
self.captures = []
# Modes
self.targets = [] # practice: the current target (x, y, radius)
self.test = None # accuracy test state
self.calib = None # calibration run state (see start_calibration)
self.calibrated_at = 0.0 # when the last calibration run finished (points since then refine it)
self.test_history = [] # (time, source, model, corrected mean error) of tests since
self.results = None
self.attempts = []
# Snap practice (see snap_layout)
self.snap = None
self.snap_log = []
# The mouse pointer in the window (the Frametop pointer). Only its movement is used:
# while a press is held, it drags the gaze pointer (click practice) or the highlight
# (snap practice). The probe never moves it: an earlier version steered it onto the
# gaze with the pointer helper's "move" commands, and lost the user's pointer when
# the helper's pointer went idle or a controller had the laser (the moves piled up).
self.pointer = None
self.practice = None # click practice: the press being held (see practice_press)
self.recent = deque(maxlen=60) # the last samples, for where you looked at a press
self.fitcheck = FitCheck() # Headset fit: how well the tracker sees each eye
self.models = {s: Correction() for s in SOURCES}
self.own = OwnTracker()
self.reseat_check = False # showing the one-dot check after the headset was off
self.reseat_skipped = False # S skipped this one
self.live = {s: LiveCorrection() for s in SOURCES}
self.steam = SteamEyeLog()
self.steam.poll()
self.saved_model = DEFAULT_MODEL
self.load_calibration()
self.build_ui()
# The model the saved calibration was fitted as, without refitting it.
self.last_model = self.saved_model
self.w_model.set_selected(MODELS.index(self.saved_model))
keys = Gtk.EventControllerKey()
keys.set_propagation_phase(Gtk.PropagationPhase.CAPTURE)
keys.connect("key-pressed", self.on_key)
keys.connect("key-released", self.on_key_up)
self.add_controller(keys)
self.reader = GazeReader(self.on_sample, self.on_status)
self.connect("close-request", self.on_close)
self.connect("realize", lambda *_: GLib.idle_add(self.place))
GLib.timeout_add(1000, self.tick_rate)
# --- UI ---
def build_ui(self):
css = Gtk.CssProvider()
css.load_from_string("""
.probe-panel { background: alpha(@window_bg_color, 0.93); border-radius: 14px; padding: 14px; }
.probe-panel label, .probe-panel button, .probe-panel dropdown { font-size: 17px; }
.probe-panel button { min-height: 40px; padding: 4px 16px; }
.probe-stats { font-family: monospace; font-size: 15px; }
""")
Gtk.StyleContext.add_provider_for_display(Gdk.Display.get_default(), css,
Gtk.STYLE_PROVIDER_PRIORITY_APPLICATION)
self.area = Gtk.DrawingArea(hexpand=True, vexpand=True)
self.area.set_draw_func(self.draw)
# Any button is the trigger, except the right one, which opens the menu.
click = Gtk.GestureClick(button=0)
click.connect("pressed", self.on_click)
click.connect("released", self.on_unclick)
self.area.add_controller(click)
motion = Gtk.EventControllerMotion()
motion.connect("motion", lambda _c, x, y: self.on_motion(x, y))
self.area.add_controller(motion)
self.build_menu()
overlay = Gtk.Overlay()
overlay.set_child(self.area)
self.panel = self.build_panel()
overlay.add_overlay(self.panel)
overlay.add_overlay(self.build_toolbar())
self.set_content(overlay)
# --- The context menu (right-click, the Menu key, or Shift+F10) ---
MODE_KEYS = ["free", "test", "practice", "snap", "fit"]
MODE_NAMES = ["Free look", "Accuracy test", "Click practice", "Snap practice", "Headset fit"]
def build_menu(self):
def action(name, fn, state=None):
"""A plain action, or a check item (boolean `state`): fn gets the requested state,
and the item's check follows what actually happened (see the notify hooks)."""
if state is None:
a = Gio.SimpleAction.new(name, None)
a.connect("activate", fn)
else:
a = Gio.SimpleAction.new_stateful(name, None, state)
a.connect("change-state", fn)
self.add_action(a)
return a
action("calibrate", lambda *_: self.start_calibration())
action("skip", lambda *_: self.calib_skip())
action("test", lambda *_: self.start_test())
action("refine", lambda *_: self.refine_calibration())
action("reset", lambda *_: self.reset_calibration())
action("clearclicks", lambda *_: self.clear_clicks())
self.model_action = Gio.SimpleAction.new_stateful("model", GLib.VariantType.new("s"),
GLib.Variant("s", DEFAULT_MODEL))
self.model_action.connect("activate", lambda a, v: self.w_model.set_selected(MODELS.index(v.get_string())))
self.add_action(self.model_action)
self.mode_action = Gio.SimpleAction.new_stateful("mode", GLib.VariantType.new("s"), GLib.Variant("s", "free"))
self.mode_action.connect("activate", lambda a, v: self.set_mode(v.get_string()))
self.add_action(self.mode_action)
self.panel_action = action("panel", self.on_panel_toggle, GLib.Variant("b", True))
self.collapse_action = action("collapse", lambda a, v: self.set_collapsed(v.get_boolean()),
GLib.Variant("b", False))
self.full_action = action("fullscreen", lambda a, v: self.toggle_fullscreen(), GLib.Variant("b", True))
action("quit", lambda *_: self.close())
menu = Gio.Menu()
calib = Gio.Menu()
calib.append("Run calibration", "win.calibrate")
calib.append("Skip this calibration dot", "win.skip")
calib.append("Start accuracy test", "win.test")
calib.append("Refine from calibration and tests", "win.refine")
calib.append("Clear click corrections", "win.clearclicks")
calib.append("Reset calibration", "win.reset")
menu.append_section(None, calib)
modes = Gio.Menu()
for key, label in zip(self.MODE_KEYS, self.MODE_NAMES):
item = Gio.MenuItem.new(label, None)
item.set_action_and_target_value("win.mode", GLib.Variant("s", key))
modes.append_item(item)
menu.append_submenu("Mode", modes)
models = Gio.Menu()
for key in MODELS:
item = Gio.MenuItem.new(key, None)
item.set_action_and_target_value("win.model", GLib.Variant("s", key))
models.append_item(item)
menu.append_submenu("Correction model", models)
view = Gio.Menu()
view.append("Panel", "win.panel")
view.append("Collapse panel", "win.collapse")
view.append("Fullscreen", "win.fullscreen")
view.append("Quit", "win.quit")
menu.append_section(None, view)
self.menu = Gtk.PopoverMenu.new_from_model(menu)
self.menu.set_parent(self.area)
self.menu.set_has_arrow(False)
self.connect("notify::fullscreened", lambda *_: self.full_action.set_state(
GLib.Variant("b", self.is_fullscreen())))
def open_menu(self, x=None, y=None):
if x is None: # from the keyboard: at the dot, or the middle
x, y = self.cursor or (self.area.get_width() / 2, self.area.get_height() / 2)
rect = Gdk.Rectangle()
rect.x, rect.y, rect.width, rect.height = int(x), int(y), 1, 1
self.menu.set_pointing_to(rect)
self.menu.popup()
def on_click(self, gesture, _n, x, y):
if gesture.get_current_button() == Gdk.BUTTON_SECONDARY:
if self.held:
return # right-click while the trigger is held: ignore
self.open_menu(x, y)
return
if self.calib or self.test:
# A mouse click goes through SteamVR's laser, and SteamVR takes every click as
# "I was looking where the laser is" for its own calibration: during a run that
# retrains it from the wrong spot and moves the raw gaze under us.
self.mouse_trigger = False
if self.calib:
self.calib["retry"] = "use Enter or Space: mouse clicks also retrain SteamVR's own calibration"
self.on_status("use Enter or Space during calibration and tests, not the mouse")
self.area.queue_draw()
return
self.pointer = (x, y)
self.mouse_trigger = True
self.trigger(True)
def on_unclick(self, gesture, _n, _x, _y):
if gesture.get_current_button() != Gdk.BUTTON_SECONDARY and getattr(self, "mouse_trigger", False):
self.mouse_trigger = False
self.trigger(False)
def on_panel_toggle(self, action, value):
self.panel.set_visible(value.get_boolean())
def set_mode(self, key):
if key in self.MODE_KEYS:
self.w_mode.set_selected(self.MODE_KEYS.index(key)) # on_setting keeps the menu in step
def build_toolbar(self):
"""Always there, top right: the keys only work once typing goes to this window."""
bar = Gtk.Box(spacing=8, halign=Gtk.Align.END, valign=Gtk.Align.START, margin_end=40, margin_top=40)
bar.add_css_class("probe-panel")
panel = Gtk.Button(label="Panel")
panel.connect("clicked", lambda *_: self.panel.set_visible(not self.panel.get_visible()))
self.panel.connect("notify::visible", lambda *_: self.panel_action.set_state(
GLib.Variant("b", self.panel.get_visible())))
self.w_windowed = Gtk.Button(label="Windowed")
self.w_windowed.connect("clicked", lambda *_: self.toggle_fullscreen())
quit_ = Gtk.Button(label="Quit")
quit_.add_css_class("destructive-action")
quit_.connect("clicked", lambda *_: self.close())
for b in (panel, self.w_windowed, quit_):
bar.append(b)
self.connect("notify::fullscreened", lambda *_: self.w_windowed.set_label(
"Windowed" if self.is_fullscreen() else "Fullscreen"))
return bar
def toggle_fullscreen(self):
if self.is_fullscreen():
m = self.monitors.get(self.screen)
size = None
if m:
g = m.get_geometry()
size = (min(1400, int(g.width * 0.8)), min(1100, int(g.height * 0.8)))
self.unfullscreen()
# GTK restores the size from before fullscreen; asking again once it's out of
# fullscreen resizes it to fit this screen.
if size:
GLib.timeout_add(150, lambda: (self.set_default_size(*size), False)[1])
# KWin would put it back where it first placed the window (off every screen).
output = f"WL-{self.screen - 1}"
for ms in (400, 1200):
GLib.timeout_add(ms, lambda: (self.kwin.center_on(output), False)[1])
else:
self.go_fullscreen()
def on_window_report(self, outputs, window):
"""KWin moved or resized us (or a periodic report): where are we on which screen."""
if not window:
return
x, y, _w, _h, full, output = window
if not (output.startswith("WL-") and output[3:].isdigit() and output in outputs):
return
screen = int(output[3:]) + 1
origin = (0.0, 0.0) if full else (x - outputs[output][0], y - outputs[output][1])
if screen != self.screen and screen in self.monitors:
self.screen = screen
self.filters.clear()
self.fix.clear()
order = sorted(self.monitors)
self.w_screen.set_selected(order.index(screen))
if origin != self.origin:
self.origin = origin
self.targets = []
if self.mode == "practice":
self.new_target()
if self.snap:
self.snap_layout()
self.area.queue_draw()
def dropdown(self, items, selected=0):
d = Gtk.DropDown.new_from_strings(items)
d.set_selected(selected)
d.connect("notify::selected", lambda *_: self.on_setting())
return d
def scale(self, lo, hi, step, value, digits=2):
s = Gtk.Scale.new_with_range(Gtk.Orientation.HORIZONTAL, lo, hi, step)
s.set_value(value)
s.set_digits(digits)
s.set_draw_value(True)
s.set_hexpand(True)
s.set_size_request(260, -1)
s.connect("value-changed", lambda *_: self.on_setting())
return s
def build_panel(self):
panel = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, halign=Gtk.Align.START,
valign=Gtk.Align.START, margin_start=40, margin_top=40)
panel.add_css_class("probe-panel")
# The title bar stays when the panel is collapsed, so the gaze dot isn't lost behind it.
head = Gtk.Box(spacing=12)
title = Gtk.Label(label="Gaze probe", xalign=0, hexpand=True)
title.add_css_class("title-2")
head.append(title)
self.w_collapse = Gtk.Button(icon_name="pan-up-symbolic", tooltip_text="Collapse the panel (C)")
self.w_collapse.connect("clicked", lambda *_: self.set_collapsed(not self.collapsed))
head.append(self.w_collapse)
panel.append(head)
box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=8, margin_top=8)
box.set_size_request(460, -1)
self.panel_body = box
panel.append(box)
# Wrapped labels ask for their whole text on one line, which made the panel as wide
# as the screen; max_width_chars keeps it to about the grid's width.
hint = Gtk.Label(label="Trigger: Enter, Space, or click. Right-click for the menu. Tab hides this panel, "
"C collapses it, Esc quits. Click the window once so the keys reach it.",
xalign=0, wrap=True, max_width_chars=45)
hint.add_css_class("dim-label")
box.append(hint)
grid = Gtk.Grid(column_spacing=12, row_spacing=6)
row = 0
def add(label, widget):
nonlocal row
grid.attach(Gtk.Label(label=label, xalign=0), 0, row, 1, 1)
grid.attach(widget, 1, row, 1, 1)
row += 1
# Which tracker drives the dot. SteamVR's gaze is still logged in the background with
# the Own tracker (for comparing), but not shown.
tracker = Gtk.Box(css_classes=["linked"])
self.w_steamvr = Gtk.ToggleButton(label="SteamVR", active=True)
self.w_own = Gtk.ToggleButton(label="Own tracker", group=self.w_steamvr)
tracker.append(self.w_steamvr)
tracker.append(self.w_own)
self.w_own.connect("toggled", lambda b: self.on_tracker())
add("Tracker", tracker)
self.w_screen = self.dropdown(["Screen 1"])
add("Screen", self.w_screen)
self.w_mode = self.dropdown(self.MODE_NAMES)
add("Mode", self.w_mode)
# mmap set 2 had the least jitter (0.25 deg vs 0.30 for the action, sitting still).
self.w_source = self.dropdown([SOURCE_NAMES[s] for s in SOURCES], 2)
self.w_source.connect("notify::selected", lambda *_: self.sync_tracker())
add("Source", self.w_source)
self.w_filter = self.dropdown(["Raw", "One Euro", "Fixation lock"], 2)
add("Smoothing", self.w_filter)
self.w_cutoff = self.scale(0.05, 3.0, 0.05, 0.5, 2)
add(" min cutoff (Hz)", self.w_cutoff)
self.w_beta = self.scale(0.0, 0.3, 0.005, 0.05, 3)
add(" beta (per deg/s)", self.w_beta)
self.w_radius = self.scale(0.3, 3.0, 0.1, 1.0, 1)
add(" fixation radius (deg)", self.w_radius)
self.w_model = self.dropdown(MODELS, MODELS.index(DEFAULT_MODEL))
add("Correction model", self.w_model)
self.w_rate = self.scale(0.05, 1.0, 0.05, 0.5, 2)
add(" learning rate", self.w_rate)
# The Frame's raw error reaches 8-9 degrees looking well up or down, so the limit is
# well above that; it's there to catch a look at the wrong spot, not real error.
self.w_limit = self.scale(1.0, 20.0, 0.5, 12.0, 1)
add(" max error (deg)", self.w_limit)
self.w_action = self.dropdown(["Freeze and look", "Nudge with head", "Nudge with eyes"])
add("Trigger", self.w_action)
self.w_settle = self.scale(100, 800, 25, 300, 0)
add(" settle (ms)", self.w_settle)
self.w_capture = self.scale(200, 2000, 50, 600, 0)
add(" capture (ms)", self.w_capture)
self.w_spread = self.scale(0.2, 3.0, 0.05, 1.0, 2)
add(" max spread (deg)", self.w_spread)
self.w_gain = self.scale(0.0, 1.5, 0.05, 1.0, 2)
add(" nudge gain", self.w_gain)
self.w_size = self.scale(8, 120, 2, 30, 0)
add("Target radius (px)", self.w_size)
self.w_grid = self.dropdown(self.TEST_LAYOUTS)
add("Test spots", self.w_grid)
self.w_ring = self.scale(4, 30, 1, 20, 0)
add("Calibration ring (deg)", self.w_ring)
box.append(grid)
checks = Gtk.FlowBox(selection_mode=Gtk.SelectionMode.NONE, max_children_per_line=2)
self.w_learn = Gtk.CheckButton(label="Learn from trigger", active=True)
self.w_autotest = Gtk.CheckButton(label="Test after calibration", active=True)
self.w_live = Gtk.CheckButton(label="Live dot while frozen", active=False)
self.w_all = Gtk.CheckButton(label="Show all sources", active=False)
self.w_rawdot = Gtk.CheckButton(label="Show raw dot", active=True)
self.w_cells = Gtk.CheckButton(label="Show degree grid", active=False)
self.w_snaps = Gtk.CheckButton(label="Learn from clicks", active=True)
for c in (self.w_learn, self.w_autotest, self.w_live, self.w_all, self.w_rawdot, self.w_cells, self.w_snaps):
c.connect("toggled", lambda *_: self.on_setting())
checks.append(c)
box.append(checks)
buttons = Gtk.FlowBox(selection_mode=Gtk.SelectionMode.NONE, max_children_per_line=3)
for label, fn in (("Run calibration", self.start_calibration), ("Start test", self.start_test),
("Refine calibration", self.refine_calibration),
("Reset calibration", self.reset_calibration), ("Clear stats", self.clear_stats)):
b = Gtk.Button(label=label)
b.connect("clicked", lambda _b, f=fn: f())
buttons.append(b)
box.append(buttons)
self.w_stats = Gtk.Label(xalign=0, yalign=0, wrap=True, selectable=False, max_width_chars=50)
self.w_stats.add_css_class("probe-stats")
box.append(self.w_stats)
return panel
@property
def collapsed(self):
return not self.panel_body.get_visible()
def set_collapsed(self, collapsed):
self.panel_body.set_visible(not collapsed)
self.w_collapse.set_icon_name("pan-down-symbolic" if collapsed else "pan-up-symbolic")
self.w_collapse.set_tooltip_text("Expand the panel (C)" if collapsed else "Collapse the panel (C)")
self.collapse_action.set_state(GLib.Variant("b", collapsed))
def under_panel(self, x0, y0, x1, y1, margin=20):
"""Whether a box on the canvas overlaps the panel, collapsed or not."""
if not self.panel.get_visible():
return False
# Collapsed: the title bar. Before the first layout: about the full panel.
px, py, pw, ph = (40, 40, 240, 70) if self.collapsed else (40, 40, 520, 960)
ok, r = self.panel.compute_bounds(self.area)
if ok and r.get_width() > 0:
px, py, pw, ph = r.get_x(), r.get_y(), r.get_width(), r.get_height()
return x0 < px + pw + margin and px - margin < x1 and y0 < py + ph + margin and py - margin < y1
def place(self):
"""Find the Frametop screens and go fullscreen on the chosen one."""
display = self.get_display()
mons = display.get_monitors()
self.monitors = {}
for i in range(mons.get_n_items()):
m = mons.get_item(i)
conn = m.get_connector() or ""
# KWin's nested outputs: WL-0 is Frametop screen 1, WL-1 screen 2, ...
if conn.startswith("WL-") and conn[3:].isdigit():
self.monitors[int(conn[3:]) + 1] = m
if not self.want_screen:
surface = self.get_surface()
here = display.get_monitor_at_surface(surface) if surface else None
for n, m in self.monitors.items():
if m == here:
self.screen = n
names = []
for n in sorted(self.monitors):
g = self.monitors[n].get_geometry()
names.append(f"Screen {n} ({g.width}x{g.height})")
order = sorted(self.monitors)
model = Gtk.StringList.new(names or ["Screen 1"])
self.w_screen.set_model(model)
if self.screen in order:
self.w_screen.set_selected(order.index(self.screen))
self.w_screen.connect("notify::selected", self.on_screen_choice)
self.go_fullscreen()
self.kwin.start()
self.reader.start()
return False
def on_screen_choice(self, *_):
order = sorted(self.monitors)
i = self.w_screen.get_selected()
if 0 <= i < len(order) and order[i] != self.screen:
self.screen = order[i]
self.filters.clear()
self.fix.clear()
self.cursor = None
self.targets = []
self.snap = None
if self.mode == "snap":
GLib.timeout_add(500, lambda: (self.snap_layout(), False)[1])
self.go_fullscreen()
def go_fullscreen(self):
m = self.monitors.get(self.screen)
if m:
self.fullscreen_on_monitor(m)
else:
self.fullscreen()
# --- Settings ---
@property
def mode(self):
return self.MODE_KEYS[self.w_mode.get_selected()]
@property
def source(self):
return SOURCES[self.w_source.get_selected()]
@property
def model_mode(self):
return MODELS[self.w_model.get_selected()]
@property
def model_label(self):
"""The model, plus the click corrections on top when they're on and have learned."""
n = len(self.live[self.source].samples)
return self.model_mode + (f" + {n} clicks" if n and self.w_snaps.get_active() else "")
def on_tracker(self):
"""The tracker toggle: the Own tracker, or SteamVR's set 2 (its quietest source)."""
own = self.w_own.get_active()
if own != (self.source == "own"):
self.w_source.set_selected(SOURCES.index("own") if own else SOURCES.index("mmap2"))
if own:
if not self.own.lease():
self.on_status("The gaze service isn't running, and it runs the Own tracker "
"(frametop-gaze.service, gaze/run.sh install)")
return
self.lease_at = time.monotonic()
reply = self.own.ask("status")
if reply.startswith("fail"):
self.on_status("Starting the Own tracker (a few seconds)…")
else:
st = json.loads(reply)
self.on_status("Own tracker: " + ("calibrated " + st["calibration"].get("made", "")
if st.get("calibration") else "not calibrated yet: Run calibration"))
def sync_tracker(self):
own = self.source == "own"
if self.w_own.get_active() != own:
(self.w_own if own else self.w_steamvr).set_active(True)
def on_setting(self):
model = self.model_mode
if model != getattr(self, "last_model", model):
self.refit_as(model)
self.refit_live()
self.last_model = model
if hasattr(self, "model_action"):
self.model_action.set_state(GLib.Variant("s", model))
for f in self.filters.values():
f.min_cutoff = self.w_cutoff.get_value()
f.beta = self.w_beta.get_value()
for f in self.fix.values():
f.radius = self.w_radius.get_value()
if self.mode != "practice":
self.targets = []
elif not self.targets:
self.new_target()
if self.mode != "test":
self.test = None
if self.mode != "practice":
self.practice = None
if self.mode != "snap":
self.snap = None
elif not self.snap:
self.snap_layout()
if self.mode == "fit" and getattr(self, "last_mode", None) != "fit" and hasattr(self, "collapse_action"):
self.set_collapsed(True) # the fit check needs the room
self.last_mode = self.mode
if hasattr(self, "mode_action"):
self.mode_action.set_state(GLib.Variant("s", self.mode))
self.update_stats()
self.area.queue_draw()
# --- Input ---
def on_key(self, _ctl, keyval, _code, state):
if keyval in TRIGGER_KEYS:
self.trigger(True)
return True
if keyval == Gdk.KEY_Menu or (keyval == Gdk.KEY_F10 and state & Gdk.ModifierType.SHIFT_MASK):
self.open_menu()
return True
if keyval == Gdk.KEY_Tab:
self.panel.set_visible(not self.panel.get_visible())
return True
if keyval in (Gdk.KEY_c, Gdk.KEY_C):
self.panel.set_visible(True)
self.set_collapsed(not self.collapsed)
return True
if keyval == Gdk.KEY_F11:
self.toggle_fullscreen()
return True
if keyval in (Gdk.KEY_r, Gdk.KEY_R) and self.mode == "fit":
self.fitcheck.reset()
self.area.queue_draw()
return True
if keyval == Gdk.KEY_BackSpace and self.mode == "snap":
self.snap_undo()
return True
if keyval in (Gdk.KEY_s, Gdk.KEY_S) and self.calib:
self.calib_skip()
return True
if keyval in (Gdk.KEY_s, Gdk.KEY_S) and self.reseat_check:
self.reseat_check, self.reseat_skipped = False, True
self.area.queue_draw()
return True
if keyval == Gdk.KEY_Escape:
if self.calib or self.test:
self.calib = self.test = None
self.panel.set_visible(True)
self.on_status("cancelled")
self.area.queue_draw()
else:
self.close()
return True
return False
def on_key_up(self, _ctl, keyval, _code, _state):
if keyval in TRIGGER_KEYS:
self.trigger(False)
return True
return False
def trigger(self, down):
if down == self.held:
return # key repeat, or a release we never saw the press of
self.held = down
if down:
self.on_press()
else:
self.on_release()
# --- Samples ---
def on_status(self, text):
self.status = text
self.update_stats()
def tick_rate(self):
now = time.monotonic()
self.rate = self.rate_count / max(1e-6, now - self.rate_since)
self.rate_count, self.rate_since = 0, now
if self.capture and now - self.capture["t0"] > 6:
self.finish_capture() # samples stopped coming (headset off?)
if self.steam.poll():
self.on_status("SteamVR's eye tracker just started again: its own calibration started over, "
"so the raw gaze may have moved")
self.poll_own()
self.update_stats()
return True
def poll_own(self):
"""With the Own tracker: after the headset was off (or the tracker restarted), its
next click starts each eye's shift over, and until then the first clicks can be
10-17 degrees off (gaze/tracker/findings.md, session 5). So ask for one look at a centre dot first,
as Varjo's headsets do each time they're put on. Also keeps the lease on the tracker."""
if self.source != "own":
return
if time.monotonic() - getattr(self, "lease_at", 0.0) > 10:
self.lease_at = time.monotonic()
self.own.lease()
if self.calib or self.capture:
return
reply = self.own.ask("status")
if reply.startswith("fail"):
return
st = json.loads(reply)
pending = bool(st.get("calibration")) and any(e.get("reseat") for e in st["eyes"].values())
if not pending:
self.reseat_skipped = False
check = pending and not self.reseat_skipped
if check != self.reseat_check:
self.reseat_check = check
self.area.queue_draw()
def reseat_press(self):
"""The press on the check dot: a click there teaches both eyes' shifts at once."""
if not self.recent:
return
t_end = self.recent[-1]["t"]
w, h = self.canvas_size()
d = self.screen_direction([smp for smp in self.recent if smp["t"] >= t_end - 0.6], (w / 2, h / 2))
if d is None:
self.on_status("no gaze samples on this screen: look at the dot and press again")
return
reply = self.own.ask(f"click {t_end:.6f} {d[0]:.4f} {d[1]:.4f}")
if reply.startswith("ok"):
self.reseat_check = False
self.on_status("Own tracker: checked after the headset was off")
else:
self.on_status(reply[5:] if reply.startswith("fail") else reply)
self.area.queue_draw()
def draw_reseat(self, cr, w, h):
cx, cy = w / 2, h / 2
cr.set_source_rgb(1, 1, 1)
cr.arc(cx, cy, 10, 0, 2 * math.pi)
cr.fill()
cr.set_source_rgb(0.1, 0.1, 0.1)
cr.arc(cx, cy, 3, 0, 2 * math.pi)
cr.fill()
ink = (0.9, 0.9, 0.9)
self.text(cr, 60, 70, "Own tracker: the headset was off, so it may sit differently now", ink, 26)
self.text(cr, 60, 108, self.status if self.status.startswith("no gaze") else
"Look at the dot and press (Enter, Space, or click). S skips.", ink, 20)
def on_sample(self, s):
self.sample = s
self.recent.append(s)
self.fitcheck.feed(s, time.monotonic())
self.rate_count += 1
self.last_arrival = time.monotonic()
t = s["t"]
self.raw = {}
for name in SOURCES:
src = s["src"].get(name) or {}
hit = src.get("hit")
if hit and hit["s"] == self.screen:
ox, oy = self.origin or (0, 0)
self.raw[name] = (hit["x"] - ox, hit["y"] - oy, hit["j"], src["hy"], src["hp"], hit["dpp"])
else:
self.raw[name] = None
head = s["head"].get("hit")
ox, oy = self.origin or (0, 0)
self.head = (head["x"] - ox, head["y"] - oy) if head and head["s"] == self.screen else None
# The Own tracker's eyes, each where it alone puts your gaze (left, right).
self.own_eyes = [(h["x"] - ox, h["y"] - oy) if h and h["s"] == self.screen else None
for h in ((s["src"].get("own") or {}).get("ehit") or [])]
r = self.raw.get(self.source)
if r is None:
self.gaze = None
if not self.held and not self.capture:
self.cursor = None
else:
x, y = self.corrected(self.source, r)
self.gaze = self.filtered(x, y, t, r[5])
if self.capture:
self.cursor = self.capture["F"]
self.collect_capture(x, y, r, s)
elif self.practice:
self.cursor = self.practice_point()
elif self.held and self.press:
self.cursor = self.adjusted()
else:
self.cursor = self.gaze
if self.snap:
self.snap_update()
if self.calib and self.calib.get("collect_until"):
self.collect_calib(s)
if self.test and self.test.get("collect_until"):
self.collect_test(s)
self.area.queue_draw()
def corrected(self, name, r):
x, y, j, hy, hp, _dpp = r
cy, cp = self.correction(name, hy, hp)
dx, dy = px_from_deg(j, cy, cp)
return x + dx, y + dy
def correction(self, name, hy, hp, snaps=None):
"""The whole correction at (hy, hp): the calibration, plus what snap and practice
clicks have taught since (when "Learn from clicks" is on). None for the Own tracker:
it learns from the clicks itself (own_click)."""
if name == "own":
return 0.0, 0.0
cy, cp = self.models[name].get(hy, hp, self.model_mode)
if snaps if snaps is not None else self.w_snaps.get_active():
ly, lp = self.live[name].get(hy, hp)
cy, cp = cy + ly, cp + lp
return cy, cp
def view(self, name):
"""The correction as summarize() looks it up (an object with get(hy, hp, mode))."""
probe = self
class View:
def get(self, hy, hp, _mode):
return probe.correction(name, hy, hp)
return View()
def refit_live(self):
"""The calibration changed under the click corrections: refit them against it."""
for name in SOURCES:
self.live[name].refit(self.models[name], self.model_mode)
def filtered(self, x, y, t, dpp):
kind = self.w_filter.get_selected()
if kind == 0:
return x, y
if kind == 1:
fx = self.filters.setdefault(("x", self.source), OneEuro(self.w_cutoff.get_value(), self.w_beta.get_value()))
fy = self.filters.setdefault(("y", self.source), OneEuro(self.w_cutoff.get_value(), self.w_beta.get_value()))
return fx(x, t, dpp), fy(y, t, dpp)
f = self.fix.setdefault(self.source, Fixation(self.w_radius.get_value()))
return f(x, y, t, dpp)
# --- Freeze and look ---
#
# The press freezes the dot at F, where the (corrected, smoothed) gaze says you're
# looking. Then you look at the frozen dot itself: it's a target right where you are
# looking, and it doesn't move. After `settle` ms (the eye getting there), the unsmoothed
# gaze is averaged for `capture` ms, longer if you keep holding. That average, L, is where
# the tracker puts your gaze while you look at F, so F - L is the tracker's error there,
# and the calibration learns it. The live dot is hidden meanwhile, so it can't pull your
# eye off the frozen one. A capture is thrown out if the gaze wandered (spread over
# `max spread`) or the error is implausible (over `max error`).
@property
def action(self):
return ["freeze", "head", "eyes"][self.w_action.get_selected()]
def start_capture(self):
if self.gaze is None:
return
r = self.raw.get(self.source)
self.capture = {"F": self.gaze, "t0": time.monotonic(), "released": None, "samples": [],
"source": self.source, "target": self.targets[0] if self.targets else None,
"press_j": r[2] if r else None}
self.result = None
self.cursor = self.gaze
self.area.queue_draw()
def collect_capture(self, x, y, r, s):
c = self.capture
elapsed = time.monotonic() - c["t0"]
settle, span = self.w_settle.get_value() / 1000, self.w_capture.get_value() / 1000
if elapsed >= settle:
c["taken"] = c.get("taken", 0) + 1
c["samples"].append(((x, y, r[2], r[3], r[4], r[5]), s))
if elapsed >= settle + span and (c["released"] is not None or elapsed >= settle + 4.0):
self.finish_capture()
def finish_capture(self):
c, self.capture = self.capture, None
self.cursor = self.gaze
# Blinks and moments the tracker lost an eye are judged over the whole look (see
# steady_samples), then dropped.
steady = {id(smp) for smp in steady_samples([smp for _, smp in c["samples"]])}
pts = [p for p, smp in c["samples"] if id(smp) in steady]
F = c["F"]
rec = {"time": time.time(), "mode": self.mode, "source": c["source"], "model": self.model_mode,
"F": F, "n": len(pts), "taken": c.get("taken", 0),
"settle_ms": self.w_settle.get_value(), "capture_ms": self.w_capture.get_value()}
verdict = None
if len(pts) < 10:
verdict = (f"only {len(pts)} of {c.get('taken', 0)} samples had both eyes tracked"
if c.get("taken", 0) >= 10 else "too few samples")
else:
# Medians (see summarize): a stray sample can't move the result or the spread far.
lx = statistics.median(p[0] for p in pts)
ly = statistics.median(p[1] for p in pts)
j = [statistics.fmean(p[2][k] for p in pts) for k in range(4)]
hy = statistics.median(p[3] for p in pts)
hp = statistics.median(p[4] for p in pts)
dpp = statistics.fmean(p[5] for p in pts)
spread = 1.4826 * statistics.median(math.hypot(p[0] - lx, p[1] - ly) for p in pts) * dpp
dy, dp = deg_from_px(j, F[0] - lx, F[1] - ly)
err = math.hypot(dy, dp)
rec.update(L=[lx, ly], hy=hy, hp=hp, dpp=dpp, spread_deg=spread, err_deg=err, delta_deg=[dy, dp])
if spread > self.w_spread.get_value():
verdict = f"gaze moved ({spread:.2f} deg spread)"
elif err > self.w_limit.get_value():
verdict = f"{err:.1f} deg: over the max (looked at the frozen dot?)"
elif self.w_learn.get_active():
self.models[c["source"]].learn(hy, hp, dy, dp, self.model_mode, self.w_rate.get_value())
self.save_calibration()
verdict = "learned"
else:
verdict = "measured"
rec["verdict"] = verdict
target = c["target"]
if target and c["press_j"]:
# Would a plain gaze click at the press have hit? That's the score to watch.
tx, ty, tr = target
ey, ep = deg_from_px(c["press_j"], tx - F[0], ty - F[1])
rec.update(target=[tx, ty, tr], hit=math.hypot(tx - F[0], ty - F[1]) <= tr,
press_err_px=math.hypot(tx - F[0], ty - F[1]), press_err_deg=math.hypot(ey, ep))
self.attempts.append(rec)
self.clicks.append((F[0], F[1], time.monotonic(), rec["hit"]))
self.clicks = self.clicks[-20:]
self.new_target()
self.captures.append(rec)
self.log("captures.jsonl", rec)
self.result = {**rec, "shown": time.monotonic()}
self.update_stats()
self.area.queue_draw()
# --- Nudge (hold, and move the frozen dot with your head or eyes) ---
def on_press(self):
if self.calib:
self.calib_press()
return
if self.reseat_check:
self.reseat_press()
return
if self.mode == "test":
self.test_press()
return
if self.mode == "snap":
self.snap_press()
return
if self.mode == "practice":
self.practice_press()
return
if self.mode == "fit":
self.fitcheck.toggle_guide(time.monotonic())
self.area.queue_draw()
return
if self.action == "freeze":
if not self.capture:
self.start_capture()
return
if self.cursor is None or self.gaze is None:
self.press = None
return
r = self.raw.get(self.source)
self.press = {"cursor": self.cursor, "gaze": self.gaze, "head": self.head,
"j": r[2] if r else None, "hy": r[3] if r else 0, "hp": r[4] if r else 0,
"dpp": r[5] if r else 0, "time": time.monotonic(),
"target": self.targets[0] if self.targets else None}
self.area.queue_draw()
def adjusted(self):
p = self.press
if not p or self.gaze is None:
return self.cursor
gain = self.w_gain.get_value()
if self.action == "eyes":
g = self.gaze
return p["cursor"][0] + gain * (g[0] - p["gaze"][0]), p["cursor"][1] + gain * (g[1] - p["gaze"][1])
if self.head and p["head"]:
return (p["cursor"][0] + gain * (self.head[0] - p["head"][0]),
p["cursor"][1] + gain * (self.head[1] - p["head"][1]))
return self.cursor
def on_release(self):
if self.mode in ("test", "fit"):
return
if self.mode == "snap":
self.snap_release()
return
if self.mode == "practice":
self.practice_release()
return
if self.capture:
self.capture["released"] = time.monotonic()
return
p, self.press = self.press, None
if not p or self.cursor is None or not p["j"]:
return
x, y = self.cursor
held = time.monotonic() - p["time"]
nx, ny = x - p["cursor"][0], y - p["cursor"][1]
dy, dp = deg_from_px(p["j"], nx, ny)
nudge = math.hypot(dy, dp)
target = p["target"]
hit = None
rec = {"time": time.time(), "mode": self.mode, "source": self.source, "model": self.model_mode,
"adjust": self.action, "gain": self.w_gain.get_value(),
"held_s": round(held, 3), "hy": p["hy"], "hp": p["hp"],
"press": p["cursor"], "release": [x, y], "nudge_deg": [dy, dp], "dpp": p["dpp"]}
if target:
tx, ty, tr = target
hit = math.hypot(x - tx, y - ty) <= tr
ey, ep = deg_from_px(p["j"], tx - p["cursor"][0], ty - p["cursor"][1])
rec.update(target=[tx, ty, tr], hit=hit, press_err_px=math.hypot(tx - p["cursor"][0], ty - p["cursor"][1]),
press_err_deg=math.hypot(ey, ep), final_err_px=math.hypot(x - tx, y - ty))
# Learn: the nudge is how far off the dot was here. A tap (under 0.2 s) adjusted
# nothing, so its "nudge" is only eye noise; a huge nudge, a long hold, or a miss is
# probably a change of mind, not calibration error.
learned = False
if (self.w_learn.get_active() and 0.2 <= held <= 5.0 and nudge <= self.w_limit.get_value()
and hit is not False):
self.models[self.source].learn(p["hy"], p["hp"], dy, dp, self.model_mode, self.w_rate.get_value())
self.save_calibration()
learned = True
rec["learned"] = learned
self.attempts.append(rec)
self.log("practice.jsonl", rec)
self.clicks.append((x, y, time.monotonic(), hit))
self.clicks = self.clicks[-20:]
if target:
self.new_target()
self.update_stats()
# --- Click practice ---
#
# A bullseye to look at. Look at its centre and press: the gap between the gaze just
# before the press and the centre is the tracker's error there, and the click
# corrections learn it (the same ones snap clicks teach). Each press is also scored:
# would a gaze click there have hit, with the correction as it was before this press?
def practice_press(self):
"""The press: the gaze pointer (the probe's dot) stops following your gaze, where it
was (F). Move the mouse to drag it onto what you were looking at, and let go
(practice_release)."""
if not self.targets:
self.new_target()
return
fix = {name: self.press_fixation(name) for name in SOURCES}
if not fix.get(self.source) or self.gaze is None:
self.on_status("no gaze on this screen at the press")
return
self.practice = {"t": time.monotonic(), "F": self.gaze, "fix": fix, "pointer0": self.pointer,
"target": self.targets[0], "t_raw": self.recent[-1]["t"] if self.recent else None}
self.area.queue_draw()
def practice_point(self):
"""Where the held gaze pointer is now: the frozen gaze point, moved by as much as the
mouse has moved since the press."""
p = self.practice
fx, fy = p["F"]
if self.pointer and p["pointer0"]:
fx += self.pointer[0] - p["pointer0"][0]
fy += self.pointer[1] - p["pointer0"][1]
return fx, fy
def practice_release(self):
p = self.practice
if not p:
return
(fx, fy), (px, py) = p["F"], self.practice_point()
self.practice = None
now = time.monotonic()
tx, ty, tr = p["target"]
mine = p["fix"][self.source]
dpp = mine["dpp"]
drag = math.hypot(px - fx, py - fy) * dpp
dragged = drag >= 0.4
rec = {"time": time.time(), "mode": "practice", "source": self.source, "model": self.model_mode,
"target": [tx, ty, tr], "press": [fx, fy], "release": [px, py], "drag_deg": drag, "dragged": dragged,
"held_s": round(now - p["t"], 3),
"hit_at_press": math.hypot(tx - fx, ty - fy) <= tr, "hit": math.hypot(tx - px, ty - py) <= tr,
"press_err_px": math.hypot(tx - fx, ty - fy),
"press_err_deg": math.hypot(*deg_from_px(mine["j"], tx - fx, ty - fy)),
# How close the drag put it to the target: how good the lesson was.
"release_err_deg": math.hypot(*deg_from_px(mine["j"], tx - px, ty - py)), "sources": {}}
learn = self.w_snaps.get_active() and dragged
verdict = "learned" if learn else "clicked (no drag, nothing to learn)" if not dragged else "measured"
for name, fix in p["fix"].items():
if not fix:
continue
# You were looking at the release point when you pressed: from the raw gaze to
# it is the whole error there.
oy, op = deg_from_px(fix["j"], px - fix["x"], py - fix["y"])
if name == "own":
rec["sources"][name] = {"hy": fix["hy"], "hp": fix["hp"], "raw": [fix["x"], fix["y"]],
"off": [oy, op], "n": fix["n"]}
if self.source == "own":
rec["own"] = self.own_click(p, fix["hy"] + oy, fix["hp"] + op, math.hypot(oy, op))
verdict = rec["own"]
continue
c = self.correction(name, fix["hy"], fix["hp"], snaps=True)
left = math.hypot(oy - c[0], op - c[1])
rec["sources"][name] = {"hy": fix["hy"], "hp": fix["hp"], "raw": [fix["x"], fix["y"]], "off": [oy, op],
"lesson_deg": left, "n": fix["n"]}
if not dragged:
continue
if left > self.SNAP_LEARN_MAX:
if name == self.source:
verdict = f"dragged {left:.1f} deg: too far to be the tracker's error, not learned"
continue
if learn:
self.live[name].add({"time": rec["time"], "hy": fix["hy"], "hp": fix["hp"], "dy": oy, "dp": op,
"wy": 1.0, "wp": 1.0, "how": "drag"}, self.models[name], self.model_mode)
if learn:
self.save_calibration()
rec["verdict"] = verdict
rec["learned"] = verdict == "learned" or verdict.startswith("taught")
self.attempts.append(rec)
self.log("practice.jsonl", rec)
# Drawn for a moment: the drag, from where the gaze put the pointer to where you let go.
self.result = {"F": [fx, fy], "L": [px, py], "err_deg": drag, "verdict": verdict, "shown": now}
self.clicks.append((px, py, now, rec["hit"]))
self.clicks = self.clicks[-20:]
self.new_target()
self.update_stats()
def own_click(self, p, yaw, pitch, off):
"""Teach the Own tracker: you were looking at (yaw, pitch) just before the press.
Every click counts, dragged or not (a click that needed no drag says so too), unless
the drag was too far to be the tracker's error."""
if not self.w_snaps.get_active():
return "clicked (learning is off)"
if off > self.OWN_LEARN_MAX:
return f"dragged {off:.1f} deg: too far to be the tracker's error, not learned"
if p.get("t_raw") is None:
return "no sample time at the press"
reply = self.own.ask(f"click {p['t_raw']:.6f} {yaw:.4f} {pitch:.4f}")
return "taught the Own tracker" if reply.startswith("ok") else reply
def on_motion(self, x, y):
self.pointer = (x, y)
if self.practice:
self.cursor = self.practice_point()
self.area.queue_draw()
def canvas_size(self):
w, h = self.area.get_width(), self.area.get_height()
if w > 0 and h > 0:
return w, h
m = self.monitors.get(self.screen)
return (m.get_geometry().width, m.get_geometry().height) if m else (1920, 1080)
def new_target(self):
w, h = self.canvas_size()
r = self.w_size.get_value()
margin = max(80, r * 2)
last = self.targets[0] if self.targets else None
for _ in range(50):
x, y = random.uniform(margin, w - margin), random.uniform(margin, h - margin)
if self.under_panel(x - r, y - r, x + r, y + r):
continue
if not last or math.hypot(x - last[0], y - last[1]) > min(w, h) * 0.25:
break
self.targets = [(x, y, r)]
self.area.queue_draw()
# --- Snap practice ---
#
# A desktop-like field of elements (toolbar icons, list rows, buttons, tiles, small
# links), some close together. The gaze snaps to the element nearest to where the
# corrected gaze is, and it's highlighted. Look at the orange one and press (Enter,
# Space, or a mouse button):
#
# tap clicks the highlighted element
# hold the highlight locks, and no longer follows your gaze. If it's on
# the wrong element, glance in the direction of the right one (a
# quick look off to that side and back): each glance steps the
# highlight to the next element that way. Or move the mouse: the
# highlight follows it from where it was. Let go on the right one.
#
# Whichever element you let go on is taken as the one you were looking at when you
# pressed, so the gap from the gaze at the press to it is the tracker's error there, and
# the click corrections (LiveCorrection) learn it. A glance works however far off the
# tracker is, because only the eye movement counts, and the tracker measures that
# well: its error barely changes over a couple of degrees. That's what it would learn in real
# use, where nothing knows which element you meant. The probe does know (the orange
# one), so each click is also scored: was the snap right at the press, was it right in
# the end, and was what got learned the element you meant. Backspace takes back the last
# click's lesson. Clicks are logged to snaps.jsonl.
SNAP_MAX = 4.0 # degrees from the gaze past which nothing is snapped
SNAP_KEEP = 0.4 # degrees another element has to be closer by to take the snap over
SNAP_LAYOUT_CLICKS = 12
FLICK = 1.5 # degrees the gaze has to move from where it was at the press to step
REARM = 0.9 # ... and come back within, before the next glance steps again
SNAP_LEARN_MAX = 6.0 # degrees: a lesson bigger than this (after the correction) is a wrong element
# The Own tracker takes bigger ones: after the headset is taken off and put back on, its
# first clicks were 11-17 degrees off, and a 6-degree limit kept them from teaching it
# (gaze/tracker/findings.md, session 5). It keeps the median of an eye's last 5 clicks, so one click
# where you changed your mind does little harm.
OWN_LEARN_MAX = 25.0
def snap_layout(self):
w, h = self.canvas_size()
r = self.raw.get(self.source)
ppd = 1 / r[5] if r and r[5] else 28.0 # pixels per degree
# Inside the calibrated area (the calibration ring), like the accuracy test.
radius = min(self.w_ring.get_value() * ppd, w / 2 - 40, h / 2 - 40)
cx, cy = w / 2, h / 2
def toolbar():
s, gap = random.uniform(1.2, 1.8), random.uniform(0.25, 0.6)
n = random.randint(6, 9)
if random.random() < 0.5:
return [(i * (s + gap), 0, s, s) for i in range(n)]
return [(0, i * (s + gap), s, s) for i in range(n - 2)]
def rows():
rh, rw = random.uniform(1.1, 1.6), random.uniform(7, 11)
return [(0, i * (rh + 0.12), rw, rh) for i in range(random.randint(5, 7))]
def buttons():
bw, bh, gap = random.uniform(3, 4.5), random.uniform(1.3, 1.8), random.uniform(0.4, 0.9)
return [(i * (bw + gap), 0, bw, bh) for i in range(random.randint(2, 3))]
def tiles():
s, gap = random.uniform(3, 4.2), 0.5
return [(c * (s + gap), k * (s + gap), s, s) for k in range(2) for c in range(3)]
def links():
return [(c * 1.6, k * 1.4, 0.9, 0.6) for k in range(2) for c in range(random.randint(2, 3))]
makers = [toolbar, rows, buttons, tiles, links, toolbar, links, buttons]
random.shuffle(makers)
elements, boxes = [], []
margin = 1.5 * ppd
for make in makers:
group = make()
gw = max(x + ew for x, _, ew, _ in group) * ppd
gh = max(y + eh for _, y, _, eh in group) * ppd
for _ in range(80):
ox = random.uniform(cx - radius, cx + radius - gw)
oy = random.uniform(cy - radius, cy + radius - gh)
corners = [(ox, oy), (ox + gw, oy), (ox, oy + gh), (ox + gw, oy + gh)]
if any(math.hypot(x - cx, y - cy) > radius * 1.05 for x, y in corners):
continue
if self.under_panel(ox, oy, ox + gw, oy + gh):
continue
if any(ox < b[2] + margin and b[0] < ox + gw + margin and oy < b[3] + margin and b[1] < oy + gh + margin
for b in boxes):
continue
boxes.append((ox, oy, ox + gw, oy + gh))
kind = make.__name__
for x, y, ew, eh in group:
elements.append({"x": ox + x * ppd, "y": oy + y * ppd, "w": ew * ppd, "h": eh * ppd, "kind": kind})
break
self.snap = {"elements": elements, "asked": None, "sel": None, "press": None, "count": 0,
"ppd": ppd, "guessed_ppd": not (r and r[5])}
self.snap_ask()
def snap_ask(self):
"""The next element to click, mostly small ones (where snapping is hard)."""
s = self.snap
els = s["elements"]
if not els:
s["asked"] = None
return
weights = [(3 if min(e["w"], e["h"]) / s["ppd"] < 2 else 1) if i != s["asked"] else 0
for i, e in enumerate(els)]
s["asked"] = random.choices(range(len(els)), weights=weights)[0] if sum(weights) else 0
@staticmethod
def centre(e):
return e["x"] + e["w"] / 2, e["y"] + e["h"] / 2
def snap_pick(self, x, y, dpp, current=None):
"""The element for a gaze at (x, y): the nearest by its edge (0 inside it), a
little by its centre to break ties; the current one stays unless another is
clearly nearer. None when nothing is within SNAP_MAX degrees."""
els = self.snap["elements"]
best, scores = None, {}
for i, e in enumerate(els):
dx = max(e["x"] - x, 0.0, x - e["x"] - e["w"])
dy = max(e["y"] - y, 0.0, y - e["y"] - e["h"])
edge = math.hypot(dx, dy) * dpp
if edge > self.SNAP_MAX:
continue
ex, ey = self.centre(e)
scores[i] = edge + 0.05 * math.hypot(x - ex, y - ey) * dpp
if best is None or scores[i] < scores[best]:
best = i
if current in scores and best is not None and scores[current] <= scores[best] + self.SNAP_KEEP:
return current
return best
def snap_update(self):
s = self.snap
r = self.raw.get(self.source)
if s.get("guessed_ppd") and r and r[5] and not s["count"] and not s["press"]:
self.snap_layout() # laid out before the first sample: again at the real scale
s = self.snap
dpp = r[5] if r and r[5] else 1 / s["ppd"]
p = s.get("press")
if not p:
if self.gaze is None:
s["sel"] = None
else:
s["sel"] = self.snap_pick(self.gaze[0], self.gaze[1], dpp, s["sel"])
return
# Held: the highlight is locked. The mouse moves it from where it was; once the mouse
# has moved, it has the highlight for the rest of the hold.
if self.pointer and p["mouse0"]:
mx, my = self.pointer[0] - p["mouse0"][0], self.pointer[1] - p["mouse0"][1]
if p["mouse_used"] or math.hypot(mx, my) * dpp > 0.5:
if not p["mouse_used"]:
p["mouse_used"] = True
p["mouse_from"] = s["sel"]
ex, ey = self.centre(s["elements"][p["mouse_from"]])
sel = self.snap_pick(ex + mx, ey + my, dpp, s["sel"])
if sel is not None and sel != s["sel"]:
s["sel"] = sel
p["steps"].append(("mouse", sel))
return
# A glance: the gaze leaves where it was at the press by FLICK degrees. One step per
# glance, toward the next element that way from the highlighted one; the gaze has to
# come back (within REARM) before the next glance counts.
if self.gaze is None or p["anchor"] is None:
return
gx, gy = self.gaze[0] - p["anchor"][0], self.gaze[1] - p["anchor"][1]
away = math.hypot(gx, gy) * dpp
if not p["armed"]:
if away < self.REARM:
p["armed"] = True
return
if away < self.FLICK:
return
p["armed"] = False
nxt = self.snap_neighbour(s["sel"], gx, gy)
if nxt is not None:
s["sel"] = nxt
p["steps"].append(("eyes", nxt))
def snap_neighbour(self, i, dx, dy):
"""The element next to element i in the direction (dx, dy): within 50 degrees of
it, nearest first, straight ahead preferred."""
els = self.snap["elements"]
cx, cy = self.centre(els[i])
norm = math.hypot(dx, dy)
best, best_cost = None, None
for k, e in enumerate(els):
if k == i:
continue
ex, ey = self.centre(e)
vx, vy = ex - cx, ey - cy
d = math.hypot(vx, vy)
if d < 1e-6:
continue
cos = (vx * dx + vy * dy) / (d * norm)
if cos < math.cos(math.radians(50)):
continue
cost = d * (1 + 2 * (1 - cos))
if best is None or cost < best_cost:
best, best_cost = k, cost
return best
@staticmethod
def steady_for(name, samples):
"""steady_samples, but for the Own tracker its own view of the eyes, not SteamVR's."""
if name != "own":
return steady_samples(samples)
return [smp for smp in samples if all((smp["src"].get("own") or {}).get("eyes") or [None])]
def press_fixation(self, name):
"""Where `name` put your gaze just before the press: the median of the last 300 ms,
without blinks, dropouts, or samples from before an eye movement in that time."""
if not self.recent:
return None
t_end = self.recent[-1]["t"]
pts = []
ox, oy = self.origin or (0, 0)
for smp in self.steady_for(name, [smp for smp in self.recent if smp["t"] >= t_end - 0.3]):
src = smp["src"].get(name) or {}
hit = src.get("hit")
if hit and hit["s"] == self.screen:
pts.append((hit["x"] - ox, hit["y"] - oy, hit["j"], src["hy"], src["hp"], hit["dpp"]))
if len(pts) < 3:
return None
for _ in range(2):
mx = statistics.median(q[0] for q in pts)
my = statistics.median(q[1] for q in pts)
near = [q for q in pts if math.hypot(q[0] - mx, q[1] - my) * q[5] <= 1.5]
if len(near) >= 3:
pts = near
return {"x": statistics.median(q[0] for q in pts), "y": statistics.median(q[1] for q in pts),
"hy": statistics.median(q[3] for q in pts), "hp": statistics.median(q[4] for q in pts),
"j": pts[-1][2], "dpp": pts[-1][5], "n": len(pts)}
def snap_press(self):
s = self.snap
if not s or s.get("press"):
return
fix = {name: self.press_fixation(name) for name in SOURCES}
if s["sel"] is None or not fix.get(self.source):
self.on_status("nothing to snap to: look at an element and press")
return
s["press"] = {"t": time.monotonic(), "via": "mouse" if getattr(self, "mouse_trigger", False) else "key",
"sel0": s["sel"], "fix": fix, "mouse0": self.pointer, "mouse_used": False, "mouse_from": None,
"anchor": self.gaze, "armed": True, "steps": []}
self.area.queue_draw()
def snap_release(self):
s = self.snap
p = s.get("press") if s else None
if not p:
return
s["press"] = None
final = s["sel"]
if final is None:
self.on_status("let go off every element: nothing clicked")
return
kinds = {k for k, _ in p["steps"]}
how = "mouse" if "mouse" in kinds else "eyes" if kinds else "tap"
e = s["elements"][final]
ex, ey = self.centre(e)
asked = s["asked"]
steam_start = self.steam.started()
rec = {"time": time.time(), "source": self.source, "model": self.model_mode, "how": how, "via": p["via"],
"steps": p["steps"], "moved_mouse": p["mouse_used"],
"held_s": round(time.monotonic() - p["t"], 3), "asked": asked, "sel0": p["sel0"], "final": final,
"right_at_press": p["sel0"] == asked, "right": final == asked,
"element": {k: e[k] for k in ("x", "y", "w", "h", "kind")},
"asked_element": {k: s["elements"][asked][k] for k in ("x", "y", "w", "h", "kind")} if asked is not None else None,
"snaps_on": self.w_snaps.get_active(), "steamvr_started": steam_start,
"usercal_since_calibration": self.steam.accepted_since(self.calibrated_at) if self.calibrated_at else None,
"usercal_since_start": self.steam.accepted_since(steam_start) if steam_start else None,
"sources": {}}
learn = self.w_snaps.get_active()
for name, fx in p["fix"].items():
if not fx:
continue
# The whole error at the press: from the raw gaze to the element you let go on.
dy, dp = deg_from_px(fx["j"], ex - fx["x"], ey - fx["y"])
# Each axis counts less the bigger the element is along it (where on it you
# looked isn't known): full weight up to about 1 degree across.
wy = 1 / (1 + (e["w"] / 2 * fx["dpp"]) ** 2)
wp = 1 / (1 + (e["h"] / 2 * fx["dpp"]) ** 2)
base = self.models[name].get(fx["hy"], fx["hp"], self.model_mode)
live = self.live[name].get(fx["hy"], fx["hp"])
g = {"raw": [fx["x"], fx["y"]], "hy": fx["hy"], "hp": fx["hp"], "n": fx["n"], "dpp": fx["dpp"],
"off": [dy, dp], "w": [wy, wp], "base": list(base), "live": list(live)}
if asked is not None:
ax, ay = self.centre(s["elements"][asked])
g["off_asked"] = list(deg_from_px(fx["j"], ax - fx["x"], ay - fx["y"]))
# Would the snap have been right with only the calibration, and with the snaps on top?
def pick(cy, cp, fx=fx):
dx, dyy = px_from_deg(fx["j"], cy, cp)
return self.snap_pick(fx["x"] + dx, fx["y"] + dyy, fx["dpp"])
g["pick_base"] = pick(*base)
g["pick_live"] = pick(base[0] + live[0], base[1] + live[1])
rec["sources"][name] = g
left = math.hypot(dy - base[0] - live[0], dp - base[1] - live[1])
g["lesson_deg"] = left
if left > self.SNAP_LEARN_MAX:
g["skipped"] = "too far off to be the element you looked at"
continue
if learn:
self.live[name].add({"time": rec["time"], "hy": fx["hy"], "hp": fx["hp"], "dy": dy, "dp": dp,
"wy": wy, "wp": wp, "how": how}, self.models[name], self.model_mode)
if learn:
self.save_calibration()
rec["learned"] = learn and not rec["sources"].get(self.source, {}).get("skipped")
self.snap_log.append(rec)
self.log("snaps.jsonl", rec)
self.clicks.append((ex, ey, time.monotonic(), final == asked))
self.clicks = self.clicks[-20:]
s["count"] += 1
if s["count"] % self.SNAP_LAYOUT_CLICKS == 0:
count = s["count"]
self.snap_layout()
self.snap["count"] = count
else:
self.snap_ask()
self.update_stats()
self.area.queue_draw()
def snap_undo(self):
"""Backspace: the last click was the wrong element, so don't learn from it."""
last = self.snap_log[-1] if self.snap_log else None
if not last or not last.get("learned") or last.get("undone"):
return
for name in last["sources"]:
self.live[name].undo(self.models[name], self.model_mode)
last["undone"] = True
self.log("snaps.jsonl", {"time": time.time(), "undo": last["time"]})
self.save_calibration()
self.on_status("took back the last click's correction")
def draw_snap(self, cr):
s = self.snap
p = s.get("press")
for i, e in enumerate(s["elements"]):
x, y, w, h = e["x"], e["y"], e["w"], e["h"]
rad = min(6.0, w / 4, h / 4)
self.round_rect(cr, x, y, w, h, rad)
cr.set_source_rgba(*((0.95, 0.55, 0.12) if i == s["asked"] else (0.26, 0.28, 0.33)), 0.95)
cr.fill()
if i == s["sel"]:
self.round_rect(cr, x - 3, y - 3, w + 6, h + 6, rad + 3)
cr.set_source_rgba(*((0.35, 0.85, 1.0) if p else (1, 1, 1)), 1)
cr.set_line_width(4 if p else 3)
cr.stroke()
elif p and i == p["sel0"]:
self.round_rect(cr, x - 3, y - 3, w + 6, h + 6, rad + 3)
cr.set_source_rgba(1, 1, 1, 0.5)
cr.set_line_width(2)
cr.set_dash([6, 5])
cr.stroke()
cr.set_dash([])
self.text(cr, 40, 60, "Snap practice: look at the orange element and tap Enter (or click).", (1, 1, 1), 26)
self.text(cr, 40, 96, "Wrong one highlighted? Hold the press, then glance toward the right one (look off to "
"that side and back) or move the mouse, and let go on it. Backspace takes back a click.",
(0.85, 0.85, 0.85), 19)
@staticmethod
def round_rect(cr, x, y, w, h, r):
cr.new_sub_path()
cr.arc(x + w - r, y + r, r, -math.pi / 2, 0)
cr.arc(x + w - r, y + h - r, r, 0, math.pi / 2)
cr.arc(x + r, y + h - r, r, math.pi / 2, math.pi)
cr.arc(x + r, y + r, r, math.pi, 1.5 * math.pi)
cr.close_path()
# --- Calibration run (after Apple Vision Pro's eye setup) ---
#
# Vision Pro's eye setup: look at one dot and pinch, then at each of six dots in a circle,
# in three rounds, each in brighter light than the one before (pupil size changes with
# brightness, and the tracker's error with it). Here the trigger is the pinch, the
# rounds dim the whole window dark, middle, then bright, and there's no gaze dot while
# it runs. Each later round turns the ring 20 degrees, so the 21 dots cover more of the
# view than the same 7 spots three times. Each dot's samples are checked (the gaze has
# to hold still); a dot that fails stays up to try again. At the end every source's
# calibration is fitted from all the dots, which replaces what it had learned; freeze
# and look then refines it from there. Keep your head facing the centre dot and move
# only your eyes, so the dots span your view.
ROUND_BG = [(0.03, 0.03, 0.035), (0.33, 0.33, 0.34), (0.8, 0.8, 0.8)]
ROUND_NAMES = ["dark", "medium", "bright"]
RING_SCALE = [1.0, 0.5, 1.0] # the middle round's ring is half the size
def calib_points(self, rnd):
"""The centre, then six on a ring of `Calibration ring` degrees (as much of it as fits
in the window), turned 20 degrees per round, and half the size in the middle round,
so the fit sees the middle, halfway out, and the edge of your view.
For the Own tracker: the centre, then eight directions on an oval out to `Calibration
ring` degrees each way (as much as fits across and up the window), turned 20 degrees
per round, and half the size in the middle round. Its fit is quadratic and goes
wrong past its dots, and the ring (limited by the window's height) left the sides
out: in practice2 (gaze/tracker/findings.md) the worst clicks were all past the ring."""
w, h = self.canvas_size()
cx, cy = w / 2, h / 2
r = self.raw.get(self.source)
px_per_deg = 1 / r[5] if r and r[5] else 32.0
reach = self.w_ring.get_value() * px_per_deg
if self.source == "own":
k = self.RING_SCALE[rnd]
rx = min(reach, cx - 80) * k
up, down = min(reach, cy - 150) * k, min(reach, cy - 80) * k # clear of the text at the top
pts = [(cx, cy)]
for i in range(8):
a = math.radians(-90 + rnd * 20 + i * 45)
pts.append((cx + rx * math.cos(a), cy + (up if math.sin(a) < 0 else down) * math.sin(a)))
return pts
radius = min(reach, w / 2 - 60, h / 2 - 60) * self.RING_SCALE[rnd]
pts = [(cx, cy)]
for k in range(6):
a = math.radians(-90 + rnd * 20 + k * 60)
pts.append((cx + radius * math.cos(a), cy + radius * math.sin(a)))
return pts
def start_calibration(self):
if self.source == "own":
# A fresh calibration for the Own tracker: it forgets the old one's clicks and
# shifts when this one is fitted.
reply = self.own.ask("calib-start")
if reply.startswith("fail"):
self.on_status(reply[5:])
return
self.test = None
self.results = None
self.capture = None
self.panel.set_visible(False)
self.calib = {"round": 0, "index": 0, "points": self.calib_points(0), "data": [], "collect_until": None,
"samples": [], "done_at": None, "retry": None,
"head": (self.sample["head"]["yaw"], self.sample["head"]["pitch"]) if self.sample else None}
self.on_status("calibrating")
self.area.queue_draw()
def calib_press(self):
c = self.calib
if c.get("collect_until") or c.get("done_at"):
return
# A retry listens longer: more samples to get past blinks and dropouts.
c["collect_until"] = time.monotonic() + (0.6 if not c.get("tries") else 1.1)
c["collect_from"] = time.monotonic() + 0.1 # skip the first 100 ms: the press itself
c["samples"] = []
c["retry"] = None
def calib_skip(self):
"""Give up on this dot (after it failed): the run goes on without it."""
c = self.calib
if not c or c.get("collect_until") or c.get("done_at"):
return
self.log("calibration-attempts.jsonl", {"time": time.time(), "round": c["round"], "index": c["index"],
"target": c["points"][c["index"]], "verdict": "skipped"})
c["done_at"] = time.monotonic()
GLib.timeout_add(150, self.calib_next)
def collect_calib(self, s):
c = self.calib
now = time.monotonic()
if now < c["collect_until"]:
if now >= c["collect_from"]:
c["samples"].append(s)
return
c["collect_until"] = None
target = c["points"][c["index"]]
entry = {"target": target, "round": c["round"], "sources": {}}
steady = steady_samples(c["samples"])
for name in SOURCES:
pts = []
for smp in steady:
src = smp["src"].get(name) or {}
hit = src.get("hit")
if hit and hit["s"] == self.screen:
ox, oy = self.origin or (0, 0)
pts.append((hit["x"] - ox, hit["y"] - oy, hit["j"], src["hy"], src["hp"], hit["dpp"], smp["t"]))
if len(pts) >= 20:
entry["sources"][name] = summarize(pts, target, self.models[name], "none")
mine = entry["sources"].get(self.source)
on_screen = sum(1 for smp in steady if ((smp["src"].get(self.source) or {}).get("hit") or {}).get("s") == self.screen)
opens = [min(o) for o in ((smp["src"].get("mmap1") or {}).get("open") for smp in c["samples"]) if o]
verg = [v for v in ((smp["src"].get("mmap1") or {}).get("lr") for smp in c["samples"]) if v is not None]
attempt = {"time": time.time(), "round": c["round"], "index": c["index"], "target": target,
"source": self.source, "samples": len(c["samples"]), "steady": len(steady), "on_screen": on_screen,
"open_median": statistics.median(opens) if opens else None,
"vergence_median": statistics.median(verg) if verg else None,
"sources": {n: {k: g[k] for k in ("err_deg", "sd_deg", "off_deg", "mean", "n")}
for n, g in entry["sources"].items()}}
verdict = None
if self.source == "own":
verdict = self.own_calib_point(c["samples"], target)
elif not mine:
if len(steady) < 20:
verdict = (f"only {len(steady)} of {len(c['samples'])} samples had both eyes tracked "
"(blinks, or the tracker lost an eye): open your eyes wide and press again")
else:
verdict = "no gaze on this screen: look at the dot and press again"
elif mine["sd_deg"] > self.w_spread.get_value():
verdict = (f"the gaze moved ({mine['sd_deg']:.2f} deg, the limit is {self.w_spread.get_value():.2f}): "
"hold it on the dot and press again")
elif mine["err_deg"] > self.w_limit.get_value():
verdict = f"{mine['err_deg']:.1f} deg off: look at the highlighted dot and press again"
elif self.other_dot(c, mine):
verdict = "that looked like another dot: look at the bright pulsing one and press again"
attempt["verdict"] = verdict or "accepted"
self.log("calibration-attempts.jsonl", attempt)
print(f"calibration round {c['round'] + 1} dot {c['index'] + 1}: {attempt['verdict']} "
f"(steady {len(steady)}/{len(c['samples'])})", file=sys.stderr, flush=True)
if verdict:
c["tries"] = c.get("tries", 0) + 1
c["retry"] = verdict + (" (S skips this dot)" if c["tries"] >= 2 else "")
return
c["tries"] = 0
if mine:
c.setdefault("seen", {})[c["index"]] = tuple(mine["mean"])
c["data"].append(entry)
c["done_at"] = now # a short pause on the filled dot, then the next one
GLib.timeout_add(350, self.calib_next)
def own_calib_point(self, samples, target):
"""Send one calibration dot to the Own tracker: the time you looked at it and its
direction. None if accepted, else why not."""
d = self.screen_direction(samples, target)
if d is None or len(samples) < 2:
return "no gaze samples on this screen: look at the dot and press again"
reply = self.own.ask(f"calib-point {samples[0]['t']:.6f} {samples[-1]['t']:.6f} {d[0]:.4f} {d[1]:.4f}")
if reply.startswith("ok"):
return None
return reply[5:] + ": look at the dot and press again"
def screen_direction(self, samples, target):
"""The head-relative direction (yaw, pitch) of a point on the canvas, from the screen
geometry around it: SteamVR's nearby gaze and its pixels-per-degree there (only the
geometry is used, not where SteamVR thinks you looked). None without samples."""
pts = []
ox, oy = self.origin or (0, 0)
for smp in samples:
for name in ("mmap1", "mmap2", "action"):
src = smp["src"].get(name) or {}
hit = src.get("hit")
if hit and hit["s"] == self.screen:
pts.append((hit["x"] - ox, hit["y"] - oy, hit["j"], src["hy"], src["hp"]))
break
if len(pts) < 5:
return None
x = statistics.median(q[0] for q in pts)
y = statistics.median(q[1] for q in pts)
j = [statistics.fmean(q[2][k] for q in pts) for k in range(4)]
oy_, op_ = deg_from_px(j, target[0] - x, target[1] - y)
return statistics.median(q[3] for q in pts) + oy_, statistics.median(q[4] for q in pts) + op_
def other_dot(self, c, g):
"""Was the gaze on another dot of this round rather than the current one?
For a dot already done this round, the tracker's answer when you looked at it is
known (`seen`), error included, so a look back at it lands in about the same place;
for the others, only where they're drawn is known. Only when it's clearly so: within
1.5 degrees of that, and less than half as far from it as from the current dot (the
raw error alone can be 8 degrees or more)."""
mx, my = g["mean"]
tx, ty = c["points"][c["index"]]
to_target = math.hypot(tx - mx, ty - my)
seen = c.get("seen", {})
for k, (x, y) in enumerate(c["points"]):
if k == c["index"] or (c["index"] == 0 and k > 0):
continue
ex, ey = seen.get(k, (x, y))
d = math.hypot(ex - mx, ey - my)
if d * g["dpp"] < 1.5 and d < 0.5 * to_target:
return True
return False
def calib_next(self):
c = self.calib
if not c:
return False
c["done_at"] = None
c["tries"] = 0
c["retry"] = None
c["index"] += 1
if c["index"] >= len(c["points"]):
c["round"] += 1
c["index"] = 0
c["seen"] = {}
if c["round"] >= len(self.ROUND_BG):
self.finish_calibration()
return False
c["points"] = self.calib_points(c["round"])
self.area.queue_draw()
return False
def finish_calibration(self):
data = self.calib["data"]
self.calib = None
mode = self.model_mode if self.model_mode != "none" else DEFAULT_MODEL
for name in SOURCES:
self.live[name] = LiveCorrection() # a new calibration: snaps start over on top of it
pts = [(e["sources"][name]["hy"], e["sources"][name]["hp"], *e["sources"][name]["off_deg"])
for e in data if name in e["sources"]]
if pts:
self.models[name].fit(pts, mode)
self.save_calibration()
# How well the fit explains the dots it came from (optimistic: the accuracy test,
# on new spots, is the honest check).
for e in data:
for name, g in e["sources"].items():
cy, cp = self.models[name].get(g["hy"], g["hp"], mode)
g["cerr_deg"] = math.hypot(g["off_deg"][0] - cy, g["off_deg"][1] - cp)
self.results = self.summarize_run(data, "calibration", mode)
self.results["rounds"] = self.ROUND_NAMES
self.log(time.strftime("calibration-%Y%m%d-%H%M%S.json"), self.results, single=True)
self.calibrated_at = self.results["time"]
self.test_history = []
self.log_points("calibration", data)
self.save_calibration()
own_note = ""
if self.source == "own":
reply = self.own.ask("calib-fit")
own_note = ("Own tracker: " + reply[3:]) if reply.startswith("ok") else ("Own tracker: " + reply)
print(f"calibration: {own_note}", file=sys.stderr, flush=True)
if self.model_mode == "none":
self.last_model = mode # just fitted
self.w_model.set_selected(MODELS.index(mode))
if self.w_autotest.get_active():
# The check on spots the calibration hasn't seen, right away: same head
# position, same session of SteamVR's own calibration.
self.on_status(f"calibrated ({mode}) from {len(data)} dots; now testing it on new spots. {own_note}")
GLib.timeout_add(1500, lambda: (self.start_test(), False)[1])
else:
self.panel.set_visible(True)
self.on_status(f"calibrated ({mode}) from {len(data)} dots; Start test to check it. {own_note}")
def draw_calib(self, cr, w, h):
c = self.calib
bg = self.ROUND_BG[c["round"]]
bright = bg[0] > 0.5
ink = (0.05, 0.05, 0.05) if bright else (1, 1, 1)
now = time.monotonic()
for k, (x, y) in enumerate(c["points"]):
if c["index"] == 0 and k > 0:
break # the centre dot comes alone first, then the ring
current = k == c["index"]
done = k < c["index"] or (current and c.get("done_at"))
# The current dot has to be the most striking thing on the screen: when finished
# dots were drawn solid, eyes went back to the (finished) centre dot in the middle
# round, whose ring is close to it. Finished dots are now faint specks.
if done and not current:
cr.set_source_rgba(*ink, 0.18)
cr.arc(x, y, 3, 0, 2 * math.pi)
cr.fill()
elif current:
collecting = c.get("collect_until")
accent = (0.1, 0.55, 0.2) if bright else (0.4, 1, 0.5)
col = accent if collecting or c.get("done_at") else ink
pulse = 1 + 0.25 * math.sin(now * 6)
cr.set_source_rgba(*col, 0.35)
cr.arc(x, y, 26 * pulse, 0, 2 * math.pi)
cr.fill()
cr.set_source_rgba(*col, 1)
cr.arc(x, y, 10, 0, 2 * math.pi)
cr.fill()
cr.set_source_rgba(*((1, 1, 1) if bright else (0, 0, 0)), 1)
cr.arc(x, y, 2.5, 0, 2 * math.pi) # a point to look at, in the middle
cr.fill()
else:
cr.set_source_rgba(*ink, 0.25)
cr.set_line_width(1.5)
cr.arc(x, y, 7, 0, 2 * math.pi)
cr.stroke()
GLib.idle_add(self.area.queue_draw) # the pulse
total = len(self.ROUND_BG) * len(c["points"])
n = c["round"] * len(c["points"]) + c["index"] + 1
self.text(cr, 60, 70, f"Calibration ({'Own tracker' if self.source == 'own' else 'SteamVR'}): round {c['round'] + 1} of 3 ({self.ROUND_NAMES[c['round']]}), dot {n} of {total}",
ink, 26)
hint = "Face the centre dot, look at the highlighted dot, and press. Esc cancels, S skips a dot."
self.text(cr, 60, 108, c.get("retry") or hint, (0.8, 0.2, 0.1) if c.get("retry") and bright else
(1, 0.55, 0.45) if c.get("retry") else ink, 20)
if self.sample and c.get("head"):
dy = abs(self.sample["head"]["yaw"] - c["head"][0])
dp = abs(self.sample["head"]["pitch"] - c["head"][1])
if max(dy, dp) > 10:
self.text(cr, 60, 140, "Your head has turned: face the centre dot again", ink, 20)
# --- Accuracy test ---
TEST_LAYOUTS = ["Calibrated area", "Window 5 x 3", "Window 7 x 4", "Window 9 x 5"]
def test_points(self):
"""Calibrated area: 15 new spots inside the calibration ring (the centre, 7 halfway
out, 7 near the ring), turned so none sits on a calibration dot, with a little
jitter. It checks the calibration where it was made, with your head facing the
centre as it was then. The window grids reach past it, to see how it holds up
beyond (on a wide screen that's far more than eyes turn without the head)."""
w, h = self.canvas_size()
layout = self.w_grid.get_selected()
if layout == 0:
cx, cy = w / 2, h / 2
r = self.raw.get(self.source)
px_per_deg = 1 / r[5] if r and r[5] else 32.0
radius = min(self.w_ring.get_value() * px_per_deg, w / 2 - 60, h / 2 - 60)
pts = [(cx, cy)]
for ring, turn in ((0.55, 10), (0.92, 40)):
for k in range(7):
a = math.radians(-90 + turn + k * 360 / 7 + random.uniform(-8, 8))
rr = radius * ring * random.uniform(0.93, 1.05)
pts.append((cx + rr * math.cos(a), cy + rr * math.sin(a)))
else:
cols, rows = [(5, 3), (7, 4), (9, 5)][layout - 1]
mx, my = w * 0.07, h * 0.1
pts = []
for r in range(rows):
for c in range(cols):
# A little jitter, so a calibration fit on one run is checked on new spots.
jx = random.uniform(-0.25, 0.25) * (w - 2 * mx) / max(1, cols - 1)
jy = random.uniform(-0.25, 0.25) * (h - 2 * my) / max(1, rows - 1)
pts.append((mx + c * (w - 2 * mx) / max(1, cols - 1) + jx,
my + r * (h - 2 * my) / max(1, rows - 1) + jy))
random.shuffle(pts)
return pts, self.TEST_LAYOUTS[layout]
def start_test(self):
self.w_mode.set_selected(1)
pts, layout = self.test_points()
self.panel.set_visible(False)
self.results = None
self.test = {"points": pts, "index": 0, "data": [], "collect_until": None, "samples": [], "layout": layout}
self.update_stats()
self.area.queue_draw()
def test_press(self):
if not self.test:
self.start_test() # the trigger in test mode starts one
return
t = self.test
if t.get("collect_until") or t["index"] >= len(t["points"]):
return
t["collect_until"] = time.monotonic() + 0.6
t["samples"] = []
def collect_test(self, s):
t = self.test
now = time.monotonic()
if now < t["collect_until"]:
if now > t["collect_until"] - 0.5: # skip the first 100 ms: the key press itself
t["samples"].append(s)
return
t["collect_until"] = None
target = t["points"][t["index"]]
entry = {"target": target, "sources": {}}
for name in SOURCES:
pts = []
ox, oy = self.origin or (0, 0)
for smp in steady_samples(t["samples"]):
src = smp["src"].get(name) or {}
hit = src.get("hit")
if hit and hit["s"] == self.screen:
pts.append((hit["x"] - ox, hit["y"] - oy, hit["j"], src["hy"], src["hp"], hit["dpp"], smp["t"]))
if len(pts) >= 5:
entry["sources"][name] = summarize(pts, target, self.view(name), self.model_mode)
t["data"].append(entry)
t["index"] += 1
if t["index"] >= len(t["points"]):
self.finish_test()
self.update_stats()
def finish_test(self):
self.results = self.summarize_run(self.test["data"], "test", self.model_label)
self.results["layout"] = self.test.get("layout")
self.results["regions"] = {n: region_errors(self.test["data"], n) for n in SOURCES}
v = self.results["summary"].get(self.source)
if v:
self.test_history.append((self.results["time"], self.source, self.model_mode, v["corrected_mean_deg"],
self.results["layout"]))
self.results["history"] = self.test_history
self.log(time.strftime("test-%Y%m%d-%H%M%S.json"), self.results, single=True)
self.log_points("test", self.test["data"])
self.test = None
self.panel.set_visible(True)
def summarize_run(self, data, kind, model):
results = {"kind": kind, "screen": self.screen, "model": model, "time": time.time(), "targets": data,
"summary": {}}
for name in SOURCES:
got = [e["sources"][name] for e in data if name in e["sources"]]
if not got:
continue
err = [g["err_deg"] for g in got]
cerr = [g["cerr_deg"] for g in got]
results["summary"][name] = {
"n": len(got), "mean_deg": statistics.fmean(err), "median_deg": statistics.median(err),
"max_deg": max(err), "within_1deg": sum(e <= 1 for e in err) / len(err),
"corrected_mean_deg": statistics.fmean(cerr),
"mean_px": statistics.fmean(g["err_px"] for g in got),
"offset_deg": [statistics.fmean(g["off_deg"][0] for g in got),
statistics.fmean(g["off_deg"][1] for g in got)],
"jitter_deg": statistics.fmean(g["jitter_deg"] for g in got),
"sd_deg": statistics.fmean(g["sd_deg"] for g in got)}
return results
REFINE_MODELS = ["offset", "affine", "quadratic", "quadratic+grid"]
def load_points(self, source):
"""(hy, hp, dy, dp) for every calibration dot and test target of `source` since the
last calibration run: dy, dp is the whole error there (raw gaze to target)."""
pts = []
try:
with open(STATE / "points.jsonl") as f:
for line in f:
r = json.loads(line)
if (r["source"] == source and r["time"] >= self.calibrated_at - 1
and r.get("layout") in (None, self.TEST_LAYOUTS[0])):
pts.append((r["hy"], r["hp"], r["off"][0], r["off"][1]))
except (OSError, ValueError):
pass
return pts
def refit_as(self, mode):
"""A model chosen by hand: fit it from the calibration dots and tests since (the
points.jsonl record), so the new terms (a grid, say) aren't left empty."""
if mode == "none":
self.on_status("correction off (none)")
return
n = 0
for name in SOURCES:
pts = self.load_points(name)
if pts:
self.models[name].fit(pts, mode)
n = max(n, len(pts))
if n:
self.save_calibration()
self.on_status(f"correction model {mode}, fitted from {n} points")
else:
self.on_status(f"correction model {mode}: no calibration points yet, run a calibration")
def refine_calibration(self):
"""Refit from the calibration dots plus every test since, with whichever model does
best on points it wasn't fitted on (leave-one-out), and use it. Test again after:
each test adds its targets, so test, refine, test is the loop."""
pts = self.load_points(self.source)
if len(pts) < 8:
self.on_status("refine needs a calibration run (and ideally a test) first")
return
scores = {}
for mode in self.REFINE_MODELS:
errs = cross_validate(pts, mode)
scores[mode] = (statistics.fmean(errs), statistics.median(errs), max(errs))
best = min(scores, key=lambda m: scores[m][0])
for name in SOURCES:
p = self.load_points(name)
if p:
self.models[name].fit(p, best)
self.last_model = best # already fitted: on_setting mustn't refit it
self.w_model.set_selected(MODELS.index(best))
self.refit_live()
self.save_calibration()
self.refine_scores = {"points": len(pts), "scores": scores, "best": best, "time": time.time()}
self.log("refinements.jsonl", {**self.refine_scores, "source": self.source})
self.on_status(f"refined from {len(pts)} points: {best}, "
f"{scores[best][0]:.2f} deg on points it wasn't fitted on; test again to check")
# --- Calibration storage and logs ---
def load_calibration(self):
try:
d = json.loads((STATE / "calibration.json").read_text())
for name in SOURCES:
if name in d:
self.models[name].from_json(d[name])
self.calibrated_at = float(d.get("_meta", {}).get("calibrated_at", 0.0))
if d.get("_meta", {}).get("model") in MODELS:
self.saved_model = d["_meta"]["model"]
for name, samples in (d.get("_live") or {}).items():
if name in self.live and isinstance(samples, list):
self.live[name].samples = samples[-LiveCorrection.KEEP:]
self.live[name].refit(self.models[name], self.saved_model)
except (OSError, ValueError):
pass
def save_calibration(self):
path = STATE / "calibration.json"
tmp = path.with_suffix(".tmp")
d = {n: m.to_json() for n, m in self.models.items()}
d["_meta"] = {"calibrated_at": self.calibrated_at,
"model": self.model_mode if hasattr(self, "w_model") else self.saved_model}
d["_live"] = {n: lc.samples for n, lc in self.live.items() if lc.samples}
tmp.write_text(json.dumps(d, indent=1))
tmp.replace(path)
def clear_clicks(self):
"""Forget what snap and practice clicks taught; the calibration stays."""
self.live = {s: LiveCorrection() for s in SOURCES}
self.save_calibration()
self.on_status("click corrections cleared")
def reset_calibration(self):
for m in self.models.values():
m.reset()
self.live = {s: LiveCorrection() for s in SOURCES}
self.save_calibration()
self.on_status("calibration reset")
def log_points(self, kind, data):
"""Every dot or target of a run, one line per source, for refining and for looking
at where the calibration is off: where in the view, the raw error, and the error
with the calibration of the time."""
lines = []
for e in data:
for name, g in e["sources"].items():
lines.append({"time": time.time(), "kind": kind, "source": name, "screen": self.screen,
"round": e.get("round"), "target": e["target"], "hy": g["hy"], "hp": g["hp"],
"off": g["off_deg"], "err": g["err_deg"], "cerr": g.get("cerr_deg"),
"model": self.model_mode, "sd": g["sd_deg"], "n": g["n"],
"layout": (self.test or {}).get("layout") if kind == "test" else None})
try:
with open(STATE / "points.jsonl", "a") as f:
for rec in lines:
f.write(json.dumps(rec) + "\n")
except OSError as e:
print(f"points: {e}", file=sys.stderr)
def clear_stats(self):
self.attempts = []
self.results = None
self.update_stats()
self.area.queue_draw()
def log(self, name, obj, single=False):
try:
if single:
(STATE / name).write_text(json.dumps(obj, indent=1))
else:
with open(STATE / name, "a") as f:
f.write(json.dumps(obj) + "\n")
except OSError as e:
print(f"log {name}: {e}", file=sys.stderr)
# --- Stats text ---
def update_stats(self):
lines = []
s = self.sample
if s is None:
lines.append(self.status or "waiting for gaze...")
else:
stale = time.monotonic() - self.last_arrival > 0.5
lines.append(f"{self.rate:4.0f} Hz age {s['age']:4.1f} ms" + (" (no samples: headset off?)" if stale else ""))
r = self.raw.get(self.source)
if r:
lines.append(f"dot {r[0]:6.0f},{r[1]:5.0f} px 1 deg = {1 / r[5]:.0f} px" if r[5] else "")
lines.append(f"view yaw {r[3]:+5.1f} pitch {r[4]:+5.1f} deg")
else:
other = (s["src"].get(self.source) or {}).get("hit")
lines.append(f"gaze on screen {other['s']}" if other else "gaze off the screens")
if self.status:
lines.append(self.status)
m = self.models[self.source]
cy, cp = m.offset()
lines.append(f"calibration {self.model_mode}: offset {cy:+.2f},{cp:+.2f} deg, {m.samples} samples")
lc = self.live[self.source]
if lc.samples:
ly, lp = lc.offset()
lines.append(f"click corrections: {len(lc.samples)} clicks, offset {ly:+.2f},{lp:+.2f} deg"
+ ("" if self.w_snaps.get_active() else " (off)"))
started = self.steam.started()
if started:
line = f"SteamVR eye tracker running since {time.strftime('%H:%M', time.localtime(started))}"
if self.calibrated_at and started > self.calibrated_at:
line += ", restarted after your calibration (its own calibration started over)"
lines.append(line)
if self.calibrated_at:
lines.append(f" it learned from {self.steam.accepted_since(self.calibrated_at)} clicks since your calibration")
snaps = [r for r in self.snap_log if r.get("source") == self.source][-30:]
if snaps:
n = len(snaps)
fixed = [r for r in snaps if r["how"] != "tap"]
lines.append(f"snaps (last {n}): right at the press {sum(r['right_at_press'] for r in snaps)}, "
f"right in the end {sum(r['right'] for r in snaps)}")
if fixed:
lines.append(f" corrected {len(fixed)} (eyes {sum(r['how'] == 'eyes' for r in fixed)}, "
f"mouse {sum(r['how'] == 'mouse' for r in fixed)}), "
f"{sum(r['right'] for r in fixed)} onto the right one")
g = [r["sources"][self.source] for r in snaps if self.source in r["sources"]]
asked = [r for r in snaps if self.source in r["sources"] and r["asked"] is not None]
if asked:
base = sum(r["sources"][self.source]["pick_base"] == r["asked"] for r in asked)
live = sum(r["sources"][self.source]["pick_live"] == r["asked"] for r in asked)
lines.append(f" snap right with calibration only {base}, with click corrections {live}")
if g:
# To the element you were asked for (what you really looked at), not the one clicked.
errs = [math.hypot(o[0] - x["base"][0] - x["live"][0], o[1] - x["base"][1] - x["live"][1])
for x in g for o in [x.get("off_asked", x["off"])]]
lines.append(f" corrected error at the press: median {statistics.median(errs):.2f} deg")
if self.test:
lines.append(f"test: target {self.test['index'] + 1} of {len(self.test['points'])}")
if self.results:
kind = self.results.get("kind", "test")
lines.append(f"{kind} ({self.results['model']}): raw error, mean / median / max, within 1 deg, jitter")
for name, v in self.results["summary"].items():
lines.append(f" {SOURCE_NAMES[name]:<15} {v['mean_deg']:.2f} / {v['median_deg']:.2f} / "
f"{v['max_deg']:.2f} deg ({v['mean_px']:.0f} px) {v['within_1deg'] * 100:3.0f}% "
f"{v['jitter_deg']:.2f}")
lines.append(f" offset {v['offset_deg'][0]:+.2f},{v['offset_deg'][1]:+.2f} deg; "
+ (f"fit leaves {v['corrected_mean_deg']:.2f} deg on these dots" if kind == "calibration"
else f"with current calibration {v['corrected_mean_deg']:.2f} deg"))
if self.results and self.results.get("regions", {}).get(self.source):
worst = self.results["regions"][self.source]
lines.append(" corrected error by region, worst first:")
lines.append(" " + ", ".join(f"{r} {m:.2f} ({n})" for r, m, n in worst[:5]))
if len(self.test_history) > 1:
lines.append("tests since calibrating: " + " -> ".join(f"{h[3]:.2f}" for h in self.test_history) + " deg")
rs = getattr(self, "refine_scores", None)
if rs:
lines.append(f"refine ({rs['points']} points), error on left-out points, mean / median / max:")
for mode, (mean, med, mx) in rs["scores"].items():
lines.append(f" {mode:<15}{mean:.2f} / {med:.2f} / {mx:.2f}" + (" <- using" if mode == rs["best"] else ""))
caps = [c for c in self.captures if c.get("source") == self.source and "err_deg" in c]
if caps:
good = [c for c in caps if c["verdict"] in ("learned", "measured")]
lines.append(f"freezes: {len(good)} of {len(caps)} kept")
if good:
first, last = good[:10], good[-10:]
lines.append(f" error: last {len(last)} median {statistics.median(c['err_deg'] for c in last):.2f} deg"
+ (f", first {len(first)} {statistics.median(c['err_deg'] for c in first):.2f}"
if len(good) > 10 else ""))
lines.append(f" spread: median {statistics.median(c['spread_deg'] for c in last):.2f} deg")
tries = [a for a in self.attempts if "hit" in a][-20:]
if tries:
hits = sum(a.get("hit_at_press", a["hit"]) for a in tries)
lines.append(f"practice (last {len(tries)}): {hits} hits at the press")
lines.append(f" off at press: median {statistics.median(a['press_err_deg'] for a in tries):.2f} deg "
f"({statistics.median(a['press_err_px'] for a in tries):.0f} px)")
drags = [a for a in tries if a.get("dragged")]
if drags:
lines.append(f" dragged {len(drags)}: median {statistics.median(a['drag_deg'] for a in drags):.2f} deg, "
f"{sum(a['hit'] for a in drags)} onto the target, "
f"{sum(a.get('learned', False) for a in drags)} learned")
nudged = [a for a in tries if "final_err_px" in a]
if nudged:
lines.append(f" miss after nudge: median {statistics.median(a['final_err_px'] for a in nudged):.0f} px, "
f"hold {statistics.median(a['held_s'] for a in nudged):.2f} s")
text = "\n".join(l for l in lines if l)
self.w_stats.set_label(text)
if os.environ.get("FT_GAZEPROBE_DEBUG"):
print(text.replace("\n", " | "), file=sys.stderr, flush=True)
# --- Drawing ---
def draw(self, _area, cr, w, h):
cr.set_source_rgb(*(self.ROUND_BG[self.calib["round"]] if self.calib else (0.11, 0.12, 0.14)))
cr.paint()
if self.calib:
self.draw_calib(cr, w, h)
return
if self.reseat_check:
self.draw_reseat(cr, w, h)
return
m = self.monitors.get(self.screen)
if m and self.is_fullscreen():
g = m.get_geometry()
if (g.width, g.height) != (w, h):
self.text(cr, 40, h - 40, f"window {w}x{h} isn't the screen's {g.width}x{g.height}: positions are off",
(1, 0.4, 0.3), 22)
elif not self.is_fullscreen() and self.origin is None:
self.text(cr, 40, h - 40, "windowed, and KWin hasn't said where: positions are off", (1, 0.4, 0.3), 22)
r = self.raw.get(self.source)
if self.w_cells.get_active() and r and r[5]:
self.degree_grid(cr, w, h, 1 / r[5])
if self.mode == "fit":
self.fitcheck.draw(cr, w, h, self.text, time.monotonic())
if self.fitcheck.guide_step(time.monotonic()):
GLib.idle_add(self.area.queue_draw)
if self.test:
self.draw_test(cr)
if self.results and not self.test and not self.snap and self.mode != "fit":
self.draw_results(cr)
if self.snap:
self.draw_snap(cr)
for x, y, tr in self.targets:
self.bullseye(cr, x, y, tr)
if self.mode == "practice" and not self.test:
self.text(cr, 40, 60, "Click practice: look at the target and press (click or Enter), and keep looking.",
(1, 1, 1), 26)
self.text(cr, 40, 96, "The white dot is your gaze pointer. Not on the target? Keep holding and move the mouse "
"to drag the dot there, then let go. The drag is learned as the tracker's error.",
(0.85, 0.85, 0.85), 19)
if self.practice:
(fx, fy), (px, py) = self.practice["F"], self.practice_point()
cr.set_source_rgba(0.35, 0.85, 1.0, 0.9)
cr.set_line_width(2)
cr.arc(fx, fy, 14, 0, 2 * math.pi)
cr.stroke()
cr.move_to(fx, fy)
cr.line_to(px, py)
cr.stroke()
now = time.monotonic()
for x, y, t, hit in self.clicks:
a = max(0.0, 1 - (now - t) / 2.0)
if a <= 0:
continue
col = (0.3, 1, 0.4) if hit else (1, 0.35, 0.3) if hit is False else (1, 1, 1)
cr.set_source_rgba(*col, a)
cr.set_line_width(3)
cr.arc(x, y, 14 + (1 - a) * 20, 0, 2 * math.pi)
cr.stroke()
frozen = self.capture is not None
live = not frozen or self.w_live.get_active()
own = self.source == "own"
if self.w_all.get_active() and live and not own:
for name in SOURCES:
rr = self.raw.get(name)
if rr:
x, y = self.corrected(name, rr)
cr.set_source_rgba(*SOURCE_COLORS[name], 0.9)
cr.arc(x, y, 7, 0, 2 * math.pi)
cr.fill()
if self.head:
cr.set_source_rgba(0.7, 0.7, 0.7, 0.8)
cr.set_line_width(2)
x, y = self.head
cr.move_to(x - 12, y)
cr.line_to(x + 12, y)
cr.move_to(x, y - 12)
cr.line_to(x, y + 12)
cr.stroke()
if self.w_rawdot.get_active() and own and live:
# Each eye alone: they should meet, with a little jitter, where you look.
for e in getattr(self, "own_eyes", []):
if e:
cr.set_source_rgba(1, 0.25, 0.25, 0.85)
cr.arc(e[0], e[1], 5, 0, 2 * math.pi)
cr.fill()
elif self.w_rawdot.get_active() and r and live:
cr.set_source_rgba(1, 0.3, 0.3, 0.8)
cr.arc(r[0], r[1], 5, 0, 2 * math.pi)
cr.fill()
if frozen and self.w_live.get_active() and self.gaze:
cr.set_source_rgba(0.4, 0.8, 1, 0.8)
cr.arc(*self.gaze, 6, 0, 2 * math.pi)
cr.fill()
if frozen:
# Progress: the ring closes over settle + capture.
c = self.capture
total = (self.w_settle.get_value() + self.w_capture.get_value()) / 1000
f = min(1.0, (now - c["t0"]) / total)
settling = now - c["t0"] < self.w_settle.get_value() / 1000
cr.set_source_rgba(*((1, 0.8, 0.3) if settling else (0.3, 1, 0.5)), 0.9)
cr.set_line_width(3)
cr.arc(c["F"][0], c["F"][1], 22, -math.pi / 2, -math.pi / 2 + 2 * math.pi * f)
cr.stroke()
GLib.idle_add(self.area.queue_draw)
res = self.result
if res and now - res["shown"] < 2.5 and "L" in res:
a = max(0.0, 1 - (now - res["shown"]) / 2.5)
ok = res["verdict"] in ("learned", "measured") or res["verdict"].startswith("taught")
(fx, fy), (lx, ly) = res["F"], res["L"]
cr.set_source_rgba(*((1, 0.85, 0.2) if ok else (1, 0.35, 0.3)), a)
cr.set_line_width(2)
cr.move_to(fx, fy)
cr.line_to(lx, ly)
cr.stroke()
cr.arc(lx, ly, 5, 0, 2 * math.pi)
cr.fill()
self.text(cr, fx + 26, fy - 26, f"{res['err_deg']:.2f}\u00b0 {res['verdict']}",
(1, 0.95, 0.7) if ok else (1, 0.5, 0.45), 18)
GLib.idle_add(self.area.queue_draw)
elif res and now - res["shown"] < 2.5:
self.text(cr, res["F"][0] + 26, res["F"][1] - 26, res["verdict"], (1, 0.5, 0.45), 18)
if self.press:
px, py = self.press["cursor"]
cr.set_source_rgba(1, 1, 1, 0.5)
cr.set_line_width(2)
cr.arc(px, py, 18, 0, 2 * math.pi)
cr.stroke()
if self.cursor:
cr.move_to(px, py)
cr.line_to(*self.cursor)
cr.stroke()
if self.cursor and self.snap:
# The highlight is the pointer here; a faint dot shows where the gaze is.
cr.set_source_rgba(1, 1, 1, 0.45)
cr.arc(*self.cursor, 4, 0, 2 * math.pi)
cr.fill()
elif self.cursor and not (self.test and self.mode == "test"):
x, y = self.cursor
cr.set_source_rgba(0.1, 0.1, 0.1, 0.9)
cr.arc(x, y, 11, 0, 2 * math.pi)
cr.fill()
cr.set_source_rgba(1, 1, 1, 0.95)
cr.arc(x, y, 8, 0, 2 * math.pi)
cr.fill()
if self.sample is None or time.monotonic() - self.last_arrival > 0.5:
self.text(cr, w / 2 - 200, h / 2, self.status or "waiting for gaze...", (1, 1, 1), 30)
def degree_grid(self, cr, w, h, px_per_deg):
"""Faint lines every degree (approximate: uses the scale where you're looking)."""
cr.set_source_rgba(1, 1, 1, 0.06)
cr.set_line_width(1)
step = px_per_deg
x = 0.0
while x < w:
cr.move_to(x, 0)
cr.line_to(x, h)
x += step
y = 0.0
while y < h:
cr.move_to(0, y)
cr.line_to(w, y)
y += step
cr.stroke()
def bullseye(self, cr, x, y, r):
cr.set_source_rgba(1, 0.85, 0.2, 0.95)
cr.set_line_width(2)
cr.arc(x, y, r, 0, 2 * math.pi)
cr.stroke()
cr.arc(x, y, max(2, r * 0.15), 0, 2 * math.pi)
cr.fill()
def draw_test(self, cr):
t = self.test
if t["index"] >= len(t["points"]):
return
x, y = t["points"][t["index"]]
collecting = t.get("collect_until")
scale = 1.0
if collecting:
scale = max(0.2, (collecting - time.monotonic()) / 0.6)
GLib.idle_add(self.area.queue_draw)
cr.set_source_rgba(1, 1, 1, 0.9)
cr.set_line_width(3)
cr.arc(x, y, 30 * scale, 0, 2 * math.pi)
cr.stroke()
cr.set_source_rgba(1, 0.2, 0.2, 1)
cr.arc(x, y, 4, 0, 2 * math.pi)
cr.fill()
self.text(cr, 40, 60, f"Testing the {self.model_label} correction. Look at the dot and press Enter or Space. "
f"{t['index'] + 1} / {len(t['points'])}", (1, 1, 1), 26)
if t.get("layout") == self.TEST_LAYOUTS[0]:
self.text(cr, 40, 96, "Face the centre of the window, as in the calibration, and move only your eyes.", (1, 1, 1), 20)
def draw_results(self, cr):
"""Each target: a faint line to the raw gaze, a solid one to where the corrected dot
was, coloured by that error (green under 1 degree, yellow under 2, red above)."""
for e in self.results["targets"]:
tx, ty = e["target"]
cr.set_source_rgba(1, 1, 1, 0.9)
cr.arc(tx, ty, 4, 0, 2 * math.pi)
cr.fill()
for name, g in e["sources"].items():
if not self.w_all.get_active() and name != self.source:
continue
cr.set_source_rgba(*SOURCE_COLORS[name], 0.35)
cr.set_line_width(1)
cr.move_to(tx, ty)
cr.line_to(*g["mean"])
cr.stroke()
ce = g.get("cerr_deg", g["err_deg"])
col = (0.3, 0.95, 0.4) if ce < 1 else (1, 0.85, 0.2) if ce < 2 else (1, 0.35, 0.3)
end = g.get("cmean", g["mean"])
cr.set_source_rgba(*col, 0.95)
cr.set_line_width(2.5)
cr.move_to(tx, ty)
cr.line_to(*end)
cr.stroke()
cr.arc(*end, 5, 0, 2 * math.pi)
cr.fill()
cr.set_line_width(1)
cr.arc(*end, max(2, g["sd_px"]), 0, 2 * math.pi) # the samples' spread
cr.stroke()
if name == self.source:
self.text(cr, tx + 8, ty - 8, f"{ce:.2f}\u00b0", col, 16)
def text(self, cr, x, y, s, rgb, size):
cr.set_source_rgb(*rgb)
cr.select_font_face("sans")
cr.set_font_size(size)
cr.move_to(x, y)
cr.show_text(s)
def on_close(self, *_):
self.reader.stop()
self.kwin.stop()
return False
def summarize(pts, target, model, mode):
"""One source at one test target: where it put your gaze, and how far off that was.
Medians, not means: a blink or a moment where the tracker lost an eye that got past
`steady_samples` would drag a mean, and the spread with it, a long way. The spread is
the median distance from the centre, scaled (x1.4826) to match a standard deviation."""
tx, ty = target
mx = statistics.median(p[0] for p in pts)
my = statistics.median(p[1] for p in pts)
j = [statistics.fmean(p[2][k] for p in pts) for k in range(4)]
hy = statistics.median(p[3] for p in pts)
hp = statistics.median(p[4] for p in pts)
dpp = statistics.fmean(p[5] for p in pts)
oy, op = deg_from_px(j, tx - mx, ty - my) # what the correction should add here
cy, cp = model.get(hy, hp, mode)
sd_px = 1.4826 * statistics.median(math.hypot(p[0] - mx, p[1] - my) for p in pts)
steps = [math.hypot(b[0] - a[0], b[1] - a[1]) for a, b in zip(pts, pts[1:])]
jitter = math.sqrt(statistics.fmean(d * d for d in steps)) * dpp if steps else 0.0
shift_x, shift_y = px_from_deg(j, cy, cp) # the correction, in pixels
return {"mean": [mx, my], "cmean": [mx + shift_x, my + shift_y], "n": len(pts), "j": j, "hy": hy, "hp": hp, "dpp": dpp,
"err_px": math.hypot(tx - mx, ty - my), "err_deg": math.hypot(oy, op), "off_deg": [oy, op],
"cerr_deg": math.hypot(oy - cy, op - cp), "sd_px": sd_px, "sd_deg": sd_px * dpp, "jitter_deg": jitter}
def view_region(hy, hp, edge=7.0):
"""Which part of your view a head-relative direction is in: centre, up, down-left, ..."""
row = "up" if hp > edge else "down" if hp < -edge else ""
col = "left" if hy > edge else "right" if hy < -edge else ""
return "-".join(p for p in (row, col) if p) or "centre"
def region_errors(targets, source, key="cerr_deg"):
"""Mean error by region of the view, worst first: [(region, mean deg, count)]."""
by = {}
for e in targets:
g = e["sources"].get(source)
if g and key in g:
by.setdefault(view_region(g["hy"], g["hp"]), []).append(g[key])
return sorted(((r, statistics.fmean(v), len(v)) for r, v in by.items()), key=lambda x: -x[1])
def main():
ap = argparse.ArgumentParser(description="Eye tracking playground for the Frametop desktop")
ap.add_argument("--screen", type=int, default=0, help="Frametop screen to open on (default: where it opens)")
ap.add_argument("--mode", choices=Probe.MODE_KEYS, help="start in this mode (fit: Headset fit)")
ap.add_argument("--source", choices=SOURCES,
help="gaze source to start with (default: own if our tracker is running, else mmap2)")
args, rest = ap.parse_known_args()
app = Adw.Application(application_id="dev.frametop.GazeProbe", flags=Gio.ApplicationFlags.NON_UNIQUE)
windows = []
def activate(a):
win = Probe(a, args.screen)
if args.source:
win.w_source.set_selected(SOURCES.index(args.source))
elif not win.own.ask("status").startswith("fail"):
# Our own tracker is running: that's what you're here to test.
win.w_source.set_selected(SOURCES.index("own"))
if args.mode:
win.set_mode(args.mode)
windows.append(win)
win.present()
def quit_on_signal():
for w in windows:
w.reader.stop()
app.quit()
return GLib.SOURCE_REMOVE
app.connect("activate", activate)
app.connect("shutdown", lambda *_: [w.reader.stop() for w in windows])
for sig in (signal.SIGINT, signal.SIGTERM, signal.SIGHUP):
GLib.unix_signal_add(GLib.PRIORITY_DEFAULT, sig, quit_on_signal)
return app.run([sys.argv[0]] + rest)
if __name__ == "__main__":
sys.exit(main())
Reference in new issue View Git Blame Copy Permalink