# status: verified 2026-08-27 against the active robium-apps Robot Navigation catalog
schema_version: "1"
assets:
  - id: world.aws-small-house
    kind: world
    name: AWS RoboMaker Small House
    storage: pointer
    manifest: worlds/aws-small-house/asset.yaml
  - id: world.tugbot-warehouse
    kind: world
    name: Tugbot in Warehouse
    storage: pointer
    manifest: worlds/tugbot-warehouse/asset.yaml

# Canonical test assets: the verified catalog

Well-known public assets to test robotics apps against: worlds, robot models,
datasets, and recordings. Every entry carries a citation line saying how and
when it was verified; keep that discipline when adding entries (see the end
of this file). Licenses shown were read from the upstream API/card on the
verification date; re-confirm at adoption time before vendoring.

## Worlds

| Asset | Upstream | License | Canonical for | Verified |
|---|---|---|---|---|
| TurtleBot3 House | [ROBOTIS-GIT/turtlebot3_simulations](https://github.com/ROBOTIS-GIT/turtlebot3_simulations) | Apache-2.0 | Indoor/house nav testing; the Nav2-tutorial-canonical world | GitHub API 2026-07-18: 515★, pushed 2025-07-14, SPDX Apache-2.0 |
| Tugbot in Warehouse (OpenRobotics) | [Gazebo Fuel](https://app.gazebosim.org/OpenRobotics/fuel/worlds/Tugbot%20in%20Warehouse) | CC-BY-NC-ND-4.0; pointer only | Industrial/warehouse scenes in modern gz | Fuel world-details API + clean-cache version 2 ZIP SHA/entrypoint check, 2026-08-27 |
| AWS RoboMaker Small House | [aws-robotics/aws-robomaker-small-house-world](https://github.com/aws-robotics/aws-robomaker-small-house-world) | MIT at pinned commit | Richer furnished multi-room house; **Gazebo-Classic era: verify modern-gz load before adopting** | Commit `ff9631ca6d1db9c1ba656498151464b5ab74aafe`; pinned license byte comparison + clean-cache archive SHA/entrypoint check, 2026-08-27 |
| AWS RoboMaker Small Warehouse | [aws-robotics/aws-robomaker-small-warehouse-world](https://github.com/aws-robotics/aws-robomaker-small-warehouse-world) | check repo at adoption | The most-recognized open warehouse world; **same Gazebo-Classic caveat** | GitHub API 2026-07-18: 487★, last pushed 2024-07-26 |

## Robot models

| Asset | Upstream | License | Canonical for | Verified |
|---|---|---|---|---|
| TurtleBot3 burger/waffle | [ROBOTIS-GIT/turtlebot3_simulations](https://github.com/ROBOTIS-GIT/turtlebot3_simulations) | Apache-2.0 | Mobile-base/Nav2 testing | GitHub API 2026-07-18 (same repo as house world) |
| Unitree Go2 (MJCF) | [google-deepmind/mujoco_menagerie](https://github.com/google-deepmind/mujoco_menagerie), dir `unitree_go2` | per-model (repo SPDX: NOASSERTION; check the model's own license file) | Quadruped testing in MuJoCo | GitHub API 2026-07-18: dir listing confirms unitree_go2 |
| Unitree Go2/G1 (URDF, ROS) | [unitreerobotics/unitree_ros](https://github.com/unitreerobotics/unitree_ros) | BSD-3-Clause | Official Unitree descriptions for ROS/Gazebo use | GitHub API 2026-07-18: 1,477★, pushed 2026-07-08, SPDX BSD-3-Clause |
| Unitree G1 (MJCF) | mujoco_menagerie dir `unitree_g1`; also [unitreerobotics/unitree_mujoco](https://github.com/unitreerobotics/unitree_mujoco) (BSD-3-Clause) | per-model (Menagerie) | Humanoid testing in MuJoCo | GitHub API 2026-07-18: dir listing confirms unitree_g1; unitree_mujoco 1,089★ |
| SO-101 arm (MJCF) | [google-deepmind/mujoco_menagerie](https://github.com/google-deepmind/mujoco_menagerie), dir `robotstudio_so101`; derived from [TheRobotStudio/SO-ARM100](https://github.com/TheRobotStudio/SO-ARM100), `Simulation/SO101/` | Apache-2.0 | The LeRobot-ecosystem arm in MuJoCo; Menagerie adds manipulation collision geometry, a camera mount, and ready scenes | GitHub contents API and both upstream READMEs on 2026-09-07 at Menagerie commit `8161bba264d7fa7c99ca301e91e7fb44737676ad`; model requires MuJoCo 3.1.3+ |

## Datasets

| Asset | Upstream | License | Canonical for | Verified |
|---|---|---|---|---|
| svla_so101_pickplace | [lerobot/svla_so101_pickplace](https://huggingface.co/datasets/lerobot/svla_so101_pickplace) | Apache-2.0 (card tag) | The canonical SO-101 sample: official LeRobot org, SmolVLA-tutorial dataset | HF API 2026-07-18: 42 likes (most-liked SO-101 dataset); every higher-download alternative is a 0-like community upload. Runner-up (sim-MuJoCo shape, inspect before trust): szk1ck/so101-pickplace-sim-mujoco |
| pusht | [lerobot/pusht](https://huggingface.co/datasets/lerobot/pusht) | MIT (card tag) | The CI-sized LeRobot standard for train-smoke tests | HF API 2026-07-18: 18,181 downloads, 54 likes |

## Recordings (replay fixtures)

| Asset | Upstream | License | Canonical for | Verified |
|---|---|---|---|---|
| EuRoC MAV sequences | [ETH ASL dataset page](https://projects.asl.ethz.ch/datasets/doku.php?id=kmavvisualinertialdatasets) | check page at adoption | Canonical VIO/drone rosbags that "just replay" | HTTP 200 on 2026-07-18; re-verify format/license at adoption |
| TUM RGB-D sequences | [TUM CVG dataset page](https://cvg.cit.tum.de/data/datasets/rgbd-dataset) | check page at adoption | The RGB-D SLAM classic, rosbag downloads | HTTP 200 on 2026-07-18; re-verify at adoption |
| Foxglove sample MCAPs | [foxglove.dev/examples](https://foxglove.dev/examples) | check page at adoption | Viz-tooling scenarios (foxglove/rerun skills) | HTTP 200 on 2026-07-18; re-verify at adoption |

## Known gaps

- **Nav2/TB3 rosbag search was inconclusive:** a 2026-07-18 public-source/API
  search did not identify a canonical bag suitable for Robium's navigation
  regression. Self-recording from a seeded scenario was the practical fallback,
  not proof that no suitable public bag exists. Search again before generating
  a new fixture.
- **Drone assets unpicked**: no vertical yet (px4 is future work); the
  matrix row is an honest gap, not an oversight.
- **Legged robots have models here but no robium skill coverage**: Go2/G1
  are test assets awaiting a future legged vertical.

## Adding an entry

1. Verify the asset against its live source (API call, direct fetch), never
   from memory, and write the citation line with method + date.
2. Read the license from the upstream repo/card at adoption time; an asset
   whose license forbids redistribution stays pointer-mode only.
3. State what the asset is canonical *for*; an entry without a clear
   testing role doesn't belong in the catalog.

# The standard test-assets layout

Use one small catalog plus one manifest directory per asset. Pointer mode is
the default for large worlds, models, and datasets: Git owns metadata and
license evidence, while an ignored cache holds downloaded bytes.

## Layout

    test-assets/
      README.md
      catalog.yaml
      scripts/fetch_assets.py
      worlds/
        example-world/
          asset.yaml
          LICENSE
      cache/                 # ignored in pointer mode
      datasets/              # per-asset manifests follow the same pattern
      bags/                  # seeded project recordings, not arbitrary downloads
      goldens/               # tolerance-band outputs from verified scenarios

## Catalog schema

    schema_version: "1"
    assets:
      - id: world.example
        kind: world
        name: Example World
        storage: pointer
        manifest: worlds/example-world/asset.yaml

IDs are stable and names are human-facing. The resolver verifies that the
catalog and manifest agree on `id`, `kind`, `name`, and `storage`.

## Per-asset manifest schema

    schema_version: "1"
    id: world.example
    kind: world
    name: Example World
    revision: "1"           # local schema revision
    storage: pointer
    license:
      id: SPDX-ID
      file: LICENSE          # checked-in evidence beside asset.yaml
      url: https://upstream.example/license
    verification:
      date: "2026-08-27"
      method: Clean-cache archive fetch, SHA-256, entrypoint, and license check.
    source:
      type: git-archive
      repository: https://upstream.example/project
      revision: immutable-upstream-revision
      url: https://upstream.example/archive.tar.gz
      sha256: 64-lowercase-hex-characters
      archive: tar.gz        # tar.gz | zip
      strip_prefix: project-revision
    entrypoints:
      world: worlds/example.world

Rules:

- Pin an immutable upstream revision and the exact archive SHA-256. A branch,
  `latest`, or download URL without a checksum is not a pointer lock.
- Keep a license record next to every manifest. Compare it with the pinned
  upstream source at adoption time; restrictive assets remain pointer-only.
- Record the verification date and concrete method. A clean-cache fetch must
  validate safe extraction and every declared entrypoint.
- Do not add a fixture without an active application or test consumer.
- Vendored mode may commit bytes only under a documented size budget and with
  redistribution terms that permit it.

## Dataset slices and goldens

- Take deterministic slices, such as the first N episodes, and record N and
  the source revision. Confirm the reduced dataset still loads.
- Compare simulated goldens with explicit tolerances and seeds. Exact byte
  checks apply only to pure replay, not re-run physics.

#!/usr/bin/env python3
# /// script
# requires-python = ">=3.12"
# dependencies = ["pyyaml>=6,<7"]
# ///
"""Fetch checksum-pinned pointer assets from a test-assets catalog."""

from __future__ import annotations

import argparse
import hashlib
import re
import shutil
import stat
import sys
import tarfile
import tempfile
import urllib.request
import zipfile
from pathlib import Path, PurePosixPath

import yaml


def fail(message: str) -> ValueError:
    return ValueError(message)


def mapping(value: object, label: str) -> dict:
    if not isinstance(value, dict):
        raise fail(f"{label} must be a mapping")
    return value


def relative_path(value: object, label: str) -> Path:
    if not isinstance(value, str) or not value:
        raise fail(f"{label} must be a non-empty relative path")
    path = PurePosixPath(value)
    if path.is_absolute() or ".." in path.parts:
        raise fail(f"{label} must be a safe relative path")
    return Path(*path.parts)


def load_catalog(path: Path) -> list[dict]:
    catalog = mapping(yaml.safe_load(path.read_text()), "catalog")
    if catalog.get("schema_version") != "1":
        raise fail("unsupported catalog schema_version")
    entries = catalog.get("assets")
    if not isinstance(entries, list) or not entries:
        raise fail("catalog assets must be a non-empty list")

    assets: list[dict] = []
    seen: set[str] = set()
    for entry_value in entries:
        entry = mapping(entry_value, "catalog asset")
        asset_id = str(entry.get("id", ""))
        if not asset_id or asset_id in seen:
            raise fail(f"missing or duplicate asset id: {asset_id!r}")
        seen.add(asset_id)
        manifest_path = path.parent / relative_path(entry.get("manifest"), "manifest")
        manifest = mapping(yaml.safe_load(manifest_path.read_text()), asset_id)
        for field in ("id", "kind", "name", "storage"):
            if manifest.get(field) != entry.get(field):
                raise fail(f"{asset_id}: catalog/manifest {field} mismatch")
        if manifest.get("storage") != "pointer":
            raise fail(f"{asset_id}: resolver accepts pointer assets only")
        if not asset_id.startswith(f"{entry.get('kind')}."):
            raise fail(f"{asset_id}: id prefix must match kind")

        license_data = mapping(manifest.get("license"), f"{asset_id}.license")
        if not license_data.get("id") or not license_data.get("url"):
            raise fail(f"{asset_id}: license id and upstream URL are required")
        license_path = manifest_path.parent / relative_path(
            license_data.get("file"), f"{asset_id}.license.file"
        )
        if not license_path.is_file():
            raise fail(f"{asset_id}: missing license evidence {license_path}")

        verification = mapping(manifest.get("verification"), f"{asset_id}.verification")
        if not verification.get("date") or not verification.get("method"):
            raise fail(f"{asset_id}: dated verification method is required")
        if re.fullmatch(r"\d{4}-\d{2}-\d{2}", str(verification["date"])) is None:
            raise fail(f"{asset_id}: verification date must use YYYY-MM-DD")
        source = mapping(manifest.get("source"), f"{asset_id}.source")
        if not all(source.get(key) for key in ("repository", "revision", "url", "sha256", "archive")):
            raise fail(f"{asset_id}: source provenance is incomplete")
        digest = str(source["sha256"])
        if len(digest) != 64 or any(c not in "0123456789abcdef" for c in digest):
            raise fail(f"{asset_id}: sha256 must be 64 lowercase hex characters")
        entrypoints = mapping(manifest.get("entrypoints"), f"{asset_id}.entrypoints")
        if not entrypoints:
            raise fail(f"{asset_id}: at least one entrypoint is required")
        manifest["_manifest_path"] = manifest_path
        assets.append(manifest)
    return assets


def safe_members(names: list[str]) -> None:
    for name in names:
        path = PurePosixPath(name)
        if path.is_absolute() or ".." in path.parts:
            raise fail(f"unsafe archive member: {name}")


def fetch(asset: dict, destination: Path) -> None:
    asset_id = asset["id"]
    source = asset["source"]
    target = destination / asset_id
    with tempfile.TemporaryDirectory(prefix="robium-asset-") as temp_name:
        temp = Path(temp_name)
        archive_path = temp / "asset.archive"
        request = urllib.request.Request(source["url"], headers={"User-Agent": "robium-assets/1"})
        digest = hashlib.sha256()
        with urllib.request.urlopen(request) as response, archive_path.open("wb") as output:
            while chunk := response.read(1024 * 1024):
                digest.update(chunk)
                output.write(chunk)
        actual = digest.hexdigest()
        if actual != source["sha256"]:
            raise fail(f"{asset_id}: checksum mismatch: expected {source['sha256']}, got {actual}")

        extracted = temp / "extracted"
        extracted.mkdir()
        if source["archive"] == "tar.gz":
            with tarfile.open(archive_path, "r:gz") as archive:
                members = archive.getmembers()
                safe_members([member.name for member in members])
                if any(member.issym() or member.islnk() for member in members):
                    raise fail(f"{asset_id}: archive contains links")
                archive.extractall(extracted, members=members, filter="data")
        elif source["archive"] == "zip":
            with zipfile.ZipFile(archive_path) as archive:
                members = archive.infolist()
                safe_members([member.filename for member in members])
                if any(stat.S_ISLNK(member.external_attr >> 16) for member in members):
                    raise fail(f"{asset_id}: archive contains links")
                archive.extractall(extracted)
        else:
            raise fail(f"{asset_id}: unsupported archive {source['archive']!r}")

        root = extracted
        if source.get("strip_prefix"):
            root = extracted / relative_path(source["strip_prefix"], "strip_prefix")
        if not root.is_dir():
            raise fail(f"{asset_id}: strip_prefix not found")
        for label, value in asset["entrypoints"].items():
            if not (root / relative_path(value, f"entrypoint {label}")).is_file():
                raise fail(f"{asset_id}: missing entrypoint {label}: {value}")
        if target.exists():
            shutil.rmtree(target)
        target.parent.mkdir(parents=True, exist_ok=True)
        shutil.copytree(root, target)
    print(f"{asset_id}: verified {source['sha256']} -> {target}")


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("asset_ids", nargs="*")
    parser.add_argument("--catalog", type=Path, default=Path("test-assets/catalog.yaml"))
    parser.add_argument("--destination", type=Path, default=Path("test-assets/cache"))
    parser.add_argument("--list", action="store_true")
    args = parser.parse_args()
    try:
        assets = load_catalog(args.catalog)
        if args.list:
            print("\n".join(asset["id"] for asset in assets))
            return 0
        selected = set(args.asset_ids)
        unknown = selected - {asset["id"] for asset in assets}
        if unknown:
            raise fail(f"unknown asset ids: {', '.join(sorted(unknown))}")
        for asset in assets:
            if not selected or asset["id"] in selected:
                fetch(asset, args.destination)
        return 0
    except (OSError, ValueError, yaml.YAMLError) as error:
        print(f"error: {error}", file=sys.stderr)
        return 1


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