[feat] (backend) add nextzen terrain xyz provider as DEM raster source, make sam predictions with RGB image made with normalized slope, elevation and curvature as channels

samgis/io/raster_helpers.py

@@ -0,0 +1,291 @@
+"""helpers for computer vision duties"""
+import numpy as np
+from numpy import ndarray
+
+from samgis import app_logger
+from samgis.utilities.type_hints import TmsTerrainProvidersNames
+
+
+def get_nextzen_terrain_rgb_formula(red: ndarray, green: ndarray, blue: ndarray) -> ndarray:
+    """
+    Compute a 32-bits 2d digital elevation model from a nextzen 'terrarium' (terrain-rgb) raster.
+    'Terrarium' format PNG tiles contain raw elevation data in meters, in Mercator projection (EPSG:3857).
+    All values are positive with a 32,768 offset, split into the red, green, and blue channels,
+    with 16 bits of integer and 8 bits of fraction. To decode:
+
+        (red * 256 + green + blue / 256) - 32768
+
+    More details on https://www.mapzen.com/blog/elevation/
+
+    Args:
+        red: red-valued channel image array
+        green: green-valued channel image array
+        blue: blue-valued channel image array
+
+    Returns:
+        ndarray: nextzen 'terrarium' 2d digital elevation model raster at 32 bits
+
+    """
+    return (red * 256 + green + blue / 256) - 32768
+
+
+def get_mapbox__terrain_rgb_formula(red: ndarray, green: ndarray, blue: ndarray) -> ndarray:
+    return ((red * 256 * 256 + green * 256 + blue) * 0.1) - 10000
+
+
+providers_terrain_rgb_formulas = {
+    TmsTerrainProvidersNames.MAPBOX_TERRAIN_TILES_NAME: get_mapbox__terrain_rgb_formula,
+    TmsTerrainProvidersNames.NEXTZEN_TERRAIN_TILES_NAME: get_nextzen_terrain_rgb_formula
+}
+
+
+def _get_2d_array_from_3d(arr: ndarray) -> ndarray:
+    return arr.reshape(arr.shape[0], arr.shape[1])
+
+
+def _channel_split(arr: ndarray) -> list[ndarray]:
+    from numpy import dsplit
+
+    return dsplit(arr, arr.shape[-1])
+
+
+def get_raster_terrain_rgb_like(arr: ndarray, xyz_provider_name, nan_value_int: int = -12000):
+    """
+    Compute a 32-bits 2d digital elevation model from a terrain-rgb raster.
+
+    Args:
+        arr: rgb raster
+        xyz_provider_name: xyz provider
+        nan_value_int: threshold int value to replace NaN
+
+    Returns:
+        ndarray: 2d digital elevation model raster at 32 bits
+    """
+    red, green, blue = _channel_split(arr)
+    dem_rgb = providers_terrain_rgb_formulas[xyz_provider_name](red, green, blue)
+    output = _get_2d_array_from_3d(dem_rgb)
+    output[output < nan_value_int] = np.NaN
+    return output
+
+
+def get_rgb_prediction_image(raster_cropped: ndarray, slope_cellsize: int, invert_image: bool = True) -> ndarray:
+    """
+    Return an RGB image from input numpy array
+    
+    Args:
+        raster_cropped: input numpy array
+        slope_cellsize: window size to calculate slope and curvature (1st and 2nd degree array derivative)
+        invert_image: 
+
+    Returns:
+        tuple of str: image filename, image path (with filename)
+    """
+    from samgis.utilities.constants import CHANNEL_EXAGGERATIONS_LIST
+
+    try:
+        slope, curvature = get_slope_curvature(raster_cropped, slope_cellsize=slope_cellsize)
+        channel0 = raster_cropped
+        channel1 = normalize_array_list(
+            [raster_cropped, slope, curvature], CHANNEL_EXAGGERATIONS_LIST, title=f"channel1_normlist")
+        channel2 = curvature
+
+        return get_rgb_image(channel0, channel1, channel2, invert_image=invert_image)
+    except ValueError as ve_get_rgb_prediction_image:
+        msg = f"ve_get_rgb_prediction_image:{ve_get_rgb_prediction_image}."
+        app_logger.error(msg)
+        raise ve_get_rgb_prediction_image
+
+
+def get_rgb_image(arr_channel0: ndarray, arr_channel1: ndarray, arr_channel2: ndarray,
+                  invert_image: bool = True) -> ndarray:
+    """
+    Return an RGB image from input R,G,B channel arrays
+
+    Args:
+        arr_channel0: channel image 0
+        arr_channel1: channel image 1
+        arr_channel2: channel image 2
+        invert_image: invert the RGB image channel order
+
+    Returns:
+        ndarray: RGB image
+
+    """
+    try:
+        # RED curvature, GREEN slope, BLUE dem, invert_image=True
+        if len(arr_channel0.shape) != 2:
+            msg = f"arr_size, wrong type:{type(arr_channel0)} or arr_size:{arr_channel0.shape}."
+            app_logger.error(msg)
+            raise ValueError(msg)
+        data_rgb = np.zeros((arr_channel0.shape[0], arr_channel0.shape[1], 3), dtype=np.uint8)
+        app_logger.debug(f"arr_container data_rgb, type:{type(data_rgb)}, arr_shape:{data_rgb.shape}.")
+        data_rgb[:, :, 0] = normalize_array(
+            arr_channel0.astype(float), high=1, norm_type="float", title=f"RGB:channel0") * 64
+        data_rgb[:, :, 1] = normalize_array(
+            arr_channel1.astype(float), high=1, norm_type="float", title=f"RGB:channel1") * 128
+        data_rgb[:, :, 2] = normalize_array(
+            arr_channel2.astype(float), high=1, norm_type="float", title=f"RGB:channel2") * 192
+        if invert_image:
+            data_rgb = np.bitwise_not(data_rgb)
+        return data_rgb
+    except ValueError as ve_get_rgb_image:
+        msg = f"ve_get_rgb_image:{ve_get_rgb_image}."
+        app_logger.error(msg)
+        raise ve_get_rgb_image
+
+
+def get_slope_curvature(dem: ndarray, slope_cellsize: int, title: str = "") -> tuple[ndarray, ndarray]:
+    """
+    Return a tuple of two numpy arrays representing slope and curvature (1st grade derivative and 2nd grade derivative)
+
+    Args:
+        dem: input numpy array
+        slope_cellsize: window size to calculate slope and curvature
+        title: array name
+
+    Returns:
+        tuple of ndarrays: slope image, curvature image
+
+    """
+
+    app_logger.info(f"dem shape:{dem.shape}, slope_cellsize:{slope_cellsize}.")
+
+    try:
+        dem = dem.astype(float)
+        app_logger.debug("get_slope_curvature:: start")
+        slope = calculate_slope(dem, slope_cellsize)
+        app_logger.debug("get_slope_curvature:: created slope raster")
+        s2c = calculate_slope(slope, slope_cellsize)
+        curvature = normalize_array(s2c, norm_type="float", title=f"SC:curvature_{title}")
+        app_logger.debug("get_slope_curvature:: created curvature raster")
+
+        return slope, curvature
+    except ValueError as ve_get_slope_curvature:
+        msg = f"ve_get_slope_curvature:{ve_get_slope_curvature}."
+        app_logger.error(msg)
+        raise ve_get_slope_curvature
+
+
+def calculate_slope(dem_array: ndarray, cell_size: int, calctype: str = "degree") -> ndarray:
+    """
+    Return a numpy array representing slope (1st grade derivative)
+
+    Args:
+        dem_array: input numpy array
+        cell_size: window size to calculate slope
+        calctype: calculus type
+
+    Returns:
+        ndarray: slope image
+
+    """
+
+    try:
+        gradx, grady = np.gradient(dem_array, cell_size)
+        dem_slope = np.sqrt(gradx ** 2 + grady ** 2)
+        if calctype == "degree":
+            dem_slope = np.degrees(np.arctan(dem_slope))
+        app_logger.debug(f"extracted slope with calctype:{calctype}.")
+        return dem_slope
+    except ValueError as ve_calculate_slope:
+        msg = f"ve_calculate_slope:{ve_calculate_slope}."
+        app_logger.error(msg)
+        raise ve_calculate_slope
+
+
+def normalize_array(arr: ndarray, high: int = 255, norm_type: str = "float", invert: bool = False, title: str = "") -> ndarray:
+    """
+    Return normalized numpy array between 0 and 'high' value. Default normalization type is int
+    
+    Args:
+        arr: input numpy array
+        high: max value to use for normalization
+        norm_type: type of normalization: could be 'float' or 'int'
+        invert: bool to choose if invert the normalized numpy array
+        title: array title name
+
+    Returns:
+        ndarray: normalized numpy array
+
+    """
+
+    h_min_arr = np.nanmin(arr)
+    h_arr_max = np.nanmax(arr)
+    try:
+        h_diff = h_arr_max - h_min_arr
+        app_logger.debug(
+            f"normalize_array:: '{title}',h_min_arr:{h_min_arr},h_arr_max:{h_arr_max},h_diff:{h_diff}, dtype:{arr.dtype}.")
+    except Exception as e_h_diff:
+        app_logger.error(f"e_h_diff:{e_h_diff}.")
+        raise e_h_diff
+
+    if check_empty_array(arr, high) or check_empty_array(arr, h_diff):
+        msg_ve = f"normalize_array::empty array '{title}',h_min_arr:{h_min_arr},h_arr_max:{h_arr_max},h_diff:{h_diff}, dtype:{arr.dtype}."
+        app_logger.error(msg_ve)
+        raise ValueError(msg_ve)
+    try:
+        normalized = high * (arr - h_min_arr) / h_diff
+        normalized = np.nanmax(normalized) - normalized if invert else normalized
+        return normalized.astype(int) if norm_type == "int" else normalized
+    except FloatingPointError as fe:
+        msg = f"normalize_array::{title}:h_arr_max:{h_arr_max},h_min_arr:{h_min_arr},fe:{fe}."
+        app_logger.error(msg)
+        raise ValueError(msg)
+
+
+def normalize_array_list(arr_list: list[ndarray], exaggerations_list: list[float] = None, title: str = "") -> ndarray:
+    """
+    Return a normalized numpy array from a list of numpy array and an optional list of exaggeration values.
+    
+    Args:
+        arr_list: list of array to use for normalization
+        exaggerations_list: list of exaggeration values
+        title: array title name
+
+    Returns:
+        ndarray: normalized numpy array
+
+    """
+
+    if not arr_list:
+        msg = f"input list can't be empty:{arr_list}."
+        app_logger.error(msg)
+        raise ValueError(msg)
+    if exaggerations_list is None:
+        exaggerations_list = list(np.ones(len(arr_list)))
+    arr_tmp = np.zeros(arr_list[0].shape)
+    for a, exaggeration in zip(arr_list, exaggerations_list):
+        app_logger.debug(f"normalize_array_list::exaggeration:{exaggeration}.")
+        arr_tmp += normalize_array(a, norm_type="float", title=f"ARRLIST:{title}.") * exaggeration
+    return arr_tmp / len(arr_list)
+
+
+def check_empty_array(arr: ndarray, val: float) -> bool:
+    """
+    Return True if the input numpy array is empy. Check if
+        - all values are all the same value (0, 1 or given 'val' input float value)
+        - all values that are not NaN are a given 'val' float value
+
+    Args:
+        arr: input numpy array
+        val: value to use for check if array is empty
+
+    Returns:
+        bool: True if the input numpy array is empty, False otherwise
+
+    """
+
+    arr_check5_tmp = np.copy(arr)
+    arr_size = arr.shape[0]
+    arr_check3 = np.ones((arr_size, arr_size))
+    check1 = np.array_equal(arr, arr_check3)
+    check2 = np.array_equal(arr, np.zeros((arr_size, arr_size)))
+    arr_check3 *= val
+    check3 = np.array_equal(arr, arr_check3)
+    arr[np.isnan(arr)] = 0
+    check4 = np.array_equal(arr, np.zeros((arr_size, arr_size)))
+    arr_check5 = np.ones((arr_size, arr_size)) * val
+    arr_check5_tmp[np.isnan(arr_check5_tmp)] = val
+    check5 = np.array_equal(arr_check5_tmp, arr_check5)
+    app_logger.debug(f"array checks:{check1}, {check2}, {check3}, {check4}, {check5}.")
+    return check1 or check2 or check3 or check4 or check5

samgis/io/{lambda_helpers.py → wrappers_helpers.py}

@@ -1,11 +1,11 @@
 """lambda helper functions"""
 from typing import Dict
-from xyzservices import providers
+from xyzservices import providers, TileProvider
 
 from samgis import app_logger
 from samgis.io.coordinates_pixel_conversion import get_latlng_to_pixel_coordinates
-from samgis.utilities.constants import CUSTOM_RESPONSE_MESSAGES
-from samgis.utilities.type_hints import ApiRequestBody, ContentTypes
+from samgis.utilities.constants import COMPLETE_URL_TILES_MAPBOX, COMPLETE_URL_TILES_NEXTZEN, CUSTOM_RESPONSE_MESSAGES
+from samgis.utilities.type_hints import ApiRequestBody, ContentTypes, TmsTerrainProvidersNames, TmsDefaultProvidersNames
 from samgis.utilities.utilities import base64_decode
 
 
@@ -152,9 +152,30 @@ def get_parsed_request_body(event: Dict or str) -> ApiRequestBody:
     return parsed_body
 
 
+mapbox_terrain_rgb = TileProvider(
+    name=TmsTerrainProvidersNames.MAPBOX_TERRAIN_TILES_NAME,
+    url=COMPLETE_URL_TILES_MAPBOX,
+    attribution=""
+)
+nextzen_terrain_rgb = TileProvider(
+    name=TmsTerrainProvidersNames.NEXTZEN_TERRAIN_TILES_NAME,
+    url=COMPLETE_URL_TILES_NEXTZEN,
+    attribution=""
+)
+
+
 def get_url_tile(source_type: str):
-    from samgis.utilities.constants import DEFAULT_TMS_NAME, DEFAULT_TMS_NAME_SHORT
+    match source_type.lower():
+        case TmsDefaultProvidersNames.DEFAULT_TILES_NAME_SHORT:
+            return providers.query_name(TmsDefaultProvidersNames.DEFAULT_TILES_NAME_SHORT)
+        case TmsTerrainProvidersNames.MAPBOX_TERRAIN_TILES_NAME:
+            return mapbox_terrain_rgb
+        case TmsTerrainProvidersNames.NEXTZEN_TERRAIN_TILES_NAME:
+            app_logger.info("nextzen_terrain_rgb:", nextzen_terrain_rgb)
+            return nextzen_terrain_rgb
 
-    if source_type.lower() == DEFAULT_TMS_NAME_SHORT:
-        return providers.query_name(DEFAULT_TMS_NAME)
     return providers.query_name(source_type)
+
+
+def check_source_type_is_terrain(source: str | TileProvider):
+    return isinstance(source, TileProvider) and source.name in list(TmsTerrainProvidersNames)

samgis/prediction_api/predictors.py

@@ -3,10 +3,14 @@ from numpy import array as np_array, uint8, zeros, ndarray
 
 from samgis import app_logger, MODEL_FOLDER
 from samgis.io.geo_helpers import get_vectorized_raster_as_geojson
+from samgis.io.raster_helpers import get_raster_terrain_rgb_like, get_rgb_prediction_image
 from samgis.io.tms2geotiff import download_extent
+from samgis.io.wrappers_helpers import check_source_type_is_terrain
 from samgis.prediction_api.sam_onnx import SegmentAnythingONNX
-from samgis.utilities.constants import MODEL_ENCODER_NAME, MODEL_DECODER_NAME, DEFAULT_TMS
-from samgis.utilities.type_hints import llist_float, dict_str_int, list_dict, tuple_ndarr_int, PIL_Image
+from samgis.utilities.constants import MODEL_ENCODER_NAME, MODEL_DECODER_NAME, DEFAULT_URL_TILES, SLOPE_CELLSIZE, \
+    DEFAULT_INPUT_SHAPE
+from samgis.utilities.type_hints import llist_float, dict_str_int, list_dict, tuple_ndarr_int, PIL_Image, \
+    TmsTerrainProvidersNames
 
 models_dict = {"fastsam": {"instance": None}}
 
@@ -16,7 +20,7 @@ def samexporter_predict(
         prompt: list_dict,
         zoom: float,
         model_name: str = "fastsam",
-        source: str = DEFAULT_TMS
+        source: str = DEFAULT_URL_TILES
 ) -> dict_str_int:
     """
     Return predictions as a geojson from a geo-referenced image using the given input prompt.
@@ -49,6 +53,14 @@ def samexporter_predict(
     pt0, pt1 = bbox
     app_logger.info(f"tile_source: {source}: downloading geo-referenced raster with bbox {bbox}, zoom {zoom}.")
     img, transform = download_extent(w=pt1[1], s=pt1[0], e=pt0[1], n=pt0[0], zoom=zoom, source=source)
+    if check_source_type_is_terrain(source):
+        app_logger.info(f"terrain-rgb like raster: transforms it into a DEM")
+        dem = get_raster_terrain_rgb_like(img, source.name)
+        # set a slope cell size proportional to the image width
+        slope_cellsize = int(img.shape[1] * SLOPE_CELLSIZE / DEFAULT_INPUT_SHAPE[1])
+        app_logger.info(f"terrain-rgb like raster: compute slope, curvature using {slope_cellsize} as cell size.")
+        img = get_rgb_prediction_image(dem, slope_cellsize)
+
     app_logger.info(
         f"img type {type(img)} with shape/size:{img.size}, transform type: {type(transform)}, transform:{transform}.")
 

samgis/prediction_api/sam_onnx.py

@@ -29,14 +29,15 @@ from cv2 import INTER_LINEAR, warpAffine
 from onnxruntime import get_available_providers, InferenceSession
 
 from samgis import app_logger
+from samgis.utilities.constants import DEFAULT_INPUT_SHAPE
 
 
 class SegmentAnythingONNX:
     """Segmentation model using SegmentAnything"""
 
     def __init__(self, encoder_model_path, decoder_model_path) -> None:
-        self.target_size = 1024
-        self.input_size = (684, 1024)
+        self.target_size = DEFAULT_INPUT_SHAPE[1]
+        self.input_size = DEFAULT_INPUT_SHAPE
 
         # Load models
         providers = get_available_providers()

samgis/utilities/constants.py

@@ -13,9 +13,6 @@ MODEL_ENCODER_NAME = "mobile_sam.encoder.onnx"
 MODEL_DECODER_NAME = "sam_vit_h_4b8939.decoder.onnx"
 TILE_SIZE = 256
 EARTH_EQUATORIAL_RADIUS = 6378137.0
-DEFAULT_TMS_NAME_SHORT = "openstreetmap"
-DEFAULT_TMS_NAME = "OpenStreetMap.Mapnik"
-DEFAULT_TMS = 'https://tile.openstreetmap.org/{z}/{x}/{y}.png'
 WKT_3857 = 'PROJCS["WGS 84 / Pseudo-Mercator",GEOGCS["WGS 84",DATUM["WGS_1984",SPHEROID["WGS 84",6378137,298.257223563,'
 WKT_3857 += 'AUTHORITY["EPSG","7030"]],AUTHORITY["EPSG","6326"]],PRIMEM["Greenwich",0,AUTHORITY["EPSG","8901"]],'
 WKT_3857 += 'UNIT["degree",0.0174532925199433,AUTHORITY["EPSG","9122"]],AUTHORITY["EPSG","4326"]],'
@@ -33,3 +30,14 @@ N_WAIT = 0
 N_MAX_RETRIES = 2
 N_CONNECTION = 2
 ZOOM_AUTO = "auto"
+DEFAULT_URL_TILES = 'https://tile.openstreetmap.org/{z}/{x}/{y}.png'
+DOMAIN_URL_TILES_MAPBOX = "api.mapbox.com"
+RELATIVE_URL_TILES_MAPBOX = "v/mapbox.terrain-rgb/{zoom}/{x}/{y}{@2x}.pngraw?access_token={TOKEN}"
+COMPLETE_URL_TILES_MAPBOX = f"https://{DOMAIN_URL_TILES_MAPBOX}/{RELATIVE_URL_TILES_MAPBOX}"
+# https://s3.amazonaws.com/elevation-tiles-prod/terrarium/13/1308/3167.png
+DOMAIN_URL_TILES_NEXTZEN = "s3.amazonaws.com"
+RELATIVE_URL_TILES_NEXTZEN = "elevation-tiles-prod/terrarium/{z}/{x}/{y}.png"  # "terrarium/{z}/{x}/{y}.png"
+COMPLETE_URL_TILES_NEXTZEN = f"https://{DOMAIN_URL_TILES_NEXTZEN}/{RELATIVE_URL_TILES_NEXTZEN}"
+CHANNEL_EXAGGERATIONS_LIST = [2.5, 1.1, 2.0]
+DEFAULT_INPUT_SHAPE = 684, 1024
+SLOPE_CELLSIZE = 61

samgis/utilities/type_hints.py

@@ -1,5 +1,5 @@
 """custom type hints"""
-from enum import IntEnum, Enum
+from enum import IntEnum, Enum, StrEnum
 from typing import TypedDict
 
 from PIL.Image import Image
@@ -7,8 +7,6 @@ from affine import Affine
 from numpy import ndarray
 from pydantic import BaseModel
 
-from samgis.utilities.constants import DEFAULT_TMS
-
 
 dict_str_int = dict[str, int]
 dict_str = dict[str]
@@ -25,6 +23,18 @@ PIL_Image = Image
 tuple_ndarray_transform = tuple[ndarray, Affine]
 
 
+class TmsDefaultProvidersNames(StrEnum):
+    """Default xyz provider names"""
+    DEFAULT_TILES_NAME_SHORT = "openstreetmap"
+    DEFAULT_TILES_NAME = "openstreetmap.mapnik"
+
+
+class TmsTerrainProvidersNames(StrEnum):
+    """Custom xyz provider names for digital elevation models"""
+    MAPBOX_TERRAIN_TILES_NAME = "mapbox.terrain-rgb"
+    NEXTZEN_TERRAIN_TILES_NAME = "nextzen.terrarium"
+
+
 class LatLngDict(BaseModel):
     """Generic geographic latitude-longitude type"""
     lat: float

wrappers/fastapi_wrapper.py

@@ -7,7 +7,7 @@ from fastapi.responses import FileResponse, JSONResponse
 from fastapi.staticfiles import StaticFiles
 
 from samgis import app_logger
-from samgis.io.lambda_helpers import get_parsed_bbox_points
+from samgis.io.wrappers_helpers import get_parsed_bbox_points
 from samgis.utilities.type_hints import ApiRequestBody
 
 app = FastAPI()

wrappers/lambda_wrapper.py

@@ -6,7 +6,7 @@ from aws_lambda_powertools.utilities.typing import LambdaContext
 from pydantic import ValidationError
 
 from samgis import app_logger
-from samgis.io.lambda_helpers import get_parsed_request_body, get_parsed_bbox_points, get_response
+from samgis.io.wrappers_helpers import get_parsed_request_body, get_parsed_bbox_points, get_response
 from samgis.prediction_api.predictors import samexporter_predict