LIBERO Python 3.11 validation helpers

README.md

LIBERO Python 3.11 validation helpers

Context:

• Upstream PR: Lifelong-Robot-Learning/LIBERO#143
• Minimal PR commit after cleanup: b03bee8

Validation result:

• Clean Python 3.11 environment installed requirements.txt.
• Editable install completed with pip install -e . --no-deps.
• pip check passed.
• Source compilation passed.
• Policy module forward/backward smoke passed.
• Full dataset render gate passed: 130/130 tasks rendered.

Files:

• smoke_check.py: repo-local smoke harness used during validation.
• modal_render_smoke.py: Modal runner that downloads LIBERO datasets to a Volume and runs the full render gate on a T4.

These files were intentionally kept out of the upstream PR to keep the review diff minimal.

modal_render_smoke.py

import os
import subprocess
import time
from pathlib import Path

import modal


APP_NAME = "libero-render-smoke"
VOLUME_NAME = "libero-render-smoke-data"
REPO_DIR = Path("/root/LIBERO")
DATASETS_DIR = Path("/data/datasets")
OUTPUT_ROOT = Path("/data/render_outputs")
HF_REPO_ID = "yifengzhu-hf/LIBERO-datasets"

EXPECTED_COUNTS = {
    "libero_object": 10,
    "libero_goal": 10,
    "libero_spatial": 10,
    "libero_10": 10,
    "libero_90": 90,
}


image = (
    modal.Image.debian_slim(python_version="3.11")
    .apt_install(
        "build-essential",
        "cmake",
        "ffmpeg",
        "git",
        "libegl1",
        "libgl1",
        "libglib2.0-0",
        "libglvnd0",
        "libgomp1",
        "libosmesa6",
        "libsm6",
        "libx11-6",
        "libxext6",
        "libxrender1",
    )
    .pip_install_from_requirements(
        "requirements.txt", env={"CMAKE_POLICY_VERSION_MINIMUM": "3.5"}
    )
    .add_local_dir("benchmark_scripts", f"{REPO_DIR}/benchmark_scripts", copy=True)
    .add_local_dir("libero", f"{REPO_DIR}/libero", copy=True)
    .add_local_dir("scripts", f"{REPO_DIR}/scripts", copy=True)
    .add_local_dir("templates", f"{REPO_DIR}/templates", copy=True)
    .add_local_file("README.md", f"{REPO_DIR}/README.md", copy=True)
    .add_local_file("pyproject.toml", f"{REPO_DIR}/pyproject.toml", copy=True)
    .add_local_file("requirements.txt", f"{REPO_DIR}/requirements.txt", copy=True)
    .add_local_file("setup.py", f"{REPO_DIR}/setup.py", copy=True)
    .run_commands(f"cd {REPO_DIR} && pip install -e . --no-deps")
    .env(
        {
            "HF_HOME": "/data/hf-cache",
            "MUJOCO_GL": "egl",
            "PYOPENGL_PLATFORM": "egl",
        }
    )
)

app = modal.App(APP_NAME, image=image)
volume = modal.Volume.from_name(VOLUME_NAME, create_if_missing=True)


def _dataset_complete() -> bool:
    for suite, expected in EXPECTED_COUNTS.items():
        suite_dir = DATASETS_DIR / suite
        if not suite_dir.exists():
            return False
        if len(list(suite_dir.glob("*.hdf5"))) != expected:
            return False
    return True


def _download_datasets(force_download: bool) -> None:
    if _dataset_complete() and not force_download:
        print("[modal] datasets already present in volume")
        return

    from huggingface_hub import snapshot_download

    DATASETS_DIR.mkdir(parents=True, exist_ok=True)
    print(f"[modal] downloading {HF_REPO_ID} into {DATASETS_DIR}")
    snapshot_download(
        repo_id=HF_REPO_ID,
        repo_type="dataset",
        local_dir=str(DATASETS_DIR),
        force_download=force_download,
    )
    if not _dataset_complete():
        counts = {
            suite: len(list((DATASETS_DIR / suite).glob("*.hdf5")))
            for suite in EXPECTED_COUNTS
        }
        raise RuntimeError(f"download incomplete, hdf5 counts: {counts}")
    print("[modal] dataset download/check passed")


@app.function(
    cpu=4,
    memory=32768,
    volumes={"/data": volume},
    timeout=60 * 60 * 8,
)
def prepare_datasets(force_download: bool = False) -> str:
    started = time.time()
    _download_datasets(force_download=force_download)
    volume.commit()
    elapsed = time.time() - started
    return f"datasets ready at {DATASETS_DIR}, elapsed_seconds={elapsed:.1f}"


@app.function(
    gpu="T4",
    cpu=4,
    memory=32768,
    volumes={"/data": volume},
    timeout=60 * 60 * 8,
)
def run_render_smoke(
    render: str = "all",
    render_limit: int = 5,
    force_download: bool = False,
) -> str:
    started = time.time()
    os.chdir(REPO_DIR)
    if force_download:
        raise ValueError("force_download is only supported by prepare_datasets")
    if not _dataset_complete():
        raise RuntimeError(
            f"datasets are missing or incomplete at {DATASETS_DIR}; "
            "run prepare_datasets first"
        )

    run_id = time.strftime("%Y%m%d-%H%M%S")
    output_dir = OUTPUT_ROOT / run_id
    command = [
        "python",
        "scripts/smoke_check.py",
        "--datasets-dir",
        str(DATASETS_DIR),
        "--require-datasets",
        "--render",
        render,
        "--render-limit",
        str(render_limit),
        "--render-output",
        str(output_dir),
    ]
    print("[modal] running:", " ".join(command))
    subprocess.run(command, cwd=REPO_DIR, check=True)
    volume.commit()

    elapsed = time.time() - started
    png_count = len(list(output_dir.glob("*.png")))
    summary = (
        f"render={render}, output_dir={output_dir}, png_count={png_count}, "
        f"elapsed_seconds={elapsed:.1f}"
    )
    print("[modal]", summary)
    return summary


@app.local_entrypoint()
def main(
    render: str = "all",
    render_limit: int = 5,
    force_download: bool = False,
):
    print(prepare_datasets.remote(force_download=force_download))
    print(
        run_render_smoke.remote(
            render=render,
            render_limit=render_limit,
            force_download=False,
        )
    )

smoke_check.py

#!/usr/bin/env python3
"""Smoke checks for Python/runtime upgrades.

The default run verifies source compilation, benchmark manifests, BDDL/init-state
paths, and init-state loading. Dataset and render checks are opt-in because they
require the downloaded demonstration assets and a working MuJoCo renderer.
"""

import argparse
import compileall
import os
import subprocess
import sys
import tempfile
from pathlib import Path


EXPECTED_SUITES = {
    "libero_object": 10,
    "libero_goal": 10,
    "libero_spatial": 10,
    "libero_10": 10,
    "libero_90": 90,
}


def repo_root() -> Path:
    return Path(__file__).resolve().parents[1]


def compile_sources(root: Path) -> None:
    targets = [
        root / "setup.py",
        root / "benchmark_scripts",
        root / "libero",
        root / "scripts",
        root / "templates",
    ]
    ok = True
    for target in targets:
        if target.is_dir():
            ok = compileall.compile_dir(str(target), quiet=1) and ok
        else:
            ok = compileall.compile_file(str(target), quiet=1) and ok
    if not ok:
        raise RuntimeError("source compilation failed")
    print("[smoke] source compilation passed")


def write_libero_config(root: Path, config_dir: Path, datasets_dir: Path | None) -> None:
    benchmark_root = root / "libero" / "libero"
    config_dir.mkdir(parents=True, exist_ok=True)
    if datasets_dir is None:
        datasets_dir = config_dir / "datasets"
        datasets_dir.mkdir(parents=True, exist_ok=True)
    config = {
        "benchmark_root": benchmark_root,
        "bddl_files": benchmark_root / "bddl_files",
        "init_states": benchmark_root / "init_files",
        "datasets": datasets_dir,
        "assets": benchmark_root / "assets",
    }
    content = "".join(f"{key}: {value}\n" for key, value in config.items())
    (config_dir / "config.yaml").write_text(content, encoding="utf-8")
    os.environ["LIBERO_CONFIG_PATH"] = str(config_dir)
    print(f"[smoke] using LIBERO_CONFIG_PATH={config_dir}")


def validate_benchmarks(require_datasets: bool, load_init_states: bool):
    from libero.libero import benchmark, get_libero_path

    bddl_root = Path(get_libero_path("bddl_files"))
    init_root = Path(get_libero_path("init_states"))
    datasets_root = Path(get_libero_path("datasets"))
    benchmark_dict = benchmark.get_benchmark_dict()

    render_tasks = []
    missing_demos = []
    for suite_name, expected_count in EXPECTED_SUITES.items():
        bench = benchmark_dict[suite_name]()
        actual_count = bench.get_num_tasks()
        if actual_count != expected_count:
            raise AssertionError(
                f"{suite_name} expected {expected_count} tasks, got {actual_count}"
            )

        for task_id in range(actual_count):
            task = bench.get_task(task_id)
            bddl_file = bddl_root / task.problem_folder / task.bddl_file
            init_file = init_root / task.problem_folder / task.init_states_file
            demo_file = datasets_root / bench.get_task_demonstration(task_id)

            if not bddl_file.exists():
                raise FileNotFoundError(bddl_file)
            if not init_file.exists():
                raise FileNotFoundError(init_file)
            if load_init_states:
                init_states = bench.get_task_init_states(task_id)
                if len(init_states) == 0:
                    raise AssertionError(f"{init_file} has no init states")

            if demo_file.exists():
                render_tasks.append((suite_name, task_id, bddl_file, demo_file))
            else:
                missing_demos.append(demo_file)

    if missing_demos and require_datasets:
        raise FileNotFoundError(
            "missing demonstration datasets:\n"
            + "\n".join(str(path) for path in missing_demos[:20])
        )
    if missing_demos:
        print(
            f"[smoke] benchmark metadata passed; skipped {len(missing_demos)} "
            "missing demo files"
        )
    else:
        print("[smoke] benchmark metadata, init states, and demo paths passed")

    return render_tasks


def run_policy_smoke() -> None:
    import torch

    from libero.lifelong.models.modules.rgb_modules import ResnetEncoder, SpatialSoftmax

    torch.manual_seed(0)
    spatial = SpatialSoftmax(in_c=4, in_h=8, in_w=8, num_kp=2)
    spatial_out = spatial(torch.randn(2, 4, 8, 8))
    if spatial_out.shape != (2, 4) or not torch.isfinite(spatial_out).all():
        raise AssertionError(f"unexpected SpatialSoftmax output: {spatial_out.shape}")

    encoder = ResnetEncoder(
        input_shape=(3, 64, 64),
        output_size=16,
        pretrained=False,
        remove_layer_num=4,
        language_fusion="none",
    )
    encoder_out = encoder(torch.randn(2, 3, 64, 64))
    if encoder_out.shape != (2, 16) or not torch.isfinite(encoder_out).all():
        raise AssertionError(f"unexpected ResnetEncoder output: {encoder_out.shape}")
    encoder_out.square().mean().backward()
    print("[smoke] policy module forward/backward passed")


def render_tasks(root: Path, tasks, mode: str, limit: int, output_dir: Path) -> None:
    if mode == "none":
        return
    selected = tasks if mode == "all" else tasks[:limit]
    if not selected:
        raise RuntimeError("no renderable tasks were found")

    output_dir.mkdir(parents=True, exist_ok=True)
    for suite_name, task_id, bddl_file, demo_file in selected:
        command = [
            sys.executable,
            str(root / "benchmark_scripts" / "render_single_task.py"),
            "--benchmark_name",
            suite_name,
            "--task_id",
            str(task_id),
            "--bddl_file",
            str(bddl_file),
            "--demo_file",
            str(demo_file),
            "--output_dir",
            str(output_dir),
        ]
        subprocess.run(command, cwd=root, check=True)

    rendered = list(output_dir.glob("*.png"))
    if len(rendered) < len(selected):
        raise AssertionError(f"expected {len(selected)} renders, found {len(rendered)}")
    print(f"[smoke] rendered {len(selected)} task(s) to {output_dir}")


def parse_args():
    parser = argparse.ArgumentParser()
    parser.add_argument("--skip-compile", action="store_true")
    parser.add_argument("--skip-benchmarks", action="store_true")
    parser.add_argument("--no-init-state-load", action="store_true")
    parser.add_argument("--require-datasets", action="store_true")
    parser.add_argument("--policy-smoke", action="store_true")
    parser.add_argument("--use-existing-config", action="store_true")
    parser.add_argument("--config-dir", type=Path)
    parser.add_argument("--datasets-dir", type=Path)
    parser.add_argument("--render", choices=["none", "sample", "all"], default="none")
    parser.add_argument("--render-limit", type=int, default=5)
    parser.add_argument("--render-output", type=Path, default=Path("benchmark_tasks"))
    return parser.parse_args()


def main() -> None:
    args = parse_args()
    root = repo_root()
    sys.path.insert(0, str(root))
    default_datasets_dir = root / "libero" / "datasets"
    datasets_dir = args.datasets_dir
    if datasets_dir is None and default_datasets_dir.exists():
        datasets_dir = default_datasets_dir
    temp_config = None
    renderable_tasks = []

    if not args.skip_compile:
        compile_sources(root)

    needs_config = not args.skip_benchmarks or args.render != "none"
    if needs_config and not args.use_existing_config:
        if args.config_dir is None:
            temp_config = tempfile.TemporaryDirectory(prefix="libero-smoke-")
            config_dir = Path(temp_config.name)
        else:
            config_dir = args.config_dir
        write_libero_config(root, config_dir, datasets_dir)

    try:
        if not args.skip_benchmarks:
            renderable_tasks = validate_benchmarks(
                require_datasets=args.require_datasets,
                load_init_states=not args.no_init_state_load,
            )
        if args.policy_smoke:
            run_policy_smoke()
        if args.render != "none":
            if not renderable_tasks:
                renderable_tasks = validate_benchmarks(
                    require_datasets=True,
                    load_init_states=not args.no_init_state_load,
                )
            output_dir = args.render_output
            if not output_dir.is_absolute():
                output_dir = root / output_dir
            render_tasks(root, renderable_tasks, args.render, args.render_limit, output_dir)
    finally:
        if temp_config is not None:
            temp_config.cleanup()

    print("[smoke] all requested checks passed")


if __name__ == "__main__":
    main()