src/mcss/data/manifest.py
"""Versioned on-disk scene manifests independent of source dataset layout."""
from __future__ import annotations
import json
from dataclasses import dataclass
from pathlib import Path, PurePosixPath
from typing import Any
import numpy as np
SCHEMA_VERSION = 1
@dataclass(frozen=True)
class ManifestFrame:
frame_id: int
rgb: str
intrinsics: list[list[float]]
c2w: list[list[float]]
depth: str | None = None
normal: str | None = None
point: str | None = None
visibility: str | None = None
def to_dict(self) -> dict[str, Any]:
result: dict[str, Any] = {
"frame_id": self.frame_id,
"rgb": self.rgb,
"intrinsics": self.intrinsics,
"c2w": self.c2w,
}
for name in ("depth", "normal", "point", "visibility"):
value = getattr(self, name)
if value is not None:
result[name] = value
return result
@dataclass(frozen=True)
class SceneManifest:
scene_id: str
image_size: tuple[int, int]
bounds: tuple[tuple[float, float, float], tuple[float, float, float]]
frames: tuple[ManifestFrame, ...]
root: Path | None = None
coordinate_convention: str = "opencv_c2w"
def to_dict(self) -> dict[str, Any]:
return {
"schema_version": SCHEMA_VERSION,
"scene_id": self.scene_id,
"coordinate_convention": self.coordinate_convention,
"image_size": list(self.image_size),
"bounds": [list(self.bounds[0]), list(self.bounds[1])],
"frames": [frame.to_dict() for frame in self.frames],
}
@classmethod
def from_dict(cls, data: dict[str, Any], root: str | Path) -> SceneManifest:
if data.get("schema_version") != SCHEMA_VERSION:
raise ValueError(f"expected manifest schema_version {SCHEMA_VERSION}")
scene_id = data.get("scene_id")
if not isinstance(scene_id, str) or not scene_id or any(char in scene_id for char in "\\/"):
raise ValueError("scene_id must be a simple non-empty name")
if data.get("coordinate_convention") != "opencv_c2w":
raise ValueError("manifest coordinate_convention must be opencv_c2w")
image_size = _parse_image_size(data.get("image_size"))
bounds = _parse_bounds(data.get("bounds"))
raw_frames = data.get("frames")
if not isinstance(raw_frames, list) or not raw_frames:
raise ValueError("manifest frames must be a non-empty list")
root_path = Path(root).resolve()
frames = tuple(_parse_frame(frame, root_path) for frame in raw_frames)
frame_ids = [frame.frame_id for frame in frames]
if len(set(frame_ids)) != len(frame_ids):
raise ValueError("manifest frame_id values must be unique")
return cls(
scene_id=scene_id,
image_size=image_size,
bounds=bounds,
frames=frames,
root=root_path,
)
def save_manifest(manifest: SceneManifest, path: str | Path) -> Path:
output = Path(path)
output.parent.mkdir(parents=True, exist_ok=True)
output.write_text(
json.dumps(manifest.to_dict(), indent=2, sort_keys=True) + "\n", encoding="utf-8"
)
return output
def load_manifest(path: str | Path) -> SceneManifest:
manifest_path = Path(path)
try:
data = json.loads(manifest_path.read_text(encoding="utf-8"))
except json.JSONDecodeError as error:
raise ValueError(f"invalid manifest JSON: {manifest_path}") from error
if not isinstance(data, dict):
raise ValueError("manifest JSON root must be an object")
return SceneManifest.from_dict(data, manifest_path.parent)
def manifest_relative_path(root: str | Path, value: str) -> Path:
"""Resolve a manifest reference while forbidding absolute and escaping paths."""
if not isinstance(value, str) or not value:
raise ValueError("manifest paths must be non-empty strings")
normalized = value.replace("\\", "/")
pure = PurePosixPath(normalized)
if pure.is_absolute() or any(part in {"", ".", ".."} for part in pure.parts):
raise ValueError("manifest paths must be safe relative paths")
if len(pure.parts[0]) >= 2 and pure.parts[0][1] == ":":
raise ValueError("manifest paths must be relative, not drive paths")
root_path = Path(root).resolve()
result = (root_path / Path(*pure.parts)).resolve()
try:
result.relative_to(root_path)
except ValueError as error:
raise ValueError("manifest paths must be safe relative paths") from error
return result
def _parse_frame(data: Any, root: Path) -> ManifestFrame:
if not isinstance(data, dict):
raise ValueError("every manifest frame must be an object")
frame_id = data.get("frame_id")
if not isinstance(frame_id, int) or frame_id < 0:
raise ValueError("frame_id must be a non-negative integer")
rgb = _parse_path(data.get("rgb"), root)
optional = {
name: _parse_optional_path(data.get(name), root)
for name in ("depth", "normal", "point", "visibility")
}
intrinsics = _parse_matrix(data.get("intrinsics"), (3, 3), "intrinsics")
c2w = _parse_matrix(data.get("c2w"), (4, 4), "c2w")
if not np.allclose(np.asarray(c2w)[3], [0.0, 0.0, 0.0, 1.0], atol=1e-5):
raise ValueError("c2w must have a homogeneous final row")
return ManifestFrame(frame_id, rgb, intrinsics, c2w, **optional)
def _parse_path(value: Any, root: Path) -> str:
manifest_relative_path(root, value)
return value.replace("\\", "/")
def _parse_optional_path(value: Any, root: Path) -> str | None:
if value is None:
return None
return _parse_path(value, root)
def _parse_matrix(value: Any, shape: tuple[int, int], name: str) -> list[list[float]]:
array = np.asarray(value, dtype=np.float64)
if array.shape != shape or not np.isfinite(array).all():
raise ValueError(f"{name} must be a finite {shape[0]}x{shape[1]} matrix")
return array.astype(float).tolist()
def _parse_image_size(value: Any) -> tuple[int, int]:
if (
not isinstance(value, list)
or len(value) != 2
or not all(isinstance(item, int) for item in value)
):
raise ValueError("image_size must be [height, width]")
height, width = value
if height < 2 or width < 2:
raise ValueError("image_size values must be at least two")
return height, width
def _parse_bounds(value: Any) -> tuple[tuple[float, float, float], tuple[float, float, float]]:
array = np.asarray(value, dtype=np.float64)
if array.shape != (2, 3) or not np.isfinite(array).all() or not np.all(array[1] > array[0]):
raise ValueError("bounds must be finite [[minx, miny, minz], [maxx, maxy, maxz]]")
return tuple(map(tuple, array.astype(float))) # type: ignore[return-value]