264 lines
8.9 KiB
Python
264 lines
8.9 KiB
Python
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)
|