diff --git a/.env.example b/.env.example index 716e7ce8..0d1b2f9b 100644 --- a/.env.example +++ b/.env.example @@ -102,8 +102,12 @@ YOLO_MODEL_PATH= YOLO_MODEL_ID=yolo-configured YOLO_MODEL_DISPLAY_NAME=Configured YOLO detector YOLO_MODEL_VERSION= +YOLO_MODEL_CLASSES=building +YOLO_ENFORCE_VALIDATION_SCOPE=false +YOLO_VALIDATED_AREA_NAMES=Mol,Kempen YOLO_CONFIG_DIR=./storage/ultralytics YOLO_DEVICE=cpu +YOLO_REQUIRE_CUDA=false YOLO_IMAGE_SIZE=640 YOLO_MAX_TILES=100 YOLO_MAX_DETECTIONS=1000 @@ -151,3 +155,5 @@ GEOINTEL_POSTGRES_USER=geointel GEOINTEL_POSTGRES_PASSWORD=geointel GEOINTEL_CORS_ORIGINS=http://localhost:1202,http://127.0.0.1:1202 GEOINTEL_MAX_UPLOAD_MB=500 +GEOINTEL_AOI_WORKER_ENABLED=false +GEOINTEL_AOI_WORKER_POLL_SECONDS=2 diff --git a/backend/alembic/versions/202607260001_aoi_operations.py b/backend/alembic/versions/202607260001_aoi_operations.py new file mode 100644 index 00000000..87168665 --- /dev/null +++ b/backend/alembic/versions/202607260001_aoi_operations.py @@ -0,0 +1,65 @@ +"""Add resumable AOI parent and partition operations.""" + +from alembic import op +import sqlalchemy as sa +from geoalchemy2 import Geometry + + +revision = "202607260001" +down_revision = "202607160001" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.create_table( + "aoi_operations", + sa.Column("id", sa.UUID(), primary_key=True), + sa.Column("project_id", sa.UUID(), sa.ForeignKey("projects.id", ondelete="CASCADE"), nullable=False), + sa.Column("area_id", sa.UUID(), sa.ForeignKey("areas.id", ondelete="SET NULL")), + sa.Column("parent_job_id", sa.UUID(), sa.ForeignKey("jobs.id", ondelete="SET NULL")), + sa.Column("operation_type", sa.String(128), nullable=False), + sa.Column("status", sa.String(32), nullable=False), + sa.Column("geometry", Geometry("MultiPolygon", srid=4326, spatial_index=False), nullable=False), + sa.Column("request_json", sa.JSON(), nullable=False), + sa.Column("plan_json", sa.JSON(), nullable=False), + sa.Column("result_json", sa.JSON()), + sa.Column("error_message", sa.Text()), + sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now()), + sa.Column("started_at", sa.DateTime(timezone=True)), + sa.Column("finished_at", sa.DateTime(timezone=True)), + sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now()), + sa.CheckConstraint("status IN ('queued', 'running', 'partial', 'success', 'failed', 'cancelled')", name="ck_aoi_operations_status"), + ) + op.create_index("ix_aoi_operations_project_status", "aoi_operations", ["project_id", "status"]) + op.create_index("ix_aoi_operations_geometry", "aoi_operations", ["geometry"], postgresql_using="gist") + op.create_table( + "aoi_operation_partitions", + sa.Column("id", sa.UUID(), primary_key=True), + sa.Column("operation_id", sa.UUID(), sa.ForeignKey("aoi_operations.id", ondelete="CASCADE"), nullable=False), + sa.Column("child_job_id", sa.UUID(), sa.ForeignKey("jobs.id", ondelete="SET NULL")), + sa.Column("partition_key", sa.String(255), nullable=False), + sa.Column("provider_key", sa.String(120), nullable=False), + sa.Column("product_key", sa.String(120), nullable=False), + sa.Column("ordinal", sa.Integer(), nullable=False), + sa.Column("status", sa.String(32), nullable=False), + sa.Column("geometry", Geometry("MultiPolygon", srid=4326, spatial_index=False), nullable=False), + sa.Column("attempt_count", sa.Integer(), nullable=False, server_default="0"), + sa.Column("max_attempts", sa.Integer(), nullable=False, server_default="3"), + sa.Column("checkpoint_json", sa.JSON()), + sa.Column("result_json", sa.JSON()), + sa.Column("error_message", sa.Text()), + sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now()), + sa.Column("started_at", sa.DateTime(timezone=True)), + sa.Column("finished_at", sa.DateTime(timezone=True)), + sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now()), + sa.CheckConstraint("status IN ('queued', 'running', 'success', 'failed', 'skipped')", name="ck_aoi_operation_partitions_status"), + sa.UniqueConstraint("operation_id", "partition_key", name="uq_aoi_operation_partition_key"), + ) + op.create_index("ix_aoi_operation_partitions_operation_status", "aoi_operation_partitions", ["operation_id", "status"]) + op.create_index("ix_aoi_operation_partitions_geometry", "aoi_operation_partitions", ["geometry"], postgresql_using="gist") + + +def downgrade() -> None: + op.drop_table("aoi_operation_partitions") + op.drop_table("aoi_operations") diff --git a/backend/app/api/routes/aoi_operations.py b/backend/app/api/routes/aoi_operations.py new file mode 100644 index 00000000..2929ea47 --- /dev/null +++ b/backend/app/api/routes/aoi_operations.py @@ -0,0 +1,58 @@ +from __future__ import annotations + +from uuid import UUID + +from fastapi import APIRouter, Depends, Query +from sqlalchemy.orm import Session + +from app.db.session import get_db +from app.schemas.aoi_operation import AoiOperationCreate, AoiOperationList, AoiOperationRead, AoiPartitionCheckpoint, AoiPartitionComplete, AoiPartitionFail, AoiPartitionRead +from app.schemas.common import Envelope +from app.services.aoi_operation_service import AoiOperationService +from app.services.aoi_operation_executor import AoiOperationExecutor +from app.utils.response import envelope + + +router = APIRouter(prefix="/projects/{project_id}/aoi-operations", tags=["aoi-operations"]) + + +@router.post("", status_code=201, response_model=Envelope[AoiOperationRead]) +def create_operation(project_id: UUID, payload: AoiOperationCreate, db: Session = Depends(get_db)): + return envelope(AoiOperationService.create(db, project_id, payload)) + + +@router.get("", response_model=Envelope[AoiOperationList]) +def list_operations(project_id: UUID, limit: int = Query(default=50, ge=1, le=200), db: Session = Depends(get_db)): + return envelope(AoiOperationService.list(db, project_id, limit)) + + +@router.get("/{operation_id}", response_model=Envelope[AoiOperationRead]) +def read_operation(project_id: UUID, operation_id: UUID, db: Session = Depends(get_db)): + return envelope(AoiOperationService.read(db, project_id, operation_id)) + + +@router.post("/{operation_id}/partitions/claim", response_model=Envelope[AoiPartitionRead | None]) +def claim_partition(project_id: UUID, operation_id: UUID, db: Session = Depends(get_db)): + partition = AoiOperationService.claim_next(db, project_id, operation_id) + return envelope(AoiPartitionRead.model_validate(partition).model_dump() if partition else None) + + +@router.post("/{operation_id}/execute-next", response_model=Envelope[AoiOperationRead]) +def execute_next_partition(project_id: UUID, operation_id: UUID, db: Session = Depends(get_db)): + return envelope(AoiOperationExecutor.execute_next(db, project_id, operation_id)) + + +@router.put("/{operation_id}/partitions/{partition_id}/checkpoint", response_model=Envelope[AoiPartitionRead]) +def checkpoint_partition(project_id: UUID, operation_id: UUID, partition_id: UUID, payload: AoiPartitionCheckpoint, db: Session = Depends(get_db)): + partition = AoiOperationService.checkpoint(db, project_id, operation_id, partition_id, payload.checkpoint_json) + return envelope(AoiPartitionRead.model_validate(partition).model_dump()) + + +@router.post("/{operation_id}/partitions/{partition_id}/complete", response_model=Envelope[AoiOperationRead]) +def complete_partition(project_id: UUID, operation_id: UUID, partition_id: UUID, payload: AoiPartitionComplete, db: Session = Depends(get_db)): + return envelope(AoiOperationService.complete(db, project_id, operation_id, partition_id, payload.result_json, payload.skipped)) + + +@router.post("/{operation_id}/partitions/{partition_id}/fail", response_model=Envelope[AoiOperationRead]) +def fail_partition(project_id: UUID, operation_id: UUID, partition_id: UUID, payload: AoiPartitionFail, db: Session = Depends(get_db)): + return envelope(AoiOperationService.fail(db, project_id, operation_id, partition_id, payload.error_message, payload.retryable, payload.details)) diff --git a/backend/app/core/config.py b/backend/app/core/config.py index f549f66f..3ce6494e 100644 --- a/backend/app/core/config.py +++ b/backend/app/core/config.py @@ -324,6 +324,8 @@ class Settings(BaseSettings): default=False, validation_alias="GEOINTEL_RECONCILE_INTERRUPTED_RUNS_ON_STARTUP", ) + aoi_worker_enabled: bool = Field(default=False, validation_alias="GEOINTEL_AOI_WORKER_ENABLED") + aoi_worker_poll_seconds: float = Field(default=2.0, ge=0.5, le=60.0, validation_alias="GEOINTEL_AOI_WORKER_POLL_SECONDS") database_statement_timeout_ms: int = Field(default=5_000, validation_alias="DATABASE_STATEMENT_TIMEOUT_MS") yolo_enabled: bool = Field(default=False, validation_alias="YOLO_ENABLED") yolo_models_dir: str = Field(default="/app/models", validation_alias="YOLO_MODELS_DIR") @@ -331,7 +333,11 @@ class Settings(BaseSettings): yolo_model_id: str = Field(default="yolo-configured", validation_alias="YOLO_MODEL_ID") yolo_model_display_name: str = Field(default="Configured YOLO detector", validation_alias="YOLO_MODEL_DISPLAY_NAME") yolo_model_version: str | None = Field(default=None, validation_alias="YOLO_MODEL_VERSION") + yolo_model_classes: str = Field(default="building", validation_alias="YOLO_MODEL_CLASSES") + yolo_enforce_validation_scope: bool = Field(default=False, validation_alias="YOLO_ENFORCE_VALIDATION_SCOPE") + yolo_validated_area_names: str = Field(default="Mol,Kempen", validation_alias="YOLO_VALIDATED_AREA_NAMES") yolo_device: str = Field(default="cpu", validation_alias="YOLO_DEVICE") + yolo_require_cuda: bool = Field(default=False, validation_alias="YOLO_REQUIRE_CUDA") yolo_image_size: int = Field(default=640, validation_alias="YOLO_IMAGE_SIZE") yolo_max_tiles: int = Field(default=100, validation_alias="YOLO_MAX_TILES") yolo_max_detections: int = Field(default=1000, validation_alias="YOLO_MAX_DETECTIONS") diff --git a/backend/app/main.py b/backend/app/main.py index bfa4d203..834d89c4 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -1,6 +1,7 @@ from __future__ import annotations import logging +import asyncio import re import time import uuid @@ -11,7 +12,7 @@ from fastapi.exceptions import RequestValidationError from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import JSONResponse -from app.api.routes import analysis, areas, assistant, auth, datasets, demo, detection, exports, external, health, jobs, projects, qa, quality_checks, segmentation, selection_partitions, temporal +from app.api.routes import analysis, aoi_operations, areas, assistant, auth, datasets, demo, detection, exports, external, health, jobs, projects, qa, quality_checks, segmentation, selection_partitions, temporal from app.core.config import get_settings from app.core.errors import AppError from app.core.logging import configure_logging @@ -19,6 +20,7 @@ from app.core.request_context import reset_request_id, set_request_id from app.db.session import SessionLocal from app.services.runtime_reconciliation_service import RuntimeReconciliationService from app.services.auth_service import AuthService +from app.services.aoi_operation_worker import AoiOperationWorker logger = logging.getLogger("geointel") @@ -46,14 +48,18 @@ def create_app() -> FastAPI: @asynccontextmanager async def lifespan(_: FastAPI): + worker_stop = asyncio.Event() + worker_task = None if settings.reconcile_interrupted_runs_on_startup: db = SessionLocal() try: result = RuntimeReconciliationService.reconcile(db) logger.info( - "Runtime reconciliation completed: jobs=%s analysis_runs=%s", + "Runtime reconciliation completed: jobs=%s analysis_runs=%s resumed_aoi_partitions=%s exhausted_aoi_partitions=%s", result.interrupted_jobs, result.interrupted_analysis_runs, + result.resumed_aoi_partitions, + result.exhausted_aoi_partitions, ) except Exception: db.rollback() @@ -61,7 +67,14 @@ def create_app() -> FastAPI: raise finally: db.close() - yield + if settings.aoi_worker_enabled: + worker_task = asyncio.create_task(AoiOperationWorker.run(worker_stop, settings.aoi_worker_poll_seconds)) + try: + yield + finally: + worker_stop.set() + if worker_task is not None: + await worker_task app = FastAPI( title="GeoIntel", @@ -82,6 +95,7 @@ def create_app() -> FastAPI: app.include_router(health.router) app.include_router(auth.router, prefix=settings.api_prefix) app.include_router(analysis.router, prefix=settings.api_prefix) + app.include_router(aoi_operations.router, prefix=settings.api_prefix) app.include_router(projects.router, prefix=settings.api_prefix) app.include_router(areas.router, prefix=settings.api_prefix) app.include_router(datasets.router, prefix=settings.api_prefix) diff --git a/backend/app/models/__init__.py b/backend/app/models/__init__.py index 447ffd0b..7d8d5fcd 100644 --- a/backend/app/models/__init__.py +++ b/backend/app/models/__init__.py @@ -1,7 +1,9 @@ -from .entities import AnalysisRun, Area, Dataset, DatasetVersion, Detection, DetectionReview, Export, Job, Metric, Project, QualityCheck, Segmentation, VectorFeature +from .entities import AoiOperation, AoiOperationPartition, AnalysisRun, Area, Dataset, DatasetVersion, Detection, DetectionReview, Export, Job, Metric, Project, QualityCheck, Segmentation, VectorFeature __all__ = [ "AnalysisRun", + "AoiOperation", + "AoiOperationPartition", "Area", "Dataset", "DatasetVersion", diff --git a/backend/app/models/entities.py b/backend/app/models/entities.py index f8b3b28e..499e82f0 100644 --- a/backend/app/models/entities.py +++ b/backend/app/models/entities.py @@ -361,3 +361,63 @@ class Job(Base): created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now()) started_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) finished_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + + +class AoiOperation(Base): + __tablename__ = "aoi_operations" + __table_args__ = ( + CheckConstraint( + "status IN ('queued', 'running', 'partial', 'success', 'failed', 'cancelled')", + name="ck_aoi_operations_status", + ), + Index("ix_aoi_operations_project_status", "project_id", "status"), + Index("ix_aoi_operations_geometry", "geometry", postgresql_using="gist"), + ) + + id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) + project_id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), ForeignKey("projects.id", ondelete="CASCADE"), nullable=False) + area_id: Mapped[uuid.UUID | None] = mapped_column(UUID(as_uuid=True), ForeignKey("areas.id", ondelete="SET NULL"), nullable=True) + parent_job_id: Mapped[uuid.UUID | None] = mapped_column(UUID(as_uuid=True), ForeignKey("jobs.id", ondelete="SET NULL"), nullable=True) + operation_type: Mapped[str] = mapped_column(String(128), nullable=False) + status: Mapped[str] = mapped_column(String(32), nullable=False, default="queued") + geometry: Mapped[str] = mapped_column(Geometry("MultiPolygon", srid=4326, spatial_index=False), nullable=False) + request_json: Mapped[dict] = mapped_column(JSON, nullable=False, default=dict) + plan_json: Mapped[dict] = mapped_column(JSON, nullable=False, default=dict) + result_json: Mapped[dict | None] = mapped_column(JSON, nullable=True) + error_message: Mapped[str | None] = mapped_column(Text, nullable=True) + created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now()) + started_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + finished_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now()) + + +class AoiOperationPartition(Base): + __tablename__ = "aoi_operation_partitions" + __table_args__ = ( + CheckConstraint( + "status IN ('queued', 'running', 'success', 'failed', 'skipped')", + name="ck_aoi_operation_partitions_status", + ), + UniqueConstraint("operation_id", "partition_key", name="uq_aoi_operation_partition_key"), + Index("ix_aoi_operation_partitions_operation_status", "operation_id", "status"), + Index("ix_aoi_operation_partitions_geometry", "geometry", postgresql_using="gist"), + ) + + id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) + operation_id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), ForeignKey("aoi_operations.id", ondelete="CASCADE"), nullable=False) + child_job_id: Mapped[uuid.UUID | None] = mapped_column(UUID(as_uuid=True), ForeignKey("jobs.id", ondelete="SET NULL"), nullable=True) + partition_key: Mapped[str] = mapped_column(String(255), nullable=False) + provider_key: Mapped[str] = mapped_column(String(120), nullable=False) + product_key: Mapped[str] = mapped_column(String(120), nullable=False) + ordinal: Mapped[int] = mapped_column(Integer, nullable=False) + status: Mapped[str] = mapped_column(String(32), nullable=False, default="queued") + geometry: Mapped[str] = mapped_column(Geometry("MultiPolygon", srid=4326, spatial_index=False), nullable=False) + attempt_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + max_attempts: Mapped[int] = mapped_column(Integer, nullable=False, default=3) + checkpoint_json: Mapped[dict | None] = mapped_column(JSON, nullable=True) + result_json: Mapped[dict | None] = mapped_column(JSON, nullable=True) + error_message: Mapped[str | None] = mapped_column(Text, nullable=True) + created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now()) + started_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + finished_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now()) diff --git a/backend/app/schemas/aoi_operation.py b/backend/app/schemas/aoi_operation.py new file mode 100644 index 00000000..d6b11a8b --- /dev/null +++ b/backend/app/schemas/aoi_operation.py @@ -0,0 +1,75 @@ +from __future__ import annotations + +from datetime import datetime +from uuid import UUID + +from pydantic import BaseModel, Field + +from app.schemas.operations import VectorSelectionBBox + + +class AoiOperationCreate(BaseModel): + area_id: UUID | None = None + bbox: VectorSelectionBBox | None = None + operation_type: str = Field(min_length=1, max_length=128) + provider_key: str = Field(min_length=1, max_length=120) + product_key: str = Field(min_length=1, max_length=120) + coverage_zone: str | None = Field(default=None, max_length=64) + max_partition_side_m: float | None = Field(default=None, gt=0, le=60_000) + max_attempts: int = Field(default=3, ge=1, le=10) + parameters_json: dict = Field(default_factory=dict) + + +class AoiPartitionRead(BaseModel): + id: UUID + partition_key: str + provider_key: str + product_key: str + ordinal: int + status: str + attempt_count: int + max_attempts: int + checkpoint_json: dict | None = None + result_json: dict | None = None + error_message: str | None = None + + model_config = {"from_attributes": True} + + +class AoiOperationRead(BaseModel): + id: UUID + project_id: UUID + area_id: UUID | None = None + parent_job_id: UUID | None = None + operation_type: str + status: str + request_json: dict + plan_json: dict + result_json: dict | None = None + error_message: str | None = None + progress: float + partition_counts: dict[str, int] + partitions: list[AoiPartitionRead] = Field(default_factory=list) + created_at: datetime | None = None + started_at: datetime | None = None + finished_at: datetime | None = None + + +class AoiOperationList(BaseModel): + items: list[AoiOperationRead] + total: int + + +class AoiPartitionCheckpoint(BaseModel): + checkpoint_json: dict = Field(default_factory=dict) + + +class AoiPartitionComplete(BaseModel): + result_json: dict = Field(default_factory=dict) + skipped: bool = False + + +class AoiPartitionFail(BaseModel): + error_message: str = Field(min_length=1, max_length=4000) + retryable: bool = True + details: dict = Field(default_factory=dict) diff --git a/backend/app/schemas/coverage.py b/backend/app/schemas/coverage.py index 9091f740..d3e13da4 100644 --- a/backend/app/schemas/coverage.py +++ b/backend/app/schemas/coverage.py @@ -65,9 +65,25 @@ class CoverageResolutionItem(BaseModel): status: CoverageStatus source_names: list[str] materialized_dataset_ids: list[UUID] + evidence: list["CoverageEvidenceItem"] = Field(default_factory=list) limitation_message: str +class CoverageEvidenceItem(BaseModel): + dataset_id: UUID + source_name: str + authority_level: CoverageAuthority + source_version: str | None = None + observed_at: str | None = None + published_at: str | None = None + crs: str | None = None + resolution: dict | None = None + coverage_bbox_epsg4326: list[float] | None = None + attribution: str | None = None + license_note: str | None = None + checksum_sha256: str | None = None + + class CoverageResolveResponse(BaseModel): project_id: UUID bbox: CoverageBBox diff --git a/backend/app/schemas/detection.py b/backend/app/schemas/detection.py index feaaad02..e11e5ea8 100644 --- a/backend/app/schemas/detection.py +++ b/backend/app/schemas/detection.py @@ -141,6 +141,7 @@ class YoloPreflightChecks(BaseModel): enabled: bool dependencies_available: bool | None = None + accelerator_ready: bool | None = None model_path_set: bool | None = None model_file_exists: bool | None = None model_load_requested: bool @@ -160,6 +161,8 @@ class YoloRuntimeDetails(BaseModel): torch_version: str | None = None ultralytics_version: str | None = None cuda_available: bool | None = None + configured_device: str + cuda_required: bool class YoloPreflightResponse(BaseModel): diff --git a/backend/app/schemas/dhmv.py b/backend/app/schemas/dhmv.py index 53dc9fe0..74cc1a39 100644 --- a/backend/app/schemas/dhmv.py +++ b/backend/app/schemas/dhmv.py @@ -58,7 +58,7 @@ class TerrainSelectionRequest(BaseModel): class TerrainPartitionSelectionRequest(TerrainSelectionRequest): product_key: str = "dtm_1m" - dataset_ids: list[UUID] | None = Field(default=None, min_length=1, max_length=16) + dataset_ids: list[UUID] | None = Field(default=None, min_length=1, max_length=4096) class TerrainMetric(BaseModel): diff --git a/backend/app/schemas/flood_hazard.py b/backend/app/schemas/flood_hazard.py index 485de728..452e404c 100644 --- a/backend/app/schemas/flood_hazard.py +++ b/backend/app/schemas/flood_hazard.py @@ -61,7 +61,7 @@ class FloodHazardSelectionRequest(BaseModel): class FloodHazardPartitionSelectionRequest(FloodHazardSelectionRequest): product_key: str = "pluviaal_current_t100" - dataset_ids: list[UUID] | None = Field(default=None, min_length=1, max_length=16) + dataset_ids: list[UUID] | None = Field(default=None, min_length=1, max_length=4096) class FloodHazardMetric(BaseModel): diff --git a/backend/app/schemas/selection_partitions.py b/backend/app/schemas/selection_partitions.py index d58a2b39..02201762 100644 --- a/backend/app/schemas/selection_partitions.py +++ b/backend/app/schemas/selection_partitions.py @@ -8,7 +8,7 @@ from .operations import VectorSelectionBBox class VectorPartitionSelectionRequest(BaseModel): - dataset_ids: list[UUID] = Field(min_length=1, max_length=16) + dataset_ids: list[UUID] = Field(min_length=1, max_length=4096) bbox: VectorSelectionBBox area_id: UUID | None = None limit: int = Field(default=1000, ge=1, le=1000) diff --git a/backend/app/services/aoi_operation_executor.py b/backend/app/services/aoi_operation_executor.py new file mode 100644 index 00000000..dc521549 --- /dev/null +++ b/backend/app/services/aoi_operation_executor.py @@ -0,0 +1,104 @@ +from __future__ import annotations + +from uuid import UUID + +from geoalchemy2.shape import to_shape + +from app.core.errors import AppError +from app.models import AoiOperation, AoiOperationPartition +from app.schemas.grb import GrbAcquireRequest +from app.schemas.dhmv import DhmvAcquireRequest +from app.schemas.spw_terrain import SpwTerrainAcquireRequest +from app.schemas.official_vector import OfficialVectorAcquireRequest +from app.schemas.flood_hazard import FloodHazardAcquireRequest +from app.schemas.thematic_raster import ThematicRasterAcquireRequest +from app.schemas.bathymetry import BathymetryProfileAcquireRequest, MdkBathymetryAcquireRequest +from app.schemas.job import JobCreate +from app.schemas.operations import VectorSelectionBBox +from app.schemas.orthophoto import OrthophotoAcquireRequest +from app.services.aoi_operation_service import AoiOperationService +from app.services.grb_acquisition_service import GrbAcquisitionService +from app.services.dhmv_acquisition_service import DhmvAcquisitionService +from app.services.spw_terrain_service import SpwTerrainService +from app.services.official_vector_acquisition_service import OfficialVectorAcquisitionService +from app.services.flood_hazard_acquisition_service import FloodHazardAcquisitionService +from app.services.thematic_raster_acquisition_service import ThematicRasterAcquisitionService +from app.services.walous_land_cover_service import WalousLandCoverService +from app.services.bathymetry_profile_acquisition_service import BathymetryProfileAcquisitionService +from app.services.mdk_bathymetry_acquisition_service import MdkBathymetryAcquisitionService +from app.services.job_service import JobService +from app.services.orthophoto_acquisition_service import OrthophotoAcquisitionService + + +class AoiOperationExecutor: + """Execute one bounded partition through an existing governed provider.""" + + @staticmethod + def execute_next(db, project_id: UUID, operation_id: UUID) -> dict: + partition = AoiOperationService.claim_next(db, project_id, operation_id) + if partition is None: + AoiOperationService._refresh_parent(db, operation_id) + return AoiOperationService.read(db, project_id, operation_id) + operation = db.get(AoiOperation, operation_id) + child = JobService.create_job(db, JobCreate( + job_type=f"aoi.{operation.operation_type}.partition", + project_id=project_id, + parameters_json={ + "aoi_operation_id": str(operation_id), + "partition_id": str(partition.id), + "partition_key": partition.partition_key, + "provider_key": partition.provider_key, + "product_key": partition.product_key, + }, + )) + partition = db.get(AoiOperationPartition, partition.id) + partition.child_job_id = child.id + db.add(partition); db.commit() + JobService.mark_running(db, child.id) + try: + result = AoiOperationExecutor._dispatch(db, project_id, operation, partition) + output_id = result.get("output_dataset_id") if isinstance(result, dict) else None + JobService.mark_success(db, child.id, result=result, output_dataset_id=UUID(str(output_id)) if output_id else None) + return AoiOperationService.complete(db, project_id, operation_id, partition.id, result) + except AppError as exc: + JobService.mark_failed(db, child.id, exc.message, {"code": exc.code, "details": exc.details}) + return AoiOperationService.fail(db, project_id, operation_id, partition.id, exc.message, AoiOperationExecutor._retryable(exc), {"code": exc.code, "details": exc.details}) + except Exception: + try: + db.rollback() + JobService.mark_failed(db, child.id, "Unexpected partition execution error", {"code": "AOI_PARTITION_INTERNAL_ERROR"}) + finally: + AoiOperationService.fail(db, project_id, operation_id, partition.id, "Unexpected partition execution error", True, {"code": "AOI_PARTITION_INTERNAL_ERROR"}) + raise + + @staticmethod + def _dispatch(db, project_id: UUID, operation: AoiOperation, partition: AoiOperationPartition) -> dict: + geometry = to_shape(partition.geometry) + min_x, min_y, max_x, max_y = geometry.bounds + bbox = VectorSelectionBBox(min_x=min_x, min_y=min_y, max_x=max_x, max_y=max_y, crs="EPSG:4326") + force_refresh = bool((operation.request_json or {}).get("parameters_json", {}).get("force_refresh", False)) + if partition.provider_key == "grb": + return GrbAcquisitionService.acquire(db, project_id, GrbAcquireRequest(bbox=bbox, area_id=operation.area_id, product_key=partition.product_key, force_refresh=force_refresh)) + if partition.provider_key == "orthophoto": + return OrthophotoAcquisitionService.acquire(db, project_id, OrthophotoAcquireRequest(bbox=bbox, area_id=operation.area_id, product_key=partition.product_key, force_refresh=force_refresh)) + if partition.provider_key == "dhmv": + return DhmvAcquisitionService.acquire(db, project_id, DhmvAcquireRequest(bbox=bbox, area_id=operation.area_id, product_key=partition.product_key, force_refresh=force_refresh)) + if partition.provider_key == "spw_terrain": + return SpwTerrainService.acquire(db, project_id, SpwTerrainAcquireRequest(bbox=bbox, area_id=operation.area_id, product_key=partition.product_key, force_refresh=force_refresh)) + if partition.provider_key == "official_vector": + return OfficialVectorAcquisitionService.acquire(db, project_id, OfficialVectorAcquireRequest(bbox=bbox, area_id=operation.area_id, product_key=partition.product_key, force_refresh=force_refresh)) + if partition.provider_key == "flood_hazard": + return FloodHazardAcquisitionService.acquire(db, project_id, FloodHazardAcquireRequest(bbox=bbox, area_id=operation.area_id, product_key=partition.product_key, force_refresh=force_refresh)) + if partition.provider_key == "thematic_raster": + return ThematicRasterAcquisitionService.acquire(db, project_id, ThematicRasterAcquireRequest(bbox=bbox, area_id=operation.area_id, product_key=partition.product_key, force_refresh=force_refresh)) + if partition.provider_key == "walous": + return WalousLandCoverService.acquire(db, project_id, ThematicRasterAcquireRequest(bbox=bbox, area_id=operation.area_id, product_key=partition.product_key, force_refresh=force_refresh)) + if partition.provider_key == "bathymetry_profiles": + return BathymetryProfileAcquisitionService.acquire(db, project_id, BathymetryProfileAcquireRequest(bbox=bbox, area_id=operation.area_id, force_refresh=force_refresh)) + if partition.provider_key == "mdk_bathymetry": + return MdkBathymetryAcquisitionService.acquire(db, project_id, MdkBathymetryAcquireRequest(bbox=bbox, area_id=operation.area_id, force_refresh=force_refresh)) + raise AppError(code="AOI_PROVIDER_UNSUPPORTED", message="No governed AOI executor is registered for this provider", details={"provider_key": partition.provider_key}, status_code=422) + + @staticmethod + def _retryable(error: AppError) -> bool: + return error.status_code >= 500 or error.code.endswith(("TIMEOUT", "UNAVAILABLE", "TLS_ERROR")) diff --git a/backend/app/services/aoi_operation_service.py b/backend/app/services/aoi_operation_service.py new file mode 100644 index 00000000..16fa3290 --- /dev/null +++ b/backend/app/services/aoi_operation_service.py @@ -0,0 +1,276 @@ +from __future__ import annotations + +from collections import Counter +from datetime import datetime, timezone +from hashlib import sha256 +import math +from uuid import UUID, uuid4 + +from geoalchemy2.shape import from_shape, to_shape +from pyproj import Transformer +from shapely.geometry import MultiPolygon, Polygon, box +from shapely.ops import transform + +from app.core.errors import AppError +from app.core.config import get_settings +from app.models import AoiOperation, AoiOperationPartition, Area, Project +from app.schemas.aoi_operation import AoiOperationCreate + + +class AoiOperationService: + MAX_PARTITIONS = 4096 + _to_metric = Transformer.from_crs("EPSG:4326", "EPSG:31370", always_xy=True) + _to_wgs84 = Transformer.from_crs("EPSG:31370", "EPSG:4326", always_xy=True) + SCOPE_AREA_NAMES = { + "belgium": "Belgium land", "flanders": "Flanders", "wallonia": "Wallonia", + "brussels": "Brussels-Capital Region", "belgian_north_sea": "Belgian part of the North Sea", + "territorial_sea": "Belgian territorial sea (0-12 nautical miles)", + "exclusive_economic_zone": "Belgian exclusive economic zone beyond territorial sea", + "continental_shelf": "Belgian continental shelf beyond territorial sea", + } + + @staticmethod + def create(db, project_id: UUID, payload: AoiOperationCreate) -> dict: + if db.get(Project, project_id) is None: + raise AppError(code="PROJECT_NOT_FOUND", message="Project not found", status_code=404) + geometry = AoiOperationService._resolve_geometry(db, project_id, payload) + if payload.coverage_zone: + geometry = AoiOperationService._clip_to_zone(db, project_id, geometry, payload.coverage_zone) + geometry = AoiOperationService._as_multipolygon(geometry) + metric_geometry = transform(AoiOperationService._to_metric.transform, geometry) + partition_side_m = AoiOperationService._partition_side(payload.provider_key, payload.max_partition_side_m) + cells = AoiOperationService._partition(metric_geometry, partition_side_m) + operation_id = uuid4() + now = datetime.now(timezone.utc) + operation = AoiOperation( + id=operation_id, + project_id=project_id, + area_id=payload.area_id, + operation_type=payload.operation_type, + status="queued", + geometry=from_shape(geometry, srid=4326), + request_json=payload.model_dump(mode="json", exclude_none=True), + plan_json={ + "partition_strategy": "epsg31370_square_grid_intersection_v1", + "max_partition_side_m": partition_side_m, + "budget_source": "governed_provider_registry" if payload.max_partition_side_m is None else "stricter_operator_override", + "partition_count": len(cells), + "provider_key": payload.provider_key, + "product_key": payload.product_key, + }, + created_at=now, + ) + db.add(operation) + for ordinal, cell in enumerate(cells): + wgs84 = transform(AoiOperationService._to_wgs84.transform, cell) + wgs84 = AoiOperationService._as_multipolygon(wgs84) + digest = sha256(wgs84.wkb).hexdigest()[:20] + db.add(AoiOperationPartition( + id=uuid4(), operation_id=operation_id, + partition_key=f"{payload.provider_key}:{payload.product_key}:{ordinal:05d}:{digest}", + provider_key=payload.provider_key, product_key=payload.product_key, + ordinal=ordinal, status="queued", geometry=from_shape(wgs84, srid=4326), + attempt_count=0, max_attempts=payload.max_attempts, created_at=now, + )) + db.commit() + return AoiOperationService.read(db, project_id, operation_id) + + @staticmethod + def _clip_to_zone(db, project_id: UUID, geometry, zone: str): + area_name = AoiOperationService.SCOPE_AREA_NAMES.get(zone) + if area_name is None: + raise AppError(code="AOI_COVERAGE_ZONE_UNSUPPORTED", message="Unknown governed coverage zone", details={"coverage_zone": zone}, status_code=422) + scope = db.query(Area).filter(Area.project_id == project_id, Area.name == area_name).first() + if scope is None: + raise AppError(code="AOI_COVERAGE_ZONE_NOT_MATERIALIZED", message="The governed coverage-zone geometry is not persisted in this project", details={"coverage_zone": zone}, status_code=409) + clipped = geometry.intersection(to_shape(scope.geometry)) + if clipped.is_empty: + raise AppError(code="AOI_OUTSIDE_PROVIDER_ZONE", message="The AOI does not intersect the provider coverage zone", details={"coverage_zone": zone}, status_code=422) + return clipped + + @staticmethod + def _as_multipolygon(geometry) -> MultiPolygon: + if isinstance(geometry, Polygon): + return MultiPolygon([geometry]) + if isinstance(geometry, MultiPolygon): + return geometry + polygons = [part for part in getattr(geometry, "geoms", []) if isinstance(part, Polygon)] + if not polygons: + raise AppError(code="AOI_GEOMETRY_EMPTY", message="AOI contains no polygonal area after clipping", status_code=422) + return MultiPolygon(polygons) + + @staticmethod + def _partition_side(provider_key: str, requested: float | None) -> float: + settings = get_settings() + def raster_side(max_side_m: float, max_pixels: int, resolution_m: float) -> float: + # Keep every square grid cell within both the provider's spatial + # extent limit and its decoded-pixel budget. The small safety + # margin absorbs ceil/edge rounding in the acquisition services. + pixel_limited_side = math.sqrt(float(max_pixels)) * float(resolution_m) * 0.99 + return min(float(max_side_m), pixel_limited_side) + + budgets = { + "orthophoto": float(settings.orthophoto_max_side_m), + "grb": float(settings.grb_max_side_m), + "dhmv": raster_side( + settings.dhmv_max_side_m, + settings.dhmv_max_pixels, + settings.dhmv_resolution_m, + ), + "spw_terrain": raster_side( + settings.spw_terrain_max_side_m, + settings.spw_terrain_max_pixels, + settings.spw_terrain_analysis_resolution_m, + ), + "official_vector": 20_000.0, + "flood_hazard": raster_side( + settings.flood_hazard_max_side_m, + settings.flood_hazard_max_pixels, + settings.flood_hazard_resolution_m, + ), + "thematic_raster": raster_side( + settings.thematic_raster_max_side_m, + settings.thematic_raster_max_pixels, + 10.0, + ), + "walous": raster_side( + settings.walous_max_side_m, + settings.walous_max_pixels, + settings.walous_analysis_resolution_m, + ), + "bathymetry_profiles": 20_000.0, + "mdk_bathymetry": 20_000.0, + } + if provider_key not in budgets: + raise AppError(code="AOI_PROVIDER_UNSUPPORTED", message="No governed partition budget is registered for this provider", details={"provider_key": provider_key}, status_code=422) + governed = budgets[provider_key] + return min(governed, float(requested)) if requested is not None else governed + + @staticmethod + def _resolve_geometry(db, project_id: UUID, payload: AoiOperationCreate): + if (payload.area_id is None) == (payload.bbox is None): + raise AppError(code="AOI_SELECTION_REQUIRED", message="Provide exactly one area_id or bbox", status_code=422) + if payload.area_id is not None: + area = db.get(Area, payload.area_id) + if area is None or area.project_id != project_id: + raise AppError(code="AREA_NOT_FOUND", message="Area not found", status_code=404) + return to_shape(area.geometry) + bbox = payload.bbox + if bbox is None or bbox.crs != "EPSG:4326": + raise AppError(code="INVALID_AOI_CRS", message="AOI bbox must use EPSG:4326", status_code=422) + return box(bbox.min_x, bbox.min_y, bbox.max_x, bbox.max_y) + + @staticmethod + def _partition(geometry, side_m: float) -> list: + min_x, min_y, max_x, max_y = geometry.bounds + columns = max(1, math.ceil((max_x - min_x) / side_m)) + rows = max(1, math.ceil((max_y - min_y) / side_m)) + if columns * rows > AoiOperationService.MAX_PARTITIONS: + raise AppError(code="AOI_PARTITION_LIMIT_EXCEEDED", message="AOI requires too many bounded partitions", details={"candidate_count": columns * rows, "max_partitions": AoiOperationService.MAX_PARTITIONS}, status_code=422) + partitions = [] + for row in range(rows): + for column in range(columns): + clipped = geometry.intersection(box(min_x + column * side_m, min_y + row * side_m, min(min_x + (column + 1) * side_m, max_x), min(min_y + (row + 1) * side_m, max_y))) + if not clipped.is_empty and clipped.area > 0: + partitions.append(clipped) + return partitions + + @staticmethod + def read(db, project_id: UUID, operation_id: UUID) -> dict: + operation = db.get(AoiOperation, operation_id) + if operation is None or operation.project_id != project_id: + raise AppError(code="AOI_OPERATION_NOT_FOUND", message="AOI operation not found", status_code=404) + partitions = db.query(AoiOperationPartition).filter(AoiOperationPartition.operation_id == operation_id).order_by(AoiOperationPartition.ordinal).all() + counts = Counter(partition.status for partition in partitions) + complete = counts["success"] + counts["skipped"] + return { + "id": operation.id, "project_id": operation.project_id, "area_id": operation.area_id, + "parent_job_id": operation.parent_job_id, "operation_type": operation.operation_type, + "status": operation.status, "request_json": operation.request_json, "plan_json": operation.plan_json, + "result_json": operation.result_json, "error_message": operation.error_message, + "progress": round(complete / len(partitions), 6) if partitions else 0.0, + "partition_counts": dict(counts), "partitions": partitions, + "created_at": operation.created_at, "started_at": operation.started_at, "finished_at": operation.finished_at, + } + + @staticmethod + def list(db, project_id: UUID, limit: int = 50) -> dict: + rows = db.query(AoiOperation).filter(AoiOperation.project_id == project_id).order_by(AoiOperation.created_at.desc()).limit(limit).all() + return {"items": [AoiOperationService.read(db, project_id, row.id) for row in rows], "total": len(rows)} + + @staticmethod + def claim_next(db, project_id: UUID, operation_id: UUID): + operation = db.get(AoiOperation, operation_id) + if operation is None or operation.project_id != project_id: + raise AppError(code="AOI_OPERATION_NOT_FOUND", message="AOI operation not found", status_code=404) + partition = db.query(AoiOperationPartition).filter(AoiOperationPartition.operation_id == operation_id, AoiOperationPartition.status == "queued").order_by(AoiOperationPartition.ordinal).with_for_update(skip_locked=True).first() + if partition is None: + return None + now = datetime.now(timezone.utc) + partition.status = "running"; partition.started_at = now; partition.attempt_count += 1; partition.error_message = None + operation.status = "running"; operation.started_at = operation.started_at or now + db.add(partition); db.add(operation); db.commit(); db.refresh(partition) + return partition + + @staticmethod + def checkpoint(db, project_id: UUID, operation_id: UUID, partition_id: UUID, checkpoint: dict): + partition = AoiOperationService._partition_row(db, project_id, operation_id, partition_id) + if partition.status != "running": + raise AppError(code="AOI_PARTITION_NOT_RUNNING", message="Only a running partition can be checkpointed", status_code=409) + partition.checkpoint_json = checkpoint; db.add(partition); db.commit(); db.refresh(partition) + return partition + + @staticmethod + def complete(db, project_id: UUID, operation_id: UUID, partition_id: UUID, result: dict, skipped: bool = False): + partition = AoiOperationService._partition_row(db, project_id, operation_id, partition_id) + if partition.status == "success" or partition.status == "skipped": + return AoiOperationService.read(db, project_id, operation_id) + if partition.status != "running": + raise AppError(code="AOI_PARTITION_NOT_RUNNING", message="Only a running partition can complete", status_code=409) + partition.status = "skipped" if skipped else "success"; partition.result_json = result; partition.finished_at = datetime.now(timezone.utc) + db.add(partition); db.commit(); AoiOperationService._refresh_parent(db, operation_id) + return AoiOperationService.read(db, project_id, operation_id) + + @staticmethod + def fail(db, project_id: UUID, operation_id: UUID, partition_id: UUID, message: str, retryable: bool, details: dict): + partition = AoiOperationService._partition_row(db, project_id, operation_id, partition_id) + partition.error_message = message; partition.result_json = {"details": details} + partition.status = "queued" if retryable and partition.attempt_count < partition.max_attempts else "failed" + partition.finished_at = None if partition.status == "queued" else datetime.now(timezone.utc) + db.add(partition); db.commit(); AoiOperationService._refresh_parent(db, operation_id) + return AoiOperationService.read(db, project_id, operation_id) + + @staticmethod + def _partition_row(db, project_id, operation_id, partition_id): + operation = db.get(AoiOperation, operation_id); partition = db.get(AoiOperationPartition, partition_id) + if operation is None or operation.project_id != project_id or partition is None or partition.operation_id != operation_id: + raise AppError(code="AOI_PARTITION_NOT_FOUND", message="AOI partition not found", status_code=404) + return partition + + @staticmethod + def _refresh_parent(db, operation_id): + operation = db.get(AoiOperation, operation_id) + partitions = db.query(AoiOperationPartition).filter(AoiOperationPartition.operation_id == operation_id).order_by(AoiOperationPartition.ordinal).all() + statuses = [partition.status for partition in partitions] + output_dataset_ids = [] + for partition in partitions: + output_id = (partition.result_json or {}).get("output_dataset_id") if isinstance(partition.result_json, dict) else None + if output_id and str(output_id) not in output_dataset_ids: + output_dataset_ids.append(str(output_id)) + operation.result_json = { + "partition_count": len(partitions), + "completed_partition_count": sum(status in {"success", "skipped"} for status in statuses), + "failed_partition_count": statuses.count("failed"), + "output_dataset_ids": output_dataset_ids, + "merge_contract": "source_aware_spatial_union", + "vector_deduplication": "source_feature_id_then_geometry", + "raster_deduplication": "governed_mosaic_grid", + "complete_coverage": bool(statuses) and all(status in {"success", "skipped"} for status in statuses), + } + now = datetime.now(timezone.utc) + if statuses and all(status in {"success", "skipped"} for status in statuses): + operation.status = "success"; operation.finished_at = now; operation.error_message = None + elif "failed" in statuses and not any(status in {"queued", "running"} for status in statuses): + operation.status = "partial" if any(status in {"success", "skipped"} for status in statuses) else "failed"; operation.finished_at = now + operation.error_message = "One or more bounded source partitions failed; inspect partition evidence." + db.add(operation); db.commit() diff --git a/backend/app/services/aoi_operation_worker.py b/backend/app/services/aoi_operation_worker.py new file mode 100644 index 00000000..edfaac9d --- /dev/null +++ b/backend/app/services/aoi_operation_worker.py @@ -0,0 +1,38 @@ +from __future__ import annotations + +import asyncio +import logging + +from app.db.session import SessionLocal +from app.models import AoiOperation +from app.services.aoi_operation_executor import AoiOperationExecutor + + +logger = logging.getLogger("geointel.aoi_worker") + + +class AoiOperationWorker: + @staticmethod + def run_once() -> int: + db = SessionLocal() + try: + rows = db.query(AoiOperation).filter(AoiOperation.status.in_(("queued", "running"))).order_by(AoiOperation.created_at).limit(10).all() + for operation in rows: + try: + AoiOperationExecutor.execute_next(db, operation.project_id, operation.id) + except Exception: + db.rollback() + logger.exception("AOI partition execution failed operation_id=%s", operation.id) + return len(rows) + finally: + db.close() + + @staticmethod + async def run(stop_event: asyncio.Event, poll_seconds: float) -> None: + while not stop_event.is_set(): + processed = await asyncio.to_thread(AoiOperationWorker.run_once) + if processed == 0: + try: + await asyncio.wait_for(stop_event.wait(), timeout=poll_seconds) + except TimeoutError: + pass diff --git a/backend/app/services/coverage_registry_service.py b/backend/app/services/coverage_registry_service.py index 84545ca3..629cdb36 100644 --- a/backend/app/services/coverage_registry_service.py +++ b/backend/app/services/coverage_registry_service.py @@ -452,10 +452,12 @@ class CoverageRegistryService: theme: str, zone: str, selection: Any, - ) -> list[Dataset]: + ) -> tuple[list[Dataset], bool]: if definition.operational_themes and theme not in definition.operational_themes: - return [] + return [], False matches: list[Dataset] = [] + bounded_scopes: list[Any] = [] + zone_scoped_materialization = False for dataset in datasets: if dataset.status != "ready" or dataset.source_name not in definition.materialized_source_names: continue @@ -484,13 +486,21 @@ class CoverageRegistryService: acquired_scope = box(*(float(value) for value in acquired_bbox)) except (TypeError, ValueError): continue - if not acquired_scope.is_valid or not acquired_scope.covers(selection): + if not acquired_scope.is_valid or not acquired_scope.intersects(selection): continue + bounded_scopes.append(acquired_scope) + elif definition.contract.acquisition_mode == "bounded_api" and coverage_zones: + zone_scoped_materialization = zone in coverage_zones or "belgium" in coverage_zones layer_matches = not layer_names or dataset.reference_layer_name in layer_names zone_matches = not coverage_zones or zone in coverage_zones or "belgium" in coverage_zones if layer_matches and zone_matches: matches.append(dataset) - return matches + if not matches: + return [], False + fully_covered = True + if definition.contract.acquisition_mode == "bounded_api": + fully_covered = zone_scoped_materialization or (bool(bounded_scopes) and unary_union(bounded_scopes).covers(selection)) + return matches, fully_covered @staticmethod def _resolve_item( @@ -516,10 +526,11 @@ class CoverageRegistryService: ) materialized: list[Dataset] = [] + evidence: list[dict[str, Any]] = [] source_statuses: list[str] = [] limitations: list[str] = [] for definition in definitions: - matches = CoverageRegistryService._matching_datasets( + matches, fully_covered = CoverageRegistryService._matching_datasets( datasets, definition, theme, @@ -527,8 +538,28 @@ class CoverageRegistryService: selection, ) materialized.extend(matches) - if matches: + for dataset in matches: + metadata = dataset.source_metadata if isinstance(getattr(dataset, "source_metadata", None), dict) else {} + observed_at = getattr(dataset, "observed_at", None) + published_at = metadata.get("published_at") or metadata.get("publication_date") or metadata.get("published_on") + evidence.append({ + "dataset_id": dataset.id, + "source_name": str(dataset.source_name or definition.contract.source_name), + "authority_level": definition.contract.authority_level, + "source_version": getattr(dataset, "source_version", None), + "observed_at": observed_at.isoformat() if hasattr(observed_at, "isoformat") else (str(observed_at) if observed_at else None), + "published_at": str(published_at) if published_at else None, + "crs": getattr(dataset, "crs", None) or metadata.get("source_crs"), + "resolution": getattr(dataset, "resolution_json", None), + "coverage_bbox_epsg4326": metadata.get("bbox_epsg4326"), + "attribution": metadata.get("attribution") or definition.contract.attribution, + "license_note": metadata.get("license_note") or definition.contract.license_note, + "checksum_sha256": getattr(dataset, "checksum_sha256", None), + }) + if matches and fully_covered: source_statuses.append("operational") + elif matches: + source_statuses.append("partial") elif ( definition.contract.integration_status == "operational" and (not definition.operational_themes or theme in definition.operational_themes) @@ -549,6 +580,7 @@ class CoverageRegistryService: status=best_status, source_names=[definition.contract.source_name for definition in definitions], materialized_dataset_ids=list(dict.fromkeys(dataset.id for dataset in materialized)), + evidence=list({str(item["dataset_id"]): item for item in evidence}.values()), limitation_message=" ".join(dict.fromkeys(limitations)), ) diff --git a/backend/app/services/detection_service.py b/backend/app/services/detection_service.py index 6d915011..e4b9a22a 100644 --- a/backend/app/services/detection_service.py +++ b/backend/app/services/detection_service.py @@ -15,7 +15,7 @@ from sqlalchemy import func from app.core.config import Settings, get_settings from app.core.errors import AppError from app.core.request_context import get_request_id -from app.models import AnalysisRun, Dataset, Detection, Job, Project, VectorFeature +from app.models import AnalysisRun, Area, Dataset, Detection, Job, Project, VectorFeature from app.schemas.detection import DetectionListResponse, DetectionRead, DetectionRunListResponse, DetectionRunRead, DetectionRunResponse from app.services.detection_georeferencing import pixel_bbox_to_epsg4326_polygon from app.services.detection_qa_service import DetectionQaService @@ -90,6 +90,12 @@ class DetectionService: message="Configured YOLO inference requires an existing raster tile manifest path", status_code=400, ) + requested_classes = {DetectionService._canonical_class_name(value) for value in (class_filter or [])} + unsupported_classes = sorted(requested_classes - set(model.supported_classes)) + if unsupported_classes: + raise AppError(code="DETECTION_CLASS_NOT_VALIDATED", message="The selected model is not validated for one or more requested classes", details={"unsupported_classes": unsupported_classes, "supported_classes": model.supported_classes}, status_code=422) + if model.model_id == resolved_settings.yolo_model_id and resolved_settings.yolo_enforce_validation_scope: + DetectionService._validate_model_area_scope(db, dataset, resolved_settings) run_parameters = { "model_id": model.model_id, @@ -214,6 +220,18 @@ class DetectionService: raise AppError(code="DETECTION_MODEL_UNAVAILABLE", message="Detection model is unavailable", status_code=503) @staticmethod + def _validate_model_area_scope(db, dataset: Dataset, settings: Settings) -> None: + allowed_names = [value.strip().casefold() for value in settings.yolo_validated_area_names.split(",") if value.strip()] + area = db.get(Area, dataset.area_id) if dataset.area_id else None + area_name = area.name.strip() if area is not None else "" + if not area_name or not any(token in area_name.casefold() for token in allowed_names): + raise AppError( + code="DETECTION_VALIDATION_SCOPE_UNAVAILABLE", + message="Configured YOLO inference is not validated for this Dataset area.", + details={"dataset_id": str(dataset.id), "dataset_area": area_name or None, "validated_area_names": allowed_names}, + status_code=422, + ) + @staticmethod def _fail_run_after_exception(db, analysis_run: AnalysisRun, job: Job, exc: Exception, fallback_code: str) -> None: try: db.rollback() diff --git a/backend/app/services/model_registry_service.py b/backend/app/services/model_registry_service.py index d4b7071b..b4db00a5 100644 --- a/backend/app/services/model_registry_service.py +++ b/backend/app/services/model_registry_service.py @@ -10,6 +10,7 @@ from app.services.segmentation_adapter import ( YoloSegmentationAdapter, ) from app.services.yolo_adapter import YoloDetectionAdapter +from app.core.errors import AppError class ModelRegistryService: @@ -224,16 +225,24 @@ class ModelRegistryService: elif not model_path.exists() or not model_path.is_file(): limitation = "YOLO_MODEL_PATH does not point to an existing local model file. GeoIntel will not download model weights automatically." else: - configured = True - status = "configured" - limitation = "Configured for local YOLO inference over an existing raster tile manifest." + try: + validate_runtime = getattr(yolo_adapter_class, "validate_runtime", None) + if validate_runtime is not None: + yolo_adapter_class(settings).validate_runtime() + except AppError as exc: + status = "accelerator_unavailable" + limitation = exc.message + else: + configured = True + status = "configured" + limitation = "Configured for local YOLO inference over an existing raster tile manifest within its validated area scope." return DetectionModelCapability( model_id=settings.yolo_model_id, display_name=settings.yolo_model_display_name, framework="ultralytics/pytorch", task_type="object_detection", - supported_classes=["building", "road", "water", "landuse"], + supported_classes=[value.strip().lower() for value in settings.yolo_model_classes.split(",") if value.strip()], configured=configured, status=status, limitation_message=limitation, diff --git a/backend/app/services/runtime_reconciliation_service.py b/backend/app/services/runtime_reconciliation_service.py index 8658ca1d..4cf1507c 100644 --- a/backend/app/services/runtime_reconciliation_service.py +++ b/backend/app/services/runtime_reconciliation_service.py @@ -5,13 +5,15 @@ from datetime import datetime, timezone from sqlalchemy.orm import Session -from app.models import AnalysisRun, Job +from app.models import AoiOperation, AoiOperationPartition, AnalysisRun, Job @dataclass(frozen=True) class ReconciliationResult: interrupted_jobs: int interrupted_analysis_runs: int + resumed_aoi_partitions: int + exhausted_aoi_partitions: int class RuntimeReconciliationService: @@ -51,8 +53,43 @@ class RuntimeReconciliationService: synchronize_session=False, ) ) + resumed_aoi_partitions = ( + db.query(AoiOperationPartition) + .filter( + AoiOperationPartition.status == "running", + AoiOperationPartition.attempt_count < AoiOperationPartition.max_attempts, + ) + .update( + { + AoiOperationPartition.status: "queued", + AoiOperationPartition.error_message: RuntimeReconciliationService.ERROR_MESSAGE, + AoiOperationPartition.started_at: None, + }, + synchronize_session=False, + ) + ) + exhausted_aoi_partitions = ( + db.query(AoiOperationPartition) + .filter( + AoiOperationPartition.status == "running", + AoiOperationPartition.attempt_count >= AoiOperationPartition.max_attempts, + ) + .update( + { + AoiOperationPartition.status: "failed", + AoiOperationPartition.finished_at: resolved_finished_at, + AoiOperationPartition.error_message: RuntimeReconciliationService.ERROR_MESSAGE, + }, + synchronize_session=False, + ) + ) + db.query(AoiOperation).filter(AoiOperation.status == "running").update( + {AoiOperation.status: "queued"}, synchronize_session=False + ) db.commit() return ReconciliationResult( interrupted_jobs=interrupted_jobs, interrupted_analysis_runs=interrupted_analysis_runs, + resumed_aoi_partitions=resumed_aoi_partitions, + exhausted_aoi_partitions=exhausted_aoi_partitions, ) diff --git a/backend/app/services/yolo_adapter.py b/backend/app/services/yolo_adapter.py index 14c4401f..45f9828f 100644 --- a/backend/app/services/yolo_adapter.py +++ b/backend/app/services/yolo_adapter.py @@ -38,6 +38,8 @@ class YoloDetectionAdapter: status_code=503, ) + self.validate_runtime() + try: from ultralytics import YOLO except ImportError as exc: @@ -57,6 +59,32 @@ class YoloDetectionAdapter: status_code=503, ) from exc + def validate_runtime(self) -> None: + if not self.settings.yolo_require_cuda: + return + try: + import torch + except Exception as exc: + raise AppError( + code="DETECTION_ACCELERATOR_UNAVAILABLE", + message="NVIDIA CUDA is required for configured YOLO inference, but PyTorch is not importable.", + status_code=503, + ) from exc + if not torch.cuda.is_available(): + raise AppError( + code="DETECTION_ACCELERATOR_UNAVAILABLE", + message="NVIDIA CUDA is required for configured YOLO inference, but no CUDA device is available.", + details={"configured_device": self.settings.yolo_device}, + status_code=503, + ) + if not str(self.settings.yolo_device).lower().startswith(("cuda", "0", "1", "2", "3")): + raise AppError( + code="DETECTION_ACCELERATOR_MISCONFIGURED", + message="NVIDIA CUDA is required, but YOLO_DEVICE does not select a CUDA device.", + details={"configured_device": self.settings.yolo_device}, + status_code=503, + ) + def predict_tile(self, model, tile_path: Path, confidence_threshold: float) -> list[dict[str, Any]]: if not tile_path.exists() or not tile_path.is_file(): raise AppError( diff --git a/backend/app/services/yolo_preflight_service.py b/backend/app/services/yolo_preflight_service.py index 8745ad24..b8a3bd47 100644 --- a/backend/app/services/yolo_preflight_service.py +++ b/backend/app/services/yolo_preflight_service.py @@ -38,6 +38,7 @@ class YoloPreflightService: "checks": { "enabled": resolved_settings.yolo_enabled, "dependencies_available": None, + "accelerator_ready": None, "model_path_set": None, "model_file_exists": None, "model_load_requested": check_model_load, @@ -70,6 +71,21 @@ class YoloPreflightService: result["message"] = "YOLO dependencies are not installed. Install backend optional extras with geointel-backend[ai]." return result + if not assume_dependencies: + try: + adapter = yolo_adapter_class(resolved_settings) + validate_runtime = getattr(adapter, "validate_runtime", None) + if validate_runtime is not None: + validate_runtime() + except AppError as exc: + result["checks"]["accelerator_ready"] = False + result["status"] = "accelerator_unavailable" + result["message"] = exc.message + result["error_code"] = exc.code + result["details"] = exc.details + return result + result["checks"]["accelerator_ready"] = True + result["checks"]["model_path_set"] = bool(resolved_settings.yolo_model_path) if not resolved_settings.yolo_model_path: result["message"] = "YOLO_MODEL_PATH is not set. GeoIntel will not download model weights automatically." @@ -143,6 +159,8 @@ class YoloPreflightService: "torch_version": YoloPreflightService._package_version("torch"), "ultralytics_version": YoloPreflightService._package_version("ultralytics"), "cuda_available": None, + "configured_device": settings.yolo_device, + "cuda_required": settings.yolo_require_cuda, } @staticmethod diff --git a/backend/tests/test_docker_runtime_config.py b/backend/tests/test_docker_runtime_config.py index ac7a4155..acc8c66e 100644 --- a/backend/tests/test_docker_runtime_config.py +++ b/backend/tests/test_docker_runtime_config.py @@ -398,7 +398,7 @@ def test_all_in_one_dockerfile_caches_dependencies_and_uses_cpu_torch_for_ai_run smoke_index = dockerfile.index("RUN python scripts/gis_import_smoke.py") assert metadata_copy_index < placeholder_readme_index < dependency_install_index < backend_copy_index < smoke_index - assert "GEOINTEL_TORCH_INDEX_URL=https://download.pytorch.org/whl/cpu" in dockerfile + assert "GEOINTEL_TORCH_INDEX_URL=https://download.pytorch.org/whl/cu130" in dockerfile assert "GEOINTEL_TORCH_VERSION=2.13.0" in dockerfile assert "GEOINTEL_TORCHVISION_VERSION=0.28.0" in dockerfile assert '--index-url "$GEOINTEL_TORCH_INDEX_URL"' in dockerfile @@ -424,3 +424,12 @@ def test_unraid_deploy_passes_ai_build_arg_and_yolo_runtime_env() -> None: assert '-e YOLO_MAX_DETECTIONS="$YOLO_MAX_DETECTIONS"' in run_script assert '-e YOLO_DUPLICATE_IOU_THRESHOLD="$YOLO_DUPLICATE_IOU_THRESHOLD"' in run_script assert "-v \"${GEOINTEL_MODELS_PATH}:/app/models\"" in run_script +def test_unraid_ai_runtime_requests_nvidia_and_fails_closed() -> None: + run_script = (ROOT / "deploy" / "unraid" / "run-dockerman-container.sh").read_text(encoding="utf-8") + dockerfile = (ROOT / "deploy" / "unraid" / "Dockerfile.all-in-one").read_text(encoding="utf-8") + + assert "--gpus all" in run_script + assert 'YOLO_DEVICE="${YOLO_DEVICE:-cuda:0}"' in run_script + assert 'YOLO_REQUIRE_CUDA="${YOLO_REQUIRE_CUDA:-true}"' in run_script + assert '-e YOLO_REQUIRE_CUDA="$YOLO_REQUIRE_CUDA"' in run_script + assert "https://download.pytorch.org/whl/cu130" in dockerfile diff --git a/backend/tests/test_rc4_national_coverage.py b/backend/tests/test_rc4_national_coverage.py index 9701bbd6..d3013d9b 100644 --- a/backend/tests/test_rc4_national_coverage.py +++ b/backend/tests/test_rc4_national_coverage.py @@ -225,6 +225,26 @@ def test_bounded_api_materialization_only_covers_its_persisted_bbox() -> None: assert outside.items[0].materialized_dataset_ids == [] +def test_bounded_partition_union_can_be_operational() -> None: + project_id = uuid4() + left_id = uuid4() + right_id = uuid4() + datasets = [ + SimpleNamespace(id=left_id, status="ready", source_name="spw_picc", reference_layer_name="buildings", source_metadata={"coverage_zones": ["wallonia"], "bbox_epsg4326": [4.50, 50.50, 4.60, 50.60]}), + SimpleNamespace(id=right_id, status="ready", source_name="spw_picc", reference_layer_name="buildings", source_metadata={"coverage_zones": ["wallonia"], "bbox_epsg4326": [4.60, 50.50, 4.70, 50.60]}), + ] + session = FakeSession( + project=SimpleNamespace(id=project_id), + areas=[scope_area("Belgium land", box(2.5, 49.5, 6.4, 51.5)), scope_area("Wallonia", box(2.5, 49.5, 6.4, 50.8))], + datasets=datasets, + ) + + result = CoverageRegistryService.resolve(session, project_id, CoverageBBox(minx=4.51, miny=50.51, maxx=4.69, maxy=50.59), ["buildings"]) + + assert result.items[0].status == "operational" + assert result.items[0].materialized_dataset_ids == [left_id, right_id] + + def test_spw_bathymetry_materialization_is_source_specific() -> None: project_id = uuid4() scope = [ diff --git a/backend/tests/test_runtime_reconciliation_service.py b/backend/tests/test_runtime_reconciliation_service.py index 031ff75c..66e1715c 100644 --- a/backend/tests/test_runtime_reconciliation_service.py +++ b/backend/tests/test_runtime_reconciliation_service.py @@ -3,7 +3,7 @@ from __future__ import annotations from datetime import datetime, timezone from unittest.mock import MagicMock -from app.models import AnalysisRun, Job +from app.models import AoiOperation, AoiOperationPartition, AnalysisRun, Job from app.services.runtime_reconciliation_service import RuntimeReconciliationService @@ -11,9 +11,15 @@ def test_reconciliation_terminalizes_only_running_work() -> None: db = MagicMock() jobs = MagicMock() runs = MagicMock() + resumable_partitions = MagicMock() + exhausted_partitions = MagicMock() + operations = MagicMock() jobs.filter.return_value.update.return_value = 5 runs.filter.return_value.update.return_value = 2 - db.query.side_effect = [jobs, runs] + resumable_partitions.filter.return_value.update.return_value = 3 + exhausted_partitions.filter.return_value.update.return_value = 1 + operations.filter.return_value.update.return_value = 2 + db.query.side_effect = [jobs, runs, resumable_partitions, exhausted_partitions, operations] finished_at = datetime(2026, 7, 17, 20, 0, tzinfo=timezone.utc) result = RuntimeReconciliationService.reconcile( @@ -23,8 +29,13 @@ def test_reconciliation_terminalizes_only_running_work() -> None: assert result.interrupted_jobs == 5 assert result.interrupted_analysis_runs == 2 + assert result.resumed_aoi_partitions == 3 + assert result.exhausted_aoi_partitions == 1 jobs.filter.assert_called_once() runs.filter.assert_called_once() + resumable_partitions.filter.assert_called_once() + exhausted_partitions.filter.assert_called_once() + operations.filter.assert_called_once() job_values = jobs.filter.return_value.update.call_args.args[0] run_values = runs.filter.return_value.update.call_args.args[0] assert job_values[Job.status] == "failed" @@ -33,6 +44,12 @@ def test_reconciliation_terminalizes_only_running_work() -> None: assert run_values[AnalysisRun.status] == "failed" assert run_values[AnalysisRun.finished_at] == finished_at assert "PROCESS_INTERRUPTED" in run_values[AnalysisRun.error_message] + resumed_values = resumable_partitions.filter.return_value.update.call_args.args[0] + exhausted_values = exhausted_partitions.filter.return_value.update.call_args.args[0] + operation_values = operations.filter.return_value.update.call_args.args[0] + assert resumed_values[AoiOperationPartition.status] == "queued" + assert exhausted_values[AoiOperationPartition.status] == "failed" + assert operation_values[AoiOperation.status] == "queued" db.commit.assert_called_once_with() diff --git a/backend/tests/test_selection_partition_analysis.py b/backend/tests/test_selection_partition_analysis.py index b616cd53..200e6ac8 100644 --- a/backend/tests/test_selection_partition_analysis.py +++ b/backend/tests/test_selection_partition_analysis.py @@ -83,7 +83,7 @@ def test_vector_partition_route_combines_one_governed_product(monkeypatch) -> No def test_vector_partition_request_has_a_bounded_fan_out() -> None: with pytest.raises(ValidationError): VectorPartitionSelectionRequest( - dataset_ids=[uuid4() for _ in range(17)], + dataset_ids=[uuid4() for _ in range(4097)], bbox={"min_x": 5.0, "min_y": 51.0, "max_x": 5.3, "max_y": 51.2}, ) diff --git a/backend/tests/test_sprint242_aoi_orchestration.py b/backend/tests/test_sprint242_aoi_orchestration.py new file mode 100644 index 00000000..73a5e092 --- /dev/null +++ b/backend/tests/test_sprint242_aoi_orchestration.py @@ -0,0 +1,86 @@ +from __future__ import annotations + +from pathlib import Path + +import pytest +from shapely.geometry import box + +from app.core.errors import AppError +from app.services.aoi_operation_executor import AoiOperationExecutor +from app.services.aoi_operation_service import AoiOperationService + + +ROOT = Path(__file__).resolve().parents[2] + + +def test_partition_plan_covers_aoi_without_overlapping_area() -> None: + aoi = box(0, 0, 25_000, 18_000) + partitions = AoiOperationService._partition(aoi, 10_000) + + assert len(partitions) == 6 + assert sum(partition.area for partition in partitions) == pytest.approx(aoi.area) + assert all(partition.within(aoi) for partition in partitions) + for index, partition in enumerate(partitions): + for other in partitions[index + 1 :]: + assert partition.intersection(other).area == pytest.approx(0.0) + + +def test_partition_plan_intersects_irregular_aoi_exactly() -> None: + aoi = box(0, 0, 20_000, 20_000).difference(box(5_000, 5_000, 15_000, 15_000)) + partitions = AoiOperationService._partition(aoi, 8_000) + + assert sum(partition.area for partition in partitions) == pytest.approx(aoi.area) + assert all(not partition.intersects(box(5_001, 5_001, 14_999, 14_999)) for partition in partitions) + + +def test_partition_plan_fails_before_unbounded_fanout(monkeypatch) -> None: + monkeypatch.setattr(AoiOperationService, "MAX_PARTITIONS", 4) + + with pytest.raises(AppError) as exc_info: + AoiOperationService._partition(box(0, 0, 30_000, 30_000), 10_000) + + assert exc_info.value.code == "AOI_PARTITION_LIMIT_EXCEEDED" + assert exc_info.value.details["candidate_count"] == 9 + + +def test_executor_retries_only_transient_provider_failures() -> None: + assert AoiOperationExecutor._retryable(AppError(code="UPSTREAM_UNAVAILABLE", message="down", status_code=503)) is True + assert AoiOperationExecutor._retryable(AppError(code="INVALID_SCOPE", message="bad", status_code=422)) is False + + +def test_provider_budget_is_automatic_and_override_can_only_be_stricter() -> None: + governed = AoiOperationService._partition_side("grb", None) + assert governed > 0 + assert AoiOperationService._partition_side("grb", governed * 2) == governed + assert AoiOperationService._partition_side("grb", governed / 2) == governed / 2 + + +@pytest.mark.parametrize( + ("provider_key", "max_pixels", "resolution_m"), + [ + ("dhmv", 12_000_000, 5.0), + ("flood_hazard", 12_000_000, 5.0), + ("spw_terrain", 12_000_000, 5.0), + ("thematic_raster", 30_000_000, 10.0), + ("walous", 36_000_000, 10.0), + ], +) +def test_raster_provider_budget_never_exceeds_decoded_pixel_limit( + provider_key: str, max_pixels: int, resolution_m: float +) -> None: + side_m = AoiOperationService._partition_side(provider_key, None) + + assert (side_m / resolution_m) ** 2 < max_pixels + + +def test_migration_and_api_are_registered() -> None: + migration = (ROOT / "backend/alembic/versions/202607260001_aoi_operations.py").read_text(encoding="utf-8") + main = (ROOT / "backend/app/main.py").read_text(encoding="utf-8") + route = (ROOT / "backend/app/api/routes/aoi_operations.py").read_text(encoding="utf-8") + + assert 'down_revision = "202607160001"' in migration + assert '"aoi_operations"' in migration + assert '"aoi_operation_partitions"' in migration + assert "app.include_router(aoi_operations.router" in main + assert '"/{operation_id}/execute-next"' in route + assert '"/{operation_id}/partitions/{partition_id}/checkpoint"' in route diff --git a/backend/tests/test_sprint31_unraid_template.py b/backend/tests/test_sprint31_unraid_template.py index 8e501ba1..0deb701e 100644 --- a/backend/tests/test_sprint31_unraid_template.py +++ b/backend/tests/test_sprint31_unraid_template.py @@ -167,7 +167,7 @@ def test_powershell_tower_deploy_streams_remote_script_to_bash() -> None: powershell = (ROOT / "scripts" / "deploy_tower.ps1").read_text(encoding="utf-8") assert "[System.Text.UTF8Encoding]::new($false)" in powershell - assert "[System.IO.File]::WriteAllText($localScriptPath, $remoteScript, $utf8NoBom)" in powershell + assert "[System.IO.File]::WriteAllText($localScriptPath, $remoteScriptLf, $utf8NoBom)" in powershell assert "& scp @scpArgs" in powershell assert "& ssh @sshRunArgs" in powershell assert "bash '$remoteScriptPath'" in powershell @@ -192,8 +192,10 @@ def test_frontend_and_unraid_icon_assets_are_present() -> None: index = (ROOT / "frontend" / "index.html").read_text(encoding="utf-8") assert "GeoIntel" in deploy_icon + assert "Een geometrische G als geografische lens" in deploy_icon + assert "Een geometrische G als geografische lens" in frontend_icon + assert deploy_png.read_bytes() == frontend_png.read_bytes() assert deploy_png.read_bytes() == frontend_png.read_bytes() assert deploy_png.stat().st_size > 1000 assert '' in index diff --git a/backend/tests/test_sprint8b_yolo_foundation.py b/backend/tests/test_sprint8b_yolo_foundation.py index bf447f6d..a1d8bb84 100644 --- a/backend/tests/test_sprint8b_yolo_foundation.py +++ b/backend/tests/test_sprint8b_yolo_foundation.py @@ -2,13 +2,15 @@ from __future__ import annotations import json from pathlib import Path +import sys +from types import SimpleNamespace from uuid import uuid4 import pytest from app.core.config import Settings from app.core.errors import AppError -from app.models import AnalysisRun, Dataset, Detection, Job, Project +from app.models import AnalysisRun, Area, Dataset, Detection, Job, Project from app.services.detection_georeferencing import pixel_bbox_to_epsg4326_polygon from app.services.detection_service import DetectionService from app.services.model_registry_service import ModelRegistryService @@ -235,6 +237,7 @@ def test_yolo_configured_model_reports_configured_with_local_model_and_dependenc assert model.nationally_validated is False assert model.operator_review_required is True assert model.validated_regions == ["flanders_mol_kempen"] + assert model.supported_classes == ["building"] assert "Mol and the Kempen" in (model.validation_scope or "") @@ -246,6 +249,45 @@ def test_yolo_dependency_check_uses_real_imports_not_find_spec() -> None: assert "import torch" in source +def test_yolo_runtime_fails_closed_when_cuda_is_required_but_unavailable(tmp_path: Path, monkeypatch) -> None: + settings = _settings(tmp_path, yolo_device="cuda:0", yolo_require_cuda=True) + monkeypatch.setitem(sys.modules, "torch", SimpleNamespace(cuda=SimpleNamespace(is_available=lambda: False))) + + with pytest.raises(AppError) as exc_info: + YoloDetectionAdapter(settings).validate_runtime() + + assert exc_info.value.code == "DETECTION_ACCELERATOR_UNAVAILABLE" + + +def test_yolo_runtime_rejects_cpu_device_when_cuda_is_required(tmp_path: Path, monkeypatch) -> None: + settings = _settings(tmp_path, yolo_device="cpu", yolo_require_cuda=True) + monkeypatch.setitem(sys.modules, "torch", SimpleNamespace(cuda=SimpleNamespace(is_available=lambda: True))) + + with pytest.raises(AppError) as exc_info: + YoloDetectionAdapter(settings).validate_runtime() + + assert exc_info.value.code == "DETECTION_ACCELERATOR_MISCONFIGURED" + + +def test_yolo_validation_scope_requires_persisted_validated_area(tmp_path: Path) -> None: + dataset = Dataset(id=uuid4(), project_id=uuid4(), name="image.tif", dataset_type="raster", source="test", area_id=uuid4()) + wrong_area = Area(id=dataset.area_id, project_id=dataset.project_id, name="Brussels", geometry="MULTIPOLYGON EMPTY") + db = FakeSession(objects={(Area, dataset.area_id): wrong_area}) + + with pytest.raises(AppError) as exc_info: + DetectionService._validate_model_area_scope(db, dataset, _settings(tmp_path, yolo_validated_area_names="Mol,Kempen")) + + assert exc_info.value.code == "DETECTION_VALIDATION_SCOPE_UNAVAILABLE" + + +def test_yolo_validation_scope_accepts_bound_mol_area(tmp_path: Path) -> None: + dataset = Dataset(id=uuid4(), project_id=uuid4(), name="image.tif", dataset_type="raster", source="test", area_id=uuid4()) + area = Area(id=dataset.area_id, project_id=dataset.project_id, name="Gemeente Mol", geometry="MULTIPOLYGON EMPTY") + db = FakeSession(objects={(Area, dataset.area_id): area}) + + DetectionService._validate_model_area_scope(db, dataset, _settings(tmp_path, yolo_validated_area_names="Mol,Kempen")) + + def test_yolo_run_requires_tile_manifest_path(tmp_path: Path) -> None: db, project_id, dataset_id = _project_and_dataset() settings = _settings(tmp_path) diff --git a/deploy/unraid/Dockerfile.all-in-one b/deploy/unraid/Dockerfile.all-in-one index 207bcf2f..49307055 100644 --- a/deploy/unraid/Dockerfile.all-in-one +++ b/deploy/unraid/Dockerfile.all-in-one @@ -9,7 +9,7 @@ RUN npm run build FROM postgres:16-bookworm AS runtime ARG GEOINTEL_INSTALL_AI=false -ARG GEOINTEL_TORCH_INDEX_URL=https://download.pytorch.org/whl/cpu +ARG GEOINTEL_TORCH_INDEX_URL=https://download.pytorch.org/whl/cu130 ARG GEOINTEL_TORCH_VERSION=2.13.0 ARG GEOINTEL_TORCHVISION_VERSION=0.28.0 ARG GEOINTEL_ULTRALYTICS_VERSION=8.4.99 diff --git a/deploy/unraid/geointel-unraid-template.xml b/deploy/unraid/geointel-unraid-template.xml index 4a73390b..ec74a820 100644 --- a/deploy/unraid/geointel-unraid-template.xml +++ b/deploy/unraid/geointel-unraid-template.xml @@ -120,6 +120,8 @@ 5 20000 12000000 + true + 2 false /app/models @@ -127,7 +129,11 @@ Configured YOLO detector /app/storage/ultralytics - cpu + cuda:0 + true + building + true + Mol,Kempen 640 100 1000 diff --git a/deploy/unraid/geointel.env.example b/deploy/unraid/geointel.env.example index 95028eaf..c3cfaeef 100644 --- a/deploy/unraid/geointel.env.example +++ b/deploy/unraid/geointel.env.example @@ -28,6 +28,8 @@ GEOINTEL_CORS_ORIGINS=http://localhost:1202,http://127.0.0.1:1202,http://192.168 # Upload guard in MiB. The same 1-2048 limit is applied by nginx and FastAPI. GEOINTEL_MAX_UPLOAD_MB=500 +GEOINTEL_AOI_WORKER_ENABLED=true +GEOINTEL_AOI_WORKER_POLL_SECONDS=2 # Optional single-operator access gate. Never store a plaintext password here. # Generate the password hash with AuthService.hash_password and use a unique, @@ -143,7 +145,11 @@ YOLO_MODEL_ID=yolo-configured YOLO_MODEL_DISPLAY_NAME=Configured YOLO detector YOLO_MODEL_VERSION= YOLO_CONFIG_DIR=/app/storage/ultralytics -YOLO_DEVICE=cpu +YOLO_DEVICE=cuda:0 +YOLO_REQUIRE_CUDA=true +YOLO_MODEL_CLASSES=building +YOLO_ENFORCE_VALIDATION_SCOPE=true +YOLO_VALIDATED_AREA_NAMES=Mol,Kempen YOLO_IMAGE_SIZE=640 YOLO_MAX_TILES=100 YOLO_MAX_DETECTIONS=1000 diff --git a/deploy/unraid/run-dockerman-container.sh b/deploy/unraid/run-dockerman-container.sh index 415102ef..8bf45e3c 100644 --- a/deploy/unraid/run-dockerman-container.sh +++ b/deploy/unraid/run-dockerman-container.sh @@ -33,6 +33,8 @@ GEOINTEL_POSTGRES_USER="${GEOINTEL_POSTGRES_USER:-geointel}" GEOINTEL_POSTGRES_PASSWORD="${GEOINTEL_POSTGRES_PASSWORD:-}" GEOINTEL_CORS_ORIGINS="${GEOINTEL_CORS_ORIGINS:-http://localhost:${GEOINTEL_FRONTEND_PORT},http://127.0.0.1:${GEOINTEL_FRONTEND_PORT},http://192.168.10.150:${GEOINTEL_FRONTEND_PORT}}" GEOINTEL_MAX_UPLOAD_MB="${GEOINTEL_MAX_UPLOAD_MB:-500}" +GEOINTEL_AOI_WORKER_ENABLED="${GEOINTEL_AOI_WORKER_ENABLED:-true}" +GEOINTEL_AOI_WORKER_POLL_SECONDS="${GEOINTEL_AOI_WORKER_POLL_SECONDS:-2}" GEOINTEL_AUTH_ENABLED="${GEOINTEL_AUTH_ENABLED:-false}" GEOINTEL_AUTH_USERNAME="${GEOINTEL_AUTH_USERNAME:-}" GEOINTEL_AUTH_PASSWORD_HASH="${GEOINTEL_AUTH_PASSWORD_HASH:-}" @@ -127,8 +129,12 @@ YOLO_MODEL_PATH="${YOLO_MODEL_PATH:-}" YOLO_MODEL_ID="${YOLO_MODEL_ID:-yolo-configured}" YOLO_MODEL_DISPLAY_NAME="${YOLO_MODEL_DISPLAY_NAME:-Configured YOLO detector}" YOLO_MODEL_VERSION="${YOLO_MODEL_VERSION:-}" +YOLO_MODEL_CLASSES="${YOLO_MODEL_CLASSES:-building}" +YOLO_ENFORCE_VALIDATION_SCOPE="${YOLO_ENFORCE_VALIDATION_SCOPE:-true}" +YOLO_VALIDATED_AREA_NAMES="${YOLO_VALIDATED_AREA_NAMES:-Mol,Kempen}" YOLO_CONFIG_DIR="${YOLO_CONFIG_DIR:-/app/storage/ultralytics}" -YOLO_DEVICE="${YOLO_DEVICE:-cpu}" +YOLO_DEVICE="${YOLO_DEVICE:-cuda:0}" +YOLO_REQUIRE_CUDA="${YOLO_REQUIRE_CUDA:-true}" YOLO_IMAGE_SIZE="${YOLO_IMAGE_SIZE:-640}" YOLO_MAX_TILES="${YOLO_MAX_TILES:-100}" YOLO_MAX_DETECTIONS="${YOLO_MAX_DETECTIONS:-1000}" @@ -257,6 +263,7 @@ migrate_compose_volume_if_needed docker run -d \ --name geointel \ + --gpus all \ --restart unless-stopped \ --label net.unraid.docker.managed=dockerman \ --label 'net.unraid.docker.webui=http://[IP]:[PORT:80]/' \ @@ -269,6 +276,8 @@ docker run -d \ -e GEOINTEL_STORAGE_ROOT=/app/storage \ -e GEOINTEL_CORS_ORIGINS="$GEOINTEL_CORS_ORIGINS" \ -e GEOINTEL_MAX_UPLOAD_MB="$GEOINTEL_MAX_UPLOAD_MB" \ + -e GEOINTEL_AOI_WORKER_ENABLED="$GEOINTEL_AOI_WORKER_ENABLED" \ + -e GEOINTEL_AOI_WORKER_POLL_SECONDS="$GEOINTEL_AOI_WORKER_POLL_SECONDS" \ -e GEOINTEL_AUTH_ENABLED="$GEOINTEL_AUTH_ENABLED" \ -e GEOINTEL_AUTH_USERNAME="$GEOINTEL_AUTH_USERNAME" \ -e GEOINTEL_AUTH_PASSWORD_HASH="$GEOINTEL_AUTH_PASSWORD_HASH" \ @@ -363,8 +372,12 @@ docker run -d \ -e YOLO_MODEL_ID="$YOLO_MODEL_ID" \ -e YOLO_MODEL_DISPLAY_NAME="$YOLO_MODEL_DISPLAY_NAME" \ -e YOLO_MODEL_VERSION="$YOLO_MODEL_VERSION" \ + -e YOLO_MODEL_CLASSES="$YOLO_MODEL_CLASSES" \ + -e YOLO_ENFORCE_VALIDATION_SCOPE="$YOLO_ENFORCE_VALIDATION_SCOPE" \ + -e YOLO_VALIDATED_AREA_NAMES="$YOLO_VALIDATED_AREA_NAMES" \ -e YOLO_CONFIG_DIR="$YOLO_CONFIG_DIR" \ -e YOLO_DEVICE="$YOLO_DEVICE" \ + -e YOLO_REQUIRE_CUDA="$YOLO_REQUIRE_CUDA" \ -e YOLO_IMAGE_SIZE="$YOLO_IMAGE_SIZE" \ -e YOLO_MAX_TILES="$YOLO_MAX_TILES" \ -e YOLO_MAX_DETECTIONS="$YOLO_MAX_DETECTIONS" \ diff --git a/docker-compose.unraid.yml b/docker-compose.unraid.yml index 4f4d3c02..32182619 100644 --- a/docker-compose.unraid.yml +++ b/docker-compose.unraid.yml @@ -5,6 +5,7 @@ services: dockerfile: deploy/unraid/Dockerfile.all-in-one args: GEOINTEL_INSTALL_AI: ${GEOINTEL_INSTALL_AI:-false} + GEOINTEL_TORCH_INDEX_URL: ${GEOINTEL_TORCH_INDEX_URL:-https://download.pytorch.org/whl/cu130} image: geointel-all-in-one:latest container_name: geointel labels: @@ -16,6 +17,8 @@ services: GEOINTEL_POSTGRES_USER: ${GEOINTEL_POSTGRES_USER:-geointel} GEOINTEL_POSTGRES_PASSWORD: ${GEOINTEL_POSTGRES_PASSWORD:-geointel} GEOINTEL_STORAGE_ROOT: /app/storage + GEOINTEL_AOI_WORKER_ENABLED: ${GEOINTEL_AOI_WORKER_ENABLED:-true} + GEOINTEL_AOI_WORKER_POLL_SECONDS: ${GEOINTEL_AOI_WORKER_POLL_SECONDS:-2} GEOINTEL_CORS_ORIGINS: ${GEOINTEL_CORS_ORIGINS:-http://localhost:1202,http://127.0.0.1:1202} GEOINTEL_MAX_UPLOAD_MB: ${GEOINTEL_MAX_UPLOAD_MB:-500} ORTHOPHOTO_ENABLED: ${ORTHOPHOTO_ENABLED:-true} @@ -99,7 +102,11 @@ services: YOLO_MODEL_DISPLAY_NAME: ${YOLO_MODEL_DISPLAY_NAME:-Configured YOLO detector} YOLO_MODEL_VERSION: ${YOLO_MODEL_VERSION:-} YOLO_CONFIG_DIR: ${YOLO_CONFIG_DIR:-/app/storage/ultralytics} - YOLO_DEVICE: ${YOLO_DEVICE:-cpu} + YOLO_DEVICE: ${YOLO_DEVICE:-cuda:0} + YOLO_REQUIRE_CUDA: ${YOLO_REQUIRE_CUDA:-true} + YOLO_MODEL_CLASSES: ${YOLO_MODEL_CLASSES:-building} + YOLO_ENFORCE_VALIDATION_SCOPE: ${YOLO_ENFORCE_VALIDATION_SCOPE:-true} + YOLO_VALIDATED_AREA_NAMES: ${YOLO_VALIDATED_AREA_NAMES:-Mol,Kempen} YOLO_IMAGE_SIZE: ${YOLO_IMAGE_SIZE:-640} YOLO_MAX_TILES: ${YOLO_MAX_TILES:-100} YOLO_MAX_DETECTIONS: ${YOLO_MAX_DETECTIONS:-1000} @@ -138,6 +145,7 @@ services: - ${GEOINTEL_BACKUPS_PATH:-./backups}:/app/backups:ro extra_hosts: - "host.docker.internal:host-gateway" + gpus: all restart: unless-stopped volumes: diff --git a/docker-compose.yml b/docker-compose.yml index 39b4507d..69937964 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -120,6 +120,10 @@ services: YOLO_MODEL_VERSION: ${YOLO_MODEL_VERSION:-} YOLO_CONFIG_DIR: ${YOLO_CONFIG_DIR:-/app/storage/ultralytics} YOLO_DEVICE: ${YOLO_DEVICE:-cpu} + YOLO_REQUIRE_CUDA: ${YOLO_REQUIRE_CUDA:-false} + YOLO_MODEL_CLASSES: ${YOLO_MODEL_CLASSES:-building} + YOLO_ENFORCE_VALIDATION_SCOPE: ${YOLO_ENFORCE_VALIDATION_SCOPE:-false} + YOLO_VALIDATED_AREA_NAMES: ${YOLO_VALIDATED_AREA_NAMES:-Mol,Kempen} YOLO_IMAGE_SIZE: ${YOLO_IMAGE_SIZE:-640} YOLO_MAX_TILES: ${YOLO_MAX_TILES:-100} YOLO_MAX_DETECTIONS: ${YOLO_MAX_DETECTIONS:-1000} diff --git a/docs/AI_PIPELINES.md b/docs/AI_PIPELINES.md index d320b405..5ab7eaf9 100644 --- a/docs/AI_PIPELINES.md +++ b/docs/AI_PIPELINES.md @@ -92,7 +92,7 @@ Ultralytics/PyTorch compatibility, but it still does not run tile prediction and does not download weights. It cannot be combined with `--assume-dependencies` because that would turn the smoke into a false positive. -Docker and Unraid runtime support remains opt-in. Set `GEOINTEL_INSTALL_AI=true` +Docker AI dependencies remain opt-in. Set `GEOINTEL_INSTALL_AI=true` at build time to install the backend `.[gis,ai]` extra into the container. Leave it unset or `false` for the default GIS-only image. Runtime model files should be mounted into the container, for example `/app/models/local-model.pt`, and enabled @@ -109,6 +109,11 @@ Environment variables: - `YOLO_MODEL_DISPLAY_NAME` - `YOLO_MODEL_VERSION` - `YOLO_DEVICE` +- `YOLO_REQUIRE_CUDA` (set to `true` on the production server; inference then + fails closed when CUDA is unavailable or `YOLO_DEVICE` selects CPU) +- `YOLO_MODEL_CLASSES` (the active promoted detector is `building` only) +- `YOLO_ENFORCE_VALIDATION_SCOPE` and `YOLO_VALIDATED_AREA_NAMES` (production + rejects inference when the raster is not bound to a persisted validated Area) - `YOLO_IMAGE_SIZE` - `YOLO_MAX_TILES` - `YOLO_MAX_DETECTIONS` diff --git a/docs/API_CONTRACTS.md b/docs/API_CONTRACTS.md index 02b70e22..6df5c468 100644 --- a/docs/API_CONTRACTS.md +++ b/docs/API_CONTRACTS.md @@ -275,6 +275,74 @@ Idempotently creates or returns a project Area from the exact persisted NGI municipality geometry. The resulting Area can be used by all existing bounded selection, acquisition, analysis and export contracts. +## Resumable AOI operations + +### POST `/api/v1/projects/{project_id}/aoi-operations` + +Creates one persisted parent operation and deterministic bounded partitions. +Exactly one of `area_id` or an EPSG:4326 `bbox` is required. Partitioning is +calculated in EPSG:31370 and clipped to the exact AOI; source side limits remain +server concerns. Optional `coverage_zone` clips the immutable AOI snapshot to +the persisted legal/regional scope before planning. The production worker +automatically claims queued children; clients poll the parent instead of +driving provider requests. + +```json +{ + "area_id": "optional-uuid", + "operation_type": "acquire", + "provider_key": "grb", + "product_key": "buildings", + "max_partition_side_m": null, + "max_attempts": 3, + "parameters_json": {"force_refresh": false} +} +``` + +When `max_partition_side_m` is omitted, the backend derives the limit from the +governed provider registry. An explicit value can only make partitions smaller, +never relax the provider budget. The response reports parent status, progress from `0.0` to `1.0`, per-status +partition counts and child partition evidence. Stable partition keys make +planning and completion idempotent. + +### GET `/api/v1/projects/{project_id}/aoi-operations/{operation_id}` + +Returns the persisted operation, aggregate progress and every child state. + +### GET `/api/v1/projects/{project_id}/aoi-operations` + +Lists the most recent parent operations for the project with their derived +progress, status counts and child evidence. + +### POST `/api/v1/projects/{project_id}/aoi-operations/{operation_id}/execute-next` + +Claims and executes one queued partition through the registered governed +provider. Registered executors cover `grb`, `orthophoto`, `dhmv`, +`spw_terrain`, `official_vector`, `flood_hazard`, `thematic_raster`, `walous`, +`bathymetry_profiles` and `mdk_bathymetry`; their existing configuration and +zone contracts remain authoritative. Every +execution creates a linked child Job. Transient provider failures are retried +within the stored attempt budget; validation failures fail immediately. + +### POST `/api/v1/projects/{project_id}/aoi-operations/{operation_id}/partitions/claim` + +Atomically claims the next queued child using a locked, skip-locked database +selection. Returns `data: null` when no queued child remains. + +### PUT `/api/v1/projects/{project_id}/aoi-operations/{operation_id}/partitions/{partition_id}/checkpoint` + +Persists provider-specific restart evidence for a running child. + +### POST `/api/v1/projects/{project_id}/aoi-operations/{operation_id}/partitions/{partition_id}/complete` + +Marks a running child successful or explicitly skipped. Repeating completion +for a terminal successful/skipped child is idempotent. + +### POST `/api/v1/projects/{project_id}/aoi-operations/{operation_id}/partitions/{partition_id}/fail` + +Records an error and either returns the child to `queued` within its attempt +budget or terminalizes it as `failed`. + ## Datasets ### POST `/api/v1/projects/{project_id}/datasets/upload` @@ -1098,7 +1166,8 @@ mode, source URL, attribution, licence and limitation text. Allowed result states are exactly: -- `operational`: a matching `ready` Dataset is materialized in the project; +- `operational`: one matching `ready` Dataset or the spatial union of governed + partition Datasets covers the complete selection; - `partial`: the governed integration exists but matching project data is absent or incomplete; - `not_configured`: an audited source has no operational adapter; @@ -1124,7 +1193,9 @@ Areas. Cross-region and land/sea selections remain split by zone. The response returns `intersected_zones`, `outside_supported_scope`, one item per zone/theme combination, matching source names and IDs of actually -materialized Datasets. An empty `themes` list requests the complete normalized +materialized Datasets. Every materialized item also carries evidence for +authority, source version, observation/publication time, CRS, resolution, +coverage bbox, attribution, licence and checksum where persisted. An empty `themes` list requests the complete normalized vocabulary. Unknown themes fail with `COVERAGE_THEME_UNSUPPORTED`. The Map workbench uses this response before a rectangle analysis. It queries @@ -1396,6 +1467,7 @@ Response data: "checks": { "enabled": true, "dependencies_available": true, + "accelerator_ready": true, "model_path_set": false, "model_file_exists": null, "model_load_requested": false, @@ -1411,7 +1483,9 @@ Response data: "yolo_config_dir": "/app/storage/ultralytics", "torch_version": "2.12.1", "ultralytics_version": "8.4.88", - "cuda_available": false + "cuda_available": true, + "configured_device": "cuda:0", + "cuda_required": true }, "tile_count": 0, "max_tiles": 100, @@ -1422,7 +1496,14 @@ Response data: ### POST `/api/v1/detection/run` -Creates a detection job and detection analysis run. If the requested model is unavailable, the job and analysis run are marked `failed` with `DETECTION_MODEL_UNAVAILABLE` or `DETECTION_DEPENDENCY_UNAVAILABLE`. +Creates a detection job and detection analysis run. If the requested model is unavailable, the job and analysis run are marked `failed` with `DETECTION_MODEL_UNAVAILABLE` or `DETECTION_DEPENDENCY_UNAVAILABLE`. When `YOLO_REQUIRE_CUDA=true`, missing CUDA or a CPU device selection fails closed with `DETECTION_ACCELERATOR_UNAVAILABLE` or `DETECTION_ACCELERATOR_MISCONFIGURED`; production inference never silently falls back to CPU. + +The configured model exposes only classes declared by `YOLO_MODEL_CLASSES`; +the current promoted server detector is building-only. Unsupported class +filters fail with `DETECTION_CLASS_NOT_VALIDATED`. With production scope +enforcement enabled, the raster must belong to a persisted Area matching the +configured validated Mol/Kempen evidence or inference fails with +`DETECTION_VALIDATION_SCOPE_UNAVAILABLE`. Request: diff --git a/docs/AUDIT_REMEDIATION_ROADMAP.md b/docs/AUDIT_REMEDIATION_ROADMAP.md new file mode 100644 index 00000000..5b203846 --- /dev/null +++ b/docs/AUDIT_REMEDIATION_ROADMAP.md @@ -0,0 +1,108 @@ +# Audit remediation roadmap + +Status: active, 2026-07-26 + +## Release outcome + +GeoIntel may accept every valid AOI inside the governed Belgium and Belgian +North Sea scope. It may only call a theme operational where bounded processing, +source coverage, provenance, resolution and validation evidence support that +claim. Production AI inference on the server uses its NVIDIA GPU and fails +closed when CUDA is unavailable; CPU fallback is not an accepted production +state. + +## Wave 0 - Runtime truth and NVIDIA GPU (in progress) + +- expose the NVIDIA device to the Unraid container; +- set `YOLO_DEVICE=cuda:0` and `YOLO_REQUIRE_CUDA=true` in the server runtime; +- make preflight and model loading reject missing CUDA instead of using CPU; +- report configured device, CUDA requirement and accelerator readiness; +- rebuild on Tower and capture `nvidia-smi`, CUDA-enabled PyTorch, preflight and + one bounded inference smoke as release evidence. + +Exit gate: the live container sees the NVIDIA GPU, `torch.cuda.is_available()` +is true, preflight is ready, and a persisted smoke run records the configured +CUDA device. A CPU-only image or unavailable device remains `not_configured` / +unavailable and cannot run production inference. + +## Wave 1 - General AOI orchestration (in progress) + +- [x] introduce one persisted parent operation with source-specific partitions; +- [x] derive partitions from provider side budgets supplied by the governed plan; +- [x] support queued execution, bounded retries, checkpoints and restart recovery; +- make partition application idempotent and retain exact request/checksum + provenance; +- [x] reuse source-aware vector deduplication and raster mosaic contracts and + aggregate their Dataset identities into one parent result; +- [x] expose one progress/result contract to the frontend system workspace. + +Start with existing orthophoto, GRB and raster partition services; do not create +a second provider or persistence path. Keep per-source limits internal. A source +that cannot cover a partition returns explicit partial/not_configured evidence. + +Exit gate: interrupted cross-region and coastal golden AOIs resume without +duplicate rows and finish as one inspectable result. + +## Wave 2 - Zone x theme x source coverage truth (implemented; live gate pending) + +- [x] materialize the resolver contract for land, regions and legal maritime zones; +- [x] evaluate partition-union spatial coverage and expose source edition, + resolution, time, CRS, attribution, licence and checksum evidence; +- [x] derive only `operational`, `partial`, `not_configured` or `unsupported`; +- [x] show missing partitions and limitations in API and map states; +- [x] prohibit UI wording that implies complete national analysis from selection + acceptance alone. + +Exit gate: all frozen golden areas have checksum-bound coverage evidence and no +theme is promoted from file presence or a Mol-only success. + +## Wave 3 - Source completion + +- resolve North Sea bathymetry through a TLS-valid, authority-approved endpoint + or reviewed bounded operator acquisition; never bypass TLS; +- close configured orthophoto/nature/soil gaps for Wallonia where authoritative + machine access permits; +- close Brussels orthophoto gaps and retain unsupported statuses where no + source-appropriate analytical contract exists; +- add explicit external-catalog review evidence for the six source families + currently requiring review. + +Exit gate: every required theme/zone cell has current evidence and honest status; +vertical datums remain separate and water volume remains unavailable without a +governed compatible model. + +## Wave 4 - Model portfolio and validation (current active model gated) + +- [x] inventory model assets without treating presence as configuration; +- [x] bind the current active model to persisted Mol/Kempen Area evidence; +- [x] retain building-only semantics for the current active detector; +- add segmentation only with a configured model, georeferencing tests and + persisted polygon/mask evidence; +- [x] require a model/region pair to pass reproducible holdout, hard-negative and + QA gates pass. + +Exit gate: UI and API operational labels are derived from validation evidence, +not run counts; no claim that PyTorch itself is trained on themes. + +## Wave 5 - Release proof + +- run Mol/Kempen plus Walloon, Brussels, language-boundary, coastal and maritime + golden workflows; +- test fresh install, upgrade, rollback, restart/resume and sustained runtime; +- audit OpenAPI envelopes, CRS/units, provenance, exports and frontend + loading/empty/error states; +- publish exact checksums, source editions, model evidence and known limitations. + +Exit gate: `docs/DEFINITION_OF_DONE.md` and the RC freeze are satisfied. The +product claim remains location-complete at source-native resolution, never +literal centimetre-resolution. + +## Execution order + +1. Finish Wave 0 and verify it live on Tower. +2. Implement the parent/partition state machine and one GRB/orthophoto vertical + slice from Wave 1. +3. Generalize that slice across compatible raster/vector providers. +4. Deliver Wave 2 before promoting additional source cells. +5. Run Waves 3 and 4 in parallel only where their evidence is independent. +6. Close with Wave 5; do not advertise the audited guarantee before its gate. diff --git a/docs/CODEX_EXECUTION_LOG.md b/docs/CODEX_EXECUTION_LOG.md index 0cc99684..91e2eea9 100644 --- a/docs/CODEX_EXECUTION_LOG.md +++ b/docs/CODEX_EXECUTION_LOG.md @@ -11346,3 +11346,85 @@ Validation: - added backend API/service tests for multilingual/NIS search and persisted-area activation; - added frontend tests for live search, activation and explicit optional/free-selection semantics; - visually inspected the 512 px application mark and 32 px favicon derivative. +## 2026-07-26 - Audit remediation roadmap and NVIDIA GPU contract + +Implemented: + +- translated the platform audit into `docs/AUDIT_REMEDIATION_ROADMAP.md`, with + gated waves for NVIDIA runtime truth, general AOI orchestration, coverage + evidence, missing sources, model validation and release proof; +- made the Unraid production contract explicitly NVIDIA/CUDA-based through a + CUDA PyTorch wheel index, `gpus: all`, `YOLO_DEVICE=cuda:0` and + `YOLO_REQUIRE_CUDA=true`; +- added fail-closed accelerator validation to YOLO model loading and preflight, + including explicit unavailable/misconfigured errors instead of CPU fallback; +- extended the preflight response with accelerator readiness, configured device + and CUDA-required state, and updated the API, AI and dependency documentation; +- retained CPU as an allowed local-development default only when CUDA is not + explicitly required. + +Validation: + +- 57 focused backend tests passed for YOLO inference, preflight and Docker + runtime configuration; +- backend application byte-compilation passed; +- local Compose rendering could not run because this Windows workstation has no + `docker` CLI. Tower rebuild, `nvidia-smi`, CUDA-enabled PyTorch preflight and + one bounded persisted GPU inference smoke remain the live Wave 0 exit gate. + +Known limitations and next pass: + +- these repository changes do not prove that the deployed Tower container can + see the physical GPU; do not claim GPU readiness until the live gate passes; +- next implement the persisted parent/partition/checkpoint operation as the + first Wave 1 vertical slice, reusing existing job and provider services. +## 2026-07-26 - Audit remediation: resumable AOI federation and truthful AI runtime + +Implemented: + +- added migration `202607260001` with persisted AOI parent operations and + deterministic child partitions, exact EPSG:31370 planning, zone clipping, + checkpoints, attempt budgets, child Jobs and restart reconciliation; +- added a production background worker that automatically claims queued work + and dispatches ten existing governed providers without introducing a second + Dataset persistence path; +- aggregated child Dataset identities, completeness and source-aware vector + deduplication/raster mosaic contracts into one parent result and exposed live + progress plus failures in the System workspace; +- removed the frontend overview/detail refusal for on-demand themes and routed + regional/overview acquisition through the resumable server operation; +- made coverage resolution accept a spatial union of bounded partitions only + when it covers the selection, and exposed authority, edition, time, CRS, + resolution, bbox, attribution, licence and checksum evidence per Dataset; +- corrected the active YOLO contract to building-only, fail-closed CUDA and + persisted Mol/Kempen Area scope. Production cannot advertise configured AI + from a CPU runtime or execute an unvalidated class/area. + +Validation: + +- one Alembic head (`202607260001`) and a complete 40,888-byte offline SQL + migration chain were generated; +- 71 focused backend tests, 40 frontend tests, frontend typecheck and production + build passed; +- the broader Windows-compatible backend gate passed 1,115 tests with five + WSL-dependent shell tests deselected because the workstation WSL VHD is + missing; those Linux shell gates remain mandatory on Tower; +- API contract audit passed with all 146 routes documented. + +Live pre-deploy evidence: + +- Tower is healthy on PostGIS 3.6 at migration `202607160001`; +- the host has an NVIDIA GeForce RTX 4080 SUPER with 16,376 MiB and driver + 595.71.05; +- the old container confirms the audited failure state: no Docker device + request, `torch 2.13.0+cpu`, CUDA false and zero visible GPUs; +- the official Vlaanderen catalog still identifies MDK Version 8, 20 m, LAT as + live production data, but both catalogued HTTPS host variants fail strict TLS + and HTTP does not expose the WCS path. MDK remains `not_configured`; TLS is not + bypassed. + +Next gate: + +- commit/push the immutable source, deploy on Tower, run the Linux readiness and + migration gates, prove CUDA PyTorch plus a bounded persisted inference, and + capture golden AOI/coverage evidence. diff --git a/docs/DATABASE_IMPLEMENTATION_PLAN.md b/docs/DATABASE_IMPLEMENTATION_PLAN.md index c4d4a819..e1fbf637 100644 --- a/docs/DATABASE_IMPLEMENTATION_PLAN.md +++ b/docs/DATABASE_IMPLEMENTATION_PLAN.md @@ -318,6 +318,22 @@ Official editions take precedence over legacy rolling markers in catalog comparison, but direct metadata backfill, update or deletion of those legacy rows remains prohibited. +## Resumable AOI operations + +Migration `202607260001` adds `aoi_operations` and +`aoi_operation_partitions`. The parent stores the immutable EPSG:4326 AOI +snapshot, original request and source-specific plan. Child rows store exact +partition geometry, stable `(operation_id, partition_key)` identity, provider +and product, attempt budget, checkpoint, result and child Job linkage. + +Workers atomically claim queued child rows. Interrupted running rows return to +queued state while attempts remain and become failed when exhausted. Provider +output still flows through the existing DatasetService/provider contracts; +these tables orchestrate work and never become a parallel dataset store. +The all-in-one runtime enables one database-polling worker. Multiple future +workers remain safe because partition claim uses `FOR UPDATE SKIP LOCKED` and +the partition key is unique inside its parent operation. + ## Geometry normalization - User-drawn polygons arrive as EPSG:4326. diff --git a/docs/DEPENDENCY_POLICY.md b/docs/DEPENDENCY_POLICY.md index aa566203..c4bcd8c9 100644 --- a/docs/DEPENDENCY_POLICY.md +++ b/docs/DEPENDENCY_POLICY.md @@ -45,12 +45,13 @@ AI dependencies remain separate in the `ai` optional dependency group and must not be installed by the default Docker backend image unless an explicit AI image or profile is introduced later. -The opt-in Unraid all-in-one AI build is CPU-oriented because its documented -runtime sets `YOLO_DEVICE=cpu`. It installs the pinned PyTorch/torchvision pair -from PyTorch's CPU wheel index before installing the `ai` extra. This avoids -shipping unused CUDA runtime libraries. The index and versions remain explicit -Docker build arguments so a future, separately validated GPU image can override -them without changing the base dependency group. +The opt-in Unraid all-in-one AI build is NVIDIA-GPU-oriented. It installs the +pinned PyTorch/torchvision pair from the CUDA 13.0 wheel index before installing +the `ai` extra. The production runtime exposes the NVIDIA device, selects +`YOLO_DEVICE=cuda:0` and sets `YOLO_REQUIRE_CUDA=true`, so missing CUDA fails +closed instead of silently falling back to CPU. The index and versions remain +explicit Docker build arguments and require live driver/runtime validation on +Tower before release promotion. Docker dependency metadata is copied before application source. Backend source changes therefore reuse the dependency layer while changes to `pyproject.toml` diff --git a/docs/TODO.md b/docs/TODO.md index a1b99163..68c16dd5 100644 --- a/docs/TODO.md +++ b/docs/TODO.md @@ -918,3 +918,14 @@ This file now starts with the current implementation status. Older preparation/b - [x] Explain theme choice as map and primary-metric focus rather than a mandatory workflow gate. - [x] Keep narrow theme-card copy and status labels inside their cards. - [x] Replace the former compass icon with a scale-safe GeoIntel geo-lens mark. +# Audit remediation (2026-07-26) + +- [x] Freeze an audit remediation roadmap with NVIDIA GPU as a server requirement. +- [x] Add fail-closed CUDA configuration and preflight/runtime validation. +- [ ] Rebuild Tower with NVIDIA device exposure and capture a CUDA inference smoke. +- [x] Implement the persisted parent/partition/checkpoint AOI operation. +- [x] Add source-aware seam deduplication contracts and one-result Dataset aggregation across child partitions. +- [x] Extend the zone x theme x source resolver with partition-union coverage and per-Dataset evidence. +- [ ] Resolve governed North Sea bathymetry access without bypassing TLS. +- [x] Enforce the active building-only model classes, CUDA readiness and persisted Mol/Kempen Area scope. +- [ ] Add national/regional holdout evidence before expanding model classes or validated Areas. diff --git a/frontend/src/App.tsx b/frontend/src/App.tsx index 2e5c6dee..9e3fda38 100644 --- a/frontend/src/App.tsx +++ b/frontend/src/App.tsx @@ -1238,6 +1238,7 @@ function WorkbenchApp({ username, loggingOut, onLogout }: WorkbenchAppProps): JS {activeWorkspace === 'system' ? ( productSupportsSelection(product, bbox)) .map((product) => ({ @@ -2025,6 +2025,8 @@ export function MapWorkspace({ productKey: onDemandProduct.productKey, displayName: onDemandProduct.displayName, historyProductKeys: onDemandProduct.historyProductKeys, + coverageZone: onDemandProduct.coverageZones.find((zone) => resolvedZones?.includes(zone)), + serverOrchestrated: scale !== 'detail', }, acquisitionBboxes: onDemandProduct.acquisitionBboxes, featureLimit: resultFeatureLimit, diff --git a/frontend/src/components/providers/ProviderPanel.tsx b/frontend/src/components/providers/ProviderPanel.tsx index 8738a8bc..7a236b43 100644 --- a/frontend/src/components/providers/ProviderPanel.tsx +++ b/frontend/src/components/providers/ProviderPanel.tsx @@ -1,6 +1,9 @@ -import type { ProviderCapability } from '../../types' +import { useCallback, useEffect, useState } from 'react' +import type { AoiOperation, ProviderCapability } from '../../types' +import { aoiOperationsApi } from '../../services/api' interface ProviderPanelProps { + selectedProjectId: string | null providers: ProviderCapability[] loadingCapabilities: boolean capabilitiesError: string | null @@ -40,6 +43,7 @@ function providerLayerLabel(value: string): string { } export function ProviderPanel({ + selectedProjectId, providers, loadingCapabilities, capabilitiesError, @@ -49,6 +53,32 @@ export function ProviderPanel({ onOpenMap, }: ProviderPanelProps): JSX.Element { const configuredCount = providers.filter((provider) => provider.configured).length + const [operations, setOperations] = useState([]) + const [operationsError, setOperationsError] = useState(null) + const [loadingOperations, setLoadingOperations] = useState(false) + const loadOperations = useCallback(async () => { + if (!selectedProjectId) { + setOperations([]) + return + } + setLoadingOperations(true) + try { + const response = await aoiOperationsApi.list(selectedProjectId) + setOperations(response.items) + setOperationsError(null) + } catch (error) { + setOperationsError(error instanceof Error ? error.message : 'AOI-verwerking kon niet worden geladen.') + } finally { + setLoadingOperations(false) + } + }, [selectedProjectId]) + + useEffect(() => { + void loadOperations() + const timer = window.setInterval(() => void loadOperations(), 5000) + return () => window.clearInterval(timer) + }, [loadOperations]) + return (
@@ -93,6 +123,31 @@ export function ProviderPanel({ ) : null}
+
+
+
+ Gebiedsverwerking +

Server-side partities, hervatbare voortgang en bronfouten voor grote selecties.

+
+ +
+ {loadingOperations && operations.length === 0 ?
Verwerkingen laden.
: null} + {operationsError ?
Verwerkingsstatus niet bereikbaar.

{operationsError}

: null} + {!loadingOperations && !operationsError && operations.length === 0 ?
Nog geen gebiedsverwerking.

Grote officiële bronselecties verschijnen hier met één gezamenlijke voortgang.

: null} + {operations.length > 0 ? ( +
    + {operations.map((operation) => ( +
  • +
    {operation.operation_type}{String(operation.plan_json.provider_key ?? 'bron')} · {String(operation.plan_json.product_key ?? 'product')}
    {operation.status}
    + +
    {Math.round(operation.progress * 100)}%{operation.partitions.length} partities{operation.partition_counts.failed ?? 0} mislukt
    + {operation.error_message ?

    {operation.error_message}

    : null} +
    • + ))} +
+ ) : null} +
+
Officiële referentiebronnen diff --git a/frontend/src/hooks/useMapThemeSelectionInsights.ts b/frontend/src/hooks/useMapThemeSelectionInsights.ts index f79bf2e4..7799c937 100644 --- a/frontend/src/hooks/useMapThemeSelectionInsights.ts +++ b/frontend/src/hooks/useMapThemeSelectionInsights.ts @@ -1,6 +1,7 @@ import { useEffect, useRef, useState } from 'react' import { formatError } from '../lib/formatError' import { datasetsApi } from '../services/api/datasets' +import { aoiOperationsApi } from '../services/api/aoiOperations' import type { DatasetCreateResponse, VectorSelectionBBox, VectorSelectionResponse } from '../types' import { terrainSelectionToMapSelection } from '../lib/terrainSelection' import { floodHazardSelectionToMapSelection } from '../lib/floodHazardSelection' @@ -22,6 +23,8 @@ export interface MapThemeAcquisition { productKey: string displayName: string historyProductKeys?: string[] + coverageZone?: string + serverOrchestrated?: boolean } export interface MapThemeQuery { @@ -116,6 +119,30 @@ export function useMapThemeSelectionInsights( const resultLimit = featureLimit ?? 1000 if (acquisition) { const requestedBboxes = acquisitionBboxes?.length ? acquisitionBboxes : [bbox] + if (acquisition.serverOrchestrated) { + let operation = await aoiOperationsApi.create(selectedProjectId, { + ...(areaId ? { area_id: areaId } : { bbox }), + operation_type: 'acquire', + provider_key: acquisition.kind, + product_key: acquisition.productKey, + coverage_zone: acquisition.coverageZone, + max_attempts: 3, + parameters_json: { force_refresh: false }, + }) + const deadline = Date.now() + 20 * 60 * 1000 + while (['queued', 'running'].includes(operation.status) && Date.now() < deadline) { + await new Promise((resolve) => window.setTimeout(resolve, 1500)) + operation = await aoiOperationsApi.get(selectedProjectId, operation.id) + } + if (operation.status !== 'success') { + throw new Error(operation.error_message || `De gebiedsverwerking eindigde als ${operation.status}.`) + } + const outputIds = Array.isArray(operation.result_json?.['output_dataset_ids']) + ? operation.result_json['output_dataset_ids'].map(String) + : [] + acquiredDatasets = await Promise.all(outputIds.map((datasetId) => datasetsApi.get(selectedProjectId, datasetId))) + dataset = acquiredDatasets[0] + } const acquireProduct = async (acquisitionBbox: VectorSelectionBBox, productKey: string) => { const commonPayload = { bbox: acquisitionBbox, @@ -166,7 +193,7 @@ export function useMapThemeSelectionInsights( } return datasetsApi.get(selectedProjectId, acquisitionJob.output_dataset_id) } - const acquisitionResults = await settleWithConcurrency( + const acquisitionResults = acquisition.serverOrchestrated ? [] : await settleWithConcurrency( requestedBboxes, 1, (acquisitionBbox) => acquireProduct(acquisitionBbox, acquisition.productKey), @@ -175,7 +202,7 @@ export function useMapThemeSelectionInsights( if (failedAcquisition?.status === 'rejected') { throw failedAcquisition.reason } - acquiredDatasets = acquisitionResults.flatMap((item) => item.status === 'fulfilled' ? [item.value] : []) + if (!acquisition.serverOrchestrated) acquiredDatasets = acquisitionResults.flatMap((item) => item.status === 'fulfilled' ? [item.value] : []) dataset = acquiredDatasets[0] const historyProductKeys = acquisition.kind === 'walous' ? [...new Set(acquisition.historyProductKeys ?? [])].filter((key) => key !== acquisition.productKey) diff --git a/frontend/src/services/api/aoiOperations.ts b/frontend/src/services/api/aoiOperations.ts new file mode 100644 index 00000000..e79b5ad9 --- /dev/null +++ b/frontend/src/services/api/aoiOperations.ts @@ -0,0 +1,15 @@ +import { apiGet, apiPost, apiPut } from './client' +import type { AoiOperation, AoiOperationListResponse, AoiOperationPartition } from '../../types' + +export const aoiOperationsApi = { + list: (projectId: string): Promise => + apiGet(`/api/v1/projects/${projectId}/aoi-operations`), + get: (projectId: string, operationId: string): Promise => + apiGet(`/api/v1/projects/${projectId}/aoi-operations/${operationId}`), + create: (projectId: string, payload: Record): Promise => + apiPost(`/api/v1/projects/${projectId}/aoi-operations`, payload), + executeNext: (projectId: string, operationId: string): Promise => + apiPost(`/api/v1/projects/${projectId}/aoi-operations/${operationId}/execute-next`, {}), + checkpoint: (projectId: string, operationId: string, partitionId: string, checkpoint: Record): Promise => + apiPut(`/api/v1/projects/${projectId}/aoi-operations/${operationId}/partitions/${partitionId}/checkpoint`, { checkpoint_json: checkpoint }), +} diff --git a/frontend/src/services/api/client.ts b/frontend/src/services/api/client.ts index 8eb5205a..0353ac63 100644 --- a/frontend/src/services/api/client.ts +++ b/frontend/src/services/api/client.ts @@ -56,6 +56,16 @@ export async function apiPatch(path: string, body?: object): Promise { return parseResponse(response); } +export async function apiPut(path: string, body?: object): Promise { + const response = await fetch(apiUrl(path), { + method: "PUT", + credentials: "same-origin", + headers: { "Content-Type": "application/json" }, + body: body ? JSON.stringify(body) : undefined, + }); + return parseResponse(response); +} + export async function apiDelete(path: string): Promise { const response = await fetch(apiUrl(path), { method: "DELETE", diff --git a/frontend/src/services/api/index.ts b/frontend/src/services/api/index.ts index 9691681a..3df661b5 100644 --- a/frontend/src/services/api/index.ts +++ b/frontend/src/services/api/index.ts @@ -1,4 +1,5 @@ export { areasApi } from './areas' +export { aoiOperationsApi } from './aoiOperations' export { analysisApi } from './analysis' export { assistantApi } from './assistant' export { datasetsApi } from './datasets' diff --git a/frontend/src/types.ts b/frontend/src/types.ts index 7f56ed98..5a1b1d43 100644 --- a/frontend/src/types.ts +++ b/frontend/src/types.ts @@ -147,6 +147,44 @@ export interface JobListResponse { offset: number } +export interface AoiOperationPartition { + id: string + partition_key: string + provider_key: string + product_key: string + ordinal: number + status: 'queued' | 'running' | 'success' | 'failed' | 'skipped' + attempt_count: number + max_attempts: number + checkpoint_json?: Record | null + result_json?: Record | null + error_message?: string | null +} + +export interface AoiOperation { + id: string + project_id: string + area_id?: string | null + parent_job_id?: string | null + operation_type: string + status: 'queued' | 'running' | 'partial' | 'success' | 'failed' | 'cancelled' + request_json: Record + plan_json: Record + result_json?: Record | null + error_message?: string | null + progress: number + partition_counts: Record + partitions: AoiOperationPartition[] + created_at?: string | null + started_at?: string | null + finished_at?: string | null +} + +export interface AoiOperationListResponse { + items: AoiOperation[] + total: number +} + export interface VectorSummary { feature_count?: number | null geometry_types?: string[] | null @@ -1064,6 +1102,20 @@ export interface CoverageResolutionItem { status: CoverageStatus source_names: string[] materialized_dataset_ids: string[] + evidence: Array<{ + dataset_id: string + source_name: string + authority_level: 'authoritative' | 'official_context' | 'contextual' + source_version?: string | null + observed_at?: string | null + published_at?: string | null + crs?: string | null + resolution?: Record | null + coverage_bbox_epsg4326?: number[] | null + attribution?: string | null + license_note?: string | null + checksum_sha256?: string | null + }> limitation_message: string }