mirror of
https://github.com/DeeJanuz/frametop.git
synced 2026-10-06 03:00:06 +02:00
DeeJanuzandClaude Opus 5.5
8998fc1ddd
Drops the quick check's second step (3629681): the user didn't want it, and it
re-entered itself every tick, re-placing and redrawing the panel, which
flickered in the headset.
POINTER_GAZE_NUDGE_MAX is the one learning limit, now 55 degrees by default:
half of the 109 the Frame shows across. The helper measures the correction
itself (raw gaze to click), not the mouse's path; past the limit it isn't
learned and the helper sends "recheck" for the quick check, for the mouse and
keyboard alike. ft-gazed's own limits (8 and 25) follow the setting.
Test, our tracker's 409 clicks since its Sep 29 calibration: a one-dot check
set from any one of them puts the next 2 minutes' clicks within 15 degrees (99%
within 4.2) and the next 10 minutes' within 25, so it gets well under 55.
The panel's dot is still and its capture ring fills in quarters, so it's drawn
again only a few times per dot.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
654 lines
27 KiB
Python
654 lines
27 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.
|
|
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
|
|
when gaze mode comes on without a calibration for the tracker in use, and on
|
|
"calibrate". 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.
|
|
|
|
A 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; 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).
|
|
|
|
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 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
|
|
DOT_TIMEOUT = 8.0 # seconds a dot of five or full waits; then it's skipped
|
|
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
|
|
|
|
|
|
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
|
|
|
|
|
|
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.gaze_heard = 0.0
|
|
self.want_full_until = 0.0 # gaze mode came on before the tracker said whether it's calibrated
|
|
self.last_quick = 0.0
|
|
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" to "gaze ?"; the panel's are dropped."""
|
|
while True:
|
|
try:
|
|
data = self.out.recv(4096).decode("utf-8", "replace")
|
|
except (BlockingIOError, OSError):
|
|
return
|
|
if data in ("ok on", "ok off"):
|
|
on = data == "ok on"
|
|
was, self.gaze_on, self.gaze_heard = self.gaze_on, on, time.monotonic()
|
|
if on and was is False:
|
|
self.on_gaze_on()
|
|
|
|
# --- 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."""
|
|
return self.eyes_seen() and time.monotonic() - self.svc.last_sample < 2
|
|
|
|
def on_gaze_on(self):
|
|
cal = self.calibrated()
|
|
if cal is False:
|
|
self.start("full", "gaze mode came on without a calibration")
|
|
elif cal is None:
|
|
self.want_full_until = time.monotonic() + 20
|
|
|
|
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 not self.can_run():
|
|
return "error the headset is off or the tracker isn't sending"
|
|
own = svc.kind == "own"
|
|
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": {}}
|
|
log(f"{kind} check: {reason}")
|
|
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 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. Left click or Meta+J: now. Right click or Meta+K: stop")
|
|
elif c["kind"] == "five":
|
|
self.to_panel(f"text Look at the dot ({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"] = [], False, None
|
|
|
|
def on_sample(self, s):
|
|
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 not c or c["done_at"]:
|
|
return
|
|
if c["own"]:
|
|
src = s["src"].get("own") or {}
|
|
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()
|
|
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 ring fills in quarters: each step is a new picture for the panel to upload, so
|
|
# few of them keep it solid.
|
|
progress = math.floor(min(1.0, held / CHECK_WINDOW) * 4) / 4
|
|
if 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 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:
|
|
steady = steady_samples(samples)
|
|
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
|
|
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.get('reply', '')})")
|
|
if c["tries"] >= 2 or c["kind"] != "full":
|
|
self.skip()
|
|
else:
|
|
self.to_panel(f"dot {yaw:.3f} {pitch:.3f} fail")
|
|
c["run"], c["accept"] = [], False
|
|
c["shown"] = time.monotonic() # settle again, then retry
|
|
return
|
|
c["captured"] += 1
|
|
c["tries"] = 0
|
|
self.to_panel(f"dot {yaw:.3f} {pitch:.3f} done")
|
|
c["done_at"] = time.monotonic() + DONE_PAUSE
|
|
|
|
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:
|
|
log(f"calibration failed: only {c['captured']} of {n} dots; the old one stays")
|
|
self.to_panel(f"text Calibration failed: only {c['captured']} of {n} dots. Try again from Input Settings")
|
|
self.to_panel("dot 0 0 off")
|
|
c["done_at"] = time.monotonic() + 3.0
|
|
c["closing"] = True
|
|
return
|
|
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 command(self, words):
|
|
"""quickcal, calibrate, calaccept, calquit -> a reply."""
|
|
cmd = words[0]
|
|
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 == "calaccept":
|
|
if self.check:
|
|
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["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["kind"] == "quick" and now - c["started"] > QUICK_TIMEOUT:
|
|
self.close("ignored")
|
|
elif c["kind"] != "quick" and now - c["shown"] > DOT_TIMEOUT:
|
|
log(f"{c['kind']} check dot {c['i'] + 1}: skipped (the gaze didn't hold still)")
|
|
self.skip()
|
|
|
|
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 ?")
|
|
if now - self.gaze_heard > 5:
|
|
self.gaze_on = None # the helper isn't answering
|
|
if self.check:
|
|
self.to_helper("calpanel 1")
|
|
if self.want_full_until:
|
|
cal = self.calibrated()
|
|
if cal is not None or now > self.want_full_until:
|
|
self.want_full_until = 0.0
|
|
if cal is False and self.gaze_on:
|
|
self.start("full", "gaze mode came on without a calibration")
|
|
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
|
|
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")
|
|
|
|
def status(self):
|
|
c = self.check
|
|
st = {"check": None, "gaze_mode": self.gaze_on, "calibrated": self.calibrated(), "eyes": self.eyes_seen(),
|
|
"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()
|