tests/test_dense_geometry_audit.py
import json
from pathlib import Path
import numpy as np
import pytest
import torch
from PIL import Image
from mcss.dense_geometry_audit import (
DenseAuditDecisionThresholds,
DenseGeometryAuditConfig,
_evaluate_camera_shuffled_control,
_spearman,
decide_dense_audit,
load_dense_audit_config,
materialize_and_evaluate_window,
render_dense_report,
run_dense_audit,
validate_development_train_root,
write_dense_audit_finalization,
)
from mcss.dense_geometry_certificates import DenseCertificateConfig
from mcss.geometry_supplier import (
CodeDependency,
GeometrySupplierRequest,
GeometrySupplierResult,
SupplierIdentity,
SupplierRunConfig,
)
from mcss.types import Cameras
def _passing_metrics() -> dict[str, float | int]:
return {
"window_count": 32,
"certified_coverage_macro": 0.20,
"nonzero_window_fraction": 0.90,
"geometry_error_median_m": 0.10,
"geometry_error_p90_m": 0.30,
"matched_confidence_median_error_m": 0.20,
"risk_error_spearman": 0.20,
"permutation_max_depth_delta_m": 0.0,
"shuffled_certified_coverage_macro": 0.05,
"shuffled_geometry_error_median_m": 1.0,
}
def test_dense_decision_requires_every_preregistered_gate() -> None:
thresholds = DenseAuditDecisionThresholds()
passing = _passing_metrics()
assert decide_dense_audit(passing, thresholds)["status"] == "GO_TO_STATE_INTEGRATION"
failures = {
"window_count": 31,
"certified_coverage_macro": 0.10,
"nonzero_window_fraction": 0.50,
"geometry_error_median_m": 0.30,
"geometry_error_p90_m": 0.60,
"matched_confidence_median_error_m": 0.05,
"risk_error_spearman": -0.1,
"permutation_max_depth_delta_m": 0.01,
}
for key, value in failures.items():
metrics = dict(passing)
metrics[key] = value
expected = "INCONCLUSIVE" if key == "window_count" else "STOP_MAPANYTHING_SUPPLIER"
assert decide_dense_audit(metrics, thresholds)["status"] == expected
no_shuffle_degradation = dict(passing)
no_shuffle_degradation["shuffled_certified_coverage_macro"] = 0.19
no_shuffle_degradation["shuffled_geometry_error_median_m"] = 0.12
assert (
decide_dense_audit(no_shuffle_degradation, thresholds)["status"]
== "STOP_MAPANYTHING_SUPPLIER"
)
def test_dense_decision_is_inconclusive_when_endpoint_is_missing() -> None:
metrics = _passing_metrics()
del metrics["risk_error_spearman"]
decision = decide_dense_audit(metrics, DenseAuditDecisionThresholds())
assert decision["status"] == "INCONCLUSIVE"
assert decision["missing"] == ["risk_error_spearman"]
def test_zero_coverage_shuffle_passes_control_without_defined_error() -> None:
metrics = _passing_metrics()
metrics["shuffled_certified_coverage_macro"] = 0.0
metrics["shuffled_geometry_error_median_m"] = None
decision = decide_dense_audit(metrics, DenseAuditDecisionThresholds())
assert decision["status"] == "GO_TO_STATE_INTEGRATION"
assert decision["gates"]["shuffled_control_degrades"] is True
def test_spearman_uses_average_ranks_for_ties() -> None:
assert _spearman([1.0, 1.0, 1.0], [1.0, 2.0, 3.0]) is None
assert _spearman([1.0, 1.0, 2.0], [1.0, 2.0, 3.0]) == pytest.approx(0.8660254)
def test_audit_root_rejects_diagnostic_and_final_partitions(tmp_path: Path) -> None:
train = tmp_path / "train"
train.mkdir()
assert validate_development_train_root(train) == train.resolve()
for name in ("diagnostic_test", "final_holdout", "val"):
root = tmp_path / name
root.mkdir()
with pytest.raises(ValueError, match="development train"):
validate_development_train_root(root)
def test_window_pipeline_opens_depth_only_after_both_caches_are_sealed(tmp_path: Path) -> None:
intrinsics = torch.tensor(
[[[4.0, 0.0, 2.0], [0.0, 4.0, 2.0], [0.0, 0.0, 1.0]]] * 2
)
cameras = Cameras(intrinsics, torch.eye(4).repeat(2, 1, 1), (5, 5))
request = GeometrySupplierRequest((0, 1), torch.ones(2, 3, 5, 5), cameras)
result = GeometrySupplierResult(
frame_ids=(0, 1),
depth_along_ray=np.ones((2, 5, 5), dtype=np.float32) * 2.0,
confidence=np.ones((2, 5, 5), dtype=np.float32),
valid_mask=np.ones((2, 5, 5), dtype=np.bool_),
intrinsics=intrinsics.numpy(),
c2w=cameras.c2w.numpy(),
)
class Supplier:
def infer(self, supplied_request):
assert not hasattr(supplied_request, "depth")
return result
events = []
def load_depth():
assert (tmp_path / "window" / "supplier" / "metadata.json").is_file()
assert (tmp_path / "window" / "certificates" / "metadata.json").is_file()
events.append("depth")
return np.ones((2, 5, 5), dtype=np.float32) * 2.0, [tmp_path / "depth.npy"]
evaluation = materialize_and_evaluate_window(
tmp_path / "window",
request=request,
supplier=Supplier(),
identity=SupplierIdentity("model", "revision", "repo", "code"),
run_config=SupplierRunConfig(),
certificate_config=DenseCertificateConfig(grid_stride=2),
depth_loader=load_depth,
)
assert events == ["depth"]
assert evaluation["certificate_before_depth"] is True
assert evaluation["geometry_error_median_m"] == pytest.approx(0.0)
assert evaluation["accessed_depth_paths"] == [str(tmp_path / "depth.npy")]
def test_window_pipeline_reuses_hash_validated_sealed_caches_after_interruption(
tmp_path: Path,
) -> None:
intrinsics = torch.tensor(
[[[4.0, 0.0, 2.0], [0.0, 4.0, 2.0], [0.0, 0.0, 1.0]]] * 2
)
cameras = Cameras(intrinsics, torch.eye(4).repeat(2, 1, 1), (5, 5))
request = GeometrySupplierRequest((0, 1), torch.ones(2, 3, 5, 5), cameras)
calls = []
class Supplier:
def infer(self, supplied_request):
calls.append(1)
return GeometrySupplierResult(
supplied_request.frame_ids,
np.ones((2, 5, 5), dtype=np.float32) * 2.0,
np.ones((2, 5, 5), dtype=np.float32),
np.ones((2, 5, 5), dtype=np.bool_),
supplied_request.cameras.intrinsics.numpy(),
supplied_request.cameras.c2w.numpy(),
)
kwargs = {
"request": request,
"supplier": Supplier(),
"identity": SupplierIdentity("model", "revision", "repo", "code"),
"run_config": SupplierRunConfig(),
"certificate_config": DenseCertificateConfig(grid_stride=2),
"depth_loader": lambda: (
np.ones((2, 5, 5), dtype=np.float32) * 2.0,
[tmp_path / "depth.npy"],
),
}
materialize_and_evaluate_window(tmp_path / "window", **kwargs)
materialize_and_evaluate_window(tmp_path / "window", **kwargs)
assert len(calls) == 1
def test_config_parser_rejects_unknown_keys_and_loads_frozen_contract(tmp_path: Path) -> None:
config_path = tmp_path / "audit.yaml"
config_path.write_text(
"""
seed: 17
dataset_root: data/train
frozen_windows: outputs/old/frozen_windows.json
output_dir: outputs/new
context_views: 4
permutation_windows: 2
supplier:
model_id: facebook/map-anything
model_revision: model-revision
code_repository: facebookresearch/map-anything
code_revision: code-revision
dependencies:
- name: dinov2
repository: facebookresearch/dinov2
revision: dinov2-revision
source_sha256: aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa
code_root: third_party/map-anything
dinov2_code_root: third_party/dinov2
device: cuda
model_cache_dir: outputs/model_cache/huggingface
memory_efficient_inference: true
minibatch_size: 1
use_amp: true
amp_dtype: bf16
apply_mask: true
mask_edges: true
apply_confidence_mask: false
use_multiview_confidence: false
certificate:
grid_stride: 4
min_depth_m: 0.1
max_depth_m: 20.0
depth_absolute_tolerance_m: 0.1
depth_relative_tolerance: 0.05
roundtrip_threshold_px: 2.0
surface_half_width_m: 0.05
free_space_margin_m: 0.05
minimum_supporting_views: 1
maximum_conflicting_views: 0
decision:
minimum_windows: 32
minimum_certified_coverage: 0.15
minimum_nonzero_window_fraction: 0.8
maximum_median_error_m: 0.25
maximum_p90_error_m: 0.5
maximum_permutation_delta_m: 0.0001
minimum_shuffled_error_ratio: 1.5
minimum_shuffled_error_margin_m: 0.05
maximum_shuffled_coverage_ratio: 0.5
minimum_risk_error_spearman: 0.0
""".strip()
+ "\n",
encoding="utf-8",
)
config = load_dense_audit_config(config_path)
assert isinstance(config, DenseGeometryAuditConfig)
assert config.supplier_identity.model_revision == "model-revision"
assert config.supplier_identity.dependencies == (
CodeDependency(
"dinov2", "facebookresearch/dinov2", "dinov2-revision", "a" * 64
),
)
assert config.dinov2_code_root == "third_party/dinov2"
assert config.run_config.minibatch_size == 1
assert config.certificate.grid_stride == 4
config_path.write_text(config_path.read_text(encoding="utf-8") + "unknown: true\n")
with pytest.raises(ValueError, match="unknown"):
load_dense_audit_config(config_path)
def test_fake_end_to_end_runner_reuses_exact_windows_and_writes_terminal_artifacts(
tmp_path: Path,
) -> None:
dataset_root = tmp_path / "data" / "train"
scene_root = dataset_root / "scene_a"
(scene_root / "rgb").mkdir(parents=True)
(scene_root / "depth").mkdir()
frames = []
intrinsics = [[4.0, 0.0, 2.0], [0.0, 4.0, 2.0], [0.0, 0.0, 1.0]]
for frame_id in range(2):
Image.fromarray(np.full((5, 5, 3), 64 + frame_id, dtype=np.uint8)).save(
scene_root / "rgb" / f"{frame_id:06d}.png"
)
np.save(scene_root / "depth" / f"{frame_id:06d}.npy", np.ones((5, 5)) * 2.0)
frames.append(
{
"frame_id": frame_id,
"rgb": f"rgb/{frame_id:06d}.png",
"depth": f"depth/{frame_id:06d}.npy",
"intrinsics": intrinsics,
"c2w": np.eye(4).tolist(),
}
)
manifest = {
"schema_version": 1,
"scene_id": "scene_a",
"coordinate_convention": "opencv_c2w",
"image_size": [5, 5],
"bounds": [[-1, -1, 0.1], [1, 1, 4]],
"frames": frames,
}
manifest_path = scene_root / "manifest.json"
manifest_path.write_text(json.dumps(manifest), encoding="utf-8")
windows_path = tmp_path / "frozen_windows.json"
windows_path.write_text(
json.dumps(
{
"partition": "train",
"seed": 17,
"selection_rule": "test fixture",
"windows": [
{
"scene_id": "scene_a",
"start": 0,
"frame_ids": [0, 1],
"manifest": str(manifest_path),
}
],
}
),
encoding="utf-8",
)
output_dir = tmp_path / "output"
config = DenseGeometryAuditConfig(
seed=17,
dataset_root=str(dataset_root),
frozen_windows=str(windows_path),
output_dir=str(output_dir),
context_views=2,
permutation_windows=1,
supplier_identity=SupplierIdentity("model", "revision", "repo", "code"),
run_config=SupplierRunConfig(),
code_root=str(tmp_path),
dinov2_code_root=str(tmp_path),
device="cpu",
model_cache_dir=str(tmp_path / "cache"),
certificate=DenseCertificateConfig(grid_stride=2),
decision=DenseAuditDecisionThresholds(minimum_windows=1),
)
inference_calls = []
class Supplier:
def infer(self, request):
inference_calls.append(request.frame_ids)
view_count = len(request.frame_ids)
height, width = request.cameras.image_size
return GeometrySupplierResult(
request.frame_ids,
np.ones((view_count, height, width), dtype=np.float32) * 2.0,
np.ones((view_count, height, width), dtype=np.float32),
np.ones((view_count, height, width), dtype=np.bool_),
request.cameras.intrinsics.numpy(),
request.cameras.c2w.numpy(),
)
result = run_dense_audit(config, supplier=Supplier())
assert result["metrics"]["window_count"] == 1
assert len(inference_calls) == 2
assert result["certificate_before_depth"] is True
assert result["final_or_diagnostic_access"] is False
assert (output_dir / "audit_result.json").is_file()
assert (output_dir / "window_metrics.jsonl").is_file()
assert (output_dir / "access_log.json").is_file()
assert (output_dir / "REPORT.md").is_file()
assert (output_dir / "artifact_hashes.json").is_file()
def test_complete_report_exposes_runtime_model_data_and_label_provenance() -> None:
result = {
"status": "STOP_MAPANYTHING_SUPPLIER",
"decision": {
"reason": "one or more gates failed",
"gates": {"certified_coverage": True, "median_geometry_error": False},
},
"metrics": {
"window_count": 32,
"certified_coverage_macro": 0.3,
"nonzero_window_fraction": 1.0,
"geometry_error_median_m": 0.6,
"geometry_error_p90_m": 3.5,
"matched_confidence_median_error_m": 0.45,
"risk_error_spearman": -0.01,
"shuffled_certified_coverage_macro": 0.07,
"shuffled_geometry_error_median_m": 1.6,
"permutation_max_depth_delta_m": 0.0,
},
"certificate_before_depth": True,
"final_or_diagnostic_access": False,
"supplier_identity": SupplierIdentity(
"facebook/map-anything",
"model-revision",
"facebookresearch/map-anything",
"code-revision",
(
CodeDependency(
"dinov2", "facebookresearch/dinov2", "dinov2-revision", "a" * 64
),
),
).to_dict(),
}
provenance = {
"audit_result_sha256": "b" * 64,
"model_files": [
{
"name": "model.safetensors",
"bytes": 123,
"sha256": "c" * 64,
}
],
"code_snapshots": [
{
"name": "mapanything",
"revision": "code-revision",
"source_sha256": "d" * 64,
}
],
"runtime": {
"python": "3.11.14",
"torch": "2.11.0+cu128",
"cuda_runtime": "12.8",
"gpu": "NVIDIA GeForce RTX 5070",
"gpu_total_bytes": 12820480000,
"cuda_peak_memory_bytes": 5685421568,
},
"disk": {
"audit_output_bytes": 1000,
"model_cache_bytes": 2000,
"d_free_bytes": 3000,
},
"training_data_disclosure": {
"datasets": ["Aria Synthetic Environments", "BlendedMVS"],
"hypersim_named_in_public_config": False,
"limitation": "No direct or indirect overlap claim is established.",
},
"label_access": {
"count": 2,
"all_after_certificate_seal": True,
"final_or_diagnostic_access": False,
"paths": [
r"D:\data\train\scene\depth\000001.npy",
r"D:\data\train\scene\depth\000002.npy",
],
},
}
report = render_dense_report(result, provenance)
assert "model.safetensors" in report
assert "NVIDIA GeForce RTX 5070" in report
assert "Aria Synthetic Environments" in report
assert "No direct or indirect overlap claim is established." in report
assert r"D:\data\train\scene\depth\000001.npy" in report
assert "Formal Research Institute: not invoked" in report
assert "Evaluation Council: not convened" in report
def test_finalization_preserves_scientific_result_and_refreshes_all_hashes(
tmp_path: Path,
) -> None:
output = tmp_path / "audit"
output.mkdir()
result_path = output / "audit_result.json"
result = {
"status": "STOP_MAPANYTHING_SUPPLIER",
"decision": {"reason": "failed", "gates": {}},
"metrics": {"window_count": 32},
"certificate_before_depth": True,
"final_or_diagnostic_access": False,
"supplier_identity": SupplierIdentity("model", "revision", "repo", "code").to_dict(),
}
result_path.write_text(json.dumps(result, indent=2) + "\n", encoding="utf-8")
result_before = result_path.read_bytes()
provenance = {
"audit_result_sha256": __import__("hashlib").sha256(result_before).hexdigest(),
"model_files": [],
"code_snapshots": [],
"runtime": {},
"disk": {},
"training_data_disclosure": {
"datasets": [],
"hypersim_named_in_public_config": False,
"limitation": "fixture",
},
"label_access": {
"count": 0,
"all_after_certificate_seal": True,
"final_or_diagnostic_access": False,
"paths": [],
},
}
write_dense_audit_finalization(output, provenance)
assert result_path.read_bytes() == result_before
hashes = json.loads((output / "artifact_hashes.json").read_text(encoding="utf-8"))["sha256"]
assert "provenance.json" in hashes
assert "REPORT.md" in hashes
for relative, expected in hashes.items():
actual = __import__("hashlib").sha256((output / relative).read_bytes()).hexdigest()
assert actual == expected
def test_camera_shuffle_control_is_scored_against_real_depth_not_a_fixed_penalty() -> None:
intrinsics = np.repeat(
np.asarray([[[4.0, 0.0, 2.0], [0.0, 4.0, 2.0], [0.0, 0.0, 1.0]]], dtype=np.float32),
2,
axis=0,
)
c2w = np.repeat(np.eye(4, dtype=np.float32)[None], 2, axis=0)
supplier = GeometrySupplierResult(
(0, 1),
np.ones((2, 5, 5), dtype=np.float32) * 2.0,
np.ones((2, 5, 5), dtype=np.float32),
np.ones((2, 5, 5), dtype=np.bool_),
intrinsics,
c2w,
)
request = GeometrySupplierRequest(
(0, 1),
torch.ones(2, 3, 5, 5),
Cameras(torch.from_numpy(intrinsics), torch.from_numpy(c2w), (5, 5)),
)
depths = np.ones((2, 5, 5), dtype=np.float32) * 2.0
control = _evaluate_camera_shuffled_control(
supplier, request, depths, DenseCertificateConfig(grid_stride=2)
)
assert control["geometry_error_median_m"] == pytest.approx(0.0)
assert control["certified_coverage"] == pytest.approx(1.0)