code/scripts/make_samples.py

#!/usr/bin/env python3
# SPDX-License-Identifier: Apache-2.0
"""Build the PandaSet sample of bevformer-p150 from the public-data workflow's redistributable PandaSet sample
(``research/bevformer/public_data/shippable/pandaset_019_f30-40``; CC BY 4.0 + PandaSet Dataset Terms).

It does NOT ship yet: the user has not approved PandaSet-derived samples (PLAN.md 6.3, research/DATASETS.md N5), so
the default destination is the git-ignored ``staging_samples_pandaset/`` of the bundle checkout (``--dest code``
writes into ``code/tt_bevformer`` once it may ship; ``staging_samples_pandaset/README.md`` has the swap-in steps).

Writes ``<dest>/samples/pandaset_019/`` (the 18 unchanged JPEGs, LICENSE.txt, SHA256SUMS),
``<dest>/samples/pandaset_019_f{30,35,40}.json`` (one request manifest per frame: the six images, the frame's
calibration inline -- each camera's camera -> LIDAR_TOP transform at its own timestamp and the 1920x1080 K --, the
stream with the ego pose, and the CAN dynamics in the training layout) and the static calibration preset
``<dest>/calib/pandaset_019.json`` (frame 30's transforms). Workspace tool: needs the research data.

    python3 code/scripts/make_samples.py [--src <shippable dir>] [--dest staging|code]
"""
from __future__ import annotations

import argparse
import hashlib
import json
import shutil
from pathlib import Path

HERE = Path(__file__).resolve().parent
PKG = HERE.parent / "tt_bevformer"
DEFAULT_SRC = Path("/home/ubuntu/experiments/tt-models/research/bevformer/public_data/shippable/pandaset_019_f30-40")
CAMS = ["CAM_FRONT", "CAM_FRONT_RIGHT", "CAM_FRONT_LEFT", "CAM_BACK", "CAM_BACK_LEFT", "CAM_BACK_RIGHT"]
ATTRIBUTION = ("Contains data from PandaSet (Scale AI and Hesai), https://pandaset.org, licensed under CC BY 4.0 and the "
               "PandaSet Dataset Terms (pandaset_019/LICENSE.txt). Changes: JPEG files renamed to the network camera "
               "slots (bytes unchanged); calibration and ego motion converted to the BEVFormer input contract "
               "(research/bevformer/public_data README section 2). Scale AI and Hesai do not endorse this work. Cite: "
               "P. Xiao et al., PandaSet: Advanced Sensor Suite Dataset for Autonomous Driving, ITSC 2021.")


def sha256(path: Path) -> str:
    return hashlib.sha256(path.read_bytes()).hexdigest()


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--src", type=Path, default=DEFAULT_SRC)
    ap.add_argument("--dest", choices=["staging", "code"], default="staging")
    a = ap.parse_args()
    global PKG
    if a.dest == "staging":
        PKG = HERE.parents[1] / "staging_samples_pandaset"
    meta = json.loads((a.src / "sample.json").read_text())
    if meta["camera_order"] != CAMS:
        raise SystemExit(f"unexpected camera order {meta['camera_order']}")
    dst = PKG / "samples" / "pandaset_019"
    (dst / "images").mkdir(parents=True, exist_ok=True)
    shutil.copyfile(a.src / "LICENSE.txt", dst / "LICENSE.txt")
    sums = []
    for fr in meta["frames"]:
        f = int(fr["pandaset_frame"])
        for cam in CAMS:
            src = a.src / fr["images"][cam]["file"]
            if sha256(src) != fr["images"][cam]["sha256"]:
                raise SystemExit(f"{src}: sha256 differs from sample.json")
            out = dst / "images" / f"f{f}_{cam}.jpg"
            shutil.copyfile(src, out)
            sums.append(f"{sha256(out)}  images/{out.name}")
    sums.append(f"{sha256(dst / 'LICENSE.txt')}  LICENSE.txt")
    (dst / "SHA256SUMS").write_text("\n".join(sums) + "\n")

    for i, fr in enumerate(meta["frames"]):
        f = int(fr["pandaset_frame"])
        cb = fr["can_bus_abs"]
        cams = {cam: {"intrinsics": fr["K_1920x1080"][k], "T_ref_from_camera": fr["T_lidar_cam"][k]}
                for k, cam in enumerate(CAMS)}
        manifest = {
            "_doc": (f"BEVFormer sample request: PandaSet sequence 019 frame {f} (6 cameras, 1920x1080 JPEG, unmodified "
                     "bytes; the node resizes them to 1600x900 and this port rescales K). model(**tt_bevformer."
                     "load_sample(path)) once the API runs; the calibration is the frame's own (each camera's camera -> "
                     "LIDAR_TOP transform at its camera timestamp: the node's per-camera ego-motion compensation), the "
                     "stream carries the ego pose and the 'ego' block the CAN dynamics in BEVFormer's training layout "
                     "(derived from the poses and the GPS, research/bevformer/public_data README section 2). Frames 30, "
                     "35, 40 form one sequence 0.5 s apart, history reset at frame 30. " + ATTRIBUTION),
            "frame": f,
            "images": {cam: f"pandaset_019/images/f{f}_{cam}.jpg" for cam in CAMS},
            "calibration": {"frame_id": "LIDAR_TOP", "image_size": [1920, 1080], "scale_intrinsics": True,
                            "cameras": cams},
            "stream": {"id": "pandaset_019", "reset": i == 0, "timestamp_s": fr["lidar_timestamp_s"],
                       "T_world_from_ego": fr["ego2global"]},
            "ego": {"accel": cb[7:10], "rotation_rate": cb[10:13], "velocity": cb[13:16]},
            "camera_stamps_s": fr["camera_timestamps_s"],
            "image_bgr_sha256_1600x900": fr["image_bgr_sha256_1600x900"],
        }
        (PKG / "samples" / f"pandaset_019_f{f}.json").write_text(json.dumps(manifest, indent=1) + "\n")

    f0 = meta["frames"][0]
    preset = {
        "_doc": ("Calibration preset of the PandaSet sequence 019 sample: per camera the intrinsics of the "
                 "1920x1080 image AS SENT (rescaled to the node's 1600x900, scale_intrinsics=true) and "
                 "T_ref_from_camera = camera optical frame -> the network's virtual nuScenes LIDAR_TOP (frame_id "
                 "LIDAR_TOP; PandaSet pose axes x right / y forward, 1.84 m above the road), taken at frame 30 (static "
                 "rig approximation: the per-frame manifests carry each frame's time-compensated transforms). "
                 + ATTRIBUTION),
        "frame_id": "LIDAR_TOP",
        "image_size": [1920, 1080],
        "scale_intrinsics": True,
        "cameras": {cam: {"intrinsics": f0["K_1920x1080"][k], "T_ref_from_camera": f0["T_lidar_cam"][k]}
                    for k, cam in enumerate(CAMS)},
    }
    (PKG / "calib").mkdir(exist_ok=True)
    (PKG / "calib" / "pandaset_019.json").write_text(json.dumps(preset, indent=1) + "\n")
    print(f"wrote {dst} ({len(sums)} files), 3 manifests, calib/pandaset_019.json")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())