diff --git a/services/ws-modules/pyeye1/pkg/et_ws_pyeye1.js b/services/ws-modules/pyeye1/pkg/et_ws_pyeye1.js
index d441af1..a4904a6 100644
--- a/services/ws-modules/pyeye1/pkg/et_ws_pyeye1.js
+++ b/services/ws-modules/pyeye1/pkg/et_ws_pyeye1.js
@@ -1,8 +1,9 @@
-// et_ws_pyeye1.js - Browser adapter for the MediaPipe FaceLandmarker eye-detection Pyodide workflow.
+// et_ws_pyeye1.js - Browser adapter for the MediaPipe FaceLandmarker eye-movement Pyodide workflow.
// Interface: default(), run(), start(), stop(), is_running()
//
// MediaPipe FaceLandmarker (a maintained, offline .task bundle that internally runs a face detector then a
-// mesh model) does the inference in JS; the Python layer turns its normalized landmarks into eye boxes.
+// mesh model) does the inference in JS; the Python layer turns its normalized landmarks into eye boxes,
+// iris circles, and the eye-misalignment / rhythmic-oscillation screening indicators this adapter renders.
const PYODIDE_BASE_PATH = "/modules/pyodide/";
@@ -14,7 +15,6 @@ const EYE_COLORS = {
let pyodide;
let py;
-let cfg;
let runtime = null;
export default async function init() {
@@ -50,7 +50,6 @@ export default async function init() {
if (globalThis.__etPyCov) await globalThis.__etPyCov.start(pyodide, "pyeye1");
py = pyodide.pyimport("pyeye1");
- cfg = py.config().toJs({ dict_converter: Object.fromEntries });
}
export const is_running = () => runtime !== null;
@@ -60,67 +59,71 @@ export async function run() {
if (!py) throw new Error("pyeye1: not initialized");
if (runtime) return;
- setStatus(py.starting_status());
- log(py.model_log_message());
-
- let client = null;
- let stream = null;
- let state = null;
-
+ const state = { client: null, stream: null, landmarker: null };
+ runtime = state;
try {
- const wasmAgent = await import("/modules/et-ws-wasm-agent/et_ws_wasm_agent.js");
- await wasmAgent.default();
- const { WsClient, WsClientConfig } = wasmAgent;
- const protocol = window.location.protocol === "https:" ? "wss:" : "ws:";
- client = new WsClient(new WsClientConfig(`${protocol}//${window.location.host}/ws`));
- client.connect();
- for (let i = 0; client.get_state() !== "connected" && i < 100; i++) await sleep(100);
- if (client.get_state() !== "connected") throw new Error("Timed out waiting for websocket connection");
- log(`websocket connected with agent_id=${client.get_agent_id()}`);
-
- stream = await navigator.mediaDevices.getUserMedia({ audio: false, video: true });
- const video = element("video-preview", HTMLVideoElement);
- video.srcObject = stream;
- video.hidden = false;
- for (let i = 0; video.videoWidth === 0 && i < 50; i++) await sleep(100);
- if (video.videoWidth === 0 || video.videoHeight === 0) throw new Error("Video stream metadata did not load");
- await video.play();
-
- // Load the MediaPipe tasks-vision runtime (served offline from its module mount) and build the landmarker.
- const vision = await import(cfg.bundle_path);
- const { FaceLandmarker, FilesetResolver } = vision;
- const fileset = await FilesetResolver.forVisionTasks(cfg.wasm_path);
- const landmarker = await FaceLandmarker.createFromOptions(fileset, {
- baseOptions: { modelAssetPath: cfg.model_path },
- runningMode: "VIDEO",
- numFaces: 1,
- });
-
- state = { client, stream, landmarker };
- runtime = state;
-
- await py.run(
- pyodide.toPy(() => inferEyes(state)),
- pyodide.toPy((message) => client.send(message)),
- pyodide.toPy(render),
- pyodide.toPy(sleep),
- pyodide.toPy(log),
- pyodide.toPy(setStatus),
- pyodide.toPy(() => runtime !== state),
- );
+ await py.run(platformFor(state));
} catch (err) {
log(`pyeye1 run failed: ${String(err)}`);
throw err;
} finally {
if (globalThis.__etPyCov) await globalThis.__etPyCov.stop(pyodide, "pyeye1");
- cleanup(state ?? { client, stream });
+ cleanup(state);
}
}
+// The Python workflow controls everything; each primitive here is one browser operation with no sequencing,
+// polling, or timeout logic (Python owns those). Members mirror the contract documented on pyeye1's run().
+function platformFor(state) {
+ return {
+ connect_ws: async () => {
+ const wasmAgent = await import("/modules/et-ws-wasm-agent/et_ws_wasm_agent.js");
+ await wasmAgent.default();
+ const { WsClient, WsClientConfig } = wasmAgent;
+ const protocol = window.location.protocol === "https:" ? "wss:" : "ws:";
+ state.client = new WsClient(new WsClientConfig(`${protocol}//${window.location.host}/ws`));
+ state.client.connect();
+ },
+ ws_state: () => state.client?.get_state() ?? "disconnected",
+ agent_id: () => state.client?.get_agent_id(),
+ send_event: (message) => state.client.send(message),
+ start_camera: async () => {
+ state.stream = await navigator.mediaDevices.getUserMedia({ audio: false, video: true });
+ const video = element("video-preview", HTMLVideoElement);
+ video.srcObject = state.stream;
+ // The raw stream stays hidden; it still feeds the landmarker, and the canvas shows the cropped view.
+ video.hidden = true;
+ },
+ video_size: () => {
+ const video = element("video-preview", HTMLVideoElement);
+ return [video.videoWidth, video.videoHeight];
+ },
+ play_video: () => element("video-preview", HTMLVideoElement).play(),
+ load_landmarker: async (modelPath, bundlePath, wasmPath) => {
+ // Load the MediaPipe tasks-vision runtime (served offline from its module mount) and build the
+ // landmarker; Python passes the asset paths so the configuration stays on the Python side.
+ const vision = await import(bundlePath);
+ const fileset = await vision.FilesetResolver.forVisionTasks(wasmPath);
+ state.landmarker = await vision.FaceLandmarker.createFromOptions(fileset, {
+ baseOptions: { modelAssetPath: modelPath },
+ runningMode: "VIDEO",
+ numFaces: 1,
+ });
+ },
+ infer: () => inferEyes(state),
+ render,
+ sleep,
+ log,
+ set_status: setStatus,
+ should_stop: () => runtime !== state,
+ cleanup: () => cleanup(state),
+ };
+}
+
export function stop() {
if (!runtime) return;
cleanup(runtime);
- log("pyeye1 eye detection demo stopped");
+ log("pyeye1 eye movement screening demo stopped");
}
async function inferEyes(state) {
@@ -141,21 +144,34 @@ function render(resultsJson) {
const video = element("video-preview", HTMLVideoElement);
if (video.videoWidth === 0 || video.videoHeight === 0) return;
+ const payload = JSON.parse(resultsJson);
+ // Python decides the visible region (the face band around the eyes); before a face is seen the crop is
+ // null and the full frame shows. Overlay coordinates stay in source pixels; the scale+translate maps them.
+ const [cropLeft, cropTop, cropRight, cropBottom] = payload.crop ?? [0, 0, video.videoWidth, video.videoHeight];
+ const cropWidth = Math.max(cropRight - cropLeft, 1);
+ const cropHeight = Math.max(cropBottom - cropTop, 1);
+
+ // Zoom the cropped band to the page width: the canvas element fills its container, and its backing store
+ // matches the displayed size so the upscaled video stays as sharp as the source allows.
const canvas = element("video-output-canvas", HTMLCanvasElement);
- const ctx = canvas.getContext("2d");
- canvas.width = video.videoWidth;
- canvas.height = video.videoHeight;
canvas.hidden = false;
- ctx.drawImage(video, 0, 0, canvas.width, canvas.height);
- ctx.font = "16px ui-monospace, monospace";
+ canvas.style.width = "100%";
+ canvas.style.height = "auto";
+ const displayWidth = Math.max(canvas.clientWidth, 1);
+ const scale = displayWidth / cropWidth;
+ canvas.width = displayWidth;
+ canvas.height = Math.max(Math.round(cropHeight * scale), 1);
- for (const result of JSON.parse(resultsJson)) {
- const [faceLeft, faceTop, faceRight, faceBottom] = result.face_box;
- ctx.lineWidth = 1;
- ctx.strokeStyle = "#7a8794";
- ctx.strokeRect(faceLeft, faceTop, Math.max(faceRight - faceLeft, 1), Math.max(faceBottom - faceTop, 1));
-
- ctx.lineWidth = 3;
+ const ctx = canvas.getContext("2d");
+ ctx.drawImage(video, cropLeft, cropTop, cropWidth, cropHeight, 0, 0, canvas.width, canvas.height);
+
+ ctx.save();
+ ctx.scale(scale, scale);
+ ctx.translate(-cropLeft, -cropTop);
+ // Divide widths/sizes by the zoom so strokes and labels keep a constant on-screen weight.
+ ctx.font = `${16 / scale}px ui-monospace, monospace`;
+ for (const result of payload.faces ?? []) {
+ ctx.lineWidth = 3 / scale;
for (const eye of result.eyes) {
const [left, top, right, bottom] = eye.box;
const color = EYE_COLORS[eye.label] ?? "#fffdfa";
@@ -164,7 +180,37 @@ function render(resultsJson) {
ctx.fillStyle = color;
ctx.fillText(eye.label === "left_eye" ? "L" : "R", left, Math.max(top - 4, 12));
}
+
+ ctx.lineWidth = 2 / scale;
+ for (const iris of result.irises ?? []) {
+ const [centerX, centerY] = iris.center;
+ ctx.strokeStyle = EYE_COLORS[iris.label] ?? "#fffdfa";
+ ctx.beginPath();
+ ctx.arc(centerX, centerY, Math.max(iris.radius, 1), 0, 2 * Math.PI);
+ ctx.stroke();
+ }
}
+ ctx.restore();
+ ctx.font = "16px ui-monospace, monospace";
+ renderAnalysis(ctx, payload.analysis);
+}
+
+// Screening-verdict overlay, top-left. Python sends analysis=null until the first window completes, and each
+// screening reports status "insufficient_data" until it has enough non-blink samples to rate.
+function renderAnalysis(ctx, analysis) {
+ const screenings = [
+ ["eye misalignment", analysis?.misalignment],
+ ["rhythmic oscillation", analysis?.oscillation],
+ ];
+ const lines = [];
+ for (const [name, metrics] of screenings) {
+ if (metrics?.status === "ok") lines.push(`${name}: ${metrics.detected ? "DETECTED" : "none"}`);
+ }
+ if (lines.length > 0) lines.push("screening demo -- not a medical diagnosis");
+ lines.forEach((line, index) => {
+ ctx.fillStyle = line.includes("DETECTED") ? "#ffb84d" : "#d7e0e8";
+ ctx.fillText(line, 8, 20 + index * 20);
+ });
}
function cleanup(state) {
diff --git a/services/ws-modules/pyeye1/pyeye1/__init__.py b/services/ws-modules/pyeye1/pyeye1/__init__.py
index 1ba455f..859afbe 100644
--- a/services/ws-modules/pyeye1/pyeye1/__init__.py
+++ b/services/ws-modules/pyeye1/pyeye1/__init__.py
@@ -1,10 +1,10 @@
-"""pyeye1: MediaPipe FaceLandmarker eye-detection support code (face -> eye bounding boxes)."""
+"""pyeye1: MediaPipe FaceLandmarker eye-movement support code (eye boxes, iris tracking, gaze screening)."""
from .eye_detection import (
EYE_MODEL_PATH,
build_results,
- config,
decode_eye_boxes,
+ decode_irises,
event_payload,
face_bounds,
model_log_message,
@@ -13,15 +13,20 @@
status_text,
stopped_status,
)
+from .gaze_analysis import analyze_window, gaze_sample, misalignment_metrics, oscillation_metrics
__all__ = [
"EYE_MODEL_PATH",
+ "analyze_window",
"build_results",
- "config",
"decode_eye_boxes",
+ "decode_irises",
"event_payload",
"face_bounds",
+ "gaze_sample",
+ "misalignment_metrics",
"model_log_message",
+ "oscillation_metrics",
"run",
"starting_status",
"status_text",
diff --git a/services/ws-modules/pyeye1/pyeye1/eye_detection.py b/services/ws-modules/pyeye1/pyeye1/eye_detection.py
index d4d6c76..9583692 100644
--- a/services/ws-modules/pyeye1/pyeye1/eye_detection.py
+++ b/services/ws-modules/pyeye1/pyeye1/eye_detection.py
@@ -1,8 +1,12 @@
-"""Eye bounding-box post-processing for the pyeye1 MediaPipe FaceLandmarker workflow.
-
-The browser shim runs Google's MediaPipe FaceLandmarker (a maintained, offline `.task` bundle that internally
-runs a face detector then a mesh model) and hands the resulting normalized face landmarks to this module, which
-turns each eye's contour-landmark cluster into a tight bounding box in source-image pixels.
+"""Eye-box and eye-movement post-processing for the pyeye1 MediaPipe FaceLandmarker workflow.
+
+This module's `run()` drives the whole workflow: WebSocket connection, camera acquisition, loading Google's
+MediaPipe FaceLandmarker (a maintained, offline `.task` bundle that internally runs a face detector then a
+mesh model), the sampling loop, and teardown. The browser shim contributes only a flat "platform" object of
+primitives -- each a single browser operation (getUserMedia, one landmarker inference, one canvas draw) with
+no sequencing, polling, or timeout logic of its own. Each frame's landmarks become per-eye bounding boxes
+plus iris circles for the overlay, and a rolling window of per-frame gaze samples feeds the eye-misalignment
+and rhythmic-oscillation screening heuristics in `gaze_analysis`.
"""
from __future__ import annotations
@@ -10,31 +14,61 @@
import json
import math
import time
+from collections import deque
from collections.abc import Iterable, Sequence
from datetime import datetime
+from statistics import fmean
from typing import Any, TypedDict
from et_ws.messages import WsClientEvent
+from .gaze_analysis import (
+ LEFT_IRIS_CENTER,
+ LEFT_IRIS_RING,
+ MESH_LANDMARK_COUNT,
+ RIGHT_IRIS_CENTER,
+ RIGHT_IRIS_RING,
+ GazeSample,
+ WindowAnalysis,
+ analyze_window,
+ gaze_sample,
+)
+
# Served by the MediaPipe tasks-vision runtime module and the model module (see config()).
EYE_MODEL_PATH = "/modules/et-model-eye1/face_landmarker.task"
VISION_BUNDLE_PATH = "/modules/@mediapipe/tasks-vision/vision_bundle.mjs"
VISION_WASM_PATH = "/modules/@mediapipe/tasks-vision/wasm"
-# FaceLandmarker returns 478 landmarks (the 468-point mesh plus 10 iris points), each x/y normalized to [0, 1]
-# against the input frame. The eye-box needs only the mesh contour indices (all < 468).
-MESH_LANDMARK_COUNT = 478
+# The eye-box decode needs only the face-mesh contour indices (all below this count); the iris circles and the
+# gaze screening additionally need the refined iris landmarks (see `gaze_analysis.MESH_LANDMARK_COUNT`).
CONTOUR_LANDMARK_COUNT = 468
# Eye-contour landmark indices in the MediaPipe mesh (subject's perspective, so "left" appears on the right of
# a non-mirrored frame). A tight per-eye box is the min/max over each cluster.
RIGHT_EYE_INDICES = (33, 7, 163, 144, 145, 153, 154, 155, 133, 173, 157, 158, 159, 160, 161, 246)
LEFT_EYE_INDICES = (263, 249, 390, 373, 374, 380, 381, 382, 362, 398, 384, 385, 386, 387, 388, 466)
-INFERENCE_INTERVAL_MS = 750
-RENDER_INTERVAL_MS = 60
-MAX_INFERENCES = 20
+# Frames are sampled continuously (the cadence must outpace the 2-10 Hz oscillations `gaze_analysis` screens
+# for), while status updates and WebSocket events go out at the slower analysis cadence.
+SAMPLE_INTERVAL_MS = 50
+MIN_SLEEP_MS = 10
+ANALYSIS_INTERVAL_MS = 1000
+ANALYSIS_WINDOW_MS = 2500
+MAX_SAMPLES = 600
MAX_RUNTIME_MS = 30_000
+# Setup polling. Python owns all sequencing and timeouts; the JS primitives never loop or wait on their own.
+POLL_INTERVAL_MS = 100
+WS_CONNECT_TIMEOUT_MS = 10_000
+VIDEO_READY_TIMEOUT_MS = 5_000
+
+# The overlay canvas shows only a face band around the eyes: the face's width (plus a small margin) by a
+# vertical band centered on the eye line. The band's half-height is a fraction of the face height (floored at
+# the eye cluster's own height), and the box is exponentially smoothed so per-frame landmark jitter and blinks
+# don't make the cropped view judder.
+CROP_HORIZONTAL_MARGIN = 0.04
+CROP_VERTICAL_EXTENT = 0.22
+CROP_SMOOTHING = 0.35
+
Box = list[float]
@@ -46,74 +80,138 @@ class Eye(TypedDict):
box: Box
+class Iris(TypedDict):
+ """One detected iris: its eye label, center point, and radius in source-image pixels."""
+
+ label: str
+ center: list[float]
+ radius: float
+
+
class FaceEyes(TypedDict):
- """One detected face: its overall landmark bounds and the two eye boxes within it."""
+ """One detected face: its overall landmark bounds plus the eye boxes and iris circles within it."""
face_box: Box
eyes: list[Eye]
+ irises: list[Iris]
-async def run(infer, send_event, render, sleep_ms, log, set_status, should_stop) -> None:
- """Run the browser eye detection workflow using JS platform callbacks.
+async def run(platform) -> None:
+ """Drive the whole eye movement screening workflow, using JS only for the browser primitives.
- `infer()` returns a JSON string `{"faces": [[x0, y0, x1, y1, ...], ...], "width": W, "height": H}` where each
- face is the flat list of normalized landmark coordinates from MediaPipe FaceLandmarker.
+ `platform` supplies the primitives Python cannot implement itself, each a single browser operation with
+ no sequencing, polling, or timeout logic. Sync members: `ws_state()`, `agent_id()`, `send_event(json)`,
+ `video_size() -> [w, h]`, `render(json)`, `log(str)`, `set_status(str)`, `should_stop()`, `cleanup()`.
+ Async members: `connect_ws()`, `start_camera()`, `play_video()`, `sleep(ms)`,
+ `load_landmarker(model_path, bundle_path, wasm_path)`, and `infer()`, which returns one FaceLandmarker
+ pass as the JSON string `{"faces": [[x0, y0, x1, y1, ...], ...], "width": W, "height": H}` where each
+ face is the flat list of normalized landmark coordinates.
"""
- inference_count = 0
+ platform.set_status(starting_status())
+ platform.log(model_log_message())
+ try:
+ await connect_websocket(platform)
+ await acquire_camera(platform)
+ await platform.load_landmarker(EYE_MODEL_PATH, VISION_BUNDLE_PATH, VISION_WASM_PATH)
+ await sample_loop(platform)
+ finally:
+ platform.cleanup()
+
+
+async def connect_websocket(platform) -> None:
+ """Open the WebSocket client and wait until the server acknowledges the agent connection."""
+ await platform.connect_ws()
+ await wait_until(
+ platform,
+ lambda: platform.ws_state() == "connected",
+ WS_CONNECT_TIMEOUT_MS,
+ "Timed out waiting for websocket connection",
+ )
+ platform.log(f"websocket connected with agent_id={platform.agent_id()}")
+
+
+async def acquire_camera(platform) -> None:
+ """Start the camera stream, wait for the video element to report its frame size, and begin playback."""
+ await platform.start_camera()
+
+ def video_ready() -> bool:
+ size = platform.video_size()
+ return size[0] > 0 and size[1] > 0
+
+ await wait_until(platform, video_ready, VIDEO_READY_TIMEOUT_MS, "Video stream metadata did not load")
+ await platform.play_video()
+
+
+async def wait_until(platform, predicate, timeout_ms: float, failure: str) -> None:
+ """Poll `predicate` every `POLL_INTERVAL_MS` until it holds, raising `RuntimeError(failure)` on timeout."""
+ waited_ms = 0.0
+ while not predicate():
+ if waited_ms >= timeout_ms:
+ raise RuntimeError(failure)
+ await platform.sleep(POLL_INTERVAL_MS)
+ waited_ms += POLL_INTERVAL_MS
+
+
+async def sample_loop(platform) -> None:
+ """Sample frames continuously; re-analyze the gaze window (status + event) per `ANALYSIS_INTERVAL_MS`."""
+ sample_count = 0
started_at = time.monotonic()
results: list[FaceEyes] = []
-
- set_status(starting_status())
-
- while not should_stop():
- elapsed_ms = (time.monotonic() - started_at) * 1000.0
- if inference_count >= MAX_INFERENCES or elapsed_ms >= MAX_RUNTIME_MS:
+ history: deque[GazeSample] = deque()
+ analysis: WindowAnalysis | None = None
+ crop: Box | None = None
+ last_analysis_ms = 0.0
+
+ while not platform.should_stop():
+ loop_started = time.monotonic()
+ if sample_count >= MAX_SAMPLES or (loop_started - started_at) * 1000.0 >= MAX_RUNTIME_MS:
break
try:
- capture = json.loads(await infer())
- results = build_results(capture["faces"], capture["width"], capture["height"])
- inference_count += 1
-
- set_status(status_text(results))
- render(results_json(results))
- send_event(client_event_json(event_payload(results, capture["width"], capture["height"])))
+ capture = json.loads(await platform.infer())
+ now_s = time.monotonic() - started_at
+ width, height = capture["width"], capture["height"]
+ results = build_results(capture["faces"], width, height)
+ sample_count += 1
+
+ # The gaze window and the cropped view track the first (and, with numFaces=1, only) face; while
+ # no face is visible the last smoothed crop is kept rather than snapping back to the full frame.
+ if results:
+ crop = smooth_crop(crop, eye_region_crop(results[0], width, height))
+ if capture["faces"]:
+ history.append(gaze_sample(capture["faces"][0], width, height, now_s))
+ while history and (now_s - history[0]["t"]) * 1000.0 > ANALYSIS_WINDOW_MS:
+ history.popleft()
+
+ if now_s * 1000.0 - last_analysis_ms >= ANALYSIS_INTERVAL_MS:
+ last_analysis_ms = now_s * 1000.0
+ analysis = analyze_window(list(history))
+ platform.set_status(status_text(results, analysis))
+ platform.send_event(client_event_json(event_payload(results, analysis, width, height)))
+ platform.render(results_json(results, analysis, crop))
except Exception as exc:
- message = f"pyeye1 eye detection: inference error\n{exc}"
- set_status(message)
- log(f"inference error: {exc}")
-
- remaining_ms = INFERENCE_INTERVAL_MS
- while remaining_ms > 0 and not should_stop():
- render(results_json(results))
- delay = min(RENDER_INTERVAL_MS, remaining_ms)
- await sleep_ms(delay)
- remaining_ms -= delay
-
- if inference_count >= MAX_INFERENCES:
- log(f"workflow finished automatically after {MAX_INFERENCES} inferences")
- elif (time.monotonic() - started_at) * 1000.0 >= MAX_RUNTIME_MS:
- log("workflow finished automatically after 30 seconds")
- set_status(stopped_status())
+ message = f"pyeye1 eye movement screening: inference error\n{exc}"
+ platform.set_status(message)
+ platform.log(f"inference error: {exc}")
+ spent_ms = (time.monotonic() - loop_started) * 1000.0
+ await platform.sleep(max(SAMPLE_INTERVAL_MS - spent_ms, MIN_SLEEP_MS))
-def config() -> dict[str, object]:
- """Return browser-facing constants: the model asset plus the MediaPipe runtime bundle + wasm roots."""
- return {
- "model_path": EYE_MODEL_PATH,
- "bundle_path": VISION_BUNDLE_PATH,
- "wasm_path": VISION_WASM_PATH,
- }
+ if sample_count >= MAX_SAMPLES:
+ platform.log(f"workflow finished automatically after {MAX_SAMPLES} samples")
+ elif (time.monotonic() - started_at) * 1000.0 >= MAX_RUNTIME_MS:
+ platform.log("workflow finished automatically after 30 seconds")
+ platform.set_status(stopped_status())
def starting_status() -> str:
"""Return the status line shown while the workflow starts up."""
- return "pyeye1 eye detection: starting"
+ return "pyeye1 eye movement screening: starting"
def stopped_status() -> str:
"""Return the status line shown once the workflow has stopped."""
- return "pyeye1 eye detection demo stopped."
+ return "pyeye1 eye movement screening demo stopped."
def model_log_message() -> str:
@@ -142,6 +240,28 @@ def eye_box(values: Sequence[float], indices: Sequence[int], width: float, heigh
return [min(xs), min(ys), max(xs), max(ys)]
+def iris_circle(
+ values: Sequence[float], label: str, center_index: int, ring: Sequence[int], width: float, height: float
+) -> Iris:
+ """Map one iris cluster (center + four-point ring) to a labelled circle in source pixels."""
+ center_x = clamp(values[center_index * 2] * width, 0.0, width)
+ center_y = clamp(values[center_index * 2 + 1] * height, 0.0, height)
+ distances = [
+ math.hypot(values[index * 2] * width - center_x, values[index * 2 + 1] * height - center_y) for index in ring
+ ]
+ return {"label": label, "center": [center_x, center_y], "radius": fmean(distances)}
+
+
+def decode_irises(values: Sequence[float], width: float, height: float) -> list[Iris]:
+ """Decode both iris circles from one face's normalized FaceLandmarker landmarks."""
+ if len(values) < MESH_LANDMARK_COUNT * 2:
+ raise ValueError("FaceLandmarker output did not contain the iris landmarks")
+ return [
+ iris_circle(values, "left_eye", LEFT_IRIS_CENTER, LEFT_IRIS_RING, width, height),
+ iris_circle(values, "right_eye", RIGHT_IRIS_CENTER, RIGHT_IRIS_RING, width, height),
+ ]
+
+
def face_bounds(landmarks: Sequence[float], width: float, height: float) -> Box:
"""Return the bounding box of all of a face's landmarks, in source pixels (context for the eye overlay)."""
xs = [clamp(landmarks[index] * width, 0.0, width) for index in range(0, len(landmarks), 2)]
@@ -150,7 +270,7 @@ def face_bounds(landmarks: Sequence[float], width: float, height: float) -> Box:
def build_results(faces: Sequence[Any], width: float, height: float) -> list[FaceEyes]:
- """Combine each detected face's landmarks into its overall bounds plus its two eye boxes."""
+ """Combine each face's landmarks into its overall bounds, its two eye boxes, and its iris circles."""
width = require_positive_finite(width, "width")
height = require_positive_finite(height, "height")
results: list[FaceEyes] = []
@@ -164,18 +284,43 @@ def build_results(faces: Sequence[Any], width: float, height: float) -> list[Fac
{"label": "left_eye", "box": boxes["left_eye"]},
{"label": "right_eye", "box": boxes["right_eye"]},
],
+ "irises": decode_irises(values, width, height),
}
)
return results
-def results_json(results: Sequence[FaceEyes]) -> str:
- """Serialise the per-face eye results to JSON for the renderer."""
- return json.dumps(list(results))
+def eye_region_crop(result: FaceEyes, width: float, height: float) -> Box:
+ """Return the source-pixel crop showing only the face band around the eyes, clamped to the frame."""
+ face = result["face_box"]
+ eye_boxes = [eye["box"] for eye in result["eyes"]]
+ eye_top = min(box[1] for box in eye_boxes)
+ eye_bottom = max(box[3] for box in eye_boxes)
+ band_center = (eye_top + eye_bottom) / 2.0
+ half_band = max(CROP_VERTICAL_EXTENT * (face[3] - face[1]), eye_bottom - eye_top)
+ margin_x = CROP_HORIZONTAL_MARGIN * (face[2] - face[0])
+ return [
+ clamp(face[0] - margin_x, 0.0, width),
+ clamp(band_center - half_band, 0.0, height),
+ clamp(face[2] + margin_x, 0.0, width),
+ clamp(band_center + half_band, 0.0, height),
+ ]
+
+
+def smooth_crop(previous: Box | None, target: Box) -> Box:
+ """Blend the new crop target into the previous one so the cropped view pans smoothly, not frame-jumpy."""
+ if previous is None:
+ return target
+ return [old + CROP_SMOOTHING * (new - old) for old, new in zip(previous, target, strict=True)]
+
+
+def results_json(results: Sequence[FaceEyes], analysis: WindowAnalysis | None, crop: Box | None = None) -> str:
+ """Serialise the per-face results, the latest window analysis, and the view crop for the renderer."""
+ return json.dumps({"faces": list(results), "analysis": analysis, "crop": crop})
def client_event_json(details: dict[str, object]) -> str:
- """Build the et-client-event JSON envelope for an eye-detection inference."""
+ """Build the et-client-event JSON envelope for an eye-screening analysis pass."""
return WsClientEvent(
type="et-client-event",
capability="eye_detection",
@@ -184,41 +329,78 @@ def client_event_json(details: dict[str, object]) -> str:
).model_dump_json()
-def status_text(results: Sequence[FaceEyes]) -> str:
- """Render the browser status text used by the eye detection demo."""
+def status_text(results: Sequence[FaceEyes], analysis: WindowAnalysis | None) -> str:
+ """Render the browser status text used by the eye movement screening demo."""
eye_count = sum(len(result["eyes"]) for result in results)
lines = [
- "pyeye1 eye detection demo",
+ "pyeye1 eye movement screening demo",
f"model file: {EYE_MODEL_PATH}",
f"faces: {len(results)}",
f"eyes: {eye_count}",
+ *analysis_lines(analysis),
+ "screening heuristics only -- not a medical diagnosis",
f"processed at: {datetime.now().strftime('%X')}",
]
-
- if results and results[0]["eyes"]:
- box = results[0]["eyes"][0]["box"]
- lines.extend(["", f"first eye: {box[0]:.1f}, {box[1]:.1f}, {box[2]:.1f}, {box[3]:.1f}"])
-
return "\n".join(lines)
-def event_payload(results: Sequence[FaceEyes], width: float, height: float) -> dict[str, object]:
- """Build the WebSocket client event payload for one eye-detection inference."""
+def analysis_lines(analysis: WindowAnalysis | None) -> list[str]:
+ """Summarise one window analysis as status lines (or a warming-up placeholder before the first window)."""
+ if analysis is None:
+ return ["analysis: warming up"]
+ seconds = analysis["window_ms"] / 1000.0
+ lines = [f"window: {analysis['valid_samples']}/{analysis['samples']} valid samples over {seconds:.1f}s"]
+
+ misalignment = analysis["misalignment"]
+ if misalignment["status"] == "ok":
+ verdict = "DETECTED" if misalignment["detected"] else "none"
+ dh = misalignment["horizontal_deviation"] or 0.0
+ dv = misalignment["vertical_deviation"] or 0.0
+ lines.append(f"alignment dh={dh:+.3f} dv={dv:+.3f} -> eye misalignment: {verdict}")
+ else:
+ lines.append("alignment: insufficient data")
+
+ oscillation = analysis["oscillation"]
+ if oscillation["status"] == "ok":
+ verdict = "DETECTED" if oscillation["detected"] else "none"
+ frequency = oscillation["frequency_hz"] or 0.0
+ amplitude = oscillation["amplitude"] or 0.0
+ axis = oscillation["axis"]
+ lines.append(f"{frequency:.1f} Hz amp {amplitude:.3f} ({axis}) -> rhythmic oscillation: {verdict}")
+ else:
+ lines.append("oscillation: insufficient data")
+ return lines
+
+
+def event_payload(
+ results: Sequence[FaceEyes], analysis: WindowAnalysis | None, width: float, height: float
+) -> dict[str, object]:
+ """Build the WebSocket client event payload for one eye-screening analysis pass."""
width = require_non_negative_finite(width, "width")
height = require_non_negative_finite(height, "height")
eye_count = sum(len(result["eyes"]) for result in results)
return {
"mode": "eye_detection",
- "detected_class": "eyes" if eye_count else "no_detection",
+ "detected_class": detected_class(eye_count, analysis),
"faces": len(results),
"eyes": eye_count,
"results": list(results),
+ "analysis": analysis,
"processed_at": datetime.now().strftime("%X"),
"model_path": EYE_MODEL_PATH,
"source_resolution": {"width": float(width), "height": float(height)},
}
+def detected_class(eye_count: int, analysis: WindowAnalysis | None) -> str:
+ """Pick the event's headline class: a flagged screening indicator wins over plain eye presence."""
+ if analysis is not None and analysis["oscillation"]["detected"]:
+ return "rhythmic_oscillation_indicator"
+ if analysis is not None and analysis["misalignment"]["detected"]:
+ return "eye_misalignment_indicator"
+ return "eyes" if eye_count else "no_detection"
+
+
def clamp(value: float, minimum: float, maximum: float) -> float:
"""Clamp `value` to the inclusive range [minimum, maximum]."""
return max(minimum, min(value, maximum))
diff --git a/services/ws-modules/pyeye1/pyeye1/gaze_analysis.py b/services/ws-modules/pyeye1/pyeye1/gaze_analysis.py
new file mode 100644
index 0000000..267d406
--- /dev/null
+++ b/services/ws-modules/pyeye1/pyeye1/gaze_analysis.py
@@ -0,0 +1,253 @@
+"""Eye-misalignment and rhythmic-oscillation screening heuristics over FaceLandmarker iris landmarks.
+
+MediaPipe FaceLandmarker's refined mesh carries five iris landmarks per eye (a center plus a four-point ring)
+alongside the eyelid and eye-corner contour points. Each video frame is reduced to one `GazeSample`: per eye,
+the iris center is projected onto the corner-to-corner eye axis (horizontal ratio) and onto the upper-to-lower
+eyelid axis (vertical ratio), both normalized so 0.5 means "centered in the eye opening". Ratios are unitless
+fractions of the eye's own width/height, which keeps them stable across face sizes, camera distances, and
+moderate head roll.
+
+Two windowed screenings run over the sample history, each named for exactly what it measures (these are the
+gaze patterns an eye-care professional would investigate as strabismus and nystagmus respectively, but no
+diagnosis is made or implied):
+
+- Eye misalignment indicator -- the two eyes point in sustainedly different directions, measured as the
+ left-minus-right ratio difference staying offset. Both eye axes are ordered image-left to image-right, so a
+ conjugate gaze shift (both eyes looking the same way) moves both ratios together and cancels in the
+ difference; only true misalignment survives the mean.
+- Rhythmic oscillation indicator -- the eyes shake rhythmically instead of holding steady, showing up as a
+ periodic signal in the conjugate (two-eye mean) ratio series. The series is detrended with a centered
+ moving mean (removing smooth pursuit and slow head drift), then rated by zero-crossing frequency and
+ RMS-derived amplitude.
+
+Samples taken mid-blink (eyelid gap under `MIN_OPENNESS` of the eye width) are dropped before analysis, since
+a closing lid drags the detected iris and would fake both indicators.
+
+These are screening heuristics over webcam landmarks -- NOT a medical diagnosis. Thresholds are conservative
+defaults chosen to reject landmark jitter, not clinically calibrated values.
+"""
+
+from __future__ import annotations
+
+import math
+from collections.abc import Iterable, Sequence
+from itertools import pairwise
+from statistics import fmean
+from typing import Any, TypedDict
+
+# FaceLandmarker returns 478 landmarks: the 468-point face mesh plus one five-point iris cluster per eye (a
+# center followed by a four-point ring), each x/y normalized to [0, 1] against the input frame.
+MESH_LANDMARK_COUNT = 478
+
+RIGHT_IRIS_CENTER = 468
+RIGHT_IRIS_RING = (469, 470, 471, 472)
+LEFT_IRIS_CENTER = 473
+LEFT_IRIS_RING = (474, 475, 476, 477)
+
+# Eye-corner pairs ordered image-left -> image-right on a non-mirrored frame ("right" is the subject's right
+# eye, which appears on the left of the image). Keeping both axes in image order makes a conjugate gaze shift
+# move both horizontal ratios the same direction, so it cancels in the misalignment left-minus-right
+# difference.
+RIGHT_EYE_AXIS = (33, 133)
+LEFT_EYE_AXIS = (362, 263)
+# Upper then lower eyelid mid landmarks; they define the vertical ratio axis and the blink guard.
+RIGHT_EYE_LIDS = (159, 145)
+LEFT_EYE_LIDS = (386, 374)
+
+# Samples where either eye's lid gap is under this fraction of the eye width count as blinks and are excluded.
+MIN_OPENNESS = 0.15
+# Both screenings need this many valid samples (and oscillation this much time span) before rating a window.
+MIN_ANALYSIS_SAMPLES = 12
+MIN_ANALYSIS_SPAN_S = 1.0
+
+# Sustained left-minus-right ratio offset (a fraction of each eye's own width/height) that flags misalignment.
+MISALIGNMENT_RATIO_THRESHOLD = 0.10
+
+# Centered moving-mean span for detrending; spans longer than one oscillation period would erase the signal.
+DETREND_SPAN_S = 0.4
+# Pathological rhythmic eye movement beats at roughly 2-10 Hz. The amplitude floor rejects landmark jitter,
+# which crosses zero often but stays tiny, and the crossing floor rejects a one-off saccade or two within the
+# window.
+OSCILLATION_MIN_HZ = 2.0
+OSCILLATION_MAX_HZ = 10.0
+OSCILLATION_MIN_AMPLITUDE = 0.04
+OSCILLATION_MIN_CROSSINGS = 6
+
+
+class EyeGaze(TypedDict):
+ """One eye's gaze ratios: iris position within the eye opening, plus how open the eyelids are."""
+
+ h: float
+ v: float
+ openness: float
+
+
+class GazeSample(TypedDict):
+ """Both eyes' gaze ratios at one video timestamp (seconds since the workflow started)."""
+
+ t: float
+ left: EyeGaze
+ right: EyeGaze
+ valid: bool
+
+
+class MisalignmentMetrics(TypedDict):
+ """Windowed eye-alignment metrics; deviations are mean left-minus-right ratio differences."""
+
+ status: str
+ detected: bool
+ horizontal_deviation: float | None
+ vertical_deviation: float | None
+
+
+class OscillationMetrics(TypedDict):
+ """Windowed rhythmic-oscillation metrics for the dominant axis of the conjugate gaze signal."""
+
+ status: str
+ detected: bool
+ axis: str | None
+ frequency_hz: float | None
+ amplitude: float | None
+
+
+class WindowAnalysis(TypedDict):
+ """One analysis pass over the sliding sample window: counts plus both screening results."""
+
+ window_ms: float
+ samples: int
+ valid_samples: int
+ misalignment: MisalignmentMetrics
+ oscillation: OscillationMetrics
+
+
+def landmark_px(values: Sequence[float], index: int, width: float, height: float) -> tuple[float, float]:
+ """Return one normalized landmark as an (x, y) point in source pixels."""
+ return (values[index * 2] * width, values[index * 2 + 1] * height)
+
+
+def eye_gaze(
+ values: Sequence[float],
+ width: float,
+ height: float,
+ axis: tuple[int, int],
+ lids: tuple[int, int],
+ iris_center: int,
+) -> EyeGaze:
+ """Project one eye's iris center onto its corner and eyelid axes, yielding normalized gaze ratios."""
+ corner_ax, corner_ay = landmark_px(values, axis[0], width, height)
+ corner_bx, corner_by = landmark_px(values, axis[1], width, height)
+ dx, dy = corner_bx - corner_ax, corner_by - corner_ay
+ axis_len_sq = dx * dx + dy * dy
+ if axis_len_sq <= 0.0:
+ raise ValueError("eye corner landmarks are degenerate")
+ iris_x, iris_y = landmark_px(values, iris_center, width, height)
+ h = ((iris_x - corner_ax) * dx + (iris_y - corner_ay) * dy) / axis_len_sq
+
+ upper_x, upper_y = landmark_px(values, lids[0], width, height)
+ lower_x, lower_y = landmark_px(values, lids[1], width, height)
+ vx, vy = lower_x - upper_x, lower_y - upper_y
+ lid_len_sq = vx * vx + vy * vy
+ # A fully closed lid collapses the vertical axis; report a centered ratio and let `openness` gate it out.
+ v = 0.5 if lid_len_sq <= 0.0 else ((iris_x - upper_x) * vx + (iris_y - upper_y) * vy) / lid_len_sq
+ return {"h": h, "v": v, "openness": math.sqrt(lid_len_sq / axis_len_sq)}
+
+
+def gaze_sample(landmarks: Iterable[Any], width: float, height: float, t: float) -> GazeSample:
+ """Build one two-eye gaze sample from a face's flat normalized FaceLandmarker landmarks."""
+ values = [float(value) for value in landmarks]
+ if len(values) < MESH_LANDMARK_COUNT * 2:
+ raise ValueError("FaceLandmarker output did not contain the iris landmarks")
+ left = eye_gaze(values, width, height, LEFT_EYE_AXIS, LEFT_EYE_LIDS, LEFT_IRIS_CENTER)
+ right = eye_gaze(values, width, height, RIGHT_EYE_AXIS, RIGHT_EYE_LIDS, RIGHT_IRIS_CENTER)
+ valid = left["openness"] >= MIN_OPENNESS and right["openness"] >= MIN_OPENNESS
+ return {"t": float(t), "left": left, "right": right, "valid": valid}
+
+
+def misalignment_metrics(samples: Sequence[GazeSample]) -> MisalignmentMetrics:
+ """Rate windowed eye alignment: a sustained left-minus-right ratio offset flags the indicator."""
+ valid = [sample for sample in samples if sample["valid"]]
+ if len(valid) < MIN_ANALYSIS_SAMPLES:
+ return {
+ "status": "insufficient_data",
+ "detected": False,
+ "horizontal_deviation": None,
+ "vertical_deviation": None,
+ }
+ dh = fmean(sample["left"]["h"] - sample["right"]["h"] for sample in valid)
+ dv = fmean(sample["left"]["v"] - sample["right"]["v"] for sample in valid)
+ return {
+ "status": "ok",
+ "detected": abs(dh) > MISALIGNMENT_RATIO_THRESHOLD or abs(dv) > MISALIGNMENT_RATIO_THRESHOLD,
+ "horizontal_deviation": round(dh, 4),
+ "vertical_deviation": round(dv, 4),
+ }
+
+
+def detrended(ts: Sequence[float], xs: Sequence[float]) -> list[float]:
+ """Subtract a centered moving mean (span `DETREND_SPAN_S`) so smooth pursuit and slow drift drop out."""
+ half_span = DETREND_SPAN_S / 2.0
+ residual: list[float] = []
+ # O(n^2) over a ~50-sample window is trivially cheap and tolerates the irregular timestamps blinks leave.
+ for t, x in zip(ts, xs, strict=True):
+ local = [value for stamp, value in zip(ts, xs, strict=True) if abs(stamp - t) <= half_span]
+ residual.append(x - fmean(local))
+ return residual
+
+
+def series_oscillation(ts: Sequence[float], xs: Sequence[float]) -> tuple[float, float, int]:
+ """Return (frequency_hz, amplitude, zero_crossings) of the detrended series via zero-crossing counting."""
+ residual = detrended(ts, xs)
+ duration = ts[-1] - ts[0]
+ crossings = sum(1 for a, b in pairwise(residual) if a * b < 0.0)
+ frequency = crossings / (2.0 * duration) if duration > 0.0 else 0.0
+ rms = math.sqrt(fmean(value * value for value in residual))
+ # For a pure sinusoid RMS * sqrt(2) recovers the peak amplitude; landmark jitter stays well below it.
+ return frequency, rms * math.sqrt(2.0), crossings
+
+
+def oscillation_metrics(samples: Sequence[GazeSample]) -> OscillationMetrics:
+ """Rate windowed rhythmic movement of the conjugate (two-eye mean) gaze signal, per axis."""
+ valid = [sample for sample in samples if sample["valid"]]
+ if len(valid) < MIN_ANALYSIS_SAMPLES or valid[-1]["t"] - valid[0]["t"] < MIN_ANALYSIS_SPAN_S:
+ return {
+ "status": "insufficient_data",
+ "detected": False,
+ "axis": None,
+ "frequency_hz": None,
+ "amplitude": None,
+ }
+
+ ts = [sample["t"] for sample in valid]
+ horizontal = [(sample["left"]["h"] + sample["right"]["h"]) / 2.0 for sample in valid]
+ vertical = [(sample["left"]["v"] + sample["right"]["v"]) / 2.0 for sample in valid]
+
+ # Rate each axis, then report a passing axis if any (the larger-amplitude one), else the dominant axis.
+ rated: list[tuple[bool, float, float, int, str]] = []
+ for axis_label, series in (("horizontal", horizontal), ("vertical", vertical)):
+ frequency, amplitude, crossings = series_oscillation(ts, series)
+ passes = (
+ OSCILLATION_MIN_HZ <= frequency <= OSCILLATION_MAX_HZ
+ and amplitude >= OSCILLATION_MIN_AMPLITUDE
+ and crossings >= OSCILLATION_MIN_CROSSINGS
+ )
+ rated.append((passes, amplitude, frequency, crossings, axis_label))
+ passes, amplitude, frequency, _crossings, axis_label = max(rated)
+ return {
+ "status": "ok",
+ "detected": passes,
+ "axis": axis_label,
+ "frequency_hz": round(frequency, 2),
+ "amplitude": round(amplitude, 4),
+ }
+
+
+def analyze_window(samples: Sequence[GazeSample]) -> WindowAnalysis:
+ """Run both screenings over the current sliding window of gaze samples."""
+ window_ms = (samples[-1]["t"] - samples[0]["t"]) * 1000.0 if samples else 0.0
+ return {
+ "window_ms": round(window_ms, 1),
+ "samples": len(samples),
+ "valid_samples": sum(1 for sample in samples if sample["valid"]),
+ "misalignment": misalignment_metrics(samples),
+ "oscillation": oscillation_metrics(samples),
+ }
diff --git a/services/ws-modules/pyeye1/pyproject.toml b/services/ws-modules/pyeye1/pyproject.toml
index 18f3cdb..9575e0e 100644
--- a/services/ws-modules/pyeye1/pyproject.toml
+++ b/services/ws-modules/pyeye1/pyproject.toml
@@ -1,6 +1,6 @@
[project]
dependencies = ["et-ws"]
-description = "Python eye detection (MediaPipe)"
+description = "Python eye screening (MediaPipe)"
license = "Apache-2.0 OR MIT"
name = "et-ws-pyeye1"
requires-python = ">=3.10"
diff --git a/services/ws-modules/pyeye1/tests/test_eye_detection.py b/services/ws-modules/pyeye1/tests/test_eye_detection.py
index 141040f..98b0d76 100644
--- a/services/ws-modules/pyeye1/tests/test_eye_detection.py
+++ b/services/ws-modules/pyeye1/tests/test_eye_detection.py
@@ -5,10 +5,27 @@
LEFT_EYE_INDICES,
MESH_LANDMARK_COUNT,
RIGHT_EYE_INDICES,
+ FaceEyes,
build_results,
decode_eye_boxes,
+ decode_irises,
+ eye_region_crop,
face_bounds,
+ smooth_crop,
)
+from pyeye1.gaze_analysis import RIGHT_IRIS_CENTER, RIGHT_IRIS_RING
+
+
+def face_eyes(face_box: list[float], left_eye: list[float], right_eye: list[float]) -> FaceEyes:
+ """Build a FaceEyes result directly from boxes, skipping landmark decoding."""
+ return {
+ "face_box": face_box,
+ "eyes": [
+ {"label": "left_eye", "box": left_eye},
+ {"label": "right_eye", "box": right_eye},
+ ],
+ "irises": [],
+ }
def landmarks_with(points: dict[int, tuple[float, float]]) -> list[float]:
@@ -75,6 +92,53 @@ def test_build_results_pairs_each_face_with_two_eyes(self) -> None:
self.assertEqual(len(results), 1)
self.assertEqual([eye["label"] for eye in results[0]["eyes"]], ["left_eye", "right_eye"])
self.assertEqual(len(results[0]["face_box"]), 4)
+ self.assertEqual([iris["label"] for iris in results[0]["irises"]], ["left_eye", "right_eye"])
+
+ def test_decode_irises_maps_center_and_ring_to_a_pixel_circle(self) -> None:
+ # Center at (0.35, 0.50) with the ring 0.01 out on each side; in a 1000x500 frame the horizontal ring
+ # points sit 10px from the center and the vertical ones 5px, so the radius averages to 7.5px.
+ points = {RIGHT_IRIS_CENTER: (0.35, 0.50)}
+ ring_offsets = ((0.01, 0.0), (0.0, -0.01), (-0.01, 0.0), (0.0, 0.01))
+ for index, (dx, dy) in zip(RIGHT_IRIS_RING, ring_offsets, strict=True):
+ points[index] = (0.35 + dx, 0.50 + dy)
+ irises = decode_irises(landmarks_with(points), 1000.0, 500.0)
+ right = next(iris for iris in irises if iris["label"] == "right_eye")
+ self.assertAlmostEqual(right["center"][0], 350.0)
+ self.assertAlmostEqual(right["center"][1], 250.0)
+ self.assertAlmostEqual(right["radius"], 7.5)
+
+ def test_decode_irises_rejects_contour_only_landmarks(self) -> None:
+ with self.assertRaisesRegex(ValueError, "iris landmarks"):
+ decode_irises([0.0] * (CONTOUR_LANDMARK_COUNT * 2), 640.0, 480.0)
+
+ def test_eye_region_crop_is_a_face_wide_band_centered_on_the_eyes(self) -> None:
+ result = face_eyes(
+ face_box=[100.0, 50.0, 300.0, 350.0],
+ left_eye=[220.0, 150.0, 260.0, 170.0],
+ right_eye=[140.0, 150.0, 180.0, 168.0],
+ )
+ crop = eye_region_crop(result, 640.0, 480.0)
+ # Face width 200 -> 4% margin of 8 each side; eye band 150..170 centers at 160, and the half-height
+ # is 22% of the 300px face height (66), which beats the 20px eye-cluster floor.
+ self.assertEqual(crop, [92.0, 94.0, 308.0, 226.0])
+
+ def test_eye_region_crop_clamps_to_the_frame(self) -> None:
+ result = face_eyes(
+ face_box=[0.0, 0.0, 200.0, 200.0],
+ left_eye=[120.0, 40.0, 160.0, 55.0],
+ right_eye=[20.0, 40.0, 60.0, 55.0],
+ )
+ crop = eye_region_crop(result, 205.0, 90.0)
+ self.assertEqual(crop[0], 0.0)
+ self.assertEqual(crop[2], 205.0)
+ self.assertAlmostEqual(crop[1], 3.5)
+ self.assertEqual(crop[3], 90.0)
+
+ def test_smooth_crop_starts_at_the_target_then_eases_toward_it(self) -> None:
+ target = [10.0, 10.0, 110.0, 110.0]
+ self.assertEqual(smooth_crop(None, target), target)
+ eased = smooth_crop([0.0, 0.0, 100.0, 100.0], target)
+ self.assertEqual(eased, [3.5, 3.5, 103.5, 103.5])
if __name__ == "__main__":
diff --git a/services/ws-modules/pyeye1/tests/test_gaze_analysis.py b/services/ws-modules/pyeye1/tests/test_gaze_analysis.py
new file mode 100644
index 0000000..9de0de1
--- /dev/null
+++ b/services/ws-modules/pyeye1/tests/test_gaze_analysis.py
@@ -0,0 +1,218 @@
+import math
+import unittest
+
+from pyeye1.gaze_analysis import (
+ LEFT_EYE_AXIS,
+ LEFT_EYE_LIDS,
+ LEFT_IRIS_CENTER,
+ LEFT_IRIS_RING,
+ MESH_LANDMARK_COUNT,
+ MIN_ANALYSIS_SAMPLES,
+ RIGHT_EYE_AXIS,
+ RIGHT_EYE_LIDS,
+ RIGHT_IRIS_CENTER,
+ RIGHT_IRIS_RING,
+ GazeSample,
+ analyze_window,
+ gaze_sample,
+ misalignment_metrics,
+ oscillation_metrics,
+)
+
+
+def landmarks_with(points: dict[int, tuple[float, float]]) -> list[float]:
+ """Build a flat 478*2 normalized landmark list, placing (x, y) at the given landmark indices."""
+ values = [0.0] * (MESH_LANDMARK_COUNT * 2)
+ for index, (x, y) in points.items():
+ values[index * 2] = x
+ values[index * 2 + 1] = y
+ return values
+
+
+def face_with_gaze(right_iris: tuple[float, float], left_iris: tuple[float, float]) -> list[float]:
+ """Build a face whose eyes sit on fixed horizontal axes with open lids and the given iris centers."""
+ return landmarks_with(
+ {
+ RIGHT_EYE_AXIS[0]: (0.30, 0.50),
+ RIGHT_EYE_AXIS[1]: (0.40, 0.50),
+ RIGHT_EYE_LIDS[0]: (0.35, 0.48),
+ RIGHT_EYE_LIDS[1]: (0.35, 0.52),
+ RIGHT_IRIS_CENTER: right_iris,
+ LEFT_EYE_AXIS[0]: (0.60, 0.50),
+ LEFT_EYE_AXIS[1]: (0.70, 0.50),
+ LEFT_EYE_LIDS[0]: (0.65, 0.48),
+ LEFT_EYE_LIDS[1]: (0.65, 0.52),
+ LEFT_IRIS_CENTER: left_iris,
+ }
+ )
+
+
+def sample(
+ t: float,
+ left_h: float,
+ right_h: float,
+ left_v: float = 0.5,
+ right_v: float = 0.5,
+ valid: bool = True,
+) -> GazeSample:
+ """Build one analysis-level gaze sample without going through landmark decoding."""
+ return {
+ "t": t,
+ "left": {"h": left_h, "v": left_v, "openness": 0.4},
+ "right": {"h": right_h, "v": right_v, "openness": 0.4},
+ "valid": valid,
+ }
+
+
+class GazeSampleTests(unittest.TestCase):
+ def test_iris_index_sets_are_within_the_refined_mesh(self) -> None:
+ indices = (RIGHT_IRIS_CENTER, *RIGHT_IRIS_RING, LEFT_IRIS_CENTER, *LEFT_IRIS_RING)
+ self.assertEqual(len(indices), 10)
+ for index in indices:
+ self.assertGreaterEqual(index, 468)
+ self.assertLess(index, MESH_LANDMARK_COUNT)
+
+ def test_centered_iris_yields_half_ratios(self) -> None:
+ face = face_with_gaze(right_iris=(0.35, 0.50), left_iris=(0.65, 0.50))
+ result = gaze_sample(face, 1000.0, 1000.0, 1.25)
+ self.assertAlmostEqual(result["right"]["h"], 0.5)
+ self.assertAlmostEqual(result["left"]["h"], 0.5)
+ self.assertAlmostEqual(result["right"]["v"], 0.5)
+ # Lid gap 0.04 over eye width 0.10 -> openness 0.4, comfortably above the blink floor.
+ self.assertAlmostEqual(result["right"]["openness"], 0.4)
+ self.assertTrue(result["valid"])
+ self.assertAlmostEqual(result["t"], 1.25)
+
+ def test_off_center_iris_yields_proportional_ratio(self) -> None:
+ face = face_with_gaze(right_iris=(0.375, 0.50), left_iris=(0.65, 0.50))
+ result = gaze_sample(face, 640.0, 480.0, 0.0)
+ self.assertAlmostEqual(result["right"]["h"], 0.75)
+
+ def test_head_roll_diagonal_axis_still_centers(self) -> None:
+ # Rotate the right eye 45 degrees: the projection onto its own axis keeps the centered ratio at 0.5.
+ face = landmarks_with(
+ {
+ RIGHT_EYE_AXIS[0]: (0.30, 0.30),
+ RIGHT_EYE_AXIS[1]: (0.40, 0.40),
+ RIGHT_EYE_LIDS[0]: (0.36, 0.34),
+ RIGHT_EYE_LIDS[1]: (0.34, 0.36),
+ RIGHT_IRIS_CENTER: (0.35, 0.35),
+ LEFT_EYE_AXIS[0]: (0.60, 0.50),
+ LEFT_EYE_AXIS[1]: (0.70, 0.50),
+ LEFT_EYE_LIDS[0]: (0.65, 0.48),
+ LEFT_EYE_LIDS[1]: (0.65, 0.52),
+ LEFT_IRIS_CENTER: (0.65, 0.50),
+ }
+ )
+ result = gaze_sample(face, 1000.0, 1000.0, 0.0)
+ self.assertAlmostEqual(result["right"]["h"], 0.5)
+
+ def test_closed_lids_invalidate_the_sample(self) -> None:
+ face = face_with_gaze(right_iris=(0.35, 0.50), left_iris=(0.65, 0.50))
+ for index in (*RIGHT_EYE_LIDS, *LEFT_EYE_LIDS):
+ face[index * 2] = 0.35
+ face[index * 2 + 1] = 0.50
+ result = gaze_sample(face, 1000.0, 1000.0, 0.0)
+ self.assertFalse(result["valid"])
+ self.assertAlmostEqual(result["right"]["v"], 0.5)
+
+ def test_rejects_landmarks_without_iris_points(self) -> None:
+ with self.assertRaisesRegex(ValueError, "iris landmarks"):
+ gaze_sample([0.0] * (468 * 2), 640.0, 480.0, 0.0)
+
+
+class MisalignmentTests(unittest.TestCase):
+ def test_sustained_horizontal_offset_is_detected(self) -> None:
+ samples = [sample(i * 0.05, left_h=0.50, right_h=0.35) for i in range(30)]
+ metrics = misalignment_metrics(samples)
+ self.assertEqual(metrics["status"], "ok")
+ self.assertTrue(metrics["detected"])
+ self.assertAlmostEqual(metrics["horizontal_deviation"] or 0.0, 0.15)
+
+ def test_sustained_vertical_offset_is_detected(self) -> None:
+ samples = [sample(i * 0.05, left_h=0.50, right_h=0.50, left_v=0.62, right_v=0.45) for i in range(30)]
+ metrics = misalignment_metrics(samples)
+ self.assertTrue(metrics["detected"])
+ self.assertAlmostEqual(metrics["vertical_deviation"] or 0.0, 0.17)
+
+ def test_conjugate_gaze_shift_is_not_misalignment(self) -> None:
+ # Both eyes sweep together from 0.3 to 0.7: a large gaze movement but zero left-right difference.
+ samples = [sample(i * 0.05, left_h=0.3 + i * 0.4 / 29, right_h=0.3 + i * 0.4 / 29) for i in range(30)]
+ metrics = misalignment_metrics(samples)
+ self.assertEqual(metrics["status"], "ok")
+ self.assertFalse(metrics["detected"])
+
+ def test_too_few_valid_samples_is_insufficient_data(self) -> None:
+ samples = [sample(i * 0.05, 0.5, 0.3, valid=i < 5) for i in range(MIN_ANALYSIS_SAMPLES + 5)]
+ metrics = misalignment_metrics(samples)
+ self.assertEqual(metrics["status"], "insufficient_data")
+ self.assertFalse(metrics["detected"])
+ self.assertIsNone(metrics["horizontal_deviation"])
+
+
+class OscillationTests(unittest.TestCase):
+ def test_conjugate_oscillation_is_detected(self) -> None:
+ # A 4 Hz, 0.08-amplitude horizontal oscillation of both eyes, sampled at 20 Hz for 2.4 seconds.
+ samples = []
+ for i in range(48):
+ t = i * 0.05
+ h = 0.5 + 0.08 * math.sin(2.0 * math.pi * 4.0 * t)
+ samples.append(sample(t, left_h=h, right_h=h))
+ metrics = oscillation_metrics(samples)
+ self.assertEqual(metrics["status"], "ok")
+ self.assertTrue(metrics["detected"])
+ self.assertEqual(metrics["axis"], "horizontal")
+ self.assertGreaterEqual(metrics["frequency_hz"] or 0.0, 3.0)
+ self.assertLessEqual(metrics["frequency_hz"] or 0.0, 5.0)
+ self.assertGreaterEqual(metrics["amplitude"] or 0.0, 0.04)
+
+ def test_vertical_oscillation_reports_the_vertical_axis(self) -> None:
+ samples = []
+ for i in range(48):
+ t = i * 0.05
+ v = 0.5 + 0.08 * math.sin(2.0 * math.pi * 4.0 * t)
+ samples.append(sample(t, left_h=0.5, right_h=0.5, left_v=v, right_v=v))
+ metrics = oscillation_metrics(samples)
+ self.assertTrue(metrics["detected"])
+ self.assertEqual(metrics["axis"], "vertical")
+
+ def test_steady_gaze_is_not_detected(self) -> None:
+ samples = [sample(i * 0.05, left_h=0.5, right_h=0.5) for i in range(48)]
+ metrics = oscillation_metrics(samples)
+ self.assertEqual(metrics["status"], "ok")
+ self.assertFalse(metrics["detected"])
+ self.assertAlmostEqual(metrics["amplitude"] or 0.0, 0.0)
+
+ def test_slow_smooth_pursuit_is_not_detected(self) -> None:
+ # A steady drift across the eye is pursuit / head motion: detrending leaves almost no residual.
+ samples = [sample(i * 0.05, left_h=0.4 + i * 0.2 / 47, right_h=0.4 + i * 0.2 / 47) for i in range(48)]
+ metrics = oscillation_metrics(samples)
+ self.assertEqual(metrics["status"], "ok")
+ self.assertFalse(metrics["detected"])
+
+ def test_short_window_is_insufficient_data(self) -> None:
+ samples = [sample(i * 0.05, left_h=0.5, right_h=0.5) for i in range(MIN_ANALYSIS_SAMPLES)]
+ metrics = oscillation_metrics(samples)
+ self.assertEqual(metrics["status"], "insufficient_data")
+ self.assertFalse(metrics["detected"])
+
+
+class AnalyzeWindowTests(unittest.TestCase):
+ def test_empty_window_reports_zero_counts(self) -> None:
+ analysis = analyze_window([])
+ self.assertEqual(analysis["samples"], 0)
+ self.assertEqual(analysis["valid_samples"], 0)
+ self.assertEqual(analysis["window_ms"], 0.0)
+ self.assertEqual(analysis["misalignment"]["status"], "insufficient_data")
+ self.assertEqual(analysis["oscillation"]["status"], "insufficient_data")
+
+ def test_window_counts_and_span(self) -> None:
+ samples = [sample(i * 0.05, 0.5, 0.5, valid=i % 2 == 0) for i in range(40)]
+ analysis = analyze_window(samples)
+ self.assertEqual(analysis["samples"], 40)
+ self.assertEqual(analysis["valid_samples"], 20)
+ self.assertAlmostEqual(analysis["window_ms"], 39 * 50.0, places=1)
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/services/ws-modules/pyeye1/tests/test_run_workflow.py b/services/ws-modules/pyeye1/tests/test_run_workflow.py
new file mode 100644
index 0000000..2cf054f
--- /dev/null
+++ b/services/ws-modules/pyeye1/tests/test_run_workflow.py
@@ -0,0 +1,165 @@
+import json
+import unittest
+from unittest.mock import patch
+
+from pyeye1 import eye_detection
+from pyeye1.eye_detection import (
+ EYE_MODEL_PATH,
+ VISION_BUNDLE_PATH,
+ VISION_WASM_PATH,
+ run,
+ starting_status,
+ stopped_status,
+)
+from pyeye1.gaze_analysis import (
+ LEFT_EYE_AXIS,
+ LEFT_EYE_LIDS,
+ LEFT_IRIS_CENTER,
+ MESH_LANDMARK_COUNT,
+ RIGHT_EYE_AXIS,
+ RIGHT_EYE_LIDS,
+ RIGHT_IRIS_CENTER,
+)
+
+
+def synthetic_face() -> list[float]:
+ """Build one flat 478*2 landmark list with open eyes and centered irises."""
+ values = [0.0] * (MESH_LANDMARK_COUNT * 2)
+ points = {
+ RIGHT_EYE_AXIS[0]: (0.30, 0.50),
+ RIGHT_EYE_AXIS[1]: (0.40, 0.50),
+ RIGHT_EYE_LIDS[0]: (0.35, 0.48),
+ RIGHT_EYE_LIDS[1]: (0.35, 0.52),
+ RIGHT_IRIS_CENTER: (0.35, 0.50),
+ LEFT_EYE_AXIS[0]: (0.60, 0.50),
+ LEFT_EYE_AXIS[1]: (0.70, 0.50),
+ LEFT_EYE_LIDS[0]: (0.65, 0.48),
+ LEFT_EYE_LIDS[1]: (0.65, 0.52),
+ LEFT_IRIS_CENTER: (0.65, 0.50),
+ }
+ for index, (x, y) in points.items():
+ values[index * 2] = x
+ values[index * 2 + 1] = y
+ return values
+
+
+class FakePlatform:
+ """Scripted double for the browser primitives the Python workflow drives."""
+
+ def __init__(self, stop_after: int = 5) -> None:
+ self.stop_after = stop_after
+ self.infer_calls = 0
+ self.events: list[str] = []
+ self.rendered: list[str] = []
+ self.statuses: list[str] = []
+ self.logs: list[str] = []
+ self.landmarker_args: tuple[str, str, str] | None = None
+ self.played = False
+ self.cleaned = False
+
+ async def connect_ws(self) -> None:
+ pass
+
+ def ws_state(self) -> str:
+ return "connected"
+
+ def agent_id(self) -> str:
+ return "fake-agent"
+
+ def send_event(self, message: str) -> None:
+ self.events.append(message)
+
+ async def start_camera(self) -> None:
+ pass
+
+ def video_size(self) -> list[int]:
+ return [640, 480]
+
+ async def play_video(self) -> None:
+ self.played = True
+
+ async def load_landmarker(self, model_path: str, bundle_path: str, wasm_path: str) -> None:
+ self.landmarker_args = (model_path, bundle_path, wasm_path)
+
+ async def infer(self) -> str:
+ self.infer_calls += 1
+ return json.dumps({"faces": [synthetic_face()], "width": 640, "height": 480})
+
+ def render(self, results: str) -> None:
+ self.rendered.append(results)
+
+ async def sleep(self, _ms: float) -> None:
+ pass
+
+ def log(self, message: str) -> None:
+ self.logs.append(message)
+
+ def set_status(self, status: str) -> None:
+ self.statuses.append(status)
+
+ def should_stop(self) -> bool:
+ return self.infer_calls >= self.stop_after
+
+ def cleanup(self) -> None:
+ self.cleaned = True
+
+
+class NeverConnectingPlatform(FakePlatform):
+ def ws_state(self) -> str:
+ return "connecting"
+
+
+class CameraDeniedPlatform(FakePlatform):
+ async def start_camera(self) -> None:
+ raise RuntimeError("Permission denied")
+
+
+class RunWorkflowTests(unittest.IsolatedAsyncioTestCase):
+ async def test_happy_path_drives_the_full_workflow(self) -> None:
+ platform = FakePlatform(stop_after=5)
+ # Analyze every loop iteration so the event path is exercised without waiting out real time.
+ with patch.object(eye_detection, "ANALYSIS_INTERVAL_MS", 0.0):
+ await run(platform)
+
+ self.assertEqual(platform.landmarker_args, (EYE_MODEL_PATH, VISION_BUNDLE_PATH, VISION_WASM_PATH))
+ self.assertTrue(platform.played)
+ self.assertEqual(platform.infer_calls, 5)
+ self.assertEqual(len(platform.rendered), 5)
+ self.assertEqual(platform.statuses[0], starting_status())
+ self.assertEqual(platform.statuses[-1], stopped_status())
+ self.assertTrue(any("websocket connected with agent_id=fake-agent" in line for line in platform.logs))
+ self.assertTrue(platform.cleaned)
+
+ event = json.loads(platform.events[0])
+ self.assertEqual(event["capability"], "eye_detection")
+ self.assertEqual(event["details"]["eyes"], 2)
+ self.assertIn("analysis", event["details"])
+
+ rendered = json.loads(platform.rendered[0])
+ self.assertEqual(len(rendered["faces"]), 1)
+ self.assertEqual(len(rendered["faces"][0]["irises"]), 2)
+ crop = rendered["crop"]
+ self.assertEqual(len(crop), 4)
+ self.assertLess(crop[0], crop[2])
+ self.assertLess(crop[1], crop[3])
+ self.assertGreaterEqual(crop[0], 0.0)
+ self.assertLessEqual(crop[2], 640.0)
+ self.assertLessEqual(crop[3], 480.0)
+
+ async def test_websocket_timeout_raises_and_still_cleans_up(self) -> None:
+ platform = NeverConnectingPlatform()
+ with self.assertRaisesRegex(RuntimeError, "websocket connection"):
+ await run(platform)
+ self.assertTrue(platform.cleaned)
+ self.assertEqual(platform.infer_calls, 0)
+
+ async def test_camera_failure_propagates_and_still_cleans_up(self) -> None:
+ platform = CameraDeniedPlatform()
+ with self.assertRaisesRegex(RuntimeError, "Permission denied"):
+ await run(platform)
+ self.assertTrue(platform.cleaned)
+ self.assertEqual(platform.infer_calls, 0)
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/services/ws-server/static/app.js b/services/ws-server/static/app.js
index 3b01559..f6f06d5 100644
--- a/services/ws-server/static/app.js
+++ b/services/ws-server/static/app.js
@@ -26,6 +26,8 @@ const append = (line) => {
const describeError = (error) => (error instanceof Error ? error.message : String(error));
const WORKFLOW_MODULES = new Map();
+// Preselected in the dropdown when the server's module list includes it; otherwise the first option stays.
+const DEFAULT_MODULE = "et-ws-pyeye1";
const populateModuleDropdown = async () => {
append("Discovering modules via /modules...");
@@ -81,6 +83,8 @@ const populateModuleDropdown = async () => {
console.error(`discovery error for ${name}:`, error);
}
}
+
+ if (WORKFLOW_MODULES.has(DEFAULT_MODULE)) moduleSelect.value = DEFAULT_MODULE;
};
const updateAgentCard = (status, agentId = currentAgentId) => {
diff --git a/services/ws-server/static/index.html b/services/ws-server/static/index.html
index a506b5f..350f47d 100644
--- a/services/ws-server/static/index.html
+++ b/services/ws-server/static/index.html
@@ -9,25 +9,18 @@
-
WASM web agent
-
- The page imports the generated module from /pkg,
- initializes the WASM runtime, and connects to /ws.
-
-
- Open the browser devtools as well if you want to inspect the module logs
- from tracing-wasm.
-
-
- Web Agent Identity
-
Waiting for websocket handshake...
- unassigned
-
+
+ Medical disclaimer: these modules are tech demos. Any
+ health-related output, such as the eye-movement screening indicators, is
+ a heuristic signal from webcam landmarks -- not a medical device, a
+ diagnosis, or clinical advice. Consult a qualified clinician about any
+ health concern.
+
@@ -38,6 +31,15 @@
WASM web agent
+
+ Web Agent Identity
+ Waiting for websocket handshake...
+ unassigned
+
+
+ Open the browser devtools as well if you want to inspect the module logs
+ from tracing-wasm.
+