220 lines
9.7 KiB
Python
220 lines
9.7 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import datetime, timezone
|
|
from typing import Any
|
|
from uuid import UUID
|
|
|
|
from geoalchemy2.shape import to_shape
|
|
from shapely.geometry import mapping
|
|
from shapely.geometry.base import BaseGeometry
|
|
from shapely.validation import make_valid
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.core.errors import AppError
|
|
from app.models import Dataset, VectorFeature
|
|
from app.schemas.analysis import ChangeDetectionSummary
|
|
from app.services.vector_operations_service import VectorOperationsService
|
|
|
|
|
|
class ChangeDetectionService:
|
|
SUPPORTED_GEOMETRY_TYPES = {"Polygon", "MultiPolygon"}
|
|
|
|
@staticmethod
|
|
def compare_vector_datasets(
|
|
db: Session,
|
|
*,
|
|
project_id: UUID,
|
|
source_dataset_id: UUID,
|
|
target_dataset_id: UUID,
|
|
iou_threshold: float = 0.8,
|
|
include_unchanged: bool = True,
|
|
) -> ChangeDetectionSummary:
|
|
if source_dataset_id == target_dataset_id:
|
|
raise AppError(code="INVALID_PARAMETERS", message="Source and target datasets must differ", status_code=400)
|
|
if iou_threshold < 0 or iou_threshold > 1:
|
|
raise AppError(code="INVALID_PARAMETERS", message="iou_threshold must be between 0 and 1", status_code=400)
|
|
|
|
source_dataset = ChangeDetectionService._get_project_vector_dataset(db, source_dataset_id, project_id, "Source")
|
|
target_dataset = ChangeDetectionService._get_project_vector_dataset(db, target_dataset_id, project_id, "Target")
|
|
|
|
source_features, source_warnings = ChangeDetectionService._load_features(db, source_dataset)
|
|
target_features, target_warnings = ChangeDetectionService._load_features(db, target_dataset)
|
|
|
|
if not source_features:
|
|
raise AppError(code="EMPTY_VECTOR_DATASET", message="Source dataset has no comparable vector features", status_code=422)
|
|
if not target_features:
|
|
raise AppError(code="EMPTY_VECTOR_DATASET", message="Target dataset has no comparable vector features", status_code=422)
|
|
|
|
matched_target_indices: set[int] = set()
|
|
unchanged: list[dict[str, Any]] = []
|
|
removed: list[dict[str, Any]] = []
|
|
|
|
for source_feature in source_features:
|
|
best_iou = 0.0
|
|
best_index: int | None = None
|
|
for target_index, target_feature in enumerate(target_features):
|
|
if target_index in matched_target_indices:
|
|
continue
|
|
candidate_iou = ChangeDetectionService._iou(source_feature["geometry"], target_feature["geometry"])
|
|
if candidate_iou > best_iou:
|
|
best_iou = candidate_iou
|
|
best_index = target_index
|
|
|
|
if best_index is not None and best_iou >= iou_threshold:
|
|
matched_target_indices.add(best_index)
|
|
if include_unchanged:
|
|
unchanged.append(
|
|
ChangeDetectionService._feature(
|
|
geometry=source_feature["geometry"],
|
|
change_type="unchanged",
|
|
source_dataset_id=source_dataset_id,
|
|
target_dataset_id=target_dataset_id,
|
|
source_feature_id=source_feature["feature_id"],
|
|
target_feature_id=target_features[best_index]["feature_id"],
|
|
iou=best_iou,
|
|
properties=source_feature["properties"],
|
|
)
|
|
)
|
|
else:
|
|
removed.append(
|
|
ChangeDetectionService._feature(
|
|
geometry=source_feature["geometry"],
|
|
change_type="removed",
|
|
source_dataset_id=source_dataset_id,
|
|
target_dataset_id=target_dataset_id,
|
|
source_feature_id=source_feature["feature_id"],
|
|
target_feature_id=None,
|
|
iou=best_iou if best_iou > 0 else None,
|
|
properties=source_feature["properties"],
|
|
)
|
|
)
|
|
|
|
added = [
|
|
ChangeDetectionService._feature(
|
|
geometry=target_feature["geometry"],
|
|
change_type="added",
|
|
source_dataset_id=source_dataset_id,
|
|
target_dataset_id=target_dataset_id,
|
|
source_feature_id=None,
|
|
target_feature_id=target_feature["feature_id"],
|
|
iou=None,
|
|
properties=target_feature["properties"],
|
|
)
|
|
for target_index, target_feature in enumerate(target_features)
|
|
if target_index not in matched_target_indices
|
|
]
|
|
|
|
geojson_features = added + removed + unchanged
|
|
return ChangeDetectionSummary(
|
|
source_dataset_id=source_dataset_id,
|
|
target_dataset_id=target_dataset_id,
|
|
source_feature_count=len(source_features),
|
|
target_feature_count=len(target_features),
|
|
added_count=len(added),
|
|
removed_count=len(removed),
|
|
unchanged_count=len(unchanged) if include_unchanged else len(matched_target_indices),
|
|
iou_threshold=iou_threshold,
|
|
warnings=source_warnings + target_warnings,
|
|
generated_at=datetime.now(timezone.utc),
|
|
geojson={"type": "FeatureCollection", "features": geojson_features},
|
|
)
|
|
|
|
@staticmethod
|
|
def _get_project_vector_dataset(db: Session, dataset_id: UUID, project_id: UUID, label: str) -> Dataset:
|
|
dataset = db.get(Dataset, dataset_id)
|
|
if not dataset:
|
|
raise AppError(code="DATASET_NOT_FOUND", message=f"{label} dataset not found", status_code=404)
|
|
if dataset.project_id != project_id:
|
|
raise AppError(code="INVALID_DATASET_SCOPE", message=f"{label} dataset does not belong to this project", status_code=400)
|
|
VectorOperationsService._require_vector_dataset(dataset)
|
|
return dataset
|
|
|
|
@staticmethod
|
|
def _load_features(db: Session, dataset: Dataset) -> tuple[list[dict[str, Any]], list[str]]:
|
|
rows = db.query(VectorFeature).filter(VectorFeature.dataset_id == dataset.id).all()
|
|
warnings: list[str] = []
|
|
if rows:
|
|
return [ChangeDetectionService._row_to_feature(row) for row in rows], warnings
|
|
|
|
warnings.append(f"Dataset {dataset.id} has no persisted vector_features; falling back to stored GeoJSON artifact")
|
|
_payload, raw_features = VectorOperationsService._load_dataset_payload(dataset)
|
|
extracted = VectorOperationsService._extract_geometries(raw_features)
|
|
return [
|
|
ChangeDetectionService._raw_feature_to_feature(index, raw_feature, geometry)
|
|
for index, (raw_feature, geometry) in enumerate(extracted)
|
|
], warnings
|
|
|
|
@staticmethod
|
|
def _row_to_feature(row: VectorFeature) -> dict[str, Any]:
|
|
geometry = ChangeDetectionService._valid_comparable_geometry(to_shape(row.geometry))
|
|
return {
|
|
"feature_id": str(row.source_feature_id or row.id),
|
|
"properties": dict(row.properties_json or {}),
|
|
"geometry": geometry,
|
|
}
|
|
|
|
@staticmethod
|
|
def _raw_feature_to_feature(index: int, raw_feature: dict[str, Any], geometry: BaseGeometry) -> dict[str, Any]:
|
|
properties = raw_feature.get("properties") if isinstance(raw_feature.get("properties"), dict) else {}
|
|
source_id = raw_feature.get("id") or properties.get("id") or properties.get("source_feature_id") or str(index)
|
|
return {
|
|
"feature_id": str(source_id),
|
|
"properties": dict(properties),
|
|
"geometry": ChangeDetectionService._valid_comparable_geometry(geometry),
|
|
}
|
|
|
|
@staticmethod
|
|
def _valid_comparable_geometry(geometry: BaseGeometry) -> BaseGeometry:
|
|
if geometry.is_empty:
|
|
raise AppError(code="INVALID_GEOMETRY", message="Empty geometry cannot be compared", 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_GEOMETRY", message="Geometry cannot be repaired for comparison", status_code=400)
|
|
if geometry.geom_type not in ChangeDetectionService.SUPPORTED_GEOMETRY_TYPES:
|
|
raise AppError(
|
|
code="UNSUPPORTED_GEOMETRY",
|
|
message="Change detection supports Polygon and MultiPolygon geometries only",
|
|
details={"geometry_type": geometry.geom_type},
|
|
status_code=422,
|
|
)
|
|
return geometry
|
|
|
|
@staticmethod
|
|
def _iou(left: BaseGeometry, right: BaseGeometry) -> float:
|
|
if left.area <= 0 or right.area <= 0:
|
|
return 0.0
|
|
intersection = left.intersection(right)
|
|
if intersection.is_empty:
|
|
return 0.0
|
|
union_area = left.area + right.area - intersection.area
|
|
if union_area <= 0:
|
|
return 0.0
|
|
return float(intersection.area / union_area)
|
|
|
|
@staticmethod
|
|
def _feature(
|
|
*,
|
|
geometry: BaseGeometry,
|
|
change_type: str,
|
|
source_dataset_id: UUID,
|
|
target_dataset_id: UUID,
|
|
source_feature_id: str | None,
|
|
target_feature_id: str | None,
|
|
iou: float | None,
|
|
properties: dict[str, Any],
|
|
) -> dict[str, Any]:
|
|
return {
|
|
"type": "Feature",
|
|
"geometry": mapping(geometry),
|
|
"properties": {
|
|
**properties,
|
|
"change_type": change_type,
|
|
"source_dataset_id": str(source_dataset_id),
|
|
"target_dataset_id": str(target_dataset_id),
|
|
"source_feature_id": source_feature_id,
|
|
"target_feature_id": target_feature_id,
|
|
"iou": iou,
|
|
},
|
|
}
|