| |
| |
| """Demo media of the card (``media/``): the p150 output, drawn over the stored fp32 CPU reference. |
| |
| # 1. on the device: run the p150 on inputs and keep the /predict bodies (one model load for all inputs) |
| bin/devrun -t 1200 -- python code/scripts/make_demo.py run --json-dir <dir> INPUT [INPUT ...] |
| # 2. no device: the p150 boxes over the CPU reference's boxes, bird's-eye view |
| python code/scripts/make_demo.py compare --input code/tt_transfusion/samples/test_pcd.npz \\ |
| --tt <dir>/test_pcd.tt.json --out media/transfusion_test_pcd_tt_vs_cpu.png |
| # 3. no device: an animated bird's-eye view of a frame sequence (p150 boxes only) |
| python code/scripts/make_demo.py gif --tt-dir <dir> --out media/<name>.gif INPUT [INPUT ...] |
| |
| ``run``: ``TransFusion.from_pretrained`` (ETH dispatch, 12x10, the traced graph; ``--weights-dir`` or the HF cache) |
| runs every input through the Python API, which returns exactly what ``POST /predict`` returns, and writes |
| ``<json-dir>/<stem>.tt.json`` (the body plus the device description). Inputs are the API's point inputs (``.npz`` / |
| ``.npy`` / ``.pcd``); a plain ``(M, 5)`` ``.npy`` array is read as x, y, z, intensity, time_lag (the research and |
| public-data layout). |
| |
| ``compare``: the reference is ``--reference``, else ``<stem>.reference.json`` next to the input. Matching is the rule |
| of the accuracy gates (same label, BEV centre distance <= 0.5 m, ``ttaw.metrics.match_detections``); the caption |
| states the result. Drawing: ``tt_transfusion.viz.render_bev`` (p150 boxes in colour, reference boxes as light outlines, |
| unmatched boxes ringed and tagged). |
| |
| numpy + Pillow only (no device for ``compare`` / ``gif``). |
| """ |
| from __future__ import annotations |
|
|
| import argparse |
| import json |
| import sys |
| from pathlib import Path |
| from typing import Any, Dict, List, Optional |
|
|
| import numpy as np |
|
|
| HERE = Path(__file__).resolve().parent |
| CODE = HERE.parent |
| if str(CODE) not in sys.path: |
| sys.path.insert(0, str(CODE)) |
|
|
| FIELDS5 = ("x", "y", "z", "intensity", "time_lag") |
| TITLE = "TransFusion-L (Autoware lidar_transfusion v2.1)" |
|
|
|
|
| def load_cloud(path: Path): |
| """The API input of a file; a plain (M, 5) float array is x, y, z, intensity, time_lag.""" |
| from tt_transfusion import io as tio |
|
|
| if path.suffix == ".npy": |
| a = np.load(path, allow_pickle=False) |
| if a.dtype.names is None and a.ndim == 2 and a.shape[1] == 5: |
| return tio.PointCloud(np.ascontiguousarray(a, np.float32), FIELDS5) |
| return str(path) |
|
|
|
|
| def xyzt(path: Path) -> np.ndarray: |
| """(N, 4) x, y, z, time_lag for drawing.""" |
| from tt_transfusion import io as tio |
|
|
| cloud = tio.load_points(load_cloud(path)) |
| names = list(cloud.fields) |
| xyz = cloud.select(("x", "y", "z")) |
| lag = cloud.select(("time_lag",)) if "time_lag" in names else np.zeros((len(xyz), 1), np.float32) |
| return np.concatenate([np.asarray(xyz, np.float32), np.asarray(lag, np.float32).reshape(-1, 1)], 1) |
|
|
|
|
| def rows(dets: List[Dict[str, Any]]): |
| boxes = np.array([d["center"] + d["size"] + [d["yaw"]] for d in dets], np.float32).reshape(-1, 7) |
| return boxes, np.array([d["label_id"] for d in dets], np.int64), np.array([d["score"] for d in dets], np.float32) |
|
|
|
|
| def agreement(tt: Dict[str, Any], ref: Dict[str, Any]) -> Dict[str, Any]: |
| from tt_transfusion.ttaw.metrics import match_detections |
|
|
| m = match_detections(*rows(tt["detections"]), *rows(ref["detections"]), max_dist=0.5) |
| out = m.to_dict() |
| out.update(dev=len(tt["detections"]), ref=len(ref["detections"]), |
| max_score_err=float(max(m.score_errors, default=0.0))) |
| return out |
|
|
|
|
| def cmd_run(a) -> int: |
| from tt_transfusion import TransFusion |
|
|
| a.json_dir.mkdir(parents=True, exist_ok=True) |
| kw = {"weights_dir": str(a.weights_dir)} if a.weights_dir else {} |
| with TransFusion.from_pretrained(device_id=a.device_id, **kw) as model: |
| device = model.info.get("device") |
| for path in a.inputs: |
| out = model(load_cloud(path)) |
| body = out.to_dict() |
| body["device"] = device |
| (a.json_dir / f"{path.stem}.tt.json").write_text(json.dumps(body, indent=1, default=float)) |
| print(f"{path.name}: {len(out)} detections, pillars {body['meta']['num_pillars']}, timing_ms " |
| f"{ {k: round(v, 1) for k, v in out.timing_ms.items()} }", flush=True) |
| return 0 |
|
|
|
|
| def cmd_compare(a) -> int: |
| from tt_transfusion.viz import render_bev |
|
|
| body = json.loads(a.tt.read_text()) |
| ref_path = a.reference or a.input.with_name(f"{a.input.stem}.reference.json") |
| ref = json.loads(Path(ref_path).read_text()) |
| ag = agreement(body, ref) |
| dev = body.get("device") or {} |
| where = f"dispatch {dev.get('dispatch', 'eth')}, grid {dev.get('grid', '12x10')}" |
| meta = body["meta"] |
| lines = [f"p150 ({where}) vs the fp32 CPU reference of the same network: {ag['matched']} of {ag['ref']} " |
| f"reference detections matched (same label, centres <= 0.5 m), {ag['dev']} on the p150; max |score " |
| f"difference| of matched pairs {ag['max_score_err']:.4f}.", |
| f"Input: {a.describe or a.input.name}; {meta.get('num_points', 0):,} points, " |
| f"{meta['num_pillars']:,} pillars. Forward up, rings every 10 m."] |
| if a.lines: |
| lines += a.lines |
| pts = xyzt(a.input) |
| img = render_bev(pts, body["detections"], reference=ref["detections"], |
| title=f"{TITLE} on one Blackhole p150 vs the fp32 CPU reference", lines=lines, |
| x_range=tuple(a.x_range) if a.x_range else None, y_range=tuple(a.y_range) if a.y_range else None, |
| width=a.width) |
| a.out.parent.mkdir(parents=True, exist_ok=True) |
| img.save(a.out, optimize=True) |
| print(f"{a.out}: {json.dumps(ag)}") |
| return 0 |
|
|
|
|
| def cmd_gif(a) -> int: |
| from PIL import Image |
|
|
| from tt_transfusion.viz import render_bev |
|
|
| frames = [] |
| for i, path in enumerate(a.inputs): |
| body = json.loads((a.tt_dir / f"{path.stem}.tt.json").read_text()) |
| label = (a.labels[i] if a.labels and i < len(a.labels) else path.stem).replace("_", " ") |
| img = render_bev(xyzt(path), body["detections"], title=f"{TITLE} on one Blackhole p150: {label}", |
| lines=[f"{len(body['detections'])} detections (p150 output); {a.caption}"], |
| x_range=tuple(a.x_range), y_range=tuple(a.y_range), width=a.width) |
| frames.append(img.convert("P", palette=Image.ADAPTIVE, colors=64)) |
| a.out.parent.mkdir(parents=True, exist_ok=True) |
| frames[0].save(a.out, save_all=True, append_images=frames[1:], duration=a.ms, loop=0, optimize=True) |
| print(f"{a.out}: {len(frames)} frames, {a.out.stat().st_size / 1e6:.2f} MB") |
| return 0 |
|
|
|
|
| def main(argv: Optional[List[str]] = None) -> int: |
| ap = argparse.ArgumentParser(description=__doc__.split("\n\n")[0]) |
| sub = ap.add_subparsers(dest="cmd", required=True) |
| r = sub.add_parser("run", help="run the p150 on inputs (device)") |
| r.add_argument("inputs", nargs="+", type=Path) |
| r.add_argument("--json-dir", type=Path, required=True) |
| r.add_argument("--weights-dir", type=Path) |
| r.add_argument("--device-id", type=int, default=0) |
| c = sub.add_parser("compare", help="p150 over the CPU reference (no device)") |
| c.add_argument("--input", type=Path, required=True) |
| c.add_argument("--tt", type=Path, required=True, help="the p150 body (make_demo.py run)") |
| c.add_argument("--reference", type=Path, help="default: <stem>.reference.json next to the input") |
| c.add_argument("--out", type=Path, required=True) |
| c.add_argument("--describe", help="input description for the caption") |
| c.add_argument("--lines", nargs="*", help="extra caption lines") |
| c.add_argument("--x-range", type=float, nargs=2) |
| c.add_argument("--y-range", type=float, nargs=2) |
| c.add_argument("--width", type=int, default=640) |
| g = sub.add_parser("gif", help="animated bird's-eye view of p150 outputs (no device)") |
| g.add_argument("inputs", nargs="+", type=Path) |
| g.add_argument("--tt-dir", type=Path, required=True) |
| g.add_argument("--out", type=Path, required=True) |
| g.add_argument("--labels", nargs="*", help="one per input; \"_\" prints as a space") |
| g.add_argument("--caption", default="") |
| g.add_argument("--x-range", type=float, nargs=2, default=(-40.0, 60.0)) |
| g.add_argument("--y-range", type=float, nargs=2, default=(-45.0, 45.0)) |
| g.add_argument("--width", type=int, default=560) |
| g.add_argument("--ms", type=int, default=100, help="frame duration") |
| a = ap.parse_args(argv) |
| return {"run": cmd_run, "compare": cmd_compare, "gif": cmd_gif}[a.cmd](a) |
|
|
|
|
| if __name__ == "__main__": |
| sys.exit(main()) |
|
|
