"""Run the frozen V8 Q0 supplier qualification on one development-train window.""" from __future__ import annotations import argparse import hashlib import json import platform import time from datetime import UTC, datetime from pathlib import Path from typing import Any import numpy as np import torch from PIL import Image from mcss.data.manifest import SceneManifest, load_manifest from mcss.geometry_supplier import GeometrySupplierRequest from mcss.types import Cameras from mcss.v8_geometry import ( V8Proposal, classify_dual_proposals, ray_distance_to_depth_z, write_v8_proposal_cache, ) from mcss.v8_suppliers import ( EMVSNET_DEPTH_FAR_M, EMVSNET_DEPTH_NEAR_M, EMVSNetSupplier, UniDepthV2SmallSupplier, ) Q0_SCHEMA_VERSION = "mcss.v8_q0.v3" REPEAT_TOLERANCE_M = 1e-5 PEAK_MEMORY_LIMIT_BYTES = 10 * 1024**3 def main() -> int: args = _parse_args() determinism = _configure_determinism() output = Path(args.output_dir).resolve() if output.exists() and any(output.iterdir()): raise FileExistsError(f"refusing to overwrite non-empty Q0 output: {output}") output.mkdir(parents=True, exist_ok=True) started = time.time() record = _initial_record(args) record["runtime"]["determinism"] = determinism _write_json(output / "q0_running.json", record) try: window = _load_window(args.window_file) manifest = load_manifest(window["manifest"]) frames = _select_frames(manifest, window["frame_ids"]) request = _load_request(manifest, frames) record["window"] = { "scene_id": window["scene_id"], "frame_ids": list(request.frame_ids), "manifest": str(Path(window["manifest"]).resolve()), "image_size": list(request.cameras.image_size), } _check_input_contract(request) record["checks"]["input_contract"] = {"status": "PASS"} uni = UniDepthV2SmallSupplier( code_root=args.unidepth_code_root, local_model_dir=args.unidepth_bundle, device=args.device, ) emv = EMVSNetSupplier( code_root=args.emvsnet_code_root, checkpoint=args.emvsnet_checkpoint, device=args.device, image_scale=0.25, depth_bins=64, max_h=512, max_w=640, evidential_method=args.evidential_method, ) checkpoint_state = _check_model_loads(uni, emv) record["checks"]["model_load"] = {"status": "PASS"} provenance = _checkpoint_method_provenance( checkpoint_state, requested_method=args.evidential_method, checkpoint_path=args.emvsnet_checkpoint, provenance_path=args.emvsnet_provenance, model_card_path=args.emvsnet_model_card, parameter_names=[name for name, _parameter in emv._load().named_parameters()], ) record["checks"]["checkpoint_method_provenance"] = provenance case_specs = [("four_view", request, [0, 1, 2, 3])] for drop in (1, 2, 3): keep = [index for index in range(4) if index != drop] case_specs.append((f"drop_source_{drop}", _select_request(request, keep), keep)) # Phase A is completely label blind. Every proposal is sealed before any depth path opens. sealed_cases = [] for name, case_request, keep in case_specs: sealed_cases.append( _run_and_seal_case( output / "cases" / name, name, case_request, keep, uni, emv, ) ) record["checks"]["all_proposals_sealed_before_context_depth"] = {"status": "PASS"} repeat = _check_repeat_stability(output, request, uni, emv) record["checks"]["repeat_stability"] = { "status": "PASS" if repeat["passed"] else "FAIL", **repeat, } if not repeat["passed"]: raise RuntimeError("STOP_EMVSNET_Q0: repeated supplier inference is not stable") full_emv = sealed_cases[0]["emvsnet_proposal"] interventions: dict[str, dict[str, Any]] = {} for case in sealed_cases[1:]: interventions[case["name"]] = _source_drop_intervention_delta( full_emv, case["emvsnet_proposal"] ) _write_json(output / "source_drop_interventions.json", interventions) record["checks"]["source_drop_interventions"] = { "status": "PASS", "cases": interventions, } # Phase B may now open development context labels for descriptive checks only. access_log: list[dict[str, Any]] = [] case_results = [] for case in sealed_cases: context_depth = _load_context_depth( case["request"], args.window_file, access_log=access_log ) intervention = interventions.get(case["name"]) intervention_delta = ( None if intervention is None else intervention["median_depth_delta_m"] ) case_results.append( _finalize_case( case, context_depth, intervention_delta_m=intervention_delta, ) ) record["cases"] = case_results _write_json(output / "access_log.json", access_log) record["access_log"] = { "path": str(output / "access_log.json"), "opened_after_all_proposal_seals": all( entry["opened_after_all_proposal_seals"] for entry in access_log ), "context_depth_file_count": len(access_log), } peak = ( int(torch.cuda.max_memory_allocated()) if torch.cuda.is_available() and args.device.startswith("cuda") else None ) record["runtime"]["cuda_peak_memory_bytes"] = peak peak_status = "PASS" if peak is None or peak <= PEAK_MEMORY_LIMIT_BYTES else "FAIL" record["checks"]["peak_memory"] = { "status": peak_status, "bytes": peak, "limit_bytes": PEAK_MEMORY_LIMIT_BYTES, } if peak_status == "FAIL": raise RuntimeError("STOP_EMVSNET_Q0: peak CUDA memory exceeded 10 GiB") record["mechanical_qualification"] = { "status": "PASS", "meaning": "both suppliers completed frozen forward and protocol checks", } _require_checkpoint_method_provenance(provenance) record["status"] = "GO_TO_Q1_LABEL_BLIND_AUDIT" except Exception as error: record["status"] = "STOP_EMVSNET_Q0" record["error"] = {"type": type(error).__name__, "message": str(error)} record["finished_seconds"] = time.time() - started _write_json(output / "q0_result.json", record) (output / "Q0_REPORT.md").write_text(_render_report(record), encoding="utf-8") try: (output / "q0_running.json").unlink() except FileNotFoundError: pass print(json.dumps({"status": record["status"], "output": str(output)}, sort_keys=True)) return 0 if record["status"] == "GO_TO_Q1_LABEL_BLIND_AUDIT" else 2 def _initial_record(args: argparse.Namespace) -> dict[str, Any]: return { "schema_version": Q0_SCHEMA_VERSION, "status": "RUNNING", "research_boundary": { "partition": "development-train", "window": "window_000", "proposal_phase_target_access": False, "postseal_context_depth_audit": True, "training": False, "sim3_or_scale_fit": False, "threshold_selection_from_labels": False, "supplier_writes_to_typed_state": False, }, "runtime": _runtime_info(), "config": { "window_file": str(Path(args.window_file).resolve()), "unidepth_code_root": str(Path(args.unidepth_code_root).resolve()), "emvsnet_code_root": str(Path(args.emvsnet_code_root).resolve()), "unidepth_bundle": str(Path(args.unidepth_bundle).resolve()), "emvsnet_checkpoint": str(Path(args.emvsnet_checkpoint).resolve()), "emvsnet_provenance": ( None if args.emvsnet_provenance is None else str(Path(args.emvsnet_provenance).resolve()) ), "emvsnet_model_card": ( None if args.emvsnet_model_card is None else str(Path(args.emvsnet_model_card).resolve()) ), "device": args.device, "image_size": [128, 160], "depth_bins": 64, "depth_range_m": [EMVSNET_DEPTH_NEAR_M, EMVSNET_DEPTH_FAR_M], "depth_range_provenance": "pre_registered_protocol_fixed_not_label_derived", "evidential_method": args.evidential_method, }, "checks": {}, "cases": [], } def _configure_determinism() -> dict[str, Any]: """Freeze CUDA algorithm choices so repeatability is an executable contract.""" torch.use_deterministic_algorithms(True) torch.backends.cudnn.deterministic = True torch.backends.cudnn.benchmark = False return { "torch_deterministic_algorithms": torch.are_deterministic_algorithms_enabled(), "cudnn_deterministic": bool(torch.backends.cudnn.deterministic), "cudnn_benchmark": bool(torch.backends.cudnn.benchmark), } def _parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser() parser.add_argument("--window-file", required=True, type=Path) parser.add_argument("--output-dir", required=True, type=Path) parser.add_argument("--unidepth-code-root", required=True, type=Path) parser.add_argument("--unidepth-bundle", required=True, type=Path) parser.add_argument("--emvsnet-code-root", required=True, type=Path) parser.add_argument("--emvsnet-checkpoint", required=True, type=Path) parser.add_argument("--emvsnet-provenance", type=Path) parser.add_argument("--emvsnet-model-card", type=Path) parser.add_argument("--device", default="cuda") parser.add_argument("--evidential-method", choices=("der", "sder"), default="der") return parser.parse_args() def _load_window(path: Path) -> dict[str, Any]: payload = json.loads(path.read_text(encoding="utf-8")) if payload.get("partition") != "train" or len(payload.get("windows", [])) != 1: raise ValueError("Q0 requires exactly one frozen development-train window") window = payload["windows"][0] if window.get("scene_id") is None or len(window.get("frame_ids", [])) != 4: raise ValueError("Q0 window must contain exactly four frame IDs") if any( token in str(window.get("manifest", "")).lower() for token in ("diagnostic_test", "final_holdout") ): raise ValueError("Q0 cannot access diagnostic-test or final-holdout manifests") return window def _select_frames(manifest: SceneManifest, frame_ids: list[int]) -> list[Any]: by_id = {frame.frame_id: frame for frame in manifest.frames} try: return [by_id[int(frame_id)] for frame_id in frame_ids] except KeyError as error: raise ValueError(f"frozen frame ID is absent from manifest: {error}") from error def _load_request(manifest: SceneManifest, frames: list[Any]) -> GeometrySupplierRequest: rgb = [] intrinsics = [] c2w = [] for frame in frames: rgb.append( torch.from_numpy( np.asarray(Image.open(manifest.root / frame.rgb).convert("RGB"), dtype=np.float32) / 255.0 ).permute(2, 0, 1) ) intrinsics.append(torch.tensor(frame.intrinsics, dtype=torch.float32)) c2w.append(torch.tensor(frame.c2w, dtype=torch.float32)) return GeometrySupplierRequest( tuple(frame.frame_id for frame in frames), torch.stack(rgb), Cameras(torch.stack(intrinsics), torch.stack(c2w), manifest.image_size), ) def _select_request(request: GeometrySupplierRequest, keep: list[int]) -> GeometrySupplierRequest: indices = torch.tensor(keep, dtype=torch.long) return GeometrySupplierRequest( tuple(request.frame_ids[index] for index in keep), request.rgb.index_select(0, indices), request.cameras.select_views(indices), ) def _check_input_contract(request: GeometrySupplierRequest) -> None: if request.rgb.shape[-2:] != (128, 160): raise ValueError("Q0 frozen input must be 128x160") if not torch.isfinite(request.rgb).all(): raise ValueError("input RGB contains nonfinite values") def _check_model_loads( uni: UniDepthV2SmallSupplier, emv: EMVSNetSupplier ) -> dict[str, Any]: uni._load() emv._load() state = torch.load(emv.checkpoint, map_location="cpu", weights_only=False) if not isinstance(state, dict): raise ValueError("EMVSNet checkpoint payload must be a mapping") return state def _checkpoint_method_provenance( checkpoint_state: dict[str, Any], *, requested_method: str, checkpoint_path: Path | None = None, provenance_path: Path | None = None, model_card_path: Path | None = None, parameter_names: list[str] | None = None, ) -> dict[str, Any]: recorded = checkpoint_state.get("evidential_method") or checkpoint_state.get("training_method") if recorded is not None and str(recorded).lower() != requested_method: return { "status": "FAIL", "reason": "training_method_mismatch", "requested_method": requested_method, "checkpoint_method": str(recorded), } if recorded is not None: return { "status": "PASS", "reason": "training_method_verified_inside_checkpoint", "requested_method": requested_method, "checkpoint_method": str(recorded), "checkpoint_internal_method": str(recorded), "evidence_kind": "checkpoint_internal_training_metadata", "claim_ceiling": "training_method_recorded_by_checkpoint", } base = { "requested_method": requested_method, "checkpoint_method": None, "checkpoint_internal_method": None, "checkpoint_keys": sorted(str(key) for key in checkpoint_state), } if checkpoint_path is None or provenance_path is None or model_card_path is None: return {"status": "FAIL", "reason": "training_method_unverified", **base} try: manifest = json.loads(Path(provenance_path).read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as error: return { "status": "FAIL", "reason": "publisher_provenance_unreadable", "detail": str(error), **base, } if manifest.get("schema_version") != "mcss.emvsnet_checkpoint_provenance.v1": return {"status": "FAIL", "reason": "publisher_provenance_schema_mismatch", **base} checkpoint = manifest.get("checkpoint") training = manifest.get("training") model_card = manifest.get("model_card") if not all(isinstance(value, dict) for value in (checkpoint, training, model_card)): return {"status": "FAIL", "reason": "publisher_provenance_incomplete", **base} assert isinstance(checkpoint, dict) assert isinstance(training, dict) assert isinstance(model_card, dict) actual_checkpoint = Path(checkpoint_path).resolve() actual_model_card = Path(model_card_path).resolve() if checkpoint.get("filename") != actual_checkpoint.name: return {"status": "FAIL", "reason": "checkpoint_filename_mismatch", **base} if checkpoint.get("sha256") != _sha256_file(actual_checkpoint): return {"status": "FAIL", "reason": "checkpoint_sha256_mismatch", **base} if model_card.get("sha256") != _sha256_file(actual_model_card): return {"status": "FAIL", "reason": "model_card_sha256_mismatch", **base} required_fragments = model_card.get("required_fragments") if not isinstance(required_fragments, list) or not required_fragments: return {"status": "FAIL", "reason": "model_card_fragments_missing", **base} model_card_text = actual_model_card.read_text(encoding="utf-8") if any(str(fragment) not in model_card_text for fragment in required_fragments): return {"status": "FAIL", "reason": "model_card_contract_mismatch", **base} released_method = str(training.get("evidential_method", "")).lower() evidence_kind = str(training.get("evidence_kind", "")) if released_method != requested_method: return {"status": "FAIL", "reason": "publisher_method_mismatch", **base} if evidence_kind != "publisher_release_inference_contract": return {"status": "FAIL", "reason": "publisher_evidence_kind_unsupported", **base} optimizer_signature = _optimizer_method_signature(checkpoint_state, parameter_names) warnings: list[dict[str, str]] = [] if optimizer_signature.get("pattern") == "low_alpha_second_moment": warnings.append( { "code": "optimizer_alpha_moment_imbalance", "severity": "WARNING", "message": ( "Independent alpha channels have much smaller Adam second moments than peer " "channels. This is an auditable training trace, but it cannot identify DER " "versus SDER and does not override the hash-pinned publisher inference " "contract." ), } ) return { "status": "PASS", "reason": "inference_method_verified_by_publisher_release_contract", "requested_method": requested_method, "checkpoint_method": None, "checkpoint_internal_method": None, "publisher_inference_method": released_method, "checkpoint_keys": base["checkpoint_keys"], "evidence_kind": evidence_kind, "claim_ceiling": "publisher_prescribed_inference_method_not_checkpoint_training_log", "checkpoint_sha256": str(checkpoint["sha256"]), "model_card_sha256": str(model_card["sha256"]), "hub_model_id": str(checkpoint.get("hub_model_id", "")), "hub_revision": str(checkpoint.get("hub_revision", "")), "provenance_manifest": str(Path(provenance_path).resolve()), "model_card": str(actual_model_card), "optimizer_signature": optimizer_signature, "warnings": warnings, } def _optimizer_method_signature( checkpoint_state: dict[str, Any], parameter_names: list[str] | None ) -> dict[str, Any]: """Audit alpha-channel Adam moments without inferring the training method. Small second moments are compatible with an unused channel, but also with saturation, branch weighting, loss scaling, or other training dynamics. The trace is therefore diagnostic only. """ optimizer = checkpoint_state.get("optimizer") if not isinstance(optimizer, dict) or parameter_names is None: return {"status": "UNAVAILABLE", "reason": "optimizer_or_parameter_names_missing"} groups = optimizer.get("param_groups") states = optimizer.get("state") if not isinstance(groups, list) or not isinstance(states, dict): return {"status": "UNAVAILABLE", "reason": "optimizer_state_incomplete"} parameter_ids = [parameter_id for group in groups for parameter_id in group.get("params", [])] if len(parameter_ids) != len(parameter_names): return {"status": "UNAVAILABLE", "reason": "optimizer_parameter_order_unverified"} heads: list[dict[str, Any]] = [] for name, parameter_id in zip(parameter_names, parameter_ids, strict=True): if not ( name.startswith("evidential.classif") and name.endswith(".2.weight") ): continue state = states.get(parameter_id) if not isinstance(state, dict): continue second_moment = state.get("exp_avg_sq") if not isinstance(second_moment, torch.Tensor) or second_moment.ndim < 1: continue if second_moment.shape[0] != 4 or not torch.isfinite(second_moment).all(): continue channel_sums = second_moment.detach().double().reshape(4, -1).sum(dim=1) other_reference = torch.median(channel_sums[[0, 1, 3]]) heads.append( { "parameter": name, "second_moment_sums": { label: float(channel_sums[index]) for index, label in enumerate(("gamma", "nu", "alpha", "beta")) }, "other_channel_reference": float(other_reference), "alpha_to_other_ratio": ( None if other_reference <= 0 else float(channel_sums[2] / other_reference) ), } ) positive_references = [ float(head["other_channel_reference"]) for head in heads if float(head["other_channel_reference"]) > 0 ] if not positive_references: return {"status": "UNAVAILABLE", "reason": "evidential_head_moments_missing"} active_floor = max(positive_references) * 1e-8 active_heads = [ head for head in heads if float(head["other_channel_reference"]) >= active_floor ] for head in heads: head["active"] = head in active_heads ratios = [float(head["alpha_to_other_ratio"]) for head in active_heads] low_alpha_pattern = len(active_heads) >= 2 and max(ratios) <= 1e-6 return { "status": "AVAILABLE", "diagnostic_semantics": ( "optimizer second moments only; records relative alpha-channel update scale but " "cannot identify the evidential training method" ), "active_head_rule": "other-channel reference >= 1e-8 of maximum head reference", "low_alpha_pattern_rule": "at least two active heads and max alpha/other ratio <= 1e-6", "active_head_count": len(active_heads), "pattern": ( "low_alpha_second_moment" if low_alpha_pattern else "no_low_alpha_second_moment" ), "method_inference": "INDETERMINATE", "hard_gate_eligible": False, "heads": heads, } def _require_checkpoint_method_provenance(provenance: dict[str, Any]) -> None: if provenance.get("status") != "PASS": raise RuntimeError(f"STOP_EMVSNET_Q0: {provenance.get('reason', 'unknown_provenance')}") def _run_and_seal_case( case_dir: Path, name: str, request: GeometrySupplierRequest, keep: list[int], uni: UniDepthV2SmallSupplier, emv: EMVSNetSupplier, ) -> dict[str, Any]: case_dir.mkdir(parents=True, exist_ok=True) if torch.cuda.is_available(): torch.cuda.reset_peak_memory_stats() started = time.time() uni_proposal = uni.infer(request) _check_proposal(uni_proposal, request) uni_cache = write_v8_proposal_cache( case_dir / "unidepth", uni_proposal, metadata={"case": name, "source_indices": keep} ) emv_proposal = emv.infer(request) _check_proposal(emv_proposal, request) emv_cache = write_v8_proposal_cache( case_dir / "emvsnet", emv_proposal, metadata={"case": name, "source_indices": keep} ) seal_times = { "unidepth": _read_seal_time(uni_cache), "emvsnet": _read_seal_time(emv_cache), } return { "case_dir": case_dir, "name": name, "request": request, "source_indices": keep, "unidepth_proposal": uni_proposal, "emvsnet_proposal": emv_proposal, "unidepth_cache": uni_cache, "emvsnet_cache": emv_cache, "seal_times_utc": seal_times, "forward_seconds": time.time() - started, "cuda_peak_memory_bytes": ( int(torch.cuda.max_memory_allocated()) if torch.cuda.is_available() else None ), } def _finalize_case( case: dict[str, Any], context_depth: np.ndarray, *, intervention_delta_m: float | None, ) -> dict[str, Any]: uni = case["unidepth_proposal"] emv = case["emvsnet_proposal"] posthoc = _posthoc_depth_check(uni, emv, context_depth) status = classify_dual_proposals( uni, emv, disagreement_threshold_m=0.25, intervention_delta_m=intervention_delta_m, ) metric = { "case": case["name"], "source_indices": case["source_indices"], "frame_ids": list(case["request"].frame_ids), "unidepth_cache": str(case["unidepth_cache"]), "emvsnet_cache": str(case["emvsnet_cache"]), "cache_seal_times_utc": case["seal_times_utc"], "dual_status": status, "posthoc_context_depth_check": posthoc, "finite_outputs": True, "forward_seconds": case["forward_seconds"], "cuda_peak_memory_bytes": case["cuda_peak_memory_bytes"], } _write_json(case["case_dir"] / "case.json", metric) return metric def _check_proposal(proposal: V8Proposal, request: GeometrySupplierRequest) -> None: canonical = request.canonical() if proposal.frame_ids != canonical.frame_ids: raise ValueError("proposal frame IDs do not match canonical request") if proposal.image_size != canonical.cameras.image_size: raise ValueError("proposal spatial size does not match request") if not np.isfinite(proposal.depth_z_m).all() or not np.isfinite(proposal.points_cam_m).all(): raise ValueError("proposal contains nonfinite values") valid = proposal.valid_mask if valid.any() and (proposal.depth_z_m[valid] <= 0).any(): raise ValueError("proposal has nonpositive valid depth") expected_intrinsics = canonical.cameras.intrinsics.detach().cpu().numpy() expected_c2w = canonical.cameras.c2w.detach().cpu().numpy() if not np.allclose(proposal.intrinsics, expected_intrinsics, atol=1e-5, rtol=1e-5): raise ValueError("proposal intrinsics do not match canonical request") if not np.allclose(proposal.c2w, expected_c2w, atol=1e-5, rtol=1e-5): raise ValueError("proposal c2w does not match canonical request") if valid.any() and not np.allclose( proposal.points_cam_m[..., 2][valid], proposal.depth_z_m[valid], atol=1e-4, rtol=1e-4, ): raise ValueError("proposal point z coordinate does not match depth") def _load_context_depth( request: GeometrySupplierRequest, window_file: Path, *, access_log: list[dict[str, Any]] | None = None, ) -> np.ndarray: payload = _load_window(window_file) manifest = load_manifest(payload["manifest"]) by_id = {frame.frame_id: frame for frame in manifest.frames} depths = [] for frame_id in request.frame_ids: frame = by_id[frame_id] path = (manifest.root / frame.depth).resolve() opened_at = datetime.now(UTC).isoformat() depths.append(np.load(path, allow_pickle=False).astype(np.float32)) if access_log is not None: access_log.append( { "frame_id": frame_id, "path": str(path), "opened_at_utc": opened_at, "opened_after_all_proposal_seals": True, "purpose": "development_context_posthoc_descriptive_only", } ) return np.stack(depths) def _posthoc_depth_check( uni: V8Proposal, emv: V8Proposal, context_ray_distance: np.ndarray ) -> dict[str, Any]: if context_ray_distance.shape != uni.depth_z_m.shape: raise ValueError("context ray distance shape does not match proposals") context_depth_z = np.stack( [ ray_distance_to_depth_z(context_ray_distance[index], uni.intrinsics[index]) for index in range(len(uni.frame_ids)) ] ) context_valid = np.isfinite(context_ray_distance) & (context_ray_distance > 0) uni_valid = uni.valid_mask & context_valid emv_valid = emv.valid_mask & context_valid return { "context_depth_opened_after_seal": True, "context_depth_semantics": "ray_distance_m_converted_to_opencv_z_m", "context_depth_shape": list(context_ray_distance.shape), "unidepth_overlap_fraction": float(uni_valid.mean()), "emvsnet_overlap_fraction": float(emv_valid.mean()), "unidepth_median_abs_error_m": _masked_median( np.abs(uni.depth_z_m - context_depth_z), uni_valid ), "emvsnet_median_abs_error_m": _masked_median( np.abs(emv.depth_z_m - context_depth_z), emv_valid ), } def _check_repeat_stability( output: Path, request: GeometrySupplierRequest, uni: UniDepthV2SmallSupplier, emv: EMVSNetSupplier, ) -> dict[str, Any]: first_uni = uni.infer(request) second_uni = uni.infer(request) first_emv = emv.infer(request) second_emv = emv.infer(request) result = _repeat_stability_metrics(first_uni, second_uni, first_emv, second_emv) _write_json(output / "repeat_stability.json", result) return result def _repeat_stability_metrics( first_uni: V8Proposal, second_uni: V8Proposal, first_emv: V8Proposal, second_emv: V8Proposal, ) -> dict[str, Any]: uni_mask = first_uni.valid_mask & second_uni.valid_mask emv_mask = first_emv.valid_mask & second_emv.valid_mask uni_delta = _masked_max(np.abs(first_uni.depth_z_m - second_uni.depth_z_m), uni_mask) emv_delta = _masked_max(np.abs(first_emv.depth_z_m - second_emv.depth_z_m), emv_mask) passed = ( uni_delta is not None and emv_delta is not None and uni_delta <= REPEAT_TOLERANCE_M and emv_delta <= REPEAT_TOLERANCE_M ) return { "max_unidepth_depth_delta_m": uni_delta, "max_emvsnet_depth_delta_m": emv_delta, "tolerance_m": REPEAT_TOLERANCE_M, "passed": passed, } def _source_drop_intervention_delta( full: V8Proposal, dropped: V8Proposal ) -> dict[str, Any]: reference = dropped.frame_ids[0] try: full_index = full.frame_ids.index(reference) dropped_index = dropped.frame_ids.index(reference) except ValueError as error: raise ValueError("source-drop proposal does not share a reference frame") from error overlap = full.valid_mask[full_index] & dropped.valid_mask[dropped_index] delta = np.abs(full.depth_z_m[full_index] - dropped.depth_z_m[dropped_index]) return { "reference_frame_id": reference, "overlap_fraction": float(overlap.mean()), "median_depth_delta_m": _masked_median(delta, overlap), "max_depth_delta_m": _masked_max(delta, overlap), } def _read_seal_time(cache: Path) -> str: metadata = json.loads((cache / "metadata.json").read_text(encoding="utf-8")) value = metadata.get("sealed_at_utc") if not isinstance(value, str) or not value: raise ValueError("proposal cache does not record sealed_at_utc") return value def _masked_median(values: np.ndarray, mask: np.ndarray) -> float | None: selected = values[mask] return float(np.median(selected)) if selected.size else None def _masked_max(values: np.ndarray, mask: np.ndarray) -> float | None: selected = values[mask] return float(np.max(selected)) if selected.size else None def _runtime_info() -> dict[str, Any]: return { "python": platform.python_version(), "torch": torch.__version__, "cuda_runtime": torch.version.cuda, "gpu": torch.cuda.get_device_name(0) if torch.cuda.is_available() else None, } def _write_json(path: Path, payload: Any) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text( json.dumps(payload, indent=2, sort_keys=True, allow_nan=False) + "\n", encoding="utf-8", ) def _sha256_file(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as handle: for block in iter(lambda: handle.read(1024 * 1024), b""): digest.update(block) return digest.hexdigest() def _render_report(record: dict[str, Any]) -> str: lines = [ "# V8 Q0 Supplier Qualification", "", f"- Final status: `{record['status']}`", f"- Schema: `{record['schema_version']}`", "- Scope: one frozen Hypersim development-train window; proposal generation is label " "blind, followed by a post-seal development context-depth audit; no training or " "typed-state write.", "", ] if "mechanical_qualification" in record: lines.extend( [ "## Mechanical Qualification", "", f"- Status: `{record['mechanical_qualification']['status']}`", "- Meaning: both external suppliers ran and passed the executable protocol checks.", "", ] ) if "error" in record: lines.extend( [ "## Final Gate", "", f"`{record['error']['type']}: {record['error']['message']}`", "", ] ) lines.extend(["## Checks", ""]) for name, value in record.get("checks", {}).items(): lines.append(f"- `{name}`: {value.get('status', 'recorded')}") warnings = [ warning for value in record.get("checks", {}).values() if isinstance(value, dict) for warning in value.get("warnings", []) if isinstance(warning, dict) ] if warnings: lines.extend(["", "## Diagnostic Warnings", ""]) for warning in warnings: severity = warning.get("severity", "WARNING") code = warning.get("code", "unspecified") message = warning.get("message", "") lines.append(f"- `[{severity}] {code}`: {message}") lines.extend( [ "", "## Interpretation", "", "`model-supported` means supplier agreement only. It is not a measurement certificate, " "correctness proof, or typed scene state. Post-hoc context-depth values are " "descriptive and were opened only after all proposal caches were sealed.", "", ] ) return "\n".join(lines) if __name__ == "__main__": raise SystemExit(main())