""" AOI (Area-of-Interest) engine — Phase 4, the gaze analytics iMotions is known for. Given a recorded gaze stream and a set of AOIs (rectangles/polygons in the same normalized [0,1] screen coords the eye tracker reports), this computes the standard eye-tracking metrics, per AOI and per stimulus: * fixations — I-DT dispersion-threshold fixation detection, * dwell time — total time gaze fell inside the AOI, * fixation count, * TTFF — time to first fixation (from stimulus onset), * hit ratio — fraction of participants/trials that fixated the AOI (here: whether the AOI was entered at all in the window), * revisits. AOIs can be authored by hand, or generated automatically from the stimulus image via YOLO/SAM (`auto_aoi`, lazy-imported) — the open replacement for iMotions' Automated AOI module. Pure geometry + numpy; no hardware. Works on any session from `analysis.load`. """ from __future__ import annotations from dataclasses import dataclass, field import numpy as np from . import load as A # -------------------------------------------------------------------------- # AOI geometry # -------------------------------------------------------------------------- @dataclass class AOI: name: str # axis-aligned rectangle in normalized coords; or a polygon (list of (x,y)). rect: tuple | None = None # (x0, y0, x1, y1) polygon: list | None = None # [(x,y), ...] def contains(self, x: float, y: float) -> bool: if x != x or y != y: # NaN gaze (blink / off-screen) return False if self.rect is not None: x0, y0, x1, y1 = self.rect return x0 <= x <= x1 and y0 <= y <= y1 if self.polygon is not None: return _point_in_poly(x, y, self.polygon) return False def _point_in_poly(x, y, poly) -> bool: inside = False n = len(poly) j = n - 1 for i in range(n): xi, yi = poly[i]; xj, yj = poly[j] if ((yi > y) != (yj > y)) and \ (x < (xj - xi) * (y - yi) / ((yj - yi) or 1e-12) + xi): inside = not inside j = i return inside # -------------------------------------------------------------------------- # fixation detection (I-DT: dispersion threshold) # -------------------------------------------------------------------------- @dataclass class Fixation: t_start: float t_end: float x: float y: float @property def duration(self) -> float: return self.t_end - self.t_start def detect_fixations(t, x, y, dispersion=0.05, min_duration=0.1) -> list[Fixation]: """I-DT fixation detection. `dispersion` in normalized units, `min_duration` s.""" t = np.asarray(t, float); x = np.asarray(x, float); y = np.asarray(y, float) fixations: list[Fixation] = [] n = len(t) i = 0 while i < n: j = i + 1 while j <= n: xs, ys = x[i:j], y[i:j] if np.any(np.isnan(xs)): break disp = (np.nanmax(xs) - np.nanmin(xs)) + (np.nanmax(ys) - np.nanmin(ys)) if disp > dispersion: break j += 1 # window [i, j-1] is a candidate fixation if (j - 1) > i and (t[j-2] - t[i]) >= min_duration: sl = slice(i, j-1) fixations.append(Fixation(float(t[i]), float(t[j-2]), float(np.nanmean(x[sl])), float(np.nanmean(y[sl])))) i = j - 1 else: i += 1 return fixations # -------------------------------------------------------------------------- # metrics # -------------------------------------------------------------------------- def aoi_metrics(session: dict, aois: list[AOI], *, gaze_stream="EyeTracker", onset: float | None = None, window: float | None = None, gaze_xy=(0, 1), **fix_kw) -> dict: """ Compute per-AOI metrics over a time window. If `onset` is given, the window is [onset, onset+window] and TTFF is measured from `onset`; otherwise the whole recording is used. """ s = session["streams"][gaze_stream] t = np.asarray(s["t"], float) X = np.asarray(s["x"], float) gx, gy = X[:, gaze_xy[0]], X[:, gaze_xy[1]] if onset is not None: hi = onset + (window if window is not None else (t[-1] - onset)) m = (t >= onset) & (t <= hi) t, gx, gy = t[m], gx[m], gy[m] fixations = detect_fixations(t, gx, gy, **fix_kw) dt = np.gradient(t) if len(t) > 1 else np.array([0.0]) out = {} for aoi in aois: inside_samples = np.array([aoi.contains(px, py) for px, py in zip(gx, gy)]) dwell = float(np.sum(dt[inside_samples])) if inside_samples.any() else 0.0 in_fix = [f for f in fixations if aoi.contains(f.x, f.y)] ttff = None if in_fix: ref = onset if onset is not None else (t[0] if len(t) else 0.0) ttff = float(min(f.t_start for f in in_fix) - ref) out[aoi.name] = { "dwell_time_s": dwell, "fixation_count": len(in_fix), "ttff_s": ttff, "hit": bool(inside_samples.any()), "revisits": _revisits(inside_samples), "mean_fix_duration_s": float(np.mean([f.duration for f in in_fix])) if in_fix else 0.0, } return out def per_stimulus(session: dict, aois: list[AOI], *, window=2.0, **kw) -> dict: """AOI metrics around every stimulus onset marker, keyed by stimulus label.""" results = {} for t, label in A.markers(session): if not str(label).startswith("stim_on"): continue name = str(label).split(":", 1)[1] if ":" in str(label) else str(label) results.setdefault(name, []).append( aoi_metrics(session, aois, onset=float(t), window=window, **kw)) return results # -------------------------------------------------------------------------- # automated AOIs (open replacement for iMotions Automated AOI) # -------------------------------------------------------------------------- def auto_aoi(image_path: str, *, model="yolov8n.pt", conf=0.3) -> list[AOI]: """ Detect objects in a stimulus image and return one AOI per detection (normalized rects). Uses Ultralytics YOLO; lazy-imported. For pixel-accurate masks, swap in SAM and convert masks to polygons. """ from ultralytics import YOLO # lazy import cv2 img = cv2.imread(image_path) h, w = img.shape[:2] res = YOLO(model)(image_path, conf=conf, verbose=False)[0] aois = [] for i, box in enumerate(res.boxes): x0, y0, x1, y1 = box.xyxy[0].tolist() cls = res.names[int(box.cls[0])] aois.append(AOI(name=f"{cls}_{i}", rect=(x0 / w, y0 / h, x1 / w, y1 / h))) return aois def _revisits(inside: np.ndarray) -> int: """Number of separate entries into the AOI (rising edges) minus the first.""" if not inside.any(): return 0 edges = np.diff(inside.astype(int)) entries = int(np.sum(edges == 1)) + (1 if inside[0] else 0) return max(0, entries - 1)