#!/usr/bin/env python from __future__ import annotations import argparse import csv import json import math import struct import sys import zlib from collections import OrderedDict from collections.abc import Iterable from pathlib import Path from typing import Any PROJECT_ROOT = Path(__file__).resolve().parents[1] if str(PROJECT_ROOT) not in sys.path: sys.path.insert(0, str(PROJECT_ROOT)) from dovla_cil.utils.io import ensure_dir, write_json # noqa: E402, I001 METRIC_ALIASES = OrderedDict( [ ( "success_rate", ( "task_success_rate", "selected_success_rate", "success_rate", "eval.task_success_rate", ), ), ( "ranking_acc", ("pairwise_ranking_accuracy", "ranking_acc", "eval.pairwise_ranking_accuracy"), ), ("top1_action_selection", ("top1_action_selection", "eval.top1_action_selection")), ( "instruction_switch_acc", ( "instruction_switch_accuracy", "instruction_switch_acc", "eval.instruction_switch_accuracy", ), ), ("effect_mae", ("effect_prediction_mae", "effect_mae", "eval.effect_prediction_mae")), ( "regret_ece", ("regret_calibration_error", "regret_ece", "eval.regret_calibration_error"), ), ("ndcg_at_k", ("ndcg_at_k", "eval.ndcg_at_k")), ] ) BASELINE_NAMES = { "expert_only_bc", "more_independent_demos", "random_negatives", "cross_state_negatives", "label_only_counterfactual", "world_model_auxiliary", "no_effect_head", "no_rank_regret", } ABLATION_NAMES = {"world_model_auxiliary", "no_effect_head", "no_rank_regret"} def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser( description="Collect DoVLA-CIL runs and create paper-ready tables, figures, and summary." ) parser.add_argument("--runs", type=Path, default=Path("runs"), help="Run directory to scan.") parser.add_argument( "--out", type=Path, default=Path("paper_artifacts"), help="Output directory." ) args = parser.parse_args(argv) summary = generate_paper_artifacts(args.runs, args.out) print(f"paper artifacts: {args.out}") print(f"result summary: {summary['result_summary']}") print(f"tables: {len(summary['tables'])}") print(f"figures: {len(summary['figures'])}") return 0 def generate_paper_artifacts(runs_dir: str | Path, out_dir: str | Path) -> dict[str, Any]: runs_path = Path(runs_dir) output_dir = ensure_dir(out_dir) figures_dir = ensure_dir(output_dir / "figures") payloads = load_result_payloads(runs_path) metric_rows = collect_metric_rows(payloads, runs_path) scaling_rows = collect_scaling_rows(payloads, runs_path) baseline_rows = collect_baseline_rows(metric_rows) ablation_rows = [row for row in baseline_rows if row.get("baseline") in ABLATION_NAMES] category_rows = collect_per_category_rows(payloads) table_paths = { "main_scaling": write_table_pair( output_dir / "main_scaling_table", scaling_rows, [ "run_name", "k", "num_states", "effective_total_records", "success_rate", "ranking_acc", "top1_action_selection", "instruction_switch_acc", "effect_mae", "regret_ece", ], title="Main Scaling Table", ), "baseline_comparison": write_table_pair( output_dir / "baseline_comparison_table", baseline_rows, [ "baseline", "run_name", "success_rate", "ranking_acc", "top1_action_selection", "instruction_switch_acc", "effect_mae", "regret_ece", ], title="Baseline Comparison Table", ), "ablation": write_table_pair( output_dir / "ablation_table", ablation_rows, [ "baseline", "run_name", "success_rate", "ranking_acc", "top1_action_selection", "instruction_switch_acc", "effect_mae", "regret_ece", ], title="Ablation Table", ), "causalstress_per_category": write_table_pair( output_dir / "causalstress_per_category_table", category_rows, [ "source", "category", "success", "selected_success", "failure_rate", "selected_failure_rate", "instruction_switch", "top1", "pair_correct", "ranking_acc", ], title="CausalStress Per-Category Table", ), } figure_paths = { "performance_vs_k": figures_dir / "performance_vs_k.png", "same_state_vs_cross_state_ranking": figures_dir / "same_state_vs_cross_state_ranking.png", "physical_outcome_vs_label_only": figures_dir / "physical_outcome_vs_label_only.png", "success_by_failure_category": figures_dir / "success_by_failure_category.png", "regret_calibration": figures_dir / "regret_calibration.png", } plot_performance_vs_k(scaling_rows, figure_paths["performance_vs_k"]) plot_same_state_vs_cross_state(baseline_rows, figure_paths["same_state_vs_cross_state_ranking"]) plot_physical_vs_label_only(baseline_rows, figure_paths["physical_outcome_vs_label_only"]) plot_success_by_failure_category(category_rows, figure_paths["success_by_failure_category"]) plot_regret_calibration(metric_rows, scaling_rows, figure_paths["regret_calibration"]) summary_md = render_result_summary( metric_rows=metric_rows, scaling_rows=scaling_rows, baseline_rows=baseline_rows, category_rows=category_rows, table_paths=table_paths, figure_paths=figure_paths, ) summary_path = output_dir / "result_summary.md" summary_path.write_text(summary_md, encoding="utf-8") manifest = { "runs_dir": str(runs_path), "out_dir": str(output_dir), "num_payloads": len(payloads), "num_metric_rows": len(metric_rows), "num_scaling_rows": len(scaling_rows), "num_baseline_rows": len(baseline_rows), "num_category_rows": len(category_rows), "tables": { key: {"csv": str(value["csv"]), "markdown": str(value["markdown"])} for key, value in table_paths.items() }, "figures": {key: str(path) for key, path in figure_paths.items()}, "result_summary": str(summary_path), } write_json(manifest, output_dir / "artifact_manifest.json") return manifest def load_result_payloads(runs_dir: Path) -> list[dict[str, Any]]: paths: OrderedDict[str, Path] = OrderedDict() for pattern in ( "**/metrics.json", "**/causalstress.json", "**/lattice_eval.json", "**/scaling_summary.json", ): for path in sorted(runs_dir.glob(pattern)): paths[str(path)] = path payloads: list[dict[str, Any]] = [] for path in paths.values(): try: payload = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): continue if isinstance(payload, dict): payloads.append({"path": path, "payload": payload}) return payloads def collect_metric_rows(payloads: list[dict[str, Any]], runs_dir: Path) -> list[dict[str, Any]]: rows: list[dict[str, Any]] = [] for item in payloads: path = Path(item["path"]) payload = item["payload"] if "runs" in payload and isinstance(payload["runs"], dict): for key, metrics in payload["runs"].items(): if isinstance(metrics, dict): rows.append(normalize_metric_row(metrics, path, runs_dir, run_name=f"k{key}")) continue rows.append(normalize_metric_row(payload, path, runs_dir)) return sorted(rows, key=lambda row: (_sort_numeric(row.get("k")), row.get("run_name", ""))) def collect_scaling_rows(payloads: list[dict[str, Any]], runs_dir: Path) -> list[dict[str, Any]]: rows: list[dict[str, Any]] = [ row for row in collect_metric_rows(payloads, runs_dir) if _is_number(row.get("k")) ] csv_rows = read_scaling_csvs(runs_dir) keyed: OrderedDict[str, dict[str, Any]] = OrderedDict() for row in rows + csv_rows: key = str(row.get("k", row.get("run_name", len(keyed)))) keyed[key] = {**keyed.get(key, {}), **row} return sorted(keyed.values(), key=lambda row: _sort_numeric(row.get("k"))) def collect_baseline_rows(metric_rows: list[dict[str, Any]]) -> list[dict[str, Any]]: rows = [ row for row in metric_rows if row.get("baseline") or row.get("run_name") in BASELINE_NAMES ] for row in rows: if not row.get("baseline") and row.get("run_name") in BASELINE_NAMES: row["baseline"] = row["run_name"] return sorted(rows, key=lambda row: str(row.get("baseline", row.get("run_name", "")))) def collect_per_category_rows(payloads: list[dict[str, Any]]) -> list[dict[str, Any]]: rows: list[dict[str, Any]] = [] for item in payloads: path = Path(item["path"]) payload = item["payload"] per_category = payload.get("per_category") if not isinstance(per_category, dict): eval_payload = payload.get("eval") if isinstance(eval_payload, dict): per_category = eval_payload.get("per_category") if not isinstance(per_category, dict): continue for category, metrics in sorted(per_category.items()): if not isinstance(metrics, dict): continue row = {"source": str(path), "category": category} for key, value in metrics.items(): if _is_number(value): row[key] = float(value) if "ranking_acc" not in row and "pair_correct" in row: row["ranking_acc"] = row["pair_correct"] rows.append(row) return rows def normalize_metric_row( payload: dict[str, Any], path: Path, runs_dir: Path, *, run_name: str | None = None ) -> dict[str, Any]: config = payload.get("config") if isinstance(payload.get("config"), dict) else {} eval_payload = payload.get("eval") if isinstance(payload.get("eval"), dict) else {} row: dict[str, Any] = { "run_name": run_name or str(payload.get("run_name") or payload.get("name") or path.parent.name), "source_path": str(path), "relative_path": str(_relative_to(path, runs_dir)), "baseline": str(payload.get("baseline") or config.get("baseline") or ""), "k": _first_present(payload, "k", "K", default=config.get("k", "")), "num_states": _first_present(payload, "num_states", default=config.get("num_states", "")), "effective_total_records": _first_present( payload, "effective_total_records", default=config.get("effective_total_records", "") ), "checkpoint": str(payload.get("checkpoint") or eval_payload.get("checkpoint") or ""), } if not row["baseline"]: inferred = infer_baseline_from_path(path) if inferred: row["baseline"] = inferred for canonical, aliases in METRIC_ALIASES.items(): row[canonical] = metric_value(payload, aliases) row["score"] = composite_score(row) return row def read_scaling_csvs(runs_dir: Path) -> list[dict[str, Any]]: rows: list[dict[str, Any]] = [] for path in sorted(runs_dir.glob("**/scaling_results.csv")): with path.open("r", encoding="utf-8", newline="") as handle: for raw in csv.DictReader(handle): row = {key: coerce_number(value) for key, value in raw.items()} row.setdefault("run_name", f"k{row.get('k', len(rows))}") row["source_path"] = str(path) row["relative_path"] = str(path) rows.append(row) return rows def write_table_pair( stem: Path, rows: list[dict[str, Any]], fieldnames: list[str], *, title: str ) -> dict[str, Path]: csv_path = stem.with_suffix(".csv") md_path = stem.with_suffix(".md") write_csv(csv_path, rows, fieldnames) md_path.write_text(markdown_table(rows, fieldnames, title=title), encoding="utf-8") return {"csv": csv_path, "markdown": md_path} def write_csv(path: Path, rows: list[dict[str, Any]], fieldnames: list[str]) -> None: ensure_dir(path.parent) with path.open("w", encoding="utf-8", newline="") as handle: writer = csv.DictWriter(handle, fieldnames=fieldnames) writer.writeheader() for row in rows: writer.writerow({field: format_cell(row.get(field, "")) for field in fieldnames}) def markdown_table(rows: list[dict[str, Any]], fieldnames: list[str], *, title: str) -> str: lines = [f"# {title}", ""] lines.append("| " + " | ".join(fieldnames) + " |") lines.append("| " + " | ".join("---" for _field in fieldnames) + " |") if not rows: lines.append("| " + " | ".join("_n/a_" for _field in fieldnames) + " |") for row in rows: cells = " | ".join(format_cell(row.get(field, "")) for field in fieldnames) lines.append(f"| {cells} |") return "\n".join(lines).rstrip() + "\n" def plot_performance_vs_k(rows: list[dict[str, Any]], path: Path) -> None: data = [row for row in rows if _is_number(row.get("k"))] if not data: write_placeholder_png(path) return plt = pyplot() if plt is None: write_placeholder_png(path) return fig, ax = plt.subplots(figsize=(6, 4)) x_values = [float(row["k"]) for row in data] for metric in ("success_rate", "ranking_acc", "instruction_switch_acc"): y_values = [float(row.get(metric, 0.0) or 0.0) for row in data] ax.plot(x_values, y_values, marker="o", label=metric) if all(value > 0 for value in x_values): ax.set_xscale("log", base=2) ax.set_xlabel("K") ax.set_ylabel("Score") ax.set_title("Performance vs K") ax.grid(True, alpha=0.3) ax.legend() save_figure(fig, plt, path) def plot_same_state_vs_cross_state(rows: list[dict[str, Any]], path: Path) -> None: cross = best_row([row for row in rows if row.get("baseline") == "cross_state_negatives"]) same = best_row( [ row for row in rows if row.get("baseline") not in {"cross_state_negatives", "label_only_counterfactual"} ] ) labels = ["same_state", "cross_state"] values = [ float((same or {}).get("ranking_acc", 0.0) or 0.0), float((cross or {}).get("ranking_acc", 0.0) or 0.0), ] plot_bar(labels, values, path, title="Same-State vs Cross-State Ranking", ylabel="ranking_acc") def plot_physical_vs_label_only(rows: list[dict[str, Any]], path: Path) -> None: label = best_row([row for row in rows if row.get("baseline") == "label_only_counterfactual"]) physical = best_row([row for row in rows if row.get("baseline") != "label_only_counterfactual"]) labels = ["physical_outcome", "label_only"] values = [ float((physical or {}).get("ranking_acc", 0.0) or 0.0), float((label or {}).get("ranking_acc", 0.0) or 0.0), ] plot_bar(labels, values, path, title="Physical Outcome vs Label-Only", ylabel="ranking_acc") def plot_success_by_failure_category(rows: list[dict[str, Any]], path: Path) -> None: labels = [str(row.get("category", "")) for row in rows] values = [ float( row.get( "selected_success", row.get("success", 1.0 - float(row.get("failure_rate", 1.0))), ) or 0.0 ) for row in rows ] plot_bar(labels, values, path, title="Success by Failure Category", ylabel="success") def plot_regret_calibration( metric_rows: list[dict[str, Any]], scaling_rows: list[dict[str, Any]], path: Path ) -> None: rows = scaling_rows or metric_rows if not rows: write_placeholder_png(path) return if any(_is_number(row.get("k")) for row in rows): plot_line( [float(row.get("k") or index + 1) for index, row in enumerate(rows)], [float(row.get("regret_ece", 0.0) or 0.0) for row in rows], path, title="Regret Calibration", xlabel="K", ylabel="regret_ece", log_x=True, ) else: plot_bar( [str(row.get("run_name", index + 1)) for index, row in enumerate(rows)], [float(row.get("regret_ece", 0.0) or 0.0) for row in rows], path, title="Regret Calibration", ylabel="regret_ece", ) def plot_bar( labels: list[str], values: list[float], path: Path, *, title: str, ylabel: str ) -> None: plt = pyplot() if plt is None or not labels: write_placeholder_png(path) return fig, ax = plt.subplots(figsize=(max(6, min(12, len(labels) * 0.8)), 4)) ax.bar(labels, values) ax.set_title(title) ax.set_ylabel(ylabel) ax.tick_params(axis="x", rotation=30) fig.tight_layout() save_figure(fig, plt, path) def plot_line( x_values: list[float], y_values: list[float], path: Path, *, title: str, xlabel: str, ylabel: str, log_x: bool = False, ) -> None: plt = pyplot() if plt is None or not x_values: write_placeholder_png(path) return fig, ax = plt.subplots(figsize=(6, 4)) ax.plot(x_values, y_values, marker="o") if log_x and all(value > 0 for value in x_values): ax.set_xscale("log", base=2) ax.set_title(title) ax.set_xlabel(xlabel) ax.set_ylabel(ylabel) ax.grid(True, alpha=0.3) save_figure(fig, plt, path) def render_result_summary( *, metric_rows: list[dict[str, Any]], scaling_rows: list[dict[str, Any]], baseline_rows: list[dict[str, Any]], category_rows: list[dict[str, Any]], table_paths: dict[str, dict[str, Path]], figure_paths: dict[str, Path], ) -> str: best = best_row(metric_rows) warnings = expected_claim_warnings(scaling_rows, baseline_rows) lines = [ "# DoVLA-CIL Paper Artifact Summary", "", "## Key Claims", "", ] if best: lines.append( "- Best detected model: " f"`{best.get('run_name')}` with ranking_acc={format_cell(best.get('ranking_acc'))} " f"and success_rate={format_cell(best.get('success_rate'))}." ) else: lines.append("- No model metrics were found.") lines.extend( [ f"- Scaling rows collected: {len(scaling_rows)}.", f"- Baseline rows collected: {len(baseline_rows)}.", f"- CausalStress category rows collected: {len(category_rows)}.", "", "## Tables", "", ] ) for key, paths in sorted(table_paths.items()): lines.append(f"- `{key}`: `{paths['csv']}`, `{paths['markdown']}`") lines.extend(["", "## Figures", ""]) for key, path in sorted(figure_paths.items()): lines.append(f"- `{key}`: `{path}`") lines.extend(["", "## Expected Claim Checks", ""]) if warnings: for warning in warnings: lines.append(f"- Warning: {warning}") else: lines.append("- No automatic warnings for the expected claims.") return "\n".join(lines).rstrip() + "\n" def expected_claim_warnings( scaling_rows: list[dict[str, Any]], baseline_rows: list[dict[str, Any]] ) -> list[str]: warnings: list[str] = [] if len(scaling_rows) >= 2: ordered = sorted(scaling_rows, key=lambda row: _sort_numeric(row.get("k"))) first = float(ordered[0].get("ranking_acc", 0.0) or 0.0) last = float(ordered[-1].get("ranking_acc", 0.0) or 0.0) if last <= first: warnings.append("ranking accuracy does not improve from smallest K to largest K.") else: warnings.append("scaling claim is unsupported because fewer than two K values were found.") cross = best_row( [row for row in baseline_rows if row.get("baseline") == "cross_state_negatives"] ) same = best_row( [ row for row in baseline_rows if row.get("baseline") not in {"cross_state_negatives", "label_only_counterfactual"} ] ) if cross and same: same_rank = float(same.get("ranking_acc", 0.0) or 0.0) cross_rank = float(cross.get("ranking_acc", 0.0) or 0.0) if same_rank <= cross_rank: warnings.append("same-state ranking does not beat cross-state negatives.") else: warnings.append("same-state vs cross-state comparison is missing a required baseline.") label = best_row( [row for row in baseline_rows if row.get("baseline") == "label_only_counterfactual"] ) physical = best_row( [row for row in baseline_rows if row.get("baseline") != "label_only_counterfactual"] ) if label and physical: physical_rank = float(physical.get("ranking_acc", 0.0) or 0.0) label_rank = float(label.get("ranking_acc", 0.0) or 0.0) if physical_rank <= label_rank: warnings.append("physical-outcome baseline does not beat label-only counterfactuals.") else: warnings.append("physical outcome vs label-only comparison is missing a required baseline.") return warnings def metric_value(payload: dict[str, Any], aliases: Iterable[str]) -> float: search_payloads = [payload] if isinstance(payload.get("eval"), dict): search_payloads.append(payload["eval"]) if isinstance(payload.get("metrics"), dict): search_payloads.append(payload["metrics"]) for item in search_payloads: for alias in aliases: value = lookup_dotted(item, alias) if _is_number(value): return float(value) return 0.0 def lookup_dotted(payload: dict[str, Any], key: str) -> Any: if key in payload: return payload[key] cursor: Any = payload for part in key.split("."): if not isinstance(cursor, dict) or part not in cursor: return None cursor = cursor[part] return cursor def _first_present(payload: dict[str, Any], *keys: str, default: Any = "") -> Any: for key in keys: if key in payload and payload[key] not in (None, ""): return payload[key] return default if default is not None else "" def infer_baseline_from_path(path: Path) -> str: for part in path.parts: if part in BASELINE_NAMES: return part return "" def composite_score(row: dict[str, Any]) -> float: values = [ float(row.get("success_rate", 0.0) or 0.0), float(row.get("ranking_acc", 0.0) or 0.0), float(row.get("top1_action_selection", 0.0) or 0.0), float(row.get("instruction_switch_acc", 0.0) or 0.0), ] return sum(values) / len(values) def best_row(rows: list[dict[str, Any]]) -> dict[str, Any] | None: if not rows: return None return max( rows, key=lambda row: ( float(row.get("ranking_acc", 0.0) or 0.0), float(row.get("success_rate", 0.0) or 0.0), float(row.get("score", 0.0) or 0.0), ), ) def coerce_number(value: Any) -> Any: if isinstance(value, int | float): return value if isinstance(value, str): stripped = value.strip() if not stripped: return "" try: if any(char in stripped for char in ".eE"): return float(stripped) return int(stripped) except ValueError: return value return value def format_cell(value: Any) -> str: if value is None: return "" if isinstance(value, float): if math.isnan(value) or math.isinf(value): return "" return f"{value:.6g}" return str(value) def _is_number(value: Any) -> bool: return isinstance(value, int | float) and not isinstance(value, bool) def _sort_numeric(value: Any) -> tuple[int, float, str]: if _is_number(value): return (0, float(value), "") try: return (0, float(str(value)), "") except (TypeError, ValueError): return (1, 0.0, str(value)) def _relative_to(path: Path, parent: Path) -> Path: try: return path.relative_to(parent) except ValueError: return path def pyplot(): try: import matplotlib matplotlib.use("Agg") import matplotlib.pyplot as plt except ImportError: return None return plt def save_figure(fig: Any, plt: Any, path: Path) -> None: ensure_dir(path.parent) fig.tight_layout() fig.savefig(path) plt.close(fig) def write_placeholder_png(path: Path) -> None: ensure_dir(path.parent) width, height = 1, 1 raw = b"\x00\xff\xff\xff" png = b"\x89PNG\r\n\x1a\n" png += png_chunk(b"IHDR", struct.pack(">IIBBBBB", width, height, 8, 2, 0, 0, 0)) png += png_chunk(b"IDAT", zlib.compress(raw)) png += png_chunk(b"IEND", b"") path.write_bytes(png) def png_chunk(kind: bytes, data: bytes) -> bytes: return ( struct.pack(">I", len(data)) + kind + data + struct.pack(">I", zlib.crc32(kind + data) & 0xFFFFFFFF) ) if __name__ == "__main__": raise SystemExit(main())