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
| """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 | |