File size: 3,607 Bytes
0914710 c311b69 fcb8c81 0914710 c311b69 0914710 c311b69 0914710 c311b69 0914710 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 |
"""handle geo-referenced raster images"""
from affine import Affine
from numpy import ndarray as np_ndarray
from samgis_core.utilities.type_hints import ListFloat, TupleFloat, DictStrInt
from samgis_lisa_on_cuda import app_logger
def load_affine_transformation_from_matrix(matrix_source_coefficients: ListFloat) -> 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: ListFloat or TupleFloat) -> 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: TupleFloat) -> DictStrInt:
"""
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
|