from __future__ import annotations import hashlib import json from pathlib import Path from typing import Any import numpy as np from fastapi import APIRouter from fastapi.routing import APIRoute from k1link.compute import e30_materialization as materialization from k1link.compute.e30_materialization import build_e30_materialization from k1link.compute.semantic_geometry_fusion import ( CameraGeometryFusionProfile, _projection_profile, _semantic_support, ) from k1link.device_plugins.xgrids_k1.analyze.calibrated_projection import ( project_map_points_kb4, ) from k1link.web.e30_review_api import build_e30_review_router def _canonical(value: object) -> bytes: return json.dumps( value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False, ).encode() def _sha256(path: Path) -> str: return hashlib.sha256(path.read_bytes()).hexdigest() def _endpoint(router: APIRouter, path: str) -> object: for route in router.routes: if ( isinstance(route, APIRoute) and route.path == path and route.methods is not None and "GET" in route.methods ): return route.endpoint raise AssertionError(f"GET {path} route is missing") class _FakeSource: def __init__(self, root: Path) -> None: self.root = root self.pack_id = root.name self.arrays = { "cloud_offsets": np.asarray([0, 4], dtype=np.int64), "cloud_points_map": np.asarray( [ [0.00, 0.00, 2.0], [0.10, 0.00, 2.0], [0.20, 0.00, 2.0], [1.50, 1.50, 2.0], ], dtype=np.float32, ), "pose_positions_map": np.zeros((1, 3), dtype=np.float64), "pose_quaternions_map_from_lidar": np.asarray( [[0.0, 0.0, 0.0, 1.0]], dtype=np.float64, ), "sample_available": np.asarray([True], dtype=np.bool_), "source_frame_indices": np.asarray([10], dtype=np.int64), "session_seconds": np.asarray([12.5], dtype=np.float64), "intrinsic_fx_fy_cx_cy": np.asarray( [100.0, 100.0, 50.0, 50.0], dtype=np.float64, ), "distortion_kb4": np.zeros(4, dtype=np.float64), "t_camera_from_lidar": np.eye(4, dtype=np.float64), } self.identity: dict[str, Any] = { "session_id": "source-session", "frame_count": 1, "point_count": 4, "source_id": "sensor.camera.right", "camera_slot": "camera_1", "projection": {"width": 100, "height": 100}, } self.manifest = {"artifact": {"sha256": "1" * 64}} @property def frame_count(self) -> int: return 1 @property def point_count(self) -> int: return 4 def close(self) -> None: pass class _FakeSurface: def __init__(self, root: Path) -> None: self.root = root self.model_id = root.name self.identity: dict[str, Any] = { "source_pack_id": "e10-lidar-pack-" + "b" * 64, "frame_count": 1, "point_count": 4, } self.arrays = { "frame_valid": np.asarray([True], dtype=np.bool_), "point_class": np.asarray([2, 2, 2, 1], dtype=np.uint8), "point_height_m": np.asarray([0.5, 0.5, 0.5, 0.0], dtype=np.float32), } self.manifest = { "artifacts": [ {"role": "local-surface", "sha256": "2" * 64}, ] } def close(self) -> None: pass def _write_source_tree(tmp_path: Path) -> tuple[dict[str, Path], str]: source_pack_id = "e10-lidar-pack-" + "b" * 64 local_surface_id = "k1-local-surface-" + "c" * 64 source_result_identity = {"schema_version": "test-source/v1"} source_result_identity_sha256 = hashlib.sha256( _canonical(source_result_identity) ).hexdigest() source_result_id = ( f"e10-integrated-perception-{source_result_identity_sha256}" ) roots = { "e29": tmp_path / "e29", "source_results": tmp_path / "source-results", "source_packs": tmp_path / "source-packs", "surfaces": tmp_path / "surfaces", "reviews": tmp_path / "reviews", "output": tmp_path / "output", } for root in roots.values(): root.mkdir() (roots["source_packs"] / source_pack_id).mkdir() (roots["surfaces"] / local_surface_id).mkdir() source_result = roots["source_results"] / source_result_id source_result.mkdir() result_document = { "result_id": source_result_id, "identity_sha256": source_result_identity_sha256, "identity": source_result_identity, } (source_result / "result.json").write_bytes(_canonical(result_document)) fusion_frame = { "schema_version": "missioncore.e10-fusion-frame/v1", "frame_index": 0, "source_frame_index": 10, "session_seconds": 12.5, "objects": [ { "source_track_id": 7, "track_id": 70, "label": "car", "association_group": "vehicle", "score": 0.9, "bbox_xyxy": [40.0, 40.0, 70.0, 60.0], "cuboid_status": "observed", "camera_motion_state": "static", "camera_motion_confidence": 0.8, "motion_state": "unknown", "motion_status": "test", } ], } fusion_path = source_result / "fusion-frames.jsonl" fusion_path.write_bytes(_canonical(fusion_frame) + b"\n") source = _FakeSource(roots["source_packs"] / source_pack_id) profile = CameraGeometryFusionProfile() points = np.asarray(source.arrays["cloud_points_map"], dtype=np.float64) projected = project_map_points_kb4( points, position_map_xyz=(0.0, 0.0, 0.0), orientation_map_from_lidar_xyzw=(0.0, 0.0, 0.0, 1.0), profile=_projection_profile(source), # type: ignore[arg-type] ) snapshot = _semantic_support( fusion_frame["objects"][0], projected=projected, frame_points_map=points, point_class=np.asarray([2, 2, 2, 1], dtype=np.uint8), point_height_m=np.asarray([0.5, 0.5, 0.5, 0.0], dtype=np.float32), source_available=True, surface_valid=True, profile=profile, ).document assert snapshot["geometry_status"] == "agree" e29_identity = { "schema_version": "missioncore.e29-camera-geometry-fusion/v1", "source_result_id": source_result_id, "source_fusion_frames_sha256": _sha256(fusion_path), "source_pack_id": source_pack_id, "local_surface_model_id": local_surface_id, "frame_count": 1, "profile": profile.to_dict(), "authority": { "commands_enabled": False, "navigation_or_safety_accepted": False, }, } e29_identity_sha256 = hashlib.sha256(_canonical(e29_identity)).hexdigest() e29_result_id = f"e29-camera-geometry-{e29_identity_sha256}" e29_result = roots["e29"] / e29_result_id e29_result.mkdir() e29_manifest = { "schema_version": "missioncore.e29-camera-geometry-fusion/v1", "result_id": e29_result_id, "identity_sha256": e29_identity_sha256, "identity": e29_identity, } (e29_result / "manifest.json").write_bytes(_canonical(e29_manifest)) review_item_id = "e30-review-item-" + "d" * 64 review_item = { "schema_version": "missioncore.e30-evidence-review-item/v1", "sequence": 0, "item_id": review_item_id, "review_key": "semantic:0:0", "stratum": "agree", "range_bucket": "near", "evidence_binding": { "frame_index": 0, "source_frame_index": 10, "session_seconds": 12.5, }, "e29_locator": { "kind": "semantic-observation", "observation_index": 0, }, "e29_snapshot": snapshot, "review": {"state": "unreviewed", "reason_code": None, "notes": None}, "authority": { "commands_enabled": False, "navigation_or_safety_accepted": False, }, } review_items = _canonical(review_item) + b"\n" review_identity = { "source": { "e29_result_id": e29_result_id, "e29_identity_sha256": e29_identity_sha256, "camera_result_id": source_result_id, "lidar_pack_id": source_pack_id, "local_surface_model_id": local_surface_id, }, "reason_taxonomy": [ "no_lidar_observation", "outside_lidar_support", "outside_camera_fov", "time_mismatch", "semantic_mismatch", "geometry_mismatch", "insufficient_evidence", "other", ], } review_identity_sha256 = hashlib.sha256(_canonical(review_identity)).hexdigest() review_result_id = f"e30-review-pack-{review_identity_sha256}" review_root = roots["reviews"] / review_result_id review_root.mkdir() items_path = review_root / "review-items.jsonl" items_path.write_bytes(review_items) review_manifest = { "schema_version": "missioncore.e30-evidence-review-pack/v1", "result_id": review_result_id, "identity_sha256": review_identity_sha256, "identity": review_identity, "human_review_complete": False, "lab_published": False, "selected_item_count": 1, "artifacts": [ { "role": "review-items", "path": items_path.name, "byte_length": items_path.stat().st_size, "sha256": _sha256(items_path), } ], "authority": { "commands_enabled": False, "navigation_or_safety_accepted": False, }, } (review_root / "manifest.json").write_bytes(_canonical(review_manifest)) roots["review_root"] = review_root return roots, review_item_id def test_materialization_replays_exact_support_and_publishes_point_indices( tmp_path: Path, monkeypatch: Any, ) -> None: roots, review_item_id = _write_source_tree(tmp_path) monkeypatch.setattr(materialization, "E10LidarFieldSource", _FakeSource) monkeypatch.setattr(materialization, "K1LocalSurfaceV1", _FakeSurface) result = build_e30_materialization( review_pack_root=roots["review_root"], e29_root=roots["e29"], source_result_root=roots["source_results"], source_pack_root=roots["source_packs"], local_surface_root=roots["surfaces"], output_root=roots["output"], ) assert result.manifest["item_count"] == 1 assert result.manifest["human_review_complete"] is False index = json.loads( (result.result_root / "materialized-items.jsonl").read_text() ) assert index["item_id"] == review_item_id assert index["materialization"]["selected_point_count"] == 3 assert index["materialization"]["source_reprojection_required"] is False artifact = result.result_root / index["artifact"]["path"] with np.load(artifact, allow_pickle=False) as arrays: assert arrays["selected_source_indices"].tolist() == [0, 1, 2] assert arrays["projected_selected_mask"].sum() == 3 assert arrays["projected_candidate_mask"].sum() >= 3 def test_e30_review_api_exposes_verified_read_only_evidence( tmp_path: Path, monkeypatch: Any, ) -> None: roots, review_item_id = _write_source_tree(tmp_path) monkeypatch.setattr(materialization, "E10LidarFieldSource", _FakeSource) monkeypatch.setattr(materialization, "K1LocalSurfaceV1", _FakeSurface) result = build_e30_materialization( review_pack_root=roots["review_root"], e29_root=roots["e29"], source_result_root=roots["source_results"], source_pack_root=roots["source_packs"], local_surface_root=roots["surfaces"], output_root=roots["output"], ) router = build_e30_review_router( materialization_root_provider=lambda: roots["output"], review_pack_root_provider=lambda: roots["reviews"], ) catalog_route = _endpoint(router, "/api/v1/laboratory/e30/reviews") items_route = _endpoint( router, "/api/v1/laboratory/e30/reviews/{result_id}/items", ) detail_route = _endpoint( router, "/api/v1/laboratory/e30/reviews/{result_id}/items/{item_id}", ) catalog = catalog_route(limit=1) # type: ignore[operator] items = items_route( # type: ignore[operator] result_id=result.result_id, stratum="agree", limit=48, cursor=0, ) detail = detail_route( # type: ignore[operator] result_id=result.result_id, item_id=review_item_id, ) assert catalog["configured"] is True assert catalog["items"][0]["access"] == "read-only" assert catalog["items"][0]["stratum_counts"]["agree"] == 1 assert items["total"] == 1 assert items["items"][0]["item_id"] == review_item_id assert detail["item"]["selected"]["source_indices"] == [0, 1, 2] assert detail["item"]["projection"]["selected_mask"] == [1, 1, 1] assert str(tmp_path) not in repr(catalog) assert str(tmp_path) not in repr(items) assert str(tmp_path) not in repr(detail) artifact = result.result_root / "items" / f"{review_item_id}.npz" artifact.write_bytes(artifact.read_bytes() + b"changed") tampered = catalog_route(limit=1) # type: ignore[operator] assert tampered["candidate_total"] == 1 assert tampered["invalid_total"] == 1 assert tampered["items"] == []