mirror of
https://github.com/DeeJanuz/frametop.git
synced 2026-10-06 02:00:06 +02:00
DeeJanuzandClaude Opus 5.5
1d0211dc5e
The correction models, filters, blink filter, and SteamVR log reader move out of the probe so the gaze service can use them too. Two additions on the way: the log reader follows the headset going on and off (SteamVR starts its eye model over each time it goes on), and LiveCorrection counts lessons from before the last time it went on less (OLD_WEAR), so the first few after relearn the offset while the shape is kept. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2560 lines
118 KiB
Python
Executable File
2560 lines
118 KiB
Python
Executable File
#!/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 three sources ft-gaze reads (SteamVR's eye tracking action and
|
|
the two gaze sets in eye-server.mmap). Three 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).
|
|
|
|
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, 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 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"
|
|
SOURCES = ["action", "mmap1", "mmap2"]
|
|
SOURCE_NAMES = {"action": "SteamVR action", "mmap1": "mmap set 1", "mmap2": "mmap set 2"}
|
|
SOURCE_COLORS = {"action": (0.2, 0.8, 1.0), "mmap1": (1.0, 0.6, 0.1), "mmap2": (0.9, 0.3, 0.9)}
|
|
TRIGGER_KEYS = {Gdk.KEY_Return, Gdk.KEY_KP_Enter, Gdk.KEY_space}
|
|
|
|
|
|
# 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)
|
|
|
|
|
|
# --- 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.models = {s: Correction() for s in SOURCES}
|
|
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"]
|
|
MODE_NAMES = ["Free look", "Accuracy test", "Click practice", "Snap practice"]
|
|
|
|
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.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("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):
|
|
box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=8, halign=Gtk.Align.START,
|
|
valign=Gtk.Align.START, margin_start=40, margin_top=40)
|
|
box.add_css_class("probe-panel")
|
|
box.set_size_request(460, -1)
|
|
|
|
title = Gtk.Label(label="Gaze probe", xalign=0)
|
|
title.add_css_class("title-2")
|
|
box.append(title)
|
|
hint = Gtk.Label(label="Trigger: Enter, Space, or click. Right-click for the menu. Tab hides this panel, Esc quits. "
|
|
"Click the window once so the keys reach it.", xalign=0, wrap=True)
|
|
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
|
|
|
|
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)
|
|
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)
|
|
self.w_stats.add_css_class("probe-stats")
|
|
box.append(self.w_stats)
|
|
return box
|
|
|
|
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_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 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 == Gdk.KEY_F11:
|
|
self.toggle_fullscreen()
|
|
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 == 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.update_stats()
|
|
return True
|
|
|
|
def on_sample(self, s):
|
|
self.sample = s
|
|
self.recent.append(s)
|
|
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
|
|
|
|
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)."""
|
|
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.mode == "test":
|
|
self.test_press()
|
|
return
|
|
if self.mode == "snap":
|
|
self.snap_press()
|
|
return
|
|
if self.mode == "practice":
|
|
self.practice_press()
|
|
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 == "test":
|
|
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]}
|
|
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"])
|
|
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"
|
|
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 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.panel.get_visible() and x < 560 and y < 1000:
|
|
continue # not under the panel
|
|
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
|
|
|
|
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.panel.get_visible() and ox < 560 and oy < 1000:
|
|
continue # not under the panel
|
|
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
|
|
|
|
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 steady_samples([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."""
|
|
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
|
|
radius = min(self.w_ring.get_value() * px_per_deg, 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):
|
|
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 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
|
|
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 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()
|
|
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")
|
|
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")
|
|
|
|
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) * 7
|
|
n = c["round"] * 7 + c["index"] + 1
|
|
self.text(cr, 60, 70, f"Calibration: 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
|
|
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.test:
|
|
self.draw_test(cr)
|
|
if self.results and not self.test and not self.snap:
|
|
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()
|
|
if self.w_all.get_active() and live:
|
|
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 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"] == "learned" or res["verdict"] == "measured"
|
|
(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)")
|
|
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)
|
|
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())
|