From 0d59ade23c436f12f27f9b5af9e6fccdb753dc11 Mon Sep 17 00:00:00 2001 From: John Vandenberg Date: Wed, 22 Jul 2026 10:04:28 +0800 Subject: [PATCH] Improve eye tracking --- .../ws-modules/pyeye1/pkg/et_ws_pyeye1.js | 176 ++++++---- services/ws-modules/pyeye1/pyeye1/__init__.py | 11 +- .../ws-modules/pyeye1/pyeye1/eye_detection.py | 324 ++++++++++++++---- .../ws-modules/pyeye1/pyeye1/gaze_analysis.py | 253 ++++++++++++++ services/ws-modules/pyeye1/pyproject.toml | 2 +- .../pyeye1/tests/test_eye_detection.py | 64 ++++ .../pyeye1/tests/test_gaze_analysis.py | 218 ++++++++++++ .../pyeye1/tests/test_run_workflow.py | 165 +++++++++ services/ws-server/static/app.js | 4 + services/ws-server/static/index.html | 30 +- services/ws-server/static/style.css | 35 +- 11 files changed, 1114 insertions(+), 168 deletions(-) create mode 100644 services/ws-modules/pyeye1/pyeye1/gaze_analysis.py create mode 100644 services/ws-modules/pyeye1/tests/test_gaze_analysis.py create mode 100644 services/ws-modules/pyeye1/tests/test_run_workflow.py 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. + 

Booting...
diff --git a/services/ws-server/static/style.css b/services/ws-server/static/style.css index 24c0576..b5c706d 100644 --- a/services/ws-server/static/style.css +++ b/services/ws-server/static/style.css @@ -28,34 +28,41 @@ main { backdrop-filter: blur(10px); } -h1 { - margin: 0 0 8px; - font-size: clamp(2rem, 5vw, 3.4rem); - line-height: 0.95; -} - p { margin: 0 0 12px; max-width: 65ch; } .agent-card { - margin: 0 0 18px; - padding: 14px 16px; + display: flex; + flex-wrap: wrap; + gap: 2px 12px; + align-items: baseline; + margin: 18px 0 0; + padding: 4px 10px; border: 1px solid rgba(24, 32, 40, 0.14); background: rgba(255, 253, 250, 0.92); -} - -.agent-card strong { - display: block; - margin-bottom: 6px; + font-size: 0.78rem; } .agent-card code { - display: block; overflow-wrap: anywhere; } +.log-note { + margin: 12px 0 0; + font-size: 0.78rem; +} + +.medical-note { + margin: 0 0 12px; + padding: 6px 10px; + border: 1px solid rgba(213, 155, 66, 0.55); + background: rgba(213, 155, 66, 0.12); + font-size: 0.78rem; + max-width: none; +} + code { font-size: 0.95em; }