from __future__ import annotations import hashlib import io import json import shutil import zipfile from pathlib import Path from types import SimpleNamespace import numpy as np from fastapi import FastAPI from fastapi.testclient import TestClient from PIL import Image import k1link.laboratory.vegetation_policy_review as policy_review_module import k1link.laboratory.vegetation_policy_video as policy_video_module import k1link.laboratory.vegetation_shadow_lab as vegetation_lab_module from k1link.laboratory import LaboratoryEvidenceRegistry from k1link.laboratory.evidence_report import verify_laboratory_evidence_result from k1link.laboratory.vegetation_policy_review import seal_vegetation_policy_review from k1link.laboratory.vegetation_shadow_lab import seal_vegetation_shadow_lab from k1link.web.vegetation_shadow_lab_api import build_vegetation_shadow_lab_router REPOSITORY_ROOT = Path(__file__).resolve().parents[1] def test_coarse_policy_masks_mark_every_outside_fov_pixel_undefined( tmp_path: Path, monkeypatch, ) -> None: monkeypatch.setattr(policy_video_module, "FRAME_COUNT", 1) monkeypatch.setattr( policy_video_module, "fine_to_policy_lut", lambda _taxonomy, _provider_map: np.full(256, 4, dtype=np.uint8), ) source = tmp_path / "fine.zip" fine_buffer = io.BytesIO() Image.new("L", (800, 600), color=1).save(fine_buffer, format="PNG") with zipfile.ZipFile(source, "w") as archive: archive.writestr("masks/frame-000001.png", fine_buffer.getvalue()) valid_fov = np.zeros((600, 800), dtype=np.uint8) valid_fov[:, :400] = 255 valid_fov_path = tmp_path / "valid-fov.png" Image.fromarray(valid_fov, mode="L").save(valid_fov_path) destination = tmp_path / "coarse.zip" counts = policy_video_module.build_policy_mask_archive( source_archive=source, destination_archive=destination, fine_taxonomy={}, provider_label_map={}, valid_fov_mask=valid_fov_path, ) with ( zipfile.ZipFile(destination) as archive, Image.open(io.BytesIO(archive.read("masks/frame-000001.png"))) as image, ): coarse = np.asarray(image.convert("L")) assert np.all(coarse[:, :400] == 4) assert np.all(coarse[:, 400:] == 9) assert counts[4] == 600 * 400 assert counts[9] == 600 * 400 def _sha256(path: Path) -> str: return hashlib.sha256(path.read_bytes()).hexdigest() def _worker_result(root: Path, *, candidate: str, mode: str, vegetation_iou: float) -> None: root.mkdir(parents=True) visual_cases = [] for index in range(12): case_id = f"case-{index:02d}" case_root = root / "cases" / case_id case_root.mkdir(parents=True) keys = ["source", "prediction_semantic", "policy_urban", "policy_rural", "policy_offroad"] if mode == "goose": keys.extend(("truth_semantic", "vegetation_material_error")) files = {} for key in keys: path = case_root / f"{key}.png" path.write_bytes(b"\x89PNG\r\n\x1a\n" + f"{candidate}:{mode}:{case_id}:{key}".encode()) files[key] = { "relative_path": path.relative_to(root).as_posix(), "sha256": _sha256(path), } visual_cases.append( { "case_id": case_id, "source_width": 800 if mode == "ravnoves" else 512, "source_height": 600 if mode == "ravnoves" else 512, "center_crop_xyxy": [100, 0, 700, 600] if mode == "ravnoves" else [0, 0, 512, 512], "outside_crop_state": "undefined" if mode == "ravnoves" else "not-applicable", "focus": { "class_name": "high_grass", "label_id": 51, "truth_pixels": 16384, "truth_fraction": 0.0625, "stratum_rank": index + 1, } if mode == "goose" else None, "files": files, } ) payload = { "schema_version": "missioncore.lab-v1-goose-vegetation-run/v1", "result_id": f"lab-v1-{mode}-{candidate}-fixture", "mode": mode, "candidate": { "candidate_key": candidate, "loaded_model_name": "ddrnet_39" if candidate == "ddrnet" else "pp_lite_t_seg", "checkpoint_sha256": ("a" if candidate == "ddrnet" else "b") * 64, }, "metrics": { "mean_iou_percent": 44.0 + vegetation_iou, "published_mean_iou_percent": 46.53 if candidate == "ddrnet" else 45.09, "vegetation_mean_iou": vegetation_iou, }, "timing": { "latency_ms_p95": 20.0 if candidate == "ddrnet" else 15.0, "throughput_fps_from_mean_inference": 55.0, }, "resource": { "peak_reserved_vram_bytes": 2_000_000_000, "gpu_name": "fixture RTX 4090", }, "visual_cases": visual_cases, "authority": { "navigation_accepted": False, "safety_accepted": False, "actuation_accepted": False, "camera_semantics_can_clear_rigid_geometry": False, }, } (root / "result.json").write_text(json.dumps(payload), encoding="utf-8") def _video_worker_result(root: Path) -> None: root.mkdir(parents=True) archive = root / "semantic-masks.zip" mask = b"\x89PNG\r\n\x1a\n" with zipfile.ZipFile(archive, "x", compression=zipfile.ZIP_STORED) as frozen: for sequence in range(4489): frozen.writestr(f"masks/frame-{sequence + 1:06d}.png", mask) taxonomy = { "schema_version": "missioncore.lab-v1-vegetation-taxonomy/v1", "classes": [ { "class_id": class_id, "label": "undefined" if class_id == 0 else f"class-{class_id}", "color_rgb": [class_id, class_id, class_id], "disposition": "undefined" if class_id == 0 else "prediction", } for class_id in range(64) ], } payload = { "schema_version": "missioncore.lab-v1-goose-vegetation-run/v1", "result_id": f"lab-v1-ravnoves-video-ddrnet-{'e' * 64}", "mode": "ravnoves-video", "candidate": {"candidate_key": "ddrnet"}, "source": { "input_count": 4489, "ground_truth_available": False, }, "video_semantics": { "base_m4_result_id": f"m4-threat-replay-{'f' * 64}", "mask_archive": { "path": "semantic-masks.zip", "sha256": _sha256(archive), "byte_length": archive.stat().st_size, "frame_count": 4489, "width": 800, "height": 600, "encoding": "uint8-class-id-png", "media_type": "application/zip", "sequence_binding": "sequence-0-to-masks/frame-000001.png", }, "taxonomy": taxonomy, "aggregate_prediction_pixels": [4489 * 800 * 600, *([0] * 63)], "center_crop_xyxy": [100, 0, 700, 600], "outside_crop_state": "undefined", }, "authority": { "navigation_accepted": False, "safety_accepted": False, "actuation_accepted": False, "camera_semantics_can_clear_rigid_geometry": False, }, } (root / "result.json").write_text(json.dumps(payload), encoding="utf-8") def test_vegetation_shadow_lab_seals_autonomous_visual_evidence( tmp_path: Path, monkeypatch, ) -> None: roots = {} for candidate, vegetation_iou in (("ddrnet", 0.64), ("ppliteseg", 0.61)): for mode in ("goose", "ravnoves"): root = tmp_path / "worker" / f"{candidate}-{mode}" _worker_result(root, candidate=candidate, mode=mode, vegetation_iou=vegetation_iou) roots[(candidate, mode)] = root video_root = tmp_path / "worker" / "ddrnet-ravnoves-video" _video_worker_result(video_root) m47_root = tmp_path / f"m47-reference-graph-lab-{'a' * 64}" m47_root.mkdir() monkeypatch.setattr( vegetation_lab_module, "read_m47_reference_graph_lab", lambda _root: SimpleNamespace( result_id=m47_root.name, report={ "source": {"source_id": "RAVNOVES00"}, "visual_evidence": { "linked_result_id": f"m4-threat-replay-{'f' * 64}", "timeline_frames": 4489, }, }, ), ) result_root = seal_vegetation_shadow_lab( ddrnet_goose_root=roots[("ddrnet", "goose")], ppliteseg_goose_root=roots[("ppliteseg", "goose")], ddrnet_ravnoves_root=roots[("ddrnet", "ravnoves")], ppliteseg_ravnoves_root=roots[("ppliteseg", "ravnoves")], output_root=tmp_path / "results", ddrnet_ravnoves_video_root=video_root, m47_reference_graph_lab_root=m47_root, ) manifest = json.loads((result_root / "result.json").read_text("utf-8")) assert manifest["decision"]["selected_candidate"] == "ddrnet" assert manifest["ground_truth"] is False assert manifest["authority"]["commands_enabled"] is False assert manifest["authority"]["navigation_or_safety_accepted"] is False assert len(manifest["catalogs"]["ravnoves"]) == 0 assert len(manifest["catalogs"]["goose"]) == 12 assert len(manifest["artifacts"]) == 78 assert manifest["route_video"]["frame_count"] == 4489 assert manifest["route_video"]["outside_crop_state"] == "undefined" assert manifest["catalogs"]["goose"][0]["focus"]["class_name"] == "high_grass" assert "ddrnet_error" in manifest["catalogs"]["goose"][0]["assets"] assert "ppliteseg_error" in manifest["catalogs"]["goose"][0]["assets"] assert "all_classes" not in manifest["metrics"]["candidates"]["ddrnet"]["validation_metrics"] assert (result_root / "result.json").stat().st_size <= 64 * 1024 registry = LaboratoryEvidenceRegistry.from_directory(REPOSITORY_ROOT / "config/laboratories") definition = next( row for row in registry.definitions if row.work_id == "lab-v1-vegetation-shadow" ) proof = verify_laboratory_evidence_result(definition, result_root) assert proof["result_id"] == result_root.name assert proof["artifact_count"] == 78 app = FastAPI() app.include_router(build_vegetation_shadow_lab_router(root_provider=lambda: result_root.parent)) client = TestClient(app) response = client.get(f"/api/v1/laboratory/vegetation-shadow/{result_root.name}") assert response.status_code == 200 assert response.json()["access"] == "read-only" asset_path = manifest["catalogs"]["goose"][0]["assets"]["ddrnet_error"]["path"] asset = client.get( f"/api/v1/laboratory/vegetation-shadow/{result_root.name}/assets/{asset_path}" ) assert asset.status_code == 200 assert asset.headers["cache-control"].endswith("immutable") mask = client.get(f"/api/v1/laboratory/vegetation-shadow/{result_root.name}/masks/0") assert mask.status_code == 200 assert mask.content == b"\x89PNG\r\n\x1a\n" assert mask.headers["cache-control"].endswith("immutable") (result_root / asset_path).write_bytes(b"tampered") assert ( client.get(f"/api/v1/laboratory/vegetation-shadow/{result_root.name}").status_code == 503 ) def test_policy_review_reuses_sealed_video_and_links_yolox_tgs( tmp_path: Path, monkeypatch, ) -> None: roots = {} for candidate, vegetation_iou in (("ddrnet", 0.64), ("ppliteseg", 0.61)): for mode in ("goose", "ravnoves"): root = tmp_path / "worker" / f"{candidate}-{mode}" _worker_result(root, candidate=candidate, mode=mode, vegetation_iou=vegetation_iou) roots[(candidate, mode)] = root video_root = tmp_path / "worker" / "ddrnet-ravnoves-video" _video_worker_result(video_root) m47_root = tmp_path / f"m47-reference-graph-lab-{'a' * 64}" m47_root.mkdir() base_m4_result_id = f"m4-threat-replay-{'f' * 64}" monkeypatch.setattr( vegetation_lab_module, "read_m47_reference_graph_lab", lambda _root: SimpleNamespace( result_id=m47_root.name, report={ "source": {"source_id": "RAVNOVES00"}, "visual_evidence": { "linked_result_id": base_m4_result_id, "timeline_frames": 4489, }, }, ), ) base_root = seal_vegetation_shadow_lab( ddrnet_goose_root=roots[("ddrnet", "goose")], ppliteseg_goose_root=roots[("ppliteseg", "goose")], ddrnet_ravnoves_root=roots[("ddrnet", "ravnoves")], ppliteseg_ravnoves_root=roots[("ppliteseg", "ravnoves")], output_root=tmp_path / "results", ddrnet_ravnoves_video_root=video_root, m47_reference_graph_lab_root=m47_root, ) tgs_result_id = f"m49-tgs-full-shadow-{'9' * 64}" monkeypatch.setattr( policy_review_module, "read_m49_tgs_full_shadow", lambda _root: SimpleNamespace( result_id=tgs_result_id, report={ "source": { "source_id": "RAVNOVES00", "linked_visual_result_id": base_m4_result_id, }, "timeline": {"frame_count": 4489}, }, ), ) def fake_policy_archive(**kwargs) -> list[int]: shutil.copyfile(kwargs["source_archive"], kwargs["destination_archive"]) assert kwargs["valid_fov_mask"].is_file() return [4489 * 800 * 600, *([0] * 9)] monkeypatch.setattr(policy_review_module, "build_policy_mask_archive", fake_policy_archive) valid_fov_mask = tmp_path / "valid-fov-mask.png" Image.new("L", (800, 600), color=255).save(valid_fov_mask) result_root = seal_vegetation_policy_review( base_lab_root=base_root, mission_policy_path=REPOSITORY_ROOT / "config/perception/lab-v1-vegetation-mission-policy-v1.json", provider_label_map_path=REPOSITORY_ROOT / "config/perception/lab-v1-vegetation-provider-label-map-v1.json", m49_tgs_full_shadow_root=tmp_path / "sealed-tgs", valid_fov_mask_path=valid_fov_mask, output_root=tmp_path / "results", created_at_utc="2026-08-28T08:00:00+00:00", ) manifest = json.loads((result_root / "result.json").read_text("utf-8")) route = manifest["route_video"] assert route["view_kind"] == "coarse-material-policy-review" assert route["linked_tgs_result_id"] == tgs_result_id assert route["fusion"]["pixel_raster_fusion"] is False assert route["fusion"]["camera_semantic_temporal_filter"] == "none" assert route["taxonomy"]["schema_version"] == ( "missioncore.lab-v1-terrain-policy-taxonomy/v1" ) assert len(route["taxonomy"]["classes"]) == 10 assert route["valid_fov"]["outside_valid_fov_class_id"] == 9 assert len(manifest["artifacts"]) == 80 assert manifest["authority"]["commands_enabled"] is False assert manifest["decision"]["multilayer_policy_review_ready"] is True app = FastAPI() app.include_router(build_vegetation_shadow_lab_router(root_provider=lambda: result_root.parent)) response = TestClient(app).get( f"/api/v1/laboratory/vegetation-shadow/{result_root.name}/masks/0" ) assert response.status_code == 200 assert response.content == b"\x89PNG\r\n\x1a\n"