geointel/backend/tests/test_segmentation_qa_coverage.py

"""Segmentation QA must score against the area it actually inferred.

Detection QA already clips both populations to the union of the persisted
inference tiles. Segmentation QA compared candidates against every reference
feature in the dataset, so every building outside the inferred tiles counted
as a false negative and recall collapsed for no modelling reason.
"""

from __future__ import annotations

import json
from pathlib import Path
from uuid import uuid4

import pytest
from geoalchemy2.shape import from_shape
from shapely.geometry import MultiPolygon, box

from app.core.errors import AppError
from app.models import AnalysisRun, Dataset, Segmentation, VectorFeature
from app.services.segmentation_service import SegmentationService

from tests.test_sprint9_segmentation_foundation import (  # noqa: F401
    FakeSession,
    _authoritative_reference,
)


def _manifest(tmp_path: Path, dataset_id, bounds: list[float]) -> str:
    manifest_path = tmp_path / "manifest.json"
    manifest_path.write_text(
        json.dumps(
            {
                "source_dataset_id": str(dataset_id),
                "crs": "EPSG:4326",
                "tiles": [{"path": "tile_0000.tif", "bounds": bounds, "crs": "EPSG:4326"}],
            }
        ),
        encoding="utf-8",
    )
    return str(manifest_path)


def _segmentation(project_id, dataset_id, analysis_run_id, geom):
    return Segmentation(
        id=uuid4(),
        project_id=project_id,
        dataset_id=dataset_id,
        analysis_run_id=analysis_run_id,
        job_id=uuid4(),
        model_name="fixture-segmenter",
        model_version="fixture-v1",
        class_name="building",
        confidence=0.9,
        geometry=from_shape(MultiPolygon([geom]), srid=4326),
    )


def _reference(dataset_id, geom) -> VectorFeature:
    return VectorFeature(
        id=uuid4(),
        dataset_id=dataset_id,
        feature_class="building",
        geometry=from_shape(geom, srid=4326),
    )


def _session(tmp_path: Path, *, with_manifest: bool):
    project_id = uuid4()
    dataset_id = uuid4()
    reference_dataset_id = uuid4()
    analysis_run_id = uuid4()

    parameters = {}
    if with_manifest:
        parameters = {"tile_manifest_path": _manifest(tmp_path, dataset_id, [0.0, 0.0, 1.0, 1.0])}

    reference_dataset = _authoritative_reference(
        Dataset(
            id=reference_dataset_id,
            project_id=project_id,
            name="reference.geojson",
            dataset_type="vector",
            source="test",
            dataset_role="reference",
        )
    )
    db = FakeSession(
        objects={
            (AnalysisRun, analysis_run_id): AnalysisRun(
                id=analysis_run_id,
                project_id=project_id,
                dataset_id=dataset_id,
                analysis_type="segmentation",
                status="success",
                parameters_json=parameters,
            ),
            (Dataset, dataset_id): Dataset(
                id=dataset_id, project_id=project_id, name="fixture.tif", dataset_type="raster", source="test"
            ),
            (Dataset, reference_dataset_id): reference_dataset,
        },
        query_rows={
            Segmentation: [_segmentation(project_id, dataset_id, analysis_run_id, box(0.1, 0.1, 0.2, 0.2))],
            VectorFeature: [
                # Inside the inferred tile: a genuine match.
                _reference(reference_dataset_id, box(0.1, 0.1, 0.2, 0.2)),
                # Far outside it: never looked at by the model.
                _reference(reference_dataset_id, box(8.0, 8.0, 8.1, 8.1)),
                _reference(reference_dataset_id, box(9.0, 9.0, 9.1, 9.1)),
            ],
        },
    )
    return db, analysis_run_id, reference_dataset_id


def test_segmentation_qa_scores_only_inside_persisted_tile_coverage(tmp_path: Path) -> None:
    db, analysis_run_id, reference_dataset_id = _session(tmp_path, with_manifest=True)

    result = SegmentationService.compare_segmentations_with_reference(
        db=db,
        analysis_run_id=analysis_run_id,
        reference_dataset_id=reference_dataset_id,
        iou_threshold=0.5,
    )

    assert result["matches"] == 1
    assert result["false_negatives"] == 0
    assert result["recall"] == 1.0
    assert result["coverage"]["applied"] is True
    assert result["coverage"]["reference_raw_count"] == 3
    assert result["coverage"]["reference_evaluated_count"] == 1
    assert result["coverage"]["reference_excluded_outside_count"] == 2
    assert any("tile" in warning for warning in result["warnings"])


def test_segmentation_qa_without_manifest_reports_unbounded_coverage(tmp_path: Path) -> None:
    db, analysis_run_id, reference_dataset_id = _session(tmp_path, with_manifest=False)

    result = SegmentationService.compare_segmentations_with_reference(
        db=db,
        analysis_run_id=analysis_run_id,
        reference_dataset_id=reference_dataset_id,
        iou_threshold=0.5,
    )

    # Unchanged behaviour, but the response now says the score was not bounded
    # by an inference footprint so the recall can be read correctly.
    assert result["false_negatives"] == 2
    assert result["coverage"]["applied"] is False
    assert result["coverage"]["mode"] == "unbounded_no_manifest"


def test_segmentation_qa_rejects_reference_entirely_outside_coverage(tmp_path: Path) -> None:
    db, analysis_run_id, reference_dataset_id = _session(tmp_path, with_manifest=True)
    db.query_rows[VectorFeature] = [
        _reference(reference_dataset_id, box(8.0, 8.0, 8.1, 8.1)),
    ]

    with pytest.raises(AppError) as exc_info:
        SegmentationService.compare_segmentations_with_reference(
            db=db,
            analysis_run_id=analysis_run_id,
            reference_dataset_id=reference_dataset_id,
            iou_threshold=0.5,
        )

    assert exc_info.value.code == "REFERENCE_FEATURES_OUTSIDE_COVERAGE"