"""Full RELLIS validation qualification and bounded operator review export.""" from __future__ import annotations import hashlib import json import os import tempfile import time import zipfile from concurrent.futures import ProcessPoolExecutor, as_completed from dataclasses import asdict, dataclass from datetime import UTC, datetime from pathlib import Path from typing import Any, Final import numpy as np from k1link.datasets.gateway import DatasetPointFrame, decode_semantic_kitti_frame from k1link.datasets.goose_qualification import ( _aggregate, _check, _provider_frame_result, _validate_result, ) from k1link.datasets.rellis_admission import ( RELLIS_ADMISSION_SCHEMA, RELLIS_ARCHIVE_ROOT, RELLIS_LABEL_ARCHIVE, RELLIS_SCAN_ARCHIVE, RellisAdmissionError, _is_worker_dataset_root, _member_index, read_rellis_splits, ) from k1link.datasets.rellis_profile import RellisPatchworkProfile from k1link.datasets.rellis_smoke import RELLIS_SOURCE_ID from k1link.ground_segmentation import ( DEFAULT_GROUND_BENCHMARK_PROFILE, PATCHWORKPP_SOURCE_COMMIT, GroundBenchmarkProfile, GroundSegmentation, GroundSegmenter, LocalPercentileGroundSegmenter, PatchworkPPGroundSegmenter, ) from k1link.simulation.contracts import ( AuthorityProfile, ProviderPin, QualificationArtifact, QualificationRun, ReproducibilityTier, RunKind, RunState, ) from k1link.simulation.run_store import ( QualificationRunConflictError, QualificationRunStore, ) RELLIS_QUALIFICATION_REPORT_SCHEMA: Final = ( "missioncore.rellis-ground-qualification-report/v1" ) RELLIS_QUALIFICATION_FRAME_SCHEMA: Final = ( "missioncore.rellis-ground-qualification-frame/v1" ) RELLIS_REVIEW_PACK_SCHEMA: Final = "missioncore.rellis-ground-review-pack/v1" RELLIS_REVIEW_FRAME_SCHEMA: Final = "missioncore.rellis-ground-review-frame/v1" RELLIS_CALIBRATION_SCHEMA: Final = "missioncore.rellis-sensor-height-calibration/v1" RELLIS_REPORT_ARTIFACT_KIND: Final = "rellis-ground-qualification-report" EXPECTED_VALIDATION_FRAMES: Final = 2_413 DEFAULT_CALIBRATION_FRAMES: Final = 64 DEFAULT_REVIEW_POINTS: Final = 12_000 MAX_REVIEW_POINTS: Final = 20_000 GROUND_IDS: Final = frozenset({1, 3, 10, 23, 33}) ARTIFICIAL_GROUND_IDS: Final = frozenset({10, 23}) NATURAL_GROUND_IDS: Final = frozenset({1, 3, 33}) NON_GROUND_IDS: Final = frozenset({4, 5, 8, 12, 15, 17, 18, 19, 27, 34}) IGNORE_IDS: Final = frozenset({0, 6, 7, 9, 31}) _WORKER_SCANS: zipfile.ZipFile | None = None _WORKER_LABELS: zipfile.ZipFile | None = None _WORKER_SCAN_INDEX: dict[str, zipfile.ZipInfo] | None = None _WORKER_LABEL_INDEX: dict[str, zipfile.ZipInfo] | None = None _WORKER_CURRENT: GroundSegmenter | None = None _WORKER_PATCHWORK: GroundSegmenter | None = None @dataclass(frozen=True, slots=True) class RellisAcceptancePolicy: """Predeclared portability gates; never an actuator/safety promotion.""" minimum_ground_iou_gain: float = 0.02 minimum_obstacle_non_ground_recall: float = 0.90 maximum_obstacle_recall_regression: float = 0.01 minimum_assigned_fraction: float = 0.999 maximum_patchwork_latency_p95_ms: float = 50.0 maximum_per_frame_iou_regression: float = 0.20 def to_dict(self) -> dict[str, float]: return asdict(self) DEFAULT_RELLIS_ACCEPTANCE: Final = RellisAcceptancePolicy() def qualify_rellis_ground( dataset_root: Path, runs_root: Path, *, mission_core_commit: str, parallel_workers: int = 8, expected_frame_count: int = EXPECTED_VALIDATION_FRAMES, calibration_frame_count: int = DEFAULT_CALIBRATION_FRAMES, preview_points: int = DEFAULT_REVIEW_POINTS, current_profile: GroundBenchmarkProfile = DEFAULT_GROUND_BENCHMARK_PROFILE, acceptance: RellisAcceptancePolicy = DEFAULT_RELLIS_ACCEPTANCE, patchwork_module_name: str = "pypatchworkpp", current_segmenter: GroundSegmenter | None = None, patchwork_segmenter: GroundSegmenter | None = None, ) -> dict[str, Any]: """Calibrate on train labels, freeze the profile, then score validation.""" root = dataset_root.expanduser().absolute() runs = runs_root.expanduser().absolute() if not _is_worker_dataset_root(root): raise RellisAdmissionError("RELLIS qualification requires the canonical worker D root") if not 1 <= parallel_workers <= 32: raise RellisAdmissionError("parallel worker count is outside the admitted range") if not 5 <= calibration_frame_count <= 256: raise RellisAdmissionError("calibration frame count is outside the admitted range") if not 1 <= preview_points <= MAX_REVIEW_POINTS: raise RellisAdmissionError("review point count is outside the admitted range") if (current_segmenter is not None or patchwork_segmenter is not None) and parallel_workers != 1: raise RellisAdmissionError("custom providers require one qualification worker") admission = _read_admission(root) splits = read_rellis_splits(root) validation = splits["validation"] if len(validation) != expected_frame_count: raise RellisAdmissionError( f"RELLIS validation contains {len(validation)} frames, not {expected_frame_count}" ) scan_path = root / RELLIS_ARCHIVE_ROOT / RELLIS_SCAN_ARCHIVE label_path = root / RELLIS_ARCHIVE_ROOT / RELLIS_LABEL_ARCHIVE calibration = calibrate_rellis_sensor_height( scan_path, label_path, splits["train"], archive_identity=str(admission["release"]["identity_sha256"]), frame_count=calibration_frame_count, ) patchwork_profile = RellisPatchworkProfile( patchwork_sensor_height_proxy_m=calibration["sensor_height_m"], calibration_identity_sha256=calibration["identity_sha256"], calibration_frame_count=calibration["frame_count"], calibration_median_absolute_deviation_m=calibration[ "median_absolute_deviation_m" ], ) local = current_segmenter or LocalPercentileGroundSegmenter(current_profile) candidate = patchwork_segmenter or PatchworkPPGroundSegmenter.load( patchwork_profile, module_name=patchwork_module_name, ) providers = { "current": dict(local.identity), "patchworkpp": dict(candidate.identity), } profile = { "schema_version": "missioncore.rellis-ground-qualification-profile/v1", "source_id": RELLIS_SOURCE_ID, "split": "validation", "expected_frame_count": expected_frame_count, "archive_identity_sha256": admission["release"]["identity_sha256"], "current_profile": current_profile.to_dict(), "patchwork_profile": patchwork_profile.to_dict(), "calibration": calibration, "acceptance": acceptance.to_dict(), "reproducibility_tier": "R1", "authority": { "qualification_only": True, "navigation_or_safety_accepted": False, }, } identity = _canonical_sha256( { "profile": profile, "mission_core_commit": mission_core_commit, "providers": providers, } ) run_id = f"rellis-ground-{identity[:20]}" store = QualificationRunStore(runs) run = _admit_or_resume_run( store, run_id=run_id, identity=identity, profile=profile, mission_core_commit=mission_core_commit, providers=providers, ) if run.state is RunState.COMPLETED: return _read_completed_report(store, run) run = _ensure_running(store, run) store.append_event( run_id, event_type="qualification.profile-admitted", observed_at_utc=_utc_now(), host_monotonic_ns=time.monotonic_ns(), payload={ "frames_total": len(validation), "calibration_split": "train", "calibration_frames": calibration["frame_count"], "validation_labels_used_for_height": False, }, ) install_root = ( root / "rellis-3d/v1.1/installs" / str(admission["release"]["identity_sha256"]) ) work_root = install_root / "qualifications" / identity / "working" cache_root = work_root / "frames" review_root = work_root / "review-frames" cache_root.mkdir(parents=True, exist_ok=True) review_root.mkdir(parents=True, exist_ok=True) records: dict[str, dict[str, Any]] = {} pending: list[tuple[int, str, str, Path]] = [] for sequence, (point_member, label_member) in enumerate(validation): frame_id = _frame_id(point_member) cache_path = cache_root / f"{frame_id}.json" review_path = review_root / f"{frame_id}.npz" cached = _read_cached(cache_path, identity) if cached is None or not review_path.is_file(): pending.append((sequence, point_member, label_member, review_path)) else: records[frame_id] = cached if pending and parallel_workers == 1: for sequence, point_member, label_member, review_path in pending: frame = _read_frame(scan_path, label_path, point_member, label_member) record = _qualify_frame( frame, sequence=sequence, point_member=point_member, identity=identity, current=local, patchwork=candidate, review_path=review_path, preview_points=preview_points, ) _write_json_once(cache_root / f"{record['frame_id']}.json", record) records[record["frame_id"]] = record elif pending: with ProcessPoolExecutor( max_workers=parallel_workers, initializer=_initialize_worker, initargs=( str(scan_path), str(label_path), current_profile, patchwork_profile, patchwork_module_name, ), ) as executor: futures = { executor.submit( _qualify_frame_worker, sequence, point_member, label_member, identity, str(review_path), preview_points, ): point_member for sequence, point_member, label_member, review_path in pending } completed_since_event = 0 for future in as_completed(futures): record = future.result() _write_json_once(cache_root / f"{record['frame_id']}.json", record) records[record["frame_id"]] = record completed_since_event += 1 if completed_since_event >= 50 or len(records) == len(validation): store.append_event( run_id, event_type="qualification.frames-progress", observed_at_utc=_utc_now(), host_monotonic_ns=time.monotonic_ns(), payload={ "completed": len(records), "total": len(validation), "fraction": len(records) / len(validation), }, ) completed_since_event = 0 ordered = [records[_frame_id(point)] for point, _ in validation] report = _build_report( identity=identity, run_id=run_id, profile=profile, providers=providers, frames=ordered, acceptance=acceptance, ) run_root = runs / run_id report_path = run_root / "evidence/qualification.json" _write_json_once(report_path, report) _register_artifact( store, run_id, report_path, run_root, artifact_id="qualification-report", kind=RELLIS_REPORT_ARTIFACT_KIND, ) _publish_review_pack( runs, run_id=run_id, identity=identity, archive_identity=str(admission["release"]["identity_sha256"]), preview_points=preview_points, records=ordered, work_review_root=review_root, ) store.append_event( run_id, event_type="qualification.decision-recorded", observed_at_utc=_utc_now(), host_monotonic_ns=time.monotonic_ns(), payload={ "status": report["decision"]["status"], "passed": report["decision"]["passed"], "frames_completed": len(ordered), }, ) current_run = store.load(run_id) current_run = store.transition( run_id, RunState.STOPPING, expected_revision=current_run.revision, observed_at_utc=_utc_now(), host_monotonic_ns=time.monotonic_ns(), ) store.transition( run_id, RunState.COMPLETED, expected_revision=current_run.revision, observed_at_utc=_utc_now(), host_monotonic_ns=time.monotonic_ns(), reason="rellis-cross-dataset-evidence-sealed", ) return report def calibrate_rellis_sensor_height( scan_archive: Path, label_archive: Path, train_pairs: tuple[tuple[str, str], ...], *, archive_identity: str, frame_count: int = DEFAULT_CALIBRATION_FRAMES, ) -> dict[str, Any]: """Estimate one frozen height from deterministic training frames only.""" if not 5 <= frame_count <= 256: raise RellisAdmissionError("RELLIS calibration frame count is outside range") selected = _stratified_pairs(train_pairs, frame_count) estimates: list[dict[str, Any]] = [] try: with zipfile.ZipFile(scan_archive) as scans, zipfile.ZipFile(label_archive) as labels: scan_index = _member_index(scans) label_index = _member_index(labels) for point_member, label_member in selected: frame = decode_semantic_kitti_frame( scans.read(scan_index[point_member]), labels.read(label_index[label_member]), ) radial = np.linalg.norm(frame.points_xyz_m[:, :2], axis=1) ground = np.isin(frame.semantic_labels, tuple(GROUND_IDS)) z = frame.points_xyz_m[:, 2] admitted = ( ground & (radial >= 2.7) & (radial <= 15.0) & (z <= -0.30) & (z >= -3.0) ) heights = -z[admitted] if heights.size < 256: raise RellisAdmissionError( "RELLIS calibration frame has insufficient near-field ground" ) estimates.append( { "frame_id": _frame_id(point_member), "ground_point_count": int(heights.size), "height_m": float(np.median(heights)), } ) except (OSError, KeyError, zipfile.BadZipFile) as exc: raise RellisAdmissionError("RELLIS train calibration cannot read its frames") from exc values = np.asarray([item["height_m"] for item in estimates], dtype=np.float64) median = float(np.median(values)) mad = float(np.median(np.abs(values - median))) if not 0.5 <= median <= 3.0 or mad > 0.40: raise RellisAdmissionError("RELLIS train-derived height is physically inconsistent") identity_document = { "schema_version": RELLIS_CALIBRATION_SCHEMA, "source_id": RELLIS_SOURCE_ID, "archive_identity_sha256": archive_identity, "split": "train", "selection": "sequence-stratified-even-index", "range_m": [2.7, 15.0], "ground_label_ids": sorted(GROUND_IDS), "frame_ids": [item["frame_id"] for item in estimates], } return { **identity_document, "identity_sha256": _canonical_sha256(identity_document), "frame_count": len(estimates), "sensor_height_m": median, "median_absolute_deviation_m": mad, "minimum_frame_estimate_m": float(np.min(values)), "maximum_frame_estimate_m": float(np.max(values)), "frame_estimates": estimates, "validation_labels_used": False, } def _stratified_pairs( pairs: tuple[tuple[str, str], ...], frame_count: int, ) -> tuple[tuple[str, str], ...]: groups: dict[str, list[tuple[str, str]]] = {} for pair in pairs: groups.setdefault(pair[0].split("/", 1)[0], []).append(pair) if sorted(groups) != ["00000", "00001", "00002", "00003", "00004"]: raise RellisAdmissionError("RELLIS train split does not cover every sequence") selected: list[tuple[str, str]] = [] remaining = frame_count sequence_ids = sorted(groups) for index, sequence in enumerate(sequence_ids): count = remaining // (len(sequence_ids) - index) group = sorted(groups[sequence]) indices = np.linspace(0, len(group) - 1, count, dtype=np.int64) selected.extend(group[int(position)] for position in indices) remaining -= count if len(selected) != frame_count or len(set(selected)) != frame_count: raise RellisAdmissionError("RELLIS calibration selection is not unique") return tuple(selected) def _initialize_worker( scan_path: str, label_path: str, current_profile: GroundBenchmarkProfile, patchwork_profile: RellisPatchworkProfile, module_name: str, ) -> None: global _WORKER_CURRENT, _WORKER_LABEL_INDEX, _WORKER_LABELS global _WORKER_PATCHWORK, _WORKER_SCAN_INDEX, _WORKER_SCANS _WORKER_SCANS = zipfile.ZipFile(scan_path) _WORKER_LABELS = zipfile.ZipFile(label_path) _WORKER_SCAN_INDEX = _member_index(_WORKER_SCANS) _WORKER_LABEL_INDEX = _member_index(_WORKER_LABELS) _WORKER_CURRENT = LocalPercentileGroundSegmenter(current_profile) _WORKER_PATCHWORK = PatchworkPPGroundSegmenter.load( patchwork_profile, module_name=module_name, ) def _qualify_frame_worker( sequence: int, point_member: str, label_member: str, identity: str, review_path: str, preview_points: int, ) -> dict[str, Any]: if ( _WORKER_SCANS is None or _WORKER_LABELS is None or _WORKER_SCAN_INDEX is None or _WORKER_LABEL_INDEX is None or _WORKER_CURRENT is None or _WORKER_PATCHWORK is None ): raise RellisAdmissionError("RELLIS qualification worker is not initialized") frame = decode_semantic_kitti_frame( _WORKER_SCANS.read(_WORKER_SCAN_INDEX[point_member]), _WORKER_LABELS.read(_WORKER_LABEL_INDEX[label_member]), ) return _qualify_frame( frame, sequence=sequence, point_member=point_member, identity=identity, current=_WORKER_CURRENT, patchwork=_WORKER_PATCHWORK, review_path=Path(review_path), preview_points=preview_points, ) def _read_frame( scan_path: Path, label_path: Path, point_member: str, label_member: str, ) -> DatasetPointFrame: try: with zipfile.ZipFile(scan_path) as scans, zipfile.ZipFile(label_path) as labels: scan_index = _member_index(scans) label_index = _member_index(labels) return decode_semantic_kitti_frame( scans.read(scan_index[point_member]), labels.read(label_index[label_member]), ) except (OSError, KeyError, zipfile.BadZipFile) as exc: raise RellisAdmissionError("RELLIS validation frame cannot be read") from exc def _qualify_frame( frame: DatasetPointFrame, *, sequence: int, point_member: str, identity: str, current: GroundSegmenter, patchwork: GroundSegmenter, review_path: Path, preview_points: int, ) -> dict[str, Any]: semantic = frame.semantic_labels categories = np.zeros(frame.point_count, dtype=np.uint8) categories[np.isin(semantic, tuple(ARTIFICIAL_GROUND_IDS))] = 2 categories[np.isin(semantic, tuple(NATURAL_GROUND_IDS))] = 3 categories[np.isin(semantic, tuple(NON_GROUND_IDS))] = 4 known = np.isin(semantic, tuple(GROUND_IDS | NON_GROUND_IDS | IGNORE_IDS)) if not bool(np.all(known)): raise RellisAdmissionError("RELLIS validation frame contains an unknown label") evaluated = categories != 0 ground_truth = np.isin(semantic, tuple(GROUND_IDS)) xyzi = np.column_stack((frame.points_xyz_m, frame.remission)).astype( np.float32, copy=False, ) current_result = current.segment(xyzi) patchwork_result = patchwork.segment(xyzi) _validate_result(current_result, frame.point_count, "current") _validate_result(patchwork_result, frame.point_count, "patchworkpp") current_metrics = _provider_frame_result( current_result, ground_truth, evaluated, categories, frame.point_count, ) patchwork_metrics = _provider_frame_result( patchwork_result, ground_truth, evaluated, categories, frame.point_count, ) review = _write_review_frame( review_path, frame, ground_truth, evaluated, current_result, patchwork_result, frame_id=_frame_id(point_member), maximum_points=preview_points, ) sequence_id, frame_number = _frame_identity(point_member) return { "schema_version": RELLIS_QUALIFICATION_FRAME_SCHEMA, "identity_sha256": identity, "sequence": sequence, "frame_id": _frame_id(point_member), "dataset_sequence_id": sequence_id, "dataset_frame_number": frame_number, "source_point_count": frame.point_count, "nominal": { "current": current_metrics, "patchworkpp": patchwork_metrics, }, "review": review, } def _write_review_frame( path: Path, frame: DatasetPointFrame, ground_truth: np.ndarray[Any, Any], evaluated: np.ndarray[Any, Any], current: GroundSegmentation, patchwork: GroundSegmentation, *, frame_id: str, maximum_points: int, ) -> dict[str, Any]: sample_count = min(frame.point_count, maximum_points) indices = np.linspace(0, frame.point_count - 1, sample_count, dtype=np.int64) points_cm = np.rint(frame.points_xyz_m[indices] * 100.0) if not np.all(np.isfinite(points_cm)) or np.any(np.abs(points_cm) > 32_767): raise RellisAdmissionError("RELLIS review point escaped signed-centimetre range") flags = ( evaluated[indices].astype(np.uint8) | (ground_truth[indices].astype(np.uint8) << 1) | (current.ground_mask[indices].astype(np.uint8) << 2) | (patchwork.ground_mask[indices].astype(np.uint8) << 3) ) remission = np.asarray(frame.remission[indices], dtype=np.float32) low = float(np.min(remission)) high = float(np.max(remission)) intensity = ( np.zeros(sample_count, dtype=np.uint8) if high <= low else np.clip(np.rint((remission - low) / (high - low) * 255), 0, 255).astype( np.uint8 ) ) path.parent.mkdir(parents=True, exist_ok=True) with tempfile.NamedTemporaryFile(dir=path.parent, delete=False) as temporary: temporary_path = Path(temporary.name) np.savez_compressed( temporary, schema=np.asarray([RELLIS_REVIEW_FRAME_SCHEMA]), frame_id=np.asarray([frame_id]), source_point_count=np.asarray([frame.point_count], dtype=np.int32), xyz_cm=np.ascontiguousarray(points_cm, dtype=" dict[str, Any]: current = _aggregate([frame["nominal"]["current"] for frame in frames]) patchwork = _aggregate([frame["nominal"]["patchworkpp"] for frame in frames]) regressions = [ { "frame_id": frame["frame_id"], "current_ground_iou": frame["nominal"]["current"]["metrics"]["ground_iou"], "patchwork_ground_iou": frame["nominal"]["patchworkpp"]["metrics"]["ground_iou"], "ground_iou_delta": ( frame["nominal"]["patchworkpp"]["metrics"]["ground_iou"] - frame["nominal"]["current"]["metrics"]["ground_iou"] ), "patchwork_natural_ground_recall": frame["nominal"]["patchworkpp"]["metrics"][ "natural_ground_recall" ], } for frame in frames ] checks = [ _check( "ground-iou-gain", patchwork["micro"]["ground_iou"] - current["micro"]["ground_iou"], acceptance.minimum_ground_iou_gain, ">=", ), _check( "obstacle-non-ground-recall", patchwork["micro"]["obstacle_non_ground_recall"], acceptance.minimum_obstacle_non_ground_recall, ">=", ), _check( "obstacle-recall-regression", current["micro"]["obstacle_non_ground_recall"] - patchwork["micro"]["obstacle_non_ground_recall"], acceptance.maximum_obstacle_recall_regression, "<=", ), _check( "assigned-fraction", patchwork["assigned_fraction"], acceptance.minimum_assigned_fraction, ">=", ), _check( "latency-p95-ms", patchwork["latency_ms"]["p95"], acceptance.maximum_patchwork_latency_p95_ms, "<=", ), _check( "catastrophic-regression-count", sum( item["ground_iou_delta"] < -acceptance.maximum_per_frame_iou_regression for item in regressions ), 0, "<=", ), ] passed = all(bool(check["passed"]) for check in checks) return { "schema_version": RELLIS_QUALIFICATION_REPORT_SCHEMA, "identity_sha256": identity, "run_id": run_id, "source_id": RELLIS_SOURCE_ID, "split": "validation", "frame_count": len(frames), "profile": profile, "providers": providers, "aggregates": {"current": current, "patchworkpp": patchwork}, "degradations": {}, "checks": checks, "worst_frames": sorted( regressions, key=lambda item: (item["ground_iou_delta"], item["patchwork_ground_iou"]), )[:20], "frames": frames, "decision": { "status": "cross-dataset-qualified" if passed else "qualification-rejected", "passed": passed, "promoted_to_navigation_or_safety": False, "reason": ( "all predeclared independent-dataset gates passed" if passed else "one or more independent-dataset gates failed" ), }, "safety": { "qualification_only": True, "actuator_authority": False, "navigation_or_safety_accepted": False, }, } def _publish_review_pack( runs_root: Path, *, run_id: str, identity: str, archive_identity: str, preview_points: int, records: list[dict[str, Any]], work_review_root: Path, ) -> None: pack_identity = _canonical_sha256( { "schema_version": RELLIS_REVIEW_PACK_SCHEMA, "source_run_id": run_id, "qualification_identity_sha256": identity, "archive_identity_sha256": archive_identity, "preview_points": preview_points, "sampling": "deterministic-even-index", } ) pack_root = ( runs_root.parent / "polygon-review-packs" / run_id / f"review-{pack_identity}" ) frame_root = pack_root / "frames" frame_root.mkdir(parents=True, exist_ok=True) frames: list[dict[str, Any]] = [] for sequence, record in enumerate(records): frame_id = str(record["frame_id"]) source = work_review_root / f"{frame_id}.npz" target = frame_root / f"{frame_id}.npz" if not target.exists(): try: os.link(source, target) except OSError: target.write_bytes(source.read_bytes()) review = record["review"] if ( target.stat().st_size != review["byte_length"] or _sha256_file(target) != review["sha256"] ): raise RellisAdmissionError("RELLIS review frame changed during publication") nominal = record["nominal"] frames.append( { "sequence": sequence, "frame_id": frame_id, "dataset_sequence_id": record["dataset_sequence_id"], "dataset_frame_number": record["dataset_frame_number"], "source_point_count": record["source_point_count"], "point_count": review["point_count"], "relative_path": f"frames/{frame_id}.npz", "sha256": review["sha256"], "byte_length": review["byte_length"], "current": _review_metrics(nominal["current"]), "patchworkpp": _review_metrics(nominal["patchworkpp"]), "ground_iou_delta": ( nominal["patchworkpp"]["metrics"]["ground_iou"] - nominal["current"]["metrics"]["ground_iou"] ), } ) manifest = { "schema_version": RELLIS_REVIEW_PACK_SCHEMA, "source_run_id": run_id, "identity_sha256": pack_identity, "source_id": RELLIS_SOURCE_ID, "preview_points": preview_points, "frame_count": len(frames), "frames": frames, "safety": { "visualization_only": True, "actuator_authority": False, "navigation_or_safety_accepted": False, }, } _write_json_once(pack_root / "manifest.json", manifest) def _review_metrics(value: dict[str, Any]) -> dict[str, float]: metrics = value["metrics"] return { "ground_iou": float(metrics["ground_iou"]), "natural_ground_recall": float(metrics["natural_ground_recall"]), "obstacle_non_ground_recall": float(metrics["obstacle_non_ground_recall"]), "latency_ms": float(value["latency_ms"]), } def _read_admission(root: Path) -> dict[str, Any]: try: value = json.loads((root / "state/rellis-3d-v1.1.json").read_text(encoding="utf-8")) except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc: raise RellisAdmissionError("RELLIS admission manifest is unavailable") from exc if ( not isinstance(value, dict) or value.get("schema_version") != RELLIS_ADMISSION_SCHEMA or value.get("source_id") != RELLIS_SOURCE_ID or value.get("status") != "dataset-ready" or not isinstance(value.get("release"), dict) or not isinstance(value["release"].get("identity_sha256"), str) ): raise RellisAdmissionError("RELLIS admission manifest is incompatible") return value def _admit_or_resume_run( store: QualificationRunStore, *, run_id: str, identity: str, profile: dict[str, Any], mission_core_commit: str, providers: dict[str, dict[str, Any]], ) -> QualificationRun: scenario = {"source_id": RELLIS_SOURCE_ID, "split": "validation"} patchwork_digest = providers["patchworkpp"].get("binary_sha256") run = QualificationRun( run_id=run_id, episode_id=f"episode-{identity[:20]}", kind=RunKind.REPLAY_SHADOW, state=RunState.ADMITTED, scenario_generation="rellis-3d-validation-v1.1", scenario_sha256=_canonical_sha256(scenario), profile_generation="rellis-ground-current-vs-patchworkpp-v1", profile_sha256=_canonical_sha256(profile), mission_core_commit=mission_core_commit, providers=( ProviderPin( identifier="missioncore-local-percentile-ground", version="v1", revision=mission_core_commit, ), ProviderPin( identifier="patchworkpp", version="v1.4.1", revision=PATCHWORKPP_SOURCE_COMMIT, digest=str(patchwork_digest) if patchwork_digest is not None else None, ), ), host_profile_id="simulation-worker-rellis-ground-v1", host_profile_sha256=_canonical_sha256( {"role": "simulation-worker", "storage": "worker-d-only"} ), seed=42, reproducibility_tier=ReproducibilityTier.R1, authority=AuthorityProfile( generation=1, command_ttl_max_ns=1, heartbeat_timeout_monotonic_ns=1, ), clock_domain="dataset:frame-index", created_at_utc=_utc_now(), ) try: return store.create(run) except QualificationRunConflictError as exc: existing = store.load(run_id) if ( existing.profile_sha256 != run.profile_sha256 or existing.scenario_sha256 != run.scenario_sha256 or existing.mission_core_commit != mission_core_commit ): raise RellisAdmissionError("existing RELLIS run has another identity") from exc return existing def _ensure_running(store: QualificationRunStore, run: QualificationRun) -> QualificationRun: if run.state is RunState.ADMITTED: run = store.transition( run.run_id, RunState.STARTING, expected_revision=run.revision, observed_at_utc=_utc_now(), host_monotonic_ns=time.monotonic_ns(), ) if run.state is RunState.STARTING: run = store.transition( run.run_id, RunState.RUNNING, expected_revision=run.revision, observed_at_utc=_utc_now(), host_monotonic_ns=time.monotonic_ns(), ) if run.state is not RunState.RUNNING: raise RellisAdmissionError("RELLIS qualification cannot resume from this state") return run def _register_artifact( store: QualificationRunStore, run_id: str, path: Path, run_root: Path, *, artifact_id: str, kind: str, ) -> None: relative = path.relative_to(run_root).as_posix() store.register_artifact( run_id, QualificationArtifact( artifact_id=artifact_id, kind=kind, relative_path=relative, sha256=_sha256_file(path), byte_length=path.stat().st_size, source_of_record=True, ), ) def _read_completed_report( store: QualificationRunStore, run: QualificationRun, ) -> dict[str, Any]: artifacts = [ item for item in run.artifacts if item.kind == RELLIS_REPORT_ARTIFACT_KIND ] if len(artifacts) != 1: raise RellisAdmissionError("completed RELLIS run has no report") path = store.root / run.run_id / artifacts[0].relative_path if _sha256_file(path) != artifacts[0].sha256: raise RellisAdmissionError("completed RELLIS report digest differs") value = json.loads(path.read_text(encoding="utf-8")) if not isinstance(value, dict): raise RellisAdmissionError("completed RELLIS report is invalid") return value def _read_cached(path: Path, identity: str) -> dict[str, Any] | None: if not path.is_file(): return None try: value = json.loads(path.read_text(encoding="utf-8")) except (OSError, UnicodeDecodeError, json.JSONDecodeError): return None if ( not isinstance(value, dict) or value.get("schema_version") != RELLIS_QUALIFICATION_FRAME_SCHEMA or value.get("identity_sha256") != identity ): return None return value def _frame_identity(point_member: str) -> tuple[str, int]: parts = point_member.split("/") if ( len(parts) != 3 or parts[0] not in {"00000", "00001", "00002", "00003", "00004"} or not parts[2].endswith(".bin") or not parts[2][:-4].isdigit() ): raise RellisAdmissionError("RELLIS frame identity is incompatible") return parts[0], int(parts[2][:-4]) def _frame_id(point_member: str) -> str: sequence, frame = _frame_identity(point_member) return f"rellis-{sequence}-{frame:06d}" def _canonical_sha256(value: dict[str, Any]) -> str: return hashlib.sha256( json.dumps(value, sort_keys=True, separators=(",", ":")).encode() ).hexdigest() def _sha256_file(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as source: for chunk in iter(lambda: source.read(1024**2), b""): digest.update(chunk) return digest.hexdigest() def _write_json_once(path: Path, value: dict[str, Any]) -> None: encoded = json.dumps(value, sort_keys=True, separators=(",", ":")).encode() + b"\n" if path.exists(): if path.is_file() and path.read_bytes() == encoded: return raise RellisAdmissionError("immutable RELLIS artifact already exists") path.parent.mkdir(parents=True, exist_ok=True) with tempfile.NamedTemporaryFile(dir=path.parent, delete=False) as temporary: temporary_path = Path(temporary.name) temporary.write(encoded) temporary.flush() os.fsync(temporary.fileno()) os.replace(temporary_path, path) def _utc_now() -> str: return datetime.now(UTC).isoformat(timespec="milliseconds").replace("+00:00", "Z")