#!/usr/bin/env python3 """Report the frozen FRESH-V2 qualification of the V7 RGB-D carriers (saved results only).""" # ruff: noqa: E501 -- Chinese scientific report prose import argparse import json import shutil from collections import defaultdict from pathlib import Path import numpy as np def read(root, name, default=None): path = root / (name + ".json") if not path.exists() and default is not None: return default return json.loads(path.read_text()) def stat(value): if value is None or value.get("mean") is None: return "UNDEFINED" lo, hi = value["ci95"] return f"{value['mean']:+.4f}，95% CI [{lo:+.4f}, {hi:+.4f}]" def per_scene(root): rows = read(root, "raw/fresh_readout_rows", []) refs = read(root, "raw/fresh_reference_rows", []) table = defaultdict(dict) for variant in ("C0", "C1", "C2"): values = defaultdict(list) for r in rows: if r["variant"] == variant and r["method"] == "direct" and r["readout"] == "RAW": values[r["scene_id"]].append(r["depth_absrel"]) for scene, v in values.items(): table[scene][variant] = float(np.mean(v)) constant = defaultdict(list) for r in refs: if r["reference"] == "REF_TRAIN_ABSREL_OPTIMAL" and r["readout"] == "RAW": constant[r["scene_id"]].append(r["depth_absrel"]) for scene, v in constant.items(): table[scene]["constant"] = float(np.mean(v)) return dict(table) def volume_of(scene): """Hypersim volume (ai_VVV_SSS -> VVV); any other id is its own volume.""" parts = scene.split("_") return parts[1] if len(parts) == 3 and parts[0] == "ai" else scene def volume_bootstrap(per_scene_values, draws=10000, seed=20260928): """Descriptive: mean of volume means, percentile CI over resampled volumes.""" volumes = defaultdict(list) for scene, value in per_scene_values.items(): volumes[volume_of(scene)].append(value) means = np.array([np.mean(v) for _, v in sorted(volumes.items())]) rng = np.random.default_rng(seed) samples = means[rng.integers(0, len(means), (draws, len(means)))].mean(axis=1) lo, hi = np.percentile(samples, [2.5, 97.5]) return { "mean": float(means.mean()), "ci95": [float(lo), float(hi)], "n_volumes": len(means), "volumes_positive": int((means > 0).sum()), "unit": "volume (descriptive only)", "draws": draws, "seed": seed, } def run(root, docs=None): root = Path(root) results = read(root, "static_results") qualification = read(root, "qualification_results") controls = read(root, "wrong_scene_results")["contrasts"] obs = read(root, "observability_analysis") reference = read(root, "reference_results", {}) or None contract = read(root, "qualification_contract") v7 = Path(contract["v7_root"]) v7_summary = (v7 / "terminal_summary.txt").read_text().splitlines() v7_fields = dict(line.split("=", 1) for line in v7_summary if "=" in line) if results["primary"] != "C1_SURFACE16": raise PermissionError("Frozen V2 estimator primary label expected (C1 = V7 primary)") groups = results["groups"] bounds = results["surface_gain"] wrong, shuffle = controls["C1|wrong_scene"], controls["C1|spatial_shuffle"] bounds_status = qualification["SURFACE_TRAINING_STATUS"] specificity = ( "SUPPORTED" if wrong["ci95"][0] > 0 and shuffle["ci95"][0] > 0 else "NOT_ESTABLISHED" ) static = qualification["STATIC_DEV_STATUS"] c1_reference = reference["carriers"]["V7_C1"]["RAW"]["depth_absrel"] if reference else None reference_gate = ( "SUPPORTED" if c1_reference is not None and c1_reference["reference_label"] == "ABOVE" else "NOT_ESTABLISHED" ) gates = (bounds_status, specificity, static, reference_gate) verdict = "QUALIFIED" if all(g == "SUPPORTED" for g in gates) else "NOT_QUALIFIED" reference_status = ( reference["summary"]["CARRIER_REFERENCE_STATUS"] if reference else "NOT_RECORDED" ) versus = c1_reference["vs_REF_TRAIN_ABSREL_OPTIMAL"] if c1_reference else None sensitivity = reference.get("sensitivity_excluding_no_hit_scenes", {}) if reference else {} volume_bounds = volume_bootstrap(bounds["per_scene"]) volume_reference = volume_bootstrap(versus["per_scene"]) if versus else None table = per_scene(root) fields = { "FRESH_SCENES": len(results["scene_ids"]), "FRESH_VOLUMES": len({volume_of(s) for s in results["scene_ids"]}), "FRESH_SCENE_IDS": results["scene_ids"], "FROZEN_BOUNDS_QUERY_ABSREL_FRESH": round( groups["C0|direct"]["query_metrics"]["depth_absrel"]["mean"], 6 ), "DEPTH_BOUNDS_QUERY_ABSREL_FRESH": round( groups["C1|direct"]["query_metrics"]["depth_absrel"]["mean"], 6 ), "DEPTH_BOUNDS_TRIM1_QUERY_ABSREL_FRESH": round( groups["C2|direct"]["query_metrics"]["depth_absrel"]["mean"], 6 ) if "C2|direct" in groups else "NOT_EVALUATED", "BOUNDS_GAIN_FRESH": round(bounds["mean"], 6), "BOUNDS_GAIN_FRESH_CI": bounds["ci95"], "BOUNDS_GAIN_FRESH_IMPROVED_TIED_WORSE": [ bounds["positive"], bounds["tied"], bounds["negative"], ], "BOUNDS_GAIN_FRESH_VOLUME_BOOTSTRAP": volume_bounds, "BOUNDS_GAIN_DEV_V7": v7_fields.get("BOUNDS_GAIN"), "C1_VS_CONSTANT_FRESH": round(versus["mean"], 6) if versus else None, "C1_VS_CONSTANT_FRESH_CI": versus["ci95"] if versus else None, "C1_VS_CONSTANT_FRESH_IMPROVED_TIED_WORSE": versus["improved_tied_worse"] if versus else None, "C1_VS_CONSTANT_FRESH_VOLUME_BOOTSTRAP": volume_reference, "C1_WRONG_SCENE_DAMAGE_FRESH": round(wrong["mean"], 6), "C1_WRONG_SCENE_DAMAGE_FRESH_CI": wrong["ci95"], "C1_SHUFFLE_DAMAGE_FRESH": round(shuffle["mean"], 6), "C1_SHUFFLE_DAMAGE_FRESH_CI": shuffle["ci95"], "C1_FULL_CONTEXT_GAIN_FRESH": round(results["full_context_gain"]["C1"]["mean"], 6), "C1_FULL_CONTEXT_GAIN_FRESH_CI": results["full_context_gain"]["C1"]["ci95"], "BOUNDS_STATUS_FRESH": bounds_status, "SCENE_SPECIFICITY_STATUS_FRESH": specificity, "STATIC_STATUS_FRESH": static, "REFERENCE_STATUS_FRESH": reference_gate, "SEED_ROBUSTNESS_FRESH": qualification["SEED_ROBUSTNESS"], "CARRIER_REFERENCE_STATUS_FRESH": reference_status, "C1_REFERENCE_LABEL_FRESH": c1_reference["reference_label"] if c1_reference else None, "C1_REFERENCE_LABEL_FRESH_HIT_SCENES": sensitivity.get("carriers", {}) .get("V7_C1", {}) .get("RAW", {}) .get("depth_absrel", {}) .get("reference_label", "NO_NO_HIT_SCENE"), "FRESH_QUALIFICATION_STATUS": verdict, "CORE_A_STATIC_CARRIER_RGBD_TRACK": verdict, "DYNAMIC_TTT_NEXT_STAGE_ALLOWED": verdict == "QUALIFIED", "FINAL_HOLDOUT_OPENED": False, "TRAINING_RUN": False, "DEVICE": "cpu", } lines = [ "# EXP-3D-RGBD-FRESH-QUALIFICATION-V2", "", f"`FRESH_QUALIFICATION_STATUS={verdict}`。", "", "对 V7 已封存的 RGB-D carrier（每个 seed 各自的 DEV 选择，不重新训练、不重新选择）在第二个独立队列 FRESH-V2 上做预注册资格验证。" f"FRESH-V2 共 {fields['FRESH_SCENES']} 个场景、{fields['FRESH_VOLUMES']} 个 volume：来自官方 train 划分、从未被观察、从未被任何实验使用，volume 与 TRAIN72 和 DEV 都不重叠，每个 volume 至多 2 个。" "每个变体用自己的 bounds 规则；所有 fresh 状态封存之后才读取 query GT；统计用 V7 修订 1 的估计器（冻结的 V2 估计器去掉一条不适用的命中比例断言）；受保护的官方 final holdout 未打开。", "", "| 项目 | FRESH-V2 | V7 DEV（参照） |", "|---|---|---|", f"| C0 冻结 bounds，query AbsRel | {fields['FROZEN_BOUNDS_QUERY_ABSREL_FRESH']:.4f} | {v7_fields.get('FROZEN_BOUNDS_QUERY_ABSREL')} |", f"| C1 测得深度 bounds，query AbsRel | {fields['DEPTH_BOUNDS_QUERY_ABSREL_FRESH']:.4f} | {v7_fields.get('DEPTH_BOUNDS_QUERY_ABSREL')} |", f"| BOUNDS_GAIN | {stat(bounds)} | {v7_fields.get('BOUNDS_GAIN')} |", f"| C1 相对 AbsRel 最优常数 | {stat(versus)} | 见 V7 报告 |", f"| C1 wrong-scene 损伤 | {stat(wrong)} | {v7_fields.get('C1_WRONG_SCENE_DAMAGE')} |", f"| C1 shuffle 损伤 | {stat(shuffle)} | {v7_fields.get('C1_SHUFFLE_DAMAGE')} |", f"| 3 视角相对 anchor | {stat(results['full_context_gain']['C1'])} | {v7_fields.get('C1_FULL_CONTEXT_GAIN')} |", f"| BOUNDS_STATUS | {bounds_status} | {v7_fields.get('BOUNDS_STATUS')} |", f"| SCENE_SPECIFICITY_STATUS | {specificity} | {v7_fields.get('SCENE_SPECIFICITY_STATUS')} |", f"| STATIC_STATUS | {static} | {v7_fields.get('STATIC_DEV_STATUS')} |", f"| REFERENCE_STATUS（C1 在全部场景上优于常数） | {reference_gate} | {'SUPPORTED' if v7_fields.get('C1_ABOVE_REFERENCE_ALL_SCENES') == 'true' else 'NOT_ESTABLISHED'} |", "", "## 逐场景（RAW AbsRel，seed 与角色平均）", "", "| 场景 | volume | 常数（TRAIN 拟合） | C0 | C1 | C2 |", "|---|---|---:|---:|---:|---:|", ] for scene in sorted(table): t = table[scene] lines.append( f"| {scene} | {volume_of(scene)} | {t.get('constant', float('nan')):.3f} | {t.get('C0', float('nan')):.3f} | " f"{t.get('C1', float('nan')):.3f} | {t.get('C2', float('nan')):.3f} |" ) lines += [ "", "## 描述性补充", "", f"- 按 volume 聚类的 bootstrap（同一 volume 内的场景可能相关）：BOUNDS_GAIN {stat(volume_bounds)}，{volume_bounds['volumes_positive']}/{volume_bounds['n_volumes']} 个 volume 为正；" + ( f"C1 相对常数 {stat(volume_reference)}，{volume_reference['volumes_positive']}/{volume_reference['n_volumes']} 个 volume 为正。" if volume_reference else "C1 相对常数未记录。" ), f"- CARRIER_REFERENCE_STATUS={reference_status}；排除 NO_HIT 场景后 C1 的标签：{fields['C1_REFERENCE_LABEL_FRESH_HIT_SCENES']}。", "- 区域诊断 C0−C1 条件 AbsRel：" + "；".join(f"{r}={stat(obs['gains'][r])}" for r in ("OBS0", "OBS1", "OBS2PLUS")) + "。", "", "## 结论", "", ( "四个冻结 gate 在第二个独立队列上全部成立：Core A 的静态 carrier 在 RGB-D 轨道上通过资格验证（首次同时满足机制复现与绝对精度优于平凡基线）。按项目规则，Core B（动态 TTT 写入）的下一阶段可以开启；官方 final holdout 仍保持关闭。" if verdict == "QUALIFIED" else "冻结 gate 未在第二个独立队列上全部成立：Core A 在 RGB-D 轨道上尚未通过资格验证，Core B 保持关闭。未通过的 gate 见上表。" ), ] summary = ( "RGBD FRESH QUALIFICATION V2\n" + "\n".join( f"{k}={str(v).lower() if isinstance(v, bool) else v}" for k, v in fields.items() ) + "\n" ) (root / "README.md").write_text("\n".join(lines) + "\n") (root / "STATUS.md").write_text("```text\n" + summary + "```\n") (root / "terminal_summary.txt").write_text(summary) (root / "fresh_qualification_results.json").write_text( json.dumps({**fields, "per_scene": table}, indent=2, ensure_ascii=False) + "\n" ) if docs: docs = Path(docs) docs.mkdir(parents=True, exist_ok=True) for p in root.iterdir(): if p.is_file() and p.suffix in (".json", ".md", ".txt"): shutil.copyfile(p, docs / p.name) print(summary) return fields if __name__ == "__main__": parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--root", type=Path, required=True) parser.add_argument("--docs", type=Path) args = parser.parse_args() run(args.root, args.docs)