"""Pinned RELLIS-3D compatibility smoke for the shared Dataset Gateway. The smoke check deliberately stops before algorithm qualification. It proves that Mission Core can read one official Ouster OS1 SemanticKITTI frame, preserve point/label alignment, apply an explicit ground-evaluation policy and publish a bounded visualization artifact. Full RELLIS archives and ROS bags remain worker-D-only inputs. """ from __future__ import annotations import hashlib import json import os import tempfile from collections import Counter from dataclasses import dataclass from pathlib import Path from typing import Any, Final, Literal import numpy as np from k1link.datasets.gateway import DatasetFrameError, DatasetPointFrame, read_semantic_kitti_frame RELLIS_SOURCE_ID: Final = "rellis-3d/v1.1" RELLIS_PREVIEW_SCHEMA: Final = "missioncore.dataset-native-scan-preview/v2" RELLIS_GROUND_POLICY_SCHEMA: Final = "missioncore.rellis-ground-target-policy/v1" RELLIS_REPOSITORY_URL: Final = "https://github.com/unmannedlab/RELLIS-3D" RELLIS_REPOSITORY_COMMIT: Final = "c17a118fcaed1559f03cc32cc3a91dedc557f8b8" RELLIS_LABEL_CONFIG_SHA256: Final = ( "573379a232ac561805987466c391a61fd5ad338be7fb9c4c2842f3f28067e0ad" ) RELLIS_EXAMPLE_POINTS_SHA256: Final = ( "ed81a9c3636d55b17d78058c72545d5d22419beecf174d50596d23ae178752af" ) RELLIS_EXAMPLE_LABELS_SHA256: Final = ( "9b8c65b710873e931af4ac6dfc7d3dd2298696514ab721e50300bd55ad5b634e" ) RELLIS_EXAMPLE_FRAME_ID: Final = "000104" RELLIS_LICENSE: Final = "CC-BY-NC-SA-3.0" MAX_RELLIS_PREVIEW_POINTS: Final = 50_000 GroundTarget = Literal["ground", "non-ground", "ignore"] @dataclass(frozen=True) class RellisClass: label_id: int class_name: str bgr: tuple[int, int, int] ground_target: GroundTarget @property def rgb(self) -> tuple[int, int, int]: blue, green, red = self.bgr return red, green, blue @property def hex(self) -> str: red, green, blue = self.rgb return f"#{red:02x}{green:02x}{blue:02x}" def to_preview_dict(self) -> dict[str, object]: return { "label_id": self.label_id, "class_name": self.class_name, "hex": self.hex, "ground_target": self.ground_target, } # IDs, names and BGR colors are pinned to the official repository config at # RELLIS_REPOSITORY_COMMIT. The final column is Mission Core's versioned, # algorithm-scoped ground evaluation policy; it is not an upstream claim. RELLIS_CLASSES: Final[tuple[RellisClass, ...]] = ( RellisClass(0, "void", (0, 0, 0), "ignore"), RellisClass(1, "dirt", (108, 64, 20), "ground"), RellisClass(3, "grass", (0, 102, 0), "ground"), RellisClass(4, "tree", (0, 255, 0), "non-ground"), RellisClass(5, "pole", (0, 153, 153), "non-ground"), RellisClass(6, "water", (0, 128, 255), "ignore"), RellisClass(7, "sky", (0, 0, 255), "ignore"), RellisClass(8, "vehicle", (255, 255, 0), "non-ground"), RellisClass(9, "object", (255, 0, 127), "ignore"), RellisClass(10, "asphalt", (64, 64, 64), "ground"), RellisClass(12, "building", (255, 0, 0), "non-ground"), RellisClass(15, "log", (102, 0, 0), "non-ground"), RellisClass(17, "person", (204, 153, 255), "non-ground"), RellisClass(18, "fence", (102, 0, 204), "non-ground"), RellisClass(19, "bush", (255, 153, 204), "non-ground"), RellisClass(23, "concrete", (170, 170, 170), "ground"), RellisClass(27, "barrier", (41, 121, 255), "non-ground"), RellisClass(31, "puddle", (134, 255, 239), "ignore"), RellisClass(33, "mud", (99, 66, 34), "ground"), RellisClass(34, "rubble", (110, 22, 138), "non-ground"), ) RELLIS_CLASS_BY_ID: Final = {item.label_id: item for item in RELLIS_CLASSES} class RellisSmokeError(RuntimeError): """The pinned official example cannot satisfy the RELLIS smoke contract.""" def build_rellis_official_smoke_preview( point_path: Path, label_path: Path, output_path: Path, *, preview_points: int = 20_000, ) -> dict[str, Any]: """Verify the official example and publish one bounded path-free preview.""" points_digest = _sha256_file(point_path) labels_digest = _sha256_file(label_path) if points_digest != RELLIS_EXAMPLE_POINTS_SHA256: raise RellisSmokeError("RELLIS official example point digest is incompatible") if labels_digest != RELLIS_EXAMPLE_LABELS_SHA256: raise RellisSmokeError("RELLIS official example label digest is incompatible") try: frame = read_semantic_kitti_frame(point_path, label_path) except DatasetFrameError as exc: raise RellisSmokeError("RELLIS official example violates XYZI/label alignment") from exc preview = rellis_native_scan_preview( frame, frame_id=RELLIS_EXAMPLE_FRAME_ID, maximum_points=preview_points, source_evidence={ "repository_url": RELLIS_REPOSITORY_URL, "repository_commit": RELLIS_REPOSITORY_COMMIT, "label_config_sha256": RELLIS_LABEL_CONFIG_SHA256, "point_sha256": points_digest, "label_sha256": labels_digest, "license": RELLIS_LICENSE, }, ) _atomic_json(output_path, preview) return preview def rellis_native_scan_preview( frame: DatasetPointFrame, *, frame_id: str, maximum_points: int, source_evidence: dict[str, str], compatibility_smoke_only: bool = True, ) -> dict[str, Any]: """Build the common viewer artifact from a validated RELLIS native frame.""" if not frame_id or not 1 <= maximum_points <= MAX_RELLIS_PREVIEW_POINTS: raise RellisSmokeError("RELLIS preview bounds are incompatible") unknown = sorted( int(value) for value in np.unique(frame.semantic_labels) if int(value) not in RELLIS_CLASS_BY_ID ) if unknown: raise RellisSmokeError("RELLIS frame contains labels absent from the pinned ontology") required_evidence = { "repository_url", "repository_commit", "label_config_sha256", "point_sha256", "label_sha256", "license", } if set(source_evidence) != required_evidence or any( not isinstance(value, str) or not value for value in source_evidence.values() ): raise RellisSmokeError("RELLIS source evidence is incomplete") sample_count = min(frame.point_count, maximum_points) indices = np.linspace(0, frame.point_count - 1, sample_count, dtype=np.int64) points = frame.points_xyz_m[indices] remission = frame.remission[indices] semantic = frame.semantic_labels[indices] remission_min = float(np.min(remission)) remission_max = float(np.max(remission)) remission_span = remission_max - remission_min if remission_span <= 0: remission_u8 = np.zeros(sample_count, dtype=np.uint8) else: remission_u8 = np.rint( (remission - remission_min) / remission_span * 255.0 ).astype(np.uint8) ground = np.zeros(sample_count, dtype=np.uint8) evaluated = np.zeros(sample_count, dtype=np.uint8) colors = np.zeros((sample_count, 3), dtype=np.uint8) for label_id in np.unique(semantic): item = RELLIS_CLASS_BY_ID[int(label_id)] mask = semantic == label_id colors[mask] = item.rgb if item.ground_target != "ignore": evaluated[mask] = 1 if item.ground_target == "ground": ground[mask] = 1 present = Counter(int(value) for value in frame.semantic_labels) present_classes = [ { **RELLIS_CLASS_BY_ID[label_id].to_preview_dict(), "source_point_count": count, } for label_id, count in sorted(present.items()) ] return { "schema_version": RELLIS_PREVIEW_SCHEMA, "source_id": RELLIS_SOURCE_ID, "frame_id": frame_id, "representation": "native-scan", "sampling": "deterministic-even-index", "source_point_count": frame.point_count, "point_count": sample_count, "points_xyz_m": points.tolist(), "remission_0_to_255": remission_u8.tolist(), "semantic_label_ids": semantic.astype(np.uint16).tolist(), "semantic_rgb_0_to_255": colors.reshape(-1).tolist(), "ground_truth_ground": ground.tolist(), "evaluation_mask": evaluated.tolist(), "classes": present_classes, "coordinate_frame": { "frame_id": "sensor/lidar/os1", "handedness": "right", "x": "forward", "y": "left", "z": "up", "transform_applied": False, }, "ground_policy": { "schema_version": RELLIS_GROUND_POLICY_SCHEMA, "ground": [ item.class_name for item in RELLIS_CLASSES if item.ground_target == "ground" ], "non_ground": [ item.class_name for item in RELLIS_CLASSES if item.ground_target == "non-ground" ], "ignore": [ item.class_name for item in RELLIS_CLASSES if item.ground_target == "ignore" ], }, "source_evidence": source_evidence, "safety": { "visualization_only": True, "compatibility_smoke_only": compatibility_smoke_only, "navigation_or_safety_accepted": False, }, } def _sha256_file(path: Path) -> str: digest = hashlib.sha256() try: with path.open("rb") as source: for chunk in iter(lambda: source.read(1024**2), b""): digest.update(chunk) except OSError as exc: raise RellisSmokeError("RELLIS official example is unavailable") from exc return digest.hexdigest() def _atomic_json(path: Path, document: dict[str, Any]) -> None: path.parent.mkdir(parents=True, exist_ok=True) with tempfile.NamedTemporaryFile( mode="w", dir=path.parent, encoding="utf-8", delete=False, ) as temporary: temporary_path = Path(temporary.name) json.dump(document, temporary, ensure_ascii=False, separators=(",", ":")) temporary.flush() os.fsync(temporary.fileno()) os.replace(temporary_path, path)