NODEDC_MISSION_CORE/tests/test_m48t_risk_quality.py

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,
        )