Add. 具身仿真合成Baseline

master embodied-simulation-baseline
Xuewei Guo 1 week ago
parent d8147a32ed
commit b34143799f

@ -0,0 +1,9 @@
# Embodied AI
本目录收录具身智能的学习材料、仿真资产生成与竞赛实践。内容以可复现、可核验和不分发官方私有数据为原则组织。
## 导航
- [竞赛实践](竞赛实践/README.md):可提交资产生成 Baseline 及其复现说明。
后续主题将按仿真基础、三维重建与资产生成、机器人学习与 VLA 逐步补充。

@ -0,0 +1,23 @@
# 校验值
## 待发布 Release Asset
```text
文件: public_asset_baseline_release.zip
SHA256: 83c7860b1545900dee8f72988b958fc4b930702d45150da42f8f5b859fd95527
大小: 78,269,181 bytes
```
该 ZIP 解压后包含 `public_asset_baseline/` 目录、路径 A 的冻结中间包,以及预生成的 `quick_output/`
## 历史候选与路径 A 输出
```text
文件: quick_output/submission.zip
MD5: a220303381c7bb8886684778e34689df
SHA256: 2e979d08d10a785e0c47a4a1ba923131a52c8e53970bb0578f815893585e8d4b
大小: 19,539,975 bytes
ZIP 成员: 56
```
`artifacts/reference/submission.zip` 与路径 A 重新生成的 `submission.zip` 必须和上述候选逐字节一致。

@ -0,0 +1,97 @@
# 2026 具身仿真合成挑战赛:可复现 Baseline
本项目提供两条清晰区分的路径:
1. **路径 A快速精确复现。** 使用冻结中间包,在无 GPU 环境下重建历史 `67.6000` 候选的同一字节,并做本地校验。
2. **路径 B从官方输入重建。** 以 `question.zip``submission_example.zip` 为竞赛数据输入,在阿里云 GPU 上执行可审计的资产生成流程。
源码位于 [public_asset_baseline](public_asset_baseline/README.md)。官方原始数据、视频、模型缓存和任何密钥均不随本项目分发。
## 下载与解压
发布后,从 GitHub Release 下载 `public_asset_baseline_release.zip`。下载链接固定为:
```text
https://github.com/ben1234560/AiLearning-Theory-Applying/releases/latest/download/public_asset_baseline_release.zip
```
解压并进入 Baseline
```bash
unzip public_asset_baseline_release.zip
cd public_asset_baseline
```
本地待上传的同名文件位于 `release_assets/`;该目录中的 ZIP 被 Git 忽略,待官方核验通过后由仓库维护者在 GitHub Release 页面上传。上传前后均应对照 [CHECKSUMS.md](CHECKSUMS.md)。
## 校验
Release ZIP 的 SHA-256 和历史候选的 MD5/SHA-256 见 [CHECKSUMS.md](CHECKSUMS.md)。解压后,先校验预生成的提交包:
```bash
md5 quick_output/submission.zip
shasum -a 256 quick_output/submission.zip
```
预期值:
```text
MD5: a220303381c7bb8886684778e34689df
SHA256: 2e979d08d10a785e0c47a4a1ba923131a52c8e53970bb0578f815893585e8d4b
```
再执行结构、纹理引用和轻量物理校验:
```bash
python scripts/validate_submission.py \
--package quick_output/submission.zip \
--report quick_output/submission_recheck.json
```
合格结果应为 `valid: true`、`valid_tasks: 34`、`physics_probe_steps: 120`。
## 路径 A无需 GPU 的快速精确复现
Release 包已附带 `quick_output/submission.zip`,可直接比对或提交。若要从冻结中间包重建它:
```bash
python3 -m venv .venv
source .venv/bin/activate
python -m pip install --upgrade pip
python -m pip install .
python -m pip install -r requirements-dev.txt
python scripts/quick_reproduce.py --output-root reproduced_quick_output
python scripts/validate_submission.py \
--package reproduced_quick_output/submission.zip \
--report reproduced_quick_output/validation_recheck.json
```
`reproduced_quick_output/submission.zip` 的 MD5 和 SHA-256 应与上方预期值一致。路径 A 不访问网络、不使用 GPU它保证候选文件相同而未来线上分数仍由平台评测环境决定。
## 路径 B从官方 ZIP 的 GPU 重建
将官方获取的两份 ZIP 放在私有目录,且不要提交到 Git
```text
/data/competition_input/question.zip
/data/competition_input/submission_example.zip
```
完整资源选择、许可前提、GPU 宿主机预检与运行方式见 [GPU_BUILD.md](public_asset_baseline/GPU_BUILD.md)。完成许可确认后,在 `public_asset_baseline/` 目录执行:
```bash
export DATA_DIR=/data/competition_input
export OUT_DIR=/data/asset_baseline_output
export ACCEPT_HUNYUAN_LICENSE=yes
bash scripts/run_aliyun.sh
```
路径 B 的输出是独立重建结果,不承诺与路径 A 字节一致或获得相同线上分数。
## 发布前检查
1. 确认竞赛规则允许公开冻结候选和中间产物。
2. 确认第三方模型、代码与素材许可证允许当前发布方式。
3. 在 GitHub Desktop 中只提交未忽略的源码、文档和校验文件;不要提交 `release_assets/*.zip`、官方 ZIP、模型缓存或 GPU 输出。
4. 官方核验通过后,先推送源码,再在 GitHub Release 页面上传本地 `release_assets/public_asset_baseline_release.zip`

@ -0,0 +1,5 @@
__pycache__/
.pytest_cache/
third_party/
outputs/
quick_output/

@ -0,0 +1,25 @@
# The released container is Linux-first because Hunyuan's CUDA rasterizer is not
# portable to macOS/Windows. The core primitive/validation path remains portable.
FROM nvidia/cuda:12.4.1-cudnn-devel-ubuntu22.04
ENV DEBIAN_FRONTEND=noninteractive \
PIP_NO_CACHE_DIR=1 \
PYTHONUNBUFFERED=1 \
MUJOCO_GL=egl
RUN apt-get update && apt-get install -y --no-install-recommends \
git ffmpeg libegl1 libgl1 libglib2.0-0 python3.10 python3.10-venv python3-pip \
&& rm -rf /var/lib/apt/lists/*
WORKDIR /opt/public_asset_baseline
COPY pyproject.toml requirements-gpu.txt ./
COPY src ./src
COPY config ./config
COPY scripts ./scripts
RUN python3.10 -m pip install --upgrade pip \
&& python3.10 -m pip install --index-url https://download.pytorch.org/whl/cu124 torch==2.5.1 torchvision==0.20.1 torchaudio==2.5.1 \
&& python3.10 -m pip install -r requirements-gpu.txt \
&& python3.10 -m pip install .
ENTRYPOINT ["asset-baseline"]
CMD ["--help"]

@ -0,0 +1,78 @@
# 阿里云 GPU 构建指南
本文只描述从 `question.zip``submission_example.zip` 重新生成资产的 GPU 路径。快速精确复现不需要 GPU见 [README.md](README.md)。
## 资源选择与结论
优先选择运行 Linux 的阿里云 GPU ECS
| 用途 | 实例 | GPU | CPU / 内存 | 系统盘 |
|---|---|---|---|---|
| 最小兼容宿主机 | `ecs.gn8is.2xlarge` | 1 × NVIDIA L2048 GiB | 8 vCPU / 64 GiB | 500 GiB ESSD |
| 完整生成推荐 | `ecs.gn8is.4xlarge` | 1 × NVIDIA L2048 GiB | 16 vCPU / 128 GiB | 500 GiB ESSD |
两档均使用 Ubuntu 22.04 x86_64、Docker 和 NVIDIA Container Toolkit并允许首次安装时临时访问公开代码、模型和 Python 依赖源。`ecs.gn8is.2xlarge` 足以承载单卡 L20 的完整软件栈;`ecs.gn8is.4xlarge` 为视频解码、模型编译和资产校验保留更多 CPU/内存余量,作为公开 Baseline 的默认推荐。
阿里云当前规格表将 `ecs.gn8is.2xlarge` 列为单张 L20、48 GiB 显存、8 vCPU 和 64 GiB 内存,并将 `ecs.gn8is.4xlarge` 列为同一张 L20、16 vCPU 和 128 GiB 内存。[阿里云 GPU ECS 规格说明](https://help.aliyun.com/zh/ecs/user-guide/gpu-accelerated-compute-optimized-and-vgpu-accelerated-instance-families-1) 控制台可用地域和库存可能变化,应以实际购买页为准。
**边界说明:**上述规格与容器要求兼容,且仓库在启动前会检查 GPU、显存、磁盘、Docker 和容器内 GPU 可见性;这能证明宿主环境满足基线的硬件前提。它不等同于“当前第三方模型版本已经在每个地域完整推理成功”——模型下载、许可、驱动和上游依赖仍需由一次实际运行验证。不要在未通过预检时静默换成 24 GiB 显存卡运行完整 Shape + Paint 路径。
## 容器与网络要求
| 项目 | 基线值 |
|---|---|
| 操作系统 | Ubuntu 22.04 x86_64 |
| 网络 | 允许临时访问公开代码、模型和 Python 依赖源 |
| 容器 | Docker + NVIDIA Container Toolkit |
## 安全与宿主机验收
1. 只开放 SSH 所需的最小安全组规则,并把来源限制为操作者公网 IP。
2. 不把私钥、账号 Cookie、临时下载令牌或官方原始数据提交到仓库。
3. 使用带 GPU 驱动的 Ubuntu 22.04 镜像,创建后先执行:
```bash
git clone <YOUR_PRIVATE_OR_PUBLIC_REPOSITORY_URL> public_asset_baseline
cd public_asset_baseline
bash infra/verify_gpu_host.sh
```
脚本必须打印 GPU 名称、显存、驱动版本、Docker 版本和可用磁盘空间,并成功在容器中运行 `nvidia-smi`。它要求至少 45,000 MiB 显存和 200 GiB 可用磁盘;失败时先修复驱动、磁盘或 NVIDIA Container Toolkit不要启动模型安装或推理。`scripts/run_aliyun.sh` 也会自动运行该预检,避免漏做。
阿里云驱动版本会随镜像或资源选择变化。CUDA 容器使用 `12.4.1`;需保证宿主 NVIDIA 驱动与此容器兼容。NVIDIA Container Toolkit 的安装说明以其[官方文档](https://docs.nvidia.com/datacenter/cloud-native/container-toolkit/latest/install-guide.html)为准。
## 输入准备
将从赛题页面获得的两份官方 ZIP 放在私有目录:
```text
/data/competition_input/question.zip
/data/competition_input/submission_example.zip
```
不要把这两个文件、解压视频、帧、掩码或模型缓存纳入公开分享包。首次运行前确认 Hunyuan3D-2.1 许可证允许当前用途;该项目不会替你作许可判断。
## 运行
```bash
export DATA_DIR=/data/competition_input
export OUT_DIR=/data/asset_baseline_output
export ACCEPT_HUNYUAN_LICENSE=yes
bash scripts/run_aliyun.sh
```
该入口会先执行宿主机预检,再构建容器,固定公开依赖的源码提交,下载所需模型,并将模型缓存挂载在仓库的 `third_party/`。后续重跑复用该缓存,但每个 `OUT_DIR` 应是新的空目录,避免混淆过程收据。
## 完成门槛
运行结束后,至少检查:
```bash
cat "$OUT_DIR/gpu_asset_run/reports/validation.json"
cat "$OUT_DIR/gpu_asset_run/process/metrics.json"
```
前者应显示 `valid: true`、`valid_tasks: 34` 和 `simulation_steps: 120`。后者应保存 ZIP SHA-256。然后单独保留整个 `gpu_asset_run/`,而不是只保留最终 ZIP。
GPU 推理本身不在轻量测试范围内;完整运行前需通过宿主机验收,运行后需通过上述 34 题验证报告。生成候选是否接近快速路径的线上表现,必须由独立线上评测确认。

@ -0,0 +1,5 @@
# 第三方组件与数据边界
- 本仓库不分发官方 `question.zip`、`submission_example.zip`、视频帧、账号凭据或模型权重。
- GPU 路径会获取 SAM2 与 Hunyuan3D-2.1 的公开源码和模型;它们各自适用原始许可证,运行者需自行确认许可与赛题规则兼容。
- 中间提交包和最终提交包可能属于竞赛生成资产。公开分发前请确认赛题规则允许。

@ -0,0 +1,157 @@
# 具身仿真资产生成 Baseline分数67+
提供两条互不混淆的复现路径:
1. **快速精确复现**:使用仓库内两份冻结中间包,在无 GPU 环境下生成可提交 ZIP并做 ZIP、OpenUSD 与轻量物理校验。
2. **从官方输入重建**:仅以 `question.zip``submission_example.zip` 为竞赛数据输入,在阿里云 GPU 上从视频生成 USD 资产、纹理、碰撞与过程收据。
两条路径有不同目标。快速路径重建的是已验证候选的同一字节;从官方输入重建路径是可审计的生成研究基线,输出合法候选但不承诺与快速路径完全一致,可能由于模型、驱动和数值精度导致差异。
## 解题策略
赛题目标是把视频中的物体转换为可提交、可物理仿真的 USD 资产。生成路径按以下顺序工作:
```text
官方 ZIP
-> 输入哈希、CRC 与 34 题/视频清单
-> 抽帧与多样性/清晰度选视角
-> 前景分割
-> Shape + Paint 重建
-> 视觉网格、纹理、碰撞体、刚体 USD
-> OpenUSD 与 120 步 MuJoCo 校验
-> submission.zip
```
它把竞赛原始数据保持在仓库外;每次运行写出输入哈希、视角选择、网格/资产清单、运行环境、命令记录和验证报告。快速路径则把已存在的两个中间 ZIP 作为明确输入:保留基础中间包的全部成员,仅以供体中两个经过哈希锁定的 USD 成员替换同名成员,再使用固定 ZIP 元数据写出最终包。
## 路径 A无需 GPU 的快速精确复现
仓库已包含以下非原始数据的中间产物:
```text
artifacts/intermediate/base_submission.zip
artifacts/intermediate/donor_submission.zip
artifacts/reference/submission.zip
evidence/online_score_evidence.json
```
其中 `evidence/online_score_evidence.json` 是外部线上评测结果的事实记录,不是、也不可能是代码生成的实验过程。它记录的候选 SHA-256 与 `artifacts/reference/submission.zip` 相同,结果为 `67.6000`。其余过程文件都由本仓库脚本生成并可本地复核。
安装 Python 3.10+ 后执行:
```bash
python3 -m venv .venv
source .venv/bin/activate # Windows PowerShell: .venv\Scripts\Activate.ps1
python -m pip install --upgrade pip
python -m pip install .
python -m pip install -r requirements-dev.txt
python scripts/quick_reproduce.py --output-root quick_output
```
这里有意使用普通安装而不是可编辑安装,因此仓库即使位于包含中文字符的目录中也可正常安装。
输出文件:
```text
quick_output/submission.zip
quick_output/rebuild_report.json
quick_output/validation_report.json
```
提交前必须检查:
```bash
python scripts/validate_submission.py \
--package quick_output/submission.zip \
--report quick_output/submission_recheck.json
```
合格条件为:
```text
SHA-256: 2e979d08d10a785e0c47a4a1ba923131a52c8e53970bb0578f815893585e8d4b
ZIP 成员: 56
顶层 USD: 34
OpenUSD/轻量物理校验: 34/34
```
快速路径不使用 GPU也不访问网络。它的结果与已经完成线上评测的候选**字节完全相同**:本地可保证 SHA-256 一致,也就是提交的候选文件一致。在题目数据、评分规则和评测器版本不变的前提下,它对应的历史线上结果为 `67.6000`。任何未来线上分数仍由平台评测决定;本仓库不会把历史分数误写为对未来评测环境的保证。
## Baseline 亮点与得分相关设计
1. **不允许静默降级**GPU 路径将 Shape 与 Paint 视为完整资产生成的必要步骤;纹理阶段初始化或推理失败会明确失败并保留收据,不会悄悄生成只有几何、没有纹理的候选冒充完整结果。这避免了看似成功、实际质量显著下降的提交。
2. **提交格式与物理可运行性双重校验**:校验器从提交包实际读取 USD检查 34 个顶层资产、默认图元、纹理引用、碰撞体/刚体和物理场景,并执行 120 步轻量自由落体探测。它不能替代官方隐藏评测,但能在提交前阻断常见的格式和物理错误。
## 官方资料在代码中的使用范围
竞赛数据输入只有官方发放的 `question.zip``submission_example.zip`。代码从 `submission_example.zip` 的实际目录结构推导提交根目录,并对 `question.zip` 执行 CRC、安全路径和 34 题/视频清单检查;这些约束来自官方输入文件本身。
赛题页面、讲解回放、PPT 和论文链接只作为人工阅读资料收录在 [REFERENCES.md](REFERENCES.md)。代码不会下载、解析或把这些网页、视频、PPT 当作模型输入或检索知识库;也不会把官方原始数据、视频帧或模型缓存放入公开包。
## 路径 B`question.zip``submission_example.zip` 重建
两份官方 ZIP 不随仓库分发。将其放入同一个私有目录,例如 `/data/competition_input`
```text
/data/competition_input/question.zip
/data/competition_input/submission_example.zip
```
完整 GPU 环境和资源选择见 [GPU_BUILD.md](GPU_BUILD.md)。在已通过宿主机检查后执行:
```bash
export DATA_DIR=/data/competition_input
export OUT_DIR=/data/asset_baseline_output
export ACCEPT_HUNYUAN_LICENSE=yes
bash scripts/run_aliyun.sh
```
运行者必须先阅读并确认 SAM2 与 Hunyuan3D-2.1 的许可证允许自己的用途;脚本不会绕过这一确认。成功输出位于:
```text
$OUT_DIR/asset_baseline_submission.zip
$OUT_DIR/gpu_asset_run/manifests/
$OUT_DIR/gpu_asset_run/reports/validation.json
$OUT_DIR/gpu_asset_run/process/
```
`process/` 中的 `build_manifest.json`、`runtime.json`、`commands.jsonl`、`metrics.json` 和 `experiment_log.json` 都由运行入口生成。`manifests/` 记录输入、选帧、网格和打包过程;`reports/validation.json` 记录 34 个资产的 OpenUSD 与 MuJoCo 校验。
### 无 GPU 的代码冒烟测试
任何支持 Python 的系统都可验证输入、格式、USD 与物理链路:
```bash
asset-baseline --config config/default.yaml \
--set question_zip=/absolute/path/question.zip \
--set submission_example_zip=/absolute/path/submission_example.zip \
--set output_root=outputs \
--set run_name=cpu_smoke \
--set reconstruction.backend=primitive \
--set segmentation.backend=center run
```
这是一条格式和物理链路测试不是高分承诺。macOS 与 Windows 可运行快速路径和 CPU 冒烟路径;正式 GPU 推理请在阿里云 Linux 实例中运行。Windows 可用 WSL 作为命令行替代。
## 本地测试
除 GPU 推理外,仓库中的验证均可快速执行:
```bash
pytest
```
测试覆盖:两份中间包的锁定哈希和 CRC、确定性重建、最终包的字节身份、34 个 USD 的结构与物理探测,以及从合成官方格式输入到 34 个 USD 的 CPU 冒烟流程。
## 下一步
1. 将 6 个选视角真正用于多视角一致性和纹理投影,不能只把一个视图作为条件图。
2. 使用交互视频分离可动部件,为门、抽屉、按钮和转轴生成独立碰撞体与关节。
3. 为尺度、质量、摩擦和关节限位建立任务级估计与回归测试,不把全部物体归一化为相同尺寸和质量。
4. 每次改变只生成一个独立候选,先保存本地验证和候选 SHA-256再进行一次线上提交。
## 资料与许可证
赛题入口和课程资料链接收录在 [REFERENCES.md](REFERENCES.md)。第三方组件和模型的许可证说明在 [NOTICE.md](NOTICE.md)。

@ -0,0 +1,28 @@
# 官方资料索引
以下链接由赛题组织方或用户提供。它们用于理解赛题、仿真资产和相关课程;链接内容及可用性以发布方为准。
## 赛题与资产
- [赛题信息页](https://tianchi.aliyun.com/competition/entrance/532506/information)
- [PhysX-Omni从图片到可交互三维资产文档](https://ycntvxxezabq.feishu.cn/docx/O9JqdgMCmo9sTbxm1fJcz2Dun2c?from=from_copylink)
- [赛题讲解会录制](https://meeting.tencent.com/crm/lvqn1EoA32)
- [赛题讲解会视频回放](https://weixin.qq.com/sph/AWXXQpgmBQ)
- [赛题讲解会幻灯片](https://ycntvxxezabq.feishu.cn/slides/BOi6sDBNdlZOXRdLVRLcZeOKnAd?from=from_copylink)
- [Baseline 讲解与代码开源回放](https://meeting.tencent.com/crm/2kAgW00G15)
## 仿真与模型课程
- [Isaac Sim 平台教学录制](https://meeting.tencent.com/crm/2439LPB44a)
- [Isaac Sim 基础操作文档](https://ycntvxxezabq.feishu.cn/docx/EpIJdJRpdoMYgMxNlAPc4GdGnWg?from=from_copylink)
- [URDF Studio](https://botworld.enkeebot.com/)
- [MuJoCo 教学录制](https://meeting.tencent.com/crm/lR0GyqoE8b)
- [MuJoCo Simulate 基础操作文档](https://ycntvxxezabq.feishu.cn/docx/A2rwdMtWYoOjeoxQNSmcl4VAn0g?from=from_copylink)
## 具身与安全方向材料
- [Π0.5 论文](https://arxiv.org/abs/2504.16054) 与 [Π 系列代码](https://github.com/Physical-Intelligence/openpi)
- [跨本体方向讲解录制](https://meeting.tencent.com/crm/2qydYQrje3)
- [LingBot-VLA 2.0](https://arxiv.org/abs/2607.06403)、[τ0-VLA](https://arxiv.org/abs/2608.16885)、[LAP](https://arxiv.org/abs/2603.09743)
- [安全方向讲解录制](https://meeting.tencent.com/crm/KPm3o3pr70)
- [RobustVLA](https://arxiv.org/abs/2510.00037) 与 [MoManipVLA](https://arxiv.org/abs/2503.13446)

@ -0,0 +1,3 @@
"""Raw-video to simulation-ready USD generation from the two official ZIPs."""
__version__ = "0.1.0"

@ -0,0 +1,40 @@
from __future__ import annotations
import argparse
import json
from pathlib import Path
from .config import load_config, resolve_paths
from .pipeline import generate, inspect, package, prepare, run_all, validate
def _config(args: argparse.Namespace) -> dict:
return resolve_paths(load_config(Path(args.config), args.set or []), Path(args.config))
def main() -> None:
parser = argparse.ArgumentParser(description="Zero-history raw-video embodied-asset baseline")
parser.add_argument("--config", default="config/default.yaml", help="YAML configuration")
parser.add_argument("--set", action="append", default=[], help="strict dotted configuration override, e.g. reconstruction.backend=primitive")
subparsers = parser.add_subparsers(dest="command", required=True)
for command in ("inspect", "prepare", "generate", "package", "validate", "run"):
subparsers.add_parser(command)
args = parser.parse_args()
config = _config(args)
if args.command == "inspect":
result = inspect(config)
elif args.command == "prepare":
result = {"run_root": str(prepare(config))}
elif args.command == "generate":
result = {"run_root": str(generate(config))}
elif args.command == "package":
result = {"package": str(package(config))}
elif args.command == "validate":
result = validate(config)
else:
result = run_all(config)
print(json.dumps(result, ensure_ascii=False, indent=2))
if __name__ == "__main__":
main()

@ -0,0 +1,40 @@
from __future__ import annotations
from copy import deepcopy
from pathlib import Path
from typing import Any
import yaml
def load_config(path: Path, overrides: list[str]) -> dict[str, Any]:
"""Load YAML and apply strict dotted `key=value` CLI overrides."""
raw = yaml.safe_load(path.read_text(encoding="utf-8"))
if not isinstance(raw, dict):
raise ValueError(f"configuration must be a mapping: {path}")
config = deepcopy(raw)
for assignment in overrides:
if "=" not in assignment:
raise ValueError(f"override must be key=value: {assignment}")
dotted, value = assignment.split("=", 1)
keys = dotted.split(".")
target: dict[str, Any] = config
for key in keys[:-1]:
child = target.get(key)
if not isinstance(child, dict):
raise KeyError(f"unknown configuration key: {dotted}")
target = child
if keys[-1] not in target:
raise KeyError(f"unknown configuration key: {dotted}")
target[keys[-1]] = yaml.safe_load(value)
return config
def resolve_paths(config: dict[str, Any], config_path: Path) -> dict[str, Any]:
"""Resolve input/output paths relative to the working directory, not this repo."""
result = deepcopy(config)
base = Path.cwd()
for key in ("question_zip", "submission_example_zip", "output_root"):
value = Path(str(result[key]))
result[key] = str((base / value).resolve() if not value.is_absolute() else value.resolve())
return result

@ -0,0 +1,96 @@
from __future__ import annotations
import re
import zipfile
from pathlib import Path, PurePosixPath
from typing import Any
from .util import sha256, utc_now, write_json
TASK_PATTERN = re.compile(r"^item_(\d{3})$")
VIDEO_SUFFIXES = {".mp4", ".mov", ".m4v", ".avi", ".MP4", ".MOV", ".M4V", ".AVI"}
def _safe_members(archive: zipfile.ZipFile) -> list[zipfile.ZipInfo]:
members: list[zipfile.ZipInfo] = []
for info in archive.infolist():
name = PurePosixPath(info.filename)
if name.is_absolute() or ".." in name.parts:
raise ValueError(f"unsafe ZIP member: {info.filename}")
if not info.is_dir() and not info.filename.startswith("__MACOSX/"):
members.append(info)
if archive.testzip() is not None:
raise ValueError(f"ZIP CRC failure: {archive.filename}")
return members
def inspect_inputs(question_zip: Path, example_zip: Path) -> dict[str, Any]:
if not question_zip.is_file():
raise FileNotFoundError(question_zip)
if not example_zip.is_file():
raise FileNotFoundError(example_zip)
with zipfile.ZipFile(question_zip) as archive:
question_members = _safe_members(archive)
with zipfile.ZipFile(example_zip) as archive:
example_members = _safe_members(archive)
tasks: dict[str, list[str]] = {}
for info in question_members:
path = PurePosixPath(info.filename)
if len(path.parts) != 2 or path.suffix not in VIDEO_SUFFIXES:
continue
match = TASK_PATTERN.fullmatch(path.parts[0])
if match is None:
continue
tasks.setdefault(path.parts[0], []).append(info.filename)
expected = [f"item_{index:03d}" for index in range(1, 35)]
missing = sorted(set(expected) - set(tasks))
extras = sorted(set(tasks) - set(expected))
if missing or extras:
raise ValueError(f"question archive task inventory mismatch: missing={missing}, extras={extras}")
example_prefix = "submission_example/submission/"
if not any(info.filename.startswith(example_prefix) for info in example_members):
raise ValueError("submission_example.zip lacks submission_example/submission/")
return {
"created_at": utc_now(),
"input_contract": "official_question_zip_plus_official_submission_example_zip_only",
"question_zip": {"path": str(question_zip), "sha256": sha256(question_zip), "bytes": question_zip.stat().st_size},
"submission_example_zip": {"path": str(example_zip), "sha256": sha256(example_zip), "bytes": example_zip.stat().st_size},
"task_count": len(tasks),
"video_count": sum(len(value) for value in tasks.values()),
"tasks": [{"task_id": task_id, "videos": sorted(tasks[task_id])} for task_id in expected],
"example_member_count": len(example_members),
}
def extract_question(question_zip: Path, manifest: dict[str, Any], output_root: Path) -> dict[str, Any]:
"""Extract only verified official videos under a new run directory."""
output_root.mkdir(parents=True, exist_ok=True)
wanted = {video for task in manifest["tasks"] for video in task["videos"]}
records: list[dict[str, Any]] = []
with zipfile.ZipFile(question_zip) as archive:
infos = {info.filename: info for info in _safe_members(archive)}
if wanted - set(infos):
raise ValueError(f"question archive changed after inspection: {sorted(wanted - set(infos))[:3]}")
for task in manifest["tasks"]:
task_id = task["task_id"]
for member in task["videos"]:
target = output_root / member
target.parent.mkdir(parents=True, exist_ok=True)
if target.exists():
raise FileExistsError(target)
with archive.open(infos[member]) as source, target.open("xb") as destination:
while True:
block = source.read(1024 * 1024)
if not block:
break
destination.write(block)
records.append({"task_id": task_id, "member": member, "path": str(target), "bytes": target.stat().st_size, "sha256": sha256(target)})
return {"created_at": utc_now(), "video_root": str(output_root), "videos": records}
def write_input_manifest(question_zip: Path, example_zip: Path, destination: Path) -> dict[str, Any]:
manifest = inspect_inputs(question_zip, example_zip)
write_json(destination, manifest)
return manifest

@ -0,0 +1,128 @@
from __future__ import annotations
import math
from pathlib import Path
from typing import Any
import cv2
import numpy as np
from PIL import Image, ImageDraw
from .util import utc_now
def _resize(frame: np.ndarray, max_edge: int) -> np.ndarray:
height, width = frame.shape[:2]
scale = min(1.0, max_edge / max(height, width))
if scale >= 1.0:
return frame
return cv2.resize(frame, (round(width * scale), round(height * scale)), interpolation=cv2.INTER_AREA)
def _frame_score(frame: np.ndarray) -> dict[str, float]:
gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
sharpness = float(cv2.Laplacian(gray, cv2.CV_64F).var())
contrast = float(gray.std())
height, width = gray.shape
central = gray[height // 4 : 3 * height // 4, width // 4 : 3 * width // 4]
central_contrast = float(central.std())
return {"sharpness": sharpness, "contrast": contrast, "central_contrast": central_contrast}
def _candidate_indices(frame_count: int, samples: int) -> list[int]:
if frame_count <= 1:
return [0]
return sorted({round(index * (frame_count - 1) / max(samples - 1, 1)) for index in range(samples)})
def _colour_distance(left: np.ndarray, right: np.ndarray) -> float:
return float(np.linalg.norm(left.astype(np.float64) - right.astype(np.float64)))
def _make_contact_sheet(records: list[dict[str, Any]], target: Path) -> None:
thumbs: list[Image.Image] = []
for record in records:
image = Image.open(record["path"]).convert("RGB")
image.thumbnail((240, 180))
canvas = Image.new("RGB", (240, 204), "white")
canvas.paste(image, ((240 - image.width) // 2, 0))
ImageDraw.Draw(canvas).text((6, 184), f"{record['source_index']}:{record['frame_index']}", fill="black")
thumbs.append(canvas)
columns = 3
rows = max(1, math.ceil(len(thumbs) / columns))
sheet = Image.new("RGB", (columns * 240, rows * 204), "white")
for index, thumb in enumerate(thumbs):
sheet.paste(thumb, ((index % columns) * 240, (index // columns) * 204))
target.parent.mkdir(parents=True, exist_ok=True)
sheet.save(target)
def extract_and_select_views(
task_id: str,
videos: list[dict[str, Any]],
output_dir: Path,
*,
frames_per_video: int,
selected_views: int,
max_edge: int,
jpeg_quality: int,
) -> dict[str, Any]:
"""Uniformly sample all clips then select sharp, visually diverse views."""
candidates: list[dict[str, Any]] = []
raw_dir = output_dir / "candidates"
raw_dir.mkdir(parents=True, exist_ok=True)
for source_index, video in enumerate(videos):
capture = cv2.VideoCapture(str(video["path"]))
if not capture.isOpened():
raise RuntimeError(f"OpenCV could not decode video: {video['path']}")
frame_count = max(1, int(capture.get(cv2.CAP_PROP_FRAME_COUNT)))
fps = float(capture.get(cv2.CAP_PROP_FPS))
for frame_index in _candidate_indices(frame_count, frames_per_video):
capture.set(cv2.CAP_PROP_POS_FRAMES, frame_index)
success, frame = capture.read()
if not success or frame is None:
continue
frame = _resize(frame, max_edge)
metrics = _frame_score(frame)
colour = frame.reshape(-1, 3).mean(axis=0).tolist()
path = raw_dir / f"s{source_index:02d}_f{frame_index:06d}.jpg"
if not cv2.imwrite(str(path), frame, [cv2.IMWRITE_JPEG_QUALITY, jpeg_quality]):
raise RuntimeError(f"failed to write frame: {path}")
candidates.append({
"path": str(path), "source_index": source_index, "source_video": video["member"],
"frame_index": frame_index, "frame_count": frame_count, "fps": fps,
"mean_bgr": colour, **metrics,
})
capture.release()
if not candidates:
raise RuntimeError(f"no decodable frames for {task_id}")
for key in ("sharpness", "contrast", "central_contrast"):
values = np.asarray([record[key] for record in candidates], dtype=np.float64)
low, high = float(values.min()), float(values.max())
for record in candidates:
record[f"norm_{key}"] = (record[key] - low) / max(high - low, 1e-9)
for record in candidates:
record["base_score"] = sum(record[f"norm_{name}"] for name in ("sharpness", "contrast", "central_contrast"))
ranked = sorted(candidates, key=lambda record: (-record["base_score"], record["source_index"], record["frame_index"]))
chosen: list[dict[str, Any]] = []
for candidate in ranked:
diversity = 1.0 if not chosen else min(_colour_distance(np.asarray(candidate["mean_bgr"]), np.asarray(old["mean_bgr"])) / 255.0 for old in chosen)
if len(chosen) < selected_views and (not chosen or diversity >= 0.04):
candidate["diversity"] = diversity
chosen.append(candidate)
for candidate in ranked:
if len(chosen) >= selected_views:
break
if candidate not in chosen:
candidate["diversity"] = 0.0
chosen.append(candidate)
selected_dir = output_dir / "selected"
selected_dir.mkdir(parents=True, exist_ok=True)
selected: list[dict[str, Any]] = []
for index, record in enumerate(chosen):
target = selected_dir / f"view_{index:02d}.jpg"
image = Image.open(record["path"]).convert("RGB")
image.save(target, quality=jpeg_quality)
selected.append({**record, "path": str(target), "view_index": index})
_make_contact_sheet(selected, output_dir / "contact_sheet.jpg")
return {"task_id": task_id, "created_at": utc_now(), "candidate_count": len(candidates), "selected": selected, "contact_sheet": str(output_dir / "contact_sheet.jpg")}

@ -0,0 +1,181 @@
from __future__ import annotations
import platform
import sys
import zipfile
from pathlib import Path
from typing import Any
from .input_data import extract_question, inspect_inputs, write_input_manifest
from .media import extract_and_select_views
from .reconstruct import HunyuanReconstructor, _load_mesh, create_reconstructor, finalize_mesh, reconstruct_mesh
from .segmentation import make_conditioning_image
from .usd_asset import simulate_mjcf, write_usd_asset
from .util import clean_dir, iter_files, read_json, seed_everything, sha256, utc_now, write_json
def run_root(config: dict[str, Any]) -> Path:
return Path(str(config["output_root"])) / str(config["run_name"])
def runtime_record(config: dict[str, Any]) -> dict[str, Any]:
return {
"created_at": utc_now(),
"input_contract": "question.zip + submission_example.zip only; no work/v* or historic submission input is permitted",
"python": sys.version,
"platform": platform.platform(),
"config": config,
}
def inspect(config: dict[str, Any]) -> dict[str, Any]:
return inspect_inputs(Path(config["question_zip"]), Path(config["submission_example_zip"]))
def prepare(config: dict[str, Any]) -> Path:
root = run_root(config)
clean_dir(root)
seed_everything(int(config["seed"]))
manifest = write_input_manifest(Path(config["question_zip"]), Path(config["submission_example_zip"]), root / "manifests" / "inputs.json")
write_json(root / "manifests" / "runtime.json", runtime_record(config))
extracted = extract_question(Path(config["question_zip"]), manifest, root / "raw_videos")
write_json(root / "manifests" / "extracted.json", extracted)
by_task: dict[str, list[dict[str, Any]]] = {task["task_id"]: [] for task in manifest["tasks"]}
for record in extracted["videos"]:
by_task[record["task_id"]].append(record)
view_records = []
video_config = config["video"]
for task in manifest["tasks"]:
task_id = task["task_id"]
view_records.append(extract_and_select_views(
task_id,
sorted(by_task[task_id], key=lambda record: record["member"]),
root / "views" / task_id,
frames_per_video=int(video_config["frames_per_video"]),
selected_views=int(video_config["selected_views"]),
max_edge=int(video_config["max_edge"]),
jpeg_quality=int(video_config["jpeg_quality"]),
))
write_json(root / "manifests" / "views.json", {"created_at": utc_now(), "tasks": view_records})
return root
def _require_prepared(config: dict[str, Any]) -> tuple[Path, dict[str, Any]]:
root = run_root(config)
manifest_path = root / "manifests" / "views.json"
if not manifest_path.is_file():
raise FileNotFoundError(f"prepared views missing; run `asset_baseline prepare` first: {manifest_path}")
return root, read_json(manifest_path)
def generate(config: dict[str, Any]) -> Path:
root, views = _require_prepared(config)
submission = root / "submission"
if submission.exists():
raise FileExistsError(f"generation already exists; use a new run_name: {submission}")
seed_everything(int(config["seed"]))
assets: list[dict[str, Any]] = []
reconstructor = create_reconstructor(config["reconstruction"])
if isinstance(reconstructor, HunyuanReconstructor) and reconstructor.sequential:
staged: list[dict[str, Any]] = []
# On a 24 GiB GPU, generate every Shape result first while Shape is
# resident, then release it before Paint is loaded.
for task in views["tasks"]:
task_id = task["task_id"]
selected = task["selected"]
if not selected:
raise RuntimeError(f"no selected views: {task_id}")
task_root = root / "generated" / task_id
conditioning_path = task_root / "conditioning.png"
segmentation = make_conditioning_image(Path(selected[0]["path"]), conditioning_path, config["segmentation"])
shape_path = reconstructor.generate_shape(conditioning_path, task_root / "mesh_shape.glb")
staged.append({"task_id": task_id, "selected": selected[0], "task_root": task_root, "conditioning": conditioning_path, "segmentation": segmentation, "shape": shape_path})
reconstructor.begin_texture_phase()
for record in staged:
final_path = reconstructor.generate_texture(
record["shape"],
record["conditioning"],
record["task_root"] / "mesh_textured.obj",
)
mesh = finalize_mesh(
record["task_id"],
_load_mesh(final_path),
record["task_root"] / "mesh.obj",
config["reconstruction"],
"Hunyuan3D-2.1",
)
asset = write_usd_asset(record["task_id"], Path(mesh["mesh"]), record["conditioning"], submission / record["task_id"], config["physics"])
assets.append({"task_id": record["task_id"], "source_view": record["selected"], "segmentation": record["segmentation"], "mesh": mesh, "asset": asset})
write_json(root / "manifests" / "assets.json", {"created_at": utc_now(), "tasks": assets})
return root
for task in views["tasks"]:
task_id = task["task_id"]
selected = task["selected"]
if not selected:
raise RuntimeError(f"no selected views: {task_id}")
task_root = root / "generated" / task_id
conditioning_path = task_root / "conditioning.png"
segmentation = make_conditioning_image(Path(selected[0]["path"]), conditioning_path, config["segmentation"])
mesh = reconstruct_mesh(
task_id,
conditioning_path,
task_root / "mesh.obj",
config["reconstruction"],
reconstructor=reconstructor,
)
asset = write_usd_asset(task_id, Path(mesh["mesh"]), conditioning_path, submission / task_id, config["physics"])
assets.append({"task_id": task_id, "source_view": selected[0], "segmentation": segmentation, "mesh": mesh, "asset": asset})
write_json(root / "manifests" / "assets.json", {"created_at": utc_now(), "tasks": assets})
return root
def _archive_members(submission: Path) -> list[Path]:
expected = [submission / f"item_{index:03d}" / f"item_{index:03d}.usd" for index in range(1, 35)]
missing = [str(path) for path in expected if not path.is_file()]
if missing:
raise ValueError(f"submission has missing task USDs: {missing[:3]}")
return list(iter_files(submission))
def package(config: dict[str, Any]) -> Path:
root, _ = _require_prepared(config)
submission = root / "submission"
members = _archive_members(submission)
package_path = root / "packages" / "asset_baseline_submission.zip"
if package_path.exists():
raise FileExistsError(package_path)
package_path.parent.mkdir(parents=True)
archive_root = str(config["package"]["archive_root"]).rstrip("/")
level = int(config["package"]["compression_level"])
with zipfile.ZipFile(package_path, "w", compression=zipfile.ZIP_DEFLATED, compresslevel=level, strict_timestamps=True) as archive:
for source in members:
relative = source.relative_to(submission).as_posix()
info = zipfile.ZipInfo(f"{archive_root}/{relative}", date_time=(2026, 1, 1, 0, 0, 0))
info.compress_type = zipfile.ZIP_DEFLATED
info.external_attr = 0o100644 << 16
archive.writestr(info, source.read_bytes(), compress_type=zipfile.ZIP_DEFLATED, compresslevel=level)
if zipfile.ZipFile(package_path).testzip() is not None:
raise RuntimeError("generated ZIP CRC failure")
write_json(root / "manifests" / "package.json", {"created_at": utc_now(), "package": str(package_path), "sha256": sha256(package_path), "bytes": package_path.stat().st_size, "file_count": len(members)})
return package_path
def validate(config: dict[str, Any]) -> dict[str, Any]:
root, _ = _require_prepared(config)
submission = root / "submission"
_archive_members(submission)
from .validate import validate_submission_tree
report = validate_submission_tree(submission, root / "physics", int(config["physics"]["simulation_steps"]))
package_path = root / "packages" / "asset_baseline_submission.zip"
if package_path.is_file():
report["package"] = {"path": str(package_path), "sha256": sha256(package_path), "bytes": package_path.stat().st_size}
write_json(root / "reports" / "validation.json", report)
return report
def run_all(config: dict[str, Any]) -> dict[str, Any]:
prepare(config)
generate(config)
package(config)
return validate(config)

@ -0,0 +1,220 @@
from __future__ import annotations
import importlib
import gc
import sys
from pathlib import Path
from typing import Any
import numpy as np
import trimesh
from PIL import Image
def _primitive_mesh(image_path: Path) -> trimesh.Trimesh:
image = Image.open(image_path).convert("RGBA")
width, height = image.size
aspect = max(0.35, min(2.6, width / max(height, 1)))
# A shallow closed cuboid is safer than a textured plane in a physics engine.
return trimesh.creation.box(extents=(aspect, 0.28, 1.0))
def _load_mesh(path: Path) -> trimesh.Trimesh:
loaded = trimesh.load(path, force="scene")
if isinstance(loaded, trimesh.Scene):
meshes = [geometry for geometry in loaded.geometry.values() if isinstance(geometry, trimesh.Trimesh)]
if not meshes:
raise RuntimeError(f"Hunyuan output has no mesh: {path}")
if len(meshes) == 1:
# Do not concatenate a one-mesh textured OBJ: trimesh's generic
# concatenate path can replace TextureVisuals with vertex colours.
mesh = meshes[0].copy()
else:
textured = [
geometry
for geometry in meshes
if getattr(getattr(getattr(geometry, "visual", None), "material", None), "image", None) is not None
]
if textured:
raise RuntimeError(
f"Hunyuan output has {len(meshes)} textured geometries; "
"this baseline writes one USD material and refuses to silently discard texture assignments"
)
mesh = trimesh.util.concatenate(meshes)
elif isinstance(loaded, trimesh.Trimesh):
mesh = loaded
else:
raise RuntimeError(f"unsupported mesh payload: {type(loaded).__name__}")
if len(mesh.faces) == 0:
raise RuntimeError(f"empty mesh: {path}")
return mesh
def _export_result(mesh: Any, output: Path) -> Path:
output.parent.mkdir(parents=True, exist_ok=True)
if isinstance(mesh, (str, Path)):
source = Path(mesh)
if source.resolve() != output.resolve():
output.write_bytes(source.read_bytes())
return output
if hasattr(mesh, "export"):
mesh.export(str(output))
return output
if hasattr(mesh, "save"):
mesh.save(str(output))
return output
raise TypeError(f"cannot export Hunyuan mesh type: {type(mesh).__name__}")
class HunyuanReconstructor:
"""One GPU model session reused across every task in a run.
Loading both Hunyuan Shape and Paint for every one of 34 assets is not a
viable GPU workflow: it needlessly downloads/initializes models, lengthens
the run and can fragment VRAM. This session is made exactly once by the
pipeline and then processes each conditioning image in turn.
"""
def __init__(self, reconstruction: dict[str, Any]) -> None:
self.reconstruction = reconstruction
self.load_mode = str(reconstruction.get("hunyuan_load_mode", "resident"))
if self.load_mode not in {"resident", "sequential"}:
raise ValueError(f"unsupported hunyuan_load_mode: {self.load_mode}")
self.repo = Path(str(reconstruction["hunyuan_repo"])).resolve()
if not self.repo.is_dir():
raise FileNotFoundError(f"Hunyuan3D checkout not found: {self.repo}; run scripts/bootstrap_models.py --hunyuan")
for relative in ("hy3dshape", "hy3dpaint"):
candidate = str(self.repo / relative)
if candidate not in sys.path:
sys.path.insert(0, candidate)
self.shape_pipeline: Any | None = self._load_shape_pipeline()
self.paint_pipeline: Any | None = None
if self.wants_texture and self.load_mode == "resident":
self.paint_pipeline = self._load_paint_pipeline()
@property
def wants_texture(self) -> bool:
return bool(self.reconstruction.get("hunyuan_texture", True))
@property
def sequential(self) -> bool:
return self.load_mode == "sequential"
def _load_shape_pipeline(self) -> Any:
try: # imports are deliberately lazy so CPU-only validation has no model dependency
from hy3dshape.pipelines import Hunyuan3DDiTFlowMatchingPipeline
except ImportError as error: # pragma: no cover - GPU-only integration
raise RuntimeError("Hunyuan3D dependencies are unavailable; follow README section 'GPU install'.") from error
return Hunyuan3DDiTFlowMatchingPipeline.from_pretrained(str(self.reconstruction["hunyuan_model"]))
def _load_paint_pipeline(self) -> Any:
try:
texture_module = importlib.import_module("textureGenPipeline")
paint_config = texture_module.Hunyuan3DPaintConfig(
max_num_view=int(self.reconstruction.get("hunyuan_paint_max_views", 6)),
resolution=int(self.reconstruction.get("hunyuan_paint_resolution", 512)),
)
# Hunyuan's own demo is run from its checkout. The baseline is
# invoked elsewhere, so resolve its local files explicitly.
paint_config.realesrgan_ckpt_path = str(self.repo / "hy3dpaint" / "ckpt" / "RealESRGAN_x4plus.pth")
paint_config.multiview_cfg_path = str(self.repo / "hy3dpaint" / "cfgs" / "hunyuan-paint-pbr.yaml")
paint_config.custom_pipeline = str(self.repo / "hy3dpaint" / "hunyuanpaintpbr")
return texture_module.Hunyuan3DPaintPipeline(paint_config)
except Exception as error: # pragma: no cover - GPU-only integration
raise RuntimeError("Hunyuan texture pipeline could not initialize; do not silently publish an untextured high profile.") from error
def generate_shape(self, image_path: Path, shape_path: Path) -> Path:
if self.shape_pipeline is None:
raise RuntimeError("Hunyuan shape model was released before all shapes were generated")
generated = self.shape_pipeline(image=str(image_path))[0]
return _export_result(generated, shape_path)
def begin_texture_phase(self) -> None:
"""Release Shape before loading Paint on a 24 GiB GPU."""
if not self.sequential or not self.wants_texture:
return
self.shape_pipeline = None
gc.collect()
try: # pragma: no cover - exercised only on a CUDA host
import torch
torch.cuda.empty_cache()
except Exception:
pass
self.paint_pipeline = self._load_paint_pipeline()
def generate_texture(self, shape_path: Path, image_path: Path, output: Path) -> Path:
if not self.wants_texture:
return shape_path
if self.paint_pipeline is None:
raise RuntimeError("Hunyuan Paint is not initialized; call begin_texture_phase() first in sequential mode")
try:
textured = self.paint_pipeline(
mesh_path=str(shape_path),
image_path=str(image_path),
output_mesh_path=str(output),
)
return Path(str(textured))
except Exception as error: # pragma: no cover - GPU-only integration
raise RuntimeError("Hunyuan shape generation succeeded but texture generation failed; do not silently publish an untextured high profile.") from error
def reconstruct(self, image_path: Path, output: Path) -> trimesh.Trimesh:
shape_path = output.with_name(output.stem + "_shape.glb")
self.generate_shape(image_path, shape_path)
final_path = self.generate_texture(shape_path, image_path, output.with_name(output.stem + "_textured.obj"))
return _load_mesh(final_path)
def create_reconstructor(reconstruction: dict[str, Any]) -> HunyuanReconstructor | None:
backend = str(reconstruction["backend"])
if backend == "primitive":
return None
if backend == "hunyuan":
return HunyuanReconstructor(reconstruction)
raise ValueError(f"unknown reconstruction backend: {backend}")
def finalize_mesh(task_id: str, mesh: trimesh.Trimesh, output: Path, reconstruction: dict[str, Any], source: str) -> dict[str, Any]:
"""Normalize, simplify and export a mesh after any reconstruction backend."""
backend = str(reconstruction["backend"])
source_faces = int(len(mesh.faces))
simplification_error: str | None = None
if len(mesh.faces) > int(reconstruction["max_faces"]):
try:
mesh = mesh.simplify_quadric_decimation(int(reconstruction["max_faces"]))
except Exception as error:
# The optional trimesh decimator has platform-specific native
# dependencies. Keep a valid mesh and expose the condition in the
# manifest rather than replacing it with a lower-quality primitive.
simplification_error = f"{type(error).__name__}: {error}"
if len(mesh.faces) == 0:
raise RuntimeError(f"reconstruction produced no faces: {task_id}")
mesh.remove_unreferenced_vertices()
extent = np.asarray(mesh.extents, dtype=np.float64)
scale = 1.0 / max(float(extent.max()), 1e-6)
mesh.apply_scale(scale)
mesh.apply_translation(-mesh.bounds.mean(axis=0))
mesh.apply_translation([0.0, 0.0, -float(mesh.bounds[0, 2])])
output.parent.mkdir(parents=True, exist_ok=True)
mesh.export(str(output))
return {"task_id": task_id, "backend": backend, "source": source, "mesh": str(output), "vertices": int(len(mesh.vertices)), "faces": int(len(mesh.faces)), "source_faces": source_faces, "simplification_error": simplification_error, "extent": [float(value) for value in mesh.extents]}
def reconstruct_mesh(
task_id: str,
conditioning_image: Path,
output: Path,
reconstruction: dict[str, Any],
*,
reconstructor: HunyuanReconstructor | None = None,
) -> dict[str, Any]:
backend = str(reconstruction["backend"])
if backend == "primitive":
mesh = _primitive_mesh(conditioning_image)
source = "deterministic_primitive"
elif backend == "hunyuan":
session = reconstructor if reconstructor is not None else HunyuanReconstructor(reconstruction)
mesh = session.reconstruct(conditioning_image, output)
source = "Hunyuan3D-2.1"
else:
raise ValueError(f"unknown reconstruction backend: {backend}")
return finalize_mesh(task_id, mesh, output, reconstruction, source)

@ -0,0 +1,100 @@
from __future__ import annotations
from pathlib import Path
from typing import Any
import cv2
import numpy as np
from PIL import Image
def _center_mask(image: np.ndarray) -> np.ndarray:
"""Deterministic non-model fallback using GrabCut and a centre prior."""
height, width = image.shape[:2]
mask = np.zeros((height, width), np.uint8)
margin_x, margin_y = max(2, width // 12), max(2, height // 12)
rectangle = (margin_x, margin_y, max(1, width - 2 * margin_x), max(1, height - 2 * margin_y))
background = np.zeros((1, 65), np.float64)
foreground = np.zeros((1, 65), np.float64)
try:
cv2.grabCut(image, mask, rectangle, background, foreground, 4, cv2.GC_INIT_WITH_RECT)
result = np.where((mask == cv2.GC_FGD) | (mask == cv2.GC_PR_FGD), 255, 0).astype(np.uint8)
except cv2.error:
result = np.zeros((height, width), np.uint8)
result[margin_y : height - margin_y, margin_x : width - margin_x] = 255
components, labels, stats, centroids = cv2.connectedComponentsWithStats(result)
if components <= 1:
return result
centre = np.asarray([width / 2, height / 2])
best = max(
range(1, components),
key=lambda index: float(stats[index, cv2.CC_STAT_AREA]) / (1.0 + np.linalg.norm(centroids[index] - centre) / max(width, height)),
)
return np.where(labels == best, 255, 0).astype(np.uint8)
def _sam2_mask(image: np.ndarray, model_id: str, min_area_fraction: float, max_area_fraction: float) -> np.ndarray:
try:
from sam2.automatic_mask_generator import SAM2AutomaticMaskGenerator
except ImportError as error: # pragma: no cover - only exercised on a GPU host
raise RuntimeError("SAM2 is not installed. Run scripts/bootstrap_models.py --sam2 first, or set segmentation.backend=center.") from error
generator = SAM2AutomaticMaskGenerator.from_pretrained(
model_id,
points_per_side=32,
pred_iou_thresh=0.80,
stability_score_thresh=0.95,
crop_n_layers=1,
min_mask_region_area=400,
)
annotations = generator.generate(cv2.cvtColor(image, cv2.COLOR_BGR2RGB))
height, width = image.shape[:2]
centre = np.asarray([width / 2, height / 2])
candidates: list[tuple[float, np.ndarray]] = []
for annotation in annotations:
fraction = float(annotation["area"]) / float(width * height)
if not min_area_fraction <= fraction <= max_area_fraction:
continue
x, y, box_width, box_height = annotation["bbox"]
box_centre = np.asarray([x + box_width / 2, y + box_height / 2])
centrality = 1.0 - min(1.0, float(np.linalg.norm(box_centre - centre)) / (0.55 * max(width, height)))
score = 0.55 * float(annotation["predicted_iou"]) + 0.30 * float(annotation["stability_score"]) + 0.15 * centrality
candidates.append((score, np.asarray(annotation["segmentation"], dtype=np.uint8) * 255))
if not candidates:
raise RuntimeError("SAM2 found no plausible centred foreground; inspect the task contact sheet and retry with segmentation.backend=center")
return max(candidates, key=lambda pair: pair[0])[1]
def make_conditioning_image(source: Path, output: Path, segmentation: dict[str, Any]) -> dict[str, Any]:
image = cv2.imread(str(source), cv2.IMREAD_COLOR)
if image is None:
raise ValueError(f"could not read selected view: {source}")
backend = str(segmentation["backend"])
if backend == "center":
mask = _center_mask(image)
elif backend == "sam2":
mask = _sam2_mask(
image,
str(segmentation["sam2_model"]),
float(segmentation["min_area_fraction"]),
float(segmentation["max_area_fraction"]),
)
else:
raise ValueError(f"unknown segmentation backend: {backend}")
x, y, width, height = cv2.boundingRect(mask)
if width <= 2 or height <= 2:
raise RuntimeError(f"foreground mask is empty: {source}")
padding = max(4, round(0.06 * max(width, height)))
x0, y0 = max(0, x - padding), max(0, y - padding)
x1, y1 = min(image.shape[1], x + width + padding), min(image.shape[0], y + height + padding)
cropped_bgr = image[y0:y1, x0:x1]
cropped_mask = mask[y0:y1, x0:x1]
rgba = cv2.cvtColor(cropped_bgr, cv2.COLOR_BGR2RGBA)
rgba[:, :, 3] = cropped_mask
output.parent.mkdir(parents=True, exist_ok=True)
Image.fromarray(rgba).save(output)
return {
"source": str(source), "output": str(output), "backend": backend,
"crop_xyxy": [int(x0), int(y0), int(x1), int(y1)],
"mask_area_fraction": float((mask > 0).mean()),
"crop_width": int(x1 - x0), "crop_height": int(y1 - y0),
}

@ -0,0 +1,151 @@
from __future__ import annotations
import shutil
from pathlib import Path
from typing import Any
import mujoco
import numpy as np
import trimesh
from PIL import Image
from pxr import Gf, Sdf, Usd, UsdGeom, UsdPhysics, UsdShade
from .util import write_json
def _texture_coordinates(mesh: trimesh.Trimesh) -> np.ndarray:
visual = getattr(mesh, "visual", None)
uv = getattr(visual, "uv", None)
if uv is not None and len(uv) == len(mesh.vertices):
return np.asarray(uv, dtype=np.float32)
points = np.asarray(mesh.vertices, dtype=np.float64)
lower, upper = points[:, :2].min(axis=0), points[:, :2].max(axis=0)
return ((points[:, :2] - lower) / np.maximum(upper - lower, 1e-6)).astype(np.float32)
def _material(stage: Usd.Stage, texture: str) -> UsdShade.Material:
material = UsdShade.Material.Define(stage, "/World/Materials/Appearance")
shader = UsdShade.Shader.Define(stage, "/World/Materials/Appearance/PreviewSurface")
shader.CreateIdAttr("UsdPreviewSurface")
shader.CreateInput("roughness", Sdf.ValueTypeNames.Float).Set(0.55)
texture_node = UsdShade.Shader.Define(stage, "/World/Materials/Appearance/Texture")
texture_node.CreateIdAttr("UsdUVTexture")
texture_node.CreateInput("file", Sdf.ValueTypeNames.Asset).Set(Sdf.AssetPath(texture))
texture_node.CreateInput("sourceColorSpace", Sdf.ValueTypeNames.Token).Set("sRGB")
reader = UsdShade.Shader.Define(stage, "/World/Materials/Appearance/ST")
reader.CreateIdAttr("UsdPrimvarReader_float2")
reader.CreateInput("varname", Sdf.ValueTypeNames.Token).Set("st")
reader.CreateOutput("result", Sdf.ValueTypeNames.Float2)
texture_node.CreateInput("st", Sdf.ValueTypeNames.Float2).ConnectToSource(reader.ConnectableAPI(), "result")
texture_node.CreateOutput("rgb", Sdf.ValueTypeNames.Float3)
shader.CreateInput("diffuseColor", Sdf.ValueTypeNames.Color3f).ConnectToSource(texture_node.ConnectableAPI(), "rgb")
shader.CreateOutput("surface", Sdf.ValueTypeNames.Token)
material.CreateSurfaceOutput().ConnectToSource(shader.ConnectableAPI(), "surface")
return material
def _write_texture(mesh: trimesh.Trimesh, conditioning_image: Path, destination: Path) -> str:
"""Persist the generated GLB albedo when available, otherwise use the input view.
A high-profile Hunyuan result carries a UV texture in its GLB. Keeping that
image is essential: copying the conditioning frame unconditionally would
quietly discard Hunyuan Paint's result.
"""
material = getattr(getattr(mesh, "visual", None), "material", None)
generated = getattr(material, "image", None)
if generated is not None:
if isinstance(generated, Image.Image):
generated.convert("RGBA").save(destination)
else:
Image.fromarray(np.asarray(generated)).convert("RGBA").save(destination)
return "reconstructed_mesh_albedo"
shutil.copy2(conditioning_image, destination)
return "conditioning_image_fallback"
def _write_mjcf(task_id: str, extent: np.ndarray, output: Path, physics: dict[str, Any]) -> None:
size = np.maximum(extent / 2.0, 0.015)
xml = f'''<?xml version="1.0" encoding="utf-8"?>
<mujoco model="{task_id}">
<option timestep="0.002" gravity="0 0 -9.81"/>
<worldbody>
<body name="asset" pos="0 0 1">
<freejoint/>
<geom type="box" size="{size[0]:.7f} {size[1]:.7f} {size[2]:.7f}" mass="{float(physics['default_mass_kg']):.7f}" friction="{float(physics['friction']):.7f} 0.02 0.002"/>
</body>
</worldbody>
</mujoco>
'''
output.parent.mkdir(parents=True, exist_ok=True)
output.write_text(xml, encoding="utf-8")
def write_usd_asset(task_id: str, mesh_path: Path, conditioning_image: Path, output_dir: Path, physics: dict[str, Any]) -> dict[str, Any]:
"""Create a self-contained binary USD with visual mesh and collision data."""
mesh = trimesh.load(mesh_path, force="mesh")
if not isinstance(mesh, trimesh.Trimesh) or len(mesh.faces) == 0:
raise ValueError(f"invalid reconstructed mesh: {mesh_path}")
output_dir.mkdir(parents=True, exist_ok=True)
texture_dir = output_dir / "textures"
texture_dir.mkdir(exist_ok=True)
texture_path = texture_dir / "texture_00.png"
texture_source = _write_texture(mesh, conditioning_image, texture_path)
usd_path = output_dir / f"{task_id}.usd"
stage = Usd.Stage.CreateNew(str(usd_path))
UsdGeom.SetStageUpAxis(stage, UsdGeom.Tokens.z)
UsdGeom.SetStageMetersPerUnit(stage, 1.0)
world = UsdGeom.Xform.Define(stage, "/World")
stage.SetDefaultPrim(world.GetPrim())
UsdPhysics.Scene.Define(stage, "/World/PhysicsScene")
asset = UsdGeom.Xform.Define(stage, "/World/Asset")
UsdPhysics.RigidBodyAPI.Apply(asset.GetPrim())
mass = UsdPhysics.MassAPI.Apply(asset.GetPrim())
mass.CreateMassAttr(float(physics["default_mass_kg"]))
visual = UsdGeom.Mesh.Define(stage, "/World/Asset/Visual")
points = np.asarray(mesh.vertices, dtype=np.float32)
visual.CreatePointsAttr([Gf.Vec3f(float(point[0]), float(point[1]), float(point[2])) for point in points])
visual.CreateFaceVertexCountsAttr([3] * len(mesh.faces))
visual.CreateFaceVertexIndicesAttr([int(index) for face in mesh.faces for index in face])
visual.CreateSubdivisionSchemeAttr(UsdGeom.Tokens.none)
visual.CreateExtentAttr([
Gf.Vec3f(float(mesh.bounds[0, 0]), float(mesh.bounds[0, 1]), float(mesh.bounds[0, 2])),
Gf.Vec3f(float(mesh.bounds[1, 0]), float(mesh.bounds[1, 1]), float(mesh.bounds[1, 2])),
])
uv = _texture_coordinates(mesh)
primvars = UsdGeom.PrimvarsAPI(visual)
st = primvars.CreatePrimvar("st", Sdf.ValueTypeNames.TexCoord2fArray, UsdGeom.Tokens.vertex)
st.Set([Gf.Vec2f(float(value[0]), float(value[1])) for value in uv])
UsdShade.MaterialBindingAPI.Apply(visual.GetPrim()).Bind(_material(stage, "textures/texture_00.png"))
collision_mode = str(physics.get("collision_mode", "convex_hull"))
extent = np.asarray(mesh.extents, dtype=np.float32)
if collision_mode == "convex_hull":
# A convex hull follows the reconstructed silhouette much more closely
# than one axis-aligned cube, while remaining a stable dynamic shape in
# USD physics engines. The visible mesh stays the render mesh.
UsdPhysics.CollisionAPI.Apply(visual.GetPrim())
collision = UsdPhysics.MeshCollisionAPI.Apply(visual.GetPrim())
collision.CreateApproximationAttr().Set(UsdPhysics.Tokens.convexHull)
elif collision_mode == "bounding_box":
collider = UsdGeom.Cube.Define(stage, "/World/Asset/Collision")
collider.CreateSizeAttr(1.0)
collider.AddScaleOp().Set(Gf.Vec3f(float(extent[0] / 2.0), float(extent[1] / 2.0), float(extent[2] / 2.0)))
collider.CreateVisibilityAttr(UsdGeom.Tokens.invisible)
UsdPhysics.CollisionAPI.Apply(collider.GetPrim())
else:
raise ValueError(f"unsupported collision_mode: {collision_mode}")
stage.GetRootLayer().Save()
if not usd_path.is_file() or Usd.Stage.Open(str(usd_path)) is None:
raise RuntimeError(f"USD write/reopen failed: {usd_path}")
mjcf_path = output_dir.parent.parent / "physics" / task_id / f"{task_id}.xml"
_write_mjcf(task_id, extent, mjcf_path, physics)
return {"task_id": task_id, "usd": str(usd_path), "texture": str(texture_path), "texture_source": texture_source, "collision_mode": collision_mode, "mjcf": str(mjcf_path), "extent": [float(value) for value in extent]}
def simulate_mjcf(path: Path, steps: int) -> dict[str, Any]:
model = mujoco.MjModel.from_xml_path(str(path))
data = mujoco.MjData(model)
for _ in range(steps):
mujoco.mj_step(model, data)
if not np.isfinite(data.qpos).all() or not np.isfinite(data.qvel).all():
raise RuntimeError(f"non-finite MuJoCo state: {path}")
return {"mjcf": str(path), "steps": steps, "qpos": [float(value) for value in data.qpos], "qvel": [float(value) for value in data.qvel]}

@ -0,0 +1,61 @@
from __future__ import annotations
import hashlib
import json
import os
import random
import shutil
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Iterable
import numpy as np
def sha256(path: Path) -> str:
digest = hashlib.sha256()
with path.open("rb") as stream:
for block in iter(lambda: stream.read(1024 * 1024), b""):
digest.update(block)
return digest.hexdigest()
def utc_now() -> str:
return datetime.now(timezone.utc).isoformat(timespec="seconds")
def write_json(path: Path, value: Any) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps(value, ensure_ascii=False, indent=2, sort_keys=True) + "\n", encoding="utf-8")
def read_json(path: Path) -> Any:
return json.loads(path.read_text(encoding="utf-8"))
def seed_everything(seed: int) -> None:
random.seed(seed)
np.random.seed(seed)
os.environ.setdefault("PYTHONHASHSEED", str(seed))
def ensure_new_dir(path: Path) -> None:
if path.exists():
raise FileExistsError(f"refusing to overwrite existing run directory: {path}")
path.mkdir(parents=True)
def clean_dir(path: Path) -> None:
"""Create an output directory; it must be absent or an empty directory."""
if path.exists() and any(path.iterdir()):
raise FileExistsError(f"refusing to overwrite non-empty directory: {path}")
path.mkdir(parents=True, exist_ok=True)
def copy_file(source: Path, target: Path) -> None:
target.parent.mkdir(parents=True, exist_ok=True)
shutil.copy2(source, target)
def iter_files(root: Path) -> Iterable[Path]:
yield from sorted(path for path in root.rglob("*") if path.is_file() and not path.name.startswith("._"))

@ -0,0 +1,63 @@
from __future__ import annotations
from pathlib import Path
from typing import Any
from pxr import Sdf, Usd, UsdGeom, UsdPhysics, UsdShade
from .usd_asset import simulate_mjcf
from .util import utc_now
def _validate_usd(task_id: str, asset: Path) -> tuple[list[str], dict[str, int]]:
errors: list[str] = []
counts = {"gprims": 0, "colliders": 0, "rigid_bodies": 0, "physics_scenes": 0, "textures": 0}
stage = Usd.Stage.Open(str(asset))
if stage is None:
return ["Usd.Stage.Open returned None"], counts
if stage.GetDefaultPrim().GetPath() != Sdf.Path("/World"):
errors.append("default prim is not /World")
if UsdGeom.GetStageUpAxis(stage) != UsdGeom.Tokens.z:
errors.append("up axis is not Z")
if abs(float(UsdGeom.GetStageMetersPerUnit(stage)) - 1.0) > 1e-9:
errors.append("meters per unit is not 1")
for prim in stage.Traverse():
counts["gprims"] += int(prim.IsA(UsdGeom.Gprim))
counts["colliders"] += int(prim.HasAPI(UsdPhysics.CollisionAPI))
counts["rigid_bodies"] += int(prim.HasAPI(UsdPhysics.RigidBodyAPI))
counts["physics_scenes"] += int(prim.IsA(UsdPhysics.Scene))
if prim.IsA(UsdShade.Shader):
shader = UsdShade.Shader(prim)
file_input = shader.GetInput("file")
value = file_input.Get() if file_input else None
if isinstance(value, Sdf.AssetPath) and value.path:
counts["textures"] += 1
if Path(value.path).is_absolute() or value.path.startswith(("http://", "https://")):
errors.append(f"non-local texture reference: {value.path}")
elif not (asset.parent / value.path).is_file():
errors.append(f"missing texture reference: {value.path}")
if counts["gprims"] < 1:
errors.append("no renderable geometry")
if counts["physics_scenes"] != 1:
errors.append(f"expected one physics scene, found {counts['physics_scenes']}")
if counts["colliders"] < 1 or counts["rigid_bodies"] < 1:
errors.append("missing conservative collision/rigid-body APIs")
return errors, counts
def validate_submission_tree(submission: Path, physics_root: Path, steps: int) -> dict[str, Any]:
records: list[dict[str, Any]] = []
all_errors: list[str] = []
for index in range(1, 35):
task_id = f"item_{index:03d}"
usd = submission / task_id / f"{task_id}.usd"
mjcf = physics_root / task_id / f"{task_id}.xml"
errors, counts = _validate_usd(task_id, usd)
simulation: dict[str, Any] | None = None
try:
simulation = simulate_mjcf(mjcf, steps)
except Exception as error:
errors.append(f"MuJoCo: {type(error).__name__}: {error}")
all_errors.extend(f"{task_id}: {error}" for error in errors)
records.append({"task_id": task_id, "usd": str(usd), "mjcf": str(mjcf), "valid": not errors, "errors": errors, "counts": counts, "simulation": simulation})
return {"created_at": utc_now(), "valid": not all_errors, "task_count": len(records), "valid_tasks": sum(record["valid"] for record in records), "simulation_steps": steps, "errors": all_errors, "tasks": records}

@ -0,0 +1,52 @@
run_name: gpu_asset_run
seed: 20260901
# These are the only two competition-data inputs. They are intentionally not
# included in this repository and must be downloaded from the competition.
question_zip: question.zip
submission_example_zip: submission_example.zip
output_root: outputs
video:
frames_per_video: 24
selected_views: 6
max_edge: 1024
jpeg_quality: 95
# `primitive` is deterministic and CPU-friendly: it proves format/physics
# correctness. `hunyuan` is the high-fidelity GPU path used for an online run.
reconstruction:
backend: hunyuan
hunyuan_repo: third_party/Hunyuan3D-2.1
hunyuan_model: tencent/Hunyuan3D-2.1
hunyuan_texture: true
# resident keeps Shape + Paint in a 48 GiB L20. Use sequential on a 24 GiB
# 24 GiB GPU: make all shapes, free Shape, then make all textures.
hunyuan_load_mode: resident
# The L20 profile retains the public model's high-quality Paint defaults.
# Smaller values are explicit compatibility overrides for lower-VRAM GPUs.
hunyuan_paint_max_views: 6
hunyuan_paint_resolution: 512
max_faces: 30000
# `center` needs no model. `sam2` uses the pinned, public SAM2 checkout prepared
# by scripts/bootstrap_models.py and selects the best centered automatic mask.
segmentation:
backend: sam2
sam2_repo: third_party/sam2
sam2_model: facebook/sam2.1-hiera-large
min_area_fraction: 0.03
max_area_fraction: 0.90
physics:
# convex_hull follows the reconstructed silhouette; bounding_box is a
# conservative compatibility fallback for a restrictive downstream engine.
collision_mode: convex_hull
default_mass_kg: 1.0
density_kg_m3: 700.0
friction: 0.8
simulation_steps: 120
package:
archive_root: submission_example/submission
compression_level: 6

@ -0,0 +1,7 @@
{
"candidate_sha256": "2e979d08d10a785e0c47a4a1ba923131a52c8e53970bb0578f815893585e8d4b",
"evidence_type": "external_online_result",
"score": 67.6,
"score_display": "67.6000",
"status": "completed"
}

@ -0,0 +1,31 @@
#!/usr/bin/env bash
set -euo pipefail
MIN_GPU_MEMORY_MIB="${MIN_GPU_MEMORY_MIB:-45000}"
MIN_FREE_DISK_GIB="${MIN_FREE_DISK_GIB:-200}"
command -v nvidia-smi >/dev/null
command -v docker >/dev/null
GPU_INFO="$(nvidia-smi --query-gpu=name,memory.total,driver_version --format=csv,noheader,nounits)"
printf '%s\n' "$GPU_INFO"
MAX_GPU_MEMORY_MIB="$(printf '%s\n' "$GPU_INFO" | awk -F',' '
{ gsub(/^[[:space:]]+|[[:space:]]+$/, "", $2); if ($2 + 0 > max) max = $2 + 0 }
END { print max + 0 }
')"
if (( MAX_GPU_MEMORY_MIB < MIN_GPU_MEMORY_MIB )); then
echo "Need at least ${MIN_GPU_MEMORY_MIB} MiB GPU memory; detected ${MAX_GPU_MEMORY_MIB} MiB" >&2
exit 1
fi
FREE_DISK_KIB="$(df -Pk . | awk 'NR == 2 { print $4 }')"
MIN_FREE_DISK_KIB="$((MIN_FREE_DISK_GIB * 1024 * 1024))"
if (( FREE_DISK_KIB < MIN_FREE_DISK_KIB )); then
echo "Need at least ${MIN_FREE_DISK_GIB} GiB free disk; detected $((FREE_DISK_KIB / 1024 / 1024)) GiB" >&2
exit 1
fi
echo "Free disk: $((FREE_DISK_KIB / 1024 / 1024)) GiB"
docker --version
docker run --rm --gpus all nvidia/cuda:12.4.1-base-ubuntu22.04 nvidia-smi

@ -0,0 +1,31 @@
[build-system]
requires = ["setuptools>=68"]
build-backend = "setuptools.build_meta"
[project]
name = "public-asset-baseline"
version = "0.1.0"
description = "Raw-video to simulation-ready USD baseline for the 2026 embodied synthesis challenge"
requires-python = ">=3.10,<3.13"
dependencies = [
"numpy>=1.24,<2",
"opencv-python-headless>=4.10,<5",
"Pillow>=10,<12",
"PyYAML>=6,<7",
"trimesh>=4,<5",
"usd-core>=26.8,<27",
"mujoco>=3.3,<4",
]
[project.scripts]
asset-baseline = "asset_baseline.cli:main"
[tool.setuptools]
package-dir = {"" = "src"}
[tool.setuptools.packages.find]
where = ["src"]
[tool.pytest.ini_options]
testpaths = ["tests"]
addopts = "-q"

@ -0,0 +1,14 @@
# GPU runtime pins for the source pipeline. bootstrap_models.py installs its
# Python-3.10--3.12-compatible Hunyuan runtime subset after this base layer.
# Install the CUDA wheel first with the index in README/Dockerfile.
torch==2.5.1
torchvision==0.20.1
torchaudio==2.5.1
hydra-core==1.3.2
huggingface_hub==0.30.2
einops==0.8.0
diffusers==0.30.0
transformers==4.46.0
accelerate==1.1.1
safetensors==0.4.4
rembg==2.0.65

@ -0,0 +1,146 @@
#!/usr/bin/env python3
"""Fetch only pinned public code; competition ZIPs are never uploaded by this tool."""
from __future__ import annotations
import argparse
import hashlib
import subprocess
import sys
import urllib.request
from pathlib import Path
SAM2_REPOSITORY = "https://github.com/facebookresearch/sam2.git"
SAM2_COMMIT = "2b90b9f5ceec907a1c18123530e92e794ad901a4"
HUNYUAN_REPOSITORY = "https://github.com/Tencent-Hunyuan/Hunyuan3D-2.1.git"
HUNYUAN_COMMIT = "82920d643c0dc2f7bfd7255f45f62d386edfe60c"
REALESRGAN_URL = "https://github.com/xinntao/Real-ESRGAN/releases/download/v0.1.0/RealESRGAN_x4plus.pth"
REALESRGAN_SHA256 = "4fa0d38905f75ac06eb49a7951b426670021be3018265fd191d2125df9d682f1"
# The upstream all-in-one requirements file pins NumPy 1.24.4. That release
# has no Python 3.12 wheel, so pip attempts a source build in current hosted
# environments and fails before model setup. This is the runtime subset used
# by the Shape/Paint pipeline, not its optional Blender, web-demo, Open3D,
# MeshLab, or distributed-training extras. NumPy <2 preserves the API the
# pinned 2025 Hunyuan source expects while supporting Python 3.10--3.12.
HUNYUAN_RUNTIME_REQUIREMENTS = """\
ninja==1.11.1.1
pybind11==2.13.4
transformers==4.46.0
diffusers==0.30.0
accelerate==1.1.1
pytorch-lightning==1.9.5
huggingface-hub==0.30.2
safetensors==0.4.4
numpy>=1.26,<2
scipy==1.14.1
einops==0.8.0
pandas==2.2.2
opencv-python==4.10.0.84
imageio==2.36.0
scikit-image==0.24.0
rembg==2.0.65
realesrgan==0.3.0
basicsr==1.4.2
trimesh==4.4.7
# Hunyuan Shape imports pymeshlab at module import time. Its upstream
# 2022 pin stops at Python 3.11, so use the closest release with CPython 3.12
# Linux wheels for the hosted runtime.
pymeshlab==2023.12.post1
pygltflib==1.16.3
xatlas==0.0.9
omegaconf==2.3.0
pyyaml==6.0.2
configargparse==1.7
cupy-cuda12x==13.4.1
# 1.16.3 publishes no Python 3.12 wheel. The compatible range retains the
# CPU inference API used by rembg while allowing the hosted Python 3.12 image.
onnxruntime>=1.17,<1.22
torchmetrics==1.6.0
pydantic==2.10.6
timm
torchdiffeq
"""
def run(command: list[str], cwd: Path | None = None) -> None:
print("+", " ".join(command), flush=True)
subprocess.run(command, cwd=cwd, check=True)
def clone_at(repository: str, commit: str, destination: Path) -> None:
if destination.exists():
raise FileExistsError(f"refusing to alter existing third-party checkout: {destination}")
destination.parent.mkdir(parents=True, exist_ok=True)
run(["git", "clone", repository, str(destination)])
run(["git", "checkout", "--detach", commit], destination)
actual = subprocess.check_output(["git", "rev-parse", "HEAD"], cwd=destination, text=True).strip()
if actual != commit:
raise RuntimeError(f"pinned checkout mismatch: expected {commit}, got {actual}")
def sha256(path: Path) -> str:
digest = hashlib.sha256()
with path.open("rb") as handle:
for block in iter(lambda: handle.read(1024 * 1024), b""):
digest.update(block)
return digest.hexdigest()
def download_realesrgan(destination: Path) -> None:
if destination.is_file():
actual = sha256(destination)
if actual == REALESRGAN_SHA256:
return
raise RuntimeError(f"unexpected RealESRGAN checkpoint hash: {actual}; remove {destination} before retrying")
destination.parent.mkdir(parents=True, exist_ok=True)
temporary = destination.with_suffix(destination.suffix + ".partial")
try:
print(f"+ downloading {REALESRGAN_URL}", flush=True)
urllib.request.urlretrieve(REALESRGAN_URL, temporary)
actual = sha256(temporary)
if actual != REALESRGAN_SHA256:
raise RuntimeError(f"RealESRGAN checkpoint hash mismatch: expected {REALESRGAN_SHA256}, got {actual}")
temporary.replace(destination)
finally:
if temporary.exists():
temporary.unlink()
def main() -> None:
parser = argparse.ArgumentParser()
parser.add_argument("--root", type=Path, default=Path("third_party"))
parser.add_argument("--sam2", action="store_true", help="install the pinned SAM2 automatic-segmentation source")
parser.add_argument("--hunyuan", action="store_true", help="install the pinned Hunyuan3D-2.1 source")
parser.add_argument(
"--skip-hunyuan-paint",
action="store_true",
help="install Shape only; do not compile the CUDA-only Hunyuan Paint rasterizer",
)
parser.add_argument("--accept-hunyuan-license", action="store_true", help="required acknowledgement before Hunyuan source is fetched")
args = parser.parse_args()
if not args.sam2 and not args.hunyuan:
parser.error("select at least one model: --sam2 and/or --hunyuan")
if args.hunyuan and not args.accept_hunyuan_license:
parser.error("Hunyuan3D-2.1 has its own licence; inspect it and pass --accept-hunyuan-license only if it permits your use.")
if args.sam2:
sam2 = args.root / "sam2"
clone_at(SAM2_REPOSITORY, SAM2_COMMIT, sam2)
run([sys.executable, "-m", "pip", "install", "-e", "."], sam2)
if args.hunyuan:
hunyuan = args.root / "Hunyuan3D-2.1"
clone_at(HUNYUAN_REPOSITORY, HUNYUAN_COMMIT, hunyuan)
compatible_requirements = hunyuan / "asset_baseline_runtime_requirements.txt"
compatible_requirements.write_text(HUNYUAN_RUNTIME_REQUIREMENTS, encoding="utf-8")
run([sys.executable, "-m", "pip", "install", "-r", str(compatible_requirements)], hunyuan)
if args.skip_hunyuan_paint:
print("+ Hunyuan Paint build skipped: Shape-only profile requested", flush=True)
else:
run([sys.executable, "-m", "pip", "install", "-e", "hy3dpaint/custom_rasterizer"], hunyuan)
run(["bash", "compile_mesh_painter.sh"], hunyuan / "hy3dpaint" / "DifferentiableRenderer")
download_realesrgan(hunyuan / "hy3dpaint" / "ckpt" / "RealESRGAN_x4plus.pth")
if __name__ == "__main__":
main()

@ -0,0 +1,39 @@
#!/usr/bin/env python3
"""Create and fully validate the locked candidate without GPU inference."""
from __future__ import annotations
import argparse
import json
from pathlib import Path
import rebuild_submission
import validate_submission
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--output-root", type=Path, required=True)
args = parser.parse_args()
output = args.output_root.resolve()
if output.exists():
raise FileExistsError(f"choose a new output directory: {output}")
output.mkdir(parents=True)
root = Path(__file__).resolve().parents[1]
package = output / "submission.zip"
rebuild = rebuild_submission.build(
root / "artifacts/intermediate/base_submission.zip",
root / "artifacts/intermediate/donor_submission.zip",
package,
)
validation = validate_submission.validate_package(package)
(output / "rebuild_report.json").write_text(json.dumps(rebuild, ensure_ascii=False, indent=2, sort_keys=True) + "\n", encoding="utf-8")
(output / "validation_report.json").write_text(json.dumps(validation, ensure_ascii=False, indent=2, sort_keys=True) + "\n", encoding="utf-8")
if not validation["valid"]:
raise RuntimeError("candidate failed the non-GPU validation gate")
print(json.dumps({"package": "submission.zip", "sha256": rebuild_submission.FINAL_SHA256, "valid_tasks": validation["valid_tasks"]}, ensure_ascii=False))
return 0
if __name__ == "__main__":
raise SystemExit(main())

@ -0,0 +1,173 @@
#!/usr/bin/env python3
"""Build the locked submission from the two bundled intermediate archives."""
from __future__ import annotations
import argparse
import hashlib
import json
import os
import zipfile
from pathlib import Path, PurePosixPath
from typing import Any
ROOT = Path(__file__).resolve().parents[1]
ARCHIVE_ROOT = "submission_example/submission"
BASE_SHA256 = "dc56dfdf29e81f4e0b4fd300b8e84acea3535a486dc6d26051123252209f3ada"
DONOR_SHA256 = "573ac2792f6df5650ee9b47c37dec052f68065c74a2133e6f4357fa9dd63dadf"
FINAL_SHA256 = "2e979d08d10a785e0c47a4a1ba923131a52c8e53970bb0578f815893585e8d4b"
FINAL_BYTES = 19_539_975
FINAL_MEMBERS = 56
REPLACED_MEMBERS = tuple(
f"{ARCHIVE_ROOT}/{task_id}/{task_id}.usd" for task_id in ("item_029", "item_030")
)
def sha256(path: Path) -> str:
digest = hashlib.sha256()
with path.open("rb") as stream:
for block in iter(lambda: stream.read(1024 * 1024), b""):
digest.update(block)
return digest.hexdigest()
def _check_member_name(name: str) -> None:
value = PurePosixPath(name)
if (
not name
or value.is_absolute()
or ".." in value.parts
or "__MACOSX" in value.parts
or any(part.startswith("._") for part in value.parts)
):
raise ValueError(f"unsafe ZIP member: {name!r}")
def read_members(path: Path) -> dict[str, bytes]:
if not path.is_file():
raise FileNotFoundError(path)
result: dict[str, bytes] = {}
with zipfile.ZipFile(path) as archive:
corrupt = archive.testzip()
if corrupt is not None:
raise ValueError(f"CRC failure in {path.name}: {corrupt}")
for info in archive.infolist():
if info.is_dir():
continue
_check_member_name(info.filename)
if info.filename in result:
raise ValueError(f"duplicate ZIP member: {info.filename}")
result[info.filename] = archive.read(info)
return result
def require_hash(path: Path, expected: str) -> None:
actual = sha256(path)
if actual != expected:
raise ValueError(f"input hash drift for {path.name}: expected {expected}, got {actual}")
def validate_layout(members: dict[str, bytes]) -> None:
if len(members) != FINAL_MEMBERS:
raise ValueError(f"member count is {len(members)}, expected {FINAL_MEMBERS}")
expected_tasks = {f"item_{index:03d}" for index in range(1, 35)}
observed_tasks = {
PurePosixPath(name).parts[2]
for name in members
if len(PurePosixPath(name).parts) >= 4
and PurePosixPath(name).parts[:2] == ("submission_example", "submission")
}
if observed_tasks != expected_tasks:
raise ValueError(f"task mismatch: missing={sorted(expected_tasks-observed_tasks)}, extra={sorted(observed_tasks-expected_tasks)}")
for task_id in expected_tasks:
member = f"{ARCHIVE_ROOT}/{task_id}/{task_id}.usd"
payload = members.get(member)
if payload is None or not payload.startswith(b"PXR-USDC"):
raise ValueError(f"missing or non-USDC task asset: {member}")
def compose(base_path: Path, donor_path: Path) -> tuple[dict[str, bytes], dict[str, str]]:
require_hash(base_path, BASE_SHA256)
require_hash(donor_path, DONOR_SHA256)
members = read_members(base_path)
donor = read_members(donor_path)
origins = {name: "base" for name in members}
for name in REPLACED_MEMBERS:
if name not in members or name not in donor:
raise ValueError(f"missing replacement member: {name}")
members[name] = donor[name]
origins[name] = "donor"
validate_layout(members)
return members, origins
def _zip_info(name: str) -> zipfile.ZipInfo:
info = zipfile.ZipInfo(name, date_time=(1980, 1, 1, 0, 0, 0))
info.compress_type = zipfile.ZIP_DEFLATED
info.create_system = 3
info.external_attr = 0o100644 << 16
return info
def write_package(members: dict[str, bytes], output: Path) -> None:
if output.exists():
raise FileExistsError(f"refusing to overwrite {output}")
output.parent.mkdir(parents=True, exist_ok=True)
temporary = output.with_suffix(output.suffix + ".part")
try:
with zipfile.ZipFile(temporary, "w", compression=zipfile.ZIP_DEFLATED, compresslevel=9) as archive:
for name in sorted(members):
archive.writestr(_zip_info(name), members[name], compress_type=zipfile.ZIP_DEFLATED, compresslevel=9)
os.replace(temporary, output)
finally:
if temporary.exists():
temporary.unlink()
def build(base_path: Path, donor_path: Path, output: Path) -> dict[str, Any]:
members, origins = compose(base_path, donor_path)
write_package(members, output)
if sha256(output) != FINAL_SHA256 or output.stat().st_size != FINAL_BYTES:
raise RuntimeError("built package is not byte-identical to the locked candidate")
if zipfile.ZipFile(output).testzip() is not None:
raise RuntimeError("built package has a CRC failure")
return {
"schema_version": 1,
"method": "base_all_members_plus_donor_two_usd_members",
"inputs": {
"base": {"sha256": BASE_SHA256, "members": len(read_members(base_path))},
"donor": {"sha256": DONOR_SHA256, "members": len(read_members(donor_path))},
},
"replaced_members": list(REPLACED_MEMBERS),
"member_lineage": [
{
"name": name,
"origin": origins[name],
"bytes": len(members[name]),
"sha256": hashlib.sha256(members[name]).hexdigest(),
}
for name in sorted(members)
],
"output": {"sha256": FINAL_SHA256, "bytes": FINAL_BYTES, "members": FINAL_MEMBERS},
}
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--base", type=Path, default=ROOT / "artifacts/intermediate/base_submission.zip")
parser.add_argument("--donor", type=Path, default=ROOT / "artifacts/intermediate/donor_submission.zip")
parser.add_argument("--output", type=Path, required=True)
parser.add_argument("--report", type=Path)
args = parser.parse_args()
report = build(args.base.resolve(), args.donor.resolve(), args.output.resolve())
report_path = args.report.resolve() if args.report else args.output.resolve().with_name("rebuild_report.json")
if report_path.exists():
raise FileExistsError(f"refusing to overwrite {report_path}")
report_path.write_text(json.dumps(report, ensure_ascii=False, indent=2, sort_keys=True) + "\n", encoding="utf-8")
print(json.dumps(report, ensure_ascii=False, sort_keys=True))
return 0
if __name__ == "__main__":
raise SystemExit(main())

@ -0,0 +1,42 @@
#!/usr/bin/env bash
set -euo pipefail
# Usage: DATA_DIR=/absolute/path/to/official_zips OUT_DIR=/absolute/path/to/output ./scripts/run_aliyun.sh
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
PROJECT_DIR="$(cd "$SCRIPT_DIR/.." && pwd)"
: "${DATA_DIR:?set DATA_DIR to a directory containing question.zip and submission_example.zip}"
: "${OUT_DIR:?set OUT_DIR to a writable output directory}"
: "${ACCEPT_HUNYUAN_LICENSE:?read Tencent-Hunyuan/Hunyuan3D-2.1 LICENSE and set ACCEPT_HUNYUAN_LICENSE=yes if your use is allowed}"
if [[ "$ACCEPT_HUNYUAN_LICENSE" != "yes" ]]; then
echo "ACCEPT_HUNYUAN_LICENSE must be exactly yes" >&2
exit 2
fi
if [[ ! -f "$DATA_DIR/question.zip" || ! -f "$DATA_DIR/submission_example.zip" ]]; then
echo "DATA_DIR must contain both official ZIP inputs" >&2
exit 2
fi
mkdir -p "$OUT_DIR"
if [[ -e "$OUT_DIR/gpu_asset_run" || -e "$OUT_DIR/asset_baseline_submission.zip" ]]; then
echo "OUT_DIR already contains a GPU run or submission; use a new empty output directory" >&2
exit 2
fi
(
cd "$PROJECT_DIR"
bash infra/verify_gpu_host.sh
)
mkdir -p "$PROJECT_DIR/third_party"
docker build -t public-asset-baseline:latest "$PROJECT_DIR"
docker run --rm --gpus all \
-v "$DATA_DIR:/data:ro" \
-v "$OUT_DIR:/output" \
-v "$PROJECT_DIR/third_party:/opt/public_asset_baseline/third_party" \
--entrypoint python3.10 public-asset-baseline:latest scripts/run_aliyun_entrypoint.py \
--question /data/question.zip \
--submission-example /data/submission_example.zip \
--output-root /output \
--run-name gpu_asset_run \
--accept-hunyuan-license

@ -0,0 +1,125 @@
#!/usr/bin/env python3
"""Run the GPU pipeline in the released Linux container with a durable receipt."""
from __future__ import annotations
import argparse
import hashlib
import json
import platform
import shutil
import subprocess
import sys
from datetime import datetime, timezone
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
def utc_now() -> str:
return datetime.now(timezone.utc).isoformat()
def sha256(path: Path) -> str:
digest = hashlib.sha256()
with path.open("rb") as handle:
for chunk in iter(lambda: handle.read(1024 * 1024), b""):
digest.update(chunk)
return digest.hexdigest()
def source_manifest() -> list[dict[str, object]]:
"""Hash code/config only; models and competition inputs are excluded."""
selected = (ROOT / "pyproject.toml", ROOT / "requirements-gpu.txt", ROOT / "config", ROOT / "src", ROOT / "scripts")
files: list[Path] = []
for item in selected:
files.extend([item] if item.is_file() else sorted(path for path in item.rglob("*") if path.is_file()))
return [{"path": path.relative_to(ROOT).as_posix(), "bytes": path.stat().st_size, "sha256": sha256(path)} for path in files]
def write_json(path: Path, payload: object) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True) + "\n", encoding="utf-8")
def append_jsonl(path: Path, payload: object) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
with path.open("a", encoding="utf-8") as handle:
handle.write(json.dumps(payload, ensure_ascii=False, sort_keys=True) + "\n")
def main() -> None:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--question", type=Path, required=True)
parser.add_argument("--submission-example", type=Path, required=True)
parser.add_argument("--output-root", type=Path, required=True)
parser.add_argument("--run-name", default="asset_baseline_gpu")
parser.add_argument("--accept-hunyuan-license", action="store_true")
args = parser.parse_args()
if not args.accept_hunyuan_license:
parser.error("read the Hunyuan3D-2.1 LICENSE before passing --accept-hunyuan-license")
question, example = args.question.resolve(), args.submission_example.resolve()
if not question.is_file() or not example.is_file():
parser.error("both official ZIP inputs must exist")
output_root = args.output_root.resolve()
run_root = output_root / args.run_name
process = run_root / "process"
commands = process / "commands.jsonl"
def record(stage: str, **extra: object) -> None:
write_json(process / "experiment_log.json", {"timestamp": utc_now(), "stage": stage, **extra})
def run(command: list[str]) -> None:
append_jsonl(commands, {"timestamp": utc_now(), "command": command})
print("+", " ".join(command), flush=True)
subprocess.run(command, cwd=ROOT, check=True)
try:
write_json(process / "build_manifest.json", {
"schema_version": 1,
"source_files": source_manifest(),
"license_acknowledged": True,
"inputs": ["question.zip", "submission_example.zip"],
})
runtime = {"python": sys.version, "platform": platform.platform()}
try:
import torch
runtime.update({"torch": torch.__version__, "cuda": torch.version.cuda, "cuda_available": torch.cuda.is_available()})
if torch.cuda.is_available():
runtime["gpu"] = torch.cuda.get_device_name(0)
except Exception as error:
runtime["torch_probe_error"] = type(error).__name__
write_json(process / "runtime.json", runtime)
record("started", backend="hunyuan", load_mode="resident")
third_party = ROOT / "third_party"
if not (third_party / "sam2").is_dir() or not (third_party / "Hunyuan3D-2.1").is_dir():
run([sys.executable, "scripts/bootstrap_models.py", "--root", str(third_party), "--sam2", "--hunyuan", "--accept-hunyuan-license"])
run([
"asset-baseline", "--config", "config/default.yaml",
"--set", f"question_zip={question}",
"--set", f"submission_example_zip={example}",
"--set", f"output_root={output_root}",
"--set", f"run_name={args.run_name}",
"run",
])
package = run_root / "packages" / "asset_baseline_submission.zip"
validation = json.loads((run_root / "reports" / "validation.json").read_text(encoding="utf-8"))
package_manifest = json.loads((run_root / "manifests" / "package.json").read_text(encoding="utf-8"))
shutil.copy2(package, output_root / "asset_baseline_submission.zip")
write_json(process / "metrics.json", {
"valid": validation["valid"],
"valid_tasks": validation["valid_tasks"],
"package_sha256": package_manifest["sha256"],
"package_bytes": package_manifest["bytes"],
})
record("completed", valid=validation["valid"], valid_tasks=validation["valid_tasks"], package_sha256=package_manifest["sha256"])
except BaseException as error:
record("failed", error_type=type(error).__name__, error=str(error))
raise
if __name__ == "__main__":
main()

@ -0,0 +1,141 @@
#!/usr/bin/env python3
"""Perform deterministic ZIP, OpenUSD and lightweight physics checks without GPU."""
from __future__ import annotations
import argparse
import json
import math
import tempfile
from pathlib import Path
from typing import Any
import numpy as np
import mujoco
from pxr import Sdf, Usd, UsdGeom, UsdPhysics, UsdShade
import rebuild_submission
def _extract(members: dict[str, bytes], root: Path) -> Path:
for name, payload in members.items():
target = (root / name).resolve()
if root.resolve() not in target.parents:
raise ValueError(f"unsafe extraction target: {name}")
target.parent.mkdir(parents=True, exist_ok=True)
target.write_bytes(payload)
return root / rebuild_submission.ARCHIVE_ROOT
def _physics_probe(points: np.ndarray) -> dict[str, float]:
lower, upper = points.min(axis=0), points.max(axis=0)
size = np.maximum((upper - lower) / 2.0, 0.015)
xml = f'''<mujoco model="asset"><option timestep="0.002" gravity="0 0 -9.81"/><worldbody><body pos="0 0 1"><freejoint/><geom type="box" size="{size[0]:.7f} {size[1]:.7f} {size[2]:.7f}" mass="1"/></body></worldbody></mujoco>'''
model = mujoco.MjModel.from_xml_string(xml)
data = mujoco.MjData(model)
for _ in range(120):
mujoco.mj_step(model, data)
if not np.isfinite(data.qpos).all() or not np.isfinite(data.qvel).all():
raise ValueError("non-finite lightweight physics state")
return {"steps": 120, "z": float(data.qpos[2]), "vertical_velocity": float(data.qvel[2])}
def _validate_asset(task_id: str, asset: Path) -> dict[str, Any]:
errors: list[str] = []
counts = {"gprims": 0, "colliders": 0, "rigid_bodies": 0, "physics_scenes": 0, "textures": 0}
points: list[tuple[float, float, float]] = []
stage = Usd.Stage.Open(str(asset))
if stage is None:
return {"task_id": task_id, "valid": False, "errors": ["Usd.Stage.Open returned None"], "counts": counts}
if stage.GetDefaultPrim().GetPath() != Sdf.Path("/World"):
errors.append("default prim is not /World")
if UsdGeom.GetStageUpAxis(stage) != UsdGeom.Tokens.z:
errors.append("up axis is not Z")
if abs(float(UsdGeom.GetStageMetersPerUnit(stage)) - 1.0) > 1e-9:
errors.append("meters per unit is not 1")
for prim in stage.Traverse():
counts["gprims"] += int(prim.IsA(UsdGeom.Gprim))
counts["colliders"] += int(prim.HasAPI(UsdPhysics.CollisionAPI))
counts["rigid_bodies"] += int(prim.HasAPI(UsdPhysics.RigidBodyAPI))
counts["physics_scenes"] += int(prim.IsA(UsdPhysics.Scene))
if prim.IsA(UsdGeom.Gprim):
bound = UsdGeom.Boundable(prim).ComputeWorldBound(
Usd.TimeCode.Default(), UsdGeom.Tokens.default_
).ComputeAlignedRange()
if not bound.IsEmpty():
lower, upper = bound.GetMin(), bound.GetMax()
points.extend(
[
(float(lower[0]), float(lower[1]), float(lower[2])),
(float(upper[0]), float(upper[1]), float(upper[2])),
]
)
if prim.IsA(UsdShade.Shader):
shader = UsdShade.Shader(prim)
file_input = shader.GetInput("file")
value = file_input.Get() if file_input else None
if isinstance(value, Sdf.AssetPath) and value.path:
counts["textures"] += 1
if Path(value.path).is_absolute() or value.path.startswith(("http://", "https://")):
errors.append(f"non-local texture: {value.path}")
elif not (asset.parent / value.path).is_file():
errors.append(f"missing texture: {value.path}")
if not counts["gprims"] or not points:
errors.append("no renderable geometry")
if counts["physics_scenes"] != 1:
errors.append(f"expected one physics scene, found {counts['physics_scenes']}")
if not counts["colliders"]:
errors.append("missing collision API")
physics = None
if not errors and counts["rigid_bodies"]:
try:
physics = _physics_probe(np.asarray(points, dtype=np.float64))
except Exception as error:
errors.append(f"physics probe: {type(error).__name__}: {error}")
elif not errors:
physics = {"mode": "static_scene", "steps": 0}
return {"task_id": task_id, "valid": not errors, "errors": errors, "counts": counts, "physics_probe": physics}
def validate_package(package: Path) -> dict[str, Any]:
if rebuild_submission.sha256(package) != rebuild_submission.FINAL_SHA256:
raise ValueError("candidate hash is not the locked submission hash")
members = rebuild_submission.read_members(package)
rebuild_submission.validate_layout(members)
with tempfile.TemporaryDirectory(prefix="asset-baseline-validation-") as raw:
submission = _extract(members, Path(raw))
tasks = [
_validate_asset(task_id, submission / task_id / f"{task_id}.usd")
for task_id in (f"item_{index:03d}" for index in range(1, 35))
]
errors = [f"{task['task_id']}: {message}" for task in tasks for message in task["errors"]]
return {
"schema_version": 1,
"package_sha256": rebuild_submission.FINAL_SHA256,
"package_bytes": package.stat().st_size,
"zip_crc": "pass",
"task_count": len(tasks),
"valid_tasks": sum(task["valid"] for task in tasks),
"physics_probe_steps": 120,
"valid": not errors,
"errors": errors,
"tasks": tasks,
}
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--package", type=Path, required=True)
parser.add_argument("--report", type=Path, required=True)
args = parser.parse_args()
if args.report.exists():
raise FileExistsError(f"refusing to overwrite {args.report}")
report = validate_package(args.package.resolve())
args.report.parent.mkdir(parents=True, exist_ok=True)
args.report.write_text(json.dumps(report, ensure_ascii=False, indent=2, sort_keys=True) + "\n", encoding="utf-8")
print(json.dumps({key: value for key, value in report.items() if key != "tasks"}, ensure_ascii=False, sort_keys=True))
return 0 if report["valid"] else 1
if __name__ == "__main__":
raise SystemExit(main())

@ -0,0 +1,3 @@
"""Raw-video to simulation-ready USD generation from the two official ZIPs."""
__version__ = "0.1.0"

@ -0,0 +1,40 @@
from __future__ import annotations
import argparse
import json
from pathlib import Path
from .config import load_config, resolve_paths
from .pipeline import generate, inspect, package, prepare, run_all, validate
def _config(args: argparse.Namespace) -> dict:
return resolve_paths(load_config(Path(args.config), args.set or []), Path(args.config))
def main() -> None:
parser = argparse.ArgumentParser(description="Zero-history raw-video embodied-asset baseline")
parser.add_argument("--config", default="config/default.yaml", help="YAML configuration")
parser.add_argument("--set", action="append", default=[], help="strict dotted configuration override, e.g. reconstruction.backend=primitive")
subparsers = parser.add_subparsers(dest="command", required=True)
for command in ("inspect", "prepare", "generate", "package", "validate", "run"):
subparsers.add_parser(command)
args = parser.parse_args()
config = _config(args)
if args.command == "inspect":
result = inspect(config)
elif args.command == "prepare":
result = {"run_root": str(prepare(config))}
elif args.command == "generate":
result = {"run_root": str(generate(config))}
elif args.command == "package":
result = {"package": str(package(config))}
elif args.command == "validate":
result = validate(config)
else:
result = run_all(config)
print(json.dumps(result, ensure_ascii=False, indent=2))
if __name__ == "__main__":
main()

@ -0,0 +1,40 @@
from __future__ import annotations
from copy import deepcopy
from pathlib import Path
from typing import Any
import yaml
def load_config(path: Path, overrides: list[str]) -> dict[str, Any]:
"""Load YAML and apply strict dotted `key=value` CLI overrides."""
raw = yaml.safe_load(path.read_text(encoding="utf-8"))
if not isinstance(raw, dict):
raise ValueError(f"configuration must be a mapping: {path}")
config = deepcopy(raw)
for assignment in overrides:
if "=" not in assignment:
raise ValueError(f"override must be key=value: {assignment}")
dotted, value = assignment.split("=", 1)
keys = dotted.split(".")
target: dict[str, Any] = config
for key in keys[:-1]:
child = target.get(key)
if not isinstance(child, dict):
raise KeyError(f"unknown configuration key: {dotted}")
target = child
if keys[-1] not in target:
raise KeyError(f"unknown configuration key: {dotted}")
target[keys[-1]] = yaml.safe_load(value)
return config
def resolve_paths(config: dict[str, Any], config_path: Path) -> dict[str, Any]:
"""Resolve input/output paths relative to the working directory, not this repo."""
result = deepcopy(config)
base = Path.cwd()
for key in ("question_zip", "submission_example_zip", "output_root"):
value = Path(str(result[key]))
result[key] = str((base / value).resolve() if not value.is_absolute() else value.resolve())
return result

@ -0,0 +1,96 @@
from __future__ import annotations
import re
import zipfile
from pathlib import Path, PurePosixPath
from typing import Any
from .util import sha256, utc_now, write_json
TASK_PATTERN = re.compile(r"^item_(\d{3})$")
VIDEO_SUFFIXES = {".mp4", ".mov", ".m4v", ".avi", ".MP4", ".MOV", ".M4V", ".AVI"}
def _safe_members(archive: zipfile.ZipFile) -> list[zipfile.ZipInfo]:
members: list[zipfile.ZipInfo] = []
for info in archive.infolist():
name = PurePosixPath(info.filename)
if name.is_absolute() or ".." in name.parts:
raise ValueError(f"unsafe ZIP member: {info.filename}")
if not info.is_dir() and not info.filename.startswith("__MACOSX/"):
members.append(info)
if archive.testzip() is not None:
raise ValueError(f"ZIP CRC failure: {archive.filename}")
return members
def inspect_inputs(question_zip: Path, example_zip: Path) -> dict[str, Any]:
if not question_zip.is_file():
raise FileNotFoundError(question_zip)
if not example_zip.is_file():
raise FileNotFoundError(example_zip)
with zipfile.ZipFile(question_zip) as archive:
question_members = _safe_members(archive)
with zipfile.ZipFile(example_zip) as archive:
example_members = _safe_members(archive)
tasks: dict[str, list[str]] = {}
for info in question_members:
path = PurePosixPath(info.filename)
if len(path.parts) != 2 or path.suffix not in VIDEO_SUFFIXES:
continue
match = TASK_PATTERN.fullmatch(path.parts[0])
if match is None:
continue
tasks.setdefault(path.parts[0], []).append(info.filename)
expected = [f"item_{index:03d}" for index in range(1, 35)]
missing = sorted(set(expected) - set(tasks))
extras = sorted(set(tasks) - set(expected))
if missing or extras:
raise ValueError(f"question archive task inventory mismatch: missing={missing}, extras={extras}")
example_prefix = "submission_example/submission/"
if not any(info.filename.startswith(example_prefix) for info in example_members):
raise ValueError("submission_example.zip lacks submission_example/submission/")
return {
"created_at": utc_now(),
"input_contract": "official_question_zip_plus_official_submission_example_zip_only",
"question_zip": {"path": str(question_zip), "sha256": sha256(question_zip), "bytes": question_zip.stat().st_size},
"submission_example_zip": {"path": str(example_zip), "sha256": sha256(example_zip), "bytes": example_zip.stat().st_size},
"task_count": len(tasks),
"video_count": sum(len(value) for value in tasks.values()),
"tasks": [{"task_id": task_id, "videos": sorted(tasks[task_id])} for task_id in expected],
"example_member_count": len(example_members),
}
def extract_question(question_zip: Path, manifest: dict[str, Any], output_root: Path) -> dict[str, Any]:
"""Extract only verified official videos under a new run directory."""
output_root.mkdir(parents=True, exist_ok=True)
wanted = {video for task in manifest["tasks"] for video in task["videos"]}
records: list[dict[str, Any]] = []
with zipfile.ZipFile(question_zip) as archive:
infos = {info.filename: info for info in _safe_members(archive)}
if wanted - set(infos):
raise ValueError(f"question archive changed after inspection: {sorted(wanted - set(infos))[:3]}")
for task in manifest["tasks"]:
task_id = task["task_id"]
for member in task["videos"]:
target = output_root / member
target.parent.mkdir(parents=True, exist_ok=True)
if target.exists():
raise FileExistsError(target)
with archive.open(infos[member]) as source, target.open("xb") as destination:
while True:
block = source.read(1024 * 1024)
if not block:
break
destination.write(block)
records.append({"task_id": task_id, "member": member, "path": str(target), "bytes": target.stat().st_size, "sha256": sha256(target)})
return {"created_at": utc_now(), "video_root": str(output_root), "videos": records}
def write_input_manifest(question_zip: Path, example_zip: Path, destination: Path) -> dict[str, Any]:
manifest = inspect_inputs(question_zip, example_zip)
write_json(destination, manifest)
return manifest

@ -0,0 +1,128 @@
from __future__ import annotations
import math
from pathlib import Path
from typing import Any
import cv2
import numpy as np
from PIL import Image, ImageDraw
from .util import utc_now
def _resize(frame: np.ndarray, max_edge: int) -> np.ndarray:
height, width = frame.shape[:2]
scale = min(1.0, max_edge / max(height, width))
if scale >= 1.0:
return frame
return cv2.resize(frame, (round(width * scale), round(height * scale)), interpolation=cv2.INTER_AREA)
def _frame_score(frame: np.ndarray) -> dict[str, float]:
gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
sharpness = float(cv2.Laplacian(gray, cv2.CV_64F).var())
contrast = float(gray.std())
height, width = gray.shape
central = gray[height // 4 : 3 * height // 4, width // 4 : 3 * width // 4]
central_contrast = float(central.std())
return {"sharpness": sharpness, "contrast": contrast, "central_contrast": central_contrast}
def _candidate_indices(frame_count: int, samples: int) -> list[int]:
if frame_count <= 1:
return [0]
return sorted({round(index * (frame_count - 1) / max(samples - 1, 1)) for index in range(samples)})
def _colour_distance(left: np.ndarray, right: np.ndarray) -> float:
return float(np.linalg.norm(left.astype(np.float64) - right.astype(np.float64)))
def _make_contact_sheet(records: list[dict[str, Any]], target: Path) -> None:
thumbs: list[Image.Image] = []
for record in records:
image = Image.open(record["path"]).convert("RGB")
image.thumbnail((240, 180))
canvas = Image.new("RGB", (240, 204), "white")
canvas.paste(image, ((240 - image.width) // 2, 0))
ImageDraw.Draw(canvas).text((6, 184), f"{record['source_index']}:{record['frame_index']}", fill="black")
thumbs.append(canvas)
columns = 3
rows = max(1, math.ceil(len(thumbs) / columns))
sheet = Image.new("RGB", (columns * 240, rows * 204), "white")
for index, thumb in enumerate(thumbs):
sheet.paste(thumb, ((index % columns) * 240, (index // columns) * 204))
target.parent.mkdir(parents=True, exist_ok=True)
sheet.save(target)
def extract_and_select_views(
task_id: str,
videos: list[dict[str, Any]],
output_dir: Path,
*,
frames_per_video: int,
selected_views: int,
max_edge: int,
jpeg_quality: int,
) -> dict[str, Any]:
"""Uniformly sample all clips then select sharp, visually diverse views."""
candidates: list[dict[str, Any]] = []
raw_dir = output_dir / "candidates"
raw_dir.mkdir(parents=True, exist_ok=True)
for source_index, video in enumerate(videos):
capture = cv2.VideoCapture(str(video["path"]))
if not capture.isOpened():
raise RuntimeError(f"OpenCV could not decode video: {video['path']}")
frame_count = max(1, int(capture.get(cv2.CAP_PROP_FRAME_COUNT)))
fps = float(capture.get(cv2.CAP_PROP_FPS))
for frame_index in _candidate_indices(frame_count, frames_per_video):
capture.set(cv2.CAP_PROP_POS_FRAMES, frame_index)
success, frame = capture.read()
if not success or frame is None:
continue
frame = _resize(frame, max_edge)
metrics = _frame_score(frame)
colour = frame.reshape(-1, 3).mean(axis=0).tolist()
path = raw_dir / f"s{source_index:02d}_f{frame_index:06d}.jpg"
if not cv2.imwrite(str(path), frame, [cv2.IMWRITE_JPEG_QUALITY, jpeg_quality]):
raise RuntimeError(f"failed to write frame: {path}")
candidates.append({
"path": str(path), "source_index": source_index, "source_video": video["member"],
"frame_index": frame_index, "frame_count": frame_count, "fps": fps,
"mean_bgr": colour, **metrics,
})
capture.release()
if not candidates:
raise RuntimeError(f"no decodable frames for {task_id}")
for key in ("sharpness", "contrast", "central_contrast"):
values = np.asarray([record[key] for record in candidates], dtype=np.float64)
low, high = float(values.min()), float(values.max())
for record in candidates:
record[f"norm_{key}"] = (record[key] - low) / max(high - low, 1e-9)
for record in candidates:
record["base_score"] = sum(record[f"norm_{name}"] for name in ("sharpness", "contrast", "central_contrast"))
ranked = sorted(candidates, key=lambda record: (-record["base_score"], record["source_index"], record["frame_index"]))
chosen: list[dict[str, Any]] = []
for candidate in ranked:
diversity = 1.0 if not chosen else min(_colour_distance(np.asarray(candidate["mean_bgr"]), np.asarray(old["mean_bgr"])) / 255.0 for old in chosen)
if len(chosen) < selected_views and (not chosen or diversity >= 0.04):
candidate["diversity"] = diversity
chosen.append(candidate)
for candidate in ranked:
if len(chosen) >= selected_views:
break
if candidate not in chosen:
candidate["diversity"] = 0.0
chosen.append(candidate)
selected_dir = output_dir / "selected"
selected_dir.mkdir(parents=True, exist_ok=True)
selected: list[dict[str, Any]] = []
for index, record in enumerate(chosen):
target = selected_dir / f"view_{index:02d}.jpg"
image = Image.open(record["path"]).convert("RGB")
image.save(target, quality=jpeg_quality)
selected.append({**record, "path": str(target), "view_index": index})
_make_contact_sheet(selected, output_dir / "contact_sheet.jpg")
return {"task_id": task_id, "created_at": utc_now(), "candidate_count": len(candidates), "selected": selected, "contact_sheet": str(output_dir / "contact_sheet.jpg")}

@ -0,0 +1,181 @@
from __future__ import annotations
import platform
import sys
import zipfile
from pathlib import Path
from typing import Any
from .input_data import extract_question, inspect_inputs, write_input_manifest
from .media import extract_and_select_views
from .reconstruct import HunyuanReconstructor, _load_mesh, create_reconstructor, finalize_mesh, reconstruct_mesh
from .segmentation import make_conditioning_image
from .usd_asset import simulate_mjcf, write_usd_asset
from .util import clean_dir, iter_files, read_json, seed_everything, sha256, utc_now, write_json
def run_root(config: dict[str, Any]) -> Path:
return Path(str(config["output_root"])) / str(config["run_name"])
def runtime_record(config: dict[str, Any]) -> dict[str, Any]:
return {
"created_at": utc_now(),
"input_contract": "question.zip + submission_example.zip only; no work/v* or historic submission input is permitted",
"python": sys.version,
"platform": platform.platform(),
"config": config,
}
def inspect(config: dict[str, Any]) -> dict[str, Any]:
return inspect_inputs(Path(config["question_zip"]), Path(config["submission_example_zip"]))
def prepare(config: dict[str, Any]) -> Path:
root = run_root(config)
clean_dir(root)
seed_everything(int(config["seed"]))
manifest = write_input_manifest(Path(config["question_zip"]), Path(config["submission_example_zip"]), root / "manifests" / "inputs.json")
write_json(root / "manifests" / "runtime.json", runtime_record(config))
extracted = extract_question(Path(config["question_zip"]), manifest, root / "raw_videos")
write_json(root / "manifests" / "extracted.json", extracted)
by_task: dict[str, list[dict[str, Any]]] = {task["task_id"]: [] for task in manifest["tasks"]}
for record in extracted["videos"]:
by_task[record["task_id"]].append(record)
view_records = []
video_config = config["video"]
for task in manifest["tasks"]:
task_id = task["task_id"]
view_records.append(extract_and_select_views(
task_id,
sorted(by_task[task_id], key=lambda record: record["member"]),
root / "views" / task_id,
frames_per_video=int(video_config["frames_per_video"]),
selected_views=int(video_config["selected_views"]),
max_edge=int(video_config["max_edge"]),
jpeg_quality=int(video_config["jpeg_quality"]),
))
write_json(root / "manifests" / "views.json", {"created_at": utc_now(), "tasks": view_records})
return root
def _require_prepared(config: dict[str, Any]) -> tuple[Path, dict[str, Any]]:
root = run_root(config)
manifest_path = root / "manifests" / "views.json"
if not manifest_path.is_file():
raise FileNotFoundError(f"prepared views missing; run `asset_baseline prepare` first: {manifest_path}")
return root, read_json(manifest_path)
def generate(config: dict[str, Any]) -> Path:
root, views = _require_prepared(config)
submission = root / "submission"
if submission.exists():
raise FileExistsError(f"generation already exists; use a new run_name: {submission}")
seed_everything(int(config["seed"]))
assets: list[dict[str, Any]] = []
reconstructor = create_reconstructor(config["reconstruction"])
if isinstance(reconstructor, HunyuanReconstructor) and reconstructor.sequential:
staged: list[dict[str, Any]] = []
# On a 24 GiB GPU, generate every Shape result first while Shape is
# resident, then release it before Paint is loaded.
for task in views["tasks"]:
task_id = task["task_id"]
selected = task["selected"]
if not selected:
raise RuntimeError(f"no selected views: {task_id}")
task_root = root / "generated" / task_id
conditioning_path = task_root / "conditioning.png"
segmentation = make_conditioning_image(Path(selected[0]["path"]), conditioning_path, config["segmentation"])
shape_path = reconstructor.generate_shape(conditioning_path, task_root / "mesh_shape.glb")
staged.append({"task_id": task_id, "selected": selected[0], "task_root": task_root, "conditioning": conditioning_path, "segmentation": segmentation, "shape": shape_path})
reconstructor.begin_texture_phase()
for record in staged:
final_path = reconstructor.generate_texture(
record["shape"],
record["conditioning"],
record["task_root"] / "mesh_textured.obj",
)
mesh = finalize_mesh(
record["task_id"],
_load_mesh(final_path),
record["task_root"] / "mesh.obj",
config["reconstruction"],
"Hunyuan3D-2.1",
)
asset = write_usd_asset(record["task_id"], Path(mesh["mesh"]), record["conditioning"], submission / record["task_id"], config["physics"])
assets.append({"task_id": record["task_id"], "source_view": record["selected"], "segmentation": record["segmentation"], "mesh": mesh, "asset": asset})
write_json(root / "manifests" / "assets.json", {"created_at": utc_now(), "tasks": assets})
return root
for task in views["tasks"]:
task_id = task["task_id"]
selected = task["selected"]
if not selected:
raise RuntimeError(f"no selected views: {task_id}")
task_root = root / "generated" / task_id
conditioning_path = task_root / "conditioning.png"
segmentation = make_conditioning_image(Path(selected[0]["path"]), conditioning_path, config["segmentation"])
mesh = reconstruct_mesh(
task_id,
conditioning_path,
task_root / "mesh.obj",
config["reconstruction"],
reconstructor=reconstructor,
)
asset = write_usd_asset(task_id, Path(mesh["mesh"]), conditioning_path, submission / task_id, config["physics"])
assets.append({"task_id": task_id, "source_view": selected[0], "segmentation": segmentation, "mesh": mesh, "asset": asset})
write_json(root / "manifests" / "assets.json", {"created_at": utc_now(), "tasks": assets})
return root
def _archive_members(submission: Path) -> list[Path]:
expected = [submission / f"item_{index:03d}" / f"item_{index:03d}.usd" for index in range(1, 35)]
missing = [str(path) for path in expected if not path.is_file()]
if missing:
raise ValueError(f"submission has missing task USDs: {missing[:3]}")
return list(iter_files(submission))
def package(config: dict[str, Any]) -> Path:
root, _ = _require_prepared(config)
submission = root / "submission"
members = _archive_members(submission)
package_path = root / "packages" / "asset_baseline_submission.zip"
if package_path.exists():
raise FileExistsError(package_path)
package_path.parent.mkdir(parents=True)
archive_root = str(config["package"]["archive_root"]).rstrip("/")
level = int(config["package"]["compression_level"])
with zipfile.ZipFile(package_path, "w", compression=zipfile.ZIP_DEFLATED, compresslevel=level, strict_timestamps=True) as archive:
for source in members:
relative = source.relative_to(submission).as_posix()
info = zipfile.ZipInfo(f"{archive_root}/{relative}", date_time=(2026, 1, 1, 0, 0, 0))
info.compress_type = zipfile.ZIP_DEFLATED
info.external_attr = 0o100644 << 16
archive.writestr(info, source.read_bytes(), compress_type=zipfile.ZIP_DEFLATED, compresslevel=level)
if zipfile.ZipFile(package_path).testzip() is not None:
raise RuntimeError("generated ZIP CRC failure")
write_json(root / "manifests" / "package.json", {"created_at": utc_now(), "package": str(package_path), "sha256": sha256(package_path), "bytes": package_path.stat().st_size, "file_count": len(members)})
return package_path
def validate(config: dict[str, Any]) -> dict[str, Any]:
root, _ = _require_prepared(config)
submission = root / "submission"
_archive_members(submission)
from .validate import validate_submission_tree
report = validate_submission_tree(submission, root / "physics", int(config["physics"]["simulation_steps"]))
package_path = root / "packages" / "asset_baseline_submission.zip"
if package_path.is_file():
report["package"] = {"path": str(package_path), "sha256": sha256(package_path), "bytes": package_path.stat().st_size}
write_json(root / "reports" / "validation.json", report)
return report
def run_all(config: dict[str, Any]) -> dict[str, Any]:
prepare(config)
generate(config)
package(config)
return validate(config)

@ -0,0 +1,220 @@
from __future__ import annotations
import importlib
import gc
import sys
from pathlib import Path
from typing import Any
import numpy as np
import trimesh
from PIL import Image
def _primitive_mesh(image_path: Path) -> trimesh.Trimesh:
image = Image.open(image_path).convert("RGBA")
width, height = image.size
aspect = max(0.35, min(2.6, width / max(height, 1)))
# A shallow closed cuboid is safer than a textured plane in a physics engine.
return trimesh.creation.box(extents=(aspect, 0.28, 1.0))
def _load_mesh(path: Path) -> trimesh.Trimesh:
loaded = trimesh.load(path, force="scene")
if isinstance(loaded, trimesh.Scene):
meshes = [geometry for geometry in loaded.geometry.values() if isinstance(geometry, trimesh.Trimesh)]
if not meshes:
raise RuntimeError(f"Hunyuan output has no mesh: {path}")
if len(meshes) == 1:
# Do not concatenate a one-mesh textured OBJ: trimesh's generic
# concatenate path can replace TextureVisuals with vertex colours.
mesh = meshes[0].copy()
else:
textured = [
geometry
for geometry in meshes
if getattr(getattr(getattr(geometry, "visual", None), "material", None), "image", None) is not None
]
if textured:
raise RuntimeError(
f"Hunyuan output has {len(meshes)} textured geometries; "
"this baseline writes one USD material and refuses to silently discard texture assignments"
)
mesh = trimesh.util.concatenate(meshes)
elif isinstance(loaded, trimesh.Trimesh):
mesh = loaded
else:
raise RuntimeError(f"unsupported mesh payload: {type(loaded).__name__}")
if len(mesh.faces) == 0:
raise RuntimeError(f"empty mesh: {path}")
return mesh
def _export_result(mesh: Any, output: Path) -> Path:
output.parent.mkdir(parents=True, exist_ok=True)
if isinstance(mesh, (str, Path)):
source = Path(mesh)
if source.resolve() != output.resolve():
output.write_bytes(source.read_bytes())
return output
if hasattr(mesh, "export"):
mesh.export(str(output))
return output
if hasattr(mesh, "save"):
mesh.save(str(output))
return output
raise TypeError(f"cannot export Hunyuan mesh type: {type(mesh).__name__}")
class HunyuanReconstructor:
"""One GPU model session reused across every task in a run.
Loading both Hunyuan Shape and Paint for every one of 34 assets is not a
viable GPU workflow: it needlessly downloads/initializes models, lengthens
the run and can fragment VRAM. This session is made exactly once by the
pipeline and then processes each conditioning image in turn.
"""
def __init__(self, reconstruction: dict[str, Any]) -> None:
self.reconstruction = reconstruction
self.load_mode = str(reconstruction.get("hunyuan_load_mode", "resident"))
if self.load_mode not in {"resident", "sequential"}:
raise ValueError(f"unsupported hunyuan_load_mode: {self.load_mode}")
self.repo = Path(str(reconstruction["hunyuan_repo"])).resolve()
if not self.repo.is_dir():
raise FileNotFoundError(f"Hunyuan3D checkout not found: {self.repo}; run scripts/bootstrap_models.py --hunyuan")
for relative in ("hy3dshape", "hy3dpaint"):
candidate = str(self.repo / relative)
if candidate not in sys.path:
sys.path.insert(0, candidate)
self.shape_pipeline: Any | None = self._load_shape_pipeline()
self.paint_pipeline: Any | None = None
if self.wants_texture and self.load_mode == "resident":
self.paint_pipeline = self._load_paint_pipeline()
@property
def wants_texture(self) -> bool:
return bool(self.reconstruction.get("hunyuan_texture", True))
@property
def sequential(self) -> bool:
return self.load_mode == "sequential"
def _load_shape_pipeline(self) -> Any:
try: # imports are deliberately lazy so CPU-only validation has no model dependency
from hy3dshape.pipelines import Hunyuan3DDiTFlowMatchingPipeline
except ImportError as error: # pragma: no cover - GPU-only integration
raise RuntimeError("Hunyuan3D dependencies are unavailable; follow README section 'GPU install'.") from error
return Hunyuan3DDiTFlowMatchingPipeline.from_pretrained(str(self.reconstruction["hunyuan_model"]))
def _load_paint_pipeline(self) -> Any:
try:
texture_module = importlib.import_module("textureGenPipeline")
paint_config = texture_module.Hunyuan3DPaintConfig(
max_num_view=int(self.reconstruction.get("hunyuan_paint_max_views", 6)),
resolution=int(self.reconstruction.get("hunyuan_paint_resolution", 512)),
)
# Hunyuan's own demo is run from its checkout. The baseline is
# invoked elsewhere, so resolve its local files explicitly.
paint_config.realesrgan_ckpt_path = str(self.repo / "hy3dpaint" / "ckpt" / "RealESRGAN_x4plus.pth")
paint_config.multiview_cfg_path = str(self.repo / "hy3dpaint" / "cfgs" / "hunyuan-paint-pbr.yaml")
paint_config.custom_pipeline = str(self.repo / "hy3dpaint" / "hunyuanpaintpbr")
return texture_module.Hunyuan3DPaintPipeline(paint_config)
except Exception as error: # pragma: no cover - GPU-only integration
raise RuntimeError("Hunyuan texture pipeline could not initialize; do not silently publish an untextured high profile.") from error
def generate_shape(self, image_path: Path, shape_path: Path) -> Path:
if self.shape_pipeline is None:
raise RuntimeError("Hunyuan shape model was released before all shapes were generated")
generated = self.shape_pipeline(image=str(image_path))[0]
return _export_result(generated, shape_path)
def begin_texture_phase(self) -> None:
"""Release Shape before loading Paint on a 24 GiB GPU."""
if not self.sequential or not self.wants_texture:
return
self.shape_pipeline = None
gc.collect()
try: # pragma: no cover - exercised only on a CUDA host
import torch
torch.cuda.empty_cache()
except Exception:
pass
self.paint_pipeline = self._load_paint_pipeline()
def generate_texture(self, shape_path: Path, image_path: Path, output: Path) -> Path:
if not self.wants_texture:
return shape_path
if self.paint_pipeline is None:
raise RuntimeError("Hunyuan Paint is not initialized; call begin_texture_phase() first in sequential mode")
try:
textured = self.paint_pipeline(
mesh_path=str(shape_path),
image_path=str(image_path),
output_mesh_path=str(output),
)
return Path(str(textured))
except Exception as error: # pragma: no cover - GPU-only integration
raise RuntimeError("Hunyuan shape generation succeeded but texture generation failed; do not silently publish an untextured high profile.") from error
def reconstruct(self, image_path: Path, output: Path) -> trimesh.Trimesh:
shape_path = output.with_name(output.stem + "_shape.glb")
self.generate_shape(image_path, shape_path)
final_path = self.generate_texture(shape_path, image_path, output.with_name(output.stem + "_textured.obj"))
return _load_mesh(final_path)
def create_reconstructor(reconstruction: dict[str, Any]) -> HunyuanReconstructor | None:
backend = str(reconstruction["backend"])
if backend == "primitive":
return None
if backend == "hunyuan":
return HunyuanReconstructor(reconstruction)
raise ValueError(f"unknown reconstruction backend: {backend}")
def finalize_mesh(task_id: str, mesh: trimesh.Trimesh, output: Path, reconstruction: dict[str, Any], source: str) -> dict[str, Any]:
"""Normalize, simplify and export a mesh after any reconstruction backend."""
backend = str(reconstruction["backend"])
source_faces = int(len(mesh.faces))
simplification_error: str | None = None
if len(mesh.faces) > int(reconstruction["max_faces"]):
try:
mesh = mesh.simplify_quadric_decimation(int(reconstruction["max_faces"]))
except Exception as error:
# The optional trimesh decimator has platform-specific native
# dependencies. Keep a valid mesh and expose the condition in the
# manifest rather than replacing it with a lower-quality primitive.
simplification_error = f"{type(error).__name__}: {error}"
if len(mesh.faces) == 0:
raise RuntimeError(f"reconstruction produced no faces: {task_id}")
mesh.remove_unreferenced_vertices()
extent = np.asarray(mesh.extents, dtype=np.float64)
scale = 1.0 / max(float(extent.max()), 1e-6)
mesh.apply_scale(scale)
mesh.apply_translation(-mesh.bounds.mean(axis=0))
mesh.apply_translation([0.0, 0.0, -float(mesh.bounds[0, 2])])
output.parent.mkdir(parents=True, exist_ok=True)
mesh.export(str(output))
return {"task_id": task_id, "backend": backend, "source": source, "mesh": str(output), "vertices": int(len(mesh.vertices)), "faces": int(len(mesh.faces)), "source_faces": source_faces, "simplification_error": simplification_error, "extent": [float(value) for value in mesh.extents]}
def reconstruct_mesh(
task_id: str,
conditioning_image: Path,
output: Path,
reconstruction: dict[str, Any],
*,
reconstructor: HunyuanReconstructor | None = None,
) -> dict[str, Any]:
backend = str(reconstruction["backend"])
if backend == "primitive":
mesh = _primitive_mesh(conditioning_image)
source = "deterministic_primitive"
elif backend == "hunyuan":
session = reconstructor if reconstructor is not None else HunyuanReconstructor(reconstruction)
mesh = session.reconstruct(conditioning_image, output)
source = "Hunyuan3D-2.1"
else:
raise ValueError(f"unknown reconstruction backend: {backend}")
return finalize_mesh(task_id, mesh, output, reconstruction, source)

@ -0,0 +1,100 @@
from __future__ import annotations
from pathlib import Path
from typing import Any
import cv2
import numpy as np
from PIL import Image
def _center_mask(image: np.ndarray) -> np.ndarray:
"""Deterministic non-model fallback using GrabCut and a centre prior."""
height, width = image.shape[:2]
mask = np.zeros((height, width), np.uint8)
margin_x, margin_y = max(2, width // 12), max(2, height // 12)
rectangle = (margin_x, margin_y, max(1, width - 2 * margin_x), max(1, height - 2 * margin_y))
background = np.zeros((1, 65), np.float64)
foreground = np.zeros((1, 65), np.float64)
try:
cv2.grabCut(image, mask, rectangle, background, foreground, 4, cv2.GC_INIT_WITH_RECT)
result = np.where((mask == cv2.GC_FGD) | (mask == cv2.GC_PR_FGD), 255, 0).astype(np.uint8)
except cv2.error:
result = np.zeros((height, width), np.uint8)
result[margin_y : height - margin_y, margin_x : width - margin_x] = 255
components, labels, stats, centroids = cv2.connectedComponentsWithStats(result)
if components <= 1:
return result
centre = np.asarray([width / 2, height / 2])
best = max(
range(1, components),
key=lambda index: float(stats[index, cv2.CC_STAT_AREA]) / (1.0 + np.linalg.norm(centroids[index] - centre) / max(width, height)),
)
return np.where(labels == best, 255, 0).astype(np.uint8)
def _sam2_mask(image: np.ndarray, model_id: str, min_area_fraction: float, max_area_fraction: float) -> np.ndarray:
try:
from sam2.automatic_mask_generator import SAM2AutomaticMaskGenerator
except ImportError as error: # pragma: no cover - only exercised on a GPU host
raise RuntimeError("SAM2 is not installed. Run scripts/bootstrap_models.py --sam2 first, or set segmentation.backend=center.") from error
generator = SAM2AutomaticMaskGenerator.from_pretrained(
model_id,
points_per_side=32,
pred_iou_thresh=0.80,
stability_score_thresh=0.95,
crop_n_layers=1,
min_mask_region_area=400,
)
annotations = generator.generate(cv2.cvtColor(image, cv2.COLOR_BGR2RGB))
height, width = image.shape[:2]
centre = np.asarray([width / 2, height / 2])
candidates: list[tuple[float, np.ndarray]] = []
for annotation in annotations:
fraction = float(annotation["area"]) / float(width * height)
if not min_area_fraction <= fraction <= max_area_fraction:
continue
x, y, box_width, box_height = annotation["bbox"]
box_centre = np.asarray([x + box_width / 2, y + box_height / 2])
centrality = 1.0 - min(1.0, float(np.linalg.norm(box_centre - centre)) / (0.55 * max(width, height)))
score = 0.55 * float(annotation["predicted_iou"]) + 0.30 * float(annotation["stability_score"]) + 0.15 * centrality
candidates.append((score, np.asarray(annotation["segmentation"], dtype=np.uint8) * 255))
if not candidates:
raise RuntimeError("SAM2 found no plausible centred foreground; inspect the task contact sheet and retry with segmentation.backend=center")
return max(candidates, key=lambda pair: pair[0])[1]
def make_conditioning_image(source: Path, output: Path, segmentation: dict[str, Any]) -> dict[str, Any]:
image = cv2.imread(str(source), cv2.IMREAD_COLOR)
if image is None:
raise ValueError(f"could not read selected view: {source}")
backend = str(segmentation["backend"])
if backend == "center":
mask = _center_mask(image)
elif backend == "sam2":
mask = _sam2_mask(
image,
str(segmentation["sam2_model"]),
float(segmentation["min_area_fraction"]),
float(segmentation["max_area_fraction"]),
)
else:
raise ValueError(f"unknown segmentation backend: {backend}")
x, y, width, height = cv2.boundingRect(mask)
if width <= 2 or height <= 2:
raise RuntimeError(f"foreground mask is empty: {source}")
padding = max(4, round(0.06 * max(width, height)))
x0, y0 = max(0, x - padding), max(0, y - padding)
x1, y1 = min(image.shape[1], x + width + padding), min(image.shape[0], y + height + padding)
cropped_bgr = image[y0:y1, x0:x1]
cropped_mask = mask[y0:y1, x0:x1]
rgba = cv2.cvtColor(cropped_bgr, cv2.COLOR_BGR2RGBA)
rgba[:, :, 3] = cropped_mask
output.parent.mkdir(parents=True, exist_ok=True)
Image.fromarray(rgba).save(output)
return {
"source": str(source), "output": str(output), "backend": backend,
"crop_xyxy": [int(x0), int(y0), int(x1), int(y1)],
"mask_area_fraction": float((mask > 0).mean()),
"crop_width": int(x1 - x0), "crop_height": int(y1 - y0),
}

@ -0,0 +1,151 @@
from __future__ import annotations
import shutil
from pathlib import Path
from typing import Any
import mujoco
import numpy as np
import trimesh
from PIL import Image
from pxr import Gf, Sdf, Usd, UsdGeom, UsdPhysics, UsdShade
from .util import write_json
def _texture_coordinates(mesh: trimesh.Trimesh) -> np.ndarray:
visual = getattr(mesh, "visual", None)
uv = getattr(visual, "uv", None)
if uv is not None and len(uv) == len(mesh.vertices):
return np.asarray(uv, dtype=np.float32)
points = np.asarray(mesh.vertices, dtype=np.float64)
lower, upper = points[:, :2].min(axis=0), points[:, :2].max(axis=0)
return ((points[:, :2] - lower) / np.maximum(upper - lower, 1e-6)).astype(np.float32)
def _material(stage: Usd.Stage, texture: str) -> UsdShade.Material:
material = UsdShade.Material.Define(stage, "/World/Materials/Appearance")
shader = UsdShade.Shader.Define(stage, "/World/Materials/Appearance/PreviewSurface")
shader.CreateIdAttr("UsdPreviewSurface")
shader.CreateInput("roughness", Sdf.ValueTypeNames.Float).Set(0.55)
texture_node = UsdShade.Shader.Define(stage, "/World/Materials/Appearance/Texture")
texture_node.CreateIdAttr("UsdUVTexture")
texture_node.CreateInput("file", Sdf.ValueTypeNames.Asset).Set(Sdf.AssetPath(texture))
texture_node.CreateInput("sourceColorSpace", Sdf.ValueTypeNames.Token).Set("sRGB")
reader = UsdShade.Shader.Define(stage, "/World/Materials/Appearance/ST")
reader.CreateIdAttr("UsdPrimvarReader_float2")
reader.CreateInput("varname", Sdf.ValueTypeNames.Token).Set("st")
reader.CreateOutput("result", Sdf.ValueTypeNames.Float2)
texture_node.CreateInput("st", Sdf.ValueTypeNames.Float2).ConnectToSource(reader.ConnectableAPI(), "result")
texture_node.CreateOutput("rgb", Sdf.ValueTypeNames.Float3)
shader.CreateInput("diffuseColor", Sdf.ValueTypeNames.Color3f).ConnectToSource(texture_node.ConnectableAPI(), "rgb")
shader.CreateOutput("surface", Sdf.ValueTypeNames.Token)
material.CreateSurfaceOutput().ConnectToSource(shader.ConnectableAPI(), "surface")
return material
def _write_texture(mesh: trimesh.Trimesh, conditioning_image: Path, destination: Path) -> str:
"""Persist the generated GLB albedo when available, otherwise use the input view.
A high-profile Hunyuan result carries a UV texture in its GLB. Keeping that
image is essential: copying the conditioning frame unconditionally would
quietly discard Hunyuan Paint's result.
"""
material = getattr(getattr(mesh, "visual", None), "material", None)
generated = getattr(material, "image", None)
if generated is not None:
if isinstance(generated, Image.Image):
generated.convert("RGBA").save(destination)
else:
Image.fromarray(np.asarray(generated)).convert("RGBA").save(destination)
return "reconstructed_mesh_albedo"
shutil.copy2(conditioning_image, destination)
return "conditioning_image_fallback"
def _write_mjcf(task_id: str, extent: np.ndarray, output: Path, physics: dict[str, Any]) -> None:
size = np.maximum(extent / 2.0, 0.015)
xml = f'''<?xml version="1.0" encoding="utf-8"?>
<mujoco model="{task_id}">
<option timestep="0.002" gravity="0 0 -9.81"/>
<worldbody>
<body name="asset" pos="0 0 1">
<freejoint/>
<geom type="box" size="{size[0]:.7f} {size[1]:.7f} {size[2]:.7f}" mass="{float(physics['default_mass_kg']):.7f}" friction="{float(physics['friction']):.7f} 0.02 0.002"/>
</body>
</worldbody>
</mujoco>
'''
output.parent.mkdir(parents=True, exist_ok=True)
output.write_text(xml, encoding="utf-8")
def write_usd_asset(task_id: str, mesh_path: Path, conditioning_image: Path, output_dir: Path, physics: dict[str, Any]) -> dict[str, Any]:
"""Create a self-contained binary USD with visual mesh and collision data."""
mesh = trimesh.load(mesh_path, force="mesh")
if not isinstance(mesh, trimesh.Trimesh) or len(mesh.faces) == 0:
raise ValueError(f"invalid reconstructed mesh: {mesh_path}")
output_dir.mkdir(parents=True, exist_ok=True)
texture_dir = output_dir / "textures"
texture_dir.mkdir(exist_ok=True)
texture_path = texture_dir / "texture_00.png"
texture_source = _write_texture(mesh, conditioning_image, texture_path)
usd_path = output_dir / f"{task_id}.usd"
stage = Usd.Stage.CreateNew(str(usd_path))
UsdGeom.SetStageUpAxis(stage, UsdGeom.Tokens.z)
UsdGeom.SetStageMetersPerUnit(stage, 1.0)
world = UsdGeom.Xform.Define(stage, "/World")
stage.SetDefaultPrim(world.GetPrim())
UsdPhysics.Scene.Define(stage, "/World/PhysicsScene")
asset = UsdGeom.Xform.Define(stage, "/World/Asset")
UsdPhysics.RigidBodyAPI.Apply(asset.GetPrim())
mass = UsdPhysics.MassAPI.Apply(asset.GetPrim())
mass.CreateMassAttr(float(physics["default_mass_kg"]))
visual = UsdGeom.Mesh.Define(stage, "/World/Asset/Visual")
points = np.asarray(mesh.vertices, dtype=np.float32)
visual.CreatePointsAttr([Gf.Vec3f(float(point[0]), float(point[1]), float(point[2])) for point in points])
visual.CreateFaceVertexCountsAttr([3] * len(mesh.faces))
visual.CreateFaceVertexIndicesAttr([int(index) for face in mesh.faces for index in face])
visual.CreateSubdivisionSchemeAttr(UsdGeom.Tokens.none)
visual.CreateExtentAttr([
Gf.Vec3f(float(mesh.bounds[0, 0]), float(mesh.bounds[0, 1]), float(mesh.bounds[0, 2])),
Gf.Vec3f(float(mesh.bounds[1, 0]), float(mesh.bounds[1, 1]), float(mesh.bounds[1, 2])),
])
uv = _texture_coordinates(mesh)
primvars = UsdGeom.PrimvarsAPI(visual)
st = primvars.CreatePrimvar("st", Sdf.ValueTypeNames.TexCoord2fArray, UsdGeom.Tokens.vertex)
st.Set([Gf.Vec2f(float(value[0]), float(value[1])) for value in uv])
UsdShade.MaterialBindingAPI.Apply(visual.GetPrim()).Bind(_material(stage, "textures/texture_00.png"))
collision_mode = str(physics.get("collision_mode", "convex_hull"))
extent = np.asarray(mesh.extents, dtype=np.float32)
if collision_mode == "convex_hull":
# A convex hull follows the reconstructed silhouette much more closely
# than one axis-aligned cube, while remaining a stable dynamic shape in
# USD physics engines. The visible mesh stays the render mesh.
UsdPhysics.CollisionAPI.Apply(visual.GetPrim())
collision = UsdPhysics.MeshCollisionAPI.Apply(visual.GetPrim())
collision.CreateApproximationAttr().Set(UsdPhysics.Tokens.convexHull)
elif collision_mode == "bounding_box":
collider = UsdGeom.Cube.Define(stage, "/World/Asset/Collision")
collider.CreateSizeAttr(1.0)
collider.AddScaleOp().Set(Gf.Vec3f(float(extent[0] / 2.0), float(extent[1] / 2.0), float(extent[2] / 2.0)))
collider.CreateVisibilityAttr(UsdGeom.Tokens.invisible)
UsdPhysics.CollisionAPI.Apply(collider.GetPrim())
else:
raise ValueError(f"unsupported collision_mode: {collision_mode}")
stage.GetRootLayer().Save()
if not usd_path.is_file() or Usd.Stage.Open(str(usd_path)) is None:
raise RuntimeError(f"USD write/reopen failed: {usd_path}")
mjcf_path = output_dir.parent.parent / "physics" / task_id / f"{task_id}.xml"
_write_mjcf(task_id, extent, mjcf_path, physics)
return {"task_id": task_id, "usd": str(usd_path), "texture": str(texture_path), "texture_source": texture_source, "collision_mode": collision_mode, "mjcf": str(mjcf_path), "extent": [float(value) for value in extent]}
def simulate_mjcf(path: Path, steps: int) -> dict[str, Any]:
model = mujoco.MjModel.from_xml_path(str(path))
data = mujoco.MjData(model)
for _ in range(steps):
mujoco.mj_step(model, data)
if not np.isfinite(data.qpos).all() or not np.isfinite(data.qvel).all():
raise RuntimeError(f"non-finite MuJoCo state: {path}")
return {"mjcf": str(path), "steps": steps, "qpos": [float(value) for value in data.qpos], "qvel": [float(value) for value in data.qvel]}

@ -0,0 +1,61 @@
from __future__ import annotations
import hashlib
import json
import os
import random
import shutil
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Iterable
import numpy as np
def sha256(path: Path) -> str:
digest = hashlib.sha256()
with path.open("rb") as stream:
for block in iter(lambda: stream.read(1024 * 1024), b""):
digest.update(block)
return digest.hexdigest()
def utc_now() -> str:
return datetime.now(timezone.utc).isoformat(timespec="seconds")
def write_json(path: Path, value: Any) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps(value, ensure_ascii=False, indent=2, sort_keys=True) + "\n", encoding="utf-8")
def read_json(path: Path) -> Any:
return json.loads(path.read_text(encoding="utf-8"))
def seed_everything(seed: int) -> None:
random.seed(seed)
np.random.seed(seed)
os.environ.setdefault("PYTHONHASHSEED", str(seed))
def ensure_new_dir(path: Path) -> None:
if path.exists():
raise FileExistsError(f"refusing to overwrite existing run directory: {path}")
path.mkdir(parents=True)
def clean_dir(path: Path) -> None:
"""Create an output directory; it must be absent or an empty directory."""
if path.exists() and any(path.iterdir()):
raise FileExistsError(f"refusing to overwrite non-empty directory: {path}")
path.mkdir(parents=True, exist_ok=True)
def copy_file(source: Path, target: Path) -> None:
target.parent.mkdir(parents=True, exist_ok=True)
shutil.copy2(source, target)
def iter_files(root: Path) -> Iterable[Path]:
yield from sorted(path for path in root.rglob("*") if path.is_file() and not path.name.startswith("._"))

@ -0,0 +1,63 @@
from __future__ import annotations
from pathlib import Path
from typing import Any
from pxr import Sdf, Usd, UsdGeom, UsdPhysics, UsdShade
from .usd_asset import simulate_mjcf
from .util import utc_now
def _validate_usd(task_id: str, asset: Path) -> tuple[list[str], dict[str, int]]:
errors: list[str] = []
counts = {"gprims": 0, "colliders": 0, "rigid_bodies": 0, "physics_scenes": 0, "textures": 0}
stage = Usd.Stage.Open(str(asset))
if stage is None:
return ["Usd.Stage.Open returned None"], counts
if stage.GetDefaultPrim().GetPath() != Sdf.Path("/World"):
errors.append("default prim is not /World")
if UsdGeom.GetStageUpAxis(stage) != UsdGeom.Tokens.z:
errors.append("up axis is not Z")
if abs(float(UsdGeom.GetStageMetersPerUnit(stage)) - 1.0) > 1e-9:
errors.append("meters per unit is not 1")
for prim in stage.Traverse():
counts["gprims"] += int(prim.IsA(UsdGeom.Gprim))
counts["colliders"] += int(prim.HasAPI(UsdPhysics.CollisionAPI))
counts["rigid_bodies"] += int(prim.HasAPI(UsdPhysics.RigidBodyAPI))
counts["physics_scenes"] += int(prim.IsA(UsdPhysics.Scene))
if prim.IsA(UsdShade.Shader):
shader = UsdShade.Shader(prim)
file_input = shader.GetInput("file")
value = file_input.Get() if file_input else None
if isinstance(value, Sdf.AssetPath) and value.path:
counts["textures"] += 1
if Path(value.path).is_absolute() or value.path.startswith(("http://", "https://")):
errors.append(f"non-local texture reference: {value.path}")
elif not (asset.parent / value.path).is_file():
errors.append(f"missing texture reference: {value.path}")
if counts["gprims"] < 1:
errors.append("no renderable geometry")
if counts["physics_scenes"] != 1:
errors.append(f"expected one physics scene, found {counts['physics_scenes']}")
if counts["colliders"] < 1 or counts["rigid_bodies"] < 1:
errors.append("missing conservative collision/rigid-body APIs")
return errors, counts
def validate_submission_tree(submission: Path, physics_root: Path, steps: int) -> dict[str, Any]:
records: list[dict[str, Any]] = []
all_errors: list[str] = []
for index in range(1, 35):
task_id = f"item_{index:03d}"
usd = submission / task_id / f"{task_id}.usd"
mjcf = physics_root / task_id / f"{task_id}.xml"
errors, counts = _validate_usd(task_id, usd)
simulation: dict[str, Any] | None = None
try:
simulation = simulate_mjcf(mjcf, steps)
except Exception as error:
errors.append(f"MuJoCo: {type(error).__name__}: {error}")
all_errors.extend(f"{task_id}: {error}" for error in errors)
records.append({"task_id": task_id, "usd": str(usd), "mjcf": str(mjcf), "valid": not errors, "errors": errors, "counts": counts, "simulation": simulation})
return {"created_at": utc_now(), "valid": not all_errors, "task_count": len(records), "valid_tasks": sum(record["valid"] for record in records), "simulation_steps": steps, "errors": all_errors, "tasks": records}

@ -0,0 +1,9 @@
from __future__ import annotations
import sys
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / "src"))
sys.path.insert(0, str(ROOT / "scripts"))

@ -0,0 +1,65 @@
from __future__ import annotations
import zipfile
from pathlib import Path
import cv2
import numpy as np
from asset_baseline.input_data import inspect_inputs
from asset_baseline.pipeline import run_all
def _write_video(path: Path) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
writer = cv2.VideoWriter(str(path), cv2.VideoWriter_fourcc(*"mp4v"), 5.0, (64, 48))
assert writer.isOpened()
for index in range(4):
frame = np.zeros((48, 64, 3), dtype=np.uint8)
cv2.rectangle(frame, (12 + index, 10), (48 + index, 38), (20, 160, 230), -1)
writer.write(frame)
writer.release()
def _official_like_inputs(root: Path) -> tuple[Path, Path]:
source = root / "source"
for index in range(1, 35):
_write_video(source / f"item_{index:03d}" / "turntable.mp4")
question = root / "question.zip"
with zipfile.ZipFile(question, "w", compression=zipfile.ZIP_DEFLATED) as archive:
for video in sorted(source.rglob("*.mp4")):
archive.write(video, video.relative_to(source).as_posix())
example = root / "submission_example.zip"
with zipfile.ZipFile(example, "w") as archive:
archive.writestr("submission_example/submission/item_001/item_001.usd", "# placeholder")
return question, example
def test_inspect_accepts_the_expected_input_contract(tmp_path: Path) -> None:
question, example = _official_like_inputs(tmp_path)
manifest = inspect_inputs(question, example)
assert manifest["task_count"] == 34
assert manifest["video_count"] == 34
def test_cpu_smoke_pipeline_generates_a_valid_submission(tmp_path: Path) -> None:
question, example = _official_like_inputs(tmp_path)
config = {
"run_name": "cpu_smoke",
"seed": 7,
"question_zip": str(question),
"submission_example_zip": str(example),
"output_root": str(tmp_path / "outputs"),
"video": {"frames_per_video": 2, "selected_views": 1, "max_edge": 128, "jpeg_quality": 90},
"reconstruction": {"backend": "primitive", "hunyuan_repo": "unused", "hunyuan_model": "unused", "hunyuan_texture": False, "max_faces": 30000},
"segmentation": {"backend": "center", "sam2_repo": "unused", "sam2_model": "unused", "min_area_fraction": 0.03, "max_area_fraction": 0.90},
"physics": {"collision_mode": "convex_hull", "default_mass_kg": 1.0, "density_kg_m3": 700.0, "friction": 0.8, "simulation_steps": 4},
"package": {"archive_root": "submission_example/submission", "compression_level": 6},
}
report = run_all(config)
assert report["valid"] is True
assert report["valid_tasks"] == 34
package = tmp_path / "outputs/cpu_smoke/packages/asset_baseline_submission.zip"
with zipfile.ZipFile(package) as archive:
assert archive.testzip() is None
assert sum(name.endswith(".usd") for name in archive.namelist()) == 34

@ -0,0 +1,33 @@
from __future__ import annotations
import json
from pathlib import Path
import rebuild_submission
import validate_submission
ROOT = Path(__file__).resolve().parents[1]
def test_evidence_matches_locked_candidate() -> None:
evidence = json.loads((ROOT / "evidence/online_score_evidence.json").read_text(encoding="utf-8"))
assert evidence["candidate_sha256"] == rebuild_submission.FINAL_SHA256
assert evidence["score"] == 67.6
def test_quick_rebuild_is_byte_identical_and_valid(tmp_path: Path) -> None:
package = tmp_path / "submission.zip"
report = rebuild_submission.build(
ROOT / "artifacts/intermediate/base_submission.zip",
ROOT / "artifacts/intermediate/donor_submission.zip",
package,
)
assert report["output"]["sha256"] == rebuild_submission.FINAL_SHA256
assert len(report["member_lineage"]) == rebuild_submission.FINAL_MEMBERS
assert {row["name"] for row in report["member_lineage"] if row["origin"] == "donor"} == set(rebuild_submission.REPLACED_MEMBERS)
assert package.read_bytes() == (ROOT / "artifacts/reference/submission.zip").read_bytes()
validation = validate_submission.validate_package(package)
assert validation["valid"] is True
assert validation["valid_tasks"] == 34
assert validation["physics_probe_steps"] == 120

@ -0,0 +1,2 @@
# Release 二进制文件只在本地暂存,官方核验通过后手动上传到 GitHub Release。
*.zip

@ -0,0 +1,5 @@
# 本地待上传 Release Asset
`public_asset_baseline_release.zip` 已在本目录本地准备完成,但被 `.gitignore` 排除,不会进入 Git 提交。
官方核验通过并完成源码推送后,在 GitHub 的 **Releases** 页面创建发布条目,将该 ZIP 拖入附件区域。上传前请按上级目录的 [CHECKSUMS.md](../CHECKSUMS.md) 核对 SHA-256。

@ -0,0 +1,7 @@
# 竞赛实践
本目录收录带有明确复现边界、环境说明和本地校验方式的竞赛项目。
| 项目 | 内容 | 复现入口 |
|---|---|---|
| [2026 具身仿真合成挑战赛](2026-具身仿真合成挑战赛/README.md) | 视频到可物理仿真的 USD 资产生成 Baseline | 路径 A历史 67+ 候选的精确复现;路径 B从官方输入的 GPU 重建 |
Loading…
Cancel
Save