"""Post-hoc readout/reference re-analysis: readouts, metric parity, constants, labels.""" import numpy as np import pytest from mcss.mechanism_pilot.readout_reanalysis import ( READOUTS, REFERENCES, SOURCES, VARIANTS, absrel_optimal_constant, analyze, analyze_with_sensitivity, apply_readout, depth_metrics, reference_values, score_predictions, score_references, validate_matrix, ) from mcss.mechanism_pilot.statistics import measurement_metrics def sample(seed=0, shape=(12, 15)): rng = np.random.default_rng(seed) gt = rng.uniform(0.5, 8.0, shape).astype(np.float32) gt[0, :4] = 0.0 # invalid GT pixels are excluded exactly as in the frozen metric depth = (gt * rng.uniform(0.4, 1.3, shape)).astype(np.float32) opacity = rng.uniform(0.2, 1.0, shape).astype(np.float32) return depth, opacity, gt def test_raw_readout_reproduces_frozen_measurement_metrics(): depth, opacity, gt = sample() rgb = np.full((*gt.shape, 3), 0.5, dtype=np.float32) frozen = measurement_metrics(rgb, rgb, depth, gt, opacity) ours = depth_metrics(apply_readout("RAW", depth, opacity, gt), gt) assert ours["depth_absrel"] == pytest.approx(frozen["depth_absrel"], abs=1e-9) assert ours["depth_delta1"] == pytest.approx(frozen["depth_delta1"], abs=1e-12) def test_normalized_and_median_scaled_readouts(): depth, opacity, gt = sample(1) opacity[1, 1] = 0.0 normalized = apply_readout("OPACITY_NORMALIZED", depth, opacity, gt) assert normalized[2, 2] == pytest.approx(depth[2, 2] / opacity[2, 2]) assert np.isfinite(normalized).all() and normalized[1, 1] == pytest.approx(depth[1, 1] / 1e-6) scaled = apply_readout("MEDIAN_SCALED", depth, opacity, gt) valid = gt > 0 expected = np.median(gt[valid].astype(np.float64)) assert np.median(scaled[valid]) == pytest.approx(expected, rel=1e-9) with pytest.raises(ValueError): apply_readout("UNKNOWN", depth, opacity, gt) def test_absrel_optimal_constant_beats_every_other_constant(): rng = np.random.default_rng(3) t = np.exp(rng.normal(1.2, 0.6, 5000)) best = absrel_optimal_constant(t) grid = np.linspace(t.min(), t.max(), 4000) losses = np.abs(grid[:, None] - t[None]) / t[None] assert np.mean(np.abs(best - t) / t) <= losses.mean(1).min() + 1e-9 assert best < np.median(t) # the 1/t weighting favours nearer depths values = reference_values([t.reshape(50, 100), np.zeros(3)]) assert values["train_pixel_count"] == 5000 assert values["REF_TRAIN_MEDIAN"] == pytest.approx(np.median(t)) def matrix(carrier_error, c1_gain, anchor_penalty=0.1): scenes = [f"s{i}" for i in range(8)] queries = {s: [14, 15] for s in scenes} rows, refs = [], [] for i, scene in enumerate(scenes): for role in ("A", "B"): for query in queries[scene]: for readout in READOUTS: for ref in REFERENCES: refs.append( { "reference": ref, "scene_id": scene, "role": role, "query_id": query, "readout": readout, "depth_absrel": 0.5, "depth_delta1": 0.4, } ) for source in SOURCES: for variant in VARIANTS: for method in ("direct", "anchor"): for seed in (1, 2): e = 0.5 + carrier_error + 0.001 * i e -= c1_gain if variant == "C1" else 0.0 e += anchor_penalty if method == "anchor" else 0.0 rows.append( { "source": source, "variant": variant, "method": method, "seed": seed, "scene_id": scene, "role": role, "query_id": query, "readout": readout, "depth_absrel": e, "depth_delta1": 0.9 - e, } ) return rows, refs, scenes, queries def test_labels_follow_sign_conventions(): rows, refs, scenes, queries = matrix(carrier_error=-0.2, c1_gain=0.05) validate_matrix(rows, refs, [1, 2], scenes, queries) result = analyze(rows, refs, draws=200) entry = result["carriers"]["V3_C1"]["RAW"]["depth_absrel"] assert entry["reference_label"] == "ABOVE" and entry["vs_REF_TRAIN_MEDIAN"]["mean"] > 0 assert result["primary_contrasts"]["V4"]["by_readout"]["RAW"]["depth_absrel"]["label"] == ( "SUPPORTED" ) assert result["static"]["V2_C0"]["RAW"]["depth_delta1"]["label"] == "ABOVE" assert result["summary"]["CARRIER_REFERENCE_STATUS"] == ( "SOME_CARRIER_ABOVE_GEOMETRY_FREE_REFERENCE" ) worse, refs, _, _ = matrix(carrier_error=0.2, c1_gain=-0.05) result = analyze(worse, refs, draws=200) assert result["carriers"]["V2_C1"]["RAW"]["depth_absrel"]["reference_label"] == "BELOW" assert result["summary"]["CARRIER_REFERENCE_STATUS"] == ( "NO_CARRIER_ABOVE_GEOMETRY_FREE_REFERENCE" ) assert result["primary_contrasts"]["V2"]["by_readout"]["RAW"]["depth_absrel"]["label"] == ( "NOT_ESTABLISHED" ) def test_scoring_reproduces_sealed_metrics_and_rejects_drift(tmp_path): depth, opacity, gt = sample(5) path = tmp_path / "prediction.npz" np.savez_compressed(path, depth=depth, opacity=opacity) sealed = depth_metrics(depth, gt) record = { "source": "V5", "variant": "C1", "seed": 1, "scene_id": "s", "role": "A", "query_id": 14, "method": "direct", "step": 5, "path": str(path), "sealed_depth_absrel": sealed["depth_absrel"], "sealed_depth_delta1": sealed["depth_delta1"], "sealed_depth_valid_count": int((gt > 0).sum()), } rows, worst = score_predictions([record], {("s", 14): gt}) assert [r["readout"] for r in rows] == list(READOUTS) and worst["absrel"] < 1e-12 drifted = {**record, "sealed_depth_absrel": sealed["depth_absrel"] + 1e-3} with pytest.raises(RuntimeError): score_predictions([drifted], {("s", 14): gt}) values = {"REF_TRAIN_ABSREL_OPTIMAL": 2.0, "REF_TRAIN_MEDIAN": 4.0} refs = score_references(values, {("s", 14): gt}, {"s": [14]}) assert len(refs) == len(REFERENCES) * len(READOUTS) * 2 scaled = [r for r in refs if r["readout"] == "MEDIAN_SCALED"] assert len({round(r["depth_absrel"], 12) for r in scaled}) == 1 def test_no_hit_views_stay_unscaled_and_feed_the_sensitivity(tmp_path): _, _, gt = sample(6) zero = np.zeros_like(gt) assert np.array_equal(apply_readout("MEDIAN_SCALED", zero, zero, gt), zero) path = tmp_path / "nohit.npz" np.savez_compressed(path, depth=zero, opacity=zero) sealed = depth_metrics(zero, gt) record = { "source": "V4", "variant": "C0", "seed": 1, "scene_id": "s0", "role": "A", "query_id": 14, "method": "direct", "step": 5, "path": str(path), "sealed_depth_absrel": sealed["depth_absrel"], "sealed_depth_delta1": sealed["depth_delta1"], "sealed_depth_valid_count": int((gt > 0).sum()), } rows, worst = score_predictions([record], {("s0", 14): gt}) assert worst["median_scale_undefined_views"] == 1 assert all(r["all_zero_prediction"] for r in rows) assert {r["depth_absrel"] for r in rows} == {1.0} grid, refs, scenes, queries = matrix(carrier_error=-0.2, c1_gain=0.05) for row in grid: row["all_zero_prediction"] = row["scene_id"] == "s0" result = analyze_with_sensitivity(grid, refs, draws=100) sensitivity = result["sensitivity_excluding_no_hit_scenes"] assert sensitivity["excluded_no_hit_scenes"] == ["s0"] assert ( result["carriers"]["V2_C1"]["RAW"]["depth_absrel"]["vs_REF_TRAIN_MEDIAN"]["n_scenes"] == 8 ) assert ( sensitivity["carriers"]["V2_C1"]["RAW"]["depth_absrel"]["vs_REF_TRAIN_MEDIAN"]["n_scenes"] == 7 ) def test_single_source_analysis(): rows, refs, scenes, queries = matrix(carrier_error=-0.2, c1_gain=0.05) only = {"V4": SOURCES["V4"]} rows = [r for r in rows if r["source"] == "V4"] validate_matrix(rows, refs, [1, 2], scenes, queries, sources=only) result = analyze(rows, refs, sources=only, draws=100) assert set(result["carriers"]) == {"V4_C0", "V4_C1", "V4_C2"} assert result["summary"]["carrier_count"] == 3 def test_incomplete_matrix_is_rejected(): rows, refs, scenes, queries = matrix(carrier_error=0.0, c1_gain=0.0) with pytest.raises(ValueError): validate_matrix(rows[:-1], refs, [1, 2], scenes, queries) with pytest.raises(ValueError): validate_matrix(rows, refs + refs[:1], [1, 2], scenes, queries)