from __future__ import annotations import hashlib import json import zipfile from pathlib import Path from types import SimpleNamespace from fastapi import FastAPI from fastapi.testclient import TestClient from k1link.laboratory import LaboratoryEvidenceRegistry import k1link.laboratory.vegetation_shadow_lab as vegetation_lab_module from k1link.laboratory.evidence_report import verify_laboratory_evidence_result 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 _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 )