from __future__ import annotations from typing import Any from pyproj import Transformer from shapely.geometry import Polygon from app.core.errors import AppError def _require_source_crs(crs: str | None, tile: dict[str, Any]) -> str: """Resolve the CRS a pixel coordinate is measured in, or fail. Falling back to EPSG:4326 turned a missing manifest field into geometry that sits in the wrong place while still looking like a valid polygon on the map. A georeferenced result without a known CRS is not a result. """ for candidate in (crs, tile.get("crs"), tile.get("source_crs")): if isinstance(candidate, str) and candidate.strip(): return candidate.strip() raise AppError( code="DETECTION_TILE_CRS_REQUIRED", message="Georeferencing a tile requires explicit CRS metadata", status_code=422, ) def pixel_bbox_to_epsg4326_polygon(bbox: list[float], tile: dict[str, Any], crs: str | None = None) -> Polygon: if len(bbox) != 4: raise AppError(code="DETECTION_INVALID_BBOX", message="YOLO detection bbox must contain four pixel coordinates", status_code=422) x_min, y_min, x_max, y_max = [float(value) for value in bbox] if x_max <= x_min or y_max <= y_min: raise AppError(code="DETECTION_INVALID_BBOX", message="YOLO detection bbox must have positive width and height", status_code=422) transform = tile.get("transform") if isinstance(transform, list) and len(transform) >= 6: corners = [ _apply_gdal_transform(transform, x_min, y_min), _apply_gdal_transform(transform, x_max, y_min), _apply_gdal_transform(transform, x_max, y_max), _apply_gdal_transform(transform, x_min, y_max), _apply_gdal_transform(transform, x_min, y_min), ] else: corners = _corners_from_bounds(bbox=[x_min, y_min, x_max, y_max], tile=tile) source_crs = _require_source_crs(crs, tile) if str(source_crs).upper() not in {"EPSG:4326", "4326"}: transformer = Transformer.from_crs(source_crs, "EPSG:4326", always_xy=True) corners = [transformer.transform(x, y) for x, y in corners] polygon = Polygon(corners) if polygon.is_empty or not polygon.is_valid: raise AppError(code="DETECTION_INVALID_GEOMETRY", message="Georeferenced detection geometry is invalid", status_code=422) return polygon def pixel_points_to_epsg4326_polygon(points: list[list[float]], tile: dict[str, Any], crs: str | None = None) -> Polygon: if not isinstance(points, list) or len(points) < 3: raise AppError( code="SEGMENTATION_INVALID_MASK", message="Segmentation mask polygon must contain at least three pixel points", status_code=422, ) try: pixel_points = [(float(point[0]), float(point[1])) for point in points] except (TypeError, ValueError, IndexError) as exc: raise AppError( code="SEGMENTATION_INVALID_MASK", message="Segmentation mask polygon points must be numeric [x, y] pairs", status_code=422, ) from exc transform = tile.get("transform") if isinstance(transform, list) and len(transform) >= 6: coordinates = [_apply_gdal_transform(transform, x, y) for x, y in pixel_points] else: coordinates = [_project_pixel_with_bounds(tile, x, y) for x, y in pixel_points] source_crs = _require_source_crs(crs, tile) if str(source_crs).upper() not in {"EPSG:4326", "4326"}: transformer = Transformer.from_crs(source_crs, "EPSG:4326", always_xy=True) coordinates = [transformer.transform(x, y) for x, y in coordinates] if coordinates[0] != coordinates[-1]: coordinates.append(coordinates[0]) polygon = Polygon(coordinates) if not polygon.is_valid: from shapely.validation import make_valid repaired = make_valid(polygon) polygon = _largest_polygon(repaired) if polygon is None or polygon.is_empty or not polygon.is_valid or polygon.area <= 0: raise AppError( code="SEGMENTATION_INVALID_GEOMETRY", message="Georeferenced segmentation geometry is invalid", status_code=422, ) return polygon def _largest_polygon(geometry: Any) -> Polygon | None: if isinstance(geometry, Polygon): return geometry candidates = [geom for geom in getattr(geometry, "geoms", []) if isinstance(geom, Polygon) and geom.area > 0] if not candidates: return None return max(candidates, key=lambda geom: geom.area) def _project_pixel_with_bounds(tile: dict[str, Any], px: float, py: float) -> tuple[float, float]: bounds = tile.get("bounds") pixel_window = tile.get("pixel_window") if not (isinstance(bounds, list) and len(bounds) == 4 and isinstance(pixel_window, list) and len(pixel_window) == 4): raise AppError( code="DETECTION_TILE_MANIFEST_INVALID", message="Tile manifest entries require transform or bounds plus pixel_window for georeferencing", status_code=422, ) left, bottom, right, top = [float(value) for value in bounds] _, _, width, height = [float(value) for value in pixel_window] if width <= 0 or height <= 0: raise AppError(code="DETECTION_TILE_MANIFEST_INVALID", message="Tile pixel_window must have positive size", status_code=422) return (left + (px / width) * (right - left), top - (py / height) * (top - bottom)) def _apply_gdal_transform(transform: list[float], x: float, y: float) -> tuple[float, float]: c, a, b, f, d, e = [float(value) for value in transform[:6]] return (a * x + b * y + c, d * x + e * y + f) def _corners_from_bounds(bbox: list[float], tile: dict[str, Any]) -> list[tuple[float, float]]: bounds = tile.get("bounds") pixel_window = tile.get("pixel_window") if not (isinstance(bounds, list) and len(bounds) == 4 and isinstance(pixel_window, list) and len(pixel_window) == 4): raise AppError( code="DETECTION_TILE_MANIFEST_INVALID", message="Tile manifest entries require transform or bounds plus pixel_window for georeferencing", status_code=422, ) x_min, y_min, x_max, y_max = bbox left, bottom, right, top = [float(value) for value in bounds] _, _, width, height = [float(value) for value in pixel_window] if width <= 0 or height <= 0: raise AppError(code="DETECTION_TILE_MANIFEST_INVALID", message="Tile pixel_window must have positive size", status_code=422) def project(px: float, py: float) -> tuple[float, float]: x = left + (px / width) * (right - left) y = top - (py / height) * (top - bottom) return (x, y) return [ project(x_min, y_min), project(x_max, y_min), project(x_max, y_max), project(x_min, y_max), project(x_min, y_min), ]