Coverage for scanpath_studio/similarity.py: 97%
113 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-07 21:10 +0000
1"""Scanpath similarity metrics for comparing scanpaths over the same text.
3The reference use case is the **Multiple Comparison** tab: score how close each
4model-generated scanpath is to the real reading of the same text.
6Currently one metric is implemented for real:
8- **NLD** — Normalized Levenshtein Distance on the area-of-interest (word-ID)
9 sequence. The underlying edit distance is Levenshtein's (Levenshtein, 1966,
10 *Binary codes capable of correcting deletions, insertions, and reversals*,
11 Soviet Physics Doklady 10:707–710): the minimal number of insertions, deletions
12 and substitutions to turn one word-index sequence into the other. Each scanpath
13 is reduced to the ordered list of word indices its fixations land on (each
14 fixation's own recorded word/AOI label when present, otherwise the word box
15 it falls in); NLD divides that edit
16 distance by the longer sequence's length, so it sits in ``[0, 1]`` (0 =
17 identical, 1 = maximally different). Lower is better. Eyettention (Deng et al.,
18 2023) uses this same ``NLD = LD / max(|S|, |T|)`` as its scanpath-similarity
19 metric and likewise cites Levenshtein (1966) for the edit distance.
21Three further metrics are registered as **labeled placeholders** (``fn=None``)
22so the comparison table is laid out and ready for ScanMatch / MultiMatch /
23Scasim (etc.) to be dropped in later — they currently render as "—".
25The module is deliberately pure (no Streamlit, no plotting) so the metrics can
26be unit-tested against the hand-traced :mod:`scanpath_studio.synthetic` trial.
27"""
29from __future__ import annotations
31from collections.abc import Callable, Sequence
32from dataclasses import dataclass
34import numpy as np
35import pandas as pd
37from .measures import (
38 _assign_word_ids_single,
39 rebased_fixation_onsets,
40)
42# -----------------------------------------------------------------------------
43# Edit distance
44# -----------------------------------------------------------------------------
47def levenshtein(a: Sequence, b: Sequence) -> int:
48 """Levenshtein (edit) distance between two sequences of hashable items.
50 Classic two-row dynamic program, O(len(a) * len(b)) time and O(len(b))
51 space. Scanpaths are tens-to-low-hundreds of fixations long, so this is
52 plenty fast.
53 """
54 n, m = len(a), len(b)
55 if n == 0:
56 return m
57 if m == 0:
58 return n
59 prev = list(range(m + 1))
60 cur = [0] * (m + 1)
61 for i in range(1, n + 1):
62 cur[0] = i
63 ai = a[i - 1]
64 for j in range(1, m + 1):
65 cost = 0 if ai == b[j - 1] else 1
66 cur[j] = min(prev[j] + 1, cur[j - 1] + 1, prev[j - 1] + cost)
67 prev, cur = cur, prev
68 return prev[m]
71def normalized_levenshtein(a: Sequence, b: Sequence) -> float:
72 """NLD in ``[0, 1]``: edit distance divided by the longer sequence's length.
74 Two empty sequences are treated as identical (0.0); one empty and one
75 non-empty are maximally different (1.0, since the distance equals the
76 non-empty length).
77 """
78 denom = max(len(a), len(b))
79 if denom == 0:
80 return 0.0
81 return levenshtein(a, b) / float(denom)
84# -----------------------------------------------------------------------------
85# Fixation -> word (AOI) sequence
86# -----------------------------------------------------------------------------
89def assign_single_trial_word_ids(
90 fixations: pd.DataFrame,
91 words: pd.DataFrame,
92) -> np.ndarray:
93 """Word id for every fixation via bounding-box containment, single trial.
95 Thin wrapper over :func:`scanpath_studio.measures._assign_word_ids_single`
96 that does **not** group by ``participant_id`` / ``trial_id`` — the generated
97 model scanpaths carry synthetic ids (e.g. ``"Model 1"``) that don't match the
98 real trial's, so the grouped :func:`measures.assign_fixations_to_words` would
99 find no matching word boxes and return all-NaN. Here every fixation is tested
100 against the one ``words`` frame passed in. A fixation outside every box gets
101 NaN.
103 Returns a float array aligned to ``fixations`` rows (NaN = out of text).
104 """
105 if fixations.empty or words.empty:
106 return np.full(len(fixations), np.nan)
107 return _assign_word_ids_single(fixations, words)
110def _ordered_fixations(fixations: pd.DataFrame) -> pd.DataFrame:
111 """Fixations in reading order (by ``timestamp_ms`` when available)."""
112 if "timestamp_ms" in fixations.columns and not fixations.empty:
113 return fixations.sort_values("timestamp_ms")
114 return fixations
117def ordered_word_ids(
118 fixations: pd.DataFrame,
119 words: pd.DataFrame,
120) -> np.ndarray:
121 """Per-fixation word id in reading order; NaN where out of text.
123 When the fixations carry word ids (the corpus AOI label, or — for generated
124 scanpaths — the word each was drawn over) they are used as given, blanks
125 included; only fixations with none at all are mapped geometrically via
126 :func:`assign_single_trial_word_ids`.
127 """
128 ordered = _ordered_fixations(fixations)
129 if ordered.empty or words.empty:
130 return np.array([], dtype=float)
131 if "word_id" in ordered.columns:
132 existing = pd.to_numeric(ordered["word_id"], errors="coerce").to_numpy(
133 dtype=float
134 )
135 if not np.isnan(existing).all():
136 return existing
137 return assign_single_trial_word_ids(ordered, words)
140def aoi_sequence(
141 fixations: pd.DataFrame,
142 words: pd.DataFrame,
143) -> list[int]:
144 """Temporal sequence of fixated word ids (out-of-text fixations dropped).
146 Fixations are read in ``timestamp_ms`` order when that column is present
147 (it always is after normalization), so the sequence reflects reading order.
148 """
149 word_ids = ordered_word_ids(fixations, words)
150 return [int(w) for w in word_ids if pd.notna(w)]
153# -----------------------------------------------------------------------------
154# Metric registry
155# -----------------------------------------------------------------------------
158@dataclass(frozen=True)
159class ScanpathMetric:
160 """One column of the similarity table.
162 ``fn`` maps ``(reference_ctx, generated_ctx)`` to a float; both contexts are
163 the dicts built by :func:`_context` (precomputed AOI sequence + the raw
164 frames). ``fn=None`` marks a placeholder — the column is shown but no value
165 is computed yet.
166 """
168 key: str
169 label: str
170 description: str
171 lower_is_better: bool
172 fn: Callable[[dict, dict], float] | None
175def _nld_metric(reference: dict, generated: dict) -> float:
176 return normalized_levenshtein(reference["aoi"], generated["aoi"])
179# Metric 1 is real (NLD); 2-4 are placeholders with the standard scanpath
180# metrics they're earmarked for, so plugging in a real implementation is just
181# swapping ``fn=None`` for the function.
182METRICS: list[ScanpathMetric] = [
183 ScanpathMetric(
184 key="nld",
185 label="NLD",
186 description=(
187 "Normalized Levenshtein Distance on the word-index (AOI) sequence — "
188 "the Levenshtein (1966) edit distance between the two fixated-word "
189 "sequences, normalized by the longer one's length. Used as the "
190 "scanpath-similarity metric by Eyettention (Deng et al. 2023). "
191 "0 = identical sequence, 1 = maximally different. Lower is better."
192 ),
193 lower_is_better=True,
194 fn=_nld_metric,
195 ),
196 ScanpathMetric(
197 key="scanmatch",
198 label="ScanMatch",
199 description=(
200 "Placeholder — ScanMatch (Cristino et al. 2010): Needleman–Wunsch "
201 "alignment with a spatial substitution matrix. Not yet computed."
202 ),
203 lower_is_better=False,
204 fn=None,
205 ),
206 ScanpathMetric(
207 key="multimatch",
208 label="MultiMatch",
209 description=(
210 "Placeholder — MultiMatch (Dewhurst et al. 2012): vector, direction, "
211 "length, position and duration sub-scores. Not yet computed."
212 ),
213 lower_is_better=False,
214 fn=None,
215 ),
216 ScanpathMetric(
217 key="scasim",
218 label="Scasim",
219 description=(
220 "Placeholder — Scasim (von der Malsburg & Vasishth 2011): a "
221 "saccade-sensitive, duration-aware similarity. Not yet computed."
222 ),
223 lower_is_better=True,
224 fn=None,
225 ),
226]
229def _context(fixations: pd.DataFrame, words: pd.DataFrame) -> dict:
230 """Precompute the per-scanpath inputs the metric functions share."""
231 return {
232 "fix": fixations,
233 "words": words,
234 "aoi": aoi_sequence(fixations, words),
235 }
238def compute_similarity_table(
239 reference_fixations: pd.DataFrame,
240 model_fixations: dict[str, pd.DataFrame],
241 words: pd.DataFrame,
242) -> pd.DataFrame:
243 """Score every model scanpath against the reference, one row per model.
245 Args:
246 reference_fixations: the real scanpath's fixations (single trial).
247 model_fixations: ordered ``{model_name: fixations_df}``.
248 words: the shared word boxes (the real trial's words).
250 Returns:
251 DataFrame with a ``Model`` column plus one column per metric label.
252 Placeholder metrics are filled with ``NaN``; a metric that raises is
253 also recorded as ``NaN`` rather than aborting the whole table.
254 """
255 reference_ctx = _context(reference_fixations, words)
256 rows = []
257 for name, fix in model_fixations.items():
258 generated_ctx = _context(fix, words)
259 row: dict[str, object] = {"Model": name}
260 for metric in METRICS:
261 if metric.fn is None:
262 row[metric.label] = np.nan
263 continue
264 try:
265 row[metric.label] = float(metric.fn(reference_ctx, generated_ctx))
266 except Exception:
267 row[metric.label] = np.nan
268 rows.append(row)
269 columns = ["Model"] + [m.label for m in METRICS]
270 return pd.DataFrame(rows, columns=columns)
273# -----------------------------------------------------------------------------
274# Cumulative metric curves (for the Multiple Comparison convergence plots)
275# -----------------------------------------------------------------------------
278def _eval_indices(n: int, max_points: int) -> list[int]:
279 """Up to ``max_points`` prefix lengths in 1..n (all of them when n is small),
280 always including 1 and n. Keeps the convergence sweep fast on long trials."""
281 if n <= 0:
282 return []
283 if n <= max_points:
284 return list(range(1, n + 1))
285 return sorted({round(v) for v in np.linspace(1, n, max_points)})
288def nld_by_fixation_index(
289 reference_fixations: pd.DataFrame,
290 model_fixations: pd.DataFrame,
291 words: pd.DataFrame,
292 *,
293 max_points: int = 80,
294) -> tuple[list[int], list[float]]:
295 """NLD between the first-*k*-fixations prefixes of the two scanpaths, vs k.
297 For each prefix length k the reference and model are each truncated to their
298 first k fixations (reading order) and scored. Returns ``(ks, nlds)``; for
299 long scanpaths the k axis is subsampled to ``max_points`` evenly spaced
300 values (kept exhaustive when the scanpath is short).
301 """
302 ref_w = ordered_word_ids(reference_fixations, words)
303 mod_w = ordered_word_ids(model_fixations, words)
304 n = max(len(ref_w), len(mod_w))
305 ks = _eval_indices(n, max_points)
306 nlds = []
307 for k in ks:
308 ref_seq = [int(w) for w in ref_w[:k] if pd.notna(w)]
309 mod_seq = [int(w) for w in mod_w[:k] if pd.notna(w)]
310 nlds.append(normalized_levenshtein(ref_seq, mod_seq))
311 return ks, nlds
314def _rebased_onsets(fixations: pd.DataFrame) -> np.ndarray:
315 """Fixation onset times (ms) rebased so the first fixation is t=0.
317 Orders the fixations by ``timestamp_ms`` and delegates the recorded-vs-
318 synthetic-timestamp heuristic to
319 :func:`scanpath_studio.measures.rebased_fixation_onsets` (shared with the
320 animation clock).
321 """
322 return rebased_fixation_onsets(_ordered_fixations(fixations))
325def nld_by_time(
326 reference_fixations: pd.DataFrame,
327 model_fixations: pd.DataFrame,
328 words: pd.DataFrame,
329 *,
330 max_points: int = 80,
331) -> tuple[list[float], list[float]]:
332 """NLD between the prefixes up to elapsed reading time t, vs t in seconds.
334 Both scanpaths are rebased to their first fixation; at each sample time the
335 reference and model are truncated to the fixations whose onset ≤ t and
336 scored. Sample times are the union of both scanpaths' fixation onsets
337 (subsampled to ``max_points`` for long readings). Returns ``(t_seconds,
338 nlds)``.
339 """
340 # ordered_word_ids and _rebased_onsets each order by timestamp_ms internally
341 # (stable sort), so their outputs stay row-aligned without a pre-sort here.
342 ref_w = ordered_word_ids(reference_fixations, words)
343 mod_w = ordered_word_ids(model_fixations, words)
344 ref_on = _rebased_onsets(reference_fixations)
345 mod_on = _rebased_onsets(model_fixations)
346 times = sorted({float(t) for t in (*ref_on, *mod_on)})
347 if not times:
348 return [], []
349 if len(times) > max_points:
350 picks = np.linspace(0, len(times) - 1, max_points)
351 times = sorted({times[round(i)] for i in picks})
352 xs, nlds = [], []
353 for t in times:
354 ref_seq = [int(w) for w, o in zip(ref_w, ref_on) if o <= t and pd.notna(w)]
355 mod_seq = [int(w) for w, o in zip(mod_w, mod_on) if o <= t and pd.notna(w)]
356 xs.append(t / 1000.0)
357 nlds.append(normalized_levenshtein(ref_seq, mod_seq))
358 return xs, nlds