mirror of
https://github.com/DeeJanuz/frametop.git
synced 2026-10-06 01:00:06 +02:00
DeeJanuzandClaude Opus 5.5
3a796b5013
Turning gaze mode on wakes our tracker, so its eyes come back just as the first calibration opens. DON_DELAY later, "the headset went on" re-armed the automatic calibration while it was still open, and when it ended, ft-eyes' status was still a moment old (not calibrated), so a second one opened at once. Seen live on 2026-10-05: 26 of 27 dots, then 27 more. The headset going on re-arms it only when no check is open. first-calibration-test.py now has the headset go on mid-calibration and checks that only one opens. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
927 lines
43 KiB
Python
927 lines
43 KiB
Python
"""gazecheck: the gaze service's checks and calibration, in the panel fixed to the headset
|
|
(gaze/panel/ft-gazepanel; ft-gazed runs it). Every kind is made of dots shown at head-relative
|
|
directions: look at each one.
|
|
|
|
quick one dot in the middle of your view. It opens when the headset goes on: eyes seen
|
|
for DON_DELAY after none for AWAY_MIN (SteamVR's tracker's variance for an eye under
|
|
EYE_LOST). SteamVR's "HMD on" can't say: it repeats every minute or so, and it can
|
|
stay on for hours with nobody in the headset. It also opens when our own tracker asks
|
|
for a click (its "reseat": the headset may sit differently on your face now), at most
|
|
once every QUICK_COOLDOWN, and on "quickcal" (Frametop Input Settings, or a mouse
|
|
button or key combination mapped to Gaze quick check), and when a click's correction
|
|
was past POINTER_GAZE_NUDGE_MAX (55 degrees; the helper's "recheck"): the tracker is
|
|
far off. Ignored, it closes after QUICK_TIMEOUT and changes nothing. The headset going
|
|
on is seen only while the gaze service is awake (gaze mode on, someone wearing it: see
|
|
ft-gazed), so it also opens when gaze mode comes on after the service idled.
|
|
five the middle and four around it, when the first FIVE_COUNT lessons after a quick check
|
|
were all over FIVE_LIMIT degrees off: the quick check didn't fix it.
|
|
full the calibration, as the gaze probe's: three rounds, dark, medium and bright (pupil
|
|
size, and the tracker's error with it, changes with brightness), each the middle and
|
|
a ring of six (SteamVR's tracker) or eight (ours, whose fit goes wrong past its dots)
|
|
RING degrees out, half that in the middle round, turned 20 degrees a round. It opens
|
|
whenever gaze mode is on without a calibration for the tracker in use and someone's
|
|
in the headset, and on "calibrate". One that closes unfinished (ignored, too few
|
|
dots) opens again only once the headset comes off and on, or gaze mode off and on.
|
|
Frametop's screens hide while it runs. Quitting it while there's still no
|
|
calibration turns gaze mode off (POINTER_GAZE=0); turning it on again reopens it.
|
|
Why gaze mode can't work yet goes in the status ("problem"), for Input Settings.
|
|
|
|
fit the headset fit check (on "fitcheck", Check headset fit on the Gaze page): live, a
|
|
card per eye (tracked or lost, the tracker's signal, how much of the last 10 s it was
|
|
seen) and hints, from the gaze probe's Headset fit (gaze/fitcheck.py), while you
|
|
adjust the headset. A left click or Meta+J runs its guided check (dots, then looks
|
|
down, up, left and right); a right click or Meta+K closes it, as does FIT_TIMEOUT.
|
|
|
|
A check asked for while the gaze service idles (quickcal, calibrate, fitcheck) wakes it and
|
|
waits until the tracker sends, at most ft-gazed's WAKE_SETTLE; then it opens, or logs why not.
|
|
|
|
The quick check's dot captures itself: from CHECK_SETTLE after it shows (the eyes getting
|
|
there), once the gaze has held within CHECK_SPREAD for CHECK_WINDOW (the probe's max spread and
|
|
capture time). It's the gaze holding still that counts, not where the tracker puts it, so it
|
|
works however far off the tracker is; a left click or Meta+J (the pointer helper's
|
|
"calaccept") takes it now. The full calibration's and five's dots wait for that click: you
|
|
click when you're looking at the dot (the user asked for that: a steady gaze isn't always on
|
|
the dot), and the gaze held still up to then is taken (ACCEPT_SPREAD). They wait as long as
|
|
it takes, up to CLICK_IDLE. A dot not taken says why in the panel's note line (reject_reason:
|
|
gazecal.steady_samples' drop counts for SteamVR's tracker, ft-eyes' reply for ours), as does a
|
|
click with nothing taken after ACCEPT_WAIT, and a failed calibration names its most common
|
|
reason there and in the status. A right click or Meta+K ("calquit") closes the panel. The pointer hides meanwhile ("calpanel 1",
|
|
renewed every second; the helper shows it again by itself when that stops).
|
|
|
|
Our tracker's first calibration: before it has one, ft-eyes publishes no gaze (it maps pupils
|
|
to a gaze only with a calibration), so there's no gaze to hold still. Its calibration runs
|
|
anyway ("blind"): someone in the headset (SteamVR's tracker sees an eye) and ft-eyes answering
|
|
are enough to start it, each dot stands in for the gaze, and a click takes the CHECK_WINDOW up
|
|
to it. ft-eyes then checks that each pupil was seen and held still in that window (calib-point)
|
|
and says why not. Without this, a fresh install could never calibrate our tracker.
|
|
|
|
What a capture teaches:
|
|
our tracker quick and five: a click ("click T YAW PITCH", like a pointer lesson); full:
|
|
calib-start, a calib-point for each dot, calib-fit (its calibration)
|
|
SteamVR's quick and five: a lesson for each eye, like the pointer's; full: each source's
|
|
calibration fitted from the dots (gazecal.Correction, the probe's way), saved
|
|
to calibration.json, and the lessons start over on top of it
|
|
Either way, how far off each eye was goes to the eye bias, and each capture to the lesson log.
|
|
"""
|
|
|
|
import json
|
|
import math
|
|
import os
|
|
import selectors
|
|
import signal
|
|
import socket
|
|
import statistics
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
from pathlib import Path
|
|
|
|
from fitcheck import MIN_REGION, FitCheck, wrap
|
|
from gazecal import DEFAULT_MODEL, EYE_LOST, STATE, steady_samples
|
|
|
|
REPO = Path(__file__).resolve().parents[1]
|
|
PANEL_PROG = REPO / "gaze" / "build" / "ft-gazepanel"
|
|
PANEL = "\0ft_gazepanel"
|
|
POINTER = "\0ft_pointer_helper"
|
|
SCREENS = "\0ft_screens"
|
|
EYES = "\0ft_eyes"
|
|
CONF = Path.home() / ".config" / "frametop.conf"
|
|
CALIBRATION = STATE / "calibration.json"
|
|
POINTS = STATE / "points.jsonl" # the probe's record of calibration dots, for its refine
|
|
CHECK_LOG = STATE / "checks.jsonl"
|
|
|
|
CHECK_SETTLE = 0.45
|
|
CHECK_WINDOW = 0.6
|
|
CHECK_SPREAD = 1.0 # degrees
|
|
ACCEPT_SPREAD = 2.5 # degrees: a capture asked for (calaccept) takes this much
|
|
CLICK_IDLE = 120.0 # seconds a dot of five or full waits for its click; then the check closes
|
|
QUICK_TIMEOUT = 6.0
|
|
QUICK_COOLDOWN = 120.0
|
|
DON_DELAY = 3.0 # seconds of eyes after AWAY_MIN without: the headset went on
|
|
AWAY_MIN = 3.0
|
|
EYES_GONE = 2.0 # seconds without eyes that close a check: the headset came off
|
|
FIVE_LIMIT = 2.0
|
|
FIVE_COUNT = 3
|
|
DONE_PAUSE = 0.35 # seconds the filled dot shows before the next
|
|
RING = 20.0
|
|
ROUND_BG = (0.03, 0.33, 0.8)
|
|
ROUND_NAMES = ("dark", "medium", "bright")
|
|
RING_SCALE = (1.0, 0.5, 1.0)
|
|
PANEL_RETRY = 10.0
|
|
FULL_RETRY = 10.0 # seconds before an automatic calibration that failed to start tries again
|
|
FIT_TIMEOUT = 300.0 # seconds the fit check stays up
|
|
FIT_EVERY = 0.5 # seconds between its cards' updates (each is a new picture for the panel)
|
|
FIT_HINT_WIDTH = 95 # characters a hint line holds in the panel
|
|
|
|
|
|
def log(msg):
|
|
print(f"ft-gazed: {msg}", file=sys.stderr, flush=True)
|
|
|
|
|
|
def check_dots(kind, own):
|
|
"""(yaw, pitch, round): head-relative degrees, yaw +left, pitch +up."""
|
|
if kind == "quick":
|
|
return [(0.0, 0.0, 0)]
|
|
if kind == "five":
|
|
return [(0.0, 0.0, 0), (12.0, 0.0, 0), (-12.0, 0.0, 0), (0.0, 9.0, 0), (0.0, -9.0, 0)]
|
|
out = []
|
|
n = 8 if own else 6
|
|
for rnd in range(3):
|
|
out.append((0.0, 0.0, rnd))
|
|
r = RING * RING_SCALE[rnd]
|
|
for i in range(n):
|
|
a = math.radians(-90 + rnd * 20 + i * 360 / n) # as the probe: x right, y down
|
|
x, y = r * math.cos(a), r * math.sin(a) * (0.9 if own else 1.0)
|
|
out.append((-x, -y, rnd))
|
|
return out
|
|
|
|
|
|
# Why SteamVR's samples for a look were dropped (gazecal.steady_samples' counts) -> (short, long):
|
|
# short for the small panel, long for the calibration's. FIT_HINT goes after the ones the
|
|
# headset's fit causes, in the calibration's panel.
|
|
STEAM_REASONS = {"lost_left": ("left eye lost", "SteamVR lost your left eye"),
|
|
"lost_right": ("right eye lost", "SteamVR lost your right eye"),
|
|
"lost_both": ("both eyes lost", "SteamVR lost both eyes"),
|
|
"blink": ("blinked", "you blinked"),
|
|
"vergence": ("eyes disagreed", "SteamVR's two eyes disagreed")}
|
|
FIT_HINT = "check the headset fit"
|
|
ACCEPT_WAIT = 1.5 # seconds after a click with no capture before the panel says what it waits for
|
|
|
|
|
|
def reject_reason(reply, why):
|
|
"""Why a dot wasn't taken -> (short, long, fit): our tracker's reply (`reply`), or SteamVR's
|
|
drop counts (`why`, when `reply` is None). `fit`: the headset's fit is the likely cause."""
|
|
if reply is None:
|
|
if not why:
|
|
return "no reading", "SteamVR sent no reading for that look", False
|
|
key = max(why, key=why.get)
|
|
return *STEAM_REASONS[key], key.startswith("lost")
|
|
words = reply.removeprefix("fail ").split()
|
|
# ft-eyes: "fail the left eye was seen in only 3 frames", "fail the left eye moved (4.2 px)"
|
|
if len(words) >= 3 and words[0] == "the" and words[2] == "eye":
|
|
eye = words[1]
|
|
if "seen" in words:
|
|
return f"{eye} eye not seen", f"our tracker saw your {eye} eye in only {words[-2]} frames", True
|
|
if "moved" in words:
|
|
return f"{eye} eye moved", f"your {eye} eye moved while you looked", False
|
|
if not reply:
|
|
return "no answer", "our tracker didn't answer", False
|
|
if reply.startswith("our tracker isn't running"):
|
|
return "tracker not running", "our tracker isn't running", False
|
|
text = reply.removeprefix("fail ")
|
|
return text[:24], f"our tracker said: {text}", False
|
|
|
|
|
|
def spread(points):
|
|
"""The median point and the spread around it (1.4826 x the median distance: a standard
|
|
deviation that one stray sample can't move far)."""
|
|
mx = statistics.median(p[0] for p in points)
|
|
my = statistics.median(p[1] for p in points)
|
|
return 1.4826 * statistics.median(math.hypot(p[0] - mx, p[1] - my) for p in points), (mx, my)
|
|
|
|
|
|
def write_gaze(on, path=CONF):
|
|
"""POINTER_GAZE=1|0 in the config, every other line kept."""
|
|
try:
|
|
lines = path.read_text().splitlines()
|
|
except OSError:
|
|
lines = []
|
|
value = f"POINTER_GAZE={1 if on else 0}"
|
|
for i, line in enumerate(lines):
|
|
if line.split("#", 1)[0].split("=", 1)[0].strip() == "POINTER_GAZE":
|
|
lines[i] = value
|
|
break
|
|
else:
|
|
lines.append(value)
|
|
tmp = path.with_suffix(".tmp")
|
|
tmp.write_text("\n".join(lines) + "\n")
|
|
tmp.replace(path)
|
|
|
|
|
|
def ask(addr, command, timeout=1.0):
|
|
"""A command to a local datagram socket, and its reply ("" without one)."""
|
|
s = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM | socket.SOCK_CLOEXEC)
|
|
try:
|
|
s.bind("")
|
|
s.settimeout(timeout)
|
|
s.sendto(command.encode(), addr)
|
|
return s.recv(4096).decode("utf-8", "replace")
|
|
except OSError:
|
|
return ""
|
|
finally:
|
|
s.close()
|
|
|
|
|
|
class Checks:
|
|
def __init__(self, svc, sel):
|
|
self.svc = svc
|
|
self.sel = sel
|
|
self.out = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM | socket.SOCK_CLOEXEC | socket.SOCK_NONBLOCK)
|
|
self.out.bind("") # the panel's and the helper's replies come back here
|
|
sel.register(self.out, selectors.EVENT_READ, "checks")
|
|
self.check = None
|
|
self.gaze_on = None
|
|
self.headset = None # someone wears it (the helper's "worn"), False "away", None unknown
|
|
self.gaze_heard = 0.0
|
|
self.pending = None # (command words, when): asked for while the gaze service idled
|
|
self.full_armed = True # gaze mode on without a calibration opens the full one (need_full)
|
|
self.full_blocked = None # why it can't open now
|
|
self.full_retry_at = 0.0
|
|
self.full_failed = None # why the last calibration failed (too few dots), for problem()
|
|
self.last_quick = 0.0
|
|
self.sample_at = 0.0 # the tracker last sent anything
|
|
self.seen_at = 0.0 # eyes last seen (SteamVR's tracker's variance for them, "unc")
|
|
self.away = True # no eyes for AWAY_MIN: their coming back is the headset going on
|
|
self.back_since = None
|
|
self.reseat_seen = False
|
|
self.after_quick = None
|
|
self.panel_proc = None
|
|
self.panel_restart_at = 0.0
|
|
self.screens_shown = None
|
|
self.last_progress = 0.0
|
|
|
|
@property
|
|
def active(self):
|
|
return self.check is not None
|
|
|
|
# --- The panel process ---
|
|
|
|
def start_panel(self):
|
|
if not PANEL_PROG.exists():
|
|
log(f"ft-gazepanel isn't built: run {REPO}/gaze/build.sh")
|
|
self.panel_restart_at = time.monotonic() + 60
|
|
return
|
|
env = dict(os.environ)
|
|
env["XDG_RUNTIME_DIR"] = f"/run/user/{os.getuid()}"
|
|
distrobox = Path.home() / ".local" / "bin" / "distrobox"
|
|
self.panel_proc = subprocess.Popen([str(distrobox), "enter", "dev", "--", str(PANEL_PROG), "--watch-stdin"],
|
|
env=env, stdin=subprocess.PIPE, stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.PIPE, start_new_session=True)
|
|
os.set_blocking(self.panel_proc.stderr.fileno(), False)
|
|
self.sel.register(self.panel_proc.stderr, selectors.EVENT_READ, "panel")
|
|
log("ft-gazepanel started")
|
|
|
|
def read_panel(self):
|
|
try:
|
|
data = os.read(self.panel_proc.stderr.fileno(), 65536)
|
|
except BlockingIOError:
|
|
return
|
|
if not data:
|
|
log(f"ft-gazepanel stopped (exit {self.panel_proc.poll()}); again in {PANEL_RETRY:.0f} s")
|
|
self.stop_panel()
|
|
self.panel_restart_at = time.monotonic() + PANEL_RETRY
|
|
if self.check:
|
|
self.close("the panel stopped")
|
|
return
|
|
for line in data.decode("utf-8", "replace").splitlines():
|
|
if line.strip():
|
|
log(line)
|
|
|
|
def stop_panel(self):
|
|
if not self.panel_proc:
|
|
return
|
|
try:
|
|
self.sel.unregister(self.panel_proc.stderr)
|
|
except (KeyError, ValueError):
|
|
pass
|
|
if self.panel_proc.stdin and not self.panel_proc.stdin.closed:
|
|
self.panel_proc.stdin.close()
|
|
try:
|
|
self.panel_proc.wait(timeout=2)
|
|
except subprocess.TimeoutExpired:
|
|
try:
|
|
os.killpg(self.panel_proc.pid, signal.SIGTERM)
|
|
except ProcessLookupError:
|
|
pass
|
|
self.panel_proc = None
|
|
|
|
def to_panel(self, command):
|
|
try:
|
|
self.out.sendto(command.encode(), PANEL)
|
|
except OSError:
|
|
pass
|
|
|
|
def to_helper(self, command):
|
|
try:
|
|
self.out.sendto(command.encode(), POINTER)
|
|
except OSError:
|
|
pass
|
|
|
|
def on_readable(self):
|
|
"""Replies on our socket: the helper's "ok on|off [worn|away]" to "gaze ? headset" (an
|
|
older helper leaves the headset out); the panel's are dropped."""
|
|
while True:
|
|
try:
|
|
words = self.out.recv(4096).decode("utf-8", "replace").split()
|
|
except (BlockingIOError, OSError):
|
|
return
|
|
if words[:1] == ["ok"] and len(words) in (2, 3) and words[1] in ("on", "off"):
|
|
on = words[1] == "on"
|
|
headset = None if len(words) == 2 else words[2] == "worn"
|
|
was = (self.gaze_on, self.headset)
|
|
self.gaze_on, self.headset, self.gaze_heard = on, headset, time.monotonic()
|
|
if on and was[0] is False:
|
|
self.on_gaze_on()
|
|
if was != (on, headset):
|
|
self.svc.update_awake()
|
|
|
|
# --- State ---
|
|
|
|
def calibrated(self):
|
|
"""Does the tracker in use have a calibration? None: our tracker hasn't said yet."""
|
|
svc = self.svc
|
|
kind = svc.kind
|
|
if kind == "own":
|
|
if time.monotonic() - svc.own_at > 5 or not svc.own:
|
|
return None
|
|
return bool((svc.own.get("calibration") or {}).get("dots"))
|
|
if kind == "eyes":
|
|
return True
|
|
return svc.models[svc.source].samples > 0
|
|
|
|
def eyes_seen(self, within=1.0):
|
|
return time.monotonic() - self.seen_at < within
|
|
|
|
def can_run(self):
|
|
"""Someone's in the headset and the tracker is sending. Our tracker sends no gaze before
|
|
its first calibration: then ft-eyes answering (calibrated() is False only once it has)
|
|
is enough, since the calibration is what it needs (see "first calibration" at the top)."""
|
|
if self.blind():
|
|
return self.eyes_seen()
|
|
return self.eyes_seen() and time.monotonic() - self.svc.last_sample < 2
|
|
|
|
def blind(self):
|
|
"""Our tracker is in use, running, and not calibrated yet: it has no gaze to send."""
|
|
return self.svc.kind == "own" and self.calibrated() is False
|
|
|
|
def on_gaze_on(self):
|
|
self.full_armed = True
|
|
self.need_full("gaze mode came on without a calibration")
|
|
|
|
def need_full(self, reason):
|
|
"""Gaze mode is on without a calibration: open the full one, once per arming (see the top),
|
|
or note why it can't open."""
|
|
cal = self.calibrated()
|
|
if cal:
|
|
self.full_armed = True # missing one later (the other tracker picked) is news again
|
|
if not self.gaze_on or self.check or not self.full_armed or cal is not False:
|
|
self.full_blocked = None
|
|
return
|
|
now = time.monotonic()
|
|
if not self.can_run() and self.svc.waking():
|
|
return # the tracker is still starting (the service idled)
|
|
if not self.can_run():
|
|
why = ("the eye tracker isn't sending" if now - self.svc.last_sample >= 2 and not self.blind()
|
|
else "no eyes seen (is the headset on?)")
|
|
elif now < self.full_retry_at:
|
|
return
|
|
else:
|
|
reply = self.start("full", reason)
|
|
why = None if reply == "ok" else reply.removeprefix("error ")
|
|
if why:
|
|
self.full_retry_at = now + FULL_RETRY
|
|
if why and why != self.full_blocked:
|
|
log(f"the calibration can't open: {why}")
|
|
self.full_blocked = why
|
|
if not why:
|
|
self.full_armed = False
|
|
|
|
def problem(self):
|
|
"""Why gaze mode, on, can't follow your eyes yet, or None. Our tracker not having said
|
|
yet is None: Input Settings has its own line for our tracker."""
|
|
if not self.gaze_on or self.calibrated() is not False:
|
|
return None
|
|
if self.check and self.check["kind"] == "full":
|
|
return "Not calibrated yet: the calibration is open in the headset"
|
|
if self.full_blocked:
|
|
return f"Not calibrated, and the calibration can't open: {self.full_blocked}"
|
|
if not self.full_armed:
|
|
if self.full_failed:
|
|
return f"Not calibrated: the calibration failed (most dots: {self.full_failed}). Use Calibrate"
|
|
return "Not calibrated: the calibration closed unfinished. Use Calibrate"
|
|
return "Not calibrated: the calibration opens in the headset"
|
|
|
|
def auto_quick(self, reason):
|
|
now = time.monotonic()
|
|
if self.check or not self.gaze_on or not self.can_run() or self.calibrated() is not True:
|
|
return
|
|
if now - self.last_quick < QUICK_COOLDOWN:
|
|
log(f"quick check skipped ({reason}): one ran {now - self.last_quick:.0f} s ago")
|
|
return
|
|
self.start("quick", reason)
|
|
|
|
# --- Running a check ---
|
|
|
|
def start(self, kind, reason):
|
|
svc = self.svc
|
|
if self.check:
|
|
return "error a check is running"
|
|
if not self.panel_proc:
|
|
return "error the panel isn't running (gaze/build.sh builds it)"
|
|
if kind == "fit":
|
|
return self.start_fit(reason)
|
|
if not self.can_run():
|
|
return "error the headset is off or the tracker isn't sending"
|
|
own = svc.kind == "own"
|
|
blind = self.blind()
|
|
if blind and kind != "full":
|
|
return "error our tracker isn't calibrated yet: use Calibrate"
|
|
if kind == "full" and own:
|
|
reply = ask(EYES, "calib-start", 3.0)
|
|
if not reply.startswith("ok"):
|
|
log(f"calibration not started: our tracker says {reply or 'nothing'}")
|
|
return f"error our tracker: {reply or 'no reply'}"
|
|
now = time.monotonic()
|
|
self.check = {"kind": kind, "reason": reason, "own": own, "dots": check_dots(kind, own), "i": 0,
|
|
"started": now, "shown": now, "run": [], "accept": False, "done_at": None, "tries": 0,
|
|
"skipped": 0, "captured": 0, "points": {}, "reasons": {}, "fit_reasons": set(), "note": "",
|
|
"blind": blind}
|
|
log(f"{kind} check: {reason}" + (" (our tracker's first: no gaze yet, so each click takes the look "
|
|
"up to it)" if blind else ""))
|
|
if kind == "full":
|
|
st = ask(SCREENS, "state", 0.5).split()
|
|
if len(st) >= 3 and st[0] == "ok":
|
|
self.screens_shown = (st[2] == "0") if st[1] == "always" else (st[2] == "1")
|
|
ask(SCREENS, "hide", 0.5)
|
|
self.to_helper("calpanel 1")
|
|
self.to_panel(f"show {'full' if kind == 'full' else 'quick'}")
|
|
self.show_dot()
|
|
if kind == "quick":
|
|
self.last_quick = now
|
|
return "ok"
|
|
|
|
def start_fit(self, reason):
|
|
# Eyes lost are what it's there to show, so it needs only the tracker sending.
|
|
if time.monotonic() - self.sample_at > 2:
|
|
return "error the headset is off or the eye tracker isn't sending"
|
|
now = time.monotonic()
|
|
self.check = {"kind": "fit", "reason": reason, "own": False, "dots": [], "i": 0, "started": now, "shown": now,
|
|
"run": [], "accept": False, "done_at": None, "tries": 0, "skipped": 0, "captured": 0,
|
|
"points": {}, "fit": FitCheck(), "drawn": {}, "drawn_at": 0.0, "step": None}
|
|
log(f"fit check: {reason}")
|
|
self.to_helper("calpanel 1")
|
|
self.to_panel("show fit")
|
|
self.to_panel("title Headset fit: adjust the headset while you watch")
|
|
self.to_panel("text Left click or Meta+J: guided check · Right click or Meta+K: done")
|
|
return "ok"
|
|
|
|
def fit_tick(self, now):
|
|
c = self.check
|
|
fit = c["fit"]
|
|
# The guided check: its dots at head-relative directions in the panel (the probe's
|
|
# screen fractions, spread over the middle of the panel), and its looks as the title.
|
|
step = fit.guide_step(now)
|
|
key = None if step is None else (step[0], step[1])
|
|
if key != c["step"]:
|
|
c["step"] = key
|
|
if step is None:
|
|
self.to_panel("dot 0 0 off")
|
|
self.to_panel("title Headset fit: adjust the headset while you watch")
|
|
elif step[0] == "dot":
|
|
fx, fy = step[1]
|
|
self.to_panel(f"dot {(0.5 - fx) * 32:.2f} {(0.5 - fy) * 24:.2f} look")
|
|
self.to_panel("title Look at the dot")
|
|
else:
|
|
self.to_panel("dot 0 0 off")
|
|
self.to_panel(f"title {step[1]}")
|
|
if now - c["drawn_at"] < FIT_EVERY:
|
|
return
|
|
c["drawn_at"] = now
|
|
drawn = c["drawn"]
|
|
for k in (0, 1):
|
|
word, (r, g, b) = fit.status(k)
|
|
sig, seen = fit.signal(k), fit.tracked_share(k, now)
|
|
cmd = (f"eye {k} {r:.2f} {g:.2f} {b:.2f} {-1 if sig is None else round(sig, 1):g} "
|
|
f"{-1 if seen is None else round(seen * 20) / 20:g} {word.capitalize()}")
|
|
if drawn.get(k) != cmd:
|
|
drawn[k] = cmd
|
|
self.to_panel(cmd)
|
|
if fit.have_eye_data and fit.samples < 3 * MIN_REGION:
|
|
hints = ["Look around slowly: up, down, left and right. Or left click (Meta+J) for a guided check."]
|
|
else:
|
|
hints = fit.hints()
|
|
lines = [line for h in hints for line in wrap(h, FIT_HINT_WIDTH)][:8]
|
|
cmd = "hints " + "|".join(lines)
|
|
if drawn.get("hints") != cmd:
|
|
drawn["hints"] = cmd
|
|
self.to_panel(cmd)
|
|
|
|
def show_dot(self):
|
|
c = self.check
|
|
yaw, pitch, rnd = c["dots"][c["i"]]
|
|
if c["kind"] == "full":
|
|
self.to_panel(f"bg {ROUND_BG[rnd]}")
|
|
self.to_panel(f"title Gaze calibration: {ROUND_NAMES[rnd]} round, {rnd + 1} of 3")
|
|
self.to_panel("text Look at the dot and click (left click or Meta+J). Right click or Meta+K: stop")
|
|
elif c["kind"] == "five":
|
|
self.to_panel(f"text Look at the dot and click ({c['i'] + 1} of {len(c['dots'])})")
|
|
else:
|
|
self.to_panel("text Look at the dot")
|
|
self.to_panel(f"dot {yaw:.3f} {pitch:.3f} look")
|
|
self.last_progress = None
|
|
c["shown"] = time.monotonic()
|
|
c["run"], c["accept"], c["done_at"], c["accept_at"] = [], False, None, None
|
|
|
|
def on_sample(self, s):
|
|
self.sample_at = time.monotonic()
|
|
if self.pending:
|
|
self.run_pending()
|
|
unc = (s["src"].get("mmap1") or {}).get("unc")
|
|
if unc and min(unc) <= EYE_LOST:
|
|
self.seen_at = time.monotonic()
|
|
if self.away and self.back_since is None:
|
|
self.back_since = self.seen_at
|
|
c = self.check
|
|
if c and c["kind"] == "fit":
|
|
c["fit"].feed(s, self.sample_at)
|
|
return
|
|
if not c or c["done_at"]:
|
|
return
|
|
if c["own"]:
|
|
src = s["src"].get("own") or {}
|
|
if c["blind"]:
|
|
# Our tracker's first calibration: no gaze yet, so the dot stands in for it (the
|
|
# gaze can't seem to move) and a click takes the look up to it. ft-eyes checks
|
|
# the pupils held still (see the top).
|
|
yaw, pitch, _ = c["dots"][c["i"]]
|
|
src = {"hy": yaw, "hp": pitch}
|
|
else:
|
|
src = s["src"].get("mmap1") or {}
|
|
if "hy" not in src:
|
|
per = [s["src"].get(n) or {} for n in ("left", "right")]
|
|
per = [p for p in per if "hy" in p]
|
|
src = {"hy": statistics.fmean(p["hy"] for p in per), "hp": statistics.fmean(p["hp"] for p in per)} if per else {}
|
|
if "hy" not in src:
|
|
return
|
|
now = time.monotonic()
|
|
c["gaze_at"] = now
|
|
if now < c["shown"] + CHECK_SETTLE:
|
|
return
|
|
g = (src["hy"], src["hp"])
|
|
run = c["run"]
|
|
if run:
|
|
recent = run[-30:]
|
|
my, mp = statistics.median(p[2] for p in recent), statistics.median(p[3] for p in recent)
|
|
if math.hypot(g[0] - my, g[1] - mp) > 2.5 * CHECK_SPREAD:
|
|
run.clear() # the eyes moved on (a saccade, a blink): start over
|
|
run.append((now, s, g[0], g[1]))
|
|
window = [p for p in run if p[0] >= now - CHECK_WINDOW]
|
|
held = now - run[0][0]
|
|
# The quick check's ring fills in quarters: each step is a new picture for the panel, so
|
|
# few of them keep it solid. The others wait for the click (see the top): no ring.
|
|
progress = math.floor(min(1.0, held / CHECK_WINDOW) * 4) / 4
|
|
if c["kind"] == "quick" and progress != self.last_progress:
|
|
yaw, pitch, _ = c["dots"][c["i"]]
|
|
self.to_panel(f"dot {yaw:.3f} {pitch:.3f} capture {progress:.2f}")
|
|
self.last_progress = progress
|
|
if c["accept"] and held >= 0.3 and len(window) >= 10:
|
|
sd, _ = spread([(p[2], p[3]) for p in window])
|
|
if sd <= ACCEPT_SPREAD:
|
|
self.capture(window)
|
|
return
|
|
if c["kind"] == "quick" and held >= CHECK_WINDOW and len(window) >= 20:
|
|
sd, _ = spread([(p[2], p[3]) for p in window])
|
|
if sd <= CHECK_SPREAD:
|
|
self.capture(window)
|
|
else:
|
|
del run[:len(run) // 2] # not steady enough yet: keep trying with the newer half
|
|
|
|
def capture(self, window):
|
|
svc = self.svc
|
|
c = self.check
|
|
yaw, pitch, rnd = c["dots"][c["i"]]
|
|
samples = [p[1] for p in window]
|
|
rec = {"time": time.time(), "check": c["kind"], "dot": c["i"], "round": rnd, "true": [yaw, pitch],
|
|
"samples": len(samples), "own": c["own"]}
|
|
ok = True
|
|
if c["own"]:
|
|
eyes = []
|
|
for k in (0, 1):
|
|
seen = [smp["src"]["own"]["eyes"][k] for smp in samples
|
|
if (smp["src"].get("own") or {}).get("eyes") and smp["src"]["own"]["eyes"][k]]
|
|
eyes.append((statistics.median(e[0] for e in seen), statistics.median(e[1] for e in seen)) if seen else None)
|
|
miss = [math.hypot(yaw - e[0], pitch - e[1]) if e else None for e in eyes]
|
|
rec.update(eyes=eyes, miss=miss)
|
|
t0, t1 = samples[0]["t"], samples[-1]["t"]
|
|
if c["kind"] == "full":
|
|
reply = ask(EYES, f"calib-point {t0:.6f} {t1:.6f} {yaw:.4f} {pitch:.4f}", 3.0)
|
|
rec["reply"] = reply
|
|
ok = reply.startswith("ok")
|
|
else:
|
|
try:
|
|
svc.eyes_sock.sendto(f"click {t1:.6f} {yaw:.4f} {pitch:.4f}".encode(), EYES)
|
|
except OSError as e:
|
|
rec["reply"], ok = f"our tracker isn't running ({e})", False
|
|
if ok:
|
|
svc.weights["own"].add(miss)
|
|
else:
|
|
why = {}
|
|
steady = steady_samples(samples, why=why)
|
|
rec["dropped"] = why
|
|
reads = {}
|
|
for name in ("action", "mmap1", "mmap2", "left", "right"):
|
|
pts = [(smp["src"][name]["hy"], smp["src"][name]["hp"]) for smp in steady
|
|
if "hy" in (smp["src"].get(name) or {})]
|
|
if len(pts) >= 15:
|
|
reads[name] = (statistics.median(p[0] for p in pts), statistics.median(p[1] for p in pts))
|
|
rec["reads"] = reads
|
|
main = ("left", "right") if svc.kind == "eyes" else (svc.source,)
|
|
if not all(n in reads for n in main):
|
|
ok = False
|
|
rec["reply"] = f"only {len(steady)} of {len(samples)} samples had both eyes"
|
|
elif c["kind"] == "full":
|
|
for name, (hy, hp) in reads.items():
|
|
c["points"].setdefault(name, []).append((hy, hp, yaw - hy, pitch - hp))
|
|
try:
|
|
with open(POINTS, "a") as f:
|
|
for name, (hy, hp) in reads.items():
|
|
f.write(json.dumps({"time": time.time(), "source": name, "hy": hy, "hp": hp,
|
|
"off": [yaw - hy, pitch - hp], "layout": None, "how": "panel"}) + "\n")
|
|
except OSError:
|
|
pass
|
|
else:
|
|
miss = []
|
|
for name in main:
|
|
hy, hp = reads[name]
|
|
cy, cp = svc.correction(name, hy, hp)
|
|
miss.append(math.hypot(yaw - hy - cy, pitch - hp - cp))
|
|
svc.lives[name].add({"time": time.time(), "hy": hy, "hp": hp, "dy": yaw - hy, "dp": pitch - hp,
|
|
"wy": 1.0, "wp": 1.0, "how": "check"}, svc.models[name], svc.mode)
|
|
rec["miss"] = miss
|
|
if svc.kind == "eyes":
|
|
svc.weights["steam"].add(miss)
|
|
svc.dirty = True
|
|
if not ok:
|
|
short, long, fit = reject_reason(rec.get("reply", "") if c["own"] else None, rec.get("dropped"))
|
|
rec["reason"] = long
|
|
c["reasons"][long] = c["reasons"].get(long, 0) + 1
|
|
if fit:
|
|
c["fit_reasons"].add(long)
|
|
rec["accepted"] = ok
|
|
self.log_check(rec)
|
|
if not ok:
|
|
c["tries"] += 1
|
|
log(f"{c['kind']} check dot {c['i'] + 1}: not taken: {rec['reason']} ({rec.get('reply', '')})")
|
|
full = c["kind"] == "full"
|
|
if c["tries"] >= 2 or not full:
|
|
self.note(f"Dot skipped: {long}" + (f" ({FIT_HINT})" if fit else "") if full else f"Skipped: {short}")
|
|
self.skip()
|
|
else:
|
|
self.note(f"Not taken: {long}. Look at the dot and click again")
|
|
self.to_panel(f"dot {yaw:.3f} {pitch:.3f} fail")
|
|
c["run"], c["accept"], c["accept_at"] = [], False, None
|
|
c["shown"] = time.monotonic() # settle again, then retry
|
|
return
|
|
c["captured"] += 1
|
|
c["tries"] = 0
|
|
self.note("")
|
|
self.to_panel(f"dot {yaw:.3f} {pitch:.3f} done")
|
|
c["done_at"] = time.monotonic() + DONE_PAUSE
|
|
|
|
def note(self, text):
|
|
"""The panel's warning line, over the instructions (empty: none). It stays until the
|
|
next dot is taken, so a skipped dot's reason is still there at the one after it."""
|
|
c = self.check
|
|
if c.get("note") != text:
|
|
c["note"] = text
|
|
self.to_panel(f"note {text}".rstrip())
|
|
|
|
def skip(self):
|
|
c = self.check
|
|
yaw, pitch, _ = c["dots"][c["i"]]
|
|
c["skipped"] += 1
|
|
c["tries"] = 0
|
|
self.to_panel(f"dot {yaw:.3f} {pitch:.3f} fail")
|
|
c["done_at"] = time.monotonic() + DONE_PAUSE
|
|
|
|
def advance(self):
|
|
c = self.check
|
|
c["i"] += 1
|
|
if c["i"] < len(c["dots"]):
|
|
self.show_dot()
|
|
return
|
|
self.finish()
|
|
|
|
def finish(self):
|
|
svc = self.svc
|
|
c = self.check
|
|
kind = c["kind"]
|
|
if kind in ("quick", "five"):
|
|
if c["captured"]:
|
|
log(f"{kind} check done ({c['captured']} of {len(c['dots'])} dots)")
|
|
self.after_quick = [] if kind == "quick" else None
|
|
self.close()
|
|
return
|
|
n = len(c["dots"])
|
|
if c["captured"] < n * 2 / 3:
|
|
reasons = c["reasons"]
|
|
main = max(reasons, key=reasons.get) if reasons else None
|
|
self.full_failed = main
|
|
log(f"calibration failed: only {c['captured']} of {n} dots; the old one stays"
|
|
+ (f". Not taken: {', '.join(f'{r} ({k})' for r, k in reasons.items())}" if reasons else ""))
|
|
self.to_panel(f"text Calibration failed: only {c['captured']} of {n} dots. Try again from Input Settings")
|
|
if main:
|
|
fit = main in c["fit_reasons"]
|
|
self.note(f"Most dots: {main}" + (". Check headset fit on the Gaze page" if fit else ""))
|
|
self.to_panel("dot 0 0 off")
|
|
c["done_at"] = time.monotonic() + (8.0 if main else 3.0) # time to read why
|
|
c["closing"] = True
|
|
return
|
|
self.full_failed = None
|
|
if c["own"]:
|
|
reply = ask(EYES, "calib-fit", 10.0)
|
|
log(f"calibration ({c['captured']} of {n} dots): our tracker says {reply or 'nothing'}")
|
|
else:
|
|
mode = svc.mode if svc.mode != "none" else DEFAULT_MODEL
|
|
for name, pts in c["points"].items():
|
|
if name in svc.models and pts:
|
|
svc.models[name].fit(pts, mode)
|
|
d = {name: m.to_json() for name, m in svc.models.items()}
|
|
d["_meta"] = {"calibrated_at": time.time(), "model": mode, "how": "panel"}
|
|
d["_live"] = {}
|
|
tmp = CALIBRATION.with_suffix(".tmp")
|
|
tmp.write_text(json.dumps(d, indent=1))
|
|
tmp.replace(CALIBRATION)
|
|
svc.forget_lessons() # a new calibration: the pointer's lessons start over on top of it
|
|
svc.load_calibration()
|
|
svc.refit()
|
|
log(f"calibration ({c['captured']} of {n} dots): {mode}, saved")
|
|
self.close()
|
|
|
|
def close(self, why=None):
|
|
if not self.check:
|
|
return
|
|
if why:
|
|
log(f"{self.check['kind']} check closed: {why}")
|
|
self.to_panel("hide")
|
|
self.to_helper("calpanel 0")
|
|
if self.check["kind"] == "full" and self.screens_shown:
|
|
ask(SCREENS, "show", 0.5)
|
|
self.screens_shown = None
|
|
self.check = None
|
|
|
|
def quit(self):
|
|
c = self.check
|
|
if not c:
|
|
return
|
|
self.close("quit")
|
|
if c["kind"] == "full" and self.calibrated() is False:
|
|
# No calibration still: gaze mode can't work, so it goes off until it's turned on again.
|
|
try:
|
|
write_gaze(False)
|
|
except OSError as e:
|
|
log(f"can't write {CONF}: {e}")
|
|
self.to_helper("gaze off")
|
|
self.gaze_on = False
|
|
log("gaze mode off: no calibration")
|
|
|
|
def log_check(self, rec):
|
|
try:
|
|
with open(CHECK_LOG, "a") as f:
|
|
f.write(json.dumps(rec) + "\n")
|
|
except OSError:
|
|
pass
|
|
|
|
# --- From the service ---
|
|
|
|
def run_pending(self):
|
|
"""A check asked for while the service idled: once the tracker sends (and our tracker has
|
|
said whether it's calibrated), or WAKE_SETTLE after waking, when its error is the real one."""
|
|
svc = self.svc
|
|
words, _ = self.pending
|
|
ready = (time.monotonic() - svc.last_sample < 2 if words[0] == "fitcheck" else self.can_run()) \
|
|
and (svc.kind != "own" or self.calibrated() is not None)
|
|
if not ready and svc.waking():
|
|
return
|
|
self.pending = None
|
|
reply = self.command(words, queue=False)
|
|
if reply != "ok":
|
|
log(f"{words[0]}, asked for while idle: {reply.removeprefix('error ')}")
|
|
|
|
def command(self, words, queue=True):
|
|
"""quickcal, calibrate, calaccept, calquit -> a reply."""
|
|
cmd = words[0]
|
|
if cmd in ("quickcal", "calibrate", "fitcheck") and queue and self.svc.waking() and not self.check:
|
|
# The tracker isn't running (or only just started): wake it, and do this once it sends.
|
|
self.pending = (words, time.monotonic())
|
|
self.svc.update_awake()
|
|
return "ok waking the eye tracker first"
|
|
if cmd == "quickcal":
|
|
return self.start("full" if self.calibrated() is False else "quick", "asked for")
|
|
if cmd == "calibrate":
|
|
return self.start("full", "asked for")
|
|
if cmd == "fitcheck":
|
|
return self.start("fit", "asked for")
|
|
if cmd == "calaccept":
|
|
if self.check and self.check["kind"] == "fit":
|
|
self.check["fit"].toggle_guide(time.monotonic())
|
|
elif self.check:
|
|
if not self.check["accept"]:
|
|
self.check["accept_at"] = time.monotonic()
|
|
self.check["accept"] = True
|
|
return "ok"
|
|
if cmd == "calquit":
|
|
self.quit()
|
|
return "ok"
|
|
if cmd == "recheck" and len(words) == 2:
|
|
self.auto_quick(f"a click was corrected {words[1]} deg")
|
|
return "ok"
|
|
return "error unknown command"
|
|
|
|
def after_lesson(self, rec):
|
|
if self.after_quick is None or "refused" in rec:
|
|
return
|
|
self.after_quick.append(rec.get("lesson_deg", 0.0))
|
|
if len(self.after_quick) < FIVE_COUNT:
|
|
return
|
|
off = self.after_quick
|
|
self.after_quick = None
|
|
if all(d > FIVE_LIMIT for d in off):
|
|
log(f"the {FIVE_COUNT} lessons after the quick check were {', '.join(f'{d:.1f}' for d in off)} deg off")
|
|
if self.gaze_on and self.can_run() and not self.check:
|
|
self.start("five", "the quick check didn't fix it")
|
|
|
|
def tick(self):
|
|
c = self.check
|
|
if not c:
|
|
return
|
|
now = time.monotonic()
|
|
if c["kind"] == "fit":
|
|
# Not closed when the eyes go: adjusting the headset loses them.
|
|
if now - c["started"] > FIT_TIMEOUT:
|
|
self.close("timed out")
|
|
else:
|
|
self.fit_tick(now)
|
|
return
|
|
if c["done_at"] and now >= c["done_at"]:
|
|
if c.get("closing"):
|
|
self.close()
|
|
else:
|
|
self.advance()
|
|
return
|
|
if not self.eyes_seen(EYES_GONE):
|
|
self.close("the headset came off")
|
|
return
|
|
if c["done_at"]:
|
|
return
|
|
if c["accept"] and c["accept_at"] and now - c["accept_at"] > ACCEPT_WAIT:
|
|
# Clicked, but no capture yet (see on_sample): say what it's waiting for.
|
|
full = c["kind"] == "full"
|
|
if now - c.get("gaze_at", 0.0) > 0.5:
|
|
self.note("Waiting: the eye tracker isn't sending a gaze" if full else "Waiting: no gaze")
|
|
else:
|
|
self.note("Waiting for your gaze to hold still on the dot" if full else "Hold your look still")
|
|
if c["kind"] == "quick" and now - c["started"] > QUICK_TIMEOUT:
|
|
self.close("ignored")
|
|
elif c["kind"] != "quick" and now - c["shown"] > CLICK_IDLE:
|
|
self.close(f"no click on dot {c['i'] + 1} for {CLICK_IDLE:.0f} s")
|
|
|
|
def periodic(self):
|
|
svc = self.svc
|
|
now = time.monotonic()
|
|
if not self.panel_proc and now >= self.panel_restart_at:
|
|
self.start_panel()
|
|
self.to_helper("gaze ? headset")
|
|
if now - self.gaze_heard > 5:
|
|
self.gaze_on = self.headset = None # the helper isn't answering
|
|
if self.pending:
|
|
self.run_pending()
|
|
if self.check:
|
|
self.to_helper("calpanel 1")
|
|
if not self.eyes_seen(AWAY_MIN):
|
|
self.away, self.back_since = True, None
|
|
elif self.back_since is not None and not self.eyes_seen():
|
|
self.back_since = None # gone again before DON_DELAY
|
|
elif self.back_since is not None and now - self.back_since >= DON_DELAY:
|
|
self.away, self.back_since = False, None
|
|
if not self.check:
|
|
# A calibration that closed unfinished opens again. Not one still open: gaze mode
|
|
# coming on wakes our tracker, so its eyes come back just as the calibration
|
|
# opens, and re-arming then opened a second one when the first ended.
|
|
self.full_armed = True
|
|
self.auto_quick("the headset went on")
|
|
if svc.kind == "own" and now - svc.own_at < 5:
|
|
reseat = any(e.get("reseat") for e in (svc.own.get("eyes") or {}).values())
|
|
if not reseat:
|
|
self.reseat_seen = False
|
|
elif not self.reseat_seen and self.can_run():
|
|
self.reseat_seen = True
|
|
self.auto_quick("our tracker asked for a click")
|
|
self.need_full("gaze mode is on without a calibration")
|
|
|
|
def status(self):
|
|
c = self.check
|
|
st = {"check": None, "gaze_mode": self.gaze_on, "headset_worn": self.headset, "pending": self.pending[0][0]
|
|
if self.pending else None, "calibrated": self.calibrated(), "eyes": self.eyes_seen(),
|
|
"problem": self.problem(),
|
|
"panel": self.panel_proc is not None,
|
|
"last_quick_s": round(time.monotonic() - self.last_quick) if self.last_quick else None}
|
|
if c:
|
|
st["check"] = {"kind": c["kind"], "dot": c["i"] + 1, "dots": len(c["dots"]), "captured": c["captured"],
|
|
"skipped": c["skipped"], "reason": c["reason"]}
|
|
return st
|
|
|
|
def stop(self):
|
|
self.close("the gaze service is stopping")
|
|
self.stop_panel()
|