DeeJanuz--frametop/gaze/gazecheck.py

"""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
            self.full_armed = True  # a calibration that closed unfinished opens again
            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()