SILVER/NODEDC_MISSION_CORE
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

feat(perception): integrate vegetation policy review

Browse Source
This commit is contained in:
DCCONSTRUCTIONS
2026-08-28 11:22:00 +03:00
parent 30080c51aa
commit c4c2392c79
17 changed files with 2388 additions and 105 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/k1link/laboratory/vegetation_policy_review.py
+281
View File
@@ -0,0 +1,281 @@
"""Seal a coarse material + YOLOX + TGS review from an immutable vegetation LAB."""
from __future__ import annotations
import argparse
import copy
import hashlib
import json
import shutil
import tempfile
from datetime import UTC, datetime
from pathlib import Path, PurePosixPath
from typing import Any, Final
from k1link.laboratory.evidence_registry import LaboratoryEvidenceDefinition
from k1link.laboratory.evidence_report import verify_laboratory_evidence_result
from k1link.laboratory.m49_tgs_full_shadow import read_m49_tgs_full_shadow
from k1link.laboratory.vegetation_mission_policy import (
load_vegetation_mission_policy,
load_vegetation_provider_label_map,
)
from k1link.laboratory.vegetation_policy_video import build_policy_mask_archive, policy_taxonomy
from k1link.laboratory.vegetation_shadow_lab import (
LAB_SCHEMA,
RESULT_PREFIX,
canonical_json,
sha256_path,
)
_DEFINITION: Final = LaboratoryEvidenceDefinition(
work_id="lab-v1-vegetation-shadow",
runtime_relative_root=PurePosixPath("lab-v1-vegetation/results"),
result_id_prefix="lab-v1-vegetation-shadow",
document_name="result.json",
result_schema_version=LAB_SCHEMA,
)
_FRAME_COUNT: Final = 4489
_MAX_RESULT_BYTES: Final = 1024 * 1024
class VegetationPolicyReviewError(ValueError):
"""The sealed inputs cannot form an honest synchronized policy review."""
def _object(value: object, label: str) -> dict[str, Any]:
if not isinstance(value, dict) or not all(isinstance(key, str) for key in value):
raise VegetationPolicyReviewError(f"{label} must be an object")
return value
def _read_base(root: Path) -> dict[str, Any]:
candidate = root.resolve(strict=True)
verify_laboratory_evidence_result(_DEFINITION, candidate)
path = candidate / "result.json"
if path.stat().st_size > _MAX_RESULT_BYTES:
raise VegetationPolicyReviewError("base vegetation LAB document is too large")
payload = _object(json.loads(path.read_text("utf-8")), "base vegetation LAB")
route = _object(payload.get("route_video"), "base route video")
authority = _object(payload.get("authority"), "base authority")
if (
payload.get("schema_version") != LAB_SCHEMA
or payload.get("result_id") != candidate.name
or route.get("frame_count") != _FRAME_COUNT
or route.get("view_kind", "fine-semantic-prediction")
!= "fine-semantic-prediction"
or route.get("base_m4_result_id") is None
or authority.get("commands_enabled") is not False
or authority.get("navigation_or_safety_accepted") is not False
or authority.get("actuation_accepted") is not False
or authority.get("camera_semantics_can_clear_rigid_geometry") is not False
):
raise VegetationPolicyReviewError("base vegetation LAB contract changed")
return payload
def _copy_verified_artifacts(
*,
source_root: Path,
destination_root: Path,
artifacts: object,
) -> list[dict[str, object]]:
if not isinstance(artifacts, list):
raise VegetationPolicyReviewError("base artifact catalog changed")
copied: list[dict[str, object]] = []
for raw in artifacts:
descriptor = _object(raw, "base artifact")
relative_text = descriptor.get("path")
expected_sha256 = descriptor.get("sha256")
if not isinstance(relative_text, str) or not isinstance(expected_sha256, str):
raise VegetationPolicyReviewError("base artifact proof changed")
relative = PurePosixPath(relative_text)
source = source_root.joinpath(*relative.parts)
destination = destination_root.joinpath(*relative.parts)
if (
relative.is_absolute()
or str(relative) != relative_text
or any(part in {"", ".", ".."} for part in relative.parts)
or source.is_symlink()
or not source.is_file()
or sha256_path(source) != expected_sha256
):
raise VegetationPolicyReviewError("base artifact changed")
destination.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
shutil.copyfile(source, destination)
copied.append(copy.deepcopy(descriptor))
return copied
def seal_vegetation_policy_review(
*,
base_lab_root: Path,
mission_policy_path: Path,
provider_label_map_path: Path,
m49_tgs_full_shadow_root: Path,
output_root: Path,
created_at_utc: str | None = None,
) -> Path:
base_root = base_lab_root.resolve(strict=True)
base = _read_base(base_root)
base_route = _object(base["route_video"], "base route video")
repository_root = mission_policy_path.resolve().parents[2]
mission_policy = load_vegetation_mission_policy(
mission_policy_path.resolve(strict=True),
repository_root=repository_root,
)
provider_map = load_vegetation_provider_label_map(
provider_label_map_path.resolve(strict=True),
policy=mission_policy,
)
tgs = read_m49_tgs_full_shadow(m49_tgs_full_shadow_root)
tgs_source = _object(tgs.report.get("source"), "full TGS source")
tgs_timeline = _object(tgs.report.get("timeline"), "full TGS timeline")
if (
tgs_source.get("source_id") != "RAVNOVES00"
or tgs_source.get("linked_visual_result_id") != base_route.get("base_m4_result_id")
or tgs_timeline.get("frame_count") != _FRAME_COUNT
):
raise VegetationPolicyReviewError("TGS and vegetation timelines differ")
raw_archive = _object(base_route.get("mask_archive"), "fine mask archive")
if raw_archive.get("path") != "video/ddrnet-semantic-masks.zip":
raise VegetationPolicyReviewError("fine mask archive identity changed")
raw_archive_path = base_root / "video" / "ddrnet-semantic-masks.zip"
fine_taxonomy = _object(base_route.get("taxonomy"), "fine taxonomy")
output_root.mkdir(mode=0o700, parents=True, exist_ok=True)
temporary = Path(tempfile.mkdtemp(prefix=".lab-v1-policy-", dir=output_root))
try:
artifacts = _copy_verified_artifacts(
source_root=base_root,
destination_root=temporary,
artifacts=base.get("artifacts"),
)
policy_archive = temporary / "video" / "coarse-material-policy-masks.zip"
policy_counts = build_policy_mask_archive(
source_archive=raw_archive_path,
destination_archive=policy_archive,
fine_taxonomy=fine_taxonomy,
provider_label_map=provider_map,
)
policy_archive_proof = {
"role": "route-coarse-material-mask-archive",
"path": "video/coarse-material-policy-masks.zip",
"byte_length": policy_archive.stat().st_size,
"sha256": sha256_path(policy_archive),
"media_type": "application/zip",
}
artifacts.append(policy_archive_proof)
route = copy.deepcopy(base_route)
route.update(
{
"view_kind": "coarse-material-policy-review",
"source_mask_archive": copy.deepcopy(raw_archive),
"mask_archive": {
"path": policy_archive_proof["path"],
"sha256": policy_archive_proof["sha256"],
"byte_length": policy_archive_proof["byte_length"],
},
"taxonomy": policy_taxonomy(),
"aggregate_prediction_pixels": policy_counts,
"linked_tgs_result_id": tgs.result_id,
"policy": {
"profile_id": mission_policy["profile_id"],
"profile_sha256": sha256_path(mission_policy_path),
"provider_label_map_id": provider_map["profile_id"],
"provider_label_map_sha256": sha256_path(provider_label_map_path),
"presets": mission_policy["presets"],
"precedence": mission_policy["precedence"],
},
"fusion": {
"mode": "synchronised-multilayer-review",
"pixel_raster_fusion": False,
"camera_material_layer": "DDRNet fine-64 to coarse material evidence",
"camera_safety_veto_layer": "frozen M4 YOLOX camera proposals",
"spatial_safety_veto_layer": "M4.9 full TGS gravity-local costmap",
"temporal_consensus_owner": "TGS causal rolling 1 s and metric obstacle tracks",
"camera_semantic_temporal_filter": "none",
"reason": "No admitted TGS-to-camera pixel projection exists.",
},
}
)
identity = copy.deepcopy(_object(base.get("identity"), "base identity"))
identity.update(
{
"base_result_id": base_root.name,
"route_video": route,
}
)
identity_sha256 = hashlib.sha256(canonical_json(identity)).hexdigest()
result_id = f"{RESULT_PREFIX}{identity_sha256}"
manifest = copy.deepcopy(base)
manifest.update(
{
"result_id": result_id,
"identity_sha256": identity_sha256,
"created_at_utc": created_at_utc or datetime.now(UTC).isoformat(),
"identity": identity,
"route_video": route,
"method": {
"completeness": "complete",
"execution_class": "ai-inference-plus-deterministic-adapter",
"pipeline_id": "goose-fine64-to-coarse-material-plus-yolox-tgs-review/v1",
},
"decision": {
**_object(base.get("decision"), "base decision"),
"multilayer_policy_review_ready": True,
"navigation_accepted": False,
"production_accepted": False,
},
"limitations": [
"GOOSE validation is external-domain qualification, not RAVNOVES ground truth.",
(
"The coarse material playback is derived from per-frame DDRNet "
"predictions and has no RAVNOVES truth."
),
(
"Vegetation semantics never clears YOLOX, LiDAR, metric obstacle "
"or TGS vetoes."
),
"Undefined pixels outside the 600x600 center crop remain fail-closed.",
(
"TGS remains in gravity-local space; no uncalibrated pixel "
"projection is fabricated."
),
(
"Temporal consensus comes from causal TGS and metric tracks; "
"the camera material mask is not temporally filtered."
),
],
"artifacts": artifacts,
}
)
(temporary / "result.json").write_bytes(canonical_json(manifest) + b"\n")
destination = output_root / result_id
if destination.exists():
raise VegetationPolicyReviewError("immutable vegetation policy result already exists")
temporary.replace(destination)
verify_laboratory_evidence_result(_DEFINITION, destination)
return destination
except Exception:
shutil.rmtree(temporary, ignore_errors=True)
raise
def main() -> None:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--base-lab-root", type=Path, required=True)
parser.add_argument("--mission-policy-path", type=Path, required=True)
parser.add_argument("--provider-label-map-path", type=Path, required=True)
parser.add_argument("--m49-tgs-full-shadow-root", type=Path, required=True)
parser.add_argument("--output-root", type=Path, required=True)
args = parser.parse_args()
print(seal_vegetation_policy_review(**vars(args)))
if __name__ == "__main__":
main()
__all__ = ["VegetationPolicyReviewError", "seal_vegetation_policy_review"]
src/k1link/laboratory/vegetation_policy_video.py
+206
View File
@@ -0,0 +1,206 @@
"""Build a deterministic coarse material-evidence video from fine GOOSE masks."""
from __future__ import annotations
import io
import zipfile
from pathlib import Path
from typing import Any, Final
import numpy as np
from PIL import Image
from k1link.laboratory.vegetation_mission_policy import map_provider_material
TAXONOMY_SCHEMA: Final = "missioncore.lab-v1-terrain-policy-taxonomy/v1"
FRAME_COUNT: Final = 4489
WIDTH: Final = 800
HEIGHT: Final = 600
POLICY_CLASSES: Final = (
{
"class_id": 0,
"label": "UNOBSERVED / NO MATERIAL CLAIM · NO_GO",
"color_rgb": [147, 151, 159],
"disposition": "ambiguous",
"material_class": None,
"evidence_state": "UNOBSERVED",
},
{
"class_id": 1,
"label": "SAFETY DETECTOR VETO · NO_GO",
"color_rgb": [255, 104, 112],
"disposition": "labeled",
"material_class": None,
"evidence_state": "RIGID_OR_UNKNOWN_OBSTACLE",
},
{
"class_id": 2,
"label": "WOODY SHRUB / TREE · NO_GO",
"color_rgb": [232, 56, 126],
"disposition": "labeled",
"material_class": "woody_or_tree",
"evidence_state": "VEGETATION_WITH_RIGID_GEOMETRY",
},
{
"class_id": 3,
"label": "CULTIVATED VEGETATION · POLICY NO_GO",
"color_rgb": [183, 112, 255],
"disposition": "labeled",
"material_class": "cultivated_vegetation",
"evidence_state": "VEGETATION_POTENTIALLY_TRAVERSABLE",
},
{
"class_id": 4,
"label": "LOW GRASS · MISSION CANDIDATE",
"color_rgb": [181, 255, 90],
"disposition": "prediction",
"material_class": "grass",
"evidence_state": "VEGETATION_POTENTIALLY_TRAVERSABLE",
},
{
"class_id": 5,
"label": "HIGH / HERBACEOUS · MISSION CANDIDATE",
"color_rgb": [113, 211, 111],
"disposition": "prediction",
"material_class": "herbaceous_vegetation",
"evidence_state": "VEGETATION_POTENTIALLY_TRAVERSABLE",
},
{
"class_id": 6,
"label": "BARE SOIL · MISSION CANDIDATE",
"color_rgb": [255, 197, 92],
"disposition": "prediction",
"material_class": "bare_soil",
"evidence_state": "SUPPORTED_GROUND",
},
{
"class_id": 7,
"label": "HARD SURFACE · MISSION CANDIDATE",
"color_rgb": [84, 169, 255],
"disposition": "prediction",
"material_class": "hard_surface",
"evidence_state": "SUPPORTED_GROUND",
},
{
"class_id": 8,
"label": "VEGETATION UNKNOWN · NO_GO",
"color_rgb": [207, 124, 255],
"disposition": "labeled",
"material_class": "vegetation_unknown",
"evidence_state": "VEGETATION_UNKNOWN",
},
)
_MATERIAL_TO_CLASS: Final = {
"hard_surface": 7,
"bare_soil": 6,
"grass": 4,
"fern": 5,
"herbaceous_vegetation": 5,
"cultivated_vegetation": 3,
"woody_shrub": 2,
"tree_or_trunk": 2,
"vegetation_unknown": 8,
}
class VegetationPolicyVideoError(ValueError):
"""The fine-mask input cannot be transformed without inventing evidence."""
def policy_taxonomy() -> dict[str, object]:
return {
"schema_version": TAXONOMY_SCHEMA,
"classes": [dict(row) for row in POLICY_CLASSES],
}
def fine_to_policy_lut(
fine_taxonomy: dict[str, object],
provider_label_map: dict[str, Any],
) -> np.ndarray:
classes = fine_taxonomy.get("classes")
if not isinstance(classes, list) or len(classes) != 64:
raise VegetationPolicyVideoError("fine taxonomy must contain 64 classes")
lut = np.zeros(256, dtype=np.uint8)
for expected_id, raw in enumerate(classes):
if not isinstance(raw, dict) or raw.get("class_id") != expected_id:
raise VegetationPolicyVideoError("fine taxonomy ordering changed")
label = raw.get("label")
if not isinstance(label, str) or not label:
raise VegetationPolicyVideoError("fine taxonomy label is invalid")
if expected_id == 0:
continue
material = map_provider_material(
provider_label_map,
provider_id="goose-fine-64",
provider_label=label,
)
lut[expected_id] = _MATERIAL_TO_CLASS.get(material, 0)
return lut
def _zip_info(name: str) -> zipfile.ZipInfo:
info = zipfile.ZipInfo(name, date_time=(1980, 1, 1, 0, 0, 0))
info.compress_type = zipfile.ZIP_STORED
info.create_system = 3
info.external_attr = 0o600 << 16
return info
def build_policy_mask_archive(
*,
source_archive: Path,
destination_archive: Path,
fine_taxonomy: dict[str, object],
provider_label_map: dict[str, Any],
) -> list[int]:
"""Map every fine mask to coarse evidence; safety vetoes remain separate layers."""
lut = fine_to_policy_lut(fine_taxonomy, provider_label_map)
counts = np.zeros(len(POLICY_CLASSES), dtype=np.int64)
destination_archive.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
try:
with zipfile.ZipFile(source_archive) as source, zipfile.ZipFile(
destination_archive,
"x",
) as destination:
for sequence in range(FRAME_COUNT):
member = f"masks/frame-{sequence + 1:06d}.png"
with source.open(member) as stream, Image.open(stream) as image:
fine = np.asarray(image.convert("L"), dtype=np.uint8)
if fine.shape != (HEIGHT, WIDTH):
raise VegetationPolicyVideoError(
f"fine mask {member} has shape {fine.shape}, expected {(HEIGHT, WIDTH)}"
)
coarse = lut[fine]
counts += np.bincount(
coarse.reshape(-1),
minlength=len(POLICY_CLASSES),
)
buffer = io.BytesIO()
Image.fromarray(coarse, mode="L").save(
buffer,
format="PNG",
compress_level=1,
optimize=False,
)
destination.writestr(_zip_info(member), buffer.getvalue())
except (KeyError, OSError, ValueError, zipfile.BadZipFile) as exc:
destination_archive.unlink(missing_ok=True)
raise VegetationPolicyVideoError("fine mask archive is invalid") from exc
return [int(value) for value in counts]
__all__ = [
"FRAME_COUNT",
"HEIGHT",
"POLICY_CLASSES",
"TAXONOMY_SCHEMA",
"VegetationPolicyVideoError",
"WIDTH",
"build_policy_mask_archive",
"fine_to_policy_lut",
"policy_taxonomy",
]
src/k1link/laboratory/vegetation_shadow_lab.py
+153 -4
View File
@@ -14,6 +14,15 @@ from pathlib import Path, PurePosixPath
from typing import Any, Final
from k1link.laboratory.m47_reference_graph import read_m47_reference_graph_lab
from k1link.laboratory.m49_tgs_full_shadow import read_m49_tgs_full_shadow
from k1link.laboratory.vegetation_mission_policy import (
load_vegetation_mission_policy,
load_vegetation_provider_label_map,
)
from k1link.laboratory.vegetation_policy_video import (
build_policy_mask_archive,
policy_taxonomy,
)
LAB_SCHEMA: Final = "missioncore.lab-v1-vegetation-shadow/v1"
WORKER_SCHEMA: Final = "missioncore.lab-v1-goose-vegetation-run/v1"
@@ -315,6 +324,9 @@ def seal_vegetation_shadow_lab(
output_root: Path,
ddrnet_ravnoves_video_root: Path | None = None,
m47_reference_graph_lab_root: Path | None = None,
mission_policy_path: Path | None = None,
provider_label_map_path: Path | None = None,
m49_tgs_full_shadow_root: Path | None = None,
) -> Path:
roots = {
("ddrnet", "goose"): ddrnet_goose_root.resolve(),
@@ -333,6 +345,17 @@ def seal_vegetation_shadow_lab(
selected = _selected_candidate(results)
if (ddrnet_ravnoves_video_root is None) != (m47_reference_graph_lab_root is None):
raise VegetationShadowLabError("full-video Worker and M4.7 roots must be paired")
policy_inputs = (
mission_policy_path,
provider_label_map_path,
m49_tgs_full_shadow_root,
)
if any(value is not None for value in policy_inputs) and not all(
value is not None for value in policy_inputs
):
raise VegetationShadowLabError("policy, provider map and full TGS roots must be paired")
if all(value is not None for value in policy_inputs) and ddrnet_ravnoves_video_root is None:
raise VegetationShadowLabError("policy review requires the full-video DDRNet result")
route_video: dict[str, object] | None = None
route_video_archive: Path | None = None
video_result: dict[str, Any] | None = None
@@ -355,6 +378,35 @@ def seal_vegetation_shadow_lab(
raise VegetationShadowLabError("M4.7 video binding differs from DDRNet source")
route_video["m47_reference_graph_result_id"] = m47.result_id
mission_policy: dict[str, Any] | None = None
provider_label_map: dict[str, Any] | None = None
linked_tgs_result_id: str | None = None
if (
mission_policy_path is not None
and provider_label_map_path is not None
and m49_tgs_full_shadow_root is not None
and route_video is not None
):
repository_root = mission_policy_path.resolve().parents[2]
mission_policy = load_vegetation_mission_policy(
mission_policy_path.resolve(),
repository_root=repository_root,
)
provider_label_map = load_vegetation_provider_label_map(
provider_label_map_path.resolve(),
policy=mission_policy,
)
tgs = read_m49_tgs_full_shadow(m49_tgs_full_shadow_root)
tgs_source = _object(tgs.report.get("source"), "M4.9 full TGS source")
tgs_timeline = _object(tgs.report.get("timeline"), "M4.9 full TGS timeline")
if (
tgs_source.get("source_id") != "RAVNOVES00"
or tgs_source.get("linked_visual_result_id") != route_video["base_m4_result_id"]
or tgs_timeline.get("frame_count") != _VIDEO_FRAME_COUNT
):
raise VegetationShadowLabError("full TGS timeline differs from vegetation video")
linked_tgs_result_id = tgs.result_id
output_root.mkdir(mode=0o700, parents=True, exist_ok=True)
temporary = Path(tempfile.mkdtemp(prefix=".lab-v1-vegetation-", dir=output_root))
artifacts: list[dict[str, object]] = []
@@ -471,14 +523,78 @@ def seal_vegetation_shadow_lab(
temporary,
"video/ddrnet-semantic-masks.zip",
artifacts,
role="route-semantic-mask-archive",
role=(
"route-fine-semantic-source-archive"
if mission_policy is not None
else "route-semantic-mask-archive"
),
media_type="application/zip",
)
route_video["mask_archive"] = {
raw_archive_proof = {
"path": archive_descriptor["path"],
"sha256": archive_descriptor["sha256"],
"byte_length": archive_descriptor["byte_length"],
}
route_video["mask_archive"] = raw_archive_proof
route_video["view_kind"] = "fine-semantic-prediction"
if (
mission_policy is not None
and provider_label_map is not None
and linked_tgs_result_id is not None
and mission_policy_path is not None
and provider_label_map_path is not None
):
policy_archive = temporary / "video" / "coarse-material-policy-masks.zip"
policy_counts = build_policy_mask_archive(
source_archive=route_video_archive,
destination_archive=policy_archive,
fine_taxonomy=_object(route_video["taxonomy"], "fine video taxonomy"),
provider_label_map=provider_label_map,
)
policy_descriptor = {
"role": "route-coarse-material-mask-archive",
"path": "video/coarse-material-policy-masks.zip",
"byte_length": policy_archive.stat().st_size,
"sha256": sha256_path(policy_archive),
"media_type": "application/zip",
}
artifacts.append(policy_descriptor)
route_video.update(
{
"view_kind": "coarse-material-policy-review",
"source_mask_archive": raw_archive_proof,
"mask_archive": {
"path": policy_descriptor["path"],
"sha256": policy_descriptor["sha256"],
"byte_length": policy_descriptor["byte_length"],
},
"taxonomy": policy_taxonomy(),
"aggregate_prediction_pixels": policy_counts,
"linked_tgs_result_id": linked_tgs_result_id,
"policy": {
"profile_id": mission_policy["profile_id"],
"profile_sha256": sha256_path(mission_policy_path),
"provider_label_map_id": provider_label_map["profile_id"],
"provider_label_map_sha256": sha256_path(
provider_label_map_path
),
"presets": mission_policy["presets"],
"precedence": mission_policy["precedence"],
},
"fusion": {
"mode": "synchronised-multilayer-review",
"pixel_raster_fusion": False,
"camera_material_layer": "DDRNet fine-64 to coarse material evidence",
"camera_safety_veto_layer": "frozen M4 YOLOX camera proposals",
"spatial_safety_veto_layer": "M4.9 full TGS gravity-local costmap",
"temporal_consensus_owner": (
"TGS causal rolling 1 s and metric obstacle tracks"
),
"camera_semantic_temporal_filter": "none",
"reason": "No admitted TGS-to-camera pixel projection exists.",
},
}
)
candidate_metrics: dict[str, object] = {}
for candidate in _CANDIDATES:
@@ -536,7 +652,11 @@ def seal_vegetation_shadow_lab(
"method": {
"completeness": "complete",
"execution_class": "ai-inference",
"pipeline_id": "goose-fine64-ready-weights-to-ravnoves-policy-shadow/v1",
"pipeline_id": (
"goose-fine64-to-coarse-material-plus-yolox-tgs-review/v1"
if mission_policy is not None
else "goose-fine64-ready-weights-to-ravnoves-policy-shadow/v1"
),
},
"metrics": {"candidates": candidate_metrics},
"decision": {
@@ -544,14 +664,37 @@ def seal_vegetation_shadow_lab(
"visual_shadow_ready": True,
"full_video_shadow_ready": route_video is not None,
"mission_policy_ready_for_configuration": True,
"multilayer_policy_review_ready": mission_policy is not None,
"navigation_accepted": False,
"production_accepted": False,
},
"limitations": [
"GOOSE validation is external-domain qualification, not RAVNOVES ground truth.",
"The full RAVNOVES DDRNet playback is prediction-only and has no independent labels.",
(
"The coarse material playback is derived from per-frame DDRNet predictions "
"and has no RAVNOVES truth."
if mission_policy is not None
else (
"The full RAVNOVES DDRNet playback is prediction-only and has "
"no independent labels."
)
),
"Vegetation semantics never clears rigid LiDAR/TGS occupancy.",
"Undefined pixels outside the 600x600 center crop remain fail-closed.",
*(
[
(
"TGS remains in gravity-local space; no uncalibrated pixel "
"projection is fabricated."
),
(
"Temporal consensus comes from causal TGS and metric tracks; "
"the camera material mask is not temporally filtered."
),
]
if mission_policy is not None
else []
),
],
"authority": authority,
"catalogs": catalogs,
@@ -577,6 +720,9 @@ def _parse_args() -> argparse.Namespace:
parser.add_argument("--output-root", type=Path, required=True)
parser.add_argument("--ddrnet-ravnoves-video-root", type=Path)
parser.add_argument("--m47-reference-graph-lab-root", type=Path)
parser.add_argument("--mission-policy-path", type=Path)
parser.add_argument("--provider-label-map-path", type=Path)
parser.add_argument("--m49-tgs-full-shadow-root", type=Path)
return parser.parse_args()
@@ -590,6 +736,9 @@ def main() -> None:
output_root=args.output_root,
ddrnet_ravnoves_video_root=args.ddrnet_ravnoves_video_root,
m47_reference_graph_lab_root=args.m47_reference_graph_lab_root,
mission_policy_path=args.mission_policy_path,
provider_label_map_path=args.provider_label_map_path,
m49_tgs_full_shadow_root=args.m49_tgs_full_shadow_root,
)
print(destination)
src/k1link/web/vegetation_shadow_lab_api.py
+24 -3
View File
@@ -93,12 +93,33 @@ def build_vegetation_shadow_lab_router(
candidate = _resolve_candidate(root_provider, result_id)
manifest = _read_verified(candidate)
route_video = manifest.get("route_video")
if not isinstance(route_video, dict) or not 0 <= sequence < 4489:
if (
not isinstance(route_video, dict)
or route_video.get("frame_count") != 4489
or not 0 <= sequence < 4489
):
raise HTTPException(status_code=404, detail="Vegetation video mask not found")
archive = route_video.get("mask_archive")
if not isinstance(archive, dict) or archive.get("path") != "video/ddrnet-semantic-masks.zip":
archive_relative = archive.get("path") if isinstance(archive, dict) else None
if not isinstance(archive_relative, str):
raise HTTPException(status_code=404, detail="Vegetation video mask not found")
archive_path = candidate / "video" / "ddrnet-semantic-masks.zip"
relative = PurePosixPath(archive_relative)
if (
relative.is_absolute()
or str(relative) != archive_relative
or any(part in {"", ".", ".."} for part in relative.parts)
or relative.suffix != ".zip"
):
raise HTTPException(status_code=404, detail="Vegetation video mask not found")
artifacts = manifest.get("artifacts")
if not isinstance(artifacts, list) or not any(
isinstance(item, dict)
and item.get("path") == archive_relative
and item.get("media_type") == "application/zip"
for item in artifacts
):
raise HTTPException(status_code=404, detail="Vegetation video mask not found")
archive_path = candidate.joinpath(*relative.parts)
member = f"masks/frame-{sequence + 1:06d}.png"
try:
before = archive_path.stat()