#!/usr/bin/env python3 # SPDX-License-Identifier: Apache-2.0 """Write the goldens of the TT port from the fp32 CPU reference (``tt_bevdet.reference``), each cross-checked against the research pipeline's ONNX Runtime goldens (``research/bevdet``: SPEC section 6 and ``public_data/``). Two destinations (licensing: nuScenes-derived tensors never ship, PLAN D17 / research/DATASETS.md section 2): - ``--out-big`` (default ``research/bevdet/goldens``, never shipped): per-module taps (``reference.model`` tap names) plus the frame inputs (crop_u8, calibration, history, transforms, flag) of * ``sample0`` nuScenes key-frame of bevdet_vendor, reset frame (flag 0) [full taps] * ``nuscenes_port2hz_0103/kf09`` mini_val scene-0103, ego motion, history full (port2hz mode) [full taps] * ``nuscenes_port2hz_0916/kf09`` mini_val scene-0916 (Singapore rig) [gate taps] * ``pandaset_019_autoware/f38..f40`` the PandaSet sample through a fresh stream (Autoware semantics) [f38 full] * ``pandaset_019_port2hz/f40`` PandaSet 2 Hz with ego motion (non-identity AlignBEV) [full taps] and ``report.json`` (PCC / max-abs of every compared tensor, box agreement). - ``--out-small`` (PandaSet-derived, CC BY 4.0; default: where ``tests/paths.py`` finds the PandaSet sample, i.e. ``code/tt_bevdet/tests/goldens`` once the sample ships, else the git-ignored ``staging_samples_pandaset/tests/ goldens``): the boxes / detections (default params) and table statistics of PandaSet 019 f38-40, and, next to the sample manifests, ``pandaset_019_f38.reference.json`` (the smoke protocol's stored CPU reference: frame 38 on a fresh stream with ``params: {"score_threshold": 0.1}``, every vendor box). ``--no-big`` writes only these (the research goldens stay untouched). The shipped synthetic sample's reference is written by ``code/scripts/make_sample.py``. Run in the research venv (torch + onnxruntime not needed; the research goldens are read as data):: cd bundles/bevdet-p150 && OMP_NUM_THREADS=3 PYTHONPATH=$PWD/code \\ /home/ubuntu/experiments/tt-models/tools/research-venv/bin/python code/scripts/make_goldens.py --threads 3 """ from __future__ import annotations import argparse import json import os import subprocess import sys import time from pathlib import Path import numpy as np CODE = Path(__file__).resolve().parents[1] if str(CODE) not in sys.path: sys.path.insert(0, str(CODE)) from tt_bevdet import __version__ # noqa: E402 from tt_bevdet.host import calib as hc # noqa: E402 from tt_bevdet.host import inputs as hin # noqa: E402 from tt_bevdet.host import postprocess as hp # noqa: E402 from tt_bevdet.reference.config import HEAD_NAMES # noqa: E402 from tt_bevdet.tests import paths # noqa: E402 from tt_bevdet.ttaw import __version__ as TTAW_VERSION # noqa: E402 from tt_bevdet.ttaw.golden import TapRegistry, save_goldens # noqa: E402 from tt_bevdet.ttaw.metrics import match_detections, pcc # noqa: E402 GATE_TAPS = ["img.input", "depth", "feat", "depthnet.gate_ctx", "depthnet.gate_depth", "bev.curr", "bev.in", "head.*_0"] def bundle_commit() -> str: try: return subprocess.run(["git", "-C", str(CODE.parent), "rev-parse", "--short", "HEAD"], capture_output=True, text=True, check=True).stdout.strip() except (OSError, subprocess.CalledProcessError): return "unknown" def compare(ours: dict, research: dict, names) -> dict: out = {} for ours_key, res_key in names: if ours_key in ours and res_key in research: a, b = np.asarray(ours[ours_key]), np.asarray(research[res_key]) out[ours_key] = {"pcc": round(pcc(a, b), 10), "max_abs": float(np.abs(a - b).max())} return out def box_agreement(det, boxes, thr: float = 0.2) -> dict: g = [b for b in boxes if b["score"] >= thr] m = match_detections(det.boxes[:, :2], det.label_ids, det.scores, np.array([[b["x"], b["y"]] for b in g]).reshape(-1, 2), np.array([b["label"] for b in g]), np.array([b["score"] for b in g]), max_dist=0.5) return {"ours": len(det), "research": len(g), "matched": int(m.matched), "recall": round(float(m.recall), 6), "precision": round(float(m.precision), 6), "max_score_err": float(np.max(np.abs(m.score_errors))) if len(m.score_errors) else None} PAIRS = [("depth", "depth"), ("feat", "feat"), ("bev.curr", "curr_bevfeat"), ("bev.aligned", "aligned"), ("bev.in", "bev_in")] + [(f"head.{k}", k) for k in HEAD_NAMES] def save_frame(path: Path, taps: TapRegistry, inputs: dict, meta: dict) -> None: tensors = dict(taps.to_dict()) tensors.update({f"input.{k}": np.asarray(v) for k, v in inputs.items() if v is not None}) path.parent.mkdir(parents=True, exist_ok=True) save_goldens(path, tensors, meta, compress=True) def calib_from_cam_params(cp: np.ndarray) -> hc.Calibration: c = cp[0] K = np.stack([np.array([[r[0], 0, r[2]], [0, r[1], r[3]], [0, 0, 1]], np.float32) for r in c]) return hc.calibration_from_arrays(K, c[:, [15, 16, 17, 19, 20, 21, 23, 24, 25]].reshape(-1, 3, 3), c[:, [18, 22, 26]]) def run_explicit(ref, name: str, full_npz: Path, inputs_npz: Path, out: Path, full: bool, report: dict) -> None: """A research 'full' golden (explicit history) through ``run_frame``.""" with np.load(full_npz) as g, np.load(inputs_npz) as i: calib = hc.calibration_from_arrays(i["K"], i["cam2ego_R"], i["cam2ego_t"]) crop, adj, trs, flag = g["crop_u8"], g["adj_feats"], g["transforms"], int(g["flag"][0, 0]) research = {k: g[k] for k in ("depth", "feat", "curr_bevfeat", "aligned", "bev_in") + HEAD_NAMES} taps = TapRegistry(include=["*"] if full else GATE_TAPS) t0 = time.perf_counter() o = ref.run_frame(crop, calib, adj, trs, flag, taps=taps) boxes = hp.box_dicts(o.boxes) boxes_json = full_npz.with_name(full_npz.name.replace("_full.npz", "_boxes.json")) rep = {"seconds": round(time.perf_counter() - t0, 1), "flag": flag, "vs_research": compare(taps.to_dict(), research, PAIRS), "boxes": box_agreement(hp.boxes_to_detections(o.boxes, ref.cfg), json.loads(boxes_json.read_text()))} report[name] = rep meta = {"frame": name, "source": str(full_npz), "flag": flag, "tables": ref.tables(calib).stats()} save_frame(out, taps, {"crop_u8": crop, "K": calib.K, "R": calib.R, "t": calib.t, "cam_params": o.cam_params, "adj_feats": adj if full else None, "transforms": trs, "flag": np.array([[flag]])}, {**report["_meta"], **meta, "comparison": rep}) out.with_name(out.stem + "_boxes.json").write_text(json.dumps(boxes, indent=0)) print(f"{name}: {json.dumps(rep)}", flush=True) def main() -> int: ap = argparse.ArgumentParser(description=__doc__.split("\n\n")[0]) ap.add_argument("--out-big", type=Path, default=paths.golden_dir()) ap.add_argument("--out-small", type=Path, default=None) ap.add_argument("--no-big", action="store_true", help="pandaset set only: small goldens + reference JSON") ap.add_argument("--threads", type=int, default=3) ap.add_argument("--sets", default="sample0,pandaset,pandaset_motion,nuscenes") a = ap.parse_args() from tt_bevdet.reference.pipeline import BEVDetReference onnx = paths.weights_onnx() if onnx is None: raise SystemExit("bevdet_one_lt_d.onnx not found (BEVDET_WEIGHTS_DIR)") ref = BEVDetReference(onnx, threads=a.threads) sets = set(a.sets.split(",")) report: dict = {"_meta": {"generator": "code/scripts/make_goldens.py", "tt_bevdet": __version__, "bundle_commit": bundle_commit(), "ttaw": TTAW_VERSION, "onnx_sha256": ref.weights.sha256, "iou_mode": ref.iou_mode, "dtype": "float32", "date": time.strftime("%Y-%m-%d")}} big = a.out_big pdir = paths.pandaset_samples() small = a.out_small or (paths.pandaset_golden().parent if paths.pandaset_golden() else paths.STAGING_PANDASET / "tests" / "goldens") big.mkdir(parents=True, exist_ok=True) small.mkdir(parents=True, exist_ok=True) pub = paths.public_data() if "sample0" in sets and paths.research_golden("sample0.npz") is not None: g0 = paths.research_golden("sample0.npz") with np.load(g0) as g: crop = g["planes_bgr_u8"][:, :, :, :] from tt_bevdet.host.preprocess import crop_u8 crop = crop_u8(crop, ref.cfg) calib = calib_from_cam_params(g["cam_params"]) adj, trs, flag = g["adj_feats"], g["transforms"], int(g["flag"][0, 0]) research = {k: g[k] for k in ("depth", "feat", "curr_bevfeat", "aligned", "bev_in") + HEAD_NAMES} taps = TapRegistry(include=["*"]) t0 = time.perf_counter() o = ref.run_frame(crop, calib, adj, trs, flag, taps=taps) rep = {"seconds": round(time.perf_counter() - t0, 1), "flag": flag, "vs_research": compare(taps.to_dict(), research, PAIRS), "boxes": box_agreement(hp.boxes_to_detections(o.boxes, ref.cfg), json.loads(g0.with_name("sample0_boxes.json").read_text()))} report["sample0"] = rep save_frame(big / "sample0" / "sample0.npz", taps, {"crop_u8": crop, "K": calib.K, "R": calib.R, "t": calib.t, "cam_params": o.cam_params, "adj_feats": adj, "transforms": trs, "flag": np.array([[flag]])}, {**report["_meta"], "frame": "sample0", "license": "nuScenes CC BY-NC-SA 4.0: internal only", "tables": ref.tables(calib).stats(), "comparison": rep}) (big / "sample0" / "sample0_boxes.json").write_text(json.dumps(hp.box_dicts(o.boxes), indent=0)) print(f"sample0: {json.dumps(rep)}", flush=True) if "pandaset" in sets: if pdir is None: raise SystemExit("PandaSet sample manifests not found: code/scripts/stage_pandaset_sample.py") gdir = pub / "pandaset" / "goldens" / "autoware__ps019_f38-40" if pub else None ref.reset() small_rec = {"_doc": ("CPU fp32 reference (tt_bevdet.reference, vendor rotated IoU) on the PandaSet 019 " "sample, frames 38-40 through ONE fresh stream (Autoware semantics: zero history on " "frame 38, identity ego pose); detections at the default params. PandaSet: CC BY 4.0 + " "Dataset Terms (samples/pandaset_019/LICENSE.txt)."), "meta": report["_meta"], "frames": {}} for n in (38, 39, 40): kw = hin.load_sample(pdir / f"pandaset_019_f{n}.json") taps = TapRegistry(include=["*"] if n == 38 else GATE_TAPS) t0 = time.perf_counter() det = ref(**kw, taps=taps) frame = hin.prepare_frame(kw["images"], kw["calibration"], ref.cfg) vb = hp.postprocess({k: taps[f"head.{k}"] for k in HEAD_NAMES}, ref.cfg, iou_mode=ref.iou_mode) rep = {"seconds": round(time.perf_counter() - t0, 1)} if gdir is not None and (gdir / f"f{n}.npz").is_file(): with np.load(gdir / f"f{n}.npz") as g: rep["vs_research"] = compare(taps.to_dict(), {k: g[k] for k in g.files}, PAIRS) rep["boxes"] = box_agreement(det, json.loads((gdir / f"f{n}_boxes.json").read_text())) report[f"pandaset_019_autoware/f{n}"] = rep plan = det.meta["stream"] if not a.no_big: save_frame(big / "pandaset_019_autoware" / f"f{n}.npz", taps, {"crop_u8": frame.crop_u8, "K": frame.calib.K, "R": frame.calib.R, "t": frame.calib.t, "transforms": np.tile(np.array([1, 0, 0, 0, 1, 0], np.float32), (1, 8, 1)), "flag": np.array([[plan["flag"]]])}, {**report["_meta"], "frame": f"pandaset_019 f{n}", "stream": plan, "license": "PandaSet CC BY 4.0 + Dataset Terms", "comparison": rep}) (big / "pandaset_019_autoware" / f"f{n}_boxes.json").write_text(json.dumps(hp.box_dicts(vb), indent=0)) tab = ref.tables(frame.calib) small_rec["frames"][f"f{n}"] = {"stream_flag": plan["flag"], "tables": tab.stats(), "boxes": hp.box_dicts(vb), "detections": det.to_dicts()} if n == 38: # the smoke protocol's stored reference: frame 38, fresh stream, score_threshold 0.1 every = hp.boxes_to_detections(vb, ref.cfg, score_threshold=0.1, model=det.model, meta={k: v for k, v in det.meta.items() if k not in ("class_names", "autoware_labels", "score_threshold")}) body = dict(every.to_dict()) body["timing_ms"] = {} body["meta"]["request"] = {"stream": "fresh", "params": {"score_threshold": 0.1}} (pdir / "pandaset_019_f38.reference.json").write_text(json.dumps(body, indent=1) + "\n") print(f"pandaset f{n}: {json.dumps(rep)}", flush=True) (small / "pandaset_019_autoware_f38-40.json").write_text(json.dumps(small_rec, separators=(",", ":")) + "\n") if "pandaset_motion" in sets and pub is not None: full = pub / "pandaset" / "goldens" / "port2hz__ps019_2hz" / "f40_full.npz" inp = pub / "pandaset" / "inputs" / "port2hz__ps019_2hz" / "f40.npz" if full.is_file() and inp.is_file(): run_explicit(ref, "pandaset_019_port2hz/f40", full, inp, big / "pandaset_019_port2hz" / "f40.npz", True, report) if "nuscenes" in sets and pub is not None: for scene, is_full in (("scene-0103", True), ("scene-0916", False)): full = pub / "nuscenes" / "goldens" / f"port2hz__{scene}" / "kf09_full.npz" inp = pub / "nuscenes" / "inputs" / f"port2hz__{scene}" / "kf09.npz" if full.is_file() and inp.is_file(): run_explicit(ref, f"nuscenes_port2hz_{scene[-4:]}/kf09", full, inp, big / f"nuscenes_port2hz_{scene[-4:]}" / "kf09.npz", is_full, report) if not a.no_big: (big / "report.json").write_text(json.dumps(report, indent=1) + "\n") worst = min((v["pcc"] for r in report.values() if isinstance(r, dict) and "vs_research" in r for v in r["vs_research"].values()), default=None) print(f"done; worst PCC vs research = {worst}; report {big / 'report.json'}") return 0 if worst is None or worst >= 0.9999 else 1 if __name__ == "__main__": os.environ.setdefault("OMP_NUM_THREADS", "3") sys.exit(main())