Jens
2cdf9c99c6
Managed validation / Managed repository validation (pull_request) Successful in 1m46s
GeoIntel release gates / Compile, test, contracts and builds (pull_request) Successful in 1m51s
GeoIntel release gates / Python and npm vulnerability policy (pull_request) Successful in 20s
GeoIntel release gates / Production AI image, SBOM and container scan (pull_request) Successful in 15m3s
GeoIntel release gates / Deploy exact gated revision to Unraid (pull_request) Skipped
98 lines
3.5 KiB
Python
98 lines
3.5 KiB
Python
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
from app.core.config import get_settings
|
|
from app.core.errors import AppError
|
|
|
|
|
|
_DTYPE_BYTES = {
|
|
"uint8": 1,
|
|
"int8": 1,
|
|
"uint16": 2,
|
|
"int16": 2,
|
|
"uint32": 4,
|
|
"int32": 4,
|
|
"float32": 4,
|
|
"uint64": 8,
|
|
"int64": 8,
|
|
"float64": 8,
|
|
"complex64": 8,
|
|
"complex128": 16,
|
|
}
|
|
|
|
|
|
def _import_rasterio():
|
|
import importlib
|
|
|
|
rasterio = importlib.import_module("rasterio")
|
|
errors = importlib.import_module("rasterio.errors")
|
|
return rasterio, errors
|
|
|
|
|
|
def extract_raster_metadata(path: str) -> dict:
|
|
try:
|
|
rasterio, errors = _import_rasterio()
|
|
except Exception as exc: # pragma: no cover - exercised via API-level fallback tests
|
|
raise AppError(
|
|
code="RASTER_PROCESSING_UNAVAILABLE",
|
|
message="Raster processing unavailable. Install rasterio and GDAL-compatible drivers to enable raster metadata extraction.",
|
|
status_code=503,
|
|
) from exc
|
|
|
|
dataset_path = Path(path)
|
|
try:
|
|
with rasterio.open(dataset_path) as dataset:
|
|
settings = get_settings()
|
|
width = int(dataset.width)
|
|
height = int(dataset.height)
|
|
band_count = int(dataset.count)
|
|
pixel_count = width * height
|
|
dtype_bytes = max((_DTYPE_BYTES.get(str(dtype), 16) for dtype in dataset.dtypes), default=16)
|
|
decoded_bytes = pixel_count * band_count * dtype_bytes
|
|
if (
|
|
width <= 0
|
|
or height <= 0
|
|
or band_count <= 0
|
|
or pixel_count > settings.max_raster_pixels
|
|
or band_count > settings.max_raster_bands
|
|
or decoded_bytes > settings.max_decoded_raster_mb * 1024 * 1024
|
|
):
|
|
raise AppError(
|
|
code="RASTER_RESOURCE_LIMIT_EXCEEDED",
|
|
message="Raster dimensions or decoded size exceed the configured processing budget.",
|
|
details={
|
|
"width": width,
|
|
"height": height,
|
|
"band_count": band_count,
|
|
"pixel_count": pixel_count,
|
|
"estimated_decoded_bytes": decoded_bytes,
|
|
},
|
|
status_code=413,
|
|
)
|
|
nodata = dataset.nodata
|
|
if isinstance(nodata, (list, tuple)):
|
|
nodata_value = [None if value is None else float(value) for value in nodata]
|
|
else:
|
|
nodata_value = None if nodata is None else float(nodata)
|
|
|
|
transform = dataset.transform.to_gdal() if hasattr(dataset, "transform") else None
|
|
return {
|
|
"driver": dataset.driver,
|
|
"width": width,
|
|
"height": height,
|
|
"band_count": band_count,
|
|
"crs": str(dataset.crs) if dataset.crs else None,
|
|
"bounds": list(dataset.bounds),
|
|
"resolution": list(dataset.res),
|
|
"dtype": list(dataset.dtypes),
|
|
"nodata": nodata_value,
|
|
"transform": list(transform) if transform is not None else None,
|
|
}
|
|
except AppError:
|
|
raise
|
|
except Exception as exc:
|
|
if isinstance(exc, errors.RasterioIOError):
|
|
raise AppError(code="INVALID_RASTER", message="Uploaded raster file is invalid", status_code=400) from exc
|
|
raise AppError(code="RASTER_METADATA_ERROR", message="Unable to read raster metadata", status_code=400) from exc
|