""" analysis.py Turns a raw InsightUX session (gaze_log.jsonl + dom_log.jsonl + screenshots/) into a self-contained HTML report: ranked attention (in plain language), a dwell timeline, and per-scroll-position heatmaps drawn directly on top of what the page actually looked like. Usage: from analysis import generate_report report_path = generate_report(session_dir) # returns abs path to html PAD_PX must match the value used in browser_session.py's injected TRACKING_JS, or the post-hoc AOI attribution here will disagree with what the user visually saw highlighted during the session. """ import os import re import json import bisect PAD_PX = 90 # keep identical to TRACKING_JS's PAD_PX in browser_session.py # ============================================================================= # LOADING # ============================================================================= def _load_jsonl(path): records = [] if not os.path.exists(path): return records with open(path, "r") as f: for line in f: line = line.strip() if not line: continue try: records.append(json.loads(line)) except json.JSONDecodeError: continue return records def load_session(session_dir): gaze = [r for r in _load_jsonl(os.path.join(session_dir, "gaze_log.jsonl")) if r.get("type") == "gaze"] dom = [r for r in _load_jsonl(os.path.join(session_dir, "dom_log.jsonl")) if r.get("type") == "dom"] gaze.sort(key=lambda r: r["t"]) dom.sort(key=lambda r: r["t"]) return gaze, dom # ============================================================================= # HUMAN-READABLE LABELS # Raw AOI labels come straight out of the DOM (tag names, CSS classes, # truncated text). Fine for matching, unreadable for a report. This maps # them to plain language without losing which element they refer to. # ============================================================================= def friendly_label(raw): if not raw: return "Unlabeled area" if raw == "navbar": return "Navigation bar" if raw == "header": return "Page header" if raw == "footer": return "Page footer" if raw == "video": return "Video" if raw == "embed": return "Embedded content (PDF, player, or widget)" if raw.startswith("img: "): return "Image — " + raw[5:] m = re.match(r"^(h[123]):\s*(.*)$", raw) if m: level = {"h1": "Main heading", "h2": "Heading", "h3": "Sub-heading"}[m.group(1)] return f"{level} — \u201c{m.group(2)}\u201d" m = re.match(r"^p \((.*)\)$", raw) if m: return f"Text — \u201c{m.group(1)}\u2026\u201d" if raw.startswith("#"): return "Section: " + raw[1:] if raw.startswith("."): return "Block: " + raw[1:] return raw[0].upper() + raw[1:] if raw else raw # ============================================================================= # GAZE -> AOI ATTRIBUTION # ============================================================================= def _find_aoi(px, py, aois): """Smallest padded AOI containing (px, py), or None. Mirrors the JS hit-test.""" best = None best_area = None for a in aois: x, y, w, h = a["x"], a["y"], a["w"], a["h"] if (x - PAD_PX) <= px <= (x + w + PAD_PX) and (y - PAD_PX) <= py <= (y + h + PAD_PX): area = w * h if best is None or area < best_area: best = a best_area = area return best["label"] if best else None def attribute_gaze(gaze, dom): """For each gaze sample, find the most recent dom snapshot at/before it and test which AOI the point falls in. Returns [(t, raw_label_or_None)].""" if not dom: return [(g["t"], None) for g in gaze] dom_times = [d["t"] for d in dom] out = [] for g in gaze: idx = bisect.bisect_right(dom_times, g["t"]) - 1 if idx < 0: out.append((g["t"], None)) continue snap = dom[idx] label = _find_aoi(g["sx"], g["sy"], snap.get("aois", [])) out.append((g["t"], label)) return out # ============================================================================= # METRICS # ============================================================================= def compute_dwell_ranking(attributed): if len(attributed) < 2: return [] totals, hits = {}, {} for i in range(len(attributed) - 1): t0, label = attributed[i] t1, _ = attributed[i + 1] dt = max(0.0, t1 - t0) if label: totals[label] = totals.get(label, 0.0) + dt hits[label] = hits.get(label, 0) + 1 total_time = sum(totals.values()) or 1.0 ranking = sorted( ({"label": friendly_label(k), "seconds": round(v, 2), "pct": round(100 * v / total_time, 1), "hits": hits[k]} for k, v in totals.items()), key=lambda r: -r["seconds"] ) return ranking def compute_timeline(attributed): segments = [] if not attributed: return segments cur_raw = attributed[0][1] seg_start = attributed[0][0] last_t = attributed[0][0] for t, label in attributed[1:]: if label != cur_raw: segments.append({"start": round(seg_start, 2), "end": round(last_t, 2), "label": friendly_label(cur_raw) if cur_raw else None}) cur_raw = label seg_start = t last_t = t segments.append({"start": round(seg_start, 2), "end": round(last_t, 2), "label": friendly_label(cur_raw) if cur_raw else None}) return segments def session_summary(gaze, dom): if not gaze: return {"duration": 0.0, "url": None, "samples": 0} duration = gaze[-1]["t"] - gaze[0]["t"] url = dom[-1]["url"] if dom else None return {"duration": round(duration, 1), "url": url, "samples": len(gaze)} # ============================================================================= # SCREENSHOT-BACKED HEATMAP SEGMENTS # Groups gaze points against whichever screenshot was on screen at the time, # so the heatmap draws on top of the real page instead of a blank canvas. # ============================================================================= def build_screenshot_segments(gaze, dom): shot_events = [d for d in dom if d.get("screenshot")] if not shot_events: return [] shot_times = [d["t"] for d in shot_events] buckets = [[] for _ in shot_events] for g in gaze: idx = bisect.bisect_right(shot_times, g["t"]) - 1 if idx < 0: idx = 0 buckets[idx].append({"sx": round(g["sx"], 1), "sy": round(g["sy"], 1)}) segments = [] for ev, pts in zip(shot_events, buckets): if not pts: continue segments.append({ "screenshot": ev["screenshot"], "scrollY": ev.get("scrollY", 0), "points": pts, "duration": 0.0, # filled below }) if segments: total_pts = sum(len(s["points"]) for s in segments) total_time = (gaze[-1]["t"] - gaze[0]["t"]) if len(gaze) > 1 else 0.0 for s in segments: share = (len(s["points"]) / total_pts) if total_pts else 0 s["duration"] = round(share * total_time, 1) return segments # ============================================================================= # HTML REPORT # ============================================================================= _TEMPLATE = r""" InsightUX Session Report

InsightUX Session Report

__URL__  ·  generated from __SAMPLES__ gaze samples
__DURATION__s
Session Length
__NUM_ELEMENTS__
Elements Fixated
__TOP_LABEL__
Most Attended

Ranked Attention

How to read this report

Ranked Attention — every part of the page you looked at for a meaningful stretch of time, ordered by how long you spent there.

Heatmap — drawn directly on a screenshot of the page as it appeared during the session. Color shows how much attention that spot received:

Little
A lot

Each card below is a different scroll position — the report splits the page automatically whenever you scrolled far enough that the view changed meaningfully.

Timeline — the same attention data laid out across time, so you can see the order things were looked at, not just the totals.

Heatmap by Scroll Position

Attention Timeline

""" def generate_report(session_dir): gaze, dom = load_session(session_dir) attributed = attribute_gaze(gaze, dom) ranking = compute_dwell_ranking(attributed) timeline = compute_timeline(attributed) segments = build_screenshot_segments(gaze, dom) summary = session_summary(gaze, dom) top_label = ranking[0]["label"] if ranking else "—" html = _TEMPLATE html = html.replace("__URL__", summary["url"] or "Unknown page") html = html.replace("__SAMPLES__", str(summary["samples"])) html = html.replace("__DURATION__", str(summary["duration"])) html = html.replace("__NUM_ELEMENTS__", str(len(ranking))) html = html.replace("__TOP_LABEL__", top_label) html = html.replace("__RANKING_JSON__", json.dumps(ranking)) html = html.replace("__TIMELINE_JSON__", json.dumps(timeline)) html = html.replace("__SEGMENTS_JSON__", json.dumps(segments)) html = html.replace("__DURATION_JSON__", json.dumps(summary["duration"])) out_path = os.path.join(session_dir, "analysis_report.html") with open(out_path, "w", encoding="utf-8") as f: f.write(html) return os.path.abspath(out_path) if __name__ == "__main__": import sys session_dir = sys.argv[1] if len(sys.argv) > 1 else os.path.join("sessions", "live") path = generate_report(session_dir) print(f"Report written to: {path}")