"""One self-contained HTML report (Simplified Chinese, tables only, no external resources) from the analysis JSONs.
python report.py --out report.html [--inputs inputs.json] --summary "最终臂=runs/autorun-X/SUMMARY.md" ... \
--val-run "基线·三字段·spread=S.json" ... --val-compare "最终 vs 基线·三字段·spread·全部对象=C.json" ... \
[--val-control "最终 vs 对照·三字段·spread·全部对象=C.json" ...] --val-decode "最终·三字段：spread vs argmax=C.json" ... \
[--test-run ... --test-compare ... --test-control ... --test-decode ...] [--llm L.json] \
[--ablation "最终=ablation/report.json" ... --ablation-sample sample.json] \
[--selection-cohort "最终（select2）=validation-X-all.jsonl" ...] [--reference "最终=DIR" ...] \
[--walltime walltime.tsv] [--notes NOTES.md]
Inputs: stratify.py / compare.py / llm_compare.py outputs, diag_ablation.py report.json, roomgenbench --reference-check
JSONs (one directory per checkpoint), the autopilot SUMMARY.md files (``[LABEL=]PATH``), run_plan's walltime.tsv and
isambard_inputs.py's inputs.json (the two arms, which one is final and why, data / model provenance, missing LLM modes).
Every number and every "all / none" statement shown is read from these files.
Decisions come ONLY from the validation comparisons (``--val-compare``: final vs baseline; ``--val-control``: final vs
control, the evidence for the choice of the final arm; ``--val-decode``: spread vs argmax). Test, LLM, ablation and
RoomGenBench sections are labelled report-only and never feed a decision. Significance:
primary = paired room-bootstrap 95% CI of the mean per-room difference (direction from it); robustness = Holm-corrected
sign test; disagreements are flagged; pooled object-weighted differences are descriptive only. A final-vs-baseline
comparison whose two sides carry the same checkpoint is flagged as a placeholder (and "differences are 0" when both
sides are also the same evaluate output). A ``--selection-cohort`` label starts with 最终 or 基线, the checkpoint that
cohort selected; the selection-bias note names a direction only when one checkpoint's cohorts hold the validation
scenes and the other's were given and do not.
"""
from __future__ import annotations
import argparse
from collections import Counter
import html
import json
import math
from pathlib import Path
import re
NAMES = {"bottom_center_error_m": "位置误差 (m) ↓", "log_size_error": "log 尺寸误差 ↓", "yaw_error_rad": "朝向误差 (rad) ↓",
"bev_iou": "BEV IoU ↑", "log_size_error_plain_convention": "log 尺寸误差·原约定 ↓",
"yaw_error_rad_plain_convention": "朝向误差·原约定 (rad) ↓", "object_collision_rate": "物体碰撞率 ↓",
"object_hard_violation_rate": "物体硬约束违例率 ↓", "gt_object_collision_rate": "真值物体碰撞率",
"room_clean_rate": "洁净房间率（有布局）↑", "room_collision_rate": "有碰撞房间率 ↓", "gt_room_clean_rate": "真值洁净房间率",
"gt_room_collision_rate": "真值有碰撞房间率", "interface_pass_rate": "接口通过率 ↑", "strict_ok_rate": "严格几何通过率 ↑",
"clean_room_rate_incl_failures": "洁净房间率（失败计不洁净）↑",
"baseline:room_center_position": "基线·房间中心 (m)", "baseline:category_mean_position": "基线·类别均值位置 (m)",
"baseline:category_median_size": "基线·类别中位尺寸", "baseline:uniform_yaw": "基线·均匀随机朝向 (rad)",
"cell_ce": "格子交叉熵 ↓", "cell_entropy": "预测格子熵", "p_gt": "p(真值格子) ↑", "argmax_hit": "argmax 命中率 ↑",
"xy_err_m": "argmax XY 误差 (m) ↓", "size_err": "尺寸误差 ↓", "yaw_ce": "朝向分类交叉熵 ↓", "yaw_err_rad": "朝向误差 (rad) ↓",
"gt_rank": "真值格子名次 ↓", "cell_tv_vs_a": "格子分布 TV 距离（相对 a）", "yaw_tv_vs_a": "朝向分布 TV 距离（相对 a）"}
KINDS = {"rule": "目标选择规则", "place": "声明位置", "family": "来源族", "objects": "物体数分箱", "all": "全部"}
RULE_NAMES = {"frozen_prep": "frozen_prep（旧对象）", "wall_anchor": "wall_anchor（新·墙面）",
"support_inside_parent": "support_inside_parent（新·内层）", "support_on_added_parent": "support_on_added_parent（新·新增父物体上）"}
ABLATION = {"b": "b 房间尺寸换成另一房间的", "c": "c 房间类型换成另一房间的", "d": "d 描述换成类别名", "e": "e 请求顺序打乱"}
MODES = {"prompt": "单次询问", "harness": "修复循环（校验后带问题清单重问）", "structured": "单次询问（结构化提示）",
"structured-harness": "修复循环（结构化提示，校验后重问）"}
CAUSES = {"IncompleteRead": "中继传输错误（HTTP 响应体读取不完整，IncompleteRead）：传输层失败，不是模型给出的无效布局",
"over_capacity": "超出模型容量 max_objects", "other": "其他推理失败"}
VERDICT = {"a_better": ("A 更好", "good"), "a_worse": ("A 更差", "bad"), None: ("无显著差异", "")}
CSS = """
:root{--bg:#fbfbfa;--fg:#1d1d1f;--muted:#6b6b70;--line:#e2e2e0;--card:#fff;--accent:#2f5d9e;--good:#1f7a4d;--bad:#a8322d;--warn:#8a5a00;--warnbg:#fff6e0}
@media (prefers-color-scheme:dark){:root:not([data-theme="light"]){--bg:#151517;--fg:#ececee;--muted:#a0a0a8;--line:#333338;--card:#1d1d20;--accent:#8fb3ec;--good:#6cc79a;--bad:#ef8b85;--warn:#e7c06a;--warnbg:#2b2416}}
:root[data-theme="dark"]{--bg:#151517;--fg:#ececee;--muted:#a0a0a8;--line:#333338;--card:#1d1d20;--accent:#8fb3ec;--good:#6cc79a;--bad:#ef8b85;--warn:#e7c06a;--warnbg:#2b2416}
*{box-sizing:border-box}body{margin:0;background:var(--bg);color:var(--fg);font:15px/1.6 -apple-system,"PingFang SC","Hiragino Sans GB","Microsoft YaHei","Noto Sans CJK SC",sans-serif}
main{max-width:1100px;margin:0 auto;padding:24px 16px 64px}h1{font-size:1.6em;margin:.2em 0}h2{font-size:1.25em;margin-top:2.2em;border-bottom:1px solid var(--line);padding-bottom:.3em}
h3{font-size:1.05em;margin-top:1.6em}p,li{max-width:80ch}.muted{color:var(--muted)}code{font-size:.9em;background:var(--card);border:1px solid var(--line);border-radius:4px;padding:0 .3em;word-break:break-all}
.wrap{overflow-x:auto;margin:.6em 0 1.2em;border:1px solid var(--line);border-radius:8px;background:var(--card)}
table{border-collapse:collapse;width:100%;font-size:.88em;font-variant-numeric:tabular-nums}th,td{padding:.45em .7em;border-bottom:1px solid var(--line);text-align:right;white-space:nowrap}
th:first-child,td:first-child{text-align:left}thead th{background:var(--bg);font-weight:600}tr:last-child td{border-bottom:0}
.ci{color:var(--muted);font-size:.85em}.good{color:var(--good);font-weight:600}.bad{color:var(--bad);font-weight:600}
.flag{background:var(--warnbg);color:var(--warn);border-left:3px solid var(--warn);padding:.6em .9em;border-radius:4px;margin:1em 0}
.tag{display:inline-block;font-size:.75em;font-weight:600;padding:.05em .5em;border-radius:4px;border:1px solid currentColor;margin-left:.5em;vertical-align:middle}
details{margin:.8em 0}summary{cursor:pointer;color:var(--accent)}nav a{color:var(--accent);margin-right:1em;text-decoration:none}
"""
REPORT_ONLY = "仅报告，不参与决策"
DECISION = "决策依据：验证集"
def esc(value):
return html.escape(str(value))
def num(x, digits=3):
if x is None or (isinstance(x, float) and not math.isfinite(x)):
return "—"
return f"{x:.{digits}g}" if abs(x) and abs(x) < 1e-3 else f"{x:.{digits}f}"
def cell(stat, digits=3):
"""mean [lo, hi] from {"mean", "ci95"} or [mean, lo, hi, ...]."""
if stat is None:
return "—"
mean, lo, hi = (stat["mean"], *stat["ci95"]) if isinstance(stat, dict) else stat[:3]
return f'{num(mean, digits)} [{num(lo, digits)}, {num(hi, digits)}]'
def pval(p):
return "—" if p is None else "<1e-300" if p == 0 else f"{p:.2g}"
def table(head, rows):
out = "
" + "".join(f"
{h}
" for h in head) + "
"
return out + "".join("
" + "".join(f"
{c}
" for c in row) + "
" for row in rows) + "
"
def markdown(text):
"""The subset SUMMARY.md uses: headings, pipe tables, bullets, paragraphs, `code`, **bold**."""
inline = lambda s: re.sub(r"\*\*(.+?)\*\*", r"\1", re.sub(r"`([^`]+)`", r"\1", esc(s)))
out, lines, i = [], text.splitlines(), 0
while i < len(lines):
line = lines[i]
if line.startswith("#"):
level = min(len(line) - len(line.lstrip("#")) + 2, 6)
out.append(f"{inline(line.lstrip('#').strip())}")
elif line.startswith("|"):
block = []
while i < len(lines) and lines[i].startswith("|"):
block.append([c.strip() for c in lines[i].strip().strip("|").split("|")])
i += 1
body = [r for r in block[1:] if not all(set(c) <= set("-: ") for c in r)]
out.append(table([inline(c) for c in block[0]], [[inline(c) for c in r] for r in body]))
continue
elif line.startswith("- "):
items = []
while i < len(lines) and lines[i].startswith("- "):
items.append(f"
{inline(lines[i][2:])}
")
i += 1
out.append("
" + "".join(items) + "
")
continue
elif line.strip():
out.append(f"
{inline(line)}
")
i += 1
return "\n".join(out)
def load(spec):
label, _, path = spec.rpartition("=") # labels may hold "=", paths do not
return label, json.loads(Path(path).read_text()), path
def short_ckpt(path):
return f"{Path(path).parent.name}/{Path(path).name}" if path else "—"
def rows_info(path, _cache={}):
"""(scene ids, objects whose selection_rule is not frozen_prep) of a rows JSONL file."""
if path not in _cache:
ids, new = set(), 0
with open(path) as stream:
for line in stream:
if line.strip():
prov = json.loads(line)["provenance"]
ids.add(prov.get("scene_id"))
new += sum(((e or {}).get("selection_rule") or "frozen_prep") != "frozen_prep" for e in prov.get("field_evidence") or [])
_cache[path] = ids, new
return _cache[path]
def scene_ids(path):
return rows_info(path)[0]
def overlaps(ids, cohorts):
"""(label, path, scenes of `ids` in the cohort) for every cohort holding any of them."""
return [(label, path, len(ids & scene_ids(path))) for label, path in cohorts if ids & scene_ids(path)]
def cohort_flags(what, ids, cohorts):
return "".join(f"
" for label, path, k in overlaps(ids, cohorts))
ROLES = ("最终", "基线") # a --selection-cohort label starts with the checkpoint it selected
def bias_note(what, ids, cohorts):
"""Selection-bias statement for the scenes `ids`, from each checkpoint's cohorts: a direction only when one
checkpoint's cohorts hold these scenes and the other's were given and do not."""
given = {r: [(l, p) for l, p in cohorts if l.startswith(r)] for r in ROLES}
hit = {r: overlaps(ids, given[r]) for r in ROLES}
cover = lambda r: "、".join(f"「{esc(l)}」{k}/{len(ids)}" for l, _, k in hit[r])
if all(hit.values()):
version = {r: {m.group(1) for _, p, _ in hit[r] if (m := re.match(r"validation-([0-9a-f]{12})-", Path(p).name))} for r in ROLES}
saw = [r for r in ROLES if any(rows_info(p)[1] for _, p, _ in hit[r])]
return (f"{esc(what)}：两个检查点都在这些场景上被选择过（最终 {cover('最终')}；基线 {cover('基线')}"
+ (f"；不同数据版本：源验证文件 sha256 前缀 最终 {'/'.join(sorted(version['最终']))}，基线 {'/'.join(sorted(version['基线']))}"
if all(version.values()) and not version["最终"] & version["基线"] else "")
+ "），选择偏差的方向无法确定。选择队列中的新对象（非 frozen_prep 规则）："
+ "；".join(f"「{esc(l)}」{rows_info(p)[1]} 个" for r in ROLES for l, p, _ in hit[r])
+ (f"：只有{saw[0]}的选择见过新对象。" if len(saw) == 1 else "。"))
for r, o in (ROLES, ROLES[::-1]):
if hit[r]:
return (f"{esc(what)}：只有{r}的选择队列含这些场景（{cover(r)}；{o}的选择队列 0/{len(ids)}）：这些行对{r}不是留出评测，比较可能偏向{r}。"
if given[o] else f"{esc(what)}：{r}的选择队列含这些场景（{cover(r)}）；未提供{o}的选择队列，选择偏差的方向无法判断。")
return ""
def run_rows(runs):
rows = []
for split, label, s in runs:
m, c = s["meta"], s["checks"]
rows.append([split, esc(label), f"{esc(short_ckpt(m['checkpoint']))}",
f"{esc(Path(m['rows']).parent.name)}/{esc(Path(m['rows']).name)}", esc(m["projection"]),
esc(m.get("grid_decode") or "—"), "头输出重解码" if m.get("head_outputs") else esc(m.get("baseline") or "—"),
f"{esc((m.get('data_sha256') or '')[:12])}",
f"{esc((m.get('forward_implementation_sha256') or '—')[:12])}",
f"{esc((m.get('implementation_sha256') or '—')[:12])}",
f"{c['requests_with_layout']}/{c['requests']}",
'是' if c["reproduces_report_exactly"] else '否',
str(c["rooms_validation_differs_from_stored"])])
return table(["集合", "运行", "检查点", "数据", "投影", "解码", "来源", "数据 sha256", "前向代码", "解码/评分代码",
"有布局/请求", "逐位复现 report.json", "校验器漂移房间"], rows)
def overall(s, metric):
return s["strata"]["all"]["all"]["metrics"].get(metric)
def placeholder(c):
"""A final-vs-baseline comparison whose two sides carry the same checkpoint."""
return c["meta"]["a"].get("checkpoint") == c["meta"]["b"].get("checkpoint")
def same_output(c):
"""Both sides are the same evaluate output: every difference is 0."""
return all(c["meta"]["a"].get(k) == c["meta"]["b"].get(k) for k in ("eval_dir", "outcomes"))
def verdict_cell(v, a="A"):
text, cls = VERDICT[v["verdict"]["primary"]]
return f'{text.replace("A", esc(a))}' if cls else text
def compare_table(c, final_vs_baseline=False):
a, b = c["meta"]["a"]["label"], c["meta"]["b"]["label"]
rows = []
for metric, v in c["metrics"].items():
st, lo, hi = v["sign_test"], *v["room_mean_diff_ci95"]
robust = VERDICT[v["verdict"]["sign_test"]][0]
rows.append([NAMES.get(metric, metric), num(v["a"]), num(v["b"]), num(v.get("a_room_mean")), num(v.get("b_room_mean")),
f"{num(v['room_mean_diff'], 4)} [{num(lo, 4)}, {num(hi, 4)}]", verdict_cell(v),
f"{st['a_greater']}/{st['b_greater']}/{st['ties']}", pval(st["p_holm"]),
robust if v["verdict"]["agree"] else f"⚠ 不一致：{robust}",
f"{num(v['pooled_diff'], 4)} [{num(v['pooled_diff_ci95'][0], 4)}, {num(v['pooled_diff_ci95'][1], 4)}]",
str(v["rooms"])])
r, where = c["rooms"], c["meta"].get("where") or {}
note = (f"
A = {esc(a)}，B = {esc(b)}；差值 = A − B；合并 = 物体加权均值，房间均值 = 先在房间内平均再对房间平均"
f"（A、B 房间均值之差即主判据的房间均值差）。房间 {r['compared']}"
+ (f"（另有 {r['excluded_by_room_filter']} 个房间因物体数 > {c['meta']['max_objects']} 被排除）" if c["meta"].get("max_objects") else "")
+ f"：两者都有布局 {r['layout_both']}，仅 A 有布局 {r['layout_only_a']}，仅 B 有布局 {r['layout_only_b']}，都没有 {r['layout_neither']}；"
"配对指标只在两者都有布局的房间上计算" + (f"；筛选 {esc(where)}" if where else "") + f"；bootstrap {c['meta']['boot']} 次（房间为单位，配对）。
")
if final_vs_baseline and placeholder(c):
note += (f"
占位：A、B 是同一个检查点 {esc(short_ckpt(c['meta']['a'].get('checkpoint')))}，差值不说明最终模型"
+ ("；A 与 B 也是同一个评测输出：差值必为 0" if same_output(c) else "") + "。
")
elif same_output(c):
note += "
A 与 B 是同一个评测输出：差值必为 0。
"
if not rows:
return note + "
没有符合筛选条件的物体（例如旧数据没有新增对象），无可比较的指标。
"
return note + table(["指标", "A（合并，物体加权）", "B（合并，物体加权）", "A 房间均值", "B 房间均值", "房间均值差 [95% CI]（主判据）",
"结论（主判据）", "符号检验 A>B/A<B/平", "p (Holm)", "稳健性：符号检验", "合并差值 [95% CI]（物体加权，仅描述）",
"房间"], rows)
def strata_tables(s, metrics):
out = []
for kind in ("rule", "place", "family", "objects"):
entries = s["strata"].get(kind, {})
rows = [[esc(RULE_NAMES.get(v, v)), str(e["rooms"]), str(e.get("rooms_with_layout", "—")), str(e["objects"])]
+ [cell(e["metrics"].get(m)) for m in metrics] for v, e in entries.items()]
out.append(f"
按{KINDS[kind]}
" + table([KINDS[kind], "房间", "有布局房间", "物体"] + [NAMES[m] for m in metrics], rows))
return "".join(out)
STRATA_METRICS = ("bottom_center_error_m", "baseline:room_center_position", "log_size_error", "yaw_error_rad",
"object_collision_rate", "room_clean_rate")
def results_table(runs):
"""Model metrics per run next to the trivial baselines of the first run's objects."""
pairs = (("bottom_center_error_m", ("baseline:room_center_position", "baseline:category_mean_position")),
("log_size_error", ("baseline:category_median_size",)), ("yaw_error_rad", ("baseline:uniform_yaw",)), ("bev_iou", ()),
("object_collision_rate", ()), ("room_clean_rate", ()))
first = runs[0]
rows = [[NAMES[metric]] + [cell(overall(s, metric)) for _, s in runs]
+ [" ".join(f"{NAMES[b]}：{cell(overall(first[1], b))}" for b in baselines) or "—"] for metric, baselines in pairs]
return table(["指标"] + [esc(label) for label, _ in runs] + [f"平凡基线（{esc(first[0])} 的同一批物体）"], rows)
def full_set_table(runs):
"""Every room of each run, including those beyond the baseline's max_objects (not paired; room sets differ)."""
rows = []
for label, s in runs:
big = s["strata"].get("objects", {}).get("129+", {})
rows.append([esc(label), f"{s['checks']['requests_with_layout']}/{s['checks']['requests']}", str(s["checks"]["over_capacity_requests"]),
f"{big.get('rooms_with_layout', 0)}/{big.get('rooms', 0)}"]
+ [cell(overall(s, m)) for m in ("bottom_center_error_m", "log_size_error", "yaw_error_rad", "object_collision_rate", "room_clean_rate")])
return table(["运行", "有布局/请求", "超出容量", ">128 物体房间：有布局/总数", NAMES["bottom_center_error_m"], NAMES["log_size_error"],
NAMES["yaw_error_rad"], NAMES["object_collision_rate"], NAMES["room_clean_rate"]], rows)
def decision_table(compares, final_vs_baseline):
rows = []
for label, c, _ in compares:
by = {key: [NAMES.get(m, m) for m, v in c["metrics"].items() if v["verdict"]["primary"] == key] for key in ("a_better", "a_worse", None)}
disagree = [f"{NAMES.get(m, m)}（符号检验：{VERDICT[v['verdict']['sign_test']][0]}）" for m, v in c["metrics"].items() if not v["verdict"]["agree"]]
rows.append([esc(label) + (" 占位：A、B 同一检查点" if final_vs_baseline and placeholder(c) else ""),
f"{esc(c['meta']['a']['label'])} / {esc(c['meta']['b']['label'])}", str(c["rooms"]["layout_both"]),
f"{'、'.join(by['a_better'])}" if by["a_better"] else "—",
f"{'、'.join(by['a_worse'])}" if by["a_worse"] else "—", str(len(by[None])),
f"⚠ {'；'.join(disagree)}" if disagree else "无"])
return table(["验证集比较", "A / B", "配对房间", "A 显著更好（主判据）", "A 显著更差（主判据）", "无显著差异的指标数", "与符号检验不一致"], rows)
def section_reference(refs):
out = []
for label, directory in refs:
rows, check_rows, versions = [], [], set()
for path in sorted(Path(directory).glob("*.json")):
r = json.loads(path.read_text())
versions.add(r["schema_version"])
for place in ("all", "floor", "wall", "on_object"):
e = r["errors"].get(place)
if e: # box_equivalent_* exist in schema v1 and v2; v2 adds matching (by_id = unmatched) and a per-place yaw baseline
rows.append([esc(r["scene_key"]) if place == "all" else "", {"all": "全部", "floor": "地面", "wall": "墙面", "on_object": "物体上"}[place],
str(e["objects"]), num(e["position_error_m"]), num(e.get("position_error_m_by_id")),
num(e["room_center_position_error_m"]), num(e["box_equivalent_log_size_error"]),
num(e["box_equivalent_yaw_error_rad"]), num(e.get("uniform_yaw_baseline_error_rad", r["uniform_yaw_baseline_error_rad"]))])
f, g = r["checks"]["fastfill"], r["checks"]["ground_truth"]
coll = lambda x: x["by_code"].get("collision", {}).get("violation", 0)
check_rows.append([esc(r["scene_key"]), str(r["objects"]), "通过" if f["ok"] else "未通过", str(f["counts"]["violation"]), str(coll(f)),
"通过" if g["ok"] else "未通过", str(g["counts"]["violation"]), str(coll(g)), str(r.get("ground_truth_tilted_objects", "—"))])
out.append(f"
"
+ table(["房间", "真值位置", "物体", "位置误差 (m)", "位置误差·按 ID (m)", "房间中心基线 (m)", "log 尺寸（盒等价）",
"朝向（盒等价, rad）", "均匀随机朝向基线 (rad)"], rows)
+ table(["房间", "物体", "FastFill 校验", "违例", "碰撞对", "真值校验", "真值违例", "真值碰撞对", "真值倾斜物体"], check_rows))
return "".join(out)
def sample_name(path, s):
return f"消融样本（{Path(path).parent.name}，抽自 {Path(s['new']).parent.name}/{Path(s['new']).name}）"
def section_ablation(ablations, sample, s, cohorts):
out = []
if s and cohorts:
out.append(cohort_flags(sample_name(sample, s), set(s["scene_ids"]), cohorts))
keys = ("cell_ce", "p_gt", "argmax_hit", "xy_err_m", "size_err", "yaw_err_rad")
for label, a, path in ablations:
summary, diffs, prior = a["summary_mean_ci95_count"], a["diff_vs_a_mean_ci95"], a["model_a_minus_prior_cell_ce"]
best = max(prior.items(), key=lambda kv: kv[1][0]) # the prior closest to the model: the largest difference
rows = [["a 原始输入"] + [cell(summary["a"].get(k)) for k in keys]]
rows += [[ABLATION[v]] + [cell(summary[v].get(k)) for k in keys] for v in "bcde"]
drows = [[ABLATION[v[0]]] + [cell(diffs[v].get(k), 4) for k in keys] for v in diffs]
prior_a = summary["a"]
out.append(f"
"
+ table(["变体"] + [NAMES[k] for k in keys], rows)
+ "
变体减 a（95% CI，房间 bootstrap）：
" + table(["变体"] + [NAMES[k] for k in keys], drows)
+ f"
模型格子交叉熵减输入无关先验（训练集频率，{a['prior']['train_rows']} 行）：最接近模型的先验 {esc(best[0])}"
f" 差值 {cell(best[1], 4)}；负值 = 模型优于该先验。{len(prior)} 种先验中差值的 CI 上界最大为 "
f"{num(max(v[2] for v in prior.values()), 4)}。
")
return "".join(out)
def section_llm(llm):
out = []
for c in llm["selection_cohort"]:
out.append(f"
")
cost = {c["method"]: c for c in llm["cost"]}
shown = {}
for m, c in cost.items():
loop = c.get("mode") and "harness" in c["mode"]
shown[m] = m + (f"（修复循环，{c['calls_per_room']:.2f} 次调用/房间）" if loop else "")
failures = []
for m, c in cost.items():
fr = c.get("failed_rooms")
if not fr:
continue
p = fr["p_all_in_largest_if_size_independent"]
sizes = "、".join(f"{x['objects']}（第 {x['rank']} 名）" for x in sorted(fr["rooms"], key=lambda x: x["rank"]))
failures.append(f"{esc(m)}：{len(fr['rooms'])} 个房间无布局——" + "；".join(f"{esc(CAUSES.get(k, k))} ×{n}" for k, n in c["failure_causes"].items())
+ f"。这些房间的物体数（{fr['of_rows']} 行中按物体数从多到少的名次）：{sizes}；全部落在物体数最多的 {fr['largest_rooms_holding_all']} 个房间内，"
f"若失败与房间大小无关，{len(fr['rooms'])} 个失败全落在其中的概率为 {pval(p)}"
+ ("：失败集中在最大的房间" + ("，可能是与房间规模相关的中继超时（传输被截断，而非模型给出无效布局）" if "IncompleteRead" in c["failure_causes"] else "")
if p < .05 else "") + "。")
if failures:
out.append("
")
rows = []
for m, c in cost.items():
tokens = c.get("tokens")
rows.append([esc(m), esc(c.get("model") or "本地 Qwen3-8B + 几何头"), esc(MODES.get(c.get("mode"), "一次前向 + 解码")),
esc(c.get("reasoning_effort") or "—"), str(llm["layouts"][m]), num(c["calls_per_room"], 2), num(c.get("mean_latency_s"), 2),
num(c.get("p95_latency_s"), 2), num(c.get("mean_latency_s_layout_rooms"), 2) + f" （{c.get('latency_rooms_layout', '—')} 房间）",
"—" if not tokens else f"{tokens['total_tokens']:,}", "—" if not tokens else f"{c['tokens_per_room']:,.0f}"])
no_tokens = [m for m, c in cost.items() if c["kind"] == "llm" and not c.get("tokens")]
source = lambda c: ("本运行" if c["latency_source"] == "this run" else f"在线运行 {esc(Path(c['latency_source']).name)}") \
+ f"（{esc(c.get('latency_grid_decode') or '—')} 解码）"
out.append("
LLM 延迟取自 predictions.jsonl 的逐房间 latency_s（全部房间的均值即 summary.json 的 mean_latency_s）："
"每房间全部 API 调用的墙钟时间，含客户端限速等待、重试与退避（修复循环还含修复调用）。FastFill 延迟为在线评测记录的 fastfill_latency_ms"
"（一次 GPU 前向 + 解码）：" + "；".join(f"{esc(m)} 取自{source(c)}" for m, c in cost.items() if c["kind"] == "fastfill")
+ "。“仅有布局房间”只平均给出布局的房间。" + (f"summary 未记录 token 用量的方法：{esc('、'.join(no_tokens))}。" if no_tokens else "") + "
")
meta = llm["meta"]["methods"]
sha = lambda x: f"{esc((x or '—')[:12])}"
out.append("复现检查与代码哈希（llm_compare 拒绝数据或 FastFill 解码/评分代码不同的方法）"
+ table(["方法", "评测目录", "检查点", "解码", "来源", "数据 sha256", "前向代码", "解码/评分代码", "逐位复现 report.json"],
[[esc(m), f"{esc(Path(x['eval_dir']).name)}", f"{esc(short_ckpt(x.get('checkpoint')))}",
esc(x.get("grid_decode") or "—"), esc(x.get("baseline") or "—"), sha(x.get("data_sha256")),
sha(x.get("forward_implementation_sha256")), sha(x.get("implementation_sha256")),
'是' if llm["checks"][m] else '否'] for m, x in meta.items()])
+ "")
views = (("common", f"口径一·共同房间：只看每种方法都给出布局的房间（{llm['rows'] - llm['common_rooms']} 个房间因任一方法无布局而对所有方法排除）"),
("failures_as_failures", "口径二·失败计入：全部房间，无布局的房间按下方退化规则计分"))
for name, title in views:
sc = llm["scenarios"][name]
metrics = list(next(iter(sc["methods"].values())))
out.append(f"
{esc(title)}：{sc['rooms']} 个房间
"
+ table(["方法"] + [NAMES[m] for m in metrics], [[esc(shown[k])] + [cell(v.get(m)) for m in metrics] for k, v in sc["methods"].items()]))
rows = []
for pair, stats in sc["pairs"].items():
for metric, v in stats.items():
lo, hi = v["room_mean_diff_ci95"]
robust = VERDICT[v["verdict"]["sign_test"]][0].replace("A", "LLM")
rows.append([esc(pair), NAMES[metric], f"{num(v['room_mean_diff'], 4)} [{num(lo, 4)}, {num(hi, 4)}]",
verdict_cell(v, "LLM"), f"{v['sign_test']['a_greater']}/{v['sign_test']['b_greater']}", pval(v["sign_test"]["p_holm"]),
robust if v["verdict"]["agree"] else f"⚠ 不一致：{robust}"])
out.append("配对比较（差值 = LLM − FastFill；主判据 = 房间均值差的配对 bootstrap CI；Holm 校正在每一对内的指标间）"
+ table(["配对", "指标", "房间均值差 [95% CI]", "结论（主判据）", "符号检验 LLM>/<", "p (Holm)", "稳健性：符号检验"], rows) + "")
out.append(f"
退化规则（口径二）：{esc(llm['meta']['fallback'])}。
")
return "".join(out)
def section_arms(inp):
"""The two autopilot arms, which one is final and why, and where data and models come from (isambard_inputs.py)."""
out = ("
" + table(["项目", "来源与校验"], [[esc(k), esc(v)] for k, v in inp["provenance"]]))
def missing_llm(inp):
missing = (inp or {}).get("llm", {}).get("missing") or []
return "、".join(f"{m}（{MODES.get(m, m)}）" for m in missing)
def walltime_table(path):
rows = [line.rstrip("\n").split("\t") for line in Path(path).read_text().splitlines() if line.strip()]
return table(["步骤", "卡", "开始 (UTC)", "结束 (UTC)", "墙钟 (min)", "退出码"],
[[esc(r[0]), esc(r[5] if len(r) > 5 else "—"), esc(r[1]), esc(r[2]), num(int(r[3]) / 60, 1), esc(r[4])] for r in rows])
def main(argv=None):
p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
p.add_argument("--out", required=True, type=Path)
p.add_argument("--title", default="FastFill v2 评测报告")
p.add_argument("--summary", action="append", default=[], metavar="[LABEL=]SUMMARY.md")
p.add_argument("--inputs", type=Path, help="isambard_inputs.py inputs.json: arms, final-arm rule, provenance")
for split in ("val", "test"):
for kind in ("run", "compare", "control", "decode"):
p.add_argument(f"--{split}-{kind}", action="append", default=[], metavar="LABEL=JSON")
p.add_argument("--llm")
p.add_argument("--ablation", action="append", default=[])
p.add_argument("--ablation-sample", type=Path, help="sample_validation.py sample.json of the ablation rows")
p.add_argument("--selection-cohort", action="append", default=[], metavar="LABEL=ROWS_JSONL")
p.add_argument("--reference", action="append", default=[])
p.add_argument("--walltime", type=Path)
p.add_argument("--notes", type=Path)
a = p.parse_args(argv)
val_runs, test_runs = [load(x) for x in a.val_run], [load(x) for x in a.test_run]
val_compare, val_decode = [load(x) for x in a.val_compare], [load(x) for x in a.val_decode]
test_compare, test_decode = [load(x) for x in a.test_compare], [load(x) for x in a.test_decode]
val_control, test_control = [load(x) for x in a.val_control], [load(x) for x in a.test_control]
inp = json.loads(a.inputs.read_text()) if a.inputs else None
ablations, refs = [load(x) for x in a.ablation], [tuple(x.rsplit("=", 1)) for x in a.reference]
cohorts = [tuple(x.rsplit("=", 1)) for x in a.selection_cohort]
llm = json.loads(Path(a.llm).read_text()) if a.llm else None
runs = [("验证集", l, s) for l, s, _ in val_runs] + [("测试集", l, s) for l, s, _ in test_runs]
sample = json.loads(a.ablation_sample.read_text()) if a.ablation_sample else None
val_ph, test_ph = [c for _, c, _ in val_compare if placeholder(c)], [c for _, c, _ in test_compare if placeholder(c)]
parts = []
if val_ph or test_ph:
where = (["3.1 节比较表", "8.1 节决策表"] if val_ph else []) + (["第 4 节比较表"] if test_ph else [])
parts.append(f"
")
# 1 setup
rules = {split: Counter() for split in ("验证集", "测试集")}
for split, _, s in runs:
if not rules[split]:
rules[split].update({r: e["objects"] for r, e in s["strata"].get("rule", {}).items()})
fits = sorted({s["meta"]["baseline_fit"] for _, _, s in runs})
online = [f"{split}·{l}" for split, l, s in runs if not s["meta"].get("head_outputs")]
heads_note = (f"{len(runs) - len(online)}/{len(runs)} 个运行由保存的头输出在 CPU 上用同一份代码重解码（evaluate --from-head-outputs，spread 与 argmax 各一次）"
+ (f"；其余为在线解码的报告：{esc('、'.join(online))}。" if online else "。"))
val_rows = sorted({s["meta"]["rows"] for _, s, _ in val_runs})
rows_name = lambda r: f"验证集行（{Path(r).parent.name}/{Path(r).name}）"
bias = "".join(f"
{x}
" for r in val_rows if (x := bias_note(rows_name(r), scene_ids(r), cohorts)))
boots = sorted({s["meta"]["boot"] for _, _, s in runs})
parts.append("
归属：误差与基线归于标签物体（匹配目标），校验器标记归于检查点名的预测物体；声明位置取完整行的支撑声明（三字段投影不向模型显示）；
选择规则取 provenance.field_evidence[i].selection_rule（缺失记为 frozen_prep）。各集合的规则分布（物体数）："""
+ "；".join(f"{split}：" + "、".join(f"{esc(r)} {n}" for r, n in c.items()) for split, c in rules.items() if c) + """。
""")
if cohorts and val_rows:
parts.append("".join(cohort_flags(rows_name(r), scene_ids(r), cohorts) for r in val_rows)
+ (f"
选择偏差：{bias}
" if bias else ""))
for label, _, path in (x.rpartition("=") for x in a.summary):
if Path(path).is_file():
parts.append(f"autopilot SUMMARY.md（原文{'：' + esc(label) if label else ''}）"
+ markdown(Path(path).read_text()) + "")
if a.walltime and a.walltime.is_file():
parts.append("GPU / CPU 步骤墙钟时间（run_plan 记录）" + walltime_table(a.walltime) + "")
# 2 interface vs geometry
metrics = ("interface_pass_rate", "strict_ok_rate", "room_clean_rate", "clean_room_rate_incl_failures", "gt_room_clean_rate",
"object_collision_rate", "gt_object_collision_rate", "room_collision_rate")
parts.append("
2 接口通过率 vs 几何通过率
接口通过 = 返回了每个请求 ID 恰好一次、尺寸为正的布局；几何通过分两级："
"严格（所有硬检查 pass，unknown 也算不通过）与洁净（无硬违例）。真值列是同一校验器对标签本身的结果：是同一校验器下的参照，"
"不是上限（模型可能比标签更洁净）。
"
+ table(["集合", "运行"] + [NAMES[m] for m in metrics], [[split, esc(l)] + [cell(overall(s, m)) for m in metrics] for split, l, s in runs]))
# 3 validation: the decisions' basis
sec = f"
3 验证集 {DECISION}
"
if not (val_compare or val_decode):
sec += "
（未提供验证集比较：本报告不作决策）
"
sec += "
3.1 最终 vs 基线（两模型都能回答的房间）
" + ("".join(f"
{esc(l)}
" + compare_table(c, True) for l, c, _ in val_compare) or "
（未提供）
")
sec += "
3.2 默认解码：spread vs argmax
" + ("".join(f"
{esc(l)}
" + compare_table(c) for l, c, _ in val_decode) or "
（未提供）
")
if val_control:
sec += ("
3.3 最终 vs 对照（同一数据与配置，只差 yaw_cls；选择最终臂的配对证据）
"
+ "".join(f"
{esc(l)}
" + compare_table(c) for l, c, _ in val_control))
if val_runs:
sec += (f"
3.{4 if val_control else 3} 各运行在全部房间上的结果（含 > 128 物体房间；房间集合不同，非配对，仅描述）
" + full_set_table([(l, s) for l, s, _ in val_runs])
+ results_table([(l, s) for l, s, _ in val_runs])
+ "".join(f"分层结果（{esc(l)}）" + strata_tables(s, STRATA_METRICS) + "" for l, s, _ in val_runs))
parts.append(sec)
# 4 test: report only
sec = f"
"
+ (section_ablation(ablations, a.ablation_sample, sample, cohorts) or "
（未提供）
"))
parts.append(f"
7 RoomGenBench 五个房间 vs 真值 {REPORT_ONLY}
五个房间、单次预测：没有统计功效，只看量级与失败模式；"
"同一校验器也跑在真值上作为参照。
" + (section_reference(refs) or "
（未提供）
"))
# 8 limitations and decisions
lim, seen = [], set()
for split, label, s in runs:
c, key = s["checks"], (s["meta"]["eval_dir"], s["meta"]["outcomes"])
if key in seen:
continue
seen.add(key)
if not c["reproduces_report_exactly"]:
lim.append(f"{split}·{esc(label)}：逐物体重算未逐位复现 report.json，该运行的数字不可直接与报告对照。")
if c["rooms_validation_differs_from_stored"]:
lim.append(f"{split}·{esc(label)}：{c['rooms_validation_differs_from_stored']} 个房间的当前校验器结果与评测时存储的不同（校验器代码已变）；本报告用当前校验器。")
if c["requests"] - c["requests_with_layout"]:
lim.append(f"{split}·{esc(label)}：{c['requests'] - c['requests_with_layout']} 个请求没有布局（其中 {c['over_capacity_requests']} 个超出模型容量 "
"max_objects），误差只在有布局的房间上计算；接口通过率中计为失败。")
if any(s["meta"]["projection"] == "minimal" for _, _, s in runs):
lim.append("三字段投影不向模型显示支撑声明；“声明位置”分层用的是完整行的声明，作为真值位置标签。")
if missing_llm(inp):
lim.append(f"第 5 节缺少 LLM 模式 {esc(missing_llm(inp))}（答案不可得）：LLM 对比只含已有答案的模式。")
if llm:
lim += [f"LLM 行与检查点选择队列 {esc(Path(c['file']).name)} 重叠 {c['rows_in_cohort']}/{c['rows']}：用该队列选出的 FastFill 检查点在这些行上不是留出评测。"
for c in llm["selection_cohort"] if c["rows_in_cohort"]]
lim += [f"第 5 节 {esc(m)}：逐物体重算未逐位复现 report.json，该方法的数字不可直接与其报告对照。" for m, ok in llm["checks"].items() if not ok]
lim += [f"第 5 节 {esc(c['method'])}：predictions.jsonl 逐房间延迟的均值 {num(c['mean_latency_s'], 4)} s 与 summary.json 的 mean_latency_s "
f"{num(c['summary_mean_latency_s'], 4)} s 不一致。" for c in llm["cost"]
if c["kind"] == "llm" and abs(c["mean_latency_s"] - c["summary_mean_latency_s"]) > 1e-6]
if ablations and sample and (hits := overlaps(set(sample["scene_ids"]), cohorts)):
lim.append(f"{esc(sample_name(a.ablation_sample, sample))}与检查点选择队列重叠："
+ "、".join(f"「{esc(l)}」{k}/{len(sample['scene_ids'])}" for l, _, k in hits) + "：这些场景对相应检查点不是留出数据。")
family_rows = []
for split, label, s in runs:
fam = s["strata"].get("family", {})
worse = [f for f, e in fam.items() if (e["metrics"].get("bottom_center_error_m") or {}).get("mean") is not None
and e["metrics"]["bottom_center_error_m"]["mean"] > (e["metrics"].get("baseline:room_center_position") or {}).get("mean", math.inf)]
family_rows.append([split, esc(label), f"{len(worse)}/{len(fam)}", esc("、".join(worse)) or "—"])
parts.append("