Files
NODEDC_MISSION_CORE/tests/test_vegetation_shadow_lab.py
T

530 lines
21 KiB
Python

from __future__ import annotations
import hashlib
import io
import json
import shutil
import struct
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 (
_canonical_route_playback_chunk_descriptor,
_mask_component_boxes,
_route_tgs_anchor_payload,
build_vegetation_shadow_lab_router,
)
REPOSITORY_ROOT = Path(__file__).resolve().parents[1]
def test_semantic_component_boxes_keep_distinct_objects_separate() -> None:
mask = np.zeros((20, 30), dtype=np.uint8)
mask[2:10, 3:8] = 4
mask[4:12, 18:24] = 4
mask[15:17, 3:5] = 4
assert _mask_component_boxes(mask, 4, minimum_pixels=20) == [
(18, 4, 24, 12, 48),
(3, 2, 8, 10, 40),
]
def test_route_playback_chunk_descriptor_seals_only_requested_binary_window() -> None:
points = np.arange(18, dtype="<f4").reshape(6, 3)
descriptor = _canonical_route_playback_chunk_descriptor(
"/api/v1/laboratory/vegetation-shadow",
f"lab-v1-vegetation-shadow-{'a' * 64}",
memoryview(points).cast("B"),
(0, 1, 1, 3, 6),
0,
)
assert descriptor is not None
assert descriptor["start"] == 0
assert descriptor["count"] == 4
assert descriptor["point_count"] == 6
assert descriptor["bytes"] == points.nbytes
assert descriptor["shape"] == [6, 3]
assert len(str(descriptor["sha256"])) == 64
def test_route_tgs_anchor_payload_preserves_metric_evidence(tmp_path: Path) -> None:
path = tmp_path / "tgs-evidence.npz"
point_counts = np.arange(1, 11, dtype=np.int64)
offsets = np.concatenate(([0], np.cumsum(point_counts)))
points = np.arange(int(offsets[-1]) * 3, dtype=np.float32).reshape(-1, 3)
centers = np.arange(2244 * 2, dtype=np.float32).reshape(2244, 2) * 0.45
states = np.tile(np.arange(2244, dtype=np.uint16) % 4, (10, 1)).astype(np.uint8)
z_bounds = np.zeros((10, 2244, 2), dtype=np.float32)
z_bounds[..., 0] = np.nan
z_bounds[..., 1] = 1.25
np.savez(
path,
source_frame_indices=np.array(
[20, 408, 789, 1189, 1609, 1992, 2380, 3190, 4810, 6381],
dtype=np.int64,
),
current_increment_point_offsets=offsets,
current_increment_points_xyz_m=points,
costmap_cell_centers_xy_m=centers,
causal_rolling_1s_costmap_states=states,
causal_rolling_1s_costmap_z_bounds_m=z_bounds,
)
payload = _route_tgs_anchor_payload(path, 409)
assert payload["schema_version"] == "missioncore.lab-v1-route-tgs-anchor/v1"
assert payload["source_sequence"] == 409
assert payload["slot"] == 1
assert len(payload["current_points_xyz_m"]) == 2
assert len(payload["costmap"]["centers_xy_m"]) == 2244
assert set(payload["costmap"]["state_codes"]) == {0, 1, 2, 3}
assert payload["costmap"]["z_bounds_m"][0] == [None, 1.25]
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")
full_archive_payloads = (b"\x89PNG\r\n\x1a\ncity", b"\x89PNG\r\n\x1a\nvegetation")
full_timeline_payload = struct.pack("<2Q", 1_000_000_000, 1_100_000_000)
full_identity = dict(manifest["identity"])
full_route = {
"frame_count": 2,
"timeline": {
"path": "video/frame-source-times-ns.bin",
"sha256": hashlib.sha256(full_timeline_payload).hexdigest(),
"byte_length": len(full_timeline_payload),
"encoding": "uint64-le-nanoseconds",
"frame_count": 2,
},
"layers": {
layer: {"mask_archive": {"path": "video/full-route-masks.zip"}}
for layer in ("city", "vegetation")
},
}
full_identity["route_full_review"] = full_route
full_identity_sha = hashlib.sha256(
json.dumps(
full_identity,
ensure_ascii=False,
sort_keys=True,
separators=(",", ":"),
).encode("utf-8")
).hexdigest()
full_result_id = f"lab-v1-vegetation-shadow-{full_identity_sha}"
full_root = result_root.parent / full_result_id
shutil.copytree(result_root, full_root)
full_archive = full_root / "video" / "full-route-masks.zip"
full_archive.parent.mkdir(exist_ok=True)
with zipfile.ZipFile(full_archive, "x", compression=zipfile.ZIP_STORED) as frozen:
for sequence, payload in enumerate(full_archive_payloads, start=1):
frozen.writestr(f"masks/frame-{sequence:06d}.png", payload)
full_timeline = full_root / "video" / "frame-source-times-ns.bin"
full_timeline.write_bytes(full_timeline_payload)
full_manifest = dict(manifest)
full_manifest["result_id"] = full_result_id
full_manifest["identity"] = full_identity
full_manifest["identity_sha256"] = full_identity_sha
full_manifest["route_full_review"] = full_route
full_manifest["artifacts"] = [
*manifest["artifacts"],
{
"role": "full-route-mask-fixture",
"path": "video/full-route-masks.zip",
"byte_length": full_archive.stat().st_size,
"sha256": _sha256(full_archive),
"media_type": "application/zip",
},
{
"role": "full-route-frame-timeline",
"path": "video/frame-source-times-ns.bin",
"byte_length": full_timeline.stat().st_size,
"sha256": _sha256(full_timeline),
"media_type": "application/octet-stream",
},
]
(full_root / "result.json").write_text(
json.dumps(full_manifest, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
+ "\n",
encoding="utf-8",
)
for layer, sequence, expected in (
("city", 0, full_archive_payloads[0]),
("vegetation", 1, full_archive_payloads[1]),
):
response = client.get(
f"/api/v1/laboratory/vegetation-shadow/{full_result_id}"
f"/route-masks/{layer}/{sequence}"
)
assert response.status_code == 200
assert response.content == expected
assert response.headers["cache-control"].endswith("immutable")
assert client.get(
f"/api/v1/laboratory/vegetation-shadow/{full_result_id}/route-masks/city/2"
).status_code == 404
timeline = client.get(
f"/api/v1/laboratory/vegetation-shadow/{full_result_id}/route-timeline"
)
assert timeline.status_code == 200
assert timeline.content == full_timeline_payload
assert timeline.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"