src/mcss/data/hypersim.py
"""Hypersim subset conversion using official RGB, camera, depth, and normal files."""
from __future__ import annotations
import csv
import re
from pathlib import Path
import h5py
import numpy as np
from PIL import Image
from mcss.data.manifest import ManifestFrame, SceneManifest, save_manifest
from mcss.data.replica import _backproject_depth, _estimate_bounds, _resize_scalar
def convert_hypersim_pose(
orientation: np.ndarray, position: np.ndarray, *, meters_per_asset_unit: float
) -> np.ndarray:
"""Convert Hypersim OpenGL camera axes (x-right/y-up/z-back) to OpenCV c2w axes."""
rotation = np.asarray(orientation, dtype=np.float32)
translation = np.asarray(position, dtype=np.float32)
if rotation.shape != (3, 3) or translation.shape != (3,):
raise ValueError("Hypersim orientation and position require shapes [3, 3] and [3]")
if not np.isfinite(rotation).all() or not np.isfinite(translation).all():
raise ValueError("Hypersim pose values must be finite")
if meters_per_asset_unit <= 0:
raise ValueError("meters_per_asset_unit must be positive")
pose = np.eye(4, dtype=np.float32)
pose[:3, :3] = rotation @ np.diag([1.0, -1.0, -1.0]).astype(np.float32)
pose[:3, 3] = translation * meters_per_asset_unit
return pose
def prepare_hypersim(
raw_root: str | Path,
output_root: str | Path,
*,
scenes: list[str] | None = None,
camera: str = "cam_00",
image_size: tuple[int, int] | None = None,
frame_stride: int = 1,
frame_limit: int | None = None,
fov_x_degrees: float = 60.0,
) -> list[Path]:
"""Convert a selectively downloaded Hypersim subset to common manifests.
The official preview images are 1024x768 with a 60 degree horizontal field of view. The
default can be overridden if a future official camera calibration export is supplied.
"""
if not re.fullmatch(r"cam_\d{2}", camera):
raise ValueError("camera must match cam_DD")
if frame_stride < 1:
raise ValueError("frame_stride must be positive")
root = Path(raw_root)
available = [
path
for path in root.iterdir()
if path.is_dir() and re.fullmatch(r"ai_\d{3}_\d{3}", path.name)
]
if re.fullmatch(r"ai_\d{3}_\d{3}", root.name):
available = [root]
requested = set(scenes) if scenes is not None else {path.name for path in available}
selected = [path for path in available if path.name in requested]
missing = requested - {path.name for path in selected}
if missing:
raise FileNotFoundError(f"Hypersim scenes not found below raw root: {sorted(missing)}")
return [
_prepare_scene(
scene, Path(output_root), camera, image_size, frame_stride, frame_limit, fov_x_degrees
)
for scene in sorted(selected)
]
def _prepare_scene(
scene: Path,
output_root: Path,
camera: str,
image_size: tuple[int, int] | None,
frame_stride: int,
frame_limit: int | None,
fov_x_degrees: float,
) -> Path:
orientations = _read_hdf5(scene / "_detail" / camera / "camera_keyframe_orientations.hdf5")
positions = _read_hdf5(scene / "_detail" / camera / "camera_keyframe_positions.hdf5")
scale = _meters_per_asset_unit(scene / "_detail" / "metadata_scene.csv")
rgb_files = _frame_files(scene / "images" / f"scene_{camera}_final_preview", ".color.jpg")
depth_files = _frame_files(
scene / "images" / f"scene_{camera}_geometry_hdf5", ".depth_meters.hdf5"
)
normal_files = _frame_files(
scene / "images" / f"scene_{camera}_geometry_hdf5", ".normal_world.hdf5"
)
frame_ids = sorted(set(rgb_files) & set(depth_files))[::frame_stride]
if frame_limit is not None:
frame_ids = frame_ids[:frame_limit]
if not frame_ids:
raise FileNotFoundError(
f"no matching preview RGB and depth frames found for {scene.name}/{camera}"
)
if max(frame_ids) >= len(orientations) or max(frame_ids) >= len(positions):
raise ValueError("Hypersim camera trajectory is shorter than selected frame ids")
destination = output_root / scene.name
(destination / "rgb").mkdir(parents=True, exist_ok=True)
(destination / "depth").mkdir(parents=True, exist_ok=True)
(destination / "normal").mkdir(parents=True, exist_ok=True)
frames: list[ManifestFrame] = []
point_sets: list[np.ndarray] = []
for frame_id in frame_ids:
image = Image.open(rgb_files[frame_id]).convert("RGB")
original_size = (image.height, image.width)
target_size = image_size or original_size
if target_size != original_size:
image = image.resize((target_size[1], target_size[0]), Image.Resampling.BICUBIC)
rgb_relative = f"rgb/{frame_id:06d}.png"
image.save(destination / rgb_relative)
depth = _read_hdf5(depth_files[frame_id]).astype(np.float32)
if depth.ndim != 2:
raise ValueError(f"Hypersim depth must be HxW: {depth_files[frame_id]}")
if target_size != depth.shape:
depth = _resize_scalar(depth, target_size)
depth[~np.isfinite(depth) | (depth <= 0)] = 0.0
depth_relative = f"depth/{frame_id:06d}.npy"
np.save(destination / depth_relative, depth)
normal_relative = None
if frame_id in normal_files:
normal = _read_hdf5(normal_files[frame_id]).astype(np.float32)
if normal.shape[-1] != 3:
raise ValueError(f"Hypersim normal must end in channel 3: {normal_files[frame_id]}")
if target_size != normal.shape[:2]:
normal = _resize_vector(normal, target_size)
normal = _normalize_vectors(normal)
normal_relative = f"normal/{frame_id:06d}.npy"
np.save(destination / normal_relative, normal)
intrinsics = _intrinsics_for_size(target_size, fov_x_degrees)
pose = convert_hypersim_pose(
orientations[frame_id], positions[frame_id], meters_per_asset_unit=scale
)
point_sets.append(_backproject_depth(depth, intrinsics, pose, stride=16))
frames.append(
ManifestFrame(
frame_id=frame_id,
rgb=rgb_relative,
intrinsics=intrinsics.tolist(),
c2w=pose.tolist(),
depth=depth_relative,
normal=normal_relative,
)
)
manifest = SceneManifest(
scene_id=scene.name,
image_size=target_size,
bounds=_estimate_bounds(point_sets),
frames=tuple(frames),
root=destination,
)
return save_manifest(manifest, destination / "manifest.json")
def _read_hdf5(path: Path) -> np.ndarray:
if not path.is_file():
raise FileNotFoundError(path)
with h5py.File(path, "r") as handle:
datasets: list[h5py.Dataset] = []
handle.visititems(
lambda _name, object_: (
datasets.append(object_) if isinstance(object_, h5py.Dataset) else None
)
)
if len(datasets) != 1:
raise ValueError(f"expected exactly one dataset in {path}, found {len(datasets)}")
return np.asarray(datasets[0])
def _frame_files(directory: Path, suffix: str) -> dict[int, Path]:
if not directory.is_dir():
return {}
result: dict[int, Path] = {}
pattern = re.compile(r"frame\.(\d{4})\.")
for path in directory.glob(f"*{suffix}"):
match = pattern.search(path.name)
if match:
result[int(match.group(1))] = path
return result
def _meters_per_asset_unit(path: Path) -> float:
if not path.is_file():
return 1.0
with path.open(newline="", encoding="utf-8") as handle:
for row in csv.reader(handle):
normalized = [cell.strip() for cell in row]
if any(cell == "meters_per_asset_unit" for cell in normalized):
for cell in reversed(normalized):
try:
value = float(cell)
except ValueError:
continue
if value > 0:
return value
return 1.0
def _intrinsics_for_size(image_size: tuple[int, int], fov_x_degrees: float) -> np.ndarray:
height, width = image_size
focal = (width - 1) / (2.0 * np.tan(np.deg2rad(fov_x_degrees) / 2.0))
return np.array(
[[focal, 0.0, (width - 1) / 2.0], [0.0, focal, (height - 1) / 2.0], [0.0, 0.0, 1.0]],
dtype=np.float32,
)
def _resize_vector(values: np.ndarray, image_size: tuple[int, int]) -> np.ndarray:
channels = [_resize_scalar(values[..., channel], image_size) for channel in range(3)]
return np.stack(channels, axis=-1)
def _normalize_vectors(values: np.ndarray) -> np.ndarray:
norm = np.linalg.norm(values, axis=-1, keepdims=True)
valid = norm > 1e-8
return np.where(valid, values / np.clip(norm, 1e-8, None), 0.0).astype(np.float32)