Blame for code/scripts/make_sample.py

#!/usr/bin/env python3
# SPDX-License-Identifier: Apache-2.0
"""Generate the shipped sample of streampetr-p150: ``code/tt_streampetr/samples/synthetic_5cam*`` (a synthetic
five-camera frame made by this script, Apache-2.0 like the code), its calibration preset ``calib/synthetic_5cam.json``,
the stored CPU-reference output the container smoke compares with, and the small head golden of the host tests.
CPU only; run it in the research venv (torch, PIL)::

    cd bundles/streampetr-p150
    PY=/home/ubuntu/experiments/tt-models/tools/research-venv/bin/python
    OMP_NUM_THREADS=4 PYTHONPATH=$PWD/code $PY code/scripts/make_sample.py optimise --out <dir>     # ~0.5-1 h
    OMP_NUM_THREADS=4 PYTHONPATH=$PWD/code $PY code/scripts/make_sample.py finalise --images <dir>/images_f32.npy

Why a synthetic frame. The redistributable real data of this model (PandaSet, CC BY 4.0) may not ship until the user
approves PandaSet-derived samples (PLAN.md 6.3, research/DATASETS.md N5), nuScenes is non-commercial, the T4 training
data are not public, and plain synthetic images give the network nothing to detect (bench.py's seeded colour fields
give zero objects): an empty reference is a weak smoke test. So the sample is a **test pattern**: five rendered road
views of a generic rig; the pixels around five chosen objects (one per class) in three cameras were then optimised
against the fp32 CPU reference (gradient ascent through the whole network: VoVNet-99-eSE + CPFPN, the head's tokens,
the six decoder layers and the branches; cold-start memory, the node's first frame) until every decoder layer reports
the five objects at their places with large margins and every other (layer, query, class) score stays well below the
node's per-class thresholds (0.36-0.43). The images look like a road with noisy texture patches; they show no real
objects. What they give is a stable reference: every published box sits far from its threshold, so bf16 numerics
cannot flip it.

- **Rig** (:data:`RIG`, round numbers of our own, nuScenes-like layout without the back camera): five 640x480 pinhole
  cameras 1.6 m above the road, yaws 0, +-55, +-110 deg, f = 480 px (67 deg horizontal field of view), principal
  point (319.5, 239.5). The images are already 640x480, so the node's resize / crop is the identity and the
  pre-processing is ``(x - mean) / std`` exactly (checked in ``finalise``).
- **Objects** (:data:`TARGETS`, ``base_link``): a car, a bicycle and a pedestrian ahead (CAM_FRONT), a bus front-left
  (CAM_FRONT_LEFT) and a truck front-right (CAM_FRONT_RIGHT), at rest. The two back cameras keep the plain render.
- **Optimisation**: Adam on the uint8-scaled float images of the three cameras; each target is matched to the query
  whose last-layer box is close and confident in any class (sticky unless another query is clearly better); its class logit is pushed up and its centre pulled to
  the target in every layer; every other logit is pushed down. Gradients are confined to the targets' projected boxes
  (dilated) plus a small share outside them.
- **Reference**: the ``/predict`` body of ``tt_streampetr.reference.StreamPETRReference`` for the sample on a fresh
  stream with the default parameters (``timing_ms`` emptied), computed from the PNG files exactly as the API reads them.
"""
from __future__ import annotations

import argparse
import hashlib
import json
import math
import sys
import time
from pathlib import Path
from typing import Dict, List, Tuple

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_streampetr.reference import config as C  # noqa: E402

PKG = CODE / "tt_streampetr"
NAME = "synthetic_5cam"
SRC_W, SRC_H = 640, 480
FOCAL, CX, CY = 480.0, 319.5, 239.5
STAMP = 1000.0                                   # the stream's anchor stamp (any value; one cold-start frame)
# camera optical frame (x right, y down, z forward) -> base_link (x forward, y left, z up)
R_OPT = np.array([[0.0, 0.0, 1.0], [-1.0, 0.0, 0.0], [0.0, -1.0, 0.0]])
# name -> (x, y, z, yaw_deg): camera position in base_link (metres), yaw about +z
RIG: Dict[str, Tuple[float, float, float, float]] = {
    "CAM_FRONT": (1.80, 0.00, 1.60, 0.0),
    "CAM_FRONT_LEFT": (1.60, 0.60, 1.60, 55.0),
    "CAM_BACK_LEFT": (0.90, 0.80, 1.60, 110.0),
    "CAM_FRONT_RIGHT": (1.60, -0.60, 1.60, -55.0),
    "CAM_BACK_RIGHT": (0.90, -0.80, 1.60, -110.0),
}
# class, x, y (base_link, m), yaw (rad, base_link), (length, width, height); bottom on the road, at rest
TARGETS: List[Tuple[str, float, float, float, Tuple[float, float, float]]] = [
    ("CAR", 12.0, 0.4, 0.0, (4.5, 1.9, 1.6)),
    ("BICYCLE", 9.5, 3.2, 0.0, (1.8, 0.6, 1.5)),
    ("PEDESTRIAN", 8.5, -2.8, 1.57, (0.7, 0.7, 1.75)),
    ("BUS", 9.0, 13.0, 0.0, (11.5, 2.9, 3.4)),
    ("TRUCK", 9.0, -13.0, 0.0, (7.5, 2.5, 3.2)),
]
# Tried and dropped: a bus behind on the right (CAM_BACK_RIGHT) and a truck 22 m ahead never rose above score 0.06 in
# 20-30 steps, while objects 15 m out in the front-side cameras converge in a few steps
OPT_CAMS = ("CAM_FRONT", "CAM_FRONT_LEFT", "CAM_FRONT_RIGHT")


# --------------------------------------------------------------------------------------------- rig and render

def camera_T(x: float, y: float, z: float, yaw_deg: float) -> np.ndarray:
    """4x4 T_base_link_from_camera_optical for a level camera at (x, y, z) looking along yaw."""
    a = math.radians(yaw_deg)
    rz = np.array([[math.cos(a), -math.sin(a), 0.0], [math.sin(a), math.cos(a), 0.0], [0.0, 0.0, 1.0]])
    T = np.eye(4)
    T[:3, :3] = rz @ R_OPT
    T[:3, 3] = (x, y, z)
    return T


def calibration() -> Dict:
    """The calibration envelope of the rig (``calib/synthetic_5cam.json``)."""
    cams = {}
    K = [[FOCAL, 0.0, CX], [0.0, FOCAL, CY], [0.0, 0.0, 1.0]]
    for name in C.CAMERA_ORDER:
        T = camera_T(*RIG[name])
        cams[name] = {"intrinsics": K, "P": [row + [0.0] for row in K],
                      "T_ref_from_camera": [[round(float(v), 12) for v in row] for row in T]}
    return {"frame_id": "base_link", "image_size": [SRC_W, SRC_H], "cameras": cams}


def render_road(K, T, seed: int = 0, noise: float = 1.5) -> np.ndarray:
    """A camera's view of a flat world: sky gradient, a 14 m wide asphalt road along base_link x with lane markings
    (solid edges, dashed separators every 3.5 m), sand-coloured shoulders, haze toward the horizon. (480, 640, 3)
    uint8 RGB (the renderer of bevdet-p150 / bevformer-p150 at this rig's size)."""
    rng = np.random.default_rng(seed)
    u, v = np.meshgrid(np.arange(SRC_W, dtype=np.float64), np.arange(SRC_H, dtype=np.float64), indexing="xy")
    d = np.stack([u, v, np.ones_like(u)], -1) @ np.linalg.inv(np.asarray(K, np.float64)).T
    R, c = np.asarray(T)[:3, :3], np.asarray(T)[:3, 3]
    db = d @ R.T
    el = np.arctan2(db[..., 2], np.hypot(db[..., 0], db[..., 1]))
    sky = np.clip(el / 0.6, 0, 1)[..., None]
    img = (1 - sky) * np.array([205.0, 210.0, 214.0]) + sky * np.array([120.0, 160.0, 205.0])
    ground = db[..., 2] < -1e-6
    with np.errstate(divide="ignore", invalid="ignore"):
        t = np.where(ground, -c[2] / np.where(ground, db[..., 2], -1.0), np.inf)
        px, py = c[0] + t * db[..., 0], c[1] + t * db[..., 1]
    dist = np.hypot(px, py)
    g = ground & (dist < 80.0)
    road = g & (np.abs(py) < 7.0)
    side = g & ~road
    tex = rng.normal(0, 4.0, size=(SRC_H, SRC_W, 1))
    img[road] = np.array([92.0, 94.0, 96.0]) + tex[road]
    img[side] = np.array([140.0, 132.0, 118.0]) + tex[side]
    for yl, dashed in ((6.8, False), (-6.8, False), (0.0, True), (3.5, True), (-3.5, True)):
        on = g & (np.abs(py - yl) < 0.08)
        if dashed:
            on &= np.mod(np.where(g, px, 0.0), 9.0) < 3.0
        img[on] = np.array([225.0, 225.0, 220.0])
    haze = np.clip((np.where(g, dist, 0.0) - 30.0) / 50.0, 0, 1)[..., None] * g[..., None]
    img = img * (1 - 0.5 * haze) + 0.5 * haze * np.array([200.0, 205.0, 210.0])
    img += rng.normal(0, noise, size=img.shape)
    return np.clip(np.rint(img), 0, 255).astype(np.uint8)


def renders(cal: Dict) -> Dict[str, np.ndarray]:
    return {n: render_road(cal["cameras"][n]["intrinsics"], np.array(cal["cameras"][n]["T_ref_from_camera"]), seed=i)
            for i, n in enumerate(C.CAMERA_ORDER)}


def box_corners(x: float, y: float, yaw: float, l: float, w: float, h: float) -> np.ndarray:
    c, s = math.cos(yaw), math.sin(yaw)
    return np.array([[x + c * dx * l - s * dy * w, y + s * dx * l + c * dy * w, dz] for dz in (0.0, h)
                     for dx, dy in ((0.5, 0.5), (0.5, -0.5), (-0.5, -0.5), (-0.5, 0.5))])


def target_mask(cal: Dict, name: str, grow: float = 0.4) -> np.ndarray:
    """[480, 640] float32: the pixels covered by the targets' projected 3D boxes (dilated) in one camera."""
    m = np.zeros((SRC_H, SRC_W), np.float32)
    K = np.asarray(cal["cameras"][name]["intrinsics"], np.float64)
    Tinv = np.linalg.inv(np.asarray(cal["cameras"][name]["T_ref_from_camera"], np.float64))
    for _, x, y, yaw, (l, w, h) in TARGETS:
        pc = box_corners(x, y, yaw, l, w, h) @ Tinv[:3, :3].T + Tinv[:3, 3]
        if (pc[:, 2] < 0.5).any():
            continue
        uv = pc @ K.T
        uv = uv[:, :2] / uv[:, 2:3]
        (u0, v0), (u1, v1) = uv.min(0), uv.max(0)
        du, dv = max((u1 - u0) * grow, 12.0), max((v1 - v0) * grow, 12.0)
        if u1 + du < 0 or u0 - du > SRC_W - 1 or v1 + dv < 0 or v0 - dv > SRC_H - 1:
            continue
        r0, r1 = int(max(0, v0 - dv)), int(min(SRC_H, v1 + dv + 1))
        c0, c1 = int(max(0, u0 - du)), int(min(SRC_W, u1 + du + 1))
        m[r0:r1, c0:c1] = 1.0
    return m


def manifest_dict(files: Dict[str, str]) -> Dict:
    return {
        "_doc": ("streampetr-p150 sample: a synthetic five-camera test pattern generated by code/scripts/make_sample.py "
                 "(rendered road views whose pixels around five chosen objects in CAM_FRONT, CAM_FRONT_LEFT and "
                 "CAM_FRONT_RIGHT were optimised so that the fp32 reference reports them; no real scene). The images "
                 "are 640x480 (the node's resize / crop is the identity); the calibration is the preset synthetic_5cam "
                 "(base_link); one cold-start frame on a fresh stream at an identity ego pose. "
                 "model(**tt_streampetr.load_sample(path)). Generated data, Apache-2.0."),
        "images": files,
        "calibration": {"preset": NAME},
        "camera_stamps_s": {n: STAMP for n in C.CAMERA_ORDER},
        "stream": {"id": NAME, "timestamp_s": STAMP, "reset": True,
                   "T_world_from_ego": [[1.0, 0.0, 0.0, 0.0], [0.0, 1.0, 0.0, 0.0], [0.0, 0.0, 1.0, 0.0],
                                        [0.0, 0.0, 0.0, 1.0]]},
        "targets": [{"label": c, "frame": "base_link", "center": [x, y, round(s[2] / 2, 3)], "size": list(s),
                     "yaw": yaw} for c, x, y, yaw, s in TARGETS],
    }


# ------------------------------------------------------------------------------------------------ optimise

def optimise(a) -> int:
    """Gradient ascent on the 640x480 images of :data:`OPT_CAMS` (module docstring)."""
    import torch
    import torch.nn.functional as F

    from tt_streampetr.host.inputs import prepare_frame
    from tt_streampetr.host.temporal import StreamState
    from tt_streampetr.reference.model import StreamPETRNet
    from tt_streampetr.tests import paths

    torch.manual_seed(0)
    torch.set_num_threads(a.threads)
    out = Path(a.out)
    out.mkdir(parents=True, exist_ok=True)
    cfg = C.default_config()
    cal = calibration()
    imgs = renders(cal)
    wdir = Path(a.weights) if a.weights else paths.weights_dir()
    net = StreamPETRNet(wdir, cfg, threads=a.threads)
    frame = prepare_frame([{"camera": n, "image": imgs[n]} for n in C.CAMERA_ORDER], cal, None, cfg)
    plan = StreamState(cfg).begin(STAMP, np.eye(4), [STAMP] * 5)
    pre = cfg.normalization
    from tt_streampetr.host.preprocess import PRESETS
    mean = torch.tensor(PRESETS[pre].mean, dtype=torch.float32).view(1, 3, 1, 1)
    std = torch.tensor(PRESETS[pre].std, dtype=torch.float32).view(1, 3, 1, 1)
    assert PRESETS[pre].planes == "rgb"

    def prep(rgb):                                   # [n, 480, 640, 3] RGB float -> [n, 3, 480, 640] normalised
        return (rgb.permute(0, 3, 1, 2) - mean) / std

    head = net.head
    with torch.no_grad():
        rig = frame.rig
        pos, cone = net.pe(rig.img_metas_pad, rig.intrinsics, rig.img2lidar)
        sa = head.spatial_alignment_coeffs(cone)
        mem_ego, mem_ref = head.memory_pre_update(plan.pre, plan.Einv)
        t = head.temporal(mem_ref, mem_ego, plan.pre["pre_memory_timestamp"], plan.pre["pre_memory_embedding"],
                          plan.pre["pre_memory_velo"])
        q = head.queries(t)
        fixed = [n for n in C.CAMERA_ORDER if n not in OPT_CAMS]
        base = {n: torch.as_tensor(imgs[n].astype(np.float32)) for n in C.CAMERA_ORDER}
        feat_fixed = net.backbone(prep(torch.stack([base[n] for n in fixed]))[None])[0]
    x0 = torch.stack([base[n] for n in OPT_CAMS])
    x = (torch.as_tensor(np.load(a.resume).astype(np.float32)) if a.resume else x0.clone()).requires_grad_(True)
    masks = torch.as_tensor(np.stack([target_mask(cal, n) for n in OPT_CAMS]))[..., None]
    print("mask pixels", [int(m.sum()) for m in masks], flush=True)
    opt = torch.optim.Adam([x], lr=a.lr)
    tg = [(C.CLASS_NAMES.index(c), xx, yy, s[2] / 2.0) for c, xx, yy, _, s in TARGETS]

    def heads(xc):
        f_opt = net.backbone(prep(xc)[None])[0]
        order = {n: i for i, n in enumerate(OPT_CAMS)}
        fi = {n: i for i, n in enumerate(fixed)}
        feat = torch.stack([f_opt[order[n]] if n in order else feat_fixed[fi[n]] for n in C.CAMERA_ORDER])[None]
        memory, kpos = head.tokens(feat, pos, cone, sa_coeffs=sa)
        outs_dec = head.decoder(q["tgt"], q["query_pos"], q["temp_memory"], q["temp_pos"], memory, kpos)
        cls, reg = head.predict(outs_dec, q["reference_points"])
        return cls[:, 0], reg[:, 0]                  # [6, 900, 5], [6, 900, 10]

    hist = []
    nq = cfg.num_query + cfg.num_propagated
    for it in range(a.iters):
        t0 = time.time()
        opt.zero_grad()
        cls, crd = heads(x)
        last_c, last_r = cls[-1], crd[-1]
        with torch.no_grad():                         # greedy matching on the last layer (module docstring)
            taken, match = set(), []
            obj = torch.sigmoid(last_c).max(1).values          # the query's strongest class: an object is
            for j, (k, tx, ty, _) in enumerate(tg):              # easier to relabel than to create
                cost = ((last_r[:, 0] - tx).abs() + (last_r[:, 1] - ty).abs() - 2.0).clamp(min=0.0) - 5.0 * obj
                for qq in taken:
                    cost[qq] = float("inf")
                best = int(torch.argmin(cost))
                prev_q = hist[-1]["queries"][j] if hist else None
                if prev_q is not None and prev_q not in taken and float(cost[prev_q]) <= float(cost[best]) + 1.0 \
                        and abs(float(last_r[prev_q, 0]) - tx) + abs(float(last_r[prev_q, 1]) - ty) < a.sticky_m:
                    best = prev_q                                # sticky unless another query is clearly better
                taken.add(best)
                match.append(best)
        loss_t, loss_c, scores = 0.0, 0.0, []
        for (k, tx, ty, tz), qq in zip(tg, match):
            lg = cls[:, qq, k]                                         # [6] every layer
            others = torch.cat([cls[:, qq, :k], cls[:, qq, k + 1:]], 1)
            w = torch.tensor([0.3, 0.3, 0.6, 0.8, 1.0, 1.0])
            loss_t = loss_t + (w * (F.softplus(4.0 - lg) + F.softplus(others.max(1).values - lg + 4.0))).sum()
            tgt = torch.tensor([tx, ty, tz])
            loss_c = loss_c + (w[:, None] * F.smooth_l1_loss(crd[:, qq, 0:3], tgt.expand(6, 3), reduction="none")
                               ).sum() + 0.2 * (crd[:, qq, 8:10] ** 2).sum() \
                + F.relu(0.7 - (crd[:, qq, 6] ** 2 + crd[:, qq, 7] ** 2).sqrt()).sum()
            scores.append(round(float(torch.sigmoid(lg[-1].detach())), 3))
        keep = torch.ones(nq, dtype=torch.bool)
        keep[match] = False
        bg = cls[:, keep]
        # every background logit above logit(0.1) = -2.2 is pushed down individually (hinge), the rest on average
        loss_bg = 4.0 * F.softplus(bg + 4.0).mean() + a.bg_weight * F.relu(bg + 2.2).sum() \
            + F.softplus(bg + 3.0).max()
        tv = ((x[:, 1:] - x[:, :-1]).abs().mean() + (x[:, :, 1:] - x[:, :, :-1]).abs().mean()) / 255.0
        dev = (((x - x0) / 255.0) ** 2 * (1.0 - masks)).mean()
        loss = loss_t + loss_c + loss_bg + a.tv * tv + 20.0 * dev
        loss.backward()
        with torch.no_grad():
            x.grad.mul_(masks + a.outside * (1.0 - masks))
            opt.step()
            x.clamp_(0.0, 255.0)
        bmax = float(torch.sigmoid(bg.detach().max()))
        with torch.no_grad():                         # where the strongest background logit is
            flat = int(torch.argmax(bg))
            li, rem = divmod(flat, bg.shape[1] * bg.shape[2])
            qi, ci = divmod(rem, bg.shape[2])
            qq = torch.nonzero(keep)[qi].item()
            worst = [li, qq, C.CLASS_NAMES[ci], round(float(crd[li, qq, 0]), 1), round(float(crd[li, qq, 1]), 1)]
            n_hi = int((bg > -2.2).sum())
        lay_min = [round(float(torch.sigmoid(cls[:, qq, k].detach()).min()), 3) for (k, *_), qq in zip(tg, match)]
        hist.append({"it": it, "loss": round(float(loss.detach()), 4), "targets": round(float(loss_t.detach()), 4),
                     "centre": round(float(loss_c.detach()), 4), "bg_max": round(bmax, 4), "bg_worst": worst,
                     "bg_n_above_0.1": n_hi, "scores": scores,
                     "layer_min": lay_min, "queries": match, "s": round(time.time() - t0, 1)})
        print(json.dumps(hist[-1]), flush=True)
        if it % 5 == 4 or it == a.iters - 1:
            np.save(out / "images_f32.npy", x.detach().numpy())
            (out / "history.json").write_text(json.dumps({"args": {k: str(v) for k, v in vars(a).items()},
                                                          "history": hist}, indent=0))
        if min(scores) > a.stop_score and bmax < a.stop_bg:
            print(f"stop: every target > {a.stop_score}, background max {bmax:.3f} < {a.stop_bg}", flush=True)
            np.save(out / "images_f32.npy", x.detach().numpy())
            break
    (out / "history.json").write_text(json.dumps({"args": {k: str(v) for k, v in vars(a).items()},
                                                  "history": hist}, indent=0))
    return 0


# ------------------------------------------------------------------------------------------------ finalise

def sha256(data: bytes) -> str:
    return hashlib.sha256(data).hexdigest()


def write_preset() -> Path:
    """``calib/synthetic_5cam.json`` (the rig of :func:`calibration` with its provenance note)."""
    preset = {"_doc": ("Calibration preset of the shipped synthetic sample (samples/synthetic_5cam): a generic "
                       "five-camera rig with round numbers of our own (code/scripts/make_sample.py RIG): 640x480 "
                       "pinhole cameras 1.6 m above the road, yaws 0 / +-55 / +-110 deg, f = 480 px, principal point "
                       "(319.5, 239.5), P = [K | 0]. intrinsics = K of the image as sent; T_ref_from_camera = camera "
                       "optical frame -> base_link. Generated data, Apache-2.0."),
              **calibration()}
    p = PKG / "calib" / f"{NAME}.json"
    p.write_text(json.dumps(preset, indent=1) + "\n")
    return p


def finalise(a) -> int:
    import torch
    from PIL import Image

    from tt_streampetr.host import inputs as hin
    from tt_streampetr.reference.pipeline import StreamPETRReference
    from tt_streampetr.tests import paths

    cal = calibration()
    base = renders(cal)
    opt_imgs = np.clip(np.rint(np.load(a.images)), 0, 255).astype(np.uint8)
    samples = PKG / "samples"
    img_dir = samples / NAME
    img_dir.mkdir(parents=True, exist_ok=True)
    files = {}
    for name in C.CAMERA_ORDER:
        im = opt_imgs[OPT_CAMS.index(name)] if name in OPT_CAMS else base[name]
        Image.fromarray(im).save(img_dir / f"{name}.png", format="PNG", optimize=True)
        files[name] = f"{NAME}/{name}.png"
    write_preset()
    man_path = samples / f"{NAME}.json"
    man_path.write_text(json.dumps(manifest_dict(files), indent=1) + "\n")
    torch.set_num_threads(a.threads)
    wdir = Path(a.weights) if a.weights else paths.weights_dir()
    ref = StreamPETRReference(wdir, threads=a.threads)
    req = hin.load_sample(man_path)
    det, outs, plan, img = ref(**req, return_outputs=True)
    assert plan.fresh and plan.skipped is None
    # the node's resize is the identity at 640x480: the network input is (x - mean) / std of the PNG pixels
    from tt_streampetr.host.preprocess import PRESETS
    p = PRESETS[ref.cfg.normalization]
    for i, name in enumerate(C.CAMERA_ORDER):
        px = np.asarray(Image.open(img_dir / f"{name}.png").convert("RGB"), np.float32)
        exp = ((px - np.array(p.mean, np.float32)) / np.array(p.std, np.float32)).transpose(2, 0, 1)
        assert np.array_equal(img[0, i], exp), name
    body = det.to_dict()
    body["timing_ms"] = {}
    (samples / f"{NAME}.reference.json").write_text(json.dumps(body, indent=1) + "\n")
    np.savez_compressed(PKG / "tests" / "goldens" / f"{NAME}_head.npz", all_cls_scores=outs["all_cls_scores"],
                        all_bbox_preds=outs["all_bbox_preds"])
    # margins: the published boxes and the strongest score that is not published
    cls = np.asarray(outs["all_cls_scores"], np.float64)[0]          # [5, 5400]
    s = 1.0 / (1.0 + np.exp(-cls))
    pub = body["detections"]
    thr = np.asarray(ref.cfg.score_thresholds)
    print(f"{NAME}: {len(pub)} detections; num_pre_nms {body['meta']['num_pre_nms']}")
    for d in pub:
        print("  ", d["label"], d["score"], d["center"], d["size"], d["yaw"], d["velocity"])
    ratio = s / thr[:, None]
    print("top score / threshold ratios", np.round(np.sort(ratio.reshape(-1))[::-1][:40], 3).tolist())
    print("img sha256", {n: sha256(img[0, i].tobytes())[:16] for i, n in enumerate(C.CAMERA_ORDER)})
    sizes = {q.name: q.stat().st_size for q in sorted(img_dir.glob("*.png"))}
    print("png bytes", sizes, sum(sizes.values()))
    return 0


def main() -> int:
    ap = argparse.ArgumentParser(description=__doc__.split("\n\n")[0])
    sub = ap.add_subparsers(dest="cmd", required=True)
    o = sub.add_parser("optimise", help="gradient ascent on three cameras -> <out>/images_f32.npy")
    o.add_argument("--out", required=True)
    o.add_argument("--iters", type=int, default=80)
    o.add_argument("--lr", type=float, default=2.0, help="Adam step in uint8 pixel units")
    o.add_argument("--outside", type=float, default=0.1, help="gradient scale outside the target masks")
    o.add_argument("--tv", type=float, default=2.0, help="total-variation weight")
    o.add_argument("--bg-weight", type=float, default=1.0, help="hinge weight of background logits above 0.1")
    o.add_argument("--sticky-m", type=float, default=4.0, help="keep a matched query while its box is this close")
    o.add_argument("--stop-score", type=float, default=0.8)
    o.add_argument("--stop-bg", type=float, default=0.12)
    o.add_argument("--resume", help="start from an images_f32.npy")
    f = sub.add_parser("finalise", help="images -> sample PNGs, preset, manifest, CPU reference, head golden")
    f.add_argument("--images", required=True)
    for p in (o, f):
        p.add_argument("--weights", help="weights dir (default: tests/paths.weights_dir())")
        p.add_argument("--threads", type=int, default=4)
    sub.add_parser("preset", help="write calib/synthetic_5cam.json only")
    a = ap.parse_args()
    if a.cmd == "preset":
        print(write_preset())
        return 0
    return optimise(a) if a.cmd == "optimise" else finalise(a)


if __name__ == "__main__":
    sys.exit(main())