Download code/scripts/make_sample.py from changh95/streampetr-p150: direct link, hf CLI and curl.
- Browser
- Download file 24.2 kB
-
https://huggingface.co/changh95/streampetr-p150/resolve/main/code/scripts/make_sample.py
- Command line
-
hf download hf://changh95/streampetr-p150/code/scripts/make_sample.py
-
curl -L -o make_sample.py https://huggingface.co/changh95/streampetr-p150/resolve/main/code/scripts/make_sample.py
24.2 kB
| #!/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()) | |