geointel/backend/app/services/model_validation_scope_service.py

"""Checksum-bound geographic validation scope for production model inference."""

from __future__ import annotations

from hashlib import sha256
import json
from pathlib import Path
import re
from typing import Any

from shapely.geometry import shape
from shapely.geometry.base import BaseGeometry

from app.core.errors import AppError


_SHA256 = re.compile(r"^[0-9a-f]{64}$")


class ModelValidationScopeService:
    """Load an immutable, model-bound AOI and prove that an input is covered."""

    SCHEMA_VERSION = "geointel.model-validation-scope/v1"

    @classmethod
    def assert_area_covered(
        cls,
        *,
        area_geometry: BaseGeometry,
        manifest_path: str | None,
        expected_manifest_sha256: str | None,
        model_id: str,
        model_path: str | None,
    ) -> dict[str, str]:
        path = Path(manifest_path).expanduser() if manifest_path else None
        expected_checksum = (expected_manifest_sha256 or "").strip().lower()
        if path is None or not expected_checksum:
            cls._raise(
                "DETECTION_VALIDATION_SCOPE_NOT_CONFIGURED",
                "Configured YOLO inference requires a checksum-bound geographic validation-scope manifest.",
            )
        if not _SHA256.fullmatch(expected_checksum):
            cls._raise(
                "DETECTION_VALIDATION_SCOPE_INVALID",
                "The configured validation-scope checksum must be a lowercase SHA-256 digest.",
                manifest_path=str(path),
            )
        try:
            raw_manifest = path.read_bytes()
            payload = json.loads(raw_manifest.decode("utf-8"))
        except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc:
            cls._raise(
                "DETECTION_VALIDATION_SCOPE_INVALID",
                "The configured validation-scope manifest is missing or unreadable.",
                manifest_path=str(path),
                error_type=type(exc).__name__,
            )
        observed_manifest_sha256 = sha256(raw_manifest).hexdigest()
        if observed_manifest_sha256 != expected_checksum:
            cls._raise(
                "DETECTION_VALIDATION_SCOPE_CHECKSUM_MISMATCH",
                "The validation-scope manifest does not match its configured checksum.",
                manifest_path=str(path),
                expected=expected_checksum,
                observed=observed_manifest_sha256,
            )
        if not isinstance(payload, dict) or payload.get("schema_version") != cls.SCHEMA_VERSION:
            cls._raise(
                "DETECTION_VALIDATION_SCOPE_INVALID",
                "The validation-scope manifest has an unsupported schema.",
                manifest_path=str(path),
            )
        if payload.get("model_id") != model_id:
            cls._raise(
                "DETECTION_VALIDATION_SCOPE_MODEL_MISMATCH",
                "The validation scope is not bound to the selected model identity.",
                expected_model_id=model_id,
                observed_model_id=payload.get("model_id"),
            )
        configured_model_path = Path(model_path).expanduser() if model_path else None
        if configured_model_path is None or not configured_model_path.is_file():
            cls._raise(
                "DETECTION_VALIDATION_SCOPE_MODEL_UNAVAILABLE",
                "The model bytes bound by the validation scope are unavailable.",
            )
        declared_model_sha256 = str(payload.get("model_sha256") or "").strip().lower()
        observed_model_sha256 = cls._file_sha256(configured_model_path)
        if not _SHA256.fullmatch(declared_model_sha256) or declared_model_sha256 != observed_model_sha256:
            cls._raise(
                "DETECTION_VALIDATION_SCOPE_MODEL_MISMATCH",
                "The validation scope is not bound to the exact selected model bytes.",
                expected=declared_model_sha256 or None,
                observed=observed_model_sha256,
            )
        if payload.get("crs") != "EPSG:4326":
            cls._raise(
                "DETECTION_VALIDATION_SCOPE_INVALID",
                "The validation-scope geometry must explicitly use EPSG:4326.",
                observed_crs=payload.get("crs"),
            )
        try:
            scope_geometry = shape(payload.get("geometry"))
        except Exception as exc:
            cls._raise(
                "DETECTION_VALIDATION_SCOPE_INVALID",
                "The validation-scope geometry is not valid GeoJSON.",
                error_type=type(exc).__name__,
            )
        if (
            scope_geometry.is_empty
            or not scope_geometry.is_valid
            or scope_geometry.geom_type not in {"Polygon", "MultiPolygon"}
        ):
            cls._raise(
                "DETECTION_VALIDATION_SCOPE_INVALID",
                "The validation scope must be a non-empty valid Polygon or MultiPolygon.",
                geometry_type=scope_geometry.geom_type,
            )
        if area_geometry.is_empty or not area_geometry.is_valid or not scope_geometry.covers(area_geometry):
            cls._raise(
                "DETECTION_VALIDATION_SCOPE_UNAVAILABLE",
                "Configured YOLO inference is not validated for the complete Dataset area.",
                scope_key=payload.get("scope_key"),
            )
        return {
            "scope_key": str(payload.get("scope_key") or "unspecified"),
            "manifest_path": str(path.resolve()),
            "manifest_sha256": observed_manifest_sha256,
            "model_sha256": observed_model_sha256,
        }

    @staticmethod
    def _file_sha256(path: Path) -> str:
        digest = sha256()
        with path.open("rb") as handle:
            for chunk in iter(lambda: handle.read(1024 * 1024), b""):
                digest.update(chunk)
        return digest.hexdigest()

    @staticmethod
    def _raise(code: str, message: str, **details: Any) -> None:
        raise AppError(code=code, message=message, details=details, status_code=422)