[refactor] refactor and remove unused code, remove unused deps

requirements.txt

@@ -2,10 +2,10 @@ aws-lambda-powertools
 awslambdaric
 bson
 geopandas
-httpx
 jmespath
 numpy
 onnxruntime
 opencv-python
 pillow
-rasterio
+rasterio
+requests

requirements_dev.txt

@@ -1,5 +1,3 @@
-aiofiles
-aiohttp
 aws-lambda-powertools
 awslambdaric
 bson
@@ -9,7 +7,5 @@ numpy
 onnxruntime
 opencv-python
 pillow
-pyproj
 rasterio
 requests
-shapely

src/app.py

@@ -27,7 +27,7 @@ def get_response(status: int, start_time: float, request_id: str, response_body:
         str: json response
 
     """
-    app_logger.info(f"response_body:{response_body}.")
+    app_logger.debug(f"response_body:{response_body}.")
     response_body["duration_run"] = time.time() - start_time
     response_body["message"] = CUSTOM_RESPONSE_MESSAGES[status]
     response_body["request_id"] = request_id
@@ -38,59 +38,36 @@ def get_response(status: int, start_time: float, request_id: str, response_body:
         "body": json.dumps(response_body),
         "isBase64Encoded": False
     }
-    app_logger.info(f"response type:{type(response)} => {response}.")
+    app_logger.debug(f"response type:{type(response)} => {response}.")
     return json.dumps(response)
 
 
 def get_parsed_bbox_points(request_input: Dict) -> Dict:
     app_logger.info(f"try to parsing input request {request_input}...")
     bbox = request_input["bbox"]
-    app_logger.info(f"request bbox: {type(bbox)}, value:{bbox}.")
+    app_logger.debug(f"request bbox: {type(bbox)}, value:{bbox}.")
     ne = bbox["ne"]
     sw = bbox["sw"]
-    app_logger.info(f"request ne: {type(ne)}, value:{ne}.")
-    app_logger.info(f"request sw: {type(sw)}, value:{sw}.")
+    app_logger.debug(f"request ne: {type(ne)}, value:{ne}.")
+    app_logger.debug(f"request sw: {type(sw)}, value:{sw}.")
     ne_latlng = [float(ne["lat"]), float(ne["lng"])]
     sw_latlng = [float(sw["lat"]), float(sw["lng"])]
     bbox = [ne_latlng, sw_latlng]
     zoom = int(request_input["zoom"])
     for prompt in request_input["prompt"]:
-        app_logger.info(f"current prompt: {type(prompt)}, value:{prompt}.")
+        app_logger.debug(f"current prompt: {type(prompt)}, value:{prompt}.")
         data = prompt["data"]
-        # if prompt["type"] == "rectangle":
-        #     app_logger.info(f"current data points: {type(data)}, value:{data}.")
-        #     data_ne = data["ne"]
-        #     app_logger.info(f"current data_ne point: {type(data_ne)}, value:{data_ne}.")
-        #     data_sw = data["sw"]
-        #     app_logger.info(f"current data_sw point: {type(data_sw)}, value:{data_sw}.")
-        #
-        #     diff_pixel_coords_origin_data_ne = get_latlng_to_pixel_coordinates(ne, sw, data_ne, zoom, "ne")
-        #     app_logger.info(f'current diff prompt ne: {type(data)}, {data} => {diff_pixel_coords_origin_data_ne}.')
-        #     diff_pixel_coords_origin_data_sw = get_latlng_to_pixel_coordinates(ne, sw, data_sw, zoom, "sw")
-        #     app_logger.info(f'current diff prompt sw: {type(data)}, {data} => {diff_pixel_coords_origin_data_sw}.')
-        #     prompt["data"] = [
-        #         diff_pixel_coords_origin_data_ne["x"], diff_pixel_coords_origin_data_ne["y"],
-        #         diff_pixel_coords_origin_data_sw["x"], diff_pixel_coords_origin_data_sw["y"]
-        #     ]
-        #     # rect_diffs_input = str(Path(ROOT) / "rect_diffs_input.json")
-        #     # with open(rect_diffs_input, "w") as jj_out3:
-        #     #     json.dump({
-        #     #         "prompt_data": serialize(prompt["data"]),
-        #     #         "diff_pixel_coords_origin_data_ne": serialize(diff_pixel_coords_origin_data_ne),
-        #     #         "diff_pixel_coords_origin_data_sw": serialize(diff_pixel_coords_origin_data_sw),
-        #     #     }, jj_out3)
-        #     # app_logger.info(f"written json:{rect_diffs_input}.")
         if prompt["type"] == "point":
             current_point = get_latlng_to_pixel_coordinates(ne, sw, data, zoom, "point")
-            app_logger.info(f"current prompt: {type(current_point)}, value:{current_point}.")
+            app_logger.debug(f"current prompt: {type(current_point)}, value:{current_point}.")
             new_prompt_data = [current_point['x'], current_point['y']]
-            app_logger.info(f"new_prompt_data: {type(new_prompt_data)}, value:{new_prompt_data}.")
+            app_logger.debug(f"new_prompt_data: {type(new_prompt_data)}, value:{new_prompt_data}.")
             prompt["data"] = new_prompt_data
         else:
             raise ValueError("valid prompt type is only 'point'")
 
-    app_logger.info(f"bbox => {bbox}.")
-    app_logger.info(f'## request_input-prompt updated => {request_input["prompt"]}.')
+    app_logger.debug(f"bbox => {bbox}.")
+    app_logger.debug(f'## request_input-prompt updated => {request_input["prompt"]}.')
 
     app_logger.info(f"unpacking elaborated {request_input}...")
     return {
@@ -117,18 +94,18 @@ def lambda_handler(event: dict, context: LambdaContext):
             app_logger.error(f"e_constants1:{e_constants1}.")
             body = event
 
-        app_logger.info(f"body, #1: {type(body)}, {body}...")
+        app_logger.debug(f"body, #1: {type(body)}, {body}...")
 
         if isinstance(body, str):
             body_decoded_str = base64_decode(body)
-            app_logger.info(f"body_decoded_str: {type(body_decoded_str)}, {body_decoded_str}...")
+            app_logger.debug(f"body_decoded_str: {type(body_decoded_str)}, {body_decoded_str}...")
             body = json.loads(body_decoded_str)
 
         app_logger.info(f"body, #2: {type(body)}, {body}...")
 
         try:
             prompt_latlng = body["prompt"]
-            app_logger.info(f"prompt_latlng:{prompt_latlng}.")
+            app_logger.debug(f"prompt_latlng:{prompt_latlng}.")
             body_request = get_parsed_bbox_points(body)
             app_logger.info(f"body_request=> {type(body_request)}, {body_request}.")
             body_response = samexporter_predict(body_request["bbox"], body_request["prompt"], body_request["zoom"])

src/io/coordinates_pixel_conversion.py

@@ -1,5 +1,5 @@
 import math
-from typing import TypedDict, List
+from typing import TypedDict
 
 from src import app_logger
 from src.utilities.constants import TILE_SIZE

src/io/geo_helpers.py

@@ -0,0 +1,70 @@
+"""Async download raster tiles"""
+from pathlib import Path
+from typing import List
+
+import numpy as np
+
+from src import app_logger, PROJECT_ROOT_FOLDER
+
+
+def load_affine_transformation_from_matrix(matrix_source_coeffs: List):
+    from affine import Affine
+
+    if len(matrix_source_coeffs) != 6:
+        raise ValueError(f"Expected 6 coefficients, found {len(matrix_source_coeffs)}; "
+                         f"argument type: {type(matrix_source_coeffs)}.")
+
+    try:
+        a, d, b, e, c, f = (float(x) for x in matrix_source_coeffs)
+        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")
+
+
+def get_vectorized_raster_as_geojson(rio_output, mask):
+    try:
+        from rasterio import open as rio_open
+        from rasterio.features import shapes
+        from geopandas import GeoDataFrame
+
+        app_logger.info(f"read downloaded geotiff:{rio_output} to create the shapes_generator...")
+
+        with rio_open(rio_output, "r", driver="GTiff") as rio_src:
+            raster = rio_src.read()
+            transform = rio_src.transform
+            crs = rio_src.crs
+
+            app_logger.debug(f"geotiff band:{raster.shape}, type: {type(raster)}, dtype: {raster.dtype}.")
+            app_logger.debug(f"rio_src crs:{crs}.")
+            app_logger.debug(f"rio_src transform:{transform}.")
+
+            # mask = band != 0
+            shapes_generator = ({
+                'properties': {'raster_val': v}, 'geometry': s}
+                for i, (s, v)
+                # in 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(f"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(f"created a GeoDataFrame, export to geojson...")
+            geojson = gpd_polygonized_raster.to_json(to_wgs84=True)
+            app_logger.info(f"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
+
+
+if __name__ == '__main__':
+    npy_file = "prediction_masks_46.27697017893455_9.616470336914064_46.11441972281433_9.264907836914064.npy"
+    prediction_masks = np.load(Path(PROJECT_ROOT_FOLDER) / "tmp" / "try_by_steps" / "t0" / npy_file)
+
+    print("#")

src/io/helpers.py

@@ -1,618 +0,0 @@
-"""Helpers dedicated to georeferencing duties"""
-import base64
-import glob
-import json
-import os
-import zlib
-from math import log, tan, radians, cos, pi, floor, degrees, atan, sinh
-
-import rasterio
-
-from src import app_logger
-from src.utilities.constants import GEOJSON_SQUARE_TEMPLATE, OUTPUT_CRS_STRING, INPUT_CRS_STRING, SKIP_CONDITIONS_LIST
-from src.utilities.type_hints import ts_llist_float2, ts_geojson, ts_dict_str2b, ts_tuple_flat2, ts_tuple_flat4, \
-    ts_list_float4, ts_llist2, ts_tuple_int4, ts_ddict2
-
-ZIPJSON_KEY = 'base64(zip(o))'
-
-
-def get_geojson_square_angles(bounding_box:ts_llist_float2, name:str="buffer", debug:bool=False) -> ts_geojson:
-    """
-    Create a geojson-like dict rectangle from the input latitude/longitude bounding box
-    
-    Args:
-        bounding_box: float latitude/longitude bounding box
-        name: geojson-like rectangle name
-        debug: bool, default=False
-            logging debug argument
-
-    Returns:
-        dict: geojson-like object rectangle
-        
-    """
-    import copy
-    #from src.surferdtm_prediction_api.utilities.utilities import setup_logging
-
-    #app_logger = setup_logging(debug)
-    app_logger.info(f"bounding_box:{bounding_box}.")
-    top = bounding_box[0][0]
-    right = bounding_box[0][1]
-    bottom = bounding_box[1][0]
-    left = bounding_box[1][1]
-    bottom_left = [left, bottom]
-    top_left = [left, top]
-    top_right = [right, top]
-    bottom_right = [right, bottom]
-    coords = [bottom_left, top_left, top_right, bottom_right]
-    app_logger.info(f"coords:{coords}.")
-    geojson = copy.copy(GEOJSON_SQUARE_TEMPLATE)
-    geojson["name"] = name
-    geojson["features"][0]["geometry"]["coordinates"] = [[coords]]
-    app_logger.info(f"geojson:{geojson}.")
-    return geojson
-
-
-def crop_raster(merged_raster_path:str, area_crop_geojson:dict, debug:bool=False) -> ts_dict_str2b:
-    """
-    Crop a raster using a geojson-like object rectangle
-    
-    Args:
-        merged_raster_path: filename path pointing string to the raster to crop
-        area_crop_geojson: geojson-like object rectangle
-        debug: bool, default=False
-            logging debug argument
-
-    Returns:
-        dict: the cropped raster numpy array and the transform object with the georeferencing reference
-
-    """
-    #from src.surferdtm_prediction_api.utilities.utilities import setup_logging
-
-    #app_logger = setup_logging(debug)
-    try:
-        import rasterio
-        from rasterio.mask import mask
-
-        app_logger.info(f"area_crop_geojson::{area_crop_geojson}.")
-        geojson_reprojected = get_geojson_reprojected(area_crop_geojson, debug=debug)
-        shapes = [feature["geometry"] for feature in geojson_reprojected["features"]]
-        app_logger.info(f"geojson_reprojected:{geojson_reprojected}.")
-
-        app_logger.info(f"reading merged_raster_path while masking it from path:{merged_raster_path}.")
-        with rasterio.open(merged_raster_path, "r") as src:
-            masked_raster, masked_transform = mask(src, shapes, crop=True)
-            masked_meta = src.meta
-            app_logger.info(f"merged_raster_path, src:{src}.")
-        masked_meta.update({
-            "driver": "GTiff", "height": masked_raster.shape[1],
-            "width": masked_raster.shape[2], "transform": masked_transform}
-        )
-        return {"masked_raster": masked_raster, "masked_meta": masked_meta, "masked_transform": masked_transform}
-    except Exception as e:
-        app_logger.error(e)
-        raise e
-
-
-def get_geojson_reprojected(geojson:dict, output_crs:str=OUTPUT_CRS_STRING, debug:bool=False) -> dict:
-    """
-    change projection for input geojson-like object polygon
-
-    Args:
-        geojson: input geojson-like object polygon
-        output_crs: output crs string - Coordinate Reference Systems
-        debug: logging debug argument
-
-    Returns:
-        dict: reprojected geojson-like object
-
-    """
-    #from src.surferdtm_prediction_api.utilities.utilities import setup_logging
-
-    #app_logger = setup_logging(debug)
-    if not isinstance(geojson, dict):
-        raise ValueError(f"geojson here should be a dict, not of type {type(geojson)}.")
-    app_logger.info(f"start reprojecting geojson:{geojson}.")
-    try:
-        features = geojson['features']
-
-        output_crs_json = {"type": "name", "properties": {"name": f"urn:ogc:def:crs:{output_crs}"}}
-        geojson_output = {'features': [], 'type': 'FeatureCollection', "name": "converted", "crs": output_crs_json}
-
-        # Iterate through each feature of the feature collection
-        for feature in features:
-            feature_out = feature.copy()
-            new_coords = []
-            feat = feature['geometry']
-            app_logger.info(f"feat:{feat}.")
-            coords = feat['coordinates']
-            app_logger.info(f"coordinates:{coords}.")
-            # iterate over "coordinates" lists with 3 nested loops, practically with only one element but last loop
-            for coord_a in coords:
-                new_coords_a = []
-                for cord_b in coord_a:
-                    new_coords_b = []
-                    # Project/transform coordinate pairs of each ring
-                    # (iteration required in case geometry type is MultiPolygon, or there are holes)
-                    for xconv, yconf in cord_b:
-                        app_logger.info(f"xconv, yconf:{xconv},{yconf}.")
-                        x2, y2 = latlon_to_mercator(xconv, yconf)
-                        app_logger.info(f"x2, y2:{x2},{y2}.")
-                        new_coords_b.append([x2, y2])
-                    new_coords_a.append(new_coords_b)
-                new_coords.append(new_coords_a)
-            feature_out['geometry']['coordinates'] = new_coords
-            geojson_output['features'].append(feature_out)
-        app_logger.info(f"geojson_output:{geojson_output}.")
-        return geojson_output
-    except KeyError as ke_get_geojson_reprojected:
-        msg = f"ke_get_geojson_reprojected:{ke_get_geojson_reprojected}."
-        app_logger.error(msg)
-        raise KeyError(msg)
-
-
-def latlon_to_mercator(
-        lat:float, lon:float, input_crs:str=INPUT_CRS_STRING, output_crs:str=OUTPUT_CRS_STRING, always_xy:bool=True, debug:bool=False
-) -> ts_tuple_flat2:
-    """
-    Return a tuple of latitude, longitude float coordinates values transformed to mercator
-
-    Args:
-
-        lat: input latitude float value
-        lon: input longitude float value
-        input_crs: string, input Coordinate Reference Systems
-        output_crs: string, output Coordinate Reference Systems
-        always_xy: bool, default=True. 
-            If true, the transform method will accept as input and return as output
-            coordinates using the traditional GIS order, that is longitude, latitude
-            for geographic CRS and easting, northing for most projected CRS.
-        debug: bool, default=False.
-            logging debug argument
-
-    Returns:
-        tuple latitude/longitude float values
-
-    """
-    #from src.surferdtm_prediction_api.utilities.utilities import setup_logging
-    #app_logger = setup_logging(debug)
-    try:
-        from pyproj import Transformer
-        app_logger.info(f"lat:{lat},lon:{lon}.")
-        transformer = Transformer.from_crs(input_crs, output_crs, always_xy=always_xy)
-        out_lat, out_lon = transformer.transform(lat, lon)
-        app_logger.info(f"out_lat:{out_lat},out_lon:{out_lon}.")
-        return out_lat, out_lon
-    except Exception as e_latlon_to_mercator:
-        app_logger.error(f"e_latlon_to_mercator:{e_latlon_to_mercator}.")
-        raise e_latlon_to_mercator
-
-
-def sec(x:float) -> float:
-    """
-    Return secant (the reciprocal of the cosine) for given value
-
-    Args:
-        x: input float value
-
-    Returns:
-        float: secant of given float value
-
-    """
-    return 1 / cos(x)
-
-
-def latlon_to_xyz(lat:float, lon:float, z:int) -> ts_tuple_flat2:
-    """
-    Return x/y coordinates points for tiles from latitude/longitude values point.
-
-    Args:
-        lon: float longitude value
-        lat: float latitude value
-        z: float zoom value
-
-    Returns:
-        tuple: x, y values tiles coordinates
-
-    """
-    tile_count = pow(2, z)
-    x = (lon + 180) / 360
-    y = (1 - log(tan(radians(lat)) + sec(radians(lat))) / pi) / 2
-    return tile_count * x, tile_count * y
-
-
-def bbox_to_xyz(lon_min:float, lon_max:float, lat_min:float, lat_max:float, z:int) -> ts_tuple_flat4:
-    """
-    Return xyz reference coordinates for tiles from latitude/longitude min and max values.
-
-    Args:
-        lon_min: float min longitude value
-        lon_max: float max longitude value
-        lat_min: float min latitude value
-        lat_max: float max latitude value
-        z: float zoom value
-
-    Returns:
-        tuple: float x min, x max, y min, y max values tiles coordinates
-
-    """
-    x_min, y_max = latlon_to_xyz(lat_min, lon_min, z)
-    x_max, y_min = latlon_to_xyz(lat_max, lon_max, z)
-    return (floor(x_min), floor(x_max),
-            floor(y_min), floor(y_max))
-
-
-def mercator_to_lat(mercator_y:float) -> float:
-    """
-    Return latitude value coordinate from mercator coordinate value
-
-    Args:
-        mercator_y: float mercator value coordinate
-
-    Returns:
-        float: latitude value coordinate
-
-    """
-    return degrees(atan(sinh(mercator_y)))
-
-
-def y_to_lat_edges(y:float, z:int) -> ts_tuple_flat2:
-    """
-    Return edge float latitude values coordinates from x,z tiles coordinates
-
-    Args:
-        y: float x tile value coordinate
-        z: float zoom tile value coordinate
-
-    Returns:
-        tuple: two float latitude values coordinates
-
-    """
-    tile_count = pow(2, z)
-    unit = 1 / tile_count
-    relative_y1 = y * unit
-    relative_y2 = relative_y1 + unit
-    lat1 = mercator_to_lat(pi * (1 - 2 * relative_y1))
-    lat2 = mercator_to_lat(pi * (1 - 2 * relative_y2))
-    return lat1, lat2
-
-
-def x_to_lon_edges(x:float, z:int) -> ts_tuple_flat2:
-    """
-    Return edge float longitude values coordinates from x,z tiles coordinates
-
-    Args:
-        x: float x tile value coordinate
-        z: float zoom tile value coordinate
-
-    Returns:
-        tuple: two float longitude values coordinates
-
-    """
-    tile_count = pow(2, z)
-    unit = 360 / tile_count
-    lon1 = -180 + x * unit
-    lon2 = lon1 + unit
-    return lon1, lon2
-
-
-def tile_edges(x:float, y:float, z:int) -> ts_list_float4:
-    """
-    Return edge float latitude/longitude value coordinates from xyz tiles coordinates
-
-    Args:
-        x: float x tile value coordinate
-        y: float y tile value coordinate
-        z: float zoom tile value coordinate
-
-    Returns:
-        tuple: float latitude/longitude values coordinates
-
-    """
-    lat1, lat2 = y_to_lat_edges(y, z)
-    lon1, lon2 = x_to_lon_edges(x, z)
-    return [lon1, lat1, lon2, lat2]
-
-
-def merge_tiles(input_pattern:str, output_path:str, temp_dir:str, debug:bool=False) -> None:
-    """
-    Merge given raster glob input pattern into one unique georeferenced raster.
-
-    Args:
-        input_pattern: input glob pattern needed for search the raster filenames
-        output_path: output path where to write the merged raster
-        temp_dir: temporary folder needed for create
-        debug: bool, default=False.
-            logging debug argument
-
-    """
-    #from src.surferdtm_prediction_api.utilities.utilities import setup_logging
-
-    #app_logger = setup_logging(debug)
-    try:
-        from osgeo import gdal
-    except ModuleNotFoundError as module_error_merge_tiles:
-        msg = f"module_error_merge_tiles:{module_error_merge_tiles}."
-        app_logger.error(msg)
-        raise module_error_merge_tiles
-
-    try:
-        vrt_path = os.path.join(temp_dir, "tiles.vrt")
-        os_list_dir1 = os.listdir(temp_dir)
-        app_logger.info(f"os_list_dir1:{os_list_dir1}.")
-
-        gdal.BuildVRT(vrt_path, glob.glob(input_pattern))
-        gdal.Translate(output_path, vrt_path)
-
-        os_list_dir2 = os.listdir(temp_dir)
-        app_logger.info(f"os_list_dir2:{os_list_dir2}.")
-    except IOError as ioe_merge_tiles:
-        msg = f"ioe_merge_tiles:{ioe_merge_tiles}."
-        app_logger.error(msg)
-        raise ioe_merge_tiles
-
-
-def get_lat_lon_coords(bounding_box: ts_llist2) -> ts_tuple_int4:
-    """
-    Return couples of float latitude/longitude values from bounding box input list.
-
-    Args:
-        bounding_box: bounding box input list of latitude/longitude coordinates
-
-    Returns:
-        tuple: float longitude min, latitude min, longitude max, longitude max values coordinates
-
-    """
-    top_right, bottom_left = bounding_box
-    lat_max, lon_max = top_right
-    lat_min, lon_min = bottom_left
-    if lon_min == lon_max or lat_min == lat_max:
-        raise ValueError(f"latitude and/or longitude coordinates should not be equal each others... {bounding_box}.")
-    return lon_min, lat_min, lon_max, lat_max
-
-
-def get_prediction_georeferenced(prediction_obj:dict, transform:rasterio.transform, skip_conditions_list:list=None, debug:bool=False) -> dict:
-    """
-    Return a georeferenced geojson-like object starting from a dict containing "predictions" -> "points" list.
-    Apply the affine transform matrix of georeferenced raster submitted to the machine learning model.  
-
-    Args:
-        prediction_obj: input dict 
-        transform: 'rasterio.transform' or dict list, affine tranform matrix
-        skip_conditions_list: dict list, skip condition list
-        debug: bool, default=False.
-            logging debug argument
-
-    Returns:
-        dict
-
-    """
-    #from src.surferdtm_prediction_api.utilities.utilities import setup_logging
-
-    if skip_conditions_list is None:
-        skip_conditions_list = SKIP_CONDITIONS_LIST
-
-    #app_logger = setup_logging(debug)
-    app_logger.info(f"prediction_obj::{prediction_obj}, transform::{transform}.")
-    crs = {"type": "name", "properties": {"name": "urn:ogc:def:crs:EPSG::3857"}}
-    geojson_obj = {'features': [], 'type': 'FeatureCollection', "name": "geojson_name", "crs": crs}
-    for n, prediction in enumerate(prediction_obj["predictions"]):
-        points_dict_ = prediction["points"]
-        points_list = [[p["x"], p["y"]] for p in points_dict_]
-        app_logger.info(f"points_list::{points_list}.")
-        # if check_skip_conditions(prediction, skip_conditions_list, debug=debug):
-        #     continue
-        feature = populate_features_geojson(n, points_list, confidence=prediction["confidence"], geomorphic_class=prediction["class"])
-        app_logger.info(f"geojson::feature:{feature}.")
-        feature["geometry"] = apply_transform(feature["geometry"], transform, debug=debug)
-        geojson_obj["features"].append(feature)
-    app_logger.info(f"geojson::post_update:{geojson_obj}.")
-    return geojson_obj
-
-
-def populate_features_geojson(idx: int, coordinates_list: list, **kwargs) -> ts_ddict2:
-    """
-    Return a list of coordinate points in a geojson-like feature-like object.
-
-    Args:
-        idx: int, feature index
-        coordinates_list: dict list, coordinate points
-        **kwargs: optional arguments to merge within the geojson properties feature
-
-    Returns:
-        dict
-
-    """
-    return {
-        "type": "Feature",
-        "properties": {"id": idx, **kwargs},
-        "geometry": {
-            "type": "MultiPolygon",
-            "coordinates": [[coordinates_list]],
-        }
-    }
-
-
-def check_skip_conditions(prediction:dict, skip_conditions_list:list, debug:bool=False) -> bool:
-    """
-    Loop over elements within skip_condition_list and return a boolean if no condition to skip (or exceptions).
-
-    Args:
-        prediction: input dict to check
-        skip_conditions_list: dict list with conditions to evaluate
-        debug: bool, default=False
-            logging debug argument
-
-    Returns:
-        bool
-
-    """
-    for obj in skip_conditions_list:
-        return skip_feature(prediction, obj["skip_key"], obj["skip_value"], obj["skip_condition"], debug=debug)
-    return False
-
-
-def skip_feature(prediction:dict, skip_key:float, skip_value:str, skip_condition:str, debug:bool=False) -> bool:
-    """
-    Return False if values from input dict shouldn't be skipped,
-    True in case of exceptions, empty skip_condition or when chosen condition meets skip_value and skip_condition.
-
-    E.g. confidence should be major than 0.8: if confidence is equal to 0.65 then return True (0.65 < 0.8) and skip!
-
-    Args:
-        prediction: input dict to check
-        skip_key: skip condition key string
-        skip_value: skip condition value string
-        skip_condition: string (major | minor | equal)
-        debug: bool, default=False
-            logging debug argument
-
-    Returns:
-        bool
-
-    """
-    #from src.surferdtm_prediction_api.utilities.utilities import setup_logging
-    #app_logger = setup_logging(debug)
-    try:
-        v = prediction[skip_key]
-        match skip_condition:
-            case "major":
-                return v > skip_value
-            case "minor":
-                return v < skip_value
-            case "equal":
-                return v == skip_value
-            case "":
-                return False
-    except KeyError as ke_filter_feature:
-        app_logger.error(f"ke_filter_feature:{ke_filter_feature}.")
-        return False
-    except Exception as e_filter_feature:
-        app_logger.error(f"e_filter_feature:{e_filter_feature}.")
-        return False
-
-
-def apply_transform(geometry:object, transform:list[object], debug:bool=False) -> dict:
-    """
-    Returns a GeoJSON-like mapping from a transformed geometry using an affine transformation matrix.
-
-    The coefficient matrix is provided as a list or tuple with 6 items
-    for 2D transformations. The 6 parameter matrix is::
-
-        [a, b, d, e, xoff, yoff]
-
-    which represents the augmented matrix::
-
-        [x']   / a  b xoff \ [x]
-        [y'] = | d  e yoff | [y]
-        [1 ]   \ 0  0   1  / [1]
-
-    or the equations for the transformed coordinates::
-
-        x' = a * x + b * y + xoff
-        y' = d * x + e * y + yoff
-
-    Args:
-        geometry: geometry value from a geojson dict
-        transform: list of float values (affine transformation matrix)
-        debug: bool, default=False
-            logging debug argument
-
-    Returns:
-        dict
-
-    """
-    #from src.surferdtm_prediction_api.utilities.utilities import setup_logging
-
-    #app_logger = setup_logging(debug)
-
-    try:
-        from shapely.affinity import affine_transform
-        from shapely.geometry import mapping, shape
-        try:
-            geometry_transformed = affine_transform(shape(geometry), [transform.a, transform.b, transform.d, transform.e, transform.xoff, transform.yoff])
-        except AttributeError as ae:
-            app_logger.warning(f"ae:{ae}.")
-            geometry_transformed = affine_transform(shape(geometry), [transform[0], transform[1], transform[2], transform[3], transform[4], transform[5]])
-        geometry_serialized = mapping(geometry_transformed)
-        app_logger.info(f"geometry_serialized:{geometry_serialized}.")
-        return geometry_serialized
-    except ImportError as ie_apply_transform:
-        app_logger.error(f"ie_apply_transform:{ie_apply_transform}.")
-        raise ie_apply_transform
-    except Exception as e_apply_transform:
-        app_logger.error(f"e_apply_transform:{e_apply_transform}.")
-        raise e_apply_transform
-
-
-def get_perc(nan_count:int, total_count:int) -> str:
-    """
-    Return a formatted string with a percentage value representing the ratio between NaN and total number elements within a numpy array
-
-    Args:
-        nan_count: NaN value elements
-        total_count: total count of elements
-
-    Returns:
-        str
-
-    """
-    return f"{100*nan_count/total_count:.2f}"
-
-
-def json_unzip(j:dict, debug:bool=False) -> str:
-    """
-    Return uncompressed content from input dict using 'zlib' library
-
-    Args:
-        j: input dict to uncompress. key must be 'base64(zip(o))'
-        debug: logging debug argument
-
-    Returns:
-        dict: uncompressed dict
-
-    """
-    from json import JSONDecodeError
-    from zlib import error as zlib_error
-
-    #from src.surferdtm_prediction_api.utilities.utilities import setup_logging
-
-    #app_logger = setup_logging(debug)
-
-    try:
-        j = zlib.decompress(base64.b64decode(j[ZIPJSON_KEY]))
-    except KeyError as ke:
-        ke_error_msg = f"Could not decode/unzip the content because of wrong/missing dict key:{ke}."
-        raise KeyError(ke_error_msg)
-    except zlib_error as zlib_error2:
-        zlib_error2_msg = f"Could not decode/unzip the content because of:{zlib_error2}."
-        app_logger.error(zlib_error2_msg)
-        raise RuntimeError(zlib_error2_msg)
-
-    try:
-        j = json.loads(j)
-    except JSONDecodeError as json_e1:
-        msg = f"Could interpret the unzipped content because of JSONDecodeError with msg:{json_e1.msg}, pos:{json_e1.pos}, broken json:'{json_e1.doc}'"
-        app_logger.error(msg)
-        raise RuntimeError(msg)
-
-    return j
-
-
-def json_zip(j:dict) -> dict[str]:
-    """
-    Return compressed content from input dict using 'zlib' library
-
-    Args:
-        j: input dict to compress
-
-    Returns:
-        dict: compressed dict
-
-    """
-    return {
-        ZIPJSON_KEY: base64.b64encode(
-            zlib.compress(
-                json.dumps(j).encode('utf-8')
-            )
-        ).decode('ascii')
-    }

src/io/tiles_to_tiff.py

@@ -1,91 +0,0 @@
-"""Async download raster tiles"""
-from pathlib import Path
-
-import numpy as np
-
-from src import app_logger, PROJECT_ROOT_FOLDER
-from src.io.tms2geotiff import download_extent
-from src.utilities.constants import COMPLETE_URL_TILES, DEFAULT_TMS
-from src.utilities.type_hints import ts_llist2
-
-
-COOKIE_SESSION = {
-    "Accept": "*/*",
-    "Accept-Encoding": "gzip, deflate",
-    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; rv:91.0) Gecko/20100101 Firefox/91.0",
-}
-
-
-def load_affine_transformation_from_matrix(matrix_source_coeffs):
-    from affine import Affine
-
-    if len(matrix_source_coeffs) != 6:
-        raise ValueError(f"Expected 6 coefficients, found {len(matrix_source_coeffs)};"
-                         f"argument type: {type(matrix_source_coeffs)}.")
-
-    try:
-        a, d, b, e, c, f = (float(x) for x in matrix_source_coeffs)
-        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")
-
-
-# @timing_decorator
-def convert(bounding_box: ts_llist2, zoom: int) -> tuple:
-    """
-    Starting from a bounding box of two couples of latitude and longitude coordinate values, recognize a stratovolcano from an RGB image. The algorithm
-    create the image composing three channels as slope, DEM (Digital Elevation Model) and curvature. In more detail:
-
-    - download a series of terrain DEM (Digital Elevation Model) raster tiles enclosed within that bounding box
-    - merge all the downloaded rasters
-    - crop the merged raster
-    - process the cropped raster to extract slope and curvature (1st and 2nd degree derivative)
-    - produce three raster channels (DEM, slope and curvature rasters) to produce an RGB raster image
-    - submit the RGB image to a remote machine learning service to try to recognize a polygon representing a stratovolcano
-    - the output of the machine learning service is a json, so we need to georeferencing it
-    - finally we return a dict as response containing
-    - uploaded_file_name
-    - bucket_name
-    - prediction georeferenced geojson-like dict
-
-    Args:
-        bounding_box: float latitude/longitude bounding box
-        zoom: integer zoom value
-
-    Returns:
-        dict: uploaded_file_name (str), bucket_name (str), prediction_georef (dict), n_total_obj_prediction (str)
-
-    """
-
-    tile_source = COMPLETE_URL_TILES
-    app_logger.info(f"start_args: tile_source:{tile_source},bounding_box:{bounding_box},zoom:{zoom}.")
-
-    try:
-        import rasterio
-
-        app_logger.info(f'tile_source: {tile_source}!')
-        pt0, pt1 = bounding_box
-        app_logger.info("downloading...")
-        img, matrix = download_extent(DEFAULT_TMS, pt0[0], pt0[1], pt1[0], pt1[1], zoom)
-
-        app_logger.info(f'img: type {type(img)}, len_matrix:{len(matrix)}, matrix {matrix}.')
-        app_logger.info(f'img: size (shape if PIL) {img.size}.')
-        try:
-            np_img = np.array(img)
-            app_logger.info(f'img: shape (numpy) {np_img.shape}.')
-        except Exception as e_shape:
-            app_logger.info(f'e_shape {e_shape}.')
-            raise e_shape
-
-        return img, matrix
-    except ImportError as e_import_convert:
-        app_logger.error(f"e0:{e_import_convert}.")
-        raise e_import_convert
-
-
-if __name__ == '__main__':
-    npy_file = "prediction_masks_46.27697017893455_9.616470336914064_46.11441972281433_9.264907836914064.npy"
-    prediction_masks = np.load(Path(PROJECT_ROOT_FOLDER) / "tmp" / "try_by_steps" / "t0" / npy_file)
-
-    print("#")

src/prediction_api/predictors.py

@@ -1,59 +1,21 @@
 # Press the green button in the gutter to run the script.
 import tempfile
 from pathlib import Path
-from typing import List
 
 import numpy as np
-import rasterio
 
 from src import app_logger, MODEL_FOLDER
-from src.io.tiles_to_tiff import convert
-from src.io.tms2geotiff import save_geotiff_gdal
+from src.io.geo_helpers import get_vectorized_raster_as_geojson
+from src.io.tms2geotiff import save_geotiff_gdal, download_extent
 from src.prediction_api.sam_onnx import SegmentAnythingONNX
-from src.utilities.constants import MODEL_ENCODER_NAME, MODEL_DECODER_NAME
+from src.utilities.constants import MODEL_ENCODER_NAME, MODEL_DECODER_NAME, DEFAULT_TMS
 
 
 models_dict = {"fastsam": {"instance": None}}
 
 
-def zip_arrays(arr1, arr2):
-    try:
-        arr1_list = arr1.tolist()
-        arr2_list = arr2.tolist()
-        # return {serialize(k): serialize(v) for k, v in zip(arr1_list, arr2_list)}
-        d = {}
-        for n1, n2 in zip(arr1_list, arr2_list):
-            app_logger.info(f"n1:{n1}, type {type(n1)}, n2:{n2}, type {type(n2)}.")
-            n1f = str(n1)
-            n2f = str(n2)
-            app_logger.info(f"n1:{n1}=>{n1f}, n2:{n2}=>{n2f}.")
-            d[n1f] = n2f
-        app_logger.info(f"zipped dict:{d}.")
-        return d
-    except Exception as e_zip_arrays:
-        app_logger.info(f"exception zip_arrays:{e_zip_arrays}.")
-        return {}
-
-
-def load_affine_transformation_from_matrix(matrix_source_coeffs: List):
-    from affine import Affine
-
-    if len(matrix_source_coeffs) != 6:
-        raise ValueError(f"Expected 6 coefficients, found {len(matrix_source_coeffs)}; argument type: {type(matrix_source_coeffs)}.")
-
-    try:
-        a, d, b, e, c, f = (float(x) for x in matrix_source_coeffs)
-        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")
-
-
 def samexporter_predict(bbox, prompt: list[dict], zoom: float, model_name: str = "fastsam") -> dict:
     try:
-        from rasterio.features import shapes
-        from geopandas import GeoDataFrame
-
         if models_dict[model_name]["instance"] is None:
             app_logger.info(f"missing instance model {model_name}, instantiating it now!")
             model_instance = SegmentAnythingONNX(
@@ -65,71 +27,47 @@ def samexporter_predict(bbox, prompt: list[dict], zoom: float, model_name: str =
         models_instance = models_dict[model_name]["instance"]
 
         with tempfile.TemporaryDirectory() as input_tmp_dir:
-            img, matrix = convert(
-                bounding_box=bbox,
-                zoom=int(zoom)
-            )
-            app_logger.debug(f"## img type {type(img)} with shape/size:{img.size}, matrix:{matrix}.")
+            app_logger.info(f'tile_source: {DEFAULT_TMS}!')
+            pt0, pt1 = bbox
+            app_logger.info("downloading...")
+            img, matrix = download_extent(DEFAULT_TMS, pt0[0], pt0[1], pt1[0], pt1[1], zoom)
+            app_logger.debug(f"img type {type(img)} with shape/size:{img.size}, matrix:{matrix}.")
 
             pt0, pt1 = bbox
             rio_output = str(Path(input_tmp_dir) / f"downloaded_rio_{pt0[0]}_{pt0[1]}_{pt1[0]}_{pt1[1]}.tif")
-            app_logger.debug(f"saving downloaded geotiff image to {rio_output}...")
+            app_logger.debug(f"saving downloaded image as geotiff using matrix {matrix} to {rio_output}...")
             save_geotiff_gdal(img, rio_output, matrix)
-            app_logger.info(f"saved downloaded geotiff image to {rio_output}...")
-
-            np_img = np.array(img)
-            app_logger.info(f"## img type {type(np_img)}, prompt:{prompt}.")
-
-            app_logger.info(f"onnxruntime input shape/size (shape if PIL) {np_img.size}.")
-            try:
-                app_logger.info(f"onnxruntime input shape (NUMPY) {np_img.shape}.")
-            except Exception as e_shape:
-                app_logger.error(f"e_shape:{e_shape}.")
-            app_logger.info(f"instantiated model {model_name}, ENCODER {MODEL_ENCODER_NAME}, "
-                            f"DECODER {MODEL_DECODER_NAME} from {MODEL_FOLDER}: Creating embedding...")
-            embedding = models_instance.encode(np_img)
-            app_logger.info(f"embedding created, running predict_masks with prompt {prompt}...")
-            prediction_masks = models_instance.predict_masks(embedding, prompt)
-            app_logger.info(f"Created {len(prediction_masks)} prediction_masks,"
-                            f"shape:{prediction_masks.shape}, dtype:{prediction_masks.dtype}.")
-
-            mask = np.zeros((prediction_masks.shape[2], prediction_masks.shape[3]), dtype=np.uint8)
-            for n, m in enumerate(prediction_masks[0, :, :, :]):
-                app_logger.debug(f"## {n} mask => m shape:{mask.shape}, {mask.dtype}.")
-                mask[m > 0.0] = 255
-
-            app_logger.info(f"read downloaded geotiff:{rio_output} to create the shapes_generator...")
-
-            with rasterio.open(rio_output, "r", driver="GTiff") as rio_src:
-                band = rio_src.read()
-                try:
-                    app_logger.debug(f"geotiff band:{band.shape}, type: {type(band)}, dtype: {band.dtype}.")
-                    app_logger.debug(f"rio_src crs:{rio_src.crs}.")
-                    app_logger.debug(f"rio_src transform:{rio_src.transform}.")
-                except Exception as e_shape_band:
-                    app_logger.error(f"e_shape_band:{e_shape_band}.")
-                    raise e_shape_band
-                # mask_band = band != 0
-                shapes_generator = ({
-                    'properties': {'raster_val': v}, 'geometry': s}
-                    for i, (s, v)
-                    # in enumerate(shapes(mask, mask=(band != 0), transform=rio_src.transform))
-                    # use mask=None to avoid using source
-                    in enumerate(shapes(mask, mask=None, transform=rio_src.transform))
-                )
-                app_logger.info(f"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(f"created a GeoDataFrame, export to geojson...")
-                geojson = gpd_polygonized_raster.to_json(to_wgs84=True)
-                app_logger.info(f"created geojson...")
-
-                return {
-                    "geojson": geojson,
-                    "n_shapes_geojson": len(shapes_list),
-                    "n_predictions": len(prediction_masks),
-                    # "n_pixels_predictions": zip_arrays(mask_unique_values, mask_unique_values_count),
-                }
-    except ImportError as e:
-        app_logger.error(f"Error trying import module:{e}.")
+            app_logger.info(f"saved downloaded geotiff image to {rio_output}, preparing inference...")
+
+            mask, prediction_masks = get_raster_inference(img, prompt, models_instance, model_name)
+            n_predictions = len(prediction_masks)
+            app_logger.info(f"created {n_predictions} masks, preparing conversion to geojson...")
+            return {
+                "n_predictions": n_predictions,
+                **get_vectorized_raster_as_geojson(rio_output, mask)
+            }
+    except ImportError as e_import_module:
+        app_logger.error(f"Error trying import module:{e_import_module}.")
+
+
+def get_raster_inference(img, prompt, models_instance, model_name):
+    np_img = np.array(img)
+    app_logger.info(f"img type {type(np_img)}, prompt:{prompt}.")
+    app_logger.debug(f"onnxruntime input shape/size (shape if PIL) {np_img.size}.")
+    try:
+        app_logger.debug(f"onnxruntime input shape (NUMPY) {np_img.shape}.")
+    except Exception as e_shape:
+        app_logger.error(f"e_shape:{e_shape}.")
+    app_logger.info(f"instantiated model {model_name}, ENCODER {MODEL_ENCODER_NAME}, "
+                    f"DECODER {MODEL_DECODER_NAME} from {MODEL_FOLDER}: Creating embedding...")
+    embedding = models_instance.encode(np_img)
+    app_logger.debug(f"embedding created, running predict_masks with prompt {prompt}...")
+    inference_out = models_instance.predict_masks(embedding, prompt)
+    app_logger.info(f"Created {len(inference_out)} prediction_masks,"
+                    f"shape:{inference_out.shape}, dtype:{inference_out.dtype}.")
+    mask = np.zeros((inference_out.shape[2], inference_out.shape[3]), dtype=np.uint8)
+    for n, m in enumerate(inference_out[0, :, :, :]):
+        app_logger.debug(f"{n}th of prediction_masks shape {inference_out.shape}"
+                         f" => mask shape:{mask.shape}, {mask.dtype}.")
+        mask[m > 0.0] = 255
+    return mask, inference_out

src/utilities/constants.py

@@ -1,27 +1,7 @@
 """Project constants"""
-CHANNEL_EXAGGERATIONS_LIST = [2.5, 1.1, 2.0]
 INPUT_CRS_STRING = "EPSG:4326"
 OUTPUT_CRS_STRING = "EPSG:3857"
-# DOMAIN_URL_TILES = "elevation-tiles-prod-eu.s3.eu-central-1.amazonaws.com"
-# RELATIVE_URL_TILES = "geotiff/{z}/{x}/{y}.tif"
-# COMPLETE_URL_TILES = f"https://{DOMAIN_URL_TILES}/{RELATIVE_URL_TILES}"
 ROOT = "/tmp"
-NODATA_VALUES = -32768
-MODEL_PROJECT_NAME = "surferdtm"
-MODEL_VERSION = 4
-SKIP_CONDITIONS_LIST = [{"skip_key": "confidence", "skip_value": 0.5, "skip_condition": "major"}]
-FEATURE_SQUARE_TEMPLATE = [
-    {'type': 'Feature', 'properties': {'id': 1},
-     'geometry': {
-         'type': 'MultiPolygon',
-         'coordinates': [[]]
-     }}
-]
-GEOJSON_SQUARE_TEMPLATE = {
-    'type': 'FeatureCollection', 'name': 'etna_wgs84p',
-    'crs': {'type': 'name', 'properties': {'name': 'urn:ogc:def:crs:OGC:1.3:CRS84'}},
-    'features': FEATURE_SQUARE_TEMPLATE
-}
 CUSTOM_RESPONSE_MESSAGES = {
     200: "ok",
     422: "Missing required parameter",
@@ -29,8 +9,6 @@ CUSTOM_RESPONSE_MESSAGES = {
 }
 MODEL_ENCODER_NAME = "mobile_sam.encoder.onnx"
 MODEL_DECODER_NAME = "sam_vit_h_4b8939.decoder.onnx"
-ZOOM = 13
-SOURCE_TYPE = "Satellite"
 TILE_SIZE = 256
 EARTH_EQUATORIAL_RADIUS = 6378137.0
 DEFAULT_TMS = 'https://tile.openstreetmap.org/{z}/{x}/{y}.png'
@@ -38,4 +16,3 @@ WKT_3857 = 'PROJCS["WGS 84 / Pseudo-Mercator",GEOGCS["WGS 84",DATUM["WGS_1984",S
 WKT_3857 += 'AUTHORITY["EPSG","8901"]],UNIT["degree",0.0174532925199433,AUTHORITY["EPSG","9122"]],AUTHORITY["EPSG","4326"]],PROJECTION["Mercator_1SP"],PARAMETER["central_meridian",0],'
 WKT_3857 += 'PARAMETER["scale_factor",1],PARAMETER["false_easting",0],PARAMETER["false_northing",0],UNIT["metre",1,AUTHORITY["EPSG","9001"]],AXIS["X",EAST],AXIS["Y",NORTH],EXTENSION["PROJ4",'
 WKT_3857 += '"+proj=merc +a=6378137 +b=6378137 +lat_ts=0.0 +lon_0=0.0 +x_0=0.0 +y_0=0 +k=1.0 +units=m +nadgrids=@null +wktext +no_defs"],AUTHORITY["EPSG","3857"]]'
-COMPLETE_URL_TILES = DEFAULT_TMS

src/utilities/type_hints.py

@@ -1,24 +1,3 @@
 """custom type hints"""
-from typing import List, Tuple
-import numpy as np
-
-ts_list_str1 = list[str]
-ts_http2 = tuple[ts_list_str1, ts_list_str1]
-ts_list_float2 = list[float, float]
-ts_llist_float2 = list[ts_list_float2, ts_list_float2]
-ts_geojson = dict[str, str, dict[str, dict[str]], list[str, dict[int], dict[str, list]]]
-ts_float64_1 = tuple[np.float64, np.float64, np.float64, np.float64, np.float64, np.float64]
-ts_float64_2 = tuple[np.float64, np.float64, np.float64, np.float64, np.float64, np.float64, np.float64]
 ts_dict_str2 = dict[str, str]
 ts_dict_str3 = dict[str, str, any]
-ts_dict_str2b = dict[str, any]
-ts_ddict1 = dict[str, dict[str, any], dict, dict, any]
-ts_ddict2 = dict[str, dict, dict[str, list]]
-ts_tuple_str2 = tuple[str, str]
-ts_tuple_arr2 = tuple[np.ndarray, np.ndarray]
-ts_tuple_flat2 = tuple[float, float]
-ts_tuple_flat4 = tuple[float, float, float, float]
-ts_list_float4 = list[float, float, float, float]
-ts_tuple_int4 = tuple[int, int, int, int]
-ts_llist2 = list[[int, int], [int, int]]
-ts_ddict3 = dict[list[dict[float | int | str]], dict[float | int]]

src/utilities/utilities.py

@@ -1,8 +1,4 @@
 """Various utilities (logger, time benchmark, args dump, numerical and stats info)"""
-import numpy as np
-
-from src import app_logger
-from src.utilities.type_hints import ts_float64_1, ts_float64_2
 
 
 def is_base64(sb):
@@ -21,22 +17,6 @@ def is_base64(sb):
         return False
 
 
-def get_system_info():
-    import multiprocessing
-    import torch.multiprocessing as mp
-    import os
-    import subprocess
-
-    app_logger.info(f"mp::cpu_count:{mp.cpu_count()}.")
-    app_logger.info(f"multiprocessing::cpu_count:{multiprocessing.cpu_count()}.")
-    app_logger.info(f"os::cpu_count:{os.cpu_count()}")
-    app_logger.info(f"os::sched_getaffinity:{len(os.sched_getaffinity(0))}")
-    lscpu_output = subprocess.run("/usr/bin/lscpu", capture_output=True)
-    app_logger.info(f"lscpu:{lscpu_output.stdout.decode('utf-8')}.")
-    free_mem_output = subprocess.run(["/usr/bin/free", "-m"], capture_output=True)
-    app_logger.info(f"free_mem_output:{free_mem_output.stdout.decode('utf-8')}.")
-
-
 def base64_decode(s):
     import base64
 
@@ -44,141 +24,3 @@ def base64_decode(s):
         return base64.b64decode(s, validate=True).decode("utf-8")
 
     return s
-
-
-def get_constants(event: dict, debug=False) -> dict:
-    """
-    Return constants we need to use from event, context and environment variables (both production and test).
-
-    Args:
-        event: request event
-        debug: logging debug argument
-
-    Returns:
-        dict: project constants object
-
-    """
-    import json
-
-    try:
-        body = event["body"]
-    except Exception as e_constants1:
-        app_logger.error(f"e_constants1:{e_constants1}.")
-        body = event
-
-    if isinstance(body, str):
-        body = json.loads(event["body"])
-
-    try:
-        debug = body["debug"]
-        app_logger.info(f"re-try get debug value:{debug}, log_level:{app_logger.level}.")
-    except KeyError:
-        app_logger.error("get_constants:: no debug key, pass...")
-    app_logger.info(f"constants debug:{debug}, log_level:{app_logger.level}, body:{body}.")
-
-    try:
-        return {
-            "bbox": body["bbox"],
-            "point": body["point"],
-            "debug": debug
-        }
-    except KeyError as e_key_constants2:
-        app_logger.error(f"e_key_constants2:{e_key_constants2}.")
-        raise KeyError(f"e_key_constants2:{e_key_constants2}.")
-    except Exception as e_constants2:
-        app_logger.error(f"e_constants2:{e_constants2}.")
-        raise e_constants2
-
-
-def get_rasters_info(rasters_list:list, names_list:list, title:str="", debug:bool=False) -> str:
-    """
-    Analyze numpy arrays' list to extract a string containing some useful information. For every raster:
-
-        - type of raster
-        - raster.dtype if that's instance of np.ndarray
-        - raster shape
-        - min of raster value, over all axis (flattening the array)
-        - max of raster value, over all axis (flattening the array)
-        - mean of raster value, over all axis (flattening the array)
-        - median of raster value, over all axis (flattening the array)
-        - standard deviation of raster value, over all axis (flattening the array)
-        - variance of raster value, over all axis (flattening the array)
-
-    Raises:
-        ValueError if raster_list and names_list have a different number of elements
-
-    Args:
-        rasters_list: list of numpy array raster to analyze
-        names_list: string list of numpy array
-        title: title of current analytic session
-        debug: logging debug argument
-
-    Returns:
-        str: the collected information
-
-    """
-
-    msg = f"get_rasters_info::title:{title},\n"
-    if not len(rasters_list) == len(names_list):
-        msg = "raster_list and names_list should have the same number of elements:\n"
-        msg += f"len(rasters_list):{len(rasters_list)}, len(names_list):{len(names_list)}."
-        raise ValueError(msg)
-    try:
-        for raster, name in zip(rasters_list, names_list):
-            try:
-                if isinstance(raster, np.ndarray):
-                    shape_or_len = raster.shape
-                elif isinstance(raster, list):
-                    shape_or_len = len(raster)
-                else:
-                    raise ValueError(f"wrong argument type:{raster}, variable:{raster}.")
-                zmin, zmax, zmean, zmedian, zstd, zvar = get_stats_raster(raster, debug=debug)
-                msg += "name:{}:type:{},dtype:{},shape:{},min:{},max:{},mean:{},median:{},std:{},var:{}\n".format(
-                    name, type(raster), raster.dtype if isinstance(raster, np.ndarray) else None, shape_or_len, zmin,
-                    zmax, zmean, zmedian, zstd, zvar
-                )
-            except Exception as get_rasters_types_e:
-                msg = f"get_rasters_types_e::{get_rasters_types_e}, type_raster:{type(raster)}."
-                app_logger.error(msg)
-                raise ValueError(msg)
-    except IndexError as get_rasters_types_ie:
-        app_logger.error(f"get_rasters_types::len:rasters_list:{len(rasters_list)}, len_names_list:{len(names_list)}.")
-        raise get_rasters_types_ie
-    return msg + "\n=============================\n"
-
-
-def get_stats_raster(raster: np.ndarray, get_rms:bool=False, debug:bool=False) -> ts_float64_1 or ts_float64_2:
-    """
-    Analyze a numpy arrays to extract a tuple of useful information:
-
-        - type of raster
-        - raster.dtype if that's instance of np.ndarray
-        - raster shape
-        - min of raster value, over all axis (flattening the array)
-        - max of raster value, over all axis (flattening the array)
-        - mean of raster value, over all axis (flattening the array)
-        - median of raster value, over all axis (flattening the array)
-        - standard deviation of raster value, over all axis (flattening the array)
-        - variance of raster value, over all axis (flattening the array)
-
-    Args:
-        raster: numpy array to analyze
-        get_rms: bool to get Root Mean Square Error
-        debug: logging debug argument
-
-    Returns:
-        tuple: float values (min, max, mean, median, standard deviation, variance of raster)
-
-    """
-    std = np.nanstd(raster)
-    if get_rms:
-        try:
-            rms = np.sqrt(np.nanmean(np.square(raster)))
-        except Exception as rms_e:
-            rms = None
-            app_logger.error(f"get_stats_raster::rms_Exception:{rms_e}.")
-        app_logger.info(f"nanmin:{type(np.nanmin(raster))}.")
-        return (np.nanmin(raster), np.nanmax(raster), np.nanmean(raster), np.nanmedian(raster), std,
-                np.nanvar(raster), rms)
-    return (np.nanmin(raster), np.nanmax(raster), np.nanmean(raster), np.nanmedian(raster), np.nanstd(raster),
-            np.nanvar(raster))