Buckets:
| #!/usr/bin/env python3 | |
| r"""make_figs.py -- regenerate the CockpitWM paper figures (python3 + matplotlib only, no seaborn). | |
| Usage | |
| python3 make_figs.py # all figures | |
| python3 make_figs.py heatmap # figs/robustness_heatmap.pdf (+ .txt sidecar); wide two-panel layout for figure* | |
| python3 make_figs.py heatmap --column # same data, panels stacked at \columnwidth for a single-column figure | |
| python3 make_figs.py heatmap --iclr # same data, side by side at the ICLR \textwidth (5.5 in), CockpitWM palette | |
| python3 make_figs.py arch # figs/architecture.pdf (+ .txt sidecar; portrait, IEEE \columnwidth) | |
| python3 make_figs.py arch --wide # landscape 5.5 x 2.94 in ICLR figure (CockpitWM palette, final model, sidecar with source lines) | |
| python3 make_figs.py arch --clean # clean landscape 5.5 x 2.58 in (owner 2026-09-16: clearer, less text; perception names / chain lists in the caption) | |
| python3 make_figs.py pareto # copy aide/results/tables/pareto_all_vs_hm.png -> figs/ (if different) | |
| python3 make_figs.py ... --out=DIR # write into DIR instead of figs/ (e.g. the ICLR figures directory); --out DIR also accepted | |
| Numbers policy (paper/NUMBERS_POLICY.md) | |
| * robustness_heatmap.pdf: every cell string (mean Delta, sd, verdict) and every clean level is parsed | |
| verbatim from results/tables/robustness_final.md section "## 1." -- nothing is recomputed; the colour | |
| is the float value of the copied mean string. The sidecar .txt dumps the parsed grid so the figure | |
| can be diffed against the table. | |
| * architecture.pdf: every parameter count / shape / constant is copied from docs/model_spec.md | |
| (S1 inputs, S2 branches, S3 interaction, S4/S4b detached heads, S5 decoders) and the warning-head | |
| block sizes from results/tables/warning_head_final_ext.md (header: dwell 11 dims, face 16 dims). | |
| Nothing is derived (no sums). The sidecar .txt lists each label with its source. | |
| Both PDFs carry source path + sha256 + timestamp in their PDF metadata (Subject / Keywords). | |
| """ | |
| from __future__ import annotations | |
| import datetime as _dt | |
| import hashlib | |
| import os | |
| import re | |
| import shutil | |
| import sys | |
| import matplotlib | |
| matplotlib.use("Agg") | |
| import matplotlib.pyplot as plt # noqa: E402 | |
| from matplotlib import colors as mcolors # noqa: E402 | |
| from matplotlib.patches import FancyArrowPatch, FancyBboxPatch # noqa: E402 | |
| HERE = os.path.dirname(os.path.abspath(__file__)) | |
| ROOT = os.path.abspath(os.path.join(HERE, "..", "..")) # .../cwm3 | |
| SRC_ROBUST = os.path.join(ROOT, "results", "tables", "robustness_final.md") | |
| SRC_SPEC = os.path.join(ROOT, "docs", "model_spec.md") | |
| SRC_WHX = os.path.join(ROOT, "results", "tables", "warning_head_final_ext.md") | |
| SRC_PARETO = os.path.join(ROOT, "aide", "results", "tables", "pareto_all_vs_hm.png") | |
| # ---- style (dataviz skill: thin marks, recessive chrome, text in ink tokens, sans everywhere) ---------- | |
| INK = "#0b0b0b" | |
| INK2 = "#52514e" | |
| MUTED = "#898781" | |
| HAIR = "#e1e0d9" | |
| SURFACE = "#fcfcfb" | |
| NEUTRAL_MID = "#f0efec" | |
| # categorical slots 1-3 of the reference palette (documented as validated all-pairs in both modes) | |
| BLUE, ORANGE, AQUA = "#2a78d6", "#eb6834", "#1baf7a" | |
| RED_STATUS = "#d03b3b" # status "critical" -- used once, with a label, for the ablated path | |
| plt.rcParams.update( | |
| { | |
| "font.family": "sans-serif", | |
| "font.sans-serif": ["Nimbus Sans", "Liberation Sans", "Helvetica", "Arial", "DejaVu Sans"], | |
| "mathtext.fontset": "custom", | |
| "mathtext.rm": "Nimbus Sans", | |
| "mathtext.it": "Nimbus Sans:italic", | |
| "mathtext.bf": "Nimbus Sans:bold", | |
| "pdf.fonttype": 42, | |
| "ps.fonttype": 42, | |
| "axes.edgecolor": HAIR, | |
| "axes.linewidth": 0.5, | |
| "text.color": INK, | |
| "axes.labelcolor": INK2, | |
| "xtick.color": INK2, | |
| "ytick.color": INK2, | |
| "figure.facecolor": "white", | |
| "savefig.facecolor": "white", | |
| } | |
| ) | |
| MINUS = "−" # typographic minus for display only; sidecar keeps the verbatim ASCII strings | |
| def _sha256(path: str) -> str: | |
| h = hashlib.sha256() | |
| with open(path, "rb") as f: | |
| for chunk in iter(lambda: f.read(1 << 20), b""): | |
| h.update(chunk) | |
| return h.hexdigest() | |
| def _now() -> str: | |
| return _dt.datetime.now(_dt.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") | |
| def _tint(hex_color: str, frac: float) -> str: | |
| """Blend a hue towards white (frac=1 -> white).""" | |
| r, g, b = mcolors.to_rgb(hex_color) | |
| return mcolors.to_hex((r + (1 - r) * frac, g + (1 - g) * frac, b + (1 - b) * frac)) | |
| # ===================================================================================================== | |
| # 1. Robustness heatmap | |
| # ===================================================================================================== | |
| CELL_RE = re.compile(r"^([+-]\d+\.\d+)\s*±\s*(\d+\.\d+)\s+(WORSE|NOT_WORSE|INCONCLUSIVE|BEATS)$") | |
| CLEAN_RE = re.compile(r"^(\d+\.\d+)\s*±\s*(\d+\.\d+)$") | |
| VERDICT_LETTER = {"WORSE": "W", "NOT_WORSE": "N", "INCONCLUSIVE": "I", "BEATS": "B"} | |
| def parse_robustness_section1(path: str) -> dict: | |
| """Return {regime: {"cols": [...], "clean": {col: (mean_str, sd_str)}, | |
| "rows": [(pert, {col: (mean_str, sd_str, verdict)}), ...]}} from section '## 1.'.""" | |
| with open(path, encoding="utf-8") as f: | |
| lines = f.read().splitlines() | |
| start = next(i for i, l in enumerate(lines) if l.startswith("## 1.")) | |
| end = next(i for i, l in enumerate(lines) if i > start and l.startswith("## ")) | |
| sec = lines[start:end] | |
| out: dict = {} | |
| regime = None | |
| table: list[str] = [] | |
| def flush(): | |
| nonlocal table | |
| if regime is None or not table: | |
| table = [] | |
| return | |
| header = [c.strip() for c in table[0].strip().strip("|").split("|")] | |
| assert header[0] == "perturbation", header | |
| cols = header[1:] | |
| clean: dict = {} | |
| rows: list = [] | |
| for line in table[2:]: # skip header + separator | |
| cells = [c.strip() for c in line.strip().strip("|").split("|")] | |
| assert len(cells) == len(header), (line, len(cells), len(header)) | |
| name, vals = cells[0], cells[1:] | |
| if name.startswith("clean level"): | |
| for c, v in zip(cols, vals): | |
| m = CLEAN_RE.match(v) | |
| assert m, v | |
| clean[c] = (m.group(1), m.group(2)) | |
| else: | |
| d = {} | |
| for c, v in zip(cols, vals): | |
| m = CELL_RE.match(v) | |
| assert m, (name, c, v) | |
| d[c] = (m.group(1), m.group(2), m.group(3)) | |
| rows.append((name, d)) | |
| out[regime] = {"cols": cols, "clean": clean, "rows": rows} | |
| table = [] | |
| for l in sec: | |
| if l.startswith("### "): | |
| flush() | |
| regime = l[4:].strip() | |
| elif l.startswith("|"): | |
| table.append(l) | |
| else: | |
| flush() | |
| flush() | |
| assert set(out) == {"manual", "engaged"}, list(out) | |
| return out | |
| def make_robustness_heatmap(out_pdf: str, layout: str = "wide") -> None: | |
| """layout='wide': two panels side by side, sized to IEEEtran \\textwidth (516 pt) for a figure* float. | |
| layout='column': two panels stacked, sized to \\columnwidth (252 pt) for a single-column figure. | |
| layout='iclr': two panels side by side, sized to the ICLR \\textwidth (396 pt = 5.5 in, single column).""" | |
| assert layout in ("wide", "column", "iclr"), layout | |
| side = layout in ("wide", "iclr") # side-by-side layouts share the row labels on the left panel | |
| data = parse_robustness_section1(SRC_ROBUST) | |
| regimes = ["manual", "engaged"] | |
| cols = data["manual"]["cols"] | |
| assert cols == data["engaged"]["cols"] | |
| perts = [r[0] for r in data["manual"]["rows"]] | |
| assert perts == [r[0] for r in data["engaged"]["rows"]] | |
| n_r, n_c = len(perts), len(cols) | |
| # Owner decision 2026-09-17 (docs/decisions.md row 443, item 3): the ENGAGED hand_leave / hand_return cells are not reported in the | |
| # ICLR manuscript. The ICLR layout keeps the shared row/column grid and renders these two cells (clean level and every perturbation) | |
| # as neutral "n/r" cells, excludes them from the colour range and writes "not reported" into the sidecar; IEEE layouts are untouched. | |
| omit = {("engaged", "hand_leave"), ("engaged", "hand_return")} if layout == "iclr" else set() | |
| # colour range: symmetric about 0 so the neutral midpoint is exactly Delta = 0 (diverging palette); | |
| # vmax = max |mean| over all cells, rounded up to 0.01 | |
| all_vals = [float(v[0]) for reg in regimes for _, d in data[reg]["rows"] for c, v in d.items() if (reg, c) not in omit] | |
| raw_max = max(abs(v) for v in all_vals) | |
| vmax = round(raw_max + 0.005, 2) if abs(raw_max * 100 - round(raw_max * 100)) > 1e-9 else round(raw_max, 2) | |
| norm = mcolors.Normalize(vmin=-vmax, vmax=vmax) | |
| cw = layout == "iclr" # ICLR: CockpitWM palette only; IEEE layouts keep the original RdBu | |
| if cw: | |
| # diverging map on the CockpitWM palette: cwPeachL (negative Delta, worse) -> white at 0 -> cwGreyL (positive) | |
| cmap = mcolors.LinearSegmentedColormap.from_list("cw_div", [CW["peachL"], "#FFFFFF", CW["greyL"]]) | |
| cmap_name = "cwPeachL #F5D8BF -> white -> cwGreyL #BFDCE7" | |
| c_ink, c_ink2, c_muted, c_hair, c_surface, c_row = CW_BLACK, CW_BLACK, CW_MUTED, CW_HAIR, "#FFFFFF", CW_ROW | |
| else: | |
| # ColorBrewer RdBu: red (loss) <-> near-white <-> blue (gain); CVD-safe diverging pair with neutral midpoint | |
| cmap = plt.get_cmap("RdBu") | |
| cmap_name = "RdBu" | |
| c_ink, c_ink2, c_muted, c_hair, c_surface, c_row = INK, INK2, MUTED, HAIR, SURFACE, NEUTRAL_MID | |
| col_labels = { | |
| "steer_phase": "steer\nphase", | |
| "gas": "gas", | |
| "brake": "brake", | |
| "veh_any": "veh\nany", | |
| "hand_leave": "hand\nleave", | |
| "hand_return": "hand\nreturn", | |
| "head_turn": "head\nturn", | |
| "P ttc_drop": "P ttc\ndrop", | |
| "P cut_in": "P cut\nin", | |
| } | |
| # ---- geometry in inches -> figure fractions ---- | |
| cell_h = 0.19 | |
| panel_h = cell_h * (n_r + 1) # + clean-level row | |
| cbar_h = 0.07 | |
| if layout == "wide": | |
| fig_w = 516.0 / 72.0 # IEEEtran conference \textwidth | |
| x_lab = 1.02 # row-label gutter (left panel only) | |
| gap_x = 0.34 | |
| pw = (fig_w - x_lab - gap_x - 0.10) / 2 | |
| panels = {"manual": (x_lab, pw), "engaged": (x_lab + pw + gap_x, pw)} | |
| y_legend, y_cblabel, y_cbar = 0.08, 0.24, 0.46 # tick labels hang ~0.1 in below the bar | |
| y_p = y_cbar + cbar_h + 0.30 | |
| y_panel = {"manual": y_p, "engaged": y_p} | |
| fig_h = y_p + panel_h + 0.50 # top tick labels + regime title | |
| fs_val, fs_let, fs_tick = 6.0, 5.2, 6.0 | |
| elif layout == "iclr": | |
| fig_w = 396.0 / 72.0 # ICLR \textwidth (5.5 in, single column) | |
| x_lab = 0.90 # row-label gutter (left panel only) | |
| gap_x = 0.20 | |
| pw = (fig_w - x_lab - gap_x - 0.06) / 2 | |
| panels = {"manual": (x_lab, pw), "engaged": (x_lab + pw + gap_x, pw)} | |
| y_legend, y_cblabel, y_cbar = 0.08, 0.24, 0.46 | |
| y_p = y_cbar + cbar_h + 0.30 | |
| y_panel = {"manual": y_p, "engaged": y_p} | |
| fig_h = y_p + panel_h + 0.50 | |
| fs_val, fs_let, fs_tick = 5.0, 4.7, 5.6 | |
| else: | |
| fig_w = 252.0 / 72.0 # \columnwidth | |
| x_lab, pw = 1.05, 3.48 - 1.05 | |
| panels = {"manual": (x_lab, pw), "engaged": (x_lab, pw)} | |
| y_legend, y_cblabel, y_cbar = 0.08, 0.29, 0.50 | |
| y_eng = y_cbar + cbar_h + 0.40 | |
| y_panel = {"engaged": y_eng, "manual": y_eng + panel_h + 0.34} | |
| fig_h = y_panel["manual"] + panel_h + 0.36 | |
| fs_val, fs_let, fs_tick = 5.6, 5.0, 6.0 | |
| fig = plt.figure(figsize=(fig_w, fig_h)) | |
| def fx(x): | |
| return x / fig_w | |
| def fy(y): | |
| return y / fig_h | |
| axes = {} | |
| for reg in regimes: | |
| x0, w = panels[reg] | |
| axes[reg] = fig.add_axes([fx(x0), fy(y_panel[reg]), fx(w), fy(panel_h)]) | |
| cell_w_in = panels["manual"][1] / n_c | |
| cell_texts: list = [] # value strings, checked against the cell width after layout | |
| for reg in regimes: | |
| ax, d = axes[reg], data[reg] | |
| ax.set_xlim(0, n_c) | |
| ax.set_ylim(n_r + 1, 0) | |
| ax.set_facecolor(c_surface) | |
| for spine in ax.spines.values(): | |
| spine.set_visible(False) | |
| # row 0: clean level (text only, neutral surface) | |
| ax.add_patch(plt.Rectangle((0, 0), n_c, 1, fc=c_row, ec="none")) | |
| for j, c in enumerate(cols): | |
| if (reg, c) in omit: | |
| cell_texts.append(ax.text(j + 0.5, 0.5, "n/r", ha="center", va="center", fontsize=fs_let, color=c_muted)) | |
| continue | |
| mean_s, sd_s = d["clean"][c] | |
| cell_texts.append(ax.text(j + 0.5, 0.5, f"{mean_s}\n±{sd_s}", ha="center", va="center", | |
| fontsize=fs_let, color=c_ink2, linespacing=1.0)) | |
| # rows 1..n_r: perturbation cells | |
| for i in range(n_r): | |
| for j, c in enumerate(cols): | |
| if (reg, c) in omit: # not reported (owner decision 2026-09-17): neutral cell, no value, no verdict letter | |
| ax.add_patch(plt.Rectangle((j + 0.02, i + 1 + 0.02), 0.96, 0.96, fc=c_row, ec="none")) | |
| cell_texts.append(ax.text(j + 0.5, i + 1 + 0.55, "n/r", ha="center", va="center", fontsize=fs_val, color=c_muted)) | |
| continue | |
| mean_s, sd_s, verdict = d["rows"][i][1][c] | |
| rgba = cmap(norm(float(mean_s))) | |
| ax.add_patch(plt.Rectangle((j + 0.02, i + 1 + 0.02), 0.96, 0.96, fc=rgba, ec="none")) | |
| lum = 0.2126 * rgba[0] + 0.7152 * rgba[1] + 0.0722 * rgba[2] | |
| tc = "white" if lum < 0.45 else c_ink | |
| cell_texts.append(ax.text(j + 0.5, i + 1 + 0.36, mean_s.replace("-", MINUS), ha="center", | |
| va="center", fontsize=fs_val, color=tc)) | |
| ax.text(j + 0.5, i + 1 + 0.75, VERDICT_LETTER[verdict], ha="center", va="center", | |
| fontsize=fs_let, color=tc, fontweight="bold" if verdict == "WORSE" else "normal") | |
| ax.set_yticks([0.5] + [i + 1.5 for i in range(n_r)]) | |
| if side and reg == "engaged": | |
| ax.set_yticklabels([]) # rows are aligned with the left panel | |
| else: | |
| ax.set_yticklabels(["clean level\n(mean ± sd)"] + perts, fontsize=fs_tick) | |
| ax.tick_params(axis="y", length=0, pad=3, labelcolor=c_ink2) | |
| ax.set_xticks([j + 0.5 for j in range(n_c)]) | |
| ax.set_xticklabels([col_labels[c] for c in cols], fontsize=fs_tick, linespacing=0.95) | |
| ax.tick_params(axis="x", length=0, pad=2, labelcolor=c_ink2) | |
| if side or reg == "manual": | |
| ax.xaxis.tick_top() | |
| # white separators between endpoint families (vehicle | driver | road pressure) | |
| for xg in (4, 7): | |
| ax.plot([xg, xg], [0, n_r + 1], color="white", lw=1.6, solid_capstyle="butt", zorder=3) | |
| # family labels just below the panel (column layout: only in the gap under the manual panel, | |
| # the engaged panel carries the bottom tick labels there) | |
| if side or reg == "manual": | |
| for xc, lab in ((2.0, "vehicle"), (5.5, "driver"), (8.0, "road pressure")): | |
| ax.text(xc, n_r + 1.55, lab, ha="center", va="center", fontsize=5.8, color=c_muted) | |
| # regime titles (bold) | |
| if side: | |
| for reg in regimes: | |
| axes[reg].text(0.0, 1.0 + 0.36 / panel_h, reg, transform=axes[reg].transAxes, ha="left", | |
| va="bottom", fontsize=6.8, fontweight="bold", color=c_ink) | |
| else: | |
| for reg in regimes: | |
| axes[reg].text(-0.02, 1.03, reg, transform=axes[reg].transAxes, ha="right", va="bottom", | |
| fontsize=6.5, fontweight="bold", color=c_ink) | |
| # colourbar (horizontal, below, centred under the panel area) | |
| x_c0 = panels["manual"][0] | |
| x_c1 = panels["engaged"][0] + panels["engaged"][1] | |
| cb_w = {"wide": 2.2, "iclr": 2.0}.get(layout, (x_c1 - x_c0) - 0.70) | |
| cax = fig.add_axes([fx((x_c0 + x_c1) / 2 - cb_w / 2), fy(y_cbar), fx(cb_w), fy(cbar_h)]) | |
| sm = plt.cm.ScalarMappable(norm=norm, cmap=cmap) | |
| cb = fig.colorbar(sm, cax=cax, orientation="horizontal") | |
| cb.outline.set_linewidth(0.4) | |
| cb.outline.set_edgecolor(c_hair) | |
| ticks = [-vmax, -0.1, 0.0, 0.1, vmax] if vmax > 0.1 else [-vmax, 0.0, vmax] | |
| cb.set_ticks(ticks) | |
| cb.set_ticklabels([("0" if t == 0 else f"{t:+.2f}".replace("-", MINUS)) for t in ticks]) | |
| cb.ax.tick_params(labelsize=5.6, length=2, width=0.4, color=c_muted, labelcolor=c_ink2, pad=1.5) | |
| fig.text(fx((x_c0 + x_c1) / 2), fy(y_cblabel), "Δ PR-AUC@3 s (perturbed − clean), mean of 3 seeds", | |
| ha="center", va="center", fontsize=6.0, color=c_ink2) | |
| legend = ("cell letter = verdict over 3 seeds: W significantly worse (bold) " | |
| "N matches clean (|Δ| < 0.01 on every seed, no CI excl. 0) I no significant difference") | |
| if layout == "column": | |
| legend = legend.replace("(bold) N", "(bold)\nN") | |
| x_leg = (x_c0 + x_c1) / 2 if side else fig_w / 2 | |
| fig.text(fx(x_leg), fy(y_legend), legend, ha="center", va="center", fontsize=5.2, color=c_ink2, linespacing=1.25) | |
| # fit check: every printed cell string must be narrower than its cell (fail loudly, never clip) | |
| fig.canvas.draw() | |
| rend = fig.canvas.get_renderer() | |
| too_wide = [] | |
| for t in cell_texts: | |
| bb = t.get_window_extent(renderer=rend) | |
| if bb.width / fig.dpi > cell_w_in - 0.01: | |
| too_wide.append((t.get_text(), round(bb.width / fig.dpi, 3), round(cell_w_in, 3))) | |
| if too_wide: | |
| for b in too_wide: | |
| print("CELL TEXT OVERFLOW:", b) | |
| raise SystemExit("robustness_heatmap.pdf: cell text wider than its cell -- reduce fs_val or widen the panel") | |
| src_sha = _sha256(SRC_ROBUST) | |
| meta = { | |
| "Title": "CockpitWM robustness heatmap: Delta PR-AUC@3s (perturbed - clean), frozen main model", | |
| "Subject": f"source: {SRC_ROBUST} section 1 (sha256 {src_sha}); layout {layout}; rendered {_now()} by make_figs.py", | |
| "Keywords": "CockpitWM, robustness_final.md, section 1, verbatim copy, no recomputation", | |
| "Creator": "paper/figs/make_figs.py (matplotlib)", | |
| } | |
| fig.savefig(out_pdf, metadata=meta) | |
| plt.close(fig) | |
| # sidecar: parsed grid, so every plotted string can be diffed against the table | |
| with open(out_pdf[:-4] + ".txt", "w", encoding="utf-8") as f: | |
| f.write(f"# {os.path.basename(out_pdf)} -- generated {_now()} by paper/figs/make_figs.py (layout {layout})\n") | |
| f.write(f"# source: {SRC_ROBUST}\n# source sha256: {src_sha}\n") | |
| f.write("# section: '## 1. Δ PR-AUC@3s (perturbed − clean) on the main endpoints' (### manual, ### engaged)\n") | |
| f.write("# every cell below is the verbatim table string 'mean ± sd VERDICT'; colour = float(mean); " | |
| "letter = W/N/I/B for WORSE/NOT_WORSE/INCONCLUSIVE/BEATS\n") | |
| f.write(f"# colour scale: {cmap_name}, symmetric, vmin=-{vmax:.2f} vmax=+{vmax:.2f} (max |mean| over all reported cells, rounded up to 0.01)\n") | |
| if omit: | |
| f.write("# not reported (owner decision 2026-09-17, docs/decisions.md row 443 item 3): " + ", ".join(f"{r}/{c}" for r, c in sorted(omit)) + " -- rendered as n/r\n") | |
| for reg in regimes: | |
| d = data[reg] | |
| f.write(f"\n[{reg}]\n") | |
| f.write("clean level (mean ± sd): " + " | ".join( | |
| (f"{c}=not reported" if (reg, c) in omit else f"{c}={d['clean'][c][0]} ± {d['clean'][c][1]}") for c in cols) + "\n") | |
| for name, row in d["rows"]: | |
| f.write(f"{name}: " + " | ".join( | |
| (f"{c}=not reported" if (reg, c) in omit else f"{c}={row[c][0]} ± {row[c][1]} {row[c][2]}") for c in cols) + "\n") | |
| print(f"wrote {out_pdf} ({layout}, {fig_w:.3f} x {fig_h:.3f} in) and sidecar; vmax={vmax:.2f}; source sha256 {src_sha[:16]}…") | |
| # ===================================================================================================== | |
| # 2. Architecture block diagram | |
| # ===================================================================================================== | |
| _BOXED: list = [] # (text artist, w_in, h_in) for the overflow check | |
| def _box(ax, x, y, w, h, text, fc, ec, fs=5.6, lw=0.7, color=INK, ls="-", rounding=0.05, weight="normal", | |
| linespacing=1.15, zorder=2, ty=None): | |
| p = FancyBboxPatch((x, y), w, h, boxstyle=f"round,pad=0,rounding_size={rounding}", fc=fc, ec=ec, lw=lw, | |
| linestyle=ls, zorder=zorder) | |
| ax.add_patch(p) | |
| if text: | |
| t = ax.text(x + w / 2, y + h / 2 if ty is None else ty, text, ha="center", va="center", fontsize=fs, | |
| color=color, linespacing=linespacing, fontweight=weight, zorder=zorder + 1) | |
| _BOXED.append((t, w, h)) | |
| return x, y, w, h | |
| def _check_overflow(fig) -> None: | |
| """Fail loudly if any boxed text is wider/taller than its box (axes are in inches, 1 unit = 1 inch).""" | |
| fig.canvas.draw() | |
| r = fig.canvas.get_renderer() | |
| bad = [] | |
| for t, w, h in _BOXED: | |
| bb = t.get_window_extent(renderer=r) | |
| tw, th = bb.width / fig.dpi, bb.height / fig.dpi | |
| if tw > w - 0.03 or th > h - 0.02: | |
| bad.append((t.get_text().splitlines()[0][:40], round(tw, 3), round(w, 3), round(th, 3), round(h, 3))) | |
| if bad: | |
| for b in bad: | |
| print("TEXT OVERFLOW:", b) | |
| raise SystemExit("architecture.pdf: text overflows its box -- shorten the label or widen the box") | |
| def _arrow(ax, p0, p1, color=INK2, lw=0.7, ls="-", rad=0.0, head=4.5, zorder=4, shrinkA=0.5, shrinkB=0.5): | |
| a = FancyArrowPatch(p0, p1, arrowstyle="-|>", mutation_scale=head, color=color, lw=lw, linestyle=ls, | |
| connectionstyle=f"arc3,rad={rad}", shrinkA=shrinkA, shrinkB=shrinkB, zorder=zorder) | |
| ax.add_patch(a) | |
| return a | |
| # every label of both architecture layouts -> its source (nothing derived); written to the .txt sidecar | |
| _ARCH_SOURCE_ROWS = [ | |
| ("V branch 1,608,347 params; TCN(6->256, k=5, dil 1/2/4/8); head [B,25,35] = 3 MDNs + gas/brake/steer_phase/resp chains", "A S2 row V"), | |
| ("U branch 159,290 params; GRU(18->128) last state h_u [B,128]; head [B,25,26] hands_on / head_turned chains", "A S2 row U"), | |
| ("E branch 880,604 params (pooled); one PCA-256 clip token per 0.5 s slot, 16 tokens; 2x causal TransformerEncoderLayer(d128, 4 heads, FFN 256); z_e [B,128]; z_vf LN-Linear(256->128); head [B,25,12] 6 demand chains; F [B,5,256]", "A S2 row E"), | |
| ("inputs: X [B,40,6] CAN 4 s @ 10 Hz + dt; Kx [B,40,18] = 11 pose kinematics + valid bit + 5 CAN + dt @ 10 Hz; tok 16 slots x 0.5 s over 8 s causal bank; vf [B,256] front PCA-256 at the anchor; mode in every branch", "A S1"), | |
| ("Road2Driver cross-attention 231,424 (query h_u, keys/values RegionTokenEmbed of the region token bank); RegionTokenEmbed 11,392", "A S3 rows 1-2"), | |
| ("Env2Vehicle (drawn as road->vehicle) [z_e ; z_vf] -> residual on z_v, 329,088", "A S3 row 3"), | |
| ("Driver2Vehicle h_u -> residual on z_v, sigmoid(Embedding(mode)) gate, 247,680", "A S3 row 4"), | |
| ("zero-init -> bitwise identity at init", "A S3 heading"), | |
| ("K head detached 89,510: dK mu / log sigma", "A S4"), | |
| ("P head detached 146,988: six road-pressure chains", "A S4b"), | |
| ("R head detached 41,394: hard-response chain", "A S4b"), | |
| ("DirectDecoder25 (direct) = flat 25-step head", "A S5"), | |
| ("main model keeps road->vehicle (Env2Vehicle) and driver->vehicle, road->driver removed", "A S3 forward order; docs/results_cwm3.md S0 (all_minus_road2drv)"), | |
| ("warning head: causal current-state features of the frozen model at the anchor, dwell/duration block 11 dims, face-landmark block 16 dims; fixed HGB; AUROC vs DMS attention chain on gold", "B header (n979dw, n979dwfc; feature sets) + results/tables/warning_head_final.md header"), | |
| ] | |
| def _write_arch_sidecar(out_pdf: str, spec_sha: str, whx_sha: str, layout: str) -> None: | |
| with open(out_pdf[:-4] + ".txt", "w", encoding="utf-8") as f: | |
| f.write(f"# {os.path.basename(out_pdf)} -- generated {_now()} by paper/figs/make_figs.py (layout {layout})\n") | |
| f.write(f"# source A: {SRC_SPEC} (sha256 {spec_sha})\n# source B: {SRC_WHX} (sha256 {whx_sha})\n") | |
| f.write("# every label -> source (nothing derived):\n") | |
| for label, src in _ARCH_SOURCE_ROWS: | |
| f.write(f"- {label}\n <- {src}\n") | |
| def make_architecture(out_pdf: str) -> None: | |
| W_IN, H_IN = 3.5, 3.72 | |
| _BOXED.clear() | |
| fig = plt.figure(figsize=(W_IN, H_IN)) | |
| ax = fig.add_axes([0, 0, 1, 1]) | |
| ax.set_xlim(0, W_IN) | |
| ax.set_ylim(0, H_IN) | |
| ax.set_aspect("equal") | |
| ax.axis("off") | |
| # columns: U (driver) | V (vehicle) | E (environment) -- V in the middle receives both surviving paths | |
| col_w = 1.08 | |
| xs = {"U": 0.05, "V": 1.21, "E": 2.37} | |
| cx = {k: v + col_w / 2 for k, v in xs.items()} | |
| hue = {"U": ORANGE, "V": BLUE, "E": AQUA} | |
| fill = {k: _tint(h, 0.86) for k, h in hue.items()} | |
| fill2 = {k: _tint(h, 0.94) for k, h in hue.items()} | |
| # ---- row 0: column titles with parameter counts (model_spec.md S2) -------------------------------- | |
| titles = { | |
| "U": "U driver branch\n159,290 params", | |
| "V": "V vehicle branch\n1,608,347 params", | |
| "E": "E environment branch\n880,604 params (pooled)", | |
| } | |
| for k in "UVE": | |
| ax.text(cx[k], 0.12, titles[k], ha="center", va="center", fontsize=5.8, color=hue[k], | |
| fontweight="bold", linespacing=1.1) | |
| # ---- row 1: inputs (model_spec.md S1) -------------------------------------------------------------- | |
| y_in, h_in = 0.28, 0.38 | |
| inputs = { | |
| "U": "driver kinematics (causal ROI)\n11 kin + valid bit + 5 CAN + Δt\n4 s @ 10 Hz [40×18]", | |
| "V": "CAN history + Δt\n4 s @ 10 Hz [40×6]", | |
| "E": "pooled road tokens (V-JEPA 2)\n1 per 0.5 s slot, 8 s causal bank\n+ front latent at the anchor [256]", | |
| } | |
| for k in "UVE": | |
| _box(ax, xs[k], y_in, col_w, h_in, inputs[k], fc="white", ec=MUTED, fs=4.6, lw=0.6, color=INK2, | |
| ls=(0, (2, 1.2)), linespacing=1.1) | |
| # ---- row 2: trunks (model_spec.md S2) -------------------------------------------------------------- | |
| y_tr, h_tr = 0.78, 0.32 | |
| trunks = { | |
| "U": "GRU (18→128)\ncausal, last state", | |
| "V": "TCN (6→256)\nk 5, dilations 1/2/4/8", | |
| "E": "2× causal Transformer\n(d 128, 4 heads, FFN 256)", | |
| } | |
| for k in "UVE": | |
| _box(ax, xs[k], y_tr, col_w, h_tr, trunks[k], fc=fill[k], ec=hue[k], fs=5.3, lw=0.8) | |
| _arrow(ax, (cx[k], y_in + h_in), (cx[k], y_tr), color=INK2) | |
| # ---- row 3: latent states + directed interaction I (model_spec.md S3) ------------------------------- | |
| y_lat, h_lat = 1.40, 0.22 | |
| lat_w = 0.62 | |
| lat = {"U": r"$h_U$ [128]", "V": r"$z_V$ [256]", "E": r"$[z_E ; z_{vf}]$ [256]"} | |
| # interaction backdrop | |
| _box(ax, 0.04, 1.14, W_IN - 0.08, 0.98, "", fc="#f6f5f2", ec=HAIR, lw=0.5, rounding=0.06, zorder=1) | |
| ax.text(0.09, 2.07, "directed interaction I\nzero-init residuals\n(identity at init)", | |
| ha="left", va="top", fontsize=4.9, color=INK2, fontweight="bold", linespacing=1.1) | |
| lx = {k: cx[k] - lat_w / 2 for k in "UVE"} | |
| for k in "UVE": | |
| _box(ax, lx[k], y_lat, lat_w, h_lat, lat[k], fc="white", ec=hue[k], fs=5.6, lw=0.8) | |
| _arrow(ax, (cx[k], y_tr + h_tr), (cx[k], y_lat), color=INK2) | |
| ym = y_lat + h_lat / 2 | |
| # surviving paths into z_V (solid, dark) | |
| _arrow(ax, (lx["U"] + lat_w, ym), (lx["V"], ym), color=INK, lw=0.9, head=5) | |
| ax.text((lx["U"] + lat_w + lx["V"]) / 2, y_lat - 0.03, "driver→vehicle\nmode-gated MLP · 247,680", | |
| ha="center", va="top", fontsize=4.8, color=INK, linespacing=1.1) | |
| _arrow(ax, (lx["E"], ym), (lx["V"] + lat_w, ym), color=INK, lw=0.9, head=5) | |
| ax.text((lx["E"] + lx["V"] + lat_w) / 2, y_lat - 0.03, "env→vehicle\nMLP · 329,088", | |
| ha="center", va="top", fontsize=4.8, color=INK, linespacing=1.1) | |
| # ablated path road->driver: dashed arc from the E column (region-token bank) to h_U, crossed out | |
| y_arc = y_lat + h_lat | |
| x0, x1 = lx["E"] + 0.10, lx["U"] + lat_w - 0.10 | |
| rad = 0.22 | |
| _arrow(ax, (x0, y_arc), (x1, y_arc), color=MUTED, lw=0.8, ls=(0, (2.2, 1.4)), rad=rad, head=4.5) | |
| # quadratic Bezier of arc3: control point = midpoint + rad*|d| perpendicular; evaluate at t for the cross | |
| t = 0.62 # left of the V column centre so the cross does not sit on the V arrow | |
| mx, my = (x0 + x1) / 2, y_arc + rad * (x0 - x1) | |
| xm = (1 - t) ** 2 * x0 + 2 * (1 - t) * t * mx + t**2 * x1 | |
| ycross = (1 - t) ** 2 * y_arc + 2 * (1 - t) * t * my + t**2 * y_arc | |
| s = 0.032 | |
| ax.plot([xm - s, xm + s], [ycross - s, ycross + s], color=RED_STATUS, lw=1.1, zorder=6) | |
| ax.plot([xm - s, xm + s], [ycross + s, ycross - s], color=RED_STATUS, lw=1.1, zorder=6) | |
| ax.text(cx["V"] + 0.30, y_arc + rad * (x0 - x1) / 2 + 0.05, | |
| "road→driver cross-attention — removed in the main model\n" | |
| r"($h_U$ attends over the region-token bank) 231,424 + 11,392 token embedding", | |
| ha="center", va="bottom", fontsize=4.6, color=INK2, linespacing=1.15) | |
| # ---- row 4: heads: direct 25-step decoders + detached heads (model_spec.md S2, S4, S4b, S5) ------- | |
| y_hd, h_hd = 2.22, 0.66 | |
| heads = { | |
| "U": "direct 25-step decoder [25×26]\nhands_on / head_turned\nhazard chains", | |
| "V": "direct 25-step decoder [25×35]\n3 MDNs (ego signals) +\ngas / brake / steer_phase / resp chains", | |
| "E": "direct 25-step decoder [25×12]\n6 demand chains\n+ future-latent head F [5×256]", | |
| } | |
| dheads = { | |
| "U": "K head (detached) 89,510\n" + r"$\Delta\hat{K}$ residual (μ, log σ)", | |
| "V": "R head (detached) 41,394\nhard-response chain", | |
| "E": "P head (detached) 146,988\n6 road-pressure chains", | |
| } | |
| h_sub = 0.24 | |
| for k in "UVE": | |
| _box(ax, xs[k], y_hd, col_w, h_hd, "", fc=fill[k], ec=hue[k], lw=0.8) | |
| ax.text(cx[k], y_hd + (h_hd - h_sub - 0.04) / 2, heads[k], ha="center", va="center", fontsize=4.9, | |
| color=INK, linespacing=1.1, zorder=3) | |
| _box(ax, xs[k] + 0.04, y_hd + h_hd - h_sub - 0.04, col_w - 0.08, h_sub, dheads[k], fc=fill2[k], ec=hue[k], | |
| fs=4.8, lw=0.7, ls=(0, (1.2, 1.0)), linespacing=1.1, zorder=3) | |
| _arrow(ax, (cx[k], y_lat + h_lat), (cx[k], y_hd), color=INK2) | |
| ax.text(W_IN - 0.06, y_hd + h_hd + 0.025, "dotted sub-box = detached head on the same branch state (no gradient to any trunk)", | |
| ha="right", va="bottom", fontsize=4.5, color=MUTED) | |
| # ---- row 5: post-hoc warning head W (warning_head_final_ext.md header; model_spec.md S8) ------------ | |
| y_w, h_w = 3.06, 0.46 | |
| _box(ax, 0.08, y_w, W_IN - 0.16, h_w, | |
| "post-hoc warning head W (fixed histogram gradient boosting; trained on the frozen model's outputs)\n" | |
| "current-state features at the anchor (no predicted quantity)\n" | |
| "+ dwell / duration block (11) + face-landmark block (16)\n" | |
| "→ warning score; AUROC vs the openpilot DMS attention chain on the human gold set", | |
| fc="white", ec=INK, fs=4.8, lw=0.8, linespacing=1.2) | |
| for k in "UVE": | |
| _arrow(ax, (cx[k], y_hd + h_hd), (cx[k], y_w), color=INK2) | |
| ax.text(W_IN / 2, y_w + h_w + 0.08, | |
| "regime m ∈ {manual, engaged} embedded in every branch solid = trained path " | |
| "dashed + × = ablated path", | |
| ha="center", va="center", fontsize=4.6, color=MUTED) | |
| spec_sha, whx_sha = _sha256(SRC_SPEC), _sha256(SRC_WHX) | |
| meta = { | |
| "Title": "CockpitWM joint model architecture (portrait)", | |
| "Subject": f"sources: {SRC_SPEC} S1-S5 (sha256 {spec_sha}); {SRC_WHX} header (sha256 {whx_sha}); rendered {_now()} by make_figs.py", | |
| "Keywords": "CockpitWM, model_spec.md, parameter counts verbatim", | |
| "Creator": "paper/figs/make_figs.py (matplotlib)", | |
| } | |
| _check_overflow(fig) | |
| fig.savefig(out_pdf, metadata=meta) | |
| plt.close(fig) | |
| _write_arch_sidecar(out_pdf, spec_sha, whx_sha, "portrait") | |
| print(f"wrote {out_pdf} and sidecar") | |
| # ---- CockpitWM ICLR palette (BATON__/ICLR/cwm_macros.tex lines 30-33) -- the ONLY hues in the ICLR figures -------- | |
| CW = { | |
| "lav": "#DDD6E5", # secondary module fill (detached auxiliary heads) | |
| "lavL": "#DFEDF2", # observed inputs | |
| "greyL": "#BFDCE7", # frozen external models / header blue | |
| "grey": "#DDD6E5", | |
| "peachL": "#F5D8BF", # ours / highlighted (heatmap: negative Delta pole) | |
| "peach": "#D9E4C2", # trainable modules | |
| "coral": "#CAC2D7", # prediction outputs | |
| "red": "#CAC2D7", | |
| "bg": "#F9EDE0", # background panels | |
| } | |
| CW_BLACK = "#000000" | |
| CW_INK2 = "#404040" # black!75 -- arrows, secondary text | |
| CW_MUTED = "#6E6E6E" # black!57 -- de-emphasised captions | |
| CW_FROZEN = "#8C8C8C" # black!45 -- frozen-module borders | |
| CW_HAIR = "#D9D9D9" # black!15 -- hairlines | |
| CW_ROW = "#EEEEEE" # black!7 -- neutral surface (heatmap clean-level row) | |
| CW_DMS = "#F2F2F2" # black!5 -- external comparator fill | |
| def _measure(fig, s: str, fs: float, weight: str = "normal") -> tuple: | |
| """Width / height in inches of a one-off text string (temporary artist, removed after measuring).""" | |
| t = fig.text(0, 0, s, fontsize=fs, fontweight=weight) | |
| fig.canvas.draw() | |
| bb = t.get_window_extent(renderer=fig.canvas.get_renderer()) | |
| t.remove() | |
| return bb.width / fig.dpi, bb.height / fig.dpi | |
| def _snowflake(ax, x, y, r=0.03, color=CW_FROZEN, lw=0.6, zorder=6) -> None: | |
| """Six-spoke snowflake glyph drawn as vector strokes (Nimbus Sans has no U+2744).""" | |
| import math | |
| for k in range(3): | |
| a = math.radians(90 + 60 * k) | |
| dx, dy = r * math.cos(a), r * math.sin(a) | |
| ax.plot([x - dx, x + dx], [y - dy, y + dy], color=color, lw=lw, solid_capstyle="round", zorder=zorder) | |
| for s in (1, -1): | |
| px, py = x + s * dx, y + s * dy | |
| for t in (1, -1): | |
| b = a + t * math.radians(40) | |
| ax.plot([px, px - s * 0.42 * r * math.cos(b)], [py, py - s * 0.42 * r * math.sin(b)], | |
| color=color, lw=lw * 0.8, solid_capstyle="round", zorder=zorder) | |
| def _dot(ax, x, y, r=0.016, color=CW_INK2, zorder=6) -> None: | |
| ax.add_patch(plt.Circle((x, y), r, fc=color, ec="none", zorder=zorder)) | |
| def _oplus(ax, x, y, r=0.045, zorder=6) -> None: | |
| ax.add_patch(plt.Circle((x, y), r, fc="white", ec=CW_BLACK, lw=0.7, zorder=zorder)) | |
| ax.plot([x - 0.55 * r, x + 0.55 * r], [y, y], color=CW_BLACK, lw=0.7, zorder=zorder + 1) | |
| ax.plot([x, x], [y - 0.55 * r, y + 0.55 * r], color=CW_BLACK, lw=0.7, zorder=zorder + 1) | |
| def _tri_down(ax, x, y, w=0.05, h=0.03, color=CW_MUTED, zorder=5) -> None: | |
| ax.add_patch(plt.Polygon([[x - w / 2, y + h / 2], [x + w / 2, y + h / 2], [x, y - h / 2]], closed=True, | |
| fc=color, ec="none", zorder=zorder)) | |
| def _elbow(ax, p0, p1, xm, color=CW_INK2, lw=0.7, head=4.5, ls="-", zorder=4) -> None: | |
| """p0 -> (xm, p0.y) -> (xm, p1.y) -> p1 (arrow head on the last leg).""" | |
| ax.plot([p0[0], xm, xm], [p0[1], p0[1], p1[1]], color=color, lw=lw, ls=ls, solid_joinstyle="round", | |
| solid_capstyle="butt", zorder=zorder) | |
| _arrow(ax, (xm, p1[1]), p1, color=color, lw=lw, head=head, ls=ls, zorder=zorder, shrinkA=0, shrinkB=0.5) | |
| # every printed label of the ICLR landscape layout -> its source line (rounded M values are printed in the boxes; | |
| # the exact count and its line are listed here; the only derived quantity is the 3.50 M total, marked DERIVED) | |
| _ARCH_WIDE_SOURCE_ROWS = [ | |
| ("U driver / GRU, d 128 / 0.16 M", "model_spec.md:25 U: GRU(18->128) last state h_u [B,128]; params 159,290 -> 0.16 M"), | |
| ("V vehicle / causal TCN / 4 dilated blocks / 1.61 M", "model_spec.md:24 V: TCN(6->256, k=5, dil 1/2/4/8); params 1,608,347 -> 1.61 M"), | |
| ("E road / 2-layer causal / Transformer / d 128 . 0.88 M", "model_spec.md:26 E pooled: 2x causal TransformerEncoderLayer(d128, 4 heads, FFN 256); params 880,604 -> 0.88 M"), | |
| ("road->vehicle / MLP . 0.33 M", "model_spec.md:36 Env2Vehicle [z_e ; z_vf] -> residual on z_v; params 329,088 -> 0.33 M"), | |
| ("driver->vehicle / regime-gated / MLP . 0.25 M", "model_spec.md:37 Driver2Vehicle h_u -> residual on z_v, sigma(Embedding(mode,256)) gate; params 247,680 -> 0.25 M"), | |
| ("zero-init residual MLPs into the vehicle state", "model_spec.md:30 (zero-init -> bitwise identity at init), :41 (forward order: z_v += env2veh; z_v += drv2veh(h_u))"), | |
| ("dashed + x = ablated road->driver path", "model_spec.md:34-35 Road2Driver 231,424 + RegionTokenEmbed 11,392 (NOT drawn as a module); main model = all_minus_road2drv: docs/results_cwm3.md:9"), | |
| ("K future kinematics 0.09 M (dotted = detached)", "model_spec.md:46-47 k_head on zU.detach(): Delta-K mu / log sigma; params 89,510 -> 0.09 M"), | |
| ("R hard response 0.04 M (dotted = detached)", "model_spec.md:54 r_head on zV.detach(): hard-response chain R >= 2; params 41,394 -> 0.04 M"), | |
| ("P road pressure 0.15 M (dotted = detached)", "model_spec.md:53 p_head on zE.detach(): six road-pressure chains; params 146,988 -> 0.15 M"), | |
| ("CockpitWM -- 3.50 M trainable parameters", "DERIVED (per the figure brief): 2,926,133 (model_spec.md:72, 'I1 none / indep' total) + 329,088 + 247,680 (= 576,768; :36-37) = 3,502,901 -> 3.50 M; not printed in model_spec.md"), | |
| ("external frozen models not counted", "model_spec.md:4 (every count is the models/cwm3.py param_report of the assembled model; YOLO11x-pose, RTMW-x-l and V-JEPA2 are not modules of it)"), | |
| ("cabin video . t-4 s ... t -> YOLO11x-pose causal box -> RTMW-x-l HALPE-136 keypoints (frozen)", "docs/decisions.md:143 (YOLO11x-pose per 10 Hz frame -> causal driver box), :140 (RTMW-x-l @10 Hz, HALPE-136); window 4 s @10 Hz: model_spec.md:13 (Kx [B,40,18] @10 Hz), :12 (t-3.9..t)"), | |
| ("kinematics 11 ch @10 Hz", "model_spec.md:13 Kx: 11 pose kinematics (standardised) + valid bit + 5 CAN + dt @10 Hz"), | |
| ("CAN bus, 5 whitelisted / signals: steering angle, / gas, brake, vEgo, aEgo / 4 s @10 Hz (t-4 s ... t)", "model_spec.md:12 X [B,40,6] CAN history + dt, can4s = t-3.9..t @10 Hz; column names data/build_can8.py:6 [steeringAngleDeg, gasPressed, brakePressed, vEgo, aEgo]"), | |
| ("road video . t-8 s ... t -> V-JEPA2 ViT-g (frozen) / clip-mean token", "model_spec.md:14 (V-JEPA2 tokens per 0.5 s slot over 8 s, causal bank), :26 (pooled: one PCA-256 clip token per 0.5 s slot); ViT-g: docs/latent_quality.md:4, :256 (facebook/vjepa2-vitg-fpc64-384)"), | |
| ("PCA-256, 16 slots @2 Hz / + front latent at t", "model_spec.md:26 (one PCA-256 clip token per 0.5 s slot, 16 tokens -> 2 Hz), :15 (vf [B,256] front PCA-256 at the anchor)"), | |
| ("regime m (pill under every branch); legend: regime m in {manual, assistance engaged}", "model_spec.md:8 (mode in {0 manual, 1 engaged} embedded in every branch), :16"), | |
| ("driver chains / hands-on / head-turned / -> onsets: hand_leave, / hand_return, head_turn", "model_spec.md:25 (hands_on / head_turned chains); onset names models/heads.py:11-12 (hand_return = hands_on.on, hand_leave = hands_on.off, head_turn_onset = head_turned.on)"), | |
| ("vehicle chains / gas, brake, steer-phase, / response; MDN: steering, / speed, acceleration", "model_spec.md:24 (3 MDNs + gas/brake/steer_phase/resp chains); MDN targets models/heads.py:4-5 (steeringAngleDeg, vEgo, aEgo)"), | |
| ("road demand + pressure / chains: TTC, TTC drop, / cut-in, lead braking, / lane excursion, BADAS", "model_spec.md:26 (6 demand chains), :53 (P chains ttc, ttc_drop, cut_in, lead_braking, lane_excursion, badas); demand chain names models/heads.py:9 (steer_rate, decel, ttc, ttc_drop, lane, badas) -- the box lists the pressure names"), | |
| ("prediction (25 x 0.2 s = 5 s)", "model_spec.md:62 (DirectDecoder25 = flat 25-step head), models/heads.py:22 (H = 25); 5 s horizon model_spec.md:94 (K-hat @1/3/5 s); 0.2 s/step DERIVED = 5 s / 25"), | |
| ("warning readout W / HGB . exploratory / inputs at anchor t: current state of the frozen model (no predicted quantity)", "results/tables/warning_head_final_ext.md header (feature sets current = no predicted quantity; K0 from the causal driver box); HGB: train/train_warning_head.py:18, :268; 'fixed HGB head' results/tables/warning_head_final.md:1034"), | |
| ("+ dwell block: 8 s kinematics history", "train/warning_features_ext.py:7-8 (dwell (11) from Kh8_c [N,80,11] t-7.9..t @10 Hz); warning_head_final_ext.md header (n979dw = current + dwell / duration block, 11 dims)"), | |
| ("+ face block: HALPE-136 face landmarks", "train/warning_features_ext.py:19-23 (face (16), our own RTMW HALPE-136 face points); warning_head_final_ext.md header (n979dwfc, 16 dims)"), | |
| ("openpilot DMS chain -- evaluation comparator only, never an input; compared (AUROC)", "results/tables/warning_head_final_ext.md header (AUROC primary, evaluator warning_gold.py); models/heads.py:13 and train/warning_features_ext.py:2-3 (no openpilot DMS field is read anywhere)"), | |
| ] | |
| def _write_arch_wide_sidecar(out_pdf: str, spec_sha: str, whx_sha: str, w_in: float, h_in: float) -> None: | |
| with open(out_pdf[:-4] + ".txt", "w", encoding="utf-8") as f: | |
| f.write(f"# {os.path.basename(out_pdf)} -- generated {_now()} by paper/figs/make_figs.py (layout wide / ICLR, " | |
| f"{w_in:.2f} x {h_in:.2f} in)\n") | |
| f.write(f"# source A: {SRC_SPEC} (sha256 {spec_sha})\n# source B: {SRC_WHX} (sha256 {whx_sha})\n") | |
| f.write("# palette: cwLav #DDD6E5, cwLavL #DFEDF2, cwGreyL #BFDCE7, cwPeachL #F5D8BF, cwPeach #D9E4C2, cwCoral #CAC2D7, " | |
| "cwBg #F9EDE0 + greys derived from black (#000000, #404040, #6E6E6E, #8C8C8C, #D9D9D9, #F2F2F2)\n") | |
| f.write("# every printed label -> source line (file:line). Rounded 'x.xx M' values are printed; exact counts listed here.\n") | |
| f.write("# DERIVED quantities (only two): the 3.50 M total and the 0.2 s step; both marked below.\n") | |
| for label, src in _ARCH_WIDE_SOURCE_ROWS: | |
| f.write(f"- {label}\n <- {src}\n") | |
| def make_architecture_wide(out_pdf: str) -> None: | |
| """Landscape CockpitWM block diagram for the ICLR single column (5.5 in text width), read left to right | |
| (past -> future): observed inputs + frozen external perception | three separately pre-trained trainable | |
| branches (U driver top, V vehicle middle, E road bottom) | zero-initialised residual interaction into the | |
| vehicle state | direct 25-step prediction heads (+ small detached heads K / R / P) | warning readout W | |
| (separate trainable box, reads the anchor-time state only) with the openpilot DMS chain drawn outside the | |
| model as an evaluation comparator. Colours: CockpitWM palette only (CW dict + greys). Every number is | |
| copied from docs/model_spec.md (see _ARCH_WIDE_SOURCE_ROWS); 3.50 M is the one derived total.""" | |
| W_IN, H_IN = 5.5, 2.94 | |
| FS = 6.5 # minimum readable size at the native 5.5 in width; nothing in the figure is smaller | |
| _BOXED.clear() | |
| fig = plt.figure(figsize=(W_IN, H_IN)) | |
| ax = fig.add_axes([0, 0, 1, 1]) | |
| ax.set_xlim(0, W_IN) | |
| ax.set_ylim(0, H_IN) | |
| ax.set_aspect("equal") | |
| ax.axis("off") | |
| # ---- columns (x, width) and rows (centre y) in inches ------------------------------------------------ | |
| x_L, w_L = 0.03, 1.20 # observed inputs + frozen perception cards | |
| x_B, w_B = 1.37, 0.78 # branches | |
| x_I, w_I = 2.22, 0.98 # interaction | |
| x_O, w_O = 3.30, 1.14 # predictions | |
| x_W, w_W = 4.60, 0.87 # warning readout (outside the model frame) | |
| pitch = 0.76 | |
| yr = {"E": 0.73, "V": 0.73 + pitch, "U": 0.73 + 2 * pitch} | |
| xF0, xF1 = x_B - 0.06, x_O + w_O + 0.05 # model frame | |
| yF0, yF1 = 0.25, H_IN - 0.04 | |
| h_B, h_O, h_D = 0.48, 0.48, 0.15 # branch box, prediction box, detached head box | |
| LS = 1.0 | |
| def box(x, y, w, h, text, fc, ec=CW_INK2, ls="-", lw=0.7, weight="normal", fs=FS, color=CW_BLACK, **kw): | |
| return _box(ax, x, y, w, h, text, fc, ec, fs=fs, lw=lw, color=color, ls=ls, rounding=0.035, weight=weight, | |
| linespacing=LS, **kw) | |
| # ---- model frame + title --------------------------------------------------------------------------------- | |
| box(xF0, yF0, xF1 - xF0, yF1 - yF0, "", CW["bg"], ec=CW_HAIR, lw=0.6, zorder=1) | |
| ax.text(xF0 + 0.07, yF1 - 0.07, "CockpitWM — 3.50 M trainable parameters (external frozen models not counted)", | |
| ha="left", va="center", fontsize=FS, color=CW_BLACK, fontweight="bold", zorder=3) | |
| # ---- left column: observed inputs (cwLavL) and frozen external perception (cwGreyL, grey border) -------- | |
| def card(k, tab, frozen_lines, feat_lines, h_frozen, h_feat): | |
| y = yr[k] | |
| top = y + 0.36 | |
| y_tab0 = top - 0.14 | |
| box(x_L, y_tab0, w_L, 0.14, tab, CW["lavL"]) | |
| y_fr1 = y_tab0 - 0.03 | |
| y_fr0 = y_fr1 - h_frozen | |
| _tri_down(ax, x_L + w_L / 2, (y_fr1 + y_tab0) / 2) | |
| box(x_L, y_fr0, w_L, h_frozen, frozen_lines, CW["greyL"], ec=CW_FROZEN, lw=0.8) | |
| y_ft1 = y_fr0 - 0.03 | |
| y_ft0 = y_ft1 - h_feat | |
| _tri_down(ax, x_L + w_L / 2, (y_ft1 + y_fr0) / 2) | |
| box(x_L, y_ft0, w_L, h_feat, feat_lines, CW["lavL"]) | |
| # snowflake in the gap left inside the '(frozen)' line: the last frozen line is 'A (frozen)' | |
| lines = frozen_lines.splitlines() | |
| j = next(i for i, l in enumerate(lines) if " " in l) | |
| left, right = lines[j].split(" ") | |
| wl, _ = _measure(fig, left, FS) | |
| wr, _ = _measure(fig, right, FS) | |
| wfull, _ = _measure(fig, lines[j], FS) | |
| line_h = FS * 1.2 * LS / 72.0 | |
| y_line = (y_fr0 + y_fr1) / 2 + ((len(lines) - 1) / 2 - j) * line_h | |
| x_gap = x_L + w_L / 2 - wfull / 2 + wl + (wfull - wl - wr) / 2 | |
| _snowflake(ax, x_gap, y_line, r=0.03) | |
| # elbow from the features tier (its centre) into the branch (row centre) | |
| _elbow(ax, (x_L + w_L, (y_ft0 + y_ft1) / 2), (x_B, y), x_L + w_L + 0.055) | |
| return y_ft0, top | |
| card("U", "cabin video · t−4 s … t", | |
| "YOLO11x-pose causal box\n→ RTMW-x-l HALPE-136\nkeypoints (frozen)", | |
| "kinematics 11 ch @10 Hz", 0.36, 0.14) | |
| card("E", "road video · t−8 s … t", | |
| "V-JEPA2 ViT-g (frozen)\nclip-mean token", | |
| "PCA-256, 16 slots @2 Hz\n+ front latent at t", 0.26, 0.26) | |
| # vehicle row: the CAN bus is observed directly (no frozen model) | |
| box(x_L, yr["V"] - 0.24, w_L, 0.48, | |
| "CAN bus, 5 whitelisted\nsignals: steering angle,\ngas, brake, vEgo, aEgo\n4 s @10 Hz (t−4 s … t)", CW["lavL"]) | |
| _arrow(ax, (x_L + w_L, yr["V"]), (x_B, yr["V"]), color=CW_INK2) | |
| # ---- branches (trainable, pre-trained independently) + regime pills ------------------------------------ | |
| branch = { | |
| "U": "GRU, d 128\n0.16 M", | |
| "V": "causal TCN\n4 dilated blocks\n1.61 M", | |
| "E": "2-layer causal\nTransformer\nd 128 · 0.88 M", | |
| } | |
| title = {"U": "U driver", "V": "V vehicle", "E": "E road"} | |
| for k in "UVE": | |
| y = yr[k] | |
| box(x_B, y - h_B / 2, w_B, h_B, "", CW["peach"]) | |
| t = ax.text(x_B + w_B / 2, y + h_B / 2 - 0.045, title[k], ha="center", va="top", fontsize=FS, | |
| fontweight="bold", color=CW_BLACK, zorder=3) | |
| _BOXED.append((t, w_B, 0.14)) | |
| body_h = h_B - 0.16 | |
| t = ax.text(x_B + w_B / 2, y - h_B / 2 + 0.02 + body_h / 2, branch[k], ha="center", va="center", fontsize=FS, | |
| color=CW_BLACK, linespacing=LS, zorder=3) | |
| _BOXED.append((t, w_B, body_h)) | |
| # regime token: small observed input under every branch | |
| pw, ph = 0.48, 0.13 | |
| px = x_B + w_B / 2 - pw / 2 | |
| py = y - h_B / 2 - 0.08 - ph | |
| box(px, py, pw, ph, "regime m", CW["lavL"], lw=0.5) | |
| _arrow(ax, (x_B + w_B / 2, py + ph), (x_B + w_B / 2, y - h_B / 2), color=CW_INK2, head=4) | |
| # main line branch -> prediction box (through the interaction column) | |
| _arrow(ax, (x_B + w_B, y), (x_O, y), color=CW_INK2, lw=0.8) | |
| # ---- interaction: zero-init residual MLPs into z_V; ablated road->driver path ---------------------------- | |
| x_M, w_M = x_I + 0.28, 0.70 | |
| x_c = x_M + w_M / 2 | |
| h_M = 0.37 | |
| y_dv = (yr["U"] + yr["V"]) / 2 | |
| y_rv = (yr["V"] + yr["E"]) / 2 | |
| box(x_M, y_dv - h_M / 2, w_M, h_M, "driver→vehicle\nregime-gated\nMLP · 0.25 M", CW["peach"]) | |
| box(x_M, y_rv - h_M / 2, w_M, h_M, "road→vehicle\nMLP · 0.33 M", CW["peach"]) | |
| _oplus(ax, x_c, yr["V"]) | |
| _dot(ax, x_c, yr["U"]) | |
| _arrow(ax, (x_c, yr["U"]), (x_c, y_dv + h_M / 2), color=CW_INK2, shrinkA=0) | |
| _arrow(ax, (x_c, y_dv - h_M / 2), (x_c, yr["V"] + 0.045), color=CW_INK2, shrinkA=0, shrinkB=0) | |
| _dot(ax, x_c, yr["E"]) | |
| _arrow(ax, (x_c, yr["E"]), (x_c, y_rv - h_M / 2), color=CW_INK2, shrinkA=0) | |
| _arrow(ax, (x_c, y_rv + h_M / 2), (x_c, yr["V"] - 0.045), color=CW_INK2, shrinkA=0, shrinkB=0) | |
| ax.text(x_I + w_I / 2, yr["E"] - 0.19, "zero-init residual MLPs\ninto the vehicle state", ha="center", va="center", | |
| fontsize=FS, color=CW_INK2, linespacing=LS, zorder=3) | |
| # ablated road->driver: dashed grey, crossed out, labelled | |
| x_a = x_I + 0.10 | |
| _dot(ax, x_a, yr["E"], color=CW_FROZEN) | |
| ax.plot([x_a, x_a], [yr["E"], yr["U"] - 0.09], color=CW_FROZEN, lw=0.8, ls=(0, (2.2, 1.4)), zorder=4) | |
| _arrow(ax, (x_a, yr["U"] - 0.09), (x_a, yr["U"]), color=CW_FROZEN, lw=0.8, shrinkA=0, shrinkB=0.5) # solid head | |
| s = 0.035 | |
| ax.plot([x_a - s, x_a + s], [y_dv - s, y_dv + s], color=CW_BLACK, lw=1.0, zorder=7) | |
| ax.plot([x_a - s, x_a + s], [y_dv + s, y_dv - s], color=CW_BLACK, lw=1.0, zorder=7) | |
| ax.text(x_a + 0.09, y_dv, "ablated", ha="center", va="center", rotation=90, fontsize=FS, color=CW_INK2, zorder=3) | |
| # ---- predictions (cwCoral) + small detached heads (cwLav, dotted) ------------------------------------------- | |
| pred_title = {"U": "driver chains", "V": "vehicle chains", "E": "road demand + pressure"} | |
| pred_body = { | |
| "U": "hands-on / head-turned\n→ onsets: hand_leave,\nhand_return, head_turn", | |
| "V": "gas, brake, steer-phase,\nresponse; MDN: steering,\nspeed, acceleration", | |
| "E": "chains: TTC, TTC drop,\ncut-in, lead braking,\nlane excursion, BADAS", | |
| } | |
| detached = {"U": "K future kinematics 0.09 M", "V": "R hard response 0.04 M", "E": "P road pressure 0.15 M"} | |
| for k in "UVE": | |
| y = yr[k] | |
| y0 = y - 0.12 | |
| box(x_O, y0, w_O, h_O, "", CW["coral"]) | |
| t = ax.text(x_O + w_O / 2, y0 + h_O - 0.045, pred_title[k], ha="center", va="top", fontsize=FS, | |
| fontweight="bold", color=CW_BLACK, zorder=3) | |
| _BOXED.append((t, w_O, 0.14)) | |
| body_h = h_O - 0.16 | |
| t = ax.text(x_O + w_O / 2, y0 + 0.02 + body_h / 2, pred_body[k], ha="center", va="center", fontsize=FS, | |
| color=CW_BLACK, linespacing=LS, zorder=3) | |
| _BOXED.append((t, w_O, body_h)) | |
| yd1 = y0 - 0.04 | |
| yd0 = yd1 - h_D | |
| box(x_O, yd0, w_O, h_D, detached[k], CW["lav"], ls=(0, (1.2, 1.0)), lw=0.6) | |
| xj = x_O - 0.09 | |
| _dot(ax, xj, y) | |
| _elbow(ax, (xj, y), (x_O, (yd0 + yd1) / 2), xj, ls=(0, (1.2, 1.0)), head=4) | |
| # ---- warning readout W (separate trainable box) + external DMS comparator --------------------------------- | |
| yW1 = H_IN - 0.16 | |
| yW0 = yr["V"] - 0.33 | |
| box(x_W, yW0, w_W, yW1 - yW0, "", CW["peach"]) | |
| t = ax.text(x_W + w_W / 2, yW1 - 0.045, "warning readout W", ha="center", va="top", fontsize=FS, color=CW_BLACK, | |
| fontweight="bold", zorder=3) | |
| _BOXED.append((t, w_W, 0.14)) | |
| w_body = ("HGB · exploratory\n\ninputs at anchor t:\ncurrent state of the\nfrozen model (no\npredicted quantity)\n" | |
| "+ dwell block: 8 s\nkinematics history\n+ face block:\nHALPE-136 face\nlandmarks\n\n→ warning score") | |
| t = ax.text(x_W + w_W / 2, yW0 + (yW1 - 0.16 - yW0) / 2, w_body, ha="center", va="center", fontsize=FS, | |
| color=CW_BLACK, linespacing=LS, zorder=3) | |
| _BOXED.append((t, w_W, yW1 - 0.16 - yW0)) | |
| # the readout reads the anchor-time state: arrow leaves the frame at the title strip, not from any prediction box | |
| y_arrow = yW1 - 0.09 | |
| ax.text(xF1 - 0.05, y_arrow, "anchor-time state", ha="right", va="center", fontsize=FS, color=CW_INK2, zorder=3) | |
| _arrow(ax, (xF1, y_arrow), (x_W, y_arrow), color=CW_INK2, lw=0.8, shrinkA=0) | |
| # openpilot DMS chain: outside the model, grey, evaluation comparator only | |
| h_dms = 0.48 | |
| yD1 = yr["E"] + 0.10 | |
| yD0 = yD1 - h_dms | |
| box(x_W, yD0, w_W, h_dms, "openpilot DMS chain\nevaluation\ncomparator only,\nnever an input", CW_DMS, ec=CW_FROZEN, | |
| lw=0.8, color=CW_INK2) | |
| xm = x_W + w_W / 2 | |
| ym = (yD1 + yW0) / 2 | |
| ax.plot([xm, xm], [yD1, ym - 0.065], color=CW_FROZEN, lw=0.7, ls=(0, (2.2, 1.4)), zorder=4) | |
| ax.plot([xm, xm], [ym + 0.065, yW0], color=CW_FROZEN, lw=0.7, ls=(0, (2.2, 1.4)), zorder=4) | |
| ax.text(xm, ym, "compared (AUROC)", ha="center", va="center", fontsize=FS, color=CW_INK2, zorder=3) | |
| # ---- legend (bottom strip): swatches laid out from measured text widths ------------------------------------ | |
| y_leg1, y_leg2 = 0.165, 0.055 | |
| items = [ | |
| (CW["lavL"], CW_INK2, "-", "input"), | |
| (CW["greyL"], CW_FROZEN, "-", "frozen external model (not counted)"), | |
| (CW["peach"], CW_INK2, "-", "trainable"), | |
| (CW["lav"], CW_INK2, (0, (1.2, 1.0)), "detached head"), | |
| (CW["coral"], CW_INK2, "-", "prediction (25 × 0.2 s = 5 s)"), | |
| ] | |
| sw_w, sw_h, gap_sw, gap_item = 0.16, 0.09, 0.04, 0.16 | |
| widths = [_measure(fig, lab, FS)[0] for *_, lab in items] | |
| total = sum(widths) + len(items) * (sw_w + gap_sw) + (len(items) - 1) * gap_item | |
| x = (W_IN - total) / 2 | |
| for (fc, ec, ls, lab), wtxt in zip(items, widths): | |
| ax.add_patch(FancyBboxPatch((x, y_leg1 - sw_h / 2), sw_w, sw_h, boxstyle="round,pad=0,rounding_size=0.015", | |
| fc=fc, ec=ec, lw=0.6, linestyle=ls, zorder=3)) | |
| if lab.startswith("frozen"): | |
| _snowflake(ax, x + sw_w / 2, y_leg1, r=0.026) | |
| ax.text(x + sw_w + gap_sw, y_leg1, lab, ha="left", va="center", fontsize=FS, color=CW_BLACK, zorder=3) | |
| x += sw_w + gap_sw + wtxt + gap_item | |
| leg2 = ("dashed grey + × = ablated road→driver path regime m ∈ {manual, assistance engaged} " | |
| "W reads no predicted quantity") | |
| t = ax.text(W_IN / 2, y_leg2, leg2, ha="center", va="center", fontsize=FS, color=CW_INK2, zorder=3) | |
| _BOXED.append((t, W_IN - 0.06, 0.14)) | |
| spec_sha, whx_sha = _sha256(SRC_SPEC), _sha256(SRC_WHX) | |
| meta = { | |
| "Title": "CockpitWM architecture (landscape, ICLR single column, CockpitWM palette)", | |
| "Subject": f"sources: {SRC_SPEC} S1-S5 (sha256 {spec_sha}); {SRC_WHX} header (sha256 {whx_sha}); rendered {_now()} by make_figs.py", | |
| "Keywords": "CockpitWM, model_spec.md, parameter counts verbatim, 3.50 M derived total, landscape", | |
| "Creator": "paper/figs/make_figs.py (matplotlib)", | |
| } | |
| _check_overflow(fig) | |
| fig.savefig(out_pdf, metadata=meta) | |
| plt.close(fig) | |
| _write_arch_wide_sidecar(out_pdf, spec_sha, whx_sha, W_IN, H_IN) | |
| print(f"wrote {out_pdf} (wide, {W_IN:.2f} x {H_IN:.2f} in) and sidecar") | |
| def _write_arch_clean_sidecar(out_pdf: str, spec_sha: str, whx_sha: str, w_in: float, h_in: float, printed: list) -> None: | |
| with open(out_pdf[:-4] + ".txt", "w", encoding="utf-8") as f: | |
| f.write(f"# {os.path.basename(out_pdf)} -- generated {_now()} by paper/figs/make_figs.py (layout clean / ICLR, " | |
| f"{w_in:.2f} x {h_in:.2f} in; owner request 2026-09-16: clearer, less text)\n") | |
| f.write(f"# source A: {SRC_SPEC} (sha256 {spec_sha})\n# source B: {SRC_WHX} (sha256 {whx_sha})\n") | |
| f.write("# palette: cwLav #DDD6E5, cwLavL #DFEDF2, cwGreyL #BFDCE7, cwPeach #D9E4C2, cwCoral #CAC2D7, cwBg #F9EDE0 " | |
| "+ greys derived from black (#000000, #404040, #6E6E6E, #8C8C8C, #D9D9D9, #F2F2F2)\n") | |
| f.write("# printed numbers (rounded 'x.xx M') -> source line; the 3.50 M total is the one DERIVED quantity.\n") | |
| for label, src in _ARCH_WIDE_SOURCE_ROWS: | |
| if any(tok in label for tok in printed): | |
| f.write(f"- {label}\n <- {src}\n") | |
| f.write("# omitted from this layout (still in the wide layout and in docs/model_spec.md): perception model names, CAN column " | |
| "names, chain lists, detached-head parameter counts, dwell / face block dimensions.\n") | |
| def make_architecture_clean(out_pdf: str) -> None: | |
| """Clean landscape CockpitWM block diagram (owner request 2026-09-16: clearer, less text). Same content | |
| order as the wide layout -- observed inputs + frozen perception | three trainable branches | zero-init | |
| residual interaction into the vehicle state | direct 5-s hazard heads (+ small auxiliary heads) | warning | |
| read-out with the openpilot DMS chain as an external comparator -- but every box carries one or two short | |
| lines, perception models and chain lists are named in the caption instead, and the type is 7 pt.""" | |
| W_IN, H_IN = 5.5, 2.58 | |
| FS, FT = 7.0, 7.5 | |
| _BOXED.clear() | |
| fig = plt.figure(figsize=(W_IN, H_IN)) | |
| ax = fig.add_axes([0, 0, 1, 1]) | |
| ax.set_xlim(0, W_IN) | |
| ax.set_ylim(0, H_IN) | |
| ax.set_aspect("equal") | |
| ax.axis("off") | |
| LS = 1.05 | |
| def box(x, y, w, h, text, fc, ec=CW_INK2, ls="-", lw=0.7, weight="normal", fs=FS, color=CW_BLACK, **kw): | |
| return _box(ax, x, y, w, h, text, fc, ec, fs=fs, lw=lw, color=color, ls=ls, rounding=0.04, weight=weight, | |
| linespacing=LS, **kw) | |
| def titled(x, y, w, h, title, body, fc, ec=CW_INK2, ls="-", lw=0.7, body_color=CW_BLACK): | |
| box(x, y, w, h, "", fc, ec=ec, ls=ls, lw=lw) | |
| t = ax.text(x + w / 2, y + h - 0.05, title, ha="center", va="top", fontsize=FT, fontweight="bold", | |
| color=CW_BLACK, zorder=3) | |
| _BOXED.append((t, w, 0.16)) | |
| if body: | |
| bh = h - 0.19 | |
| t = ax.text(x + w / 2, y + 0.02 + bh / 2, body, ha="center", va="center", fontsize=FS, color=body_color, | |
| linespacing=LS, zorder=3) | |
| _BOXED.append((t, w, bh)) | |
| # ---- geometry (inches) ----------------------------------------------------------------------------------- | |
| x_L, w_L = 0.04, 1.04 # inputs + frozen perception | |
| x_B, w_B = 1.26, 0.96 # branches | |
| x_I, w_I = 2.30, 0.92 # interaction | |
| x_O, w_O = 3.30, 1.04 # predictions | |
| x_W, w_W = 4.48, 0.98 # warning read-out (outside the frame) | |
| pitch = 0.70 | |
| yr = {"E": 0.66, "V": 0.66 + pitch, "U": 0.66 + 2 * pitch} | |
| xF0, xF1 = x_B - 0.07, x_O + w_O + 0.06 | |
| yF0, yF1 = 0.30, H_IN - 0.04 | |
| h_B, h_O = 0.46, 0.46 | |
| # ---- frame + title ----------------------------------------------------------------------------------------- | |
| box(xF0, yF0, xF1 - xF0, yF1 - yF0, "", CW["bg"], ec=CW_HAIR, lw=0.6, zorder=1) | |
| ax.text(xF0 + 0.08, yF1 - 0.08, "CockpitWM · 3.50 M trainable parameters", ha="left", va="center", fontsize=FT, | |
| fontweight="bold", color=CW_BLACK, zorder=3) | |
| # ---- inputs: three cards, one short line per tier ----------------------------------------------------------- | |
| def frozen_box(x, y, w, h, label): | |
| box(x, y, w, h, "", CW["greyL"], ec=CW_FROZEN, lw=0.8) | |
| tw, _ = _measure(fig, label, FS) | |
| cx = x + w / 2 - 0.06 | |
| t = ax.text(cx, y + h / 2, label, ha="center", va="center", fontsize=FS, color=CW_BLACK, zorder=3) | |
| _BOXED.append((t, w - 0.16, h)) | |
| _snowflake(ax, cx + tw / 2 + 0.08, y + h / 2, r=0.03) | |
| def card(k, tiers): | |
| y = yr[k] | |
| n = len(tiers) | |
| h_t, gap = 0.15, 0.035 | |
| total = n * h_t + (n - 1) * gap | |
| top = y + total / 2 | |
| for i, (kind, label) in enumerate(tiers): | |
| y1 = top - i * (h_t + gap) | |
| y0 = y1 - h_t | |
| if kind == "frozen": | |
| frozen_box(x_L, y0, w_L, h_t, label) | |
| else: | |
| box(x_L, y0, w_L, h_t, label, CW["lavL"]) | |
| if i < n - 1: | |
| _tri_down(ax, x_L + w_L / 2, y0 - gap / 2, w=0.045, h=0.025) | |
| # last tier -> branch | |
| y_last = top - (n - 1) * (h_t + gap) - h_t / 2 | |
| if abs(y_last - y) < 1e-6: | |
| _arrow(ax, (x_L + w_L, y), (x_B, y), color=CW_INK2) | |
| else: | |
| _elbow(ax, (x_L + w_L, y_last), (x_B, y), x_L + w_L + 0.07) | |
| card("U", [("in", "Cabin video · 4 s"), ("frozen", "Pose keypoints"), ("in", "Driver kinematics")]) | |
| card("V", [("in", "Vehicle bus · 4 s"), ("in", "5 CAN signals")]) | |
| card("E", [("in", "Road video · 8 s"), ("frozen", "Video encoder"), ("in", "Road tokens · 2 Hz")]) | |
| # ---- branches --------------------------------------------------------------------------------------------------- | |
| branch = {"U": ("Driver", "GRU\n0.16 M"), "V": ("Vehicle", "causal TCN\n1.61 M"), | |
| "E": ("Road", "causal Transformer\n0.88 M")} | |
| for k in "UVE": | |
| y = yr[k] | |
| titled(x_B, y - h_B / 2, w_B, h_B, branch[k][0], branch[k][1], CW["peach"], body_color=CW_INK2) | |
| # regime token: one small pill on the upper-right corner of every branch | |
| pw, ph = 0.17, 0.12 | |
| box(x_B + w_B - pw + 0.04, y + h_B / 2 - ph / 2, pw, ph, "m", CW["lavL"], lw=0.5, fs=FS) | |
| _arrow(ax, (x_B + w_B, y), (x_O, y), color=CW_INK2, lw=0.8) | |
| # ---- interaction: zero-init residual MLPs into the vehicle state; dropped road -> driver path ----------------- | |
| x_M, w_M, h_M = x_I + 0.17, 0.76, 0.30 | |
| x_c = x_M + w_M / 2 | |
| y_dv = (yr["U"] + yr["V"]) / 2 | |
| y_rv = (yr["V"] + yr["E"]) / 2 | |
| box(x_M, y_dv - h_M / 2, w_M, h_M, "driver → vehicle\nMLP · 0.25 M", CW["peach"]) | |
| box(x_M, y_rv - h_M / 2, w_M, h_M, "road → vehicle\nMLP · 0.33 M", CW["peach"]) | |
| ax.text(x_I + w_I / 2 + 0.03, yr["E"] - 0.20, "zero-init residuals", ha="center", va="center", fontsize=FS, | |
| color=CW_INK2, zorder=3) | |
| _oplus(ax, x_c, yr["V"]) | |
| _dot(ax, x_c, yr["U"]) | |
| _arrow(ax, (x_c, yr["U"]), (x_c, y_dv + h_M / 2), color=CW_INK2, shrinkA=0) | |
| _arrow(ax, (x_c, y_dv - h_M / 2), (x_c, yr["V"] + 0.045), color=CW_INK2, shrinkA=0, shrinkB=0) | |
| _dot(ax, x_c, yr["E"]) | |
| _arrow(ax, (x_c, yr["E"]), (x_c, y_rv - h_M / 2), color=CW_INK2, shrinkA=0) | |
| _arrow(ax, (x_c, y_rv + h_M / 2), (x_c, yr["V"] - 0.045), color=CW_INK2, shrinkA=0, shrinkB=0) | |
| x_a = x_I + 0.06 | |
| _dot(ax, x_a, yr["E"], color=CW_FROZEN) | |
| ax.plot([x_a, x_a], [yr["E"], yr["U"] - 0.09], color=CW_FROZEN, lw=0.8, ls=(0, (2.2, 1.4)), zorder=4) | |
| _arrow(ax, (x_a, yr["U"] - 0.09), (x_a, yr["U"]), color=CW_FROZEN, lw=0.8, shrinkA=0, shrinkB=0.5) | |
| sx = 0.035 | |
| ax.plot([x_a - sx, x_a + sx], [y_dv - sx, y_dv + sx], color=CW_BLACK, lw=1.0, zorder=7) | |
| ax.plot([x_a - sx, x_a + sx], [y_dv + sx, y_dv - sx], color=CW_BLACK, lw=1.0, zorder=7) | |
| ax.text(x_a - 0.075, y_dv, "dropped", ha="center", va="center", rotation=90, fontsize=FS, color=CW_INK2, zorder=3) | |
| # ---- predictions: 25-step hazards, one short body line; small auxiliary heads ------------------------------------ | |
| pred = {"U": ("Driver hazards", "hands, head turn"), "V": ("Vehicle hazards", "gas, brake, steering"), | |
| "E": ("Road hazards", "TTC, cut-in, lane")} | |
| aux = {"U": "aux. head K", "V": "aux. head R", "E": "aux. head P"} | |
| for k in "UVE": | |
| y = yr[k] | |
| y0 = y - 0.10 | |
| titled(x_O, y0, w_O, h_O, pred[k][0], pred[k][1], CW["coral"]) | |
| yd1, h_D = y0 - 0.04, 0.13 | |
| yd0 = yd1 - h_D | |
| box(x_O + 0.18, yd0, w_O - 0.36, h_D, aux[k], CW["lav"], ls=(0, (1.2, 1.0)), lw=0.6, color=CW_INK2) | |
| xj = x_O - 0.09 | |
| _dot(ax, xj, y) | |
| _elbow(ax, (xj, y), (x_O + 0.18, (yd0 + yd1) / 2), xj, ls=(0, (1.2, 1.0)), head=4) | |
| # ---- warning read-out + external DMS comparator --------------------------------------------------------------- | |
| yW1 = H_IN - 0.06 | |
| yW0 = yr["V"] - 0.20 | |
| titled(x_W, yW0, w_W, yW1 - yW0, "Warning read-out", "reads the state at t\n\ncurrent state\n+ dwell block\n+ face block\n\n→ warning score", | |
| CW["peach"]) | |
| y_arrow = yF1 - 0.08 | |
| _arrow(ax, (xF1, y_arrow), (x_W, y_arrow), color=CW_INK2, lw=0.8, shrinkA=0) | |
| h_dms = 0.40 | |
| yD1 = yr["E"] + 0.12 | |
| yD0 = yD1 - h_dms | |
| box(x_W, yD0, w_W, h_dms, "openpilot DMS chain\ncomparison only", CW_DMS, ec=CW_FROZEN, lw=0.8, color=CW_INK2) | |
| xm = x_W + w_W / 2 | |
| ym = (yD1 + yW0) / 2 | |
| ax.plot([xm, xm], [yD1, ym - 0.06], color=CW_FROZEN, lw=0.7, ls=(0, (2.2, 1.4)), zorder=4) | |
| ax.plot([xm, xm], [ym + 0.06, yW0], color=CW_FROZEN, lw=0.7, ls=(0, (2.2, 1.4)), zorder=4) | |
| ax.text(xm, ym, "AUROC", ha="center", va="center", fontsize=FS, color=CW_INK2, zorder=3) | |
| # ---- legend -------------------------------------------------------------------------------------------------- | |
| y_leg1, y_leg2 = 0.19, 0.07 | |
| items = [(CW["lavL"], CW_INK2, "-", "input"), (CW["greyL"], CW_FROZEN, "-", "frozen model"), | |
| (CW["peach"], CW_INK2, "-", "trainable"), (CW["coral"], CW_INK2, "-", "prediction · 5 s ahead, 25 steps"), | |
| (CW["lav"], CW_INK2, (0, (1.2, 1.0)), "auxiliary head")] | |
| sw_w, sw_h, gap_sw, gap_item = 0.16, 0.09, 0.04, 0.22 | |
| widths = [_measure(fig, lab, FS)[0] for *_, lab in items] | |
| total = sum(widths) + len(items) * (sw_w + gap_sw) + (len(items) - 1) * gap_item | |
| x = (W_IN - total) / 2 | |
| for (fc, ec, ls, lab), wtxt in zip(items, widths): | |
| ax.add_patch(FancyBboxPatch((x, y_leg1 - sw_h / 2), sw_w, sw_h, boxstyle="round,pad=0,rounding_size=0.015", | |
| fc=fc, ec=ec, lw=0.6, linestyle=ls, zorder=3)) | |
| if lab.startswith("frozen"): | |
| _snowflake(ax, x + sw_w / 2, y_leg1, r=0.026) | |
| ax.text(x + sw_w + gap_sw, y_leg1, lab, ha="left", va="center", fontsize=FS, color=CW_BLACK, zorder=3) | |
| x += sw_w + gap_sw + wtxt + gap_item | |
| t = ax.text(W_IN / 2, y_leg2, "m = regime token (manual / assistance engaged) dashed + × = road → driver path, evaluated and dropped", | |
| ha="center", va="center", fontsize=FS, color=CW_INK2, zorder=3) | |
| _BOXED.append((t, W_IN - 0.06, 0.14)) | |
| spec_sha, whx_sha = _sha256(SRC_SPEC), _sha256(SRC_WHX) | |
| meta = {"Title": "CockpitWM architecture (clean landscape, ICLR single column)", | |
| "Subject": f"sources: {SRC_SPEC} S1-S5 (sha256 {spec_sha}); {SRC_WHX} header (sha256 {whx_sha}); rendered {_now()} by make_figs.py", | |
| "Keywords": "CockpitWM, model_spec.md, parameter counts verbatim, 3.50 M derived total, clean layout", | |
| "Creator": "paper/figs/make_figs.py (matplotlib)"} | |
| _check_overflow(fig) | |
| fig.savefig(out_pdf, metadata=meta) | |
| fig.savefig(out_pdf[:-4] + ".png", dpi=300) | |
| plt.close(fig) | |
| _write_arch_clean_sidecar(out_pdf, spec_sha, whx_sha, W_IN, H_IN, ["0.16 M", "1.61 M", "0.88 M", "0.25 M", "0.33 M", "3.50 M", "0.2 s"]) | |
| print(f"wrote {out_pdf} (clean, {W_IN:.2f} x {H_IN:.2f} in), .png and sidecar") | |
| # ===================================================================================================== | |
| # 3. Pareto PNG copy | |
| # ===================================================================================================== | |
| def copy_pareto(dst_dir: str) -> None: | |
| dst = os.path.join(dst_dir, os.path.basename(SRC_PARETO)) | |
| if os.path.exists(dst) and _sha256(dst) == _sha256(SRC_PARETO): | |
| print(f"{dst} already byte-identical to source (sha256 {_sha256(dst)[:16]}…)") | |
| return | |
| shutil.copyfile(SRC_PARETO, dst) | |
| print(f"copied {SRC_PARETO} -> {dst} (sha256 {_sha256(dst)[:16]}…)") | |
| def main(argv: list[str]) -> int: | |
| argv = list(argv[1:]) | |
| out_dir = HERE | |
| if "--out" in argv: # space form: --out DIR | |
| k = argv.index("--out") | |
| out_dir = os.path.abspath(argv[k + 1]) | |
| del argv[k:k + 2] | |
| args = [a for a in argv if not a.startswith("--")] | |
| flags = {a for a in argv if a.startswith("--")} | |
| for f in flags: | |
| if f.startswith("--out="): | |
| out_dir = os.path.abspath(f[len("--out="):]) | |
| os.makedirs(out_dir, exist_ok=True) | |
| which = set(args) or {"heatmap", "arch", "pareto"} | |
| if "heatmap" in which: | |
| layout = "column" if "--column" in flags else ("iclr" if "--iclr" in flags else "wide") | |
| make_robustness_heatmap(os.path.join(out_dir, "robustness_heatmap.pdf"), layout=layout) | |
| if "arch" in which: | |
| if "--clean" in flags: | |
| make_architecture_clean(os.path.join(out_dir, "architecture.pdf")) | |
| elif "--wide" in flags: | |
| make_architecture_wide(os.path.join(out_dir, "architecture.pdf")) | |
| else: | |
| make_architecture(os.path.join(out_dir, "architecture.pdf")) | |
| if "pareto" in which: | |
| copy_pareto(out_dir) | |
| return 0 | |
| if __name__ == "__main__": | |
| sys.exit(main(sys.argv)) | |
Xet Storage Details
- Size:
- 69.2 kB
- Xet hash:
- 469f3fd1849067ea1cdf11277d209ebe1128a23a6fa91b3be3840213dd0ceb6d
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.