mirror of
https://github.com/DeeJanuz/frametop.git
synced 2026-10-06 03:00:06 +02:00
ft-gazepanel (gaze/panel) is a SteamVR overlay that stays put in your view and shows dots at known head-relative directions; the gaze service runs it and drives it (gaze/gazecheck.py): - a one-dot quick check 3 s after the headset goes on, when our tracker asks for a click (reseat), at most every 2 minutes, and from Quick check on the Gaze page; five dots follow if the next 3 lessons are still over 2 degrees off - the full calibration (the probe's three rounds, dark to bright) when gaze mode comes on without one, or from Calibrate; quitting it with still no calibration turns gaze mode off Each dot takes the gaze once it has held still for 0.6 s; a left click or Meta+J takes it at once, a right click or Meta+K closes the panel (the pointer helper hides its dot and passes those presses on while "calpanel" lasts). Our tracker gets clicks and its own calibration; SteamVR's gets lessons per eye, or a new calibration.json. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
637 lines
26 KiB
Python
637 lines
26 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 DON_DELAY after the headset goes on
|
|
(SteamVR's eye tracking log says when), 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). 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, 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
|
|
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.worn_seen = svc.steam.worn()
|
|
self.don_at = 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 can_run(self):
|
|
"""The headset is on and the tracker is sending."""
|
|
return self.svc.steam.wearing() is not False 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")
|
|
c["shown"] = time.monotonic()
|
|
c["run"], c["accept"], c["done_at"] = [], False, None
|
|
|
|
def on_sample(self, s):
|
|
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]
|
|
if now - self.last_progress > 0.07:
|
|
yaw, pitch, _ = c["dots"][c["i"]]
|
|
self.to_panel(f"dot {yaw:.3f} {pitch:.3f} capture {min(1.0, held / CHECK_WINDOW):.2f}")
|
|
self.last_progress = now
|
|
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"
|
|
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 self.svc.steam.wearing() is False:
|
|
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")
|
|
worn = svc.steam.worn()
|
|
if worn and worn != self.worn_seen:
|
|
self.worn_seen = worn
|
|
self.don_at = worn + DON_DELAY
|
|
# Both wait for the tracker to send (it takes a moment after the headset goes on).
|
|
if self.don_at and time.time() >= self.don_at:
|
|
if self.can_run():
|
|
self.don_at = None
|
|
self.auto_quick("the headset went on")
|
|
elif time.time() > self.don_at + 20:
|
|
self.don_at = None
|
|
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(),
|
|
"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()
|