"""Browse a recorded CUA dataset: stats, filters, and a step-through player.
python scripts/view_cua.py artifacts/cua # writes artifacts/cua/index.html
python -m http.server -d artifacts/cua 8097 # then open http://localhost:8097
The page reads `cua.jsonl` (inlined) and the dataset's own image files (by
relative path), so what you step through is exactly what a trainer would load.
"""
from __future__ import annotations
import json
import sys
from collections import Counter
from pathlib import Path
NAMES = {"figma": "Studio", "canva": "Template", "photoshop": "Pixel", "slides": "Deck", "powerpoint": "Present",
"inkscape": "Vector"}
def _calls(msg):
out = []
for c in msg.get("tool_calls", []):
args = json.loads(c["function"]["arguments"])
parts = [f"{k}={json.dumps(v, ensure_ascii=False)}" for k, v in args.items() if v is not None]
out.append(f"{c['function']['name']}({', '.join(parts)})")
return out
def summarise(traj: dict) -> dict:
msgs = traj["messages"]
user_imgs = [c["image"] for c in msgs[1]["content"] if c["type"] == "image"]
steps, frames = [], [user_imgs[-1]]
i = 2
while i < len(msgs):
a = msgs[i]
n = len(a.get("tool_calls", []))
shot, extra = None, []
for r in msgs[i + 1:i + 1 + n]:
if isinstance(r["content"], list):
ims = [c["image"] for c in r["content"] if c["type"] == "image"]
if ims:
shot = ims[-1]
extra += ims[:-1]
steps.append({"t": a.get("content") or "", "c": _calls(a), "m": a.get("train") is False,
"f": len(frames) if shot else None, "x": extra})
if shot:
frames.append(shot)
i += 1 + n
sub = next((k for k, s in enumerate(steps) if any(w in s["t"] for w in ("Export", "export", "Submit", "submit"))),
len(steps) - 1)
done = next((s["f"] for s in reversed(steps[:sub]) if s["f"] is not None), len(frames) - 1)
return {"meta": traj["meta"], "target": user_imgs[0] if traj["meta"]["mode"] == "reference" else None,
"frames": frames, "steps": steps, "done": done}
PAGE = """
CUA dataset viewer
CUA dataset viewer
Trajectories
Player
"""
def build(root: Path) -> Path:
trajs = [summarise(json.loads(line)) for line in (root / "cua.jsonl").read_text().splitlines() if line.strip()]
metas = [t["meta"] for t in trajs]
summary = json.loads((root / "summary.json").read_text()) if (root / "summary.json").exists() else {}
failed = len((root / "failures.jsonl").read_text().splitlines()) if (root / "failures.jsonl").exists() else 0
stats = {
"n": len(trajs), "skins": dict(Counter(m["skin"] for m in metas)), "modes": dict(Counter(m["mode"] for m in metas)),
"themes": dict(Counter(m["theme"] for m in metas)), "mirrored": sum(m["layout"] == "swap" for m in metas),
"mean_turns": round(sum(m["turns"] for m in metas) / max(1, len(metas)), 1),
"mean_actions": round(sum(m["actions"] for m in metas) / max(1, len(metas)), 1),
"images": sum(len(t["frames"]) for t in trajs), "slips": sum(len(m["mistakes"]) for m in metas),
"failed": failed, "skipped": summary.get("skipped", {}),
}
data = {"trajs": trajs, "stats": stats, "names": NAMES, "source": str(root / "cua.jsonl")}
out = root / "index.html"
out.write_text(PAGE.replace("__DATA__", json.dumps(data, ensure_ascii=False)))
return out
if __name__ == "__main__":
print(build(Path(sys.argv[1] if len(sys.argv) > 1 else "artifacts/cua")))