feat(perception): qualify M4.8T semantic identity

2026-08-25 23:44:17 +03:00
parent 209d0bfc26
commit 9a956c318a
7 changed files with 2328 additions and 0 deletions
@@ -0,0 +1,168 @@
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+import pytest
+
+from k1link.perception.m48t_risk_quality import (
+    CocoRiskImage,
+    M48TRiskQualityError,
+    RiskPrediction,
+    RiskTruth,
+    load_coco_risk_truth,
+    load_m48t_risk_quality_profile,
+    score_risk_quality,
+)
+
+REPOSITORY_ROOT = Path(__file__).resolve().parents[1]
+PROFILE_PATH = REPOSITORY_ROOT / "config/perception/m48t-risk-quality-temporal-v1.json"
+
+
+def _truth(
+    annotation_id: int,
+    class_name: str,
+    family: str,
+    bbox: tuple[float, float, float, float],
+    *,
+    size_band: str = "medium",
+) -> RiskTruth:
+    return RiskTruth(
+        image_id=1,
+        annotation_id=annotation_id,
+        class_name=class_name,
+        family=family,
+        bbox_xyxy=bbox,
+        projected_area_pixels=(bbox[2] - bbox[0]) * (bbox[3] - bbox[1]),
+        size_band=size_band,
+    )
+
+
+def _prediction(
+    prediction_id: str,
+    class_name: str,
+    family: str,
+    bbox: tuple[float, float, float, float],
+    score: float = 0.9,
+) -> RiskPrediction:
+    return RiskPrediction(
+        image_id=1,
+        prediction_id=prediction_id,
+        class_name=class_name,
+        family=family,
+        score=score,
+        bbox_xyxy=bbox,
+    )
+
+
+def test_m48t_profile_pins_candidate_and_class_independent_temporal_policy() -> None:
+    profile = load_m48t_risk_quality_profile(PROFILE_PATH)
+
+    assert profile.minimum_score == 0.25
+    assert profile.model_id == "rf_detr_large:1"
+    assert profile.class_to_family["dog"] == "animal"
+    assert profile.temporal.initial_confirmation_observations == 2
+    assert profile.temporal.switch_confirmation_observations == 3
+
+
+def test_m48t_profile_rejects_candidate_threshold_tuning(tmp_path: Path) -> None:
+    document = json.loads(PROFILE_PATH.read_text("utf-8"))
+    document["candidate"]["minimum_score"] = 0.2
+    changed = tmp_path / "changed.json"
+    changed.write_text(json.dumps(document), "utf-8")
+
+    with pytest.raises(M48TRiskQualityError, match="tuned in place"):
+        load_m48t_risk_quality_profile(changed)
+
+
+def test_coco_truth_is_projected_to_actual_source_contract_and_filtered(
+    tmp_path: Path,
+) -> None:
+    profile = load_m48t_risk_quality_profile(PROFILE_PATH)
+    annotations = {
+        "images": [{"id": 1, "file_name": "one.jpg", "width": 400, "height": 300}],
+        "categories": [
+            {"id": index, "name": class_name}
+            for index, class_name in enumerate(profile.class_to_family, start=1)
+        ],
+        "annotations": [
+            {"id": 1, "image_id": 1, "category_id": 1, "bbox": [10, 20, 20, 30], "iscrowd": 0},
+            {"id": 2, "image_id": 1, "category_id": 1, "bbox": [1, 1, 1, 1], "iscrowd": 0},
+            {"id": 3, "image_id": 1, "category_id": 1, "bbox": [10, 20, 20, 30], "iscrowd": 1},
+        ],
+    }
+    path = tmp_path / "instances.json"
+    path.write_text(json.dumps(annotations), "utf-8")
+
+    images, truth = load_coco_risk_truth(path, profile)
+
+    assert images == (CocoRiskImage(image_id=1, file_name="one.jpg", width=400, height=300),)
+    assert len(truth) == 1
+    assert truth[0].bbox_xyxy == (20.0, 40.0, 60.0, 100.0)
+    assert truth[0].projected_area_pixels == 2400.0
+
+
+def test_quality_separates_exact_class_family_and_failure_buckets() -> None:
+    profile = load_m48t_risk_quality_profile(PROFILE_PATH)
+    images = (CocoRiskImage(image_id=1, file_name="one.jpg", width=800, height=600),)
+    truth = (
+        _truth(1, "person", "person", (10.0, 10.0, 110.0, 210.0), size_band="large"),
+        _truth(2, "dog", "animal", (200.0, 100.0, 260.0, 170.0)),
+        _truth(3, "car", "vehicle", (400.0, 200.0, 600.0, 350.0), size_band="large"),
+        _truth(4, "bicycle", "light-road-user", (650.0, 200.0, 760.0, 350.0)),
+    )
+    predictions = (
+        _prediction("p1", "person", "person", (10.0, 10.0, 110.0, 210.0)),
+        _prediction("p2", "cat", "animal", (200.0, 100.0, 260.0, 170.0)),
+        _prediction("p3", "car", "vehicle", (520.0, 300.0, 700.0, 450.0)),
+        _prediction("p4", "truck", "vehicle", (300.0, 20.0, 390.0, 100.0)),
+    )
+
+    result = score_risk_quality(
+        images=images,
+        truth=truth,
+        predictions=predictions,
+        profile=profile,
+    )
+
+    assert result.report["counts"] == {
+        "predictions": 4,
+        "true_positive": 1,
+        "false_positive": 3,
+        "false_negative": 3,
+        "empty_prediction_risk_images": 0,
+    }
+    metrics = result.report["metrics"]
+    assert isinstance(metrics, dict)
+    families = metrics["families"]
+    assert isinstance(families, dict)
+    assert families["animal"]["exact_class_recall"] == 0.0
+    assert families["animal"]["family_recall"] == 1.0
+    buckets = result.report["failure_buckets"]
+    assert buckets["same-family-class-confusion"] == 1
+    assert buckets["localization"] == 1
+    assert buckets["missed"] == 1
+    assert buckets["unmatched-prediction"] == 3
+    assert result.report["quality_gates"]["passed"] is False
+    assert result.report["authority"]["candidate_accepted"] is False
+
+
+def test_quality_rejects_predictions_below_frozen_threshold() -> None:
+    profile = load_m48t_risk_quality_profile(PROFILE_PATH)
+    images = (CocoRiskImage(image_id=1, file_name="one.jpg", width=800, height=600),)
+
+    with pytest.raises(M48TRiskQualityError, match="escaped the frozen score"):
+        score_risk_quality(
+            images=images,
+            truth=(_truth(1, "person", "person", (10.0, 10.0, 100.0, 200.0)),),
+            predictions=(
+                _prediction(
+                    "p1",
+                    "person",
+                    "person",
+                    (10.0, 10.0, 100.0, 200.0),
+                    score=0.24,
+                ),
+            ),
+            profile=profile,
+        )
@@ -0,0 +1,104 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+from k1link.perception.m48t_risk_quality import (
+    BoundedTemporalSemanticIdentity,
+    M48TRiskQualityError,
+    TemporalSemanticObservation,
+    load_m48t_risk_quality_profile,
+)
+
+REPOSITORY_ROOT = Path(__file__).resolve().parents[1]
+PROFILE_PATH = REPOSITORY_ROOT / "config/perception/m48t-risk-quality-temporal-v1.json"
+
+
+def _observation(
+    time_ns: int,
+    raw_class: str | None,
+    *,
+    component_id: str = "temporal-000001",
+    currentness: str = "current",
+) -> TemporalSemanticObservation:
+    return TemporalSemanticObservation(
+        component_id=component_id,
+        evidence_time_ns=time_ns,
+        raw_class_name=raw_class,
+        currentness=currentness,
+    )
+
+
+def test_initial_class_requires_two_observations_and_identity_stays_geometry_owned() -> None:
+    stabilizer = BoundedTemporalSemanticIdentity(load_m48t_risk_quality_profile(PROFILE_PATH))
+
+    first = stabilizer.update(_observation(0, "person"))
+    second = stabilizer.update(_observation(100_000_000, "person"))
+
+    assert first.resolution == "pending"
+    assert first.selected_class_name is None
+    assert second.resolution == "confirmed"
+    assert second.selected_class_name == "person"
+    assert second.component_id == "temporal-000001"
+    assert second.association_uses_semantic_class is False
+    assert second.occupancy_uses_semantic_class is False
+
+
+def test_cross_family_switch_falls_back_unknown_until_third_confirmation() -> None:
+    stabilizer = BoundedTemporalSemanticIdentity(load_m48t_risk_quality_profile(PROFILE_PATH))
+    stabilizer.update(_observation(0, "person"))
+    stabilizer.update(_observation(100_000_000, "person"))
+
+    first_conflict = stabilizer.update(_observation(200_000_000, "car"))
+    second_conflict = stabilizer.update(_observation(300_000_000, "car"))
+    switched = stabilizer.update(_observation(400_000_000, "car"))
+
+    assert first_conflict.resolution == "conflict"
+    assert first_conflict.selected_class_name is None
+    assert second_conflict.resolution == "conflict"
+    assert switched.resolution == "confirmed"
+    assert switched.selected_class_name == "car"
+    assert stabilizer.snapshot().class_switches == 1
+
+
+def test_same_family_switch_holds_previous_class_until_confirmed() -> None:
+    stabilizer = BoundedTemporalSemanticIdentity(load_m48t_risk_quality_profile(PROFILE_PATH))
+    stabilizer.update(_observation(0, "dog"))
+    stabilizer.update(_observation(100_000_000, "dog"))
+
+    pending = stabilizer.update(_observation(200_000_000, "cat"))
+    stabilizer.update(_observation(300_000_000, "cat"))
+    switched = stabilizer.update(_observation(400_000_000, "cat"))
+
+    assert pending.resolution == "pending"
+    assert pending.selected_class_name == "dog"
+    assert switched.selected_class_name == "cat"
+
+
+def test_semantic_hold_is_bounded_then_degrades_to_unknown() -> None:
+    stabilizer = BoundedTemporalSemanticIdentity(load_m48t_risk_quality_profile(PROFILE_PATH))
+    stabilizer.update(_observation(0, "person"))
+    stabilizer.update(_observation(100_000_000, "person"))
+
+    held = stabilizer.update(_observation(300_000_000, None, currentness="held"))
+    unknown = stabilizer.update(_observation(500_000_001, None, currentness="held"))
+
+    assert held.resolution == "held"
+    assert held.selected_class_name == "person"
+    assert unknown.resolution == "unknown"
+    assert unknown.selected_class_name is None
+
+
+def test_explicit_expiry_removes_state_and_out_of_order_evidence_is_rejected() -> None:
+    stabilizer = BoundedTemporalSemanticIdentity(load_m48t_risk_quality_profile(PROFILE_PATH))
+    stabilizer.update(_observation(100, "person"))
+    with pytest.raises(M48TRiskQualityError, match="moved backwards"):
+        stabilizer.update(_observation(99, "person"))
+
+    expired = stabilizer.update(_observation(200, None, currentness="expired"))
+    restarted = stabilizer.update(_observation(300, "person"))
+
+    assert expired.resolution == "expired"
+    assert restarted.resolution == "pending"
+    assert stabilizer.snapshot().active_components == 1