Files
geointel/backend/app/services/segmentation_service.py
T
JensandClaude Opus 5 5b3839dc89 page detection and segmentation results instead of returning all of them
/detection/runs/{id}/detections and its GeoJSON sibling returned every
persisted detection, as did the segmentation equivalents. A regional run holds
tens of thousands, and these are the endpoints the results table and the map
overlay call after every run.

They now take limit and offset, default to 2.000, and report total, limit,
offset and truncated so the complete population stays visible while what is
transferred does not. The GeoJSON responses carry the same window in a
geointel_result_window foreign member.

Rows are ordered by confidence, so a capped overlay draws the strongest
detections rather than an arbitrary slice, and the lab says how many of how
many are being shown rather than silently presenting a page as the whole run.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-08-22 15:24:46 +02:00

974 lines
43 KiB
Python

from __future__ import annotations
import uuid
from datetime import UTC, datetime
from pathlib import Path
from typing import Any
from geoalchemy2.shape import from_shape, to_shape
from shapely.geometry import MultiPolygon, Polygon, mapping, shape
from shapely.validation import make_valid
from sqlalchemy import func
from app.core.config import Settings, get_settings
from app.core.errors import AppError
from app.models import AnalysisRun, Dataset, Job, Project, Segmentation, VectorFeature
from app.schemas.segmentation import (
SegmentationListResponse,
SegmentationRead,
SegmentationRunListResponse,
SegmentationRunRead,
SegmentationRunResponse,
)
from app.services.detection_georeferencing import pixel_points_to_epsg4326_polygon
from app.services.detection_qa_service import DetectionQaService
from app.services.detection_service import DetectionService
from app.services.dataset_consumption_gate_service import DatasetConsumptionGate
from app.services.model_registry_service import ModelRegistryService
from app.services.qa_service import QaService
from app.services.quality_service import QualityService
from app.services.runtime_model_provenance_service import RuntimeModelProvenance, RuntimeModelProvenanceService
from app.services.segmentation_adapter import (
FixtureSegmentationAdapter,
SamSegmentationAdapter,
YoloSegmentationAdapter,
)
class SegmentationService:
@staticmethod
def _now() -> datetime:
return datetime.now(UTC)
@staticmethod
def run_segmentation(
db,
project_id: uuid.UUID,
dataset_id: uuid.UUID,
model_id: str,
confidence_threshold: float,
class_filter: list[str] | None = None,
tile_manifest_path: str | None = None,
parameters_json: dict[str, Any] | None = None,
settings: Settings | None = None,
yolo_seg_adapter_class: type[YoloSegmentationAdapter] = YoloSegmentationAdapter,
sam_adapter_class: type[SamSegmentationAdapter] = SamSegmentationAdapter,
existing_job: Job | None = None,
) -> SegmentationRunResponse:
parameters = dict(parameters_json or {})
resolved_settings = settings or get_settings()
dataset = SegmentationService._validate_run_request(db, project_id=project_id, dataset_id=dataset_id)
model = ModelRegistryService.get_model_capability(
model_id,
settings=resolved_settings,
task_type="segmentation",
yolo_seg_adapter_class=yolo_seg_adapter_class,
sam_adapter_class=sam_adapter_class,
)
if model is None:
raise AppError(code="SEGMENTATION_MODEL_NOT_FOUND", message="Segmentation model not found", status_code=404)
if model.model_id == "fixture-segmenter" and parameters.get("fixture_mode") is not True:
raise AppError(
code="FIXTURE_MODE_REQUIRED",
message="Fixture segmenter requires explicit fixture_mode=true",
status_code=400,
)
configured_model_ids = {resolved_settings.yolo_seg_model_id, resolved_settings.sam_model_id}
if model.model_id in configured_model_ids and model.configured and not tile_manifest_path:
raise AppError(
code="SEGMENTATION_TILE_MANIFEST_REQUIRED",
message="Configured segmentation inference requires an existing raster tile manifest path",
status_code=400,
)
# Production segmentation must consume only a passed, complete and
# non-quarantined dataset. The fixture segmenter is QA/test-only and
# cannot be classified as production inference.
if model.model_id == "fixture-segmenter":
DatasetConsumptionGate.assert_eligible(
dataset,
purpose="quality_assessment",
fixture_mode=True,
)
elif model.model_id in configured_model_ids and model.configured:
DatasetConsumptionGate.assert_eligible(dataset, purpose="production_inference")
run_parameters = {
"model_id": model.model_id,
"confidence_threshold": confidence_threshold,
"class_filter": class_filter or [],
"tile_manifest_path": tile_manifest_path,
"parameters_json": parameters,
}
job = SegmentationService._create_job(db, project_id, dataset_id, run_parameters, existing_job=existing_job)
analysis_run = SegmentationService._create_analysis_run(db, project_id, dataset_id, job.id, model, run_parameters)
if not model.configured:
message = model.limitation_message
SegmentationService._mark_failed(
db,
analysis_run,
job,
code="SEGMENTATION_MODEL_UNAVAILABLE",
message=message,
)
return SegmentationRunResponse(
analysis_run_id=analysis_run.id,
job_id=job.id,
project_id=project_id,
dataset_id=dataset_id,
model_id=model.model_id,
status="failed",
segmentation_count=0,
error_code="SEGMENTATION_MODEL_UNAVAILABLE",
message=message,
)
if model.model_id == "fixture-segmenter":
try:
segmentations = SegmentationService._persist_fixture_segmentations(
db=db,
project_id=project_id,
dataset_id=dataset_id,
analysis_run=analysis_run,
job=job,
model_name=model.model_id,
model_version=model.version,
raw_segmentations=parameters.get("fixture_segmentations"),
confidence_threshold=confidence_threshold,
class_filter=class_filter or [],
settings=resolved_settings,
)
except Exception as exc:
# A rejected fixture payload must never leave the run stuck in "running".
SegmentationService._fail_run_after_exception(db, analysis_run, job, exc, fallback_code="SEGMENTATION_INTERNAL_ERROR")
raise
SegmentationService._mark_success(db, analysis_run, job, segmentation_count=len(segmentations))
return SegmentationRunResponse(
analysis_run_id=analysis_run.id,
job_id=job.id,
project_id=project_id,
dataset_id=dataset_id,
model_id=model.model_id,
status="success",
segmentation_count=len(segmentations),
message="Fixture segmentations persisted.",
)
if model.model_id in configured_model_ids:
try:
segmentations, postprocess_summary = SegmentationService._run_configured_segmentation(
db=db,
project_id=project_id,
dataset_id=dataset_id,
analysis_run=analysis_run,
job=job,
model_name=model.model_id,
model_version=model.version,
tile_manifest_path=tile_manifest_path,
confidence_threshold=confidence_threshold,
class_filter=class_filter or [],
settings=resolved_settings,
yolo_seg_adapter_class=yolo_seg_adapter_class,
sam_adapter_class=sam_adapter_class,
)
except AppError as exc:
SegmentationService._mark_failed(db, analysis_run, job, code=exc.code, message=exc.message)
return SegmentationRunResponse(
analysis_run_id=analysis_run.id,
job_id=job.id,
project_id=project_id,
dataset_id=dataset_id,
model_id=model.model_id,
status="failed",
segmentation_count=0,
error_code=exc.code,
message=exc.message,
)
except Exception as exc:
# An unexpected inference error must never leave the run stuck in "running".
SegmentationService._fail_run_after_exception(db, analysis_run, job, exc, fallback_code="SEGMENTATION_INTERNAL_ERROR")
raise
SegmentationService._mark_success(
db,
analysis_run,
job,
segmentation_count=len(segmentations),
extra_result=postprocess_summary,
)
return SegmentationRunResponse(
analysis_run_id=analysis_run.id,
job_id=job.id,
project_id=project_id,
dataset_id=dataset_id,
model_id=model.model_id,
status="success",
segmentation_count=len(segmentations),
message="Configured segmentation inference persisted georeferenced masks.",
)
SegmentationService._mark_failed(
db,
analysis_run,
job,
code="SEGMENTATION_MODEL_UNAVAILABLE",
message="Segmentation model is unavailable",
)
raise AppError(code="SEGMENTATION_MODEL_UNAVAILABLE", message="Segmentation model is unavailable", status_code=503)
@staticmethod
def _fail_run_after_exception(db, analysis_run: AnalysisRun, job: Job, exc: Exception, fallback_code: str) -> None:
try:
db.rollback()
except Exception:
pass
code = getattr(exc, "code", None) or fallback_code
message = getattr(exc, "message", None) or "Unexpected internal error during analysis run"
try:
SegmentationService._mark_failed(db, analysis_run, job, code=str(code), message=str(message))
except Exception:
pass
@staticmethod
def get_run(db, analysis_run_id: uuid.UUID) -> SegmentationRunRead:
run = db.get(AnalysisRun, analysis_run_id)
if not run or run.analysis_type != "segmentation":
raise AppError(code="SEGMENTATION_RUN_NOT_FOUND", message="Segmentation run not found", status_code=404)
return SegmentationRunRead.model_validate(run)
@staticmethod
def list_runs(
db,
*,
project_id: uuid.UUID | None = None,
dataset_id: uuid.UUID | None = None,
) -> SegmentationRunListResponse:
query = db.query(AnalysisRun).filter(AnalysisRun.analysis_type == "segmentation")
if project_id is not None:
query = query.filter(AnalysisRun.project_id == project_id)
if dataset_id is not None:
query = query.filter(AnalysisRun.dataset_id == dataset_id)
rows = query.order_by(AnalysisRun.created_at.desc()).all()
return SegmentationRunListResponse(items=[SegmentationRunRead.model_validate(row) for row in rows], total=len(rows))
@staticmethod
def list_segmentations(
db,
analysis_run_id: uuid.UUID | None = None,
*,
dataset_id: uuid.UUID | None = None,
class_name: str | None = None,
min_confidence: float | None = None,
) -> SegmentationListResponse:
if analysis_run_id is not None:
run = db.get(AnalysisRun, analysis_run_id)
if not run or run.analysis_type != "segmentation":
raise AppError(code="SEGMENTATION_RUN_NOT_FOUND", message="Segmentation run not found", status_code=404)
rows = SegmentationService._query_segmentation_rows(
db,
analysis_run_id=analysis_run_id,
dataset_id=dataset_id,
class_name=class_name,
min_confidence=min_confidence,
)
items = [SegmentationRead.model_validate(row) for row in rows]
return SegmentationListResponse(items=items, total=len(items))
@staticmethod
def get_segmentation(db, segmentation_id: uuid.UUID) -> SegmentationRead:
segmentation = db.get(Segmentation, segmentation_id)
if not segmentation:
raise AppError(code="SEGMENTATION_NOT_FOUND", message="Segmentation not found", status_code=404)
return SegmentationRead.model_validate(segmentation)
@staticmethod
def segmentations_to_geojson(
db,
*,
analysis_run_id: uuid.UUID | None = None,
dataset_id: uuid.UUID | None = None,
class_name: str | None = None,
min_confidence: float | None = None,
limit: int | None = None,
) -> dict[str, Any]:
rows = SegmentationService._query_segmentation_rows(
db,
analysis_run_id=analysis_run_id,
dataset_id=dataset_id,
class_name=class_name,
min_confidence=min_confidence,
)
resolved_limit = DetectionService.DEFAULT_RESULT_LIMIT if limit is None else int(limit)
segmentations, total, truncated = DetectionService.paginate(rows, limit=resolved_limit, offset=0)
return {
"type": "FeatureCollection",
"geointel_result_window": {
"feature_count": len(segmentations),
"total_feature_count": total,
"limit": resolved_limit,
"truncated": truncated,
},
"features": [
{
"type": "Feature",
"id": str(segmentation.id),
"properties": SegmentationService._segmentation_properties(segmentation),
"geometry": mapping(to_shape(segmentation.geometry)),
}
for segmentation in segmentations
],
}
@staticmethod
def compare_segmentations_with_reference(
db,
analysis_run_id: uuid.UUID,
reference_dataset_id: uuid.UUID,
iou_threshold: float = 0.5,
class_name: str | None = None,
min_confidence: float | None = None,
) -> dict[str, Any]:
run = db.get(AnalysisRun, analysis_run_id)
if not run or run.analysis_type != "segmentation":
raise AppError(code="SEGMENTATION_RUN_NOT_FOUND", message="Segmentation run not found", status_code=404)
reference_dataset = db.get(Dataset, reference_dataset_id)
if not reference_dataset:
raise AppError(code="DATASET_NOT_FOUND", message="Reference dataset not found", status_code=404)
if reference_dataset.project_id != run.project_id:
raise AppError(code="INVALID_DATASET_SCOPE", message="Reference dataset does not belong to segmentation project", status_code=400)
if reference_dataset.dataset_type not in {"vector", "geojson"}:
raise AppError(code="INVALID_DATASET_TYPE", message="Reference dataset must be vector data", status_code=400)
candidate_dataset = db.get(Dataset, run.dataset_id)
if not candidate_dataset:
raise AppError(code="DATASET_NOT_FOUND", message="Segmentation source dataset not found", status_code=404)
run_parameters = run.parameters_json if isinstance(run.parameters_json, dict) else {}
fixture_parameters = run_parameters.get("parameters_json")
fixture_mode = bool(
run.model_name == "fixture-segmenter"
and isinstance(fixture_parameters, dict)
and fixture_parameters.get("fixture_mode") is True
)
DatasetConsumptionGate.assert_eligible(
candidate_dataset,
purpose="quality_assessment",
fixture_mode=fixture_mode,
)
DatasetConsumptionGate.assert_eligible(
reference_dataset,
purpose="reference_validation",
reference_task="building_validation",
)
segmentations = SegmentationService._query_segmentation_rows(
db,
analysis_run_id=analysis_run_id,
dataset_id=run.dataset_id,
class_name=class_name,
min_confidence=min_confidence,
)
if not segmentations:
raise AppError(
code="SEGMENTATIONS_NOT_FOUND",
message="Segmentation run has no persisted geometries for QA",
status_code=422,
)
# Score against the footprint the model actually saw. Without this the
# whole reference dataset is the denominator for recall, and every
# building outside the inferred tiles is counted as a miss.
manifest_path = DetectionQaService.tile_manifest_path(run_parameters)
coverage = None
if manifest_path:
manifest = DetectionService._load_tile_manifest(manifest_path, get_settings().yolo_max_tiles)
coverage = DetectionQaService.build_tile_coverage(
manifest,
manifest_path=manifest_path,
expected_dataset_id=run.dataset_id,
)
reference_query = db.query(VectorFeature).filter(VectorFeature.dataset_id == reference_dataset_id)
if coverage is not None and hasattr(reference_query, "count"):
reference_raw_count = reference_query.count()
references = reference_query.filter(
func.ST_Intersects(VectorFeature.geometry, from_shape(coverage.geometry, srid=4326))
).all()
else:
references = reference_query.all()
reference_raw_count = len(references)
if reference_raw_count == 0:
raise AppError(
code="REFERENCE_FEATURES_NOT_FOUND",
message="Reference dataset has no persisted vector features for QA",
status_code=422,
)
raw_candidate_geometries = [
(
{"id": str(row.id), "class_name": row.class_name, "confidence": row.confidence},
to_shape(row.geometry),
)
for row in segmentations
]
raw_reference_geometries = [
({"id": str(row.id), "feature_class": row.feature_class}, to_shape(row.geometry)) for row in references
]
candidate_geometries = raw_candidate_geometries
reference_geometries = raw_reference_geometries
coverage_summary: dict[str, Any] = {
"applied": False,
"mode": "unbounded_no_manifest",
"manifest_path": None,
"tile_count": 0,
"source_crs_values": [],
"candidate_raw_count": len(raw_candidate_geometries),
"candidate_evaluated_count": len(raw_candidate_geometries),
"candidate_excluded_outside_count": 0,
"candidate_clipped_boundary_count": 0,
"reference_raw_count": reference_raw_count,
"reference_evaluated_count": len(raw_reference_geometries),
"reference_excluded_outside_count": 0,
"reference_clipped_boundary_count": 0,
}
coverage_warnings: list[str] = []
if coverage is not None:
candidate_population = DetectionQaService.filter_population(raw_candidate_geometries, coverage)
reference_population = DetectionQaService.filter_population(
raw_reference_geometries,
coverage,
raw_count=reference_raw_count,
)
candidate_geometries = candidate_population.geometries
reference_geometries = reference_population.geometries
if not reference_geometries:
raise AppError(
code="REFERENCE_FEATURES_OUTSIDE_COVERAGE",
message="Reference dataset has no polygon features inside persisted inference tile coverage",
status_code=422,
)
coverage_summary = {
"applied": True,
"mode": "persisted_tile_manifest_union",
"manifest_path": coverage.manifest_path,
"tile_count": coverage.tile_count,
"source_crs_values": list(coverage.source_crs_values),
"candidate_raw_count": candidate_population.raw_count,
"candidate_evaluated_count": candidate_population.evaluated_count,
"candidate_excluded_outside_count": candidate_population.excluded_outside_count,
"candidate_clipped_boundary_count": candidate_population.clipped_boundary_count,
"reference_raw_count": reference_population.raw_count,
"reference_evaluated_count": reference_population.evaluated_count,
"reference_excluded_outside_count": reference_population.excluded_outside_count,
"reference_clipped_boundary_count": reference_population.clipped_boundary_count,
}
coverage_warnings.append(
"QA populations were clipped to the union of persisted inference tile footprints before matching."
)
evidence = QaService._match_io_u_evidence(
candidate_geometries,
reference_geometries,
iou_threshold,
)
mean_iou = None if not evidence.match_iou_values else sum(evidence.match_iou_values) / len(evidence.match_iou_values)
precision = evidence.matches / (evidence.matches + evidence.false_positives) if evidence.matches + evidence.false_positives > 0 else None
recall = evidence.matches / (evidence.matches + evidence.false_negatives) if evidence.matches + evidence.false_negatives > 0 else None
f1_score = None
if precision is not None and recall is not None:
f1_score = (2 * precision * recall) / (precision + recall) if precision + recall > 0 else 0.0
status = "unsupported" if evidence.unsupported else "ok"
quality_check = QualityService.persist_quality_check(
db=db,
project_id=run.project_id,
analysis_run_id=analysis_run_id,
candidate_dataset_id=run.dataset_id,
reference_dataset_id=reference_dataset_id,
check_type="segmentations_vs_reference",
status=status,
score=f1_score,
parameters={
"analysis_run_id": str(analysis_run_id),
"reference_dataset_id": str(reference_dataset_id),
"iou_threshold": iou_threshold,
"class_name": class_name,
"min_confidence": min_confidence,
"coverage_policy": coverage_summary["mode"],
},
findings={
"matches": evidence.matches,
"false_positives": evidence.false_positives,
"false_negatives": evidence.false_negatives,
"warnings": coverage_warnings + evidence.warnings,
"unsupported_geometry": evidence.unsupported,
"coverage": coverage_summary,
"match_evidence": evidence.match_evidence,
"false_positive_evidence": evidence.false_positive_evidence,
"false_negative_evidence": evidence.false_negative_evidence,
},
metrics={
"precision": precision,
"recall": recall,
"f1": f1_score,
"mean_iou": mean_iou,
"false_positive_count": evidence.false_positives,
"false_negative_count": evidence.false_negatives,
},
)
return {
"status": status,
"quality_check_id": str(quality_check.id),
"analysis_run_id": str(analysis_run_id),
"reference_dataset_id": str(reference_dataset_id),
"candidate_feature_count": len(candidate_geometries),
"reference_feature_count": len(reference_geometries),
"candidate_feature_count_raw": len(raw_candidate_geometries),
"reference_feature_count_raw": reference_raw_count,
"matches": evidence.matches,
"false_positives": evidence.false_positives,
"false_negatives": evidence.false_negatives,
"precision": precision,
"recall": recall,
"f1_score": f1_score,
"mean_iou": mean_iou,
"iou_threshold": iou_threshold,
"warnings": coverage_warnings + evidence.warnings,
"coverage": coverage_summary,
"match_evidence": evidence.match_evidence,
"false_positive_evidence": evidence.false_positive_evidence,
"false_negative_evidence": evidence.false_negative_evidence,
}
@staticmethod
def mask_artifact_path(storage_root: str, project_id: uuid.UUID, analysis_run_id: uuid.UUID, tile_index: int | None, segmentation_id: uuid.UUID) -> str:
tile_folder = f"tile_{tile_index if tile_index is not None else 0}"
return (Path(storage_root) / "masks" / str(project_id) / str(analysis_run_id) / tile_folder / f"mask_{segmentation_id}.png").as_posix()
@staticmethod
def _create_job(
db,
project_id: uuid.UUID,
dataset_id: uuid.UUID,
parameters: dict[str, Any],
existing_job: Job | None = None,
) -> Job:
if existing_job is not None:
# Reuse the queued job so the operator polls one identifier.
existing_job.status = "running"
existing_job.dataset_id = dataset_id
existing_job.input_dataset_id = dataset_id
existing_job.parameters_json = {**(existing_job.parameters_json or {}), **parameters}
existing_job.started_at = SegmentationService._now()
db.add(existing_job)
db.commit()
db.refresh(existing_job)
return existing_job
job = Job(
id=uuid.uuid4(),
job_type="segmentation.run",
status="running",
project_id=project_id,
dataset_id=dataset_id,
input_dataset_id=dataset_id,
parameters_json=parameters,
started_at=SegmentationService._now(),
)
db.add(job)
db.commit()
db.refresh(job)
return job
@staticmethod
def _validate_run_request(db, *, project_id: uuid.UUID, dataset_id: uuid.UUID) -> Dataset:
project = db.get(Project, project_id)
if not project:
raise AppError(code="PROJECT_NOT_FOUND", message="Project not found", status_code=404)
dataset = db.get(Dataset, dataset_id)
if not dataset or dataset.project_id != project_id:
raise AppError(code="DATASET_NOT_FOUND", message="Dataset not found", status_code=404)
if dataset.dataset_type != "raster":
raise AppError(
code="INVALID_DATASET_TYPE",
message="Segmentation requires a raster dataset",
details={"dataset_type": dataset.dataset_type},
status_code=400,
)
return dataset
@staticmethod
def enqueue_segmentation(
db,
project_id: uuid.UUID,
dataset_id: uuid.UUID,
model_id: str,
confidence_threshold: float,
class_filter: list[str] | None = None,
tile_manifest_path: str | None = None,
parameters_json: dict[str, Any] | None = None,
) -> Job:
"""Accept a segmentation run for background execution."""
SegmentationService._validate_run_request(db, project_id=project_id, dataset_id=dataset_id)
job = Job(
id=uuid.uuid4(),
job_type="segmentation.run",
status="queued",
project_id=project_id,
dataset_id=dataset_id,
input_dataset_id=dataset_id,
parameters_json={
"project_id": str(project_id),
"dataset_id": str(dataset_id),
"model_id": model_id,
"confidence_threshold": confidence_threshold,
"class_filter": class_filter or [],
"tile_manifest_path": tile_manifest_path,
"parameters_json": dict(parameters_json or {}),
},
)
db.add(job)
db.commit()
db.refresh(job)
return job
@staticmethod
def _create_analysis_run(db, project_id, dataset_id, job_id, model, parameters: dict[str, Any]) -> AnalysisRun:
analysis_run = AnalysisRun(
id=uuid.uuid4(),
project_id=project_id,
dataset_id=dataset_id,
job_id=job_id,
analysis_type="segmentation",
status="running",
model_name=model.model_id,
model_version=model.version,
parameters_json=parameters,
started_at=SegmentationService._now(),
)
db.add(analysis_run)
db.commit()
db.refresh(analysis_run)
return analysis_run
@staticmethod
def _mark_failed(db, analysis_run: AnalysisRun, job: Job, code: str, message: str) -> None:
result = {"error_code": code, "message": message, "segmentation_count": 0}
analysis_run.status = "failed"
analysis_run.finished_at = SegmentationService._now()
analysis_run.error_message = message
analysis_run.result_json = result
job.status = "failed"
job.finished_at = analysis_run.finished_at
job.error_message = message
job.result_json = result
db.add(analysis_run)
db.add(job)
db.commit()
db.refresh(analysis_run)
db.refresh(job)
@staticmethod
def _mark_success(db, analysis_run: AnalysisRun, job: Job, segmentation_count: int, extra_result: dict[str, Any] | None = None) -> None:
result = {"segmentation_count": segmentation_count}
if extra_result:
result.update(extra_result)
analysis_run.status = "success"
analysis_run.finished_at = SegmentationService._now()
analysis_run.result_json = result
job.status = "success"
job.finished_at = analysis_run.finished_at
job.result_json = result
db.add(analysis_run)
db.add(job)
db.commit()
db.refresh(analysis_run)
db.refresh(job)
@staticmethod
def _run_configured_segmentation(
db,
project_id: uuid.UUID,
dataset_id: uuid.UUID,
analysis_run: AnalysisRun,
job: Job,
model_name: str,
model_version: str | None,
tile_manifest_path: str | None,
confidence_threshold: float,
class_filter: list[str],
settings: Settings,
yolo_seg_adapter_class: type[YoloSegmentationAdapter],
sam_adapter_class: type[SamSegmentationAdapter],
) -> tuple[list[Segmentation], dict[str, Any]]:
manifest = DetectionService._load_tile_manifest(tile_manifest_path, settings.yolo_max_tiles)
if model_name == settings.sam_model_id:
model_path = Path(settings.sam_model_path or "").expanduser()
allowed_frameworks = ("ultralytics/sam", "sam", "ultralytics", "pytorch")
adapter = sam_adapter_class(settings)
else:
model_path = Path(settings.yolo_seg_model_path or "").expanduser()
allowed_frameworks = ("ultralytics/pytorch", "ultralytics", "pytorch")
adapter = yolo_seg_adapter_class(settings)
runtime_model_provenance = RuntimeModelProvenanceService.validate_for_production_runtime(
db=db,
model_path=model_path,
model_id=model_name,
task_type="segmentation",
expected_model_version=model_version,
allowed_frameworks=allowed_frameworks,
)
SegmentationService._attach_runtime_model_provenance(
analysis_run,
job,
runtime_model_provenance,
)
model = adapter.load_model(model_path)
allowed_classes = {DetectionService._canonical_class_name(value) for value in class_filter if DetectionService._canonical_class_name(value)}
manifest_crs = DetectionService._require_manifest_crs(manifest)
candidates: list[dict[str, Any]] = []
for tile in manifest["tiles"]:
tile_path = DetectionService._resolve_tile_path(tile, Path(tile_manifest_path or "").expanduser())
for raw in adapter.predict_tile(model, tile_path, confidence_threshold):
model_class_name = str(raw.get("class_name") or "").strip()
class_name = DetectionService._canonical_class_name(model_class_name)
confidence = raw.get("confidence")
confidence = float(confidence) if confidence is not None else None
if allowed_classes and class_name not in allowed_classes:
continue
if confidence is not None and confidence < confidence_threshold:
continue
points = raw.get("points")
if not isinstance(points, list) or len(points) < 3:
continue
geometry = pixel_points_to_epsg4326_polygon(points=points, tile=tile, crs=tile.get("crs") or manifest_crs)
properties = dict(raw.get("properties") or {})
if model_class_name and model_class_name != class_name:
properties.setdefault("model_class_name", model_class_name)
candidates.append(
{
"class_name": class_name,
"confidence": confidence if confidence is not None else 0.0,
"reported_confidence": confidence,
"geometry": geometry,
"bbox": raw.get("bbox"),
"source_tile_path": str(tile_path),
"tile_index": tile.get("index"),
"properties": {**properties, "tile_index": tile.get("index")},
}
)
filtered_candidates = DetectionService._suppress_duplicate_candidates(
candidates,
iou_threshold=float(settings.segmentation_duplicate_iou_threshold),
)
persisted: list[Segmentation] = []
for candidate in filtered_candidates:
geometry = candidate["geometry"]
if isinstance(geometry, Polygon):
geometry = MultiPolygon([geometry])
bbox = candidate.get("bbox")
bbox_json = None
if isinstance(bbox, list) and len(bbox) == 4:
bbox_json = {
"x_min": float(bbox[0]),
"y_min": float(bbox[1]),
"x_max": float(bbox[2]),
"y_max": float(bbox[3]),
}
segmentation = Segmentation(
id=uuid.uuid4(),
project_id=project_id,
dataset_id=dataset_id,
analysis_run_id=analysis_run.id,
job_id=job.id,
model_name=model_name,
model_version=model_version,
class_name=candidate["class_name"],
confidence=candidate["reported_confidence"],
geometry=from_shape(geometry, srid=4326),
bbox_json=bbox_json,
area_m2=SegmentationService._geodesic_area_m2(geometry),
mask_path=None,
source_tile_path=candidate["source_tile_path"],
tile_index=candidate["tile_index"] if isinstance(candidate["tile_index"], int) else None,
properties_json=candidate["properties"],
provenance_json={
"inference": "local",
"model_id": model_name,
"tile_manifest_path": str(Path(tile_manifest_path or "").expanduser()),
"tile_index": candidate["tile_index"],
"device": settings.yolo_device,
"runtime_model_provenance": runtime_model_provenance.as_dict(),
},
)
db.add(segmentation)
persisted.append(segmentation)
db.commit()
for segmentation in persisted:
db.refresh(segmentation)
return persisted, {
"raw_segmentation_count": len(candidates),
"suppressed_segmentation_count": len(candidates) - len(filtered_candidates),
"duplicate_iou_threshold": float(settings.segmentation_duplicate_iou_threshold),
"tile_manifest_path": str(Path(tile_manifest_path or "").expanduser()),
"runtime_model_provenance": runtime_model_provenance.as_dict(),
}
@staticmethod
def _attach_runtime_model_provenance(
analysis_run: AnalysisRun,
job: Job,
provenance: RuntimeModelProvenance,
) -> None:
"""Record immutable model evidence with a configured segmentation run."""
evidence = provenance.as_dict()
analysis_parameters = dict(analysis_run.parameters_json or {})
analysis_parameters["runtime_model_provenance"] = evidence
analysis_run.parameters_json = analysis_parameters
job_parameters = dict(job.parameters_json or {})
job_parameters["runtime_model_provenance"] = evidence
job.parameters_json = job_parameters
@staticmethod
def _geodesic_area_m2(geometry: MultiPolygon | Polygon) -> float | None:
try:
from pyproj import Geod
area, _ = Geod(ellps="WGS84").geometry_area_perimeter(geometry)
return abs(float(area))
except Exception:
return None
@staticmethod
def _persist_fixture_segmentations(
db,
project_id: uuid.UUID,
dataset_id: uuid.UUID,
analysis_run: AnalysisRun,
job: Job,
model_name: str,
model_version: str | None,
raw_segmentations: Any,
confidence_threshold: float,
class_filter: list[str],
settings: Settings,
) -> list[Segmentation]:
if not isinstance(raw_segmentations, list):
raise AppError(code="INVALID_FIXTURE_SEGMENTATIONS", message="fixture_segmentations must be a list", status_code=400)
adapter = FixtureSegmentationAdapter()
adapter_results = adapter.segment(raw_segmentations)
if len(adapter_results) != len(raw_segmentations):
raise AppError(code="INVALID_FIXTURE_SEGMENTATION", message="Each fixture segmentation must be an object", status_code=400)
persisted: list[Segmentation] = []
allowed_classes = set(class_filter)
for raw in adapter_results:
class_name = raw.class_name
confidence = raw.confidence
if allowed_classes and class_name not in allowed_classes:
continue
if confidence is not None and confidence < confidence_threshold:
continue
if not isinstance(raw.geometry, dict):
raise AppError(code="INVALID_FIXTURE_SEGMENTATION", message="Fixture segmentation geometry is required", status_code=400)
geometry = SegmentationService._validated_multipolygon(raw.geometry)
segmentation_id = uuid.uuid4()
mask_path = raw.mask_path or SegmentationService.mask_artifact_path(
settings.storage_root,
project_id,
analysis_run.id,
raw.tile_index,
segmentation_id,
)
segmentation = Segmentation(
id=segmentation_id,
project_id=project_id,
dataset_id=dataset_id,
analysis_run_id=analysis_run.id,
job_id=job.id,
model_name=model_name,
model_version=model_version,
class_name=class_name,
confidence=confidence,
geometry=from_shape(geometry, srid=4326),
bbox_json=raw.bbox_json,
area_m2=raw.area_m2,
mask_path=mask_path,
source_tile_path=raw.source_tile_path,
tile_index=raw.tile_index,
properties_json=raw.properties_json,
provenance_json={**dict(raw.provenance_json or {}), "fixture_mode": True},
)
db.add(segmentation)
persisted.append(segmentation)
db.commit()
for segmentation in persisted:
db.refresh(segmentation)
return persisted
@staticmethod
def _validated_multipolygon(geometry_payload: dict[str, Any]) -> MultiPolygon:
try:
geometry = shape(geometry_payload)
except Exception as exc:
raise AppError(code="INVALID_FIXTURE_GEOMETRY", message="Fixture segmentation geometry must be valid GeoJSON", status_code=400) from exc
if geometry.is_empty:
raise AppError(code="INVALID_FIXTURE_GEOMETRY", message="Fixture segmentation geometry must not be empty", status_code=400)
if not geometry.is_valid:
geometry = make_valid(geometry)
if geometry.is_empty or not geometry.is_valid:
raise AppError(code="INVALID_FIXTURE_GEOMETRY", message="Fixture segmentation geometry must be valid", status_code=400)
if isinstance(geometry, Polygon):
geometry = MultiPolygon([geometry])
if not isinstance(geometry, MultiPolygon):
raise AppError(code="INVALID_FIXTURE_GEOMETRY", message="Fixture segmentation geometry must be Polygon or MultiPolygon", status_code=400)
if geometry.area <= 0:
raise AppError(code="INVALID_FIXTURE_GEOMETRY", message="Fixture segmentation geometry must have positive area", status_code=400)
return geometry
@staticmethod
def _query_segmentation_rows(
db,
*,
analysis_run_id: uuid.UUID | None = None,
dataset_id: uuid.UUID | None = None,
class_name: str | None = None,
min_confidence: float | None = None,
) -> list[Segmentation]:
query = db.query(Segmentation)
if analysis_run_id is not None:
query = query.filter(Segmentation.analysis_run_id == analysis_run_id)
if dataset_id is not None:
query = query.filter(Segmentation.dataset_id == dataset_id)
if class_name:
query = query.filter(Segmentation.class_name == class_name)
if min_confidence is not None:
query = query.filter(Segmentation.confidence >= min_confidence)
# One transaction timestamp is shared by every row in a run, so
# ordering by it alone leaves the row order — and therefore the QA
# score — undefined. See DetectionService._query_detection_rows.
return query.order_by(
Segmentation.confidence.desc(),
Segmentation.created_at.desc(),
Segmentation.id.asc(),
).all()
@staticmethod
def _segmentation_properties(segmentation: Segmentation) -> dict[str, Any]:
return {
"segmentation_id": str(segmentation.id),
"class_name": segmentation.class_name,
"confidence": segmentation.confidence,
"area_m2": segmentation.area_m2,
"model_name": segmentation.model_name,
"model_version": segmentation.model_version,
"analysis_run_id": str(segmentation.analysis_run_id) if segmentation.analysis_run_id else None,
"dataset_id": str(segmentation.dataset_id) if segmentation.dataset_id else None,
"job_id": str(segmentation.job_id) if segmentation.job_id else None,
"source_tile_path": segmentation.source_tile_path,
"tile_index": segmentation.tile_index,
"mask_path": segmentation.mask_path,
"bbox_json": segmentation.bbox_json,
"provenance_json": segmentation.provenance_json,
}