src/mcss/data/dl3dv_benchmark.py
"""Metadata and lazy RGB access for the pinned DL3DV-140 protocol.
The benchmark has two deliberately separate surfaces:
* :func:`prepare_benchmark_metadata` reads only ``transforms.json`` files and
the pinned Long-LRM split. It never opens an image.
* :class:`Dl3dvOnlineObservationSource` opens one image when that observation
arrives. Query images are exposed through callbacks only after a
:class:`~mcss.dynamic.types.SealedScene` has been supplied.
The coordinate convention and image processing follow the pinned Long-LRM/
tttLRM converter. The online gauge is intentionally different from tttLRM:
it is a frozen warmup-only baseline normalization, because query and future
poses are not available to a causal episode.
"""
from __future__ import annotations
import hashlib
import json
import math
import struct
from collections.abc import Iterable, Mapping, Sequence
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Literal
import numpy as np
import torch
from PIL import Image
from mcss.dynamic.types import OnlineObservation, SealedScene
from mcss.evaluation.sealed_rgb import RGBQuery
from mcss.types import Cameras
DL3DV_SCENE_COUNT = 140
DL3DV_TARGET_STRIDE = 8
DL3DV_WARMUP_LENGTH = 4
DL3DV_SOURCE_IMAGE_SIZE = (540, 960) # H, W; images_4
DL3DV_IMAGES4_DOWNSAMPLE = 4
DL3DV_EVAL_IMAGE_SIZE = (536, 960) # H, W; tttLRM evaluation config
DL3DV_WARMUP_TARGET_EXTENT_M = 1.9428480059945201
DL3DV_MIN_WARMUP_BASELINE = 1.0e-6
DL3DV_LOCAL_BOUNDS_M = ((-6.0, -4.0, 0.05), (6.0, 4.0, 12.05))
DL3DV_METADATA_SCHEMA = "mcss.dl3dv.benchmark_metadata.v1"
DL3DV_GAUGE_SCHEMA = "mcss.dl3dv.warmup_gauge.v1"
# These identify the public Long-LRM split, rather than the separate DL3DV
# dataset revision used for the RGB files.
DL3DV_SPLIT_REVISION = "0e89431a964b8a7e2fdec24e3ca5ec74a26dcafc"
DL3DV_SPLIT_FILE_SHA1 = "ae0ddcd09b64753210c46978124b926dd223fa98"
# Ordinary SHA-256 of the pinned split bytes. The 40-character value above is
# the Git blob SHA-1 and must not be used as a SHA-256 substitute.
DL3DV_SPLIT_SHA256 = "4b052aef605fda183683175de66250aaaf337e50003131dbf0113ea781a3145a"
DL3DV_RIGID_ROTATION_TOLERANCE = 1e-3
DL3DV_GAUGE_CALIBRATION = "warmup_gauge_calibration_v2.json"
DL3DV_GAUGE_VALID_WINDOWS = 332
DL3DV_GAUGE_EXCLUDED_WINDOWS = 31
ProtocolMode = Literal["full", "ar"]
_FULL_INPUT_COUNTS = frozenset({16, 32, 64})
_AR_INPUT_COUNTS = frozenset({32, 64})
class Dl3dvProtocolError(ValueError):
"""The pinned split or online episode violates the DL3DV protocol."""
def _as_float_tuple(values: Sequence[Any], length: int, name: str) -> tuple[float, ...]:
if not isinstance(values, Sequence) or len(values) != length:
raise Dl3dvProtocolError(f"{name} must contain exactly {length} values")
result = tuple(float(value) for value in values)
if not all(math.isfinite(value) for value in result):
raise Dl3dvProtocolError(f"{name} must contain finite values")
return result
def _validate_homogeneous(matrix: np.ndarray, name: str) -> np.ndarray:
value = np.asarray(matrix, dtype=np.float64)
if value.shape != (4, 4) or not np.isfinite(value).all():
raise Dl3dvProtocolError(f"{name} must be a finite 4x4 matrix")
if not np.allclose(value[3], (0.0, 0.0, 0.0, 1.0), atol=1e-6, rtol=1e-6):
raise Dl3dvProtocolError(f"{name} must have homogeneous final row [0,0,0,1]")
try:
np.linalg.inv(value)
except np.linalg.LinAlgError as error:
raise Dl3dvProtocolError(f"{name} must be invertible") from error
return value
def _validate_rigid_camera(matrix: np.ndarray, name: str) -> np.ndarray:
"""Validate a camera pose as a proper rigid transform in ``SE(3)``."""
value = _validate_homogeneous(matrix, name)
rotation = value[:3, :3]
if not np.allclose(
rotation.T @ rotation,
np.eye(3, dtype=np.float64),
atol=DL3DV_RIGID_ROTATION_TOLERANCE,
rtol=DL3DV_RIGID_ROTATION_TOLERANCE,
):
raise Dl3dvProtocolError(
f"{name} rotation is not orthonormal within "
f"{DL3DV_RIGID_ROTATION_TOLERANCE:g}"
)
determinant = float(np.linalg.det(rotation))
if not math.isclose(
determinant,
1.0,
abs_tol=DL3DV_RIGID_ROTATION_TOLERANCE,
rel_tol=DL3DV_RIGID_ROTATION_TOLERANCE,
):
raise Dl3dvProtocolError(
f"{name} rotation must have determinant +1, got {determinant:.6g}"
)
return value
def official_pose_transform(transform_matrix: Sequence[Sequence[float]]) -> np.ndarray:
"""Apply the exact c2w transform used by Long-LRM's DL3DV converter."""
c2w = _validate_homogeneous(np.asarray(transform_matrix), "transform_matrix").copy()
c2w[0:3, 1:3] *= -1.0
c2w = c2w[[1, 0, 2, 3], :]
c2w[2, :] *= -1.0
return _validate_homogeneous(c2w, "converted c2w")
def official_intrinsics(
fxfycxcy: Sequence[float], *, source_size: tuple[int, int]
) -> np.ndarray:
"""Build the raw OpenCV intrinsic matrix from ``fx, fy, cx, cy``."""
values = _as_float_tuple(fxfycxcy, 4, "fxfycxcy")
height, width = source_size
if height < 2 or width < 2:
raise Dl3dvProtocolError("source image size must be at least 2x2")
fx, fy, cx, cy = values
if fx <= 0.0 or fy <= 0.0:
raise Dl3dvProtocolError("source focal lengths must be positive")
return np.asarray([[fx, 0.0, cx], [0.0, fy, cy], [0.0, 0.0, 1.0]], dtype=np.float64)
def _cv2() -> Any:
try:
import cv2
except ImportError as error: # pragma: no cover - environment dependent
raise RuntimeError("opencv-python is required for DL3DV undistortion") from error
return cv2
def undistorted_intrinsics(
raw_intrinsics: np.ndarray,
distortion: Sequence[float],
*,
source_size: tuple[int, int],
) -> np.ndarray:
"""Return the converter's ``getOptimalNewCameraMatrix(alpha=0)`` result.
This operation depends only on camera metadata and image dimensions. It
does not read any image pixels, so it is safe during metadata preparation.
"""
height, width = source_size
if height < 2 or width < 2:
raise Dl3dvProtocolError("source image size must be at least 2x2")
cv2 = _cv2()
distortion_array = np.asarray(_as_float_tuple(distortion, 4, "distortion"), dtype=np.float64)
matrix, _ = cv2.getOptimalNewCameraMatrix(
np.asarray(raw_intrinsics, dtype=np.float64),
distortion_array,
(width, height),
0,
(width, height),
)
result = np.asarray(matrix, dtype=np.float64)
if result.shape != (3, 3) or not np.isfinite(result).all():
raise Dl3dvProtocolError("undistorted intrinsic matrix is invalid")
return result
def resize_and_center_crop(
image: Image.Image,
target_size: tuple[int, int] = DL3DV_EVAL_IMAGE_SIZE,
fxfycxcy: Sequence[float] | None = None,
) -> tuple[Image.Image, tuple[float, float, float, float] | None]:
"""Match tttLRM's literal anisotropic resize/crop implementation.
The public notes describe this as resizing to cover. The pinned source
actually uses independent ``scale_x`` and ``scale_y`` values, rounds each
resized dimension, and then center-crops. This function preserves that
behavior and applies the same transform to intrinsics.
"""
target_height, target_width = target_size
if target_height < 2 or target_width < 2:
raise Dl3dvProtocolError("target image size must be at least 2x2")
original_width, original_height = image.size
if original_width < 2 or original_height < 2:
raise Dl3dvProtocolError("source image must be at least 2x2")
scale_x = target_width / original_width
scale_y = target_height / original_height
new_width = int(round(original_width * scale_x))
new_height = int(round(original_height * scale_y))
resized = image.resize((new_width, new_height), Image.Resampling.LANCZOS)
left = (new_width - target_width) // 2
top = (new_height - target_height) // 2
cropped = resized.crop((left, top, left + target_width, top + target_height)).convert("RGB")
if fxfycxcy is None:
return cropped, None
fx, fy, cx, cy = _as_float_tuple(fxfycxcy, 4, "fxfycxcy")
return cropped, (
fx * scale_x,
fy * scale_y,
cx * scale_x - left,
cy * scale_y - top,
)
def _codec_roundtrip(image_bgr: np.ndarray, suffix: str) -> np.ndarray:
"""Emulate converter ``cv2.imwrite`` followed by dataset PIL loading."""
cv2 = _cv2()
extension = suffix.lower() or ".png"
if extension not in {".jpg", ".jpeg", ".png", ".webp", ".bmp", ".tif", ".tiff"}:
raise Dl3dvProtocolError(f"unsupported DL3DV image extension: {suffix!r}")
ok, encoded = cv2.imencode(extension, image_bgr)
if not ok:
raise Dl3dvProtocolError(f"OpenCV could not encode undistorted image as {extension}")
decoded = cv2.imdecode(encoded, cv2.IMREAD_COLOR)
if decoded is None:
raise Dl3dvProtocolError("OpenCV could not decode its undistorted image round-trip")
return decoded
def preprocess_dl3dv_rgb(
image_path: str | Path,
frame: Dl3dvFrame,
*,
target_size: tuple[int, int] = DL3DV_EVAL_IMAGE_SIZE,
) -> tuple[torch.Tensor, np.ndarray]:
"""Load and preprocess one frame using the official converter/data path.
The returned tensor is channel-first RGB ``float32`` in ``[0, 1]`` and the
returned intrinsic matrix matches the resized/cropped image. This function
is intentionally called only by an observation or query callback.
"""
cv2 = _cv2()
source = cv2.imread(str(image_path), cv2.IMREAD_COLOR)
if source is None:
raise FileNotFoundError(f"DL3DV RGB image could not be read: {image_path}")
height, width = int(source.shape[0]), int(source.shape[1])
expected_height, expected_width = frame.source_image_size
if (height, width) != (expected_height, expected_width):
raise Dl3dvProtocolError(
f"DL3DV source image size differs from transforms.json: {(height, width)} != "
f"{(expected_height, expected_width)}"
)
raw_intrinsics = frame.raw_intrinsics_matrix
distortion = np.asarray(frame.distortion, dtype=np.float64)
new_intrinsics, _ = cv2.getOptimalNewCameraMatrix(
raw_intrinsics, distortion, (width, height), 0, (width, height)
)
undistorted = cv2.undistort(source, raw_intrinsics, distortion, None, new_intrinsics)
# The converter writes the undistorted BGR array using the original suffix,
# after which tttLRM opens the resulting file with PIL. Reproduce that
# codec boundary so JPEG scenes retain the same small loss as the official path.
roundtripped = _codec_roundtrip(undistorted, Path(frame.rgb_relpath).suffix)
rgb = Image.fromarray(cv2.cvtColor(roundtripped, cv2.COLOR_BGR2RGB), mode="RGB")
cropped, fxfycxcy = resize_and_center_crop(
rgb,
target_size,
(
float(new_intrinsics[0, 0]),
float(new_intrinsics[1, 1]),
float(new_intrinsics[0, 2]),
float(new_intrinsics[1, 2]),
),
)
assert fxfycxcy is not None
array = np.asarray(cropped, dtype=np.float32).copy()
tensor = torch.from_numpy(array).permute(2, 0, 1).contiguous().div(255.0)
intrinsic = official_intrinsics(fxfycxcy, source_size=target_size)
return tensor, intrinsic
@dataclass(frozen=True)
class Dl3dvProtocol:
"""One supported Long-LRM input selection condition."""
mode: ProtocolMode
input_count: int
fold_size: int = DL3DV_TARGET_STRIDE
warmup_length: int = DL3DV_WARMUP_LENGTH
def __post_init__(self) -> None:
if self.mode not in ("full", "ar"):
raise Dl3dvProtocolError("DL3DV mode must be 'full' or 'ar'")
allowed = _FULL_INPUT_COUNTS if self.mode == "full" else _AR_INPUT_COUNTS
if self.input_count not in allowed:
raise Dl3dvProtocolError(
f"{self.mode} mode supports input counts {sorted(allowed)}, got {self.input_count}"
)
if self.fold_size != DL3DV_TARGET_STRIDE or self.fold_size < 1:
raise Dl3dvProtocolError("the pinned DL3DV protocol uses fold_size=8")
if self.warmup_length != DL3DV_WARMUP_LENGTH:
raise Dl3dvProtocolError("the causal DL3DV adapter requires four warmup views")
@property
def split_key(self) -> str:
return f"fold_{self.fold_size}_kmeans_{self.input_count}_input"
@property
def name(self) -> str:
return f"{self.mode}{self.input_count}"
@dataclass(frozen=True)
class LongLRMSplitRecord:
scene_name: str
input_indices_by_count: Mapping[int, tuple[int, ...]]
def input_indices(self, input_count: int) -> tuple[int, ...]:
try:
return self.input_indices_by_count[input_count]
except KeyError as error:
raise Dl3dvProtocolError(
f"Long-LRM split has no input array for N={input_count}: {self.scene_name}"
) from error
def to_dict(self) -> dict[str, Any]:
return {
"scene_name": self.scene_name,
**{
f"fold_8_kmeans_{count}_input": list(indices)
for count, indices in sorted(self.input_indices_by_count.items())
},
}
def load_long_lrm_split(
path: str | Path,
*,
expected_scene_count: int = DL3DV_SCENE_COUNT,
expected_sha256: str | None = DL3DV_SPLIT_SHA256,
) -> tuple[LongLRMSplitRecord, ...]:
"""Read, authenticate, and structurally validate the Long-LRM split JSON.
The default path is intentionally pinned to the public split bytes used by
this benchmark. ``expected_sha256=None`` is available only for isolated
structural fixtures; production metadata preparation keeps the default.
"""
source = Path(path)
split_bytes = source.read_bytes()
actual_sha256 = hashlib.sha256(split_bytes).hexdigest()
if expected_sha256 is not None and actual_sha256 != expected_sha256:
raise Dl3dvProtocolError(
"Long-LRM split SHA256 does not match the pinned revision: "
f"expected={expected_sha256}, actual={actual_sha256}"
)
try:
value = json.loads(split_bytes.decode("utf-8"))
except json.JSONDecodeError as error:
raise Dl3dvProtocolError(f"invalid Long-LRM split JSON: {source}") from error
if not isinstance(value, list):
raise Dl3dvProtocolError("Long-LRM split root must be a list")
if expected_scene_count is not None and len(value) != expected_scene_count:
raise Dl3dvProtocolError(
f"expected {expected_scene_count} Long-LRM scenes, got {len(value)}"
)
records: list[LongLRMSplitRecord] = []
seen: set[str] = set()
for row_index, row in enumerate(value):
if not isinstance(row, Mapping):
raise Dl3dvProtocolError(f"Long-LRM split row {row_index} must be an object")
scene_name = row.get("scene_name")
if not isinstance(scene_name, str) or not scene_name or any(
char in scene_name for char in "\\/"
):
raise Dl3dvProtocolError(f"invalid scene_name at Long-LRM row {row_index}")
if scene_name in seen:
raise Dl3dvProtocolError(f"duplicate Long-LRM scene_name: {scene_name}")
seen.add(scene_name)
arrays: dict[int, tuple[int, ...]] = {}
for count in (16, 32, 64, 128):
key = f"fold_8_kmeans_{count}_input"
raw_indices = row.get(key)
if not isinstance(raw_indices, list):
raise Dl3dvProtocolError(f"{scene_name} is missing {key}")
indices = tuple(raw_indices)
if any(type(index) is not int or index < 0 for index in indices):
raise Dl3dvProtocolError(f"{scene_name} has invalid indices in {key}")
if len(indices) != count or len(set(indices)) != count:
raise Dl3dvProtocolError(f"{scene_name} has invalid length or duplicates in {key}")
arrays[count] = indices
records.append(LongLRMSplitRecord(scene_name, arrays))
return tuple(records)
def target_frame_indices(frame_count: int, stride: int = DL3DV_TARGET_STRIDE) -> tuple[int, ...]:
"""Return the exact every-eighth target order used by tttLRM."""
if type(frame_count) is not int or frame_count < 1:
raise Dl3dvProtocolError("frame_count must be a positive integer")
if type(stride) is not int or stride < 1:
raise Dl3dvProtocolError("target stride must be a positive integer")
return tuple(range(0, frame_count, stride))
def _validate_scene_indices(
frame_count: int,
input_indices: Sequence[int],
*,
target_stride: int,
) -> tuple[tuple[int, ...], tuple[int, ...]]:
targets = target_frame_indices(frame_count, target_stride)
target_set = set(targets)
inputs = tuple(input_indices)
if len(set(inputs)) != len(inputs):
raise Dl3dvProtocolError("DL3DV input indices must be unique")
if any(type(index) is not int or index < 0 or index >= frame_count for index in inputs):
raise Dl3dvProtocolError("DL3DV input indices must be in the scene frame range")
overlap = sorted(target_set.intersection(inputs))
if overlap:
raise Dl3dvProtocolError(f"DL3DV context/query overlap at frame {overlap[0]}")
available = frame_count - len(targets)
if len(inputs) > available:
raise Dl3dvProtocolError(
f"DL3DV scene has {available} non-target frames but needs {len(inputs)} inputs"
)
return inputs, targets
def validate_complete_coverage(
scenes: Sequence[Dl3dvSceneMetadata],
*,
expected_scene_count: int = DL3DV_SCENE_COUNT,
expected_scene_names: Iterable[str] | None = None,
) -> dict[str, Any]:
"""Validate that metadata covers the complete benchmark exactly once."""
if len(scenes) != expected_scene_count:
raise Dl3dvProtocolError(
f"expected complete DL3DV coverage of {expected_scene_count} scenes, got {len(scenes)}"
)
names = [scene.scene_id for scene in scenes]
if len(names) != len(set(names)):
raise Dl3dvProtocolError("DL3DV metadata contains duplicate scene IDs")
if expected_scene_names is not None:
expected = set(expected_scene_names)
actual = set(names)
if actual != expected:
missing = sorted(expected - actual)
extra = sorted(actual - expected)
raise Dl3dvProtocolError(
"DL3DV scene coverage differs from pinned split; "
f"missing={missing[:3]}, extra={extra[:3]}"
)
return {
"scene_count": len(scenes),
"scene_ids": tuple(names),
"target_stride": scenes[0].protocol.fold_size if scenes else DL3DV_TARGET_STRIDE,
"query_counts": {scene.scene_id: len(scene.query_indices) for scene in scenes},
}
@dataclass(frozen=True)
class Dl3dvFrame:
scene_id: str
frame_id: int
rgb_relpath: str
c2w: tuple[tuple[float, ...], ...]
raw_fxfycxcy: tuple[float, float, float, float]
distortion: tuple[float, float, float, float]
source_image_size: tuple[int, int]
def __post_init__(self) -> None:
if not isinstance(self.scene_id, str) or not self.scene_id or any(
char in self.scene_id for char in "\\/"
):
raise Dl3dvProtocolError("DL3DV frame scene_id must be a simple identifier")
if type(self.frame_id) is not int or self.frame_id < 0:
raise Dl3dvProtocolError("DL3DV frame_id must be a non-negative integer")
_frame_path(self.rgb_relpath)
source_size = self.source_image_size
if (
not isinstance(source_size, tuple)
or len(source_size) != 2
or any(type(value) is not int or value < 2 for value in source_size)
):
raise Dl3dvProtocolError(
"DL3DV source_image_size must be a tuple of integers at least two"
)
_validate_rigid_camera(np.asarray(self.c2w, dtype=np.float64), "frame c2w")
_as_float_tuple(self.raw_fxfycxcy, 4, "raw_fxfycxcy")
_as_float_tuple(self.distortion, 4, "distortion")
@property
def raw_intrinsics_matrix(self) -> np.ndarray:
return official_intrinsics(self.raw_fxfycxcy, source_size=self.source_image_size)
@property
def c2w_array(self) -> np.ndarray:
return _validate_rigid_camera(np.asarray(self.c2w, dtype=np.float64), "frame c2w")
def to_dict(self) -> dict[str, Any]:
return {
"scene_id": self.scene_id,
"frame_id": self.frame_id,
"rgb_relpath": self.rgb_relpath,
"c2w": [list(row) for row in self.c2w],
"raw_fxfycxcy": list(self.raw_fxfycxcy),
"distortion": list(self.distortion),
"source_image_size": list(self.source_image_size),
}
@classmethod
def from_dict(cls, value: Mapping[str, Any]) -> Dl3dvFrame:
return cls(
str(value["scene_id"]),
int(value["frame_id"]),
_frame_path(value["rgb_relpath"]),
tuple(tuple(float(item) for item in row) for row in value["c2w"]),
tuple(_as_float_tuple(value["raw_fxfycxcy"], 4, "raw_fxfycxcy")),
tuple(_as_float_tuple(value["distortion"], 4, "distortion")),
tuple(int(item) for item in value["source_image_size"]),
)
@dataclass(frozen=True)
class Dl3dvSceneMetadata:
scene_id: str
scene_root: Path
frames: tuple[Dl3dvFrame, ...]
input_indices: tuple[int, ...]
query_indices: tuple[int, ...]
protocol: Dl3dvProtocol
split_id: str = "test"
target_size: tuple[int, int] = DL3DV_EVAL_IMAGE_SIZE
source_schema: str = "Long-LRM/tttLRM transforms.json"
def __post_init__(self) -> None:
if not self.scene_id or any(char in self.scene_id for char in "\\/"):
raise Dl3dvProtocolError("DL3DV scene_id must be a simple non-empty identifier")
if not isinstance(self.protocol, Dl3dvProtocol):
raise Dl3dvProtocolError("DL3DV scene metadata requires a Dl3dvProtocol")
if not self.frames:
raise Dl3dvProtocolError("DL3DV scene metadata requires at least one frame")
if self.source_schema != "Long-LRM/tttLRM transforms.json":
raise Dl3dvProtocolError(
"DL3DV metadata source_schema is not the pinned transforms format"
)
rgb_paths = [frame.rgb_relpath for frame in self.frames]
if any(
not isinstance(path, str)
or not path.startswith("images_4/")
or Path(path).is_absolute()
or ".." in Path(path).parts
for path in rgb_paths
):
raise Dl3dvProtocolError("DL3DV RGB paths must stay under images_4")
if len(rgb_paths) != len(set(rgb_paths)):
raise Dl3dvProtocolError("DL3DV frames must not alias RGB paths")
for expected_id, frame in enumerate(self.frames):
if frame.scene_id != self.scene_id or frame.frame_id != expected_id:
raise Dl3dvProtocolError("DL3DV frame IDs and scene IDs must follow source order")
if frame.source_image_size != self.frames[0].source_image_size:
raise Dl3dvProtocolError("DL3DV frames must share one source image size")
_validate_rigid_camera(np.asarray(frame.c2w, dtype=np.float64), "frame c2w")
if (
not isinstance(self.target_size, tuple)
or len(self.target_size) != 2
or any(type(value) is not int or value < 2 for value in self.target_size)
):
raise Dl3dvProtocolError("DL3DV target_size must be a tuple of integers at least two")
if len(self.input_indices) != self.protocol.input_count:
raise Dl3dvProtocolError(
"DL3DV metadata input_indices must match the protocol input_count"
)
inputs, targets = _validate_scene_indices(
len(self.frames), self.input_indices, target_stride=self.protocol.fold_size
)
if tuple(self.query_indices) != targets:
raise Dl3dvProtocolError("DL3DV query_indices must be every eighth source frame")
object.__setattr__(self, "input_indices", inputs)
object.__setattr__(self, "query_indices", targets)
@property
def arrival_indices(self) -> tuple[int, ...]:
return tuple(sorted(self.input_indices))
@property
def warmup_indices(self) -> tuple[int, ...]:
return self.arrival_indices[: self.protocol.warmup_length]
@property
def stream_indices(self) -> tuple[int, ...]:
return self.arrival_indices[self.protocol.warmup_length :]
@property
def episode_id(self) -> str:
return f"dl3dv140--{self.scene_id}--{self.protocol.name}"
@property
def query_vault_id(self) -> str:
return f"dl3dv140-query--{self.scene_id}--{self.protocol.name}"
@property
def raw_source_size(self) -> tuple[int, int]:
return self.frames[0].source_image_size
def frame(self, frame_id: int) -> Dl3dvFrame:
if type(frame_id) is not int or frame_id < 0 or frame_id >= len(self.frames):
raise Dl3dvProtocolError(f"frame_id is outside scene metadata: {frame_id}")
frame = self.frames[frame_id]
if frame.frame_id != frame_id:
raise Dl3dvProtocolError("DL3DV frame IDs must follow source frame order")
return frame
def to_dict(self) -> dict[str, Any]:
return {
"scene_id": self.scene_id,
"scene_root": str(self.scene_root),
"frames": [frame.to_dict() for frame in self.frames],
"input_indices": list(self.input_indices),
"query_indices": list(self.query_indices),
"protocol": {
"mode": self.protocol.mode,
"input_count": self.protocol.input_count,
"fold_size": self.protocol.fold_size,
"warmup_length": self.protocol.warmup_length,
},
"split_id": self.split_id,
"target_size": list(self.target_size),
"source_schema": self.source_schema,
"coordinate_protocol": {
"pose_transform": "official Long-LRM c2w conversion",
"gauge": "warmup-only max baseline, frozen before stream",
"target_extent_m": DL3DV_WARMUP_TARGET_EXTENT_M,
"gauge_calibration": DL3DV_GAUGE_CALIBRATION,
"gauge_valid_rigid_train_windows": DL3DV_GAUGE_VALID_WINDOWS,
"gauge_excluded_invalid_rigid_windows": DL3DV_GAUGE_EXCLUDED_WINDOWS,
"metric_depth_known": False,
"rgb_only": True,
"local_bounds_m": [list(v) for v in DL3DV_LOCAL_BOUNDS_M],
"tttLRM_scale_difference": (
"tttLRM normalizes all selected input and target poses; this adapter uses "
"the first four warmup poses only"
),
},
}
@classmethod
def from_dict(cls, value: Mapping[str, Any]) -> Dl3dvSceneMetadata:
protocol_value = value["protocol"]
protocol = Dl3dvProtocol(
str(protocol_value["mode"]),
int(protocol_value["input_count"]),
int(protocol_value.get("fold_size", DL3DV_TARGET_STRIDE)),
int(protocol_value.get("warmup_length", DL3DV_WARMUP_LENGTH)),
)
return cls(
str(value["scene_id"]),
Path(value["scene_root"]),
tuple(Dl3dvFrame.from_dict(frame) for frame in value["frames"]),
tuple(int(item) for item in value["input_indices"]),
tuple(int(item) for item in value["query_indices"]),
protocol,
str(value.get("split_id", "test")),
tuple(int(item) for item in value.get("target_size", DL3DV_EVAL_IMAGE_SIZE)),
str(value.get("source_schema", "Long-LRM/tttLRM transforms.json")),
)
def _frame_path(value: Any) -> str:
if not isinstance(value, str) or not value:
raise Dl3dvProtocolError("DL3DV frame file_path must be a non-empty string")
normalized = value.replace("\\", "/")
path = Path(normalized)
if path.is_absolute() or any(part == ".." for part in path.parts):
raise Dl3dvProtocolError(f"unsafe DL3DV frame file_path: {value!r}")
parts = tuple(part for part in path.parts if part not in ("", "."))
if not parts or len(parts) == 1:
raise Dl3dvProtocolError(
"DL3DV frame file_path must point to an image filename"
)
# The pinned converter discards any parent directories and keeps only the
# basename under ``images_4``.
return f"images_4/{parts[-1]}"
def _scene_fxfycxcy(scene: Mapping[str, Any], frame: Mapping[str, Any]) -> tuple[float, ...]:
values = [frame.get(name, scene.get(name)) for name in ("fl_x", "fl_y", "cx", "cy")]
if any(value is None for value in values):
raise Dl3dvProtocolError("transforms.json must provide fl_x, fl_y, cx, and cy")
return _as_float_tuple(values, 4, "frame intrinsics")
def _scene_distortion(scene: Mapping[str, Any], frame: Mapping[str, Any]) -> tuple[float, ...]:
values = [frame.get(name, scene.get(name, 0.0)) for name in ("k1", "k2", "p1", "p2")]
return _as_float_tuple(values, 4, "frame distortion")
def _resolve_images4_geometry(
transforms_size: tuple[int, int], *, require_source_size: bool
) -> tuple[tuple[int, int], tuple[float, float]]:
"""Resolve the ``images_4`` size and metadata-to-pixel intrinsic scales.
DL3DV's transforms retain the original 4K dimensions while the benchmark
tree stores the level-4 images at 960x540. The public Long-LRM converter
scales intrinsics from the transform dimensions to the decoded image
dimensions before undistortion. This inference uses metadata only.
"""
height, width = transforms_size
target_height, target_width = DL3DV_SOURCE_IMAGE_SIZE
if transforms_size == DL3DV_SOURCE_IMAGE_SIZE:
return transforms_size, (1.0, 1.0)
factor = DL3DV_IMAGES4_DOWNSAMPLE
if height % factor == 0 and width % factor == 0:
source_size = (height // factor, width // factor)
if min(source_size) >= 2:
scale = 1.0 / factor
return source_size, (scale, scale)
if require_source_size:
raise Dl3dvProtocolError(
"DL3DV transforms dimensions must describe an images_4 source "
f"or a {factor}x original-resolution multiple, got {transforms_size}"
)
return transforms_size, (1.0, 1.0)
def _read_png_size_header(path: Path) -> tuple[int, int] | None:
"""Read PNG dimensions without decoding or loading image pixels."""
try:
with path.open("rb") as handle:
header = handle.read(24)
except FileNotFoundError:
return None
if len(header) < 24 or header[:8] != b"\x89PNG\r\n\x1a\n" or header[12:16] != b"IHDR":
return None
width, height = struct.unpack(">II", header[16:24])
if width < 2 or height < 2:
raise Dl3dvProtocolError(f"DL3DV PNG has invalid dimensions: {path}")
return int(height), int(width)
def prepare_scene_metadata(
scene_root: str | Path,
split_record: LongLRMSplitRecord | Mapping[str, Any],
protocol: Dl3dvProtocol,
*,
split_id: str = "test",
target_size: tuple[int, int] = DL3DV_EVAL_IMAGE_SIZE,
require_source_size: bool = False,
) -> Dl3dvSceneMetadata:
"""Prepare one scene from metadata only; no RGB file is opened."""
root = Path(scene_root)
transforms_path = root / "nerfstudio" / "transforms.json"
try:
transforms = json.loads(transforms_path.read_text(encoding="utf-8"))
except FileNotFoundError:
raise
except json.JSONDecodeError as error:
raise Dl3dvProtocolError(f"invalid DL3DV transforms JSON: {transforms_path}") from error
if not isinstance(transforms, Mapping):
raise Dl3dvProtocolError("DL3DV transforms.json root must be an object")
try:
width = int(transforms["w"])
height = int(transforms["h"])
except (KeyError, TypeError, ValueError) as error:
raise Dl3dvProtocolError("transforms.json must provide integer w and h") from error
transforms_size = (height, width)
if min(transforms_size) < 2:
raise Dl3dvProtocolError("DL3DV source image size must be at least 2x2")
source_size, intrinsic_scales = _resolve_images4_geometry(
transforms_size, require_source_size=require_source_size
)
raw_frames = transforms.get("frames")
if not isinstance(raw_frames, list) or not raw_frames:
raise Dl3dvProtocolError("transforms.json must contain a non-empty frames list")
scene_id = root.name
if isinstance(split_record, LongLRMSplitRecord):
record = split_record
else:
scene_name = split_record.get("scene_name")
arrays = {
count: tuple(split_record[f"fold_8_kmeans_{count}_input"])
for count in (16, 32, 64, 128)
if f"fold_8_kmeans_{count}_input" in split_record
}
if not isinstance(scene_name, str):
raise Dl3dvProtocolError("split_record must contain scene_name")
record = LongLRMSplitRecord(scene_name, arrays)
if record.scene_name != scene_id:
raise Dl3dvProtocolError(
f"scene folder and Long-LRM scene_name differ: {scene_id!r} != {record.scene_name!r}"
)
raw_metadata_frames: list[Dl3dvFrame] = []
for frame_id, raw_frame in enumerate(raw_frames):
if not isinstance(raw_frame, Mapping):
raise Dl3dvProtocolError(f"DL3DV frame {frame_id} must be an object")
path = _frame_path(raw_frame.get("file_path"))
frame_source_size = source_size
frame_intrinsic_scales = intrinsic_scales
header_size = _read_png_size_header(root / "nerfstudio" / Path(path))
if header_size is not None:
frame_source_size = header_size
frame_intrinsic_scales = (
frame_source_size[1] / width,
frame_source_size[0] / height,
)
c2w = official_pose_transform(raw_frame.get("transform_matrix"))
raw_intrinsics = _scene_fxfycxcy(transforms, raw_frame)
scale_x, scale_y = frame_intrinsic_scales
raw_intrinsics = (
raw_intrinsics[0] * scale_x,
raw_intrinsics[1] * scale_y,
raw_intrinsics[2] * scale_x,
raw_intrinsics[3] * scale_y,
)
raw_metadata_frames.append(
Dl3dvFrame(
scene_id,
frame_id,
path,
tuple(tuple(float(item) for item in row) for row in c2w),
raw_intrinsics,
_scene_distortion(transforms, raw_frame),
frame_source_size,
)
)
input_indices = record.input_indices(protocol.input_count)
input_indices, query_indices = _validate_scene_indices(
len(raw_metadata_frames), input_indices, target_stride=protocol.fold_size
)
if len(input_indices) < protocol.warmup_length:
raise Dl3dvProtocolError("DL3DV input count is smaller than the warmup prefix")
return Dl3dvSceneMetadata(
scene_id,
root,
tuple(raw_metadata_frames),
input_indices,
query_indices,
protocol,
split_id,
target_size,
)
@dataclass(frozen=True)
class Dl3dvBenchmarkMetadata:
scenes: tuple[Dl3dvSceneMetadata, ...]
raw_root: Path
split_path: Path
split_sha256: str
protocol: Dl3dvProtocol
split_revision: str = DL3DV_SPLIT_REVISION
def coverage(self) -> dict[str, Any]:
return validate_complete_coverage(
self.scenes,
expected_scene_names=(scene.scene_name for scene in self.split_records),
)
@property
def split_records(self) -> tuple[LongLRMSplitRecord, ...]:
return load_long_lrm_split(self.split_path)
def to_dict(self) -> dict[str, Any]:
return {
"schema": DL3DV_METADATA_SCHEMA,
"raw_root": str(self.raw_root),
"split_path": str(self.split_path),
"split_sha256": self.split_sha256,
"split_revision": self.split_revision,
"split_file_sha1": DL3DV_SPLIT_FILE_SHA1,
"protocol": {
"mode": self.protocol.mode,
"input_count": self.protocol.input_count,
"fold_size": self.protocol.fold_size,
"warmup_length": self.protocol.warmup_length,
},
"coverage": self.coverage(),
"scenes": [scene.to_dict() for scene in self.scenes],
}
def prepare_benchmark_metadata(
raw_root: str | Path,
split_path: str | Path,
protocol: Dl3dvProtocol,
*,
output_path: str | Path | None = None,
split_id: str = "test",
target_size: tuple[int, int] = DL3DV_EVAL_IMAGE_SIZE,
require_source_size: bool = True,
) -> Dl3dvBenchmarkMetadata:
"""Prepare all 140 scene manifests without touching benchmark image pixels."""
root = Path(raw_root)
split_file = Path(split_path)
split_records = load_long_lrm_split(split_file)
expected_names = {record.scene_name for record in split_records}
scenes: list[Dl3dvSceneMetadata] = []
for record in split_records:
scene_root = root / record.scene_name
if not scene_root.is_dir():
raise FileNotFoundError(f"DL3DV scene directory is missing: {scene_root}")
scenes.append(
prepare_scene_metadata(
scene_root,
record,
protocol,
split_id=split_id,
target_size=target_size,
require_source_size=require_source_size,
)
)
metadata = Dl3dvBenchmarkMetadata(
tuple(scenes),
root,
split_file,
hashlib.sha256(split_file.read_bytes()).hexdigest(),
protocol,
)
validate_complete_coverage(
metadata.scenes,
expected_scene_names=expected_names,
)
if output_path is not None:
destination = Path(output_path)
destination.parent.mkdir(parents=True, exist_ok=True)
destination.write_text(
json.dumps(metadata.to_dict(), indent=2, sort_keys=True, ensure_ascii=True) + "\n",
encoding="utf-8",
)
return metadata
def load_benchmark_metadata(path: str | Path) -> Dl3dvBenchmarkMetadata:
source = Path(path)
try:
value = json.loads(source.read_text(encoding="utf-8"))
except json.JSONDecodeError as error:
raise Dl3dvProtocolError(f"invalid DL3DV metadata JSON: {source}") from error
if not isinstance(value, Mapping) or value.get("schema") != DL3DV_METADATA_SCHEMA:
raise Dl3dvProtocolError(f"unexpected DL3DV metadata schema: {source}")
split_revision = value.get("split_revision")
if split_revision != DL3DV_SPLIT_REVISION:
raise Dl3dvProtocolError(
"DL3DV metadata split_revision is not the pinned Long-LRM revision: "
f"expected={DL3DV_SPLIT_REVISION}, actual={split_revision}"
)
stored_split_hash = value.get("split_sha256")
if stored_split_hash != DL3DV_SPLIT_SHA256:
raise Dl3dvProtocolError(
"DL3DV metadata split_sha256 is not the pinned split digest: "
f"expected={DL3DV_SPLIT_SHA256}, actual={stored_split_hash}"
)
stored_split_sha1 = value.get("split_file_sha1", DL3DV_SPLIT_FILE_SHA1)
if stored_split_sha1 != DL3DV_SPLIT_FILE_SHA1:
raise Dl3dvProtocolError(
"DL3DV metadata split_file_sha1 is not the pinned Git blob identity: "
f"expected={DL3DV_SPLIT_FILE_SHA1}, actual={stored_split_sha1}"
)
protocol_value = value.get("protocol")
if not isinstance(protocol_value, Mapping):
raise Dl3dvProtocolError("DL3DV metadata protocol must be an object")
protocol = Dl3dvProtocol(
str(protocol_value["mode"]),
int(protocol_value["input_count"]),
int(protocol_value.get("fold_size", DL3DV_TARGET_STRIDE)),
int(protocol_value.get("warmup_length", DL3DV_WARMUP_LENGTH)),
)
try:
split_path = Path(value["split_path"])
raw_root = Path(value["raw_root"])
raw_scene_values = value["scenes"]
except (KeyError, TypeError, ValueError) as error:
raise Dl3dvProtocolError("DL3DV metadata is missing required paths or scenes") from error
if not split_path.is_file():
raise FileNotFoundError(f"DL3DV split file is missing: {split_path}")
actual_split_hash = hashlib.sha256(split_path.read_bytes()).hexdigest()
if actual_split_hash != stored_split_hash or actual_split_hash != DL3DV_SPLIT_SHA256:
raise Dl3dvProtocolError("DL3DV split file does not match the pinned SHA256")
split_records = load_long_lrm_split(split_path)
records_by_scene = {record.scene_name: record for record in split_records}
if not isinstance(raw_scene_values, list):
raise Dl3dvProtocolError("DL3DV metadata scenes must be a list")
scenes = tuple(Dl3dvSceneMetadata.from_dict(item) for item in raw_scene_values)
if protocol != Dl3dvProtocol(
str(protocol_value["mode"]),
int(protocol_value["input_count"]),
int(protocol_value.get("fold_size", DL3DV_TARGET_STRIDE)),
int(protocol_value.get("warmup_length", DL3DV_WARMUP_LENGTH)),
):
raise Dl3dvProtocolError("DL3DV metadata protocol is internally inconsistent")
for scene in scenes:
record = records_by_scene.get(scene.scene_id)
if record is None:
raise Dl3dvProtocolError(
f"DL3DV metadata scene is absent from the pinned split: {scene.scene_id}"
)
expected_root = (raw_root / scene.scene_id).resolve(strict=False)
if Path(scene.scene_root).resolve(strict=False) != expected_root:
raise Dl3dvProtocolError(
f"DL3DV scene_root does not match raw_root for {scene.scene_id}"
)
if scene.protocol != protocol:
raise Dl3dvProtocolError("DL3DV metadata scenes do not share the top-level protocol")
if scene.input_indices != record.input_indices(protocol.input_count):
raise Dl3dvProtocolError(
f"DL3DV input_indices differ from the pinned split for {scene.scene_id}"
)
if scene.query_indices != target_frame_indices(len(scene.frames), protocol.fold_size):
raise Dl3dvProtocolError(
f"DL3DV query_indices differ from the pinned target rule for {scene.scene_id}"
)
metadata = Dl3dvBenchmarkMetadata(
scenes,
raw_root,
split_path,
str(stored_split_hash),
protocol,
str(split_revision),
)
validate_complete_coverage(
metadata.scenes,
expected_scene_names=records_by_scene,
)
return metadata
@dataclass(frozen=True)
class WarmupGauge:
"""Frozen scale and first-camera frame for one causal episode."""
target_extent_m: float
observed_max_baseline: float
scale: float
anchor_c2w: np.ndarray
def __post_init__(self) -> None:
for name, value in (
("target_extent_m", self.target_extent_m),
("observed_max_baseline", self.observed_max_baseline),
("scale", self.scale),
):
if not math.isfinite(float(value)) or float(value) <= 0.0:
raise Dl3dvProtocolError(f"warmup gauge {name} must be finite and positive")
object.__setattr__(
self,
"anchor_c2w",
_validate_rigid_camera(
np.asarray(self.anchor_c2w, dtype=np.float64), "gauge anchor"
).copy(),
)
def transform(self, c2w: np.ndarray) -> np.ndarray:
"""Express a pose in the first warmup frame and apply frozen translation scale."""
relative = np.linalg.solve(self.anchor_c2w, _validate_rigid_camera(c2w, "c2w")).copy()
relative[:3, 3] *= self.scale
return _validate_homogeneous(relative, "gauged c2w")
def to_dict(self) -> dict[str, Any]:
return {
"schema": DL3DV_GAUGE_SCHEMA,
"target_extent_m": self.target_extent_m,
"calibration": DL3DV_GAUGE_CALIBRATION,
"calibration_valid_rigid_train_windows": DL3DV_GAUGE_VALID_WINDOWS,
"calibration_excluded_invalid_rigid_windows": DL3DV_GAUGE_EXCLUDED_WINDOWS,
"observed_max_baseline": self.observed_max_baseline,
"scale": self.scale,
"anchor_c2w": self.anchor_c2w.tolist(),
"metric_depth_known": False,
"rgb_only": True,
}
def compute_warmup_gauge(
c2ws: Sequence[np.ndarray],
*,
target_extent_m: float = DL3DV_WARMUP_TARGET_EXTENT_M,
minimum_baseline: float = DL3DV_MIN_WARMUP_BASELINE,
) -> WarmupGauge:
"""Compute the frozen four-view baseline gauge without future/query poses."""
if len(c2ws) != DL3DV_WARMUP_LENGTH:
raise Dl3dvProtocolError("warmup gauge requires exactly four prefix poses")
if not math.isfinite(float(target_extent_m)) or float(target_extent_m) <= 0.0:
raise Dl3dvProtocolError("target_extent_m must be finite and positive")
if not math.isfinite(float(minimum_baseline)) or float(minimum_baseline) <= 0.0:
raise Dl3dvProtocolError("minimum_baseline must be finite and positive")
poses = [_validate_rigid_camera(np.asarray(value), "warmup c2w") for value in c2ws]
first_center = poses[0][:3, 3]
observed = max(float(np.linalg.norm(pose[:3, 3] - first_center)) for pose in poses)
if observed < minimum_baseline:
raise Dl3dvProtocolError(
f"warmup camera baseline is too small for a stable gauge: {observed:.3e}"
)
scale = float(target_extent_m) / observed
return WarmupGauge(float(target_extent_m), observed, scale, poses[0])
class Dl3dvOnlineObservationSource:
"""Causal DL3DV observations with a post-seal lazy query callback seam."""
def __init__(
self,
scene: Dl3dvSceneMetadata,
*,
device: str | torch.device = "cpu",
target_extent_m: float = DL3DV_WARMUP_TARGET_EXTENT_M,
minimum_baseline: float = DL3DV_MIN_WARMUP_BASELINE,
) -> None:
self.metadata = scene
self.device = torch.device(device)
warmup_frames = [scene.frame(index) for index in scene.warmup_indices]
self.gauge = compute_warmup_gauge(
[frame.c2w_array for frame in warmup_frames],
target_extent_m=target_extent_m,
minimum_baseline=minimum_baseline,
)
self._warmup_cache: tuple[OnlineObservation, ...] | None = None
self._stream_cursor = 0
self._sealed: SealedScene | None = None
self._query_cache: dict[int, OnlineObservation] = {}
@property
def episode_id(self) -> str:
return self.metadata.episode_id
@property
def scene_id(self) -> str:
return self.metadata.scene_id
@property
def split_id(self) -> str:
return self.metadata.split_id
@property
def query_vault_id(self) -> str:
return self.metadata.query_vault_id
@property
def image_size(self) -> tuple[int, int]:
return self.metadata.target_size
@property
def stream_steps(self) -> int:
return len(self.metadata.stream_indices)
@property
def query_count(self) -> int:
return len(self.metadata.query_indices)
@property
def remaining_steps(self) -> int:
return self.stream_steps - self._stream_cursor
@property
def sealed(self) -> bool:
return self._sealed is not None
def warmup(self) -> tuple[OnlineObservation, ...]:
if self._sealed is not None:
raise ValueError("DL3DV source is sealed")
if self._warmup_cache is None:
self._warmup_cache = tuple(
self._load_observation(self.metadata.frame(index))
for index in self.metadata.warmup_indices
)
return self._warmup_cache
def next_observation(self) -> OnlineObservation:
if self._sealed is not None:
raise ValueError("DL3DV source is sealed")
if self._stream_cursor >= self.stream_steps:
raise StopIteration("DL3DV online stream is exhausted")
frame = self.metadata.frame(self.metadata.stream_indices[self._stream_cursor])
self._stream_cursor += 1
return self._load_observation(frame)
def _load_observation(self, frame: Dl3dvFrame) -> OnlineObservation:
image_path = self.metadata.scene_root / "nerfstudio" / Path(frame.rgb_relpath)
rgb, intrinsic = preprocess_dl3dv_rgb(
image_path,
frame,
target_size=self.metadata.target_size,
)
c2w = self.gauge.transform(frame.c2w_array)
camera = Cameras(
torch.from_numpy(intrinsic).to(self.device, dtype=torch.float32),
torch.from_numpy(c2w).to(self.device, dtype=torch.float32),
self.metadata.target_size,
)
return OnlineObservation(
self.metadata.scene_id,
frame.frame_id,
rgb.to(self.device, dtype=torch.float32),
camera,
)
def queries_for_sealed(self, sealed: SealedScene) -> tuple[RGBQuery, ...]:
"""Return lazy target callbacks after the runner has sealed this episode."""
if not isinstance(sealed, SealedScene):
raise TypeError("queries_for_sealed requires a SealedScene")
if self._sealed is not None:
raise ValueError("DL3DV query callbacks were already opened")
if self._stream_cursor != self.stream_steps:
raise ValueError("DL3DV query callbacks require the online stream to be exhausted")
expected_observed = tuple(self.metadata.arrival_indices)
if tuple(sealed.observed_ids) != expected_observed:
raise ValueError("sealed observed_ids do not match DL3DV context arrival order")
if (
sealed.episode_id != self.episode_id
or sealed.scene_id != self.scene_id
or sealed.split_id != self.split_id
or sealed.query_vault_id != self.query_vault_id
):
raise ValueError("sealed scene identity does not match DL3DV metadata")
self._sealed = sealed
result: list[RGBQuery] = []
for frame_id in self.metadata.query_indices:
result.append(
RGBQuery.for_sealed(
sealed,
frame_id,
lambda frame_id=frame_id: self._query_observation(frame_id).camera,
lambda frame_id=frame_id: self._query_observation(frame_id).rgb,
self.metadata.target_size,
)
)
return tuple(result)
# Explicit alias for callers that use the source/evaluator terminology.
open_query_callbacks = queries_for_sealed
def _query_observation(self, frame_id: int) -> OnlineObservation:
if self._sealed is None:
raise RuntimeError("DL3DV query supplier is unavailable before seal")
if frame_id not in self.metadata.query_indices:
raise Dl3dvProtocolError(f"frame {frame_id} is not a DL3DV query frame")
if frame_id not in self._query_cache:
self._query_cache[frame_id] = self._load_observation(self.metadata.frame(frame_id))
return self._query_cache[frame_id]
# Compatibility names make the boundary easy to discover without changing the
# existing generic OnlineEpisodeSource contract.
DL3DVProtocol = Dl3dvProtocol
DL3DVSceneMetadata = Dl3dvSceneMetadata
DL3DVOnlineObservationSource = Dl3dvOnlineObservationSource
Dl3dvEpisodeSource = Dl3dvOnlineObservationSource
DL3DVBenchmarkMetadata = Dl3dvBenchmarkMetadata
prepare_dl3dv_metadata = prepare_benchmark_metadata
__all__ = [
"DL3DV_EVAL_IMAGE_SIZE",
"DL3DV_GAUGE_CALIBRATION",
"DL3DV_GAUGE_EXCLUDED_WINDOWS",
"DL3DV_GAUGE_SCHEMA",
"DL3DV_GAUGE_VALID_WINDOWS",
"DL3DV_IMAGES4_DOWNSAMPLE",
"DL3DV_LOCAL_BOUNDS_M",
"DL3DV_METADATA_SCHEMA",
"DL3DV_MIN_WARMUP_BASELINE",
"DL3DV_RIGID_ROTATION_TOLERANCE",
"DL3DV_SCENE_COUNT",
"DL3DV_SOURCE_IMAGE_SIZE",
"DL3DV_SPLIT_REVISION",
"DL3DV_SPLIT_FILE_SHA1",
"DL3DV_SPLIT_SHA256",
"DL3DV_TARGET_STRIDE",
"DL3DV_WARMUP_LENGTH",
"DL3DV_WARMUP_TARGET_EXTENT_M",
"Dl3dvBenchmarkMetadata",
"DL3DVOnlineObservationSource",
"DL3DVProtocol",
"DL3DVSceneMetadata",
"DL3DVBenchmarkMetadata",
"Dl3dvEpisodeSource",
"Dl3dvFrame",
"Dl3dvOnlineObservationSource",
"Dl3dvProtocol",
"Dl3dvProtocolError",
"Dl3dvSceneMetadata",
"LongLRMSplitRecord",
"WarmupGauge",
"compute_warmup_gauge",
"load_benchmark_metadata",
"load_long_lrm_split",
"official_intrinsics",
"official_pose_transform",
"preprocess_dl3dv_rgb",
"prepare_benchmark_metadata",
"prepare_dl3dv_metadata",
"prepare_scene_metadata",
"resize_and_center_crop",
"target_frame_indices",
"undistorted_intrinsics",
"validate_complete_coverage",
]