Lumen

"""
Blink detection for Lumen on the Arduino UNO Q (Dragonwing QRB2210, aarch64 Debian).

Two things matter for AAC use and they are different problems:

  1. Detecting that the eye is closed.  Cheap geometry, no learning needed.
  2. Deciding the closure was INTENTIONAL.  A spontaneous blink lasts roughly
     100-400 ms.  A deliberate "I am choosing this" blink is held longer.
     Lumen only fires a selection when the eye stays shut past DWELL_MS, which
     is what keeps the interface from firing on every involuntary blink.

Primary detector is MediaPipe FaceMesh eye-aspect-ratio (EAR).
Fallback is an OpenCV Haar cascade, used only if MediaPipe will not import or
no wheel is available for aarch64.  The fallback is less accurate and is there
so the demo degrades instead of crashing.
"""

import math
import time
from collections import deque

import cv2
import numpy as np

# FaceMesh landmark indices, in EAR order: corner, upper, upper, corner, lower, lower
LEFT_EYE = (362, 385, 387, 263, 373, 380)
RIGHT_EYE = (33, 160, 158, 133, 153, 144)


def _ear(points):
    """Eye aspect ratio. Falls toward 0 as the lid closes."""
    p1, p2, p3, p4, p5, p6 = points
    vertical = math.dist(p2, p6) + math.dist(p3, p5)
    horizontal = 2.0 * math.dist(p1, p4)
    if horizontal < 1e-6:
        return 0.0
    return vertical / horizontal


class BlinkDetector:
    """Frame in, blink state out.

    update(frame) returns a dict:
        face        bool   a face was found this frame
        ear         float  smoothed eye aspect ratio (0 when no face)
        closed      bool   eye is currently below threshold
        closed_ms   int    how long the current closure has lasted
        select      bool   True on the single frame a deliberate blink completes
    """

    def __init__(self, dwell_ms=600, max_dwell_ms=2500, ratio=0.72, smoothing=3):
        # Threshold is relative to the user's own open-eye baseline rather than
        # a fixed number. Eye shape varies enough between people that a hard
        # constant like 0.21 fails on a meaningful fraction of users.
        self.ratio = ratio
        self.dwell_ms = dwell_ms
        self.max_dwell_ms = max_dwell_ms

        self.baseline = None
        self._baseline_samples = deque(maxlen=90)
        self._ear_window = deque(maxlen=smoothing)

        self._closed_since = None
        self._fired = False

        self._mesh = None
        self._cascade = None
        self.backend = "none"
        self._init_backend()

    # ---------- backend selection ----------

    def _init_backend(self):
        try:
            import mediapipe as mp

            self._mesh = mp.solutions.face_mesh.FaceMesh(
                max_num_faces=1,
                refine_landmarks=True,
                min_detection_confidence=0.5,
                min_tracking_confidence=0.5,
            )
            self.backend = "mediapipe"
            return
        except Exception as exc:  # noqa: BLE001 - any import/runtime failure falls back
            print(f"[lumen] mediapipe unavailable ({exc}), falling back to Haar")

        path = cv2.data.haarcascades + "haarcascade_eye.xml"
        self._cascade = cv2.CascadeClassifier(path)
        self.backend = "haar"

    # ---------- measurement ----------

    def _measure_mediapipe(self, frame):
        h, w = frame.shape[:2]
        result = self._mesh.process(cv2.cvtColor(frame, cv2.COLOR_BGR2RGB))
        if not result.multi_face_landmarks:
            return None
        lm = result.multi_face_landmarks[0].landmark

        def pts(idx):
            return [(lm[i].x * w, lm[i].y * h) for i in idx]

        # Average both eyes. One-eye users can pass a single tuple instead.
        return 0.5 * (_ear(pts(LEFT_EYE)) + _ear(pts(RIGHT_EYE)))

    def _measure_haar(self, frame):
        gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
        eyes = self._cascade.detectMultiScale(gray, 1.1, 5, minSize=(30, 30))
        if len(eyes) == 0:
            return None
        # Crude proxy: open eyes are detected, closed eyes usually are not.
        # Height/width of the detected box stands in for EAR.
        h_over_w = np.mean([h / float(w) for (_, _, w, h) in eyes])
        return float(h_over_w)

    # ---------- main entry ----------

    def update(self, frame):
        raw = (
            self._measure_mediapipe(frame)
            if self.backend == "mediapipe"
            else self._measure_haar(frame)
        )

        if raw is None:
            self._closed_since = None
            self._fired = False
            return {"face": False, "ear": 0.0, "closed": False, "closed_ms": 0, "select": False}

        self._ear_window.append(raw)
        ear = float(np.mean(self._ear_window))

        # Rolling baseline from open-eye frames only, so it tracks lighting and
        # head pose drift without being dragged down by the blinks themselves.
        if self.baseline is None or ear > self.baseline * 0.9:
            self._baseline_samples.append(ear)
        if len(self._baseline_samples) >= 15:
            self.baseline = float(np.percentile(self._baseline_samples, 75))

        if self.baseline is None:
            return {"face": True, "ear": ear, "closed": False, "closed_ms": 0, "select": False}

        closed = ear < self.baseline * self.ratio
        now = time.monotonic()
        select = False

        if closed:
            if self._closed_since is None:
                self._closed_since = now
                self._fired = False
            held_ms = int((now - self._closed_since) * 1000)
            # Very long closures are rest, sleep, or a dry eye. Do not select.
            if not self._fired and self.dwell_ms <= held_ms <= self.max_dwell_ms:
                select = True
                self._fired = True
        else:
            held_ms = 0
            self._closed_since = None
            self._fired = False

        return {
            "face": True,
            "ear": ear,
            "closed": closed,
            "closed_ms": held_ms,
            "select": select,
        }

    def close(self):
        if self._mesh is not None:
            self._mesh.close()

"""
Single-switch row scanning, the standard AAC access method.

The highlight steps through phrases on a fixed dwell. One deliberate blink
selects whatever is highlighted. That is the entire input vocabulary: the user
needs exactly one reliable voluntary movement, which is the point.

Two groups so a mis-selection is cheap to recover from:
  - a short CARE group of urgent needs, scanned first and faster
  - a longer CONVERSATION group

The scanner also emits a "dwell warning" a beat before it moves on, which the
MCU turns into a haptic tick. Users report the tick matters more than the
visual highlight when the screen is out of their field of view.
"""

import time

CARE = [
    "Yes",
    "No",
    "I am in pain",
    "I need the bathroom",
    "I am thirsty",
    "Please call my nurse",
]

CONVERSATION = [
    "Hello",
    "Thank you",
    "Please wait",
    "I want to rest",
    "Say that again",
    "I love you",
]

GROUPS = [("Care", CARE), ("Conversation", CONVERSATION)]


class Scanner:
    def __init__(self, step_ms=1400, on_select=None, on_step=None):
        self.step_ms = step_ms
        self.on_select = on_select or (lambda text: None)
        self.on_step = on_step or (lambda: None)

        self.group_index = 0
        self.item_index = 0
        self.in_group = False  # False = choosing a group, True = choosing a phrase
        self.last_step = time.monotonic()
        self.spoken = []

    # ---------- current highlight ----------

    @property
    def group_name(self):
        return GROUPS[self.group_index][0]

    @property
    def items(self):
        return GROUPS[self.group_index][1]

    def highlight(self):
        if not self.in_group:
            return f"[{self.group_name}]"
        return self.items[self.item_index]

    # ---------- driven by main loop ----------

    def tick(self):
        """Advance the highlight if the dwell has elapsed."""
        now = time.monotonic()
        if (now - self.last_step) * 1000 < self.step_ms:
            return
        self.last_step = now

        if not self.in_group:
            self.group_index = (self.group_index + 1) % len(GROUPS)
        else:
            self.item_index += 1
            # One full pass through a group, then back out to group level.
            # Without this escape hatch a wrong group selection traps the user.
            if self.item_index >= len(self.items):
                self.item_index = 0
                self.in_group = False
        self.on_step()

    def select(self):
        """A deliberate blink landed. Descend, or emit the phrase."""
        self.last_step = time.monotonic()
        if not self.in_group:
            self.in_group = True
            self.item_index = 0
            return None

        phrase = self.items[self.item_index]
        self.spoken.append({"text": phrase, "at": time.time()})
        self.spoken = self.spoken[-25:]
        self.in_group = False
        self.item_index = 0
        self.on_select(phrase)
        return phrase

    def state(self):
        return {
            "group": self.group_name,
            "in_group": self.in_group,
            "highlight": self.highlight(),
            "items": self.items,
            "item_index": self.item_index if self.in_group else -1,
            "spoken": list(reversed(self.spoken)),
        }