327 lines
13 KiB
Python
327 lines
13 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import shutil
|
|
import zipfile
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
|
|
import k1link.laboratory.vegetation_policy_review as policy_review_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 _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"])
|
|
return [4489 * 800 * 600, *([0] * 8)]
|
|
|
|
monkeypatch.setattr(policy_review_module, "build_policy_mask_archive", fake_policy_archive)
|
|
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",
|
|
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"]) == 9
|
|
assert len(manifest["artifacts"]) == 79
|
|
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"
|