feat(perception): add DDRNet full-video vegetation replay

This commit is contained in:
DCCONSTRUCTIONS
2026-08-28 01:13:18 +03:00
parent 67e6ae98e2
commit 30080c51aa
10 changed files with 729 additions and 64 deletions
+91 -3
View File
@@ -2,12 +2,15 @@ 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
@@ -87,19 +90,98 @@ def _worker_result(root: Path, *, candidate: str, mode: str, vegetation_iou: flo
(root / "result.json").write_text(json.dumps(payload), encoding="utf-8")
def test_vegetation_shadow_lab_seals_autonomous_visual_evidence(tmp_path: Path) -> None:
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"
@@ -108,7 +190,9 @@ def test_vegetation_shadow_lab_seals_autonomous_visual_evidence(tmp_path: Path)
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"]) == 76
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"]
@@ -121,7 +205,7 @@ def test_vegetation_shadow_lab_seals_autonomous_visual_evidence(tmp_path: Path)
)
proof = verify_laboratory_evidence_result(definition, result_root)
assert proof["result_id"] == result_root.name
assert proof["artifact_count"] == 76
assert proof["artifact_count"] == 78
app = FastAPI()
app.include_router(build_vegetation_shadow_lab_router(root_provider=lambda: result_root.parent))
@@ -135,6 +219,10 @@ def test_vegetation_shadow_lab_seals_autonomous_visual_evidence(tmp_path: 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 (