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)