File size: 3,110 Bytes
6f2f547
1499e36
2c3171a
4d26ef2
778da92
 
4d26ef2
 
778da92
1499e36
 
6f2f547
 
1499e36
6f2f547
 
1499e36
6f2f547
4d26ef2
1499e36
 
 
4d26ef2
 
1499e36
4d26ef2
 
 
 
5a246f7
4d26ef2
 
778da92
6f2f547
 
 
1499e36
6f2f547
 
1499e36
6f2f547
1499e36
6f2f547
4d26ef2
e897c81
6f2f547
1499e36
4d26ef2
6f2f547
 
e897c81
4d26ef2
6f2f547
1499e36
6f2f547
6c70a52
 
 
4d26ef2
e897c81
3041467
adbb63f
6c70a52
 
 
1499e36
6c70a52
 
 
adbb63f
6c70a52
 
 
adbb63f
6c70a52
adbb63f
6c70a52
 
 
 
4d26ef2
 
 
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
"""handle geo-referenced raster images"""
from affine import Affine
from numpy import ndarray as np_ndarray

from src import app_logger
from src.utilities.type_hints import list_float, tuple_float, dict_str_int


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.error(f"exception:{e}, check https://github.com/rasterio/affine project for updates")
        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:
        app_logger.error(f"e_shape_band:{e_shape_band}.")
        raise e_shape_band