318 lines
11 KiB
Python
318 lines
11 KiB
Python
from __future__ import annotations
|
|
|
|
import copy
|
|
import json
|
|
from dataclasses import replace
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from k1link.perception.contracts import (
|
|
EvidenceBasis,
|
|
EvidenceCurrentness,
|
|
MetricGeometry,
|
|
MotionState,
|
|
ObstacleObservation,
|
|
)
|
|
from k1link.perception.object_understanding import (
|
|
AdvisoryResponse,
|
|
AdvisoryRiskAssessment,
|
|
AgencyState,
|
|
EvidenceKind,
|
|
EvidenceProvenance,
|
|
ObjectStateEstimate,
|
|
ObjectUnderstanding,
|
|
ObjectUnderstandingError,
|
|
RiskBasis,
|
|
RiskLevel,
|
|
SemanticDecision,
|
|
SemanticHypothesis,
|
|
SemanticResolution,
|
|
StateBasis,
|
|
load_object_semantic_vocabulary,
|
|
validate_object_understanding,
|
|
)
|
|
|
|
REPOSITORY_ROOT = Path(__file__).resolve().parents[1]
|
|
VOCABULARY_PATH = REPOSITORY_ROOT / "config/perception/object-semantic-vocabulary-v0.json"
|
|
|
|
|
|
def _observation(*, semantic_hint: str | None = None) -> ObstacleObservation:
|
|
return ObstacleObservation(
|
|
observation_id="observation-vehicle-1",
|
|
occupancy_key="occupied-component-17",
|
|
source_id="RAVNOVES00",
|
|
frame_id="frame-000253",
|
|
evidence_time_ns=35_421_857_292,
|
|
basis=EvidenceBasis.FUSED,
|
|
currentness=EvidenceCurrentness.CURRENT,
|
|
occupied_support=True,
|
|
source_point_ids=(4, 7, 9),
|
|
metric_geometry=MetricGeometry(
|
|
coordinate_frame="map",
|
|
centroid_xyz_m=(4.0, 1.0, 0.5),
|
|
range_m=4.15,
|
|
covariance_diagonal_m2=(0.04, 0.04, 0.09),
|
|
),
|
|
proposal_ids=("proposal-vehicle-1",),
|
|
semantic_hint=semantic_hint,
|
|
reason_codes=("current-qualified-points",),
|
|
)
|
|
|
|
|
|
def _detector_evidence(*, frame_id: str = "frame-000253") -> EvidenceProvenance:
|
|
return EvidenceProvenance(
|
|
evidence_id="evidence-grounding-dino-1",
|
|
kind=EvidenceKind.DETECTOR,
|
|
source_id="RAVNOVES00",
|
|
frame_id=frame_id,
|
|
provider_id="grounding-dino-open-vocabulary/v1",
|
|
model_id="grounding-dino-tensorrt",
|
|
model_revision="worker-006-probe-20260825",
|
|
preprocess_id="kb4-rectified-rgb/v1",
|
|
prompt_set_id="urban-risk-groups/v0",
|
|
)
|
|
|
|
|
|
def _temporal_evidence() -> EvidenceProvenance:
|
|
return EvidenceProvenance(
|
|
evidence_id="evidence-temporal-1",
|
|
kind=EvidenceKind.TEMPORAL,
|
|
source_id="RAVNOVES00",
|
|
frame_id="frame-000253",
|
|
provider_id="temporal-occupied/v1",
|
|
model_id=None,
|
|
model_revision=None,
|
|
preprocess_id=None,
|
|
)
|
|
|
|
|
|
def _car_understanding() -> ObjectUnderstanding:
|
|
detector = _detector_evidence()
|
|
temporal = _temporal_evidence()
|
|
return ObjectUnderstanding(
|
|
understanding_id="understanding-vehicle-1",
|
|
vocabulary_id="missioncore.urban-object-semantics/v0",
|
|
generated_monotonic_ns=1_020_000,
|
|
observation=_observation(semantic_hint="car"),
|
|
hypotheses=(
|
|
SemanticHypothesis(
|
|
rank=1,
|
|
class_id="vehicle.car",
|
|
raw_label="car",
|
|
confidence=0.79,
|
|
evidence_ids=(detector.evidence_id,),
|
|
),
|
|
SemanticHypothesis(
|
|
rank=2,
|
|
class_id="vehicle.heavy",
|
|
raw_label="heavy vehicle",
|
|
confidence=0.12,
|
|
evidence_ids=(detector.evidence_id,),
|
|
),
|
|
),
|
|
semantic=SemanticDecision(
|
|
resolution=SemanticResolution.SELECTED,
|
|
selected_class_id="vehicle.car",
|
|
selected_confidence=0.79,
|
|
reason_codes=("top-hypothesis-qualified",),
|
|
),
|
|
state=ObjectStateEstimate(
|
|
motion=MotionState.STATIONARY,
|
|
motion_confidence=0.85,
|
|
agency=AgencyState.SELF_PROPELLED,
|
|
agency_basis=StateBasis.CLASS_PRIOR,
|
|
evidence_ids=(detector.evidence_id, temporal.evidence_id),
|
|
reason_codes=("stationary-observed-vehicle-prior-retained",),
|
|
),
|
|
risk=AdvisoryRiskAssessment(
|
|
policy_id="urban-object-risk/v0",
|
|
level=RiskLevel.ELEVATED,
|
|
confidence=0.71,
|
|
basis=RiskBasis.FUSED,
|
|
responses=(
|
|
AdvisoryResponse.MONITOR,
|
|
AdvisoryResponse.REDUCE_SPEED,
|
|
AdvisoryResponse.ROUTE_AROUND,
|
|
),
|
|
evidence_ids=(detector.evidence_id, temporal.evidence_id),
|
|
reason_codes=("stationary-vehicle-may-start-moving",),
|
|
),
|
|
provenance=(detector, temporal),
|
|
)
|
|
|
|
|
|
def _unknown_understanding() -> ObjectUnderstanding:
|
|
return ObjectUnderstanding(
|
|
understanding_id="understanding-unknown-1",
|
|
vocabulary_id="missioncore.urban-object-semantics/v0",
|
|
generated_monotonic_ns=1_020_000,
|
|
observation=_observation(),
|
|
hypotheses=(),
|
|
semantic=SemanticDecision(
|
|
resolution=SemanticResolution.UNRESOLVED,
|
|
selected_class_id=None,
|
|
selected_confidence=None,
|
|
reason_codes=("semantic-evidence-unavailable",),
|
|
),
|
|
state=ObjectStateEstimate(
|
|
motion=MotionState.UNKNOWN,
|
|
motion_confidence=0.0,
|
|
agency=AgencyState.UNKNOWN,
|
|
agency_basis=StateBasis.UNKNOWN,
|
|
evidence_ids=(),
|
|
reason_codes=("state-evidence-unavailable",),
|
|
),
|
|
risk=AdvisoryRiskAssessment(
|
|
policy_id="urban-object-risk/v0",
|
|
level=RiskLevel.UNKNOWN,
|
|
confidence=0.0,
|
|
basis=RiskBasis.UNKNOWN,
|
|
responses=(AdvisoryResponse.ROUTE_AROUND,),
|
|
evidence_ids=(),
|
|
reason_codes=("unknown-object-remains-occupied",),
|
|
),
|
|
provenance=(),
|
|
)
|
|
|
|
|
|
def test_object_understanding_round_trip_is_strict_and_keeps_v1_geometry() -> None:
|
|
value = _car_understanding()
|
|
observation_before = value.observation.to_dict()
|
|
document = json.loads(json.dumps(value.to_dict()))
|
|
|
|
restored = ObjectUnderstanding.from_dict(document)
|
|
|
|
assert restored == value
|
|
assert restored.observation.to_dict() == observation_before
|
|
assert restored.occupancy_identity == "occupied-component-17"
|
|
assert restored.observation.metric_geometry is not None
|
|
assert restored.observation.metric_geometry.range_m == pytest.approx(4.15)
|
|
document["unexpected"] = True
|
|
with pytest.raises(ObjectUnderstandingError, match="fields are incompatible"):
|
|
ObjectUnderstanding.from_dict(document)
|
|
|
|
|
|
def test_stationary_vehicle_keeps_separate_self_propelled_prior_and_advisory_risk() -> None:
|
|
value = _car_understanding()
|
|
|
|
assert value.state.motion is MotionState.STATIONARY
|
|
assert value.state.agency is AgencyState.SELF_PROPELLED
|
|
assert value.state.agency_basis is StateBasis.CLASS_PRIOR
|
|
assert value.risk.level is RiskLevel.ELEVATED
|
|
assert AdvisoryResponse.REDUCE_SPEED in value.risk.responses
|
|
assert value.authority.commands_enabled is False
|
|
assert value.authority.actuation_allowed is False
|
|
assert value.authority.navigation_or_safety_accepted is False
|
|
|
|
|
|
def test_unknown_semantics_never_erase_metric_occupancy() -> None:
|
|
value = _unknown_understanding()
|
|
|
|
assert value.semantic.resolution is SemanticResolution.UNRESOLVED
|
|
assert value.hypotheses == ()
|
|
assert value.observation.occupied_support is True
|
|
assert value.observation.source_point_ids == (4, 7, 9)
|
|
assert value.occupancy_identity == value.observation.occupancy_identity
|
|
assert value.risk.responses == (AdvisoryResponse.ROUTE_AROUND,)
|
|
|
|
|
|
def test_ranked_hypotheses_and_selected_class_must_agree() -> None:
|
|
value = _car_understanding()
|
|
with pytest.raises(ObjectUnderstandingError, match="ordered by confidence"):
|
|
replace(
|
|
value,
|
|
hypotheses=(
|
|
replace(value.hypotheses[0], confidence=0.10),
|
|
replace(value.hypotheses[1], confidence=0.90),
|
|
),
|
|
)
|
|
with pytest.raises(ObjectUnderstandingError, match="match one ranked hypothesis"):
|
|
replace(
|
|
value,
|
|
semantic=replace(
|
|
value.semantic,
|
|
selected_class_id="animal.dog",
|
|
selected_confidence=0.79,
|
|
),
|
|
)
|
|
with pytest.raises(ObjectUnderstandingError, match="two hypotheses"):
|
|
replace(
|
|
value,
|
|
hypotheses=value.hypotheses[:1],
|
|
semantic=SemanticDecision(
|
|
resolution=SemanticResolution.CONFLICT,
|
|
selected_class_id=None,
|
|
selected_confidence=None,
|
|
reason_codes=("provider-disagreement",),
|
|
),
|
|
)
|
|
|
|
|
|
def test_evidence_cannot_escape_geometry_frame_or_be_fabricated() -> None:
|
|
value = _car_understanding()
|
|
with pytest.raises(ObjectUnderstandingError, match="escaped"):
|
|
replace(value, provenance=(_detector_evidence(frame_id="frame-000254"),))
|
|
with pytest.raises(ObjectUnderstandingError, match="unknown evidence"):
|
|
replace(
|
|
value,
|
|
hypotheses=(
|
|
replace(value.hypotheses[0], evidence_ids=("missing-evidence",)),
|
|
value.hypotheses[1],
|
|
),
|
|
)
|
|
|
|
|
|
def test_executable_vocabulary_resolves_current_urban_labels_and_hierarchy() -> None:
|
|
vocabulary = load_object_semantic_vocabulary(VOCABULARY_PATH)
|
|
|
|
assert vocabulary.vocabulary_id == "missioncore.urban-object-semantics/v0"
|
|
assert vocabulary.resolve_label("trash bin") == "static.trash-bin"
|
|
assert vocabulary.resolve_label("Dog") == "animal.dog"
|
|
assert vocabulary.resolve_label("sidewalk-curb") == "terrain.curb"
|
|
assert {
|
|
label: vocabulary.resolve_label(label)
|
|
for label in ("car", "person", "bicycle", "road sign")
|
|
} == {
|
|
"car": "vehicle.car",
|
|
"person": "human.unknown",
|
|
"bicycle": "vehicle.bicycle",
|
|
"road sign": "static.road-sign",
|
|
}
|
|
assert vocabulary.resolve_label("unseen alien object") is None
|
|
assert vocabulary.ancestors("human.child") == (
|
|
"human.unknown",
|
|
"object.unknown",
|
|
)
|
|
validate_object_understanding(_car_understanding(), vocabulary)
|
|
|
|
|
|
def test_vocabulary_validation_rejects_undeclared_class_and_authority_change(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
vocabulary = load_object_semantic_vocabulary(VOCABULARY_PATH)
|
|
value = _car_understanding()
|
|
with pytest.raises(ObjectUnderstandingError, match="undeclared classes"):
|
|
validate_object_understanding(
|
|
replace(
|
|
value,
|
|
hypotheses=(
|
|
replace(value.hypotheses[0], class_id="vehicle.hovercraft"),
|
|
value.hypotheses[1],
|
|
),
|
|
semantic=replace(
|
|
value.semantic,
|
|
selected_class_id="vehicle.hovercraft",
|
|
),
|
|
),
|
|
vocabulary,
|
|
)
|
|
|
|
document = json.loads(VOCABULARY_PATH.read_text("utf-8"))
|
|
incompatible = copy.deepcopy(document)
|
|
incompatible["policies"]["planner_command_authority"] = True
|
|
path = tmp_path / "vocabulary.json"
|
|
path.write_text(json.dumps(incompatible), "utf-8")
|
|
with pytest.raises(ObjectUnderstandingError, match="authority policy"):
|
|
load_object_semantic_vocabulary(path)
|