Coverage for scanpath_studio/model_scanpaths.py: 97%

1"""Synthetic "model-generated" scanpaths over a real text's word boxes. 

2 

3This is **placeholder data** for the *Multiple Comparison* tab. The intended 

4end state is: for a given participant reading a given text, several scanpath 

5*models* each predict a scanpath, and the tab shows the real reading alongside 

6each model's prediction plus similarity scores. Until those real model outputs 

7are connected, this module fabricates a reading-like scanpath per "model" so the 

8whole tab — grid layout, similarity table — can be built and demoed. 

9 

10Design goals: 

11 

12- **Reading-like, not uniform noise.** Each scanpath is a left-to-right walk 

13 over the word boxes with model-specific skip / regression / refixation rates, 

14 so the panels actually look like scanpaths and their NLD vs. the real reading 

15 spreads out across models. 

16- **Reproducible.** A scanpath is deterministic in ``(trial_id, model_index, 

17 nonce)``. Streamlit reruns the whole script on every interaction, so a 

18 non-seeded generator would reshuffle every panel on each click. The tab's 

19 "Regenerate" button bumps the ``nonce`` to deliberately re-draw. 

20- **Canonical output.** Frames carry the same columns 

21 :func:`scanpath_studio.data.normalize_fixations` produces, so they drop 

22 straight into :func:`scanpath_studio.plots.make_scanpath_figure` and the 

23 similarity metrics without special-casing. 

24""" 

25 

26from __future__ import annotations 

27 

28import hashlib 

29from dataclasses import dataclass 

30 

31import numpy as np 

32import pandas as pd 

33 

34from .measures import cluster_word_lines, word_box_bounds 

35 

36 

37@dataclass(frozen=True) 

38class ModelProfile: 

39 """Behaviour knobs for one synthetic model. 

40 

41 ``skip_prob`` / ``regress_prob`` / ``refix_prob`` are per-step probabilities 

42 of skipping the next word, regressing to an earlier word, or refixating the 

43 current word. ``mean_dur_ms`` / ``dur_sd_ms`` shape fixation durations; 

44 ``jitter_frac`` is the Gaussian landing-position jitter as a fraction of the 

45 word-box size. 

46 """ 

47 

48 name: str 

49 skip_prob: float 

50 regress_prob: float 

51 refix_prob: float 

52 mean_dur_ms: float 

53 dur_sd_ms: float 

54 jitter_frac: float 

55 

56 

57# Ordered from most reading-like (Model 1: few skips/regressions, durations near 

58# typical reading) to progressively noisier. The tab slices this to the 

59# requested model count. Tuned so the panels look visibly different and the NLD 

60# vs. the real scanpath spreads across a useful range. 

61MODEL_PROFILES = [ 

62 ModelProfile("Model 1", 0.08, 0.06, 0.10, 230, 60, 0.12), 

63 ModelProfile("Model 2", 0.15, 0.10, 0.12, 215, 70, 0.16), 

64 ModelProfile("Model 3", 0.22, 0.14, 0.10, 250, 80, 0.20), 

65 ModelProfile("Model 4", 0.30, 0.05, 0.18, 200, 65, 0.18), 

66 ModelProfile("Model 5", 0.18, 0.22, 0.14, 240, 90, 0.24), 

67 ModelProfile("Model 6", 0.35, 0.18, 0.20, 190, 75, 0.26), 

68 ModelProfile("Model 7", 0.12, 0.30, 0.16, 260, 85, 0.22), 

69 ModelProfile("Model 8", 0.45, 0.25, 0.22, 175, 70, 0.30), 

70] 

71 

72MAX_MODELS = len(MODEL_PROFILES) 

73DEFAULT_N_MODELS = 6 

74 

75# Inter-fixation gap folded into timestamps, approximating saccade + planning 

76# time so the synthetic real-time clock reads plausibly in the animation/metrics. 

77SACCADE_GAP_MS = 25 

78 

79# Fixation durations are clamped to a plausible reading band (ms). 

80_MIN_DUR_MS = 50 

81_MAX_DUR_MS = 800 

82 

83# Canonical fixation columns produced by data.normalize_fixations that the 

84# figure builder and metrics rely on. Kept here so an empty frame still has the 

85# right shape. 

86_FIX_COLUMNS = [ 

87 "participant_id", 

88 "trial_id", 

89 "text_id", 

90 "x", 

91 "y", 

92 "duration_ms", 

93 "timestamp_ms", 

94 "fixation_id", 

95 "word_id", 

96 "pass_index", 

97 "order_in_trial", 

98 "eye", 

99 "saccade_type", 

100] 

101 

102 

103def _seed(trial_id: object, model_index: int, nonce: int) -> int: 

104 """Stable 32-bit seed from (trial, model, nonce). 

105 

106 Uses md5 rather than the builtin ``hash`` because ``hash`` of a str is salted 

107 per process, which would make scanpaths non-reproducible across reruns. 

108 """ 

109 raw = f"{trial_id}|{model_index}|{nonce}".encode() 

110 return int(hashlib.md5(raw).hexdigest()[:8], 16) 

111 

112 

113def _empty_fix_frame() -> pd.DataFrame: 

114 return pd.DataFrame({col: [] for col in _FIX_COLUMNS}) 

115 

116 

117def _ordered_word_rows(words: pd.DataFrame) -> pd.DataFrame: 

118 """Words in reading order: by ``word_id`` when usable, else line-then-x. 

119 

120 ``word_id`` is reading order in OneStop. When it's missing or has gaps we 

121 fall back to clustering rows into visual lines by ``y`` (tolerance ~half a 

122 word height) and reading each line left-to-right. 

123 """ 

124 if "word_id" in words.columns and words["word_id"].notna().all(): 

125 return words.sort_values("word_id").reset_index(drop=True) 

126 # Cluster rows into visual lines by vertical position (shared with the 

127 # reading-measure geometry), then read each line left-to-right. NaN-safe 

128 # line pitch and the all-NaN-height guard live in ``cluster_word_lines``. 

129 w = words.copy() 

130 w["_line"] = cluster_word_lines(w) 

131 return w.sort_values(["_line", "x"]).drop(columns="_line").reset_index(drop=True) 

132 

133 

134def _walk_word_indices(n_words: int, profile: ModelProfile, rng) -> list[int]: 

135 """Random reading-like walk producing a list of fixated word positions. 

136 

137 Positions index into the reading-ordered word list (0..n_words-1). Advances 

138 one word by default; with the profile's probabilities it skips ahead two, 

139 regresses one or two words, or refixates the current word. Bounded by a cap 

140 so heavy refixation/regression can't loop forever. 

141 """ 

142 if n_words <= 0: 

143 return [0] 

144 seq: list[int] = [] 

145 pos = 0 

146 cap = max(8, n_words * 3) 

147 consecutive_refix = 0 

148 while pos < n_words and len(seq) < cap: 

149 seq.append(pos) 

150 # Refixation: stay on the same word (bounded run so we always progress). 

151 if rng.random() < profile.refix_prob and consecutive_refix < 2: 

152 consecutive_refix += 1 

153 continue 

154 consecutive_refix = 0 

155 roll = rng.random() 

156 if roll < profile.regress_prob and pos > 0: 

157 pos = max(0, pos - int(rng.integers(1, 3))) 

158 elif roll < profile.regress_prob + profile.skip_prob: 

159 pos += 2 

160 else: 

161 pos += 1 

162 return seq or [0] 

163 

164 

165def generate_model_scanpath( 

166 words: pd.DataFrame, 

167 profile: ModelProfile, 

168 *, 

169 model_index: int, 

170 reference_trial_id: object, 

171 text_id: object | None = None, 

172 nonce: int = 0, 

173) -> pd.DataFrame: 

174 """One model's synthetic scanpath over ``words``, in canonical fixation form. 

175 

176 Deterministic in ``(reference_trial_id, model_index, nonce)``. 

177 """ 

178 if words is None or words.empty: 

179 return _empty_fix_frame() 

180 

181 rng = np.random.default_rng(_seed(reference_trial_id, model_index, nonce)) 

182 ordered = _ordered_word_rows(words) 

183 # Drop words whose box geometry failed to export (NaN x/y/width/height) so 

184 # generated fixations always land at finite coordinates — np.maximum(nan, 1) 

185 # is nan, so the jitter floor wouldn't otherwise sanitize them. 

186 finite_geom = ( 

187 pd.to_numeric(ordered["x"], errors="coerce").notna() 

188 & pd.to_numeric(ordered["y"], errors="coerce").notna() 

189 & pd.to_numeric(ordered["width"], errors="coerce").notna() 

190 & pd.to_numeric(ordered["height"], errors="coerce").notna() 

191 ) 

192 ordered = ordered[finite_geom.to_numpy()].reset_index(drop=True) 

193 if ordered.empty: 

194 return _empty_fix_frame() 

195 n_words = len(ordered) 

196 

197 # Generate inside the experiment's own boxes (BUG-83), so a model fixation 

198 # is assigned to the word it was generated for. 

199 x_box, y_box, x_right, y_bottom = word_box_bounds(ordered) 

200 w_box = x_right - x_box 

201 h_box = y_bottom - y_box 

202 if "text" in ordered.columns: 

203 text_len = ordered["text"].astype(str).str.len().to_numpy(dtype=float) 

204 else: 

205 text_len = np.full(n_words, 5.0) 

206 word_ids = ( 

207 ordered["word_id"].to_numpy() 

208 if "word_id" in ordered.columns 

209 else np.full(n_words, np.nan) 

210 ) 

211 

212 seq = _walk_word_indices(n_words, profile, rng) 

213 seq_idx = np.asarray(seq, dtype=int) 

214 k = len(seq_idx) 

215 

216 # Landing position: word-box centre + Gaussian jitter. Vertical jitter is 

217 # scaled down (box height encodes the line pitch, ~3 text lines in OneStop) 

218 # so a fixation stays near its own line rather than drifting onto neighbours. 

219 cx = x_box[seq_idx] + w_box[seq_idx] / 2.0 

220 cy = y_box[seq_idx] + h_box[seq_idx] / 2.0 

221 jitter_x = rng.normal(0.0, profile.jitter_frac * np.maximum(w_box[seq_idx], 1.0)) 

222 jitter_y = rng.normal( 

223 0.0, profile.jitter_frac * np.maximum(h_box[seq_idx], 1.0) / 3.0 

224 ) 

225 fix_x = cx + jitter_x 

226 fix_y = cy + jitter_y 

227 

228 # Duration: profile mean ± sd, nudged longer for longer words, then clamped. 

229 mean_len = float(text_len.mean()) if text_len.size else 5.0 

230 durations = rng.normal(profile.mean_dur_ms, profile.dur_sd_ms, size=k) 

231 durations = durations + (text_len[seq_idx] - mean_len) * 4.0 

232 durations = np.clip(durations, _MIN_DUR_MS, _MAX_DUR_MS).round().astype(int) 

233 

234 # Cumulative timestamps with a saccade gap between fixations. 

235 timestamps = np.empty(k, dtype=int) 

236 acc = 0 

237 for i in range(k): 

238 timestamps[i] = acc 

239 acc += int(durations[i]) + SACCADE_GAP_MS 

240 

241 para = text_id if text_id is not None else reference_trial_id 

242 return pd.DataFrame( 

243 { 

244 "participant_id": [profile.name] * k, 

245 "trial_id": [str(reference_trial_id)] * k, 

246 "text_id": [str(para)] * k, 

247 "x": fix_x, 

248 "y": fix_y, 

249 "duration_ms": durations, 

250 "timestamp_ms": timestamps, 

251 "fixation_id": np.arange(1, k + 1), 

252 "word_id": word_ids[seq_idx], 

253 "pass_index": [1] * k, 

254 "order_in_trial": np.arange(1, k + 1), 

255 "eye": ["Both"] * k, 

256 "saccade_type": ["unknown"] * k, 

257 } 

258 ) 

259 

260 

261def generate_model_scanpaths( 

262 words: pd.DataFrame, 

263 *, 

264 n_models: int = DEFAULT_N_MODELS, 

265 reference_trial_id: object, 

266 text_id: object | None = None, 

267 nonce: int = 0, 

268) -> dict[str, pd.DataFrame]: 

269 """Generate ``n_models`` synthetic scanpaths over ``words``. 

270 

271 Returns an insertion-ordered ``{model_name: fixations_df}`` (Python dicts 

272 preserve order), so the tab grid and the similarity table list models in the 

273 same Model 1..N order. 

274 """ 

275 count = max(1, min(int(n_models), MAX_MODELS)) 

276 profiles = MODEL_PROFILES[:count] 

277 return { 

278 profile.name: generate_model_scanpath( 

279 words, 

280 profile, 

281 model_index=index, 

282 reference_trial_id=reference_trial_id, 

283 text_id=text_id, 

284 nonce=nonce, 

285 ) 

286 for index, profile in enumerate(profiles) 

287 }