""" 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 # ============================================================================= # MOUSE ACTIVITY (from mouse_log.jsonl — batches pushed by the in-page Mouse # Tracker overlay, same trail/heatmap/click/dwell shape as the standalone # "Mouse Tracker & Heatmap" Chrome extension this was ported from) # ============================================================================= def load_mouse_batches(session_dir): return [r for r in _load_jsonl(os.path.join(session_dir, "mouse_log.jsonl")) if r.get("type") == "mouse_batch"] def summarize_mouse(session_dir): batches = load_mouse_batches(session_dir) dwell_totals = {} clicks = [] trail_points = 0 heatmap_points = 0 for b in batches: for item in (b.get("dwell") or []): element = item.get("element") duration = item.get("duration", 0) if not element: continue dwell_totals[element] = dwell_totals.get(element, 0) + duration for c in (b.get("click") or []): clicks.append(c) trail_points += len(b.get("trail") or []) heatmap_points += len(b.get("heatmap") or []) interests = sorted( ({"element": k, "seconds": round(v / 1000.0, 1)} for k, v in dwell_totals.items()), key=lambda r: -r["seconds"] )[:10] clicks.sort(key=lambda c: c.get("timestamp", "")) return { "interests": interests, "clicks": clicks[-50:], "click_count": len(clicks), "trail_points": trail_points, "heatmap_points": heatmap_points, } # ============================================================================= # 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

Mouse — Most Interacted Elements

Mouse — Click Log

""" 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) mouse = summarize_mouse(session_dir) 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"])) html = html.replace("__MOUSE_INTERESTS_JSON__", json.dumps(mouse["interests"])) html = html.replace("__MOUSE_CLICKS_JSON__", json.dumps(mouse["clicks"])) 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}")