alessandro trinca tornidor
[refactor] rename samgis module tosamgis_lisa_on_cuda, bump to version 1.3.0, update samgis and lisa_on_cuda dependencies
fcb8c81
raw
history blame
3.62 kB
"""handle geo-referenced raster images"""
from affine import Affine
from numpy import ndarray as np_ndarray
from samgis_core.utilities.type_hints import list_float, tuple_float, dict_str_int
from samgis_lisa_on_cuda import app_logger
def load_affine_transformation_from_matrix(matrix_source_coefficients: list_float) -> Affine:
"""
Wrapper for rasterio.Affine.from_gdal() method
Args:
matrix_source_coefficients: 6 floats ordered by GDAL.
Returns:
Affine transform
"""
if len(matrix_source_coefficients) != 6:
raise ValueError(f"Expected 6 coefficients, found {len(matrix_source_coefficients)}; "
f"argument type: {type(matrix_source_coefficients)}.")
try:
a, d, b, e, c, f = (float(x) for x in matrix_source_coefficients)
center = tuple.__new__(Affine, [a, b, c, d, e, f, 0.0, 0.0, 1.0])
return center * Affine.translation(-0.5, -0.5)
except Exception as e:
app_logger.exception(f"exception:{e}, check updates on https://github.com/rasterio/affine",
extra=e,
stack_info=True, exc_info=True)
raise e
def get_affine_transform_from_gdal(matrix_source_coefficients: list_float or tuple_float) -> Affine:
"""wrapper for rasterio Affine from_gdal method
Args:
matrix_source_coefficients: 6 floats ordered by GDAL.
Returns:
Affine transform
"""
return Affine.from_gdal(*matrix_source_coefficients)
def get_vectorized_raster_as_geojson(mask: np_ndarray, transform: tuple_float) -> dict_str_int:
"""
Get shapes and values of connected regions in a dataset or array
Args:
mask: numpy mask
transform: tuple of float to transform into an Affine transform
Returns:
dict containing the output geojson and the predictions number
"""
try:
from rasterio.features import shapes
from geopandas import GeoDataFrame
app_logger.debug(f"matrix to consume with rasterio.shapes: {type(transform)}, {transform}.")
# old value for mask => band != 0
shapes_generator = ({
'properties': {'raster_val': v}, 'geometry': s}
for i, (s, v)
# instead of `enumerate(shapes(mask, mask=(band != 0), transform=rio_src.transform))`
# use mask=None to avoid using source
in enumerate(shapes(mask, mask=None, transform=transform))
)
app_logger.info("created shapes_generator, transform it to a polygon list...")
shapes_list = list(shapes_generator)
app_logger.info(f"created {len(shapes_list)} polygons.")
gpd_polygonized_raster = GeoDataFrame.from_features(shapes_list, crs="EPSG:3857")
app_logger.info("created a GeoDataFrame, export to geojson...")
geojson = gpd_polygonized_raster.to_json(to_wgs84=True)
app_logger.info("created geojson, preparing API response...")
return {
"geojson": geojson,
"n_shapes_geojson": len(shapes_list)
}
except Exception as e_shape_band:
try:
app_logger.error(f"mask type:{type(mask)}.")
app_logger.error(f"transform type:{type(transform)}, {transform}.")
app_logger.error(f"mask shape:{mask.shape}, dtype:{mask.dtype}.")
except Exception as e_shape_dtype:
app_logger.exception(f"mask shape or dtype not found:{e_shape_dtype}.", exc_info=True)
app_logger.exception(f"e_shape_band:{e_shape_band}.", exc_info=True)
raise e_shape_band