feat(perception): add vegetation mission policy lab
This commit is contained in:
DCCONSTRUCTIONS
@@ -0,0 +1,219 @@
|
||||
"""Validate and resolve the planning-only vegetation mission-policy contract."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from collections.abc import Mapping
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import Any, Final
|
||||
|
||||
SCHEMA: Final = "missioncore.vegetation-mission-policy/v1"
|
||||
VEHICLE_SCHEMA: Final = "missioncore.m49-physical-safety-shadow-profile/v1"
|
||||
_MAX_JSON_BYTES: Final = 1024 * 1024
|
||||
|
||||
|
||||
class VegetationMissionPolicyError(RuntimeError):
|
||||
"""The vegetation mission policy is absent, inconsistent, or unsafe."""
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class TerrainPolicyDecision:
|
||||
preset_id: str
|
||||
material_class: str
|
||||
evidence_state: str
|
||||
requested_action: str
|
||||
effective_action: str
|
||||
policy_source: str
|
||||
safety_reason: str | None
|
||||
actuation_authority: bool = False
|
||||
|
||||
|
||||
def _json(path: Path, label: str) -> dict[str, Any]:
|
||||
if path.is_symlink() or not path.is_file() or path.stat().st_size > _MAX_JSON_BYTES:
|
||||
raise VegetationMissionPolicyError(f"{label} is unavailable")
|
||||
try:
|
||||
value = json.loads(path.read_text(encoding="utf-8-sig"))
|
||||
except (OSError, UnicodeDecodeError, json.JSONDecodeError) as error:
|
||||
raise VegetationMissionPolicyError(f"{label} is invalid") from error
|
||||
if not isinstance(value, dict):
|
||||
raise VegetationMissionPolicyError(f"{label} is invalid")
|
||||
return value
|
||||
|
||||
|
||||
def _string_set(value: object, label: str) -> set[str]:
|
||||
if (
|
||||
not isinstance(value, list)
|
||||
or not value
|
||||
or any(not isinstance(item, str) or not item for item in value)
|
||||
or len(set(value)) != len(value)
|
||||
):
|
||||
raise VegetationMissionPolicyError(f"{label} is invalid")
|
||||
return set(value)
|
||||
|
||||
|
||||
def _validate_vehicle_profile(policy: dict[str, Any], repository_root: Path) -> None:
|
||||
reference = policy.get("vehicle_profile")
|
||||
if not isinstance(reference, dict) or not isinstance(reference.get("path"), str):
|
||||
raise VegetationMissionPolicyError("vehicle profile reference is invalid")
|
||||
relative = Path(reference["path"])
|
||||
if relative.is_absolute() or ".." in relative.parts:
|
||||
raise VegetationMissionPolicyError("vehicle profile path escapes the repository")
|
||||
path = (repository_root / relative).resolve()
|
||||
try:
|
||||
path.relative_to(repository_root.resolve())
|
||||
except ValueError as error:
|
||||
raise VegetationMissionPolicyError("vehicle profile path escapes the repository") from error
|
||||
vehicle_profile = _json(path, "vehicle profile")
|
||||
vehicle = vehicle_profile.get("vehicle")
|
||||
authority = vehicle_profile.get("authority")
|
||||
if (
|
||||
vehicle_profile.get("schema_version") != VEHICLE_SCHEMA
|
||||
or vehicle_profile.get("profile_id") != reference.get("profile_id")
|
||||
or not isinstance(vehicle, dict)
|
||||
or not isinstance(authority, dict)
|
||||
or authority.get("commands_enabled") is not False
|
||||
or authority.get("actuation_accepted") is not False
|
||||
):
|
||||
raise VegetationMissionPolicyError("vehicle profile authority changed")
|
||||
expected_vehicle = {
|
||||
"body_length_m": 1.0,
|
||||
"body_width_m": 0.8,
|
||||
"body_height_m": 0.4,
|
||||
"ground_clearance_m": 0.15,
|
||||
"nominal_speed_mps": 0.2777777778,
|
||||
"maximum_operating_speed_mps": 0.5555555556,
|
||||
}
|
||||
if any(vehicle.get(key) != expected for key, expected in expected_vehicle.items()):
|
||||
raise VegetationMissionPolicyError("vehicle planning facts changed")
|
||||
|
||||
|
||||
def load_vegetation_mission_policy(
|
||||
path: Path,
|
||||
*,
|
||||
repository_root: Path | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Load the profile and fail closed if precedence or safety locks drift."""
|
||||
|
||||
policy = _json(path, "vegetation mission policy")
|
||||
if (
|
||||
policy.get("schema_version") != SCHEMA
|
||||
or policy.get("status") != "planning-shadow-only-unqualified"
|
||||
or policy.get("precedence")
|
||||
!= [
|
||||
"hard_safety_interlock",
|
||||
"explicit_mission_rule",
|
||||
"selected_preset_default",
|
||||
]
|
||||
):
|
||||
raise VegetationMissionPolicyError("policy identity or precedence changed")
|
||||
|
||||
actions = _string_set(policy.get("actions"), "actions")
|
||||
if actions != {"ALLOW", "HIGH_COST", "NO_GO"}:
|
||||
raise VegetationMissionPolicyError("actions changed")
|
||||
materials = _string_set(policy.get("material_classes"), "material classes")
|
||||
overridable = _string_set(
|
||||
policy.get("mission_overridable_materials"), "mission-overridable materials"
|
||||
)
|
||||
if not overridable < materials:
|
||||
raise VegetationMissionPolicyError("mission-overridable materials are invalid")
|
||||
|
||||
presets = policy.get("presets")
|
||||
if not isinstance(presets, dict) or set(presets) != {"urban", "rural", "offroad"}:
|
||||
raise VegetationMissionPolicyError("presets changed")
|
||||
for preset_id, rules in presets.items():
|
||||
if (
|
||||
not isinstance(rules, dict)
|
||||
or set(rules) != materials
|
||||
or any(action not in actions for action in rules.values())
|
||||
):
|
||||
raise VegetationMissionPolicyError(f"preset {preset_id} is invalid")
|
||||
|
||||
evidence = policy.get("evidence_policy")
|
||||
authority = policy.get("authority")
|
||||
mission_contract = policy.get("mission_configuration_contract")
|
||||
if (
|
||||
not isinstance(evidence, dict)
|
||||
or not isinstance(authority, dict)
|
||||
or not isinstance(mission_contract, dict)
|
||||
):
|
||||
raise VegetationMissionPolicyError("policy boundaries are invalid")
|
||||
eligible = _string_set(evidence.get("policy_eligible_states"), "eligible evidence states")
|
||||
force_states = _string_set(evidence.get("force_no_go_states"), "force-no-go states")
|
||||
force_materials = _string_set(
|
||||
evidence.get("force_no_go_materials"), "force-no-go materials"
|
||||
)
|
||||
if (
|
||||
eligible & force_states
|
||||
or not force_materials <= materials
|
||||
or overridable & force_materials
|
||||
or evidence.get("missing_material_is") not in force_materials
|
||||
or evidence.get("missing_evidence_is") not in force_states
|
||||
or evidence.get("camera_semantics_can_clear_rigid_geometry") is not False
|
||||
or evidence.get("mission_rule_can_override_hard_safety_interlock") is not False
|
||||
or mission_contract.get("explicit_rules_are_sparse_overrides") is not True
|
||||
or mission_contract.get("automatic_switch_may_weaken_explicit_mission_rule") is not False
|
||||
or mission_contract.get("operator_must_review_effective_rules_before_start") is not True
|
||||
or any(authority.get(key) is not False for key in authority)
|
||||
):
|
||||
raise VegetationMissionPolicyError("fail-closed policy boundary changed")
|
||||
|
||||
root = repository_root if repository_root is not None else path.resolve().parents[2]
|
||||
_validate_vehicle_profile(policy, root)
|
||||
return policy
|
||||
|
||||
|
||||
def resolve_terrain_policy(
|
||||
policy: dict[str, Any],
|
||||
*,
|
||||
preset_id: str,
|
||||
material_class: str | None,
|
||||
evidence_state: str | None,
|
||||
mission_rules: Mapping[str, str] | None = None,
|
||||
) -> TerrainPolicyDecision:
|
||||
"""Resolve preset and mission intent while retaining hard safety precedence."""
|
||||
|
||||
presets = policy["presets"]
|
||||
if preset_id not in presets:
|
||||
raise VegetationMissionPolicyError(f"unknown preset: {preset_id}")
|
||||
evidence = policy["evidence_policy"]
|
||||
material = material_class or evidence["missing_material_is"]
|
||||
state = evidence_state or evidence["missing_evidence_is"]
|
||||
if material not in policy["material_classes"]:
|
||||
material = evidence["missing_material_is"]
|
||||
|
||||
explicit_rules = dict(mission_rules or {})
|
||||
actions = set(policy["actions"])
|
||||
overridable = set(policy["mission_overridable_materials"])
|
||||
for rule_material, action in explicit_rules.items():
|
||||
if rule_material not in overridable:
|
||||
raise VegetationMissionPolicyError(
|
||||
f"material cannot be overridden by a mission: {rule_material}"
|
||||
)
|
||||
if action not in actions:
|
||||
raise VegetationMissionPolicyError(f"unknown mission action: {action}")
|
||||
|
||||
if material in explicit_rules:
|
||||
requested_action = explicit_rules[material]
|
||||
source = "explicit_mission_rule"
|
||||
else:
|
||||
requested_action = presets[preset_id][material]
|
||||
source = "selected_preset_default"
|
||||
|
||||
reason: str | None = None
|
||||
if state in evidence["force_no_go_states"]:
|
||||
reason = f"hard_safety_interlock:{state}"
|
||||
elif material in evidence["force_no_go_materials"]:
|
||||
reason = f"hard_safety_interlock:{material}"
|
||||
elif state not in evidence["policy_eligible_states"]:
|
||||
reason = "hard_safety_interlock:unrecognized_evidence_state"
|
||||
|
||||
return TerrainPolicyDecision(
|
||||
preset_id=preset_id,
|
||||
material_class=material,
|
||||
evidence_state=state,
|
||||
requested_action=requested_action,
|
||||
effective_action="NO_GO" if reason is not None else requested_action,
|
||||
policy_source=source,
|
||||
safety_reason=reason,
|
||||
)
|
||||
Reference in New Issue
Block a user