from __future__ import annotations import json from pathlib import Path import pytest from k1link.laboratory.semantic_object_quality import ( SemanticObjectQualityError, SemanticTruthLabel, load_semantic_object_quality_profile, score_semantic_object_quality, ) from k1link.perception.contracts import ( EvidenceBasis, EvidenceCurrentness, MetricGeometry, MotionState, ObstacleObservation, ) from k1link.perception.object_understanding import ( AdvisoryResponse, AdvisoryRiskAssessment, AgencyState, EvidenceKind, EvidenceProvenance, ObjectStateEstimate, ObjectUnderstanding, RiskBasis, RiskLevel, SemanticDecision, SemanticHypothesis, SemanticResolution, StateBasis, load_object_semantic_vocabulary, ) REPOSITORY_ROOT = Path(__file__).resolve().parents[1] VOCABULARY_PATH = REPOSITORY_ROOT / "config/perception/object-semantic-vocabulary-v0.json" PROFILE_PATH = REPOSITORY_ROOT / "config/perception/m48s-semantic-object-quality-v0.json" def _observation(index: int) -> ObstacleObservation: return ObstacleObservation( observation_id=f"observation-{index}", occupancy_key=f"occupied-{index}", source_id="RAVNOVES00", frame_id=f"frame-{index:06d}", evidence_time_ns=index * 1_000_000, basis=EvidenceBasis.FUSED, currentness=EvidenceCurrentness.CURRENT, occupied_support=True, source_point_ids=(index,), metric_geometry=MetricGeometry( coordinate_frame="map", centroid_xyz_m=(float(index), 0.0, 0.5), range_m=float(index + 1), covariance_diagonal_m2=(0.04, 0.04, 0.09), ), proposal_ids=(f"proposal-{index}",), semantic_hint=None, reason_codes=("current-qualified-points",), ) def _prediction( index: int, *, hypotheses: tuple[tuple[str, str, float], ...], selected_class_id: str | None, ) -> ObjectUnderstanding: observation = _observation(index) evidence = EvidenceProvenance( evidence_id=f"semantic-evidence-{index}", kind=EvidenceKind.DETECTOR, source_id=observation.source_id, frame_id=observation.frame_id, provider_id="semantic-candidate/v0", model_id="semantic-model", model_revision="candidate-1", preprocess_id="rgb/v1", prompt_set_id="urban-risk-groups/v0", ) ranked = tuple( SemanticHypothesis( rank=rank, class_id=class_id, raw_label=raw_label, confidence=confidence, evidence_ids=(evidence.evidence_id,), ) for rank, (class_id, raw_label, confidence) in enumerate(hypotheses, start=1) ) selected = next( (item for item in ranked if item.class_id == selected_class_id), None, ) return ObjectUnderstanding( understanding_id=f"understanding-{index}", vocabulary_id="missioncore.urban-object-semantics/v0", generated_monotonic_ns=index * 1_000_000 + 1, observation=observation, hypotheses=ranked, semantic=SemanticDecision( resolution=( SemanticResolution.SELECTED if selected is not None else SemanticResolution.UNRESOLVED ), selected_class_id=None if selected is None else selected.class_id, selected_confidence=None if selected is None else selected.confidence, reason_codes=( "top-hypothesis-qualified" if selected is not None else "semantic-evidence-insufficient", ), ), state=ObjectStateEstimate( motion=MotionState.UNKNOWN, motion_confidence=0.0, agency=AgencyState.UNKNOWN, agency_basis=StateBasis.UNKNOWN, evidence_ids=(), reason_codes=("motion-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=("semantic-quality-does-not-score-risk",), ), provenance=(evidence,), ) def _truth(index: int, class_id: str) -> SemanticTruthLabel: return SemanticTruthLabel( label_id=f"semantic-label-{index}", observation_id=f"observation-{index}", source_id="RAVNOVES00", frame_id=f"frame-{index:06d}", class_id=class_id, reviewer_count=2, adjudicated=True, ) def test_semantic_quality_scores_exact_group_topk_and_unresolved_separately() -> None: vocabulary = load_object_semantic_vocabulary(VOCABULARY_PATH) profile = load_semantic_object_quality_profile(PROFILE_PATH) predictions = ( _prediction( 1, hypotheses=( ("human.unknown", "person", 0.70), ("human.child", "child", 0.65), ), selected_class_id="human.unknown", ), _prediction( 2, hypotheses=(("animal.dog", "dog", 0.80),), selected_class_id="animal.dog", ), _prediction(3, hypotheses=(), selected_class_id=None), ) truth = ( _truth(1, "human.child"), _truth(2, "animal.dog"), _truth(3, "vehicle.car"), ) result = score_semantic_object_quality( predictions=predictions, truth=truth, vocabulary=vocabulary, profile=profile, ) metrics = result.report["metrics"] assert isinstance(metrics, dict) assert metrics["prediction_coverage"] == pytest.approx(1.0) assert metrics["exact_top1_accuracy"] == pytest.approx(1 / 3) assert metrics["coarse_group_accuracy"] == pytest.approx(2 / 3) assert metrics["exact_top_k_recall"] == pytest.approx(2 / 3) assert metrics["unresolved_fraction"] == pytest.approx(1 / 3) assert result.report["candidate_semantic_gate_passed"] is False assert result.cases[0].exact_top1_correct is False assert result.cases[0].coarse_group_correct is True assert result.cases[0].exact_top_k_hit is True def test_semantic_quality_reports_missing_projection_without_calling_it_detection() -> None: vocabulary = load_object_semantic_vocabulary(VOCABULARY_PATH) profile = load_semantic_object_quality_profile(PROFILE_PATH) result = score_semantic_object_quality( predictions=( _prediction( 1, hypotheses=(("human.child", "child", 0.90),), selected_class_id="human.child", ), ), truth=(_truth(1, "human.child"), _truth(2, "animal.dog")), vocabulary=vocabulary, profile=profile, ) metrics = result.report["metrics"] assert isinstance(metrics, dict) assert metrics["prediction_coverage"] == pytest.approx(0.5) assert result.cases[1].resolution == "unavailable" scope = result.report["scope"] assert isinstance(scope, dict) assert scope["object_presence_scored"] is False def test_semantic_truth_is_strict_and_requires_independent_adjudication() -> None: label = _truth(1, "human.child") document = json.loads(json.dumps(label.to_dict())) assert SemanticTruthLabel.from_dict(document) == label document["category"] = "child" with pytest.raises(SemanticObjectQualityError, match="fields are incompatible"): SemanticTruthLabel.from_dict(document) with pytest.raises(SemanticObjectQualityError, match="two independent reviewers"): SemanticTruthLabel( label_id="semantic-label-1", observation_id="observation-1", source_id="RAVNOVES00", frame_id="frame-000001", class_id="human.child", reviewer_count=1, adjudicated=True, ) with pytest.raises(SemanticObjectQualityError, match="must be adjudicated"): SemanticTruthLabel( label_id="semantic-label-1", observation_id="observation-1", source_id="RAVNOVES00", frame_id="frame-000001", class_id="human.child", reviewer_count=2, adjudicated=False, ) def test_semantic_quality_profile_stays_separate_from_presence_and_risk( tmp_path: Path, ) -> None: profile = load_semantic_object_quality_profile(PROFILE_PATH) assert profile.profile_id == "m48s-urban-semantic-object-quality/v0" assert profile.vocabulary_id == "missioncore.urban-object-semantics/v0" assert profile.top_k == 5 document = json.loads(PROFILE_PATH.read_text("utf-8")) document["scope"]["object_presence_scored"] = True changed = tmp_path / "m48s-profile-changed.json" changed.write_text(json.dumps(document), "utf-8") with pytest.raises(SemanticObjectQualityError, match="scope changed"): load_semantic_object_quality_profile(changed)