tests/test_hypersim_expansion.py
import csv
from pathlib import Path
import pytest
from mcss.config import load_config
from mcss.data.hypersim_expansion import build_expansion_plan, select_materialization_scenes
def _write_csv(
path: Path,
fieldnames: list[str],
rows: list[dict[str, object]],
*,
encoding: str = "utf-8",
) -> None:
with path.open("w", newline="", encoding=encoding) as handle:
writer = csv.DictWriter(handle, fieldnames=fieldnames)
writer.writeheader()
writer.writerows(rows)
def test_expansion_plan_preserves_official_splits_and_seals_unseen_test_scenes(
tmp_path: Path,
) -> None:
metadata = tmp_path / "metadata.csv"
subset = tmp_path / "observed.csv"
metadata_rows: list[dict[str, object]] = []
for scene, split, cameras in (
("ai_001_001", "train", {"cam_00": 3}),
("ai_001_002", "train", {"cam_00": 2}),
("ai_002_001", "val", {"cam_00": 2}),
("ai_003_001", "test", {"cam_00": 2}),
("ai_003_002", "test", {"cam_00": 2}),
("ai_003_003", "test", {"cam_02": 3, "cam_01": 2}),
):
for camera, frame_count in cameras.items():
for frame_id in range(frame_count):
metadata_rows.append(
{
"scene_name": scene,
"camera_name": camera,
"frame_id": frame_id,
"included_in_public_release": "True",
"exclude_reason": "",
"split_partition_name": split,
}
)
metadata_rows.append(
{
"scene_name": "ai_999_999",
"camera_name": "cam_00",
"frame_id": 0,
"included_in_public_release": "False",
"exclude_reason": "not public",
"split_partition_name": "train",
}
)
_write_csv(
metadata,
[
"scene_name",
"camera_name",
"frame_id",
"included_in_public_release",
"exclude_reason",
"split_partition_name",
],
metadata_rows,
)
_write_csv(
subset,
["scene_name", "split_partition_name", "frame_count"],
[
{
"scene_name": "ai_001_001",
"split_partition_name": "train",
"frame_count": 3,
},
{
"scene_name": "ai_003_001",
"split_partition_name": "test",
"frame_count": 2,
},
],
encoding="utf-8-sig",
)
plan = build_expansion_plan(metadata, subset)
assert plan.counts == {"train": 2, "val": 1, "diagnostic_test": 1, "final_holdout": 2}
by_scene = {row.scene_name: row for row in plan.scenes}
assert by_scene["ai_001_001"].protocol_partition == "train"
assert by_scene["ai_002_001"].protocol_partition == "val"
assert by_scene["ai_003_001"].protocol_partition == "diagnostic_test"
assert by_scene["ai_003_002"].protocol_partition == "final_holdout"
assert by_scene["ai_003_003"].selected_camera == "cam_02"
assert by_scene["ai_003_003"].frame_count == 3
assert {row.scene_name for row in plan.exclusions} == {"ai_003_001"}
assert not ({row.scene_name for row in plan.exclusions} & plan.final_holdout_scenes)
def test_expansion_plan_rejects_subset_split_drift(tmp_path: Path) -> None:
metadata = tmp_path / "metadata.csv"
subset = tmp_path / "observed.csv"
_write_csv(
metadata,
[
"scene_name",
"camera_name",
"frame_id",
"included_in_public_release",
"exclude_reason",
"split_partition_name",
],
[
{
"scene_name": "ai_001_001",
"camera_name": "cam_00",
"frame_id": 0,
"included_in_public_release": "True",
"exclude_reason": "",
"split_partition_name": "train",
}
],
)
_write_csv(
subset,
["scene_name", "split_partition_name", "frame_count"],
[
{
"scene_name": "ai_001_001",
"split_partition_name": "test",
"frame_count": 1,
}
],
)
with pytest.raises(ValueError, match="split mismatch"):
build_expansion_plan(metadata, subset)
def test_evidence_residual_configs_are_protocol_separated() -> None:
root = Path(__file__).parents[1]
train = load_config(root / "configs/hypersim_er_train.yaml")
validation = load_config(root / "configs/hypersim_er_val.yaml")
diagnostic = load_config(root / "configs/hypersim_er_diagnostic_test.yaml")
final_holdout = load_config(root / "configs/hypersim_er_final_holdout.yaml")
assert train.dataset.root == "data/hypersim_er_prepared/train"
assert validation.dataset.root == "data/hypersim_er_prepared/val"
assert diagnostic.dataset.root == "data/hypersim_er_prepared/diagnostic_test"
assert final_holdout.dataset.root == "data/hypersim_er_prepared/final_holdout"
assert train.model.state_architecture == "evidence_residual"
assert train.dataset.context_views == 4
assert train.dataset.target_views == 2
assert train.dataset.sample_stride == 2
assert train.training.max_steps == 150_000
assert train.training.gradient_accumulation == 2
assert validation.model == train.model
assert diagnostic.model == train.model
assert final_holdout.model == train.model
def test_v5_appearance_configs_preserve_data_protocol_and_isolate_outputs() -> None:
root = Path(__file__).parents[1]
train = load_config(root / "configs/hypersim_er_v5_appearance2x_train.yaml")
validation = load_config(root / "configs/hypersim_er_v5_appearance2x_val.yaml")
overfit = load_config(root / "configs/hypersim_er_v5_appearance2x_overfit.yaml")
probe = load_config(root / "configs/hypersim_er_v5_appearance2x_probe.yaml")
assert train.dataset.root == "data/hypersim_er_prepared/train"
assert validation.dataset.root == "data/hypersim_er_prepared/val"
assert train.model == validation.model
assert train.model.appearance_resolution_scale == 2
assert train.model.voxel_resolution == (48, 32, 48)
assert train.training.max_steps == 150_000
assert overfit.dataset.length == 1
assert overfit.training.max_steps == 5_000
assert probe.training.max_steps == 200
assert all(
"hypersim_er_v5_appearance2x" in config.training.output_dir
for config in (train, validation, overfit, probe)
)
def test_materialization_requires_explicit_final_holdout_release(tmp_path: Path) -> None:
metadata = tmp_path / "metadata.csv"
subset = tmp_path / "observed.csv"
_write_csv(
metadata,
[
"scene_name",
"camera_name",
"frame_id",
"included_in_public_release",
"exclude_reason",
"split_partition_name",
],
[
{
"scene_name": "ai_001_001",
"camera_name": "cam_00",
"frame_id": 0,
"included_in_public_release": "True",
"exclude_reason": "",
"split_partition_name": "train",
},
{
"scene_name": "ai_002_001",
"camera_name": "cam_00",
"frame_id": 0,
"included_in_public_release": "True",
"exclude_reason": "",
"split_partition_name": "test",
},
],
)
_write_csv(subset, ["scene_name", "split_partition_name"], [])
plan = build_expansion_plan(metadata, subset)
selected = select_materialization_scenes(plan, ("train",))
assert [scene.scene_name for scene in selected] == ["ai_001_001"]
with pytest.raises(ValueError, match="sealed"):
select_materialization_scenes(plan, ("final_holdout",))
released = select_materialization_scenes(
plan, ("final_holdout",), allow_final_holdout=True
)
assert [scene.scene_name for scene in released] == ["ai_002_001"]