refactor: add and remove all needed files because of switch from docker to gradio sdk and samgis-lisa

dockerfiles/dockerfile-lisa-predictions

@@ -23,7 +23,6 @@ RUN ls -l ${LAMBDA_TASK_ROOT}
 RUN ls -ld ${LAMBDA_TASK_ROOT}
 RUN ls -l ${LAMBDA_TASK_ROOT}/machine_learning_models
 RUN python -c "import sys; print(sys.path)"
-RUN python -c "import cv2"
 RUN python -c "import fastapi"
 RUN python -c "import geopandas"
 RUN python -c "import loguru"

requirements_no_versions.txt

@@ -0,0 +1,2 @@
+samgis-lisa
+samgis_lisa

samgis_lisa_on_cuda/__init__.py

@@ -1,27 +0,0 @@
-"""Get machine learning predictions from geodata raster images"""
-import os
-
-# not used here but contextily_tile is imported in samgis_lisa_on_cuda.io.tms2geotiff
-from contextily import tile as contextily_tile
-from pathlib import Path
-from samgis_lisa_on_cuda.utilities.constants import SERVICE_NAME
-
-ROOT = Path(globals().get("__file__", "./_")).absolute().parent.parent
-PROJECT_ROOT_FOLDER = Path(os.getenv("PROJECT_ROOT_FOLDER", ROOT))
-WORKDIR = Path(os.getenv("WORKDIR", ROOT))
-MODEL_FOLDER_PROJECT_ROOT_FOLDER = Path(PROJECT_ROOT_FOLDER / "machine_learning_models")
-MODEL_FOLDER = Path(os.getenv("MODEL_FOLDER", MODEL_FOLDER_PROJECT_ROOT_FOLDER))
-
-IS_AWS_LAMBDA = bool(os.getenv("IS_AWS_LAMBDA", ""))
-
-if IS_AWS_LAMBDA:
-    try:
-        from aws_lambda_powertools import Logger
-
-        app_logger = Logger(service=SERVICE_NAME)
-    except ModuleNotFoundError:
-        print("this should be AWS LAMBDA environment but we miss the required aws lambda powertools package")
-else:
-    from samgis_core.utilities.fastapi_logger import setup_logging
-
-    app_logger = setup_logging(debug=True)

samgis_lisa_on_cuda/__version__.py

@@ -1,4 +0,0 @@
-import importlib.metadata
-
-
-__version__ = importlib.metadata.version(__package__ or __name__)

samgis_lisa_on_cuda/io/__init__.py

@@ -1 +0,0 @@
-"""input/output helpers functions"""

samgis_lisa_on_cuda/io/coordinates_pixel_conversion.py

@@ -1,99 +0,0 @@
-"""functions useful to convert to/from latitude-longitude coordinates to pixel image coordinates"""
-from samgis_core.utilities.type_hints import TupleFloat, TupleFloatAny
-
-from samgis_lisa_on_cuda import app_logger
-from samgis_lisa_on_cuda.utilities.constants import TILE_SIZE, EARTH_EQUATORIAL_RADIUS
-from samgis_lisa_on_cuda.utilities.type_hints import ImagePixelCoordinates
-from samgis_lisa_on_cuda.utilities.type_hints import LatLngDict
-
-
-def _get_latlng2pixel_projection(latlng: LatLngDict) -> ImagePixelCoordinates:
-    from math import log, pi, sin
-
-    app_logger.debug(f"latlng: {type(latlng)}, value:{latlng}.")
-    app_logger.debug(f'latlng lat: {type(latlng.lat)}, value:{latlng.lat}.')
-    app_logger.debug(f'latlng lng: {type(latlng.lng)}, value:{latlng.lng}.')
-    try:
-        sin_y: float = sin(latlng.lat * pi / 180)
-        app_logger.debug(f"sin_y, #1:{sin_y}.")
-        sin_y = min(max(sin_y, -0.9999), 0.9999)
-        app_logger.debug(f"sin_y, #2:{sin_y}.")
-        x = TILE_SIZE * (0.5 + latlng.lng / 360)
-        app_logger.debug(f"x:{x}.")
-        y = TILE_SIZE * (0.5 - log((1 + sin_y) / (1 - sin_y)) / (4 * pi))
-        app_logger.debug(f"y:{y}.")
-
-        return {"x": x, "y": y}
-    except Exception as e_get_latlng2pixel_projection:
-        app_logger.error(f'args type:{type(latlng)}, {latlng}.')
-        app_logger.exception(f'e_get_latlng2pixel_projection:{e_get_latlng2pixel_projection}.', exc_info=True)
-        raise e_get_latlng2pixel_projection
-
-
-def _get_point_latlng_to_pixel_coordinates(latlng: LatLngDict, zoom: int | float) -> ImagePixelCoordinates:
-    from math import floor
-
-    try:
-        world_coordinate: ImagePixelCoordinates = _get_latlng2pixel_projection(latlng)
-        app_logger.debug(f"world_coordinate:{world_coordinate}.")
-        scale: int = pow(2, zoom)
-        app_logger.debug(f"scale:{scale}.")
-        return ImagePixelCoordinates(
-            x=floor(world_coordinate["x"] * scale),
-            y=floor(world_coordinate["y"] * scale)
-        )
-    except Exception as e_format_latlng_to_pixel_coordinates:
-        app_logger.error(f'latlng type:{type(latlng)}, {latlng}.')
-        app_logger.error(f'zoom type:{type(zoom)}, {zoom}.')
-        app_logger.exception(f'e_format_latlng_to_pixel_coordinates:{e_format_latlng_to_pixel_coordinates}.',
-                             exc_info=True)
-        raise e_format_latlng_to_pixel_coordinates
-
-
-def get_latlng_to_pixel_coordinates(
-        latlng_origin_ne: LatLngDict,
-        latlng_origin_sw: LatLngDict,
-        latlng_current_point: LatLngDict,
-        zoom: int | float,
-        k: str
-) -> ImagePixelCoordinates:
-    """
-    Parse the input request lambda event
-
-    Args:
-        latlng_origin_ne: NE latitude-longitude origin point
-        latlng_origin_sw: SW latitude-longitude origin point
-        latlng_current_point: latitude-longitude prompt point
-        zoom: Level of detail
-        k: prompt type
-
-    Returns:
-        ImagePixelCoordinates: pixel image coordinate point
-    """
-    app_logger.debug(f"latlng_origin - {k}: {type(latlng_origin_ne)}, value:{latlng_origin_ne}.")
-    app_logger.debug(f"latlng_current_point - {k}: {type(latlng_current_point)}, value:{latlng_current_point}.")
-    latlng_map_origin_ne = _get_point_latlng_to_pixel_coordinates(latlng_origin_ne, zoom)
-    latlng_map_origin_sw = _get_point_latlng_to_pixel_coordinates(latlng_origin_sw, zoom)
-    latlng_map_current_point = _get_point_latlng_to_pixel_coordinates(latlng_current_point, zoom)
-    diff_coord_x = abs(latlng_map_origin_sw["x"] - latlng_map_current_point["x"])
-    diff_coord_y = abs(latlng_map_origin_ne["y"] - latlng_map_current_point["y"])
-    point = ImagePixelCoordinates(x=diff_coord_x, y=diff_coord_y)
-    app_logger.debug(f"point type - {k}: {point}.")
-    return point
-
-
-def _from4326_to3857(lat: float, lon: float) -> TupleFloat or TupleFloatAny:
-    from math import radians, log, tan
-
-    x_tile: float = radians(lon) * EARTH_EQUATORIAL_RADIUS
-    y_tile: float = log(tan(radians(45 + lat / 2.0))) * EARTH_EQUATORIAL_RADIUS
-    return x_tile, y_tile
-
-
-def _deg2num(lat: float, lon: float, zoom: int):
-    from math import radians, pi, asinh, tan
-
-    n = 2 ** zoom
-    x_tile = ((lon + 180) / 360 * n)
-    y_tile = (1 - asinh(tan(radians(lat))) / pi) * n / 2
-    return x_tile, y_tile

samgis_lisa_on_cuda/io/geo_helpers.py

@@ -1,91 +0,0 @@
-"""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

samgis_lisa_on_cuda/io/raster_helpers.py

@@ -1,331 +0,0 @@
-"""helpers for computer vision duties"""
-import numpy as np
-from numpy import ndarray, bitwise_not
-from rasterio import open as rasterio_open
-
-from samgis_lisa_on_cuda import PROJECT_ROOT_FOLDER
-from samgis_lisa_on_cuda import app_logger
-from samgis_lisa_on_cuda.utilities.constants import OUTPUT_CRS_STRING
-from samgis_lisa_on_cuda.utilities.type_hints import XYZTerrainProvidersNames
-
-
-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 = {
-    XYZTerrainProvidersNames.MAPBOX_TERRAIN_TILES_NAME: get_mapbox__terrain_rgb_formula,
-    XYZTerrainProvidersNames.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_lisa_on_cuda.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="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="RGB:channel0") * 64
-        data_rgb[:, :, 1] = normalize_array(
-            arr_channel1.astype(float), high=1, norm_type="float", title="RGB:channel1") * 128
-        data_rgb[:, :, 2] = normalize_array(
-            arr_channel2.astype(float), high=1, norm_type="float", title="RGB:channel2") * 192
-        if invert_image:
-            app_logger.debug(f"data_rgb:{type(data_rgb)}, {data_rgb.dtype}.")
-            data_rgb = 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
-
-    """
-    np.seterr("raise")
-
-    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 ValueError(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
-
-
-def write_raster_png(arr, transform, prefix: str, suffix: str, folder_output_path="/tmp"):
-    from pathlib import Path
-    from rasterio.plot import reshape_as_raster
-
-    output_filename = Path(folder_output_path) / f"{prefix}_{suffix}.png"
-
-    with rasterio_open(
-            output_filename, 'w', driver='PNG',
-            height=arr.shape[0],
-            width=arr.shape[1],
-            count=3,
-            dtype=str(arr.dtype),
-            crs=OUTPUT_CRS_STRING,
-            transform=transform) as dst:
-        dst.write(reshape_as_raster(arr))
-    app_logger.info(f"written:{output_filename} as PNG, use {OUTPUT_CRS_STRING} as CRS.")
-
-
-def write_raster_tiff(arr, transform, prefix: str, suffix: str, folder_output_path="/tmp"):
-    from pathlib import Path
-    output_filename = Path(folder_output_path) / f"{prefix}_{suffix}.tiff"
-
-    with rasterio_open(
-            output_filename, 'w', driver='GTiff',
-            height=arr.shape[0],
-            width=arr.shape[1],
-            count=1,
-            dtype=str(arr.dtype),
-            crs=OUTPUT_CRS_STRING,
-            transform=transform) as dst:
-        dst.write(arr, 1)
-    app_logger.info(f"written:{output_filename} as TIFF, use {OUTPUT_CRS_STRING} as CRS.")

samgis_lisa_on_cuda/io/tms2geotiff.py

@@ -1,181 +0,0 @@
-import os
-
-from numpy import ndarray
-from samgis_core.utilities.type_hints import TupleFloat
-from xyzservices import TileProvider
-
-from samgis_lisa_on_cuda import app_logger
-from samgis_lisa_on_cuda.utilities.constants import (OUTPUT_CRS_STRING, DRIVER_RASTERIO_GTIFF, N_MAX_RETRIES, N_CONNECTION, N_WAIT,
-                                        ZOOM_AUTO, BOOL_USE_CACHE)
-from samgis_lisa_on_cuda.utilities.type_hints import tuple_ndarray_transform
-
-
-bool_use_cache = int(os.getenv("BOOL_USE_CACHE", BOOL_USE_CACHE))
-n_connection = int(os.getenv("N_CONNECTION", N_CONNECTION))
-n_max_retries = int(os.getenv("N_MAX_RETRIES", N_MAX_RETRIES))
-n_wait = int(os.getenv("N_WAIT", N_WAIT))
-zoom_auto_string = os.getenv("ZOOM_AUTO", ZOOM_AUTO)
-
-
-def download_extent(w: float, s: float, e: float, n: float, zoom: int or str = zoom_auto_string,
-                    source: TileProvider or str = None,
-                    wait: int = n_wait, max_retries: int = n_max_retries, n_connections: int = n_connection,
-                    use_cache: bool = bool_use_cache) -> tuple_ndarray_transform:
-    """
-    Download, merge and crop a list of tiles into a single geo-referenced image or a raster geodata
-
-    Args:
-        w: West edge
-        s: South edge
-        e: East edge
-        n: North edge
-        zoom: Level of detail
-        source: The tile source: web tile provider or path to local file. The web tile provider can be in the form of
-            a :class:`xyzservices.TileProvider` object or a URL. The placeholders for the XYZ in the URL need to be
-            `{x}`, `{y}`, `{z}`, respectively. For local file paths, the file is read with `rasterio` and all bands are
-            loaded into the basemap. IMPORTANT: tiles are assumed to be in the Spherical Mercator projection
-            (EPSG:3857), unless the `crs` keyword is specified.
-        wait: if the tile API is rate-limited, the number of seconds to wait
-            between a failed request and the next try
-        max_retries: total number of rejected requests allowed before contextily will stop trying to fetch more tiles
-            from a rate-limited API.
-        n_connections: Number of connections for downloading tiles in parallel. Be careful not to overload the tile
-            server and to check the tile provider's terms of use before increasing this value. E.g., OpenStreetMap has
-            a max. value of 2 (https://operations.osmfoundation.org/policies/tiles/). If allowed to download in
-            parallel, a recommended value for n_connections is 16, and should never be larger than 64.
-        use_cache: If False, caching of the downloaded tiles will be disabled. This can be useful in resource
-            constrained environments, especially when using n_connections > 1, or when a tile provider's terms of use
-            don't allow caching.
-
-    Returns:
-        parsed request input
-    """
-    try:
-        from samgis_lisa_on_cuda import contextily_tile
-        from samgis_lisa_on_cuda.io.coordinates_pixel_conversion import _from4326_to3857
-
-        app_logger.info(f"connection number:{n_connections}, type:{type(n_connections)}.")
-        app_logger.info(f"zoom:{zoom}, type:{type(zoom)}.")
-        app_logger.debug(f"download raster from source:{source} with bounding box w:{w}, s:{s}, e:{e}, n:{n}.")
-        app_logger.debug(f"types w:{type(w)}, s:{type(s)}, e:{type(e)}, n:{type(n)}.")
-        downloaded_raster, bbox_raster = contextily_tile.bounds2img(
-            w, s, e, n, zoom=zoom, source=source, ll=True, wait=wait, max_retries=max_retries,
-            n_connections=n_connections, use_cache=use_cache)
-        xp0, yp0 = _from4326_to3857(n, e)
-        xp1, yp1 = _from4326_to3857(s, w)
-        cropped_image_ndarray, cropped_transform = crop_raster(yp1, xp1, yp0, xp0, downloaded_raster, bbox_raster)
-        return cropped_image_ndarray, cropped_transform
-    except Exception as e_download_extent:
-        app_logger.exception(f"e_download_extent:{e_download_extent}.", exc_info=True)
-        raise e_download_extent
-
-
-def crop_raster(w: float, s: float, e: float, n: float, raster: ndarray, raster_bbox: TupleFloat,
-                crs: str = OUTPUT_CRS_STRING, driver: str = DRIVER_RASTERIO_GTIFF) -> tuple_ndarray_transform:
-    """
-    Crop a raster using given bounding box (w, s, e, n) values
-
-    Args:
-        w: cropping west edge
-        s: cropping south edge
-        e: cropping east edge
-        n: cropping north edge
-        raster: raster image to crop
-        raster_bbox: bounding box of raster to crop
-        crs: The coordinate reference system. Required in 'w' or 'w+' modes, it is ignored in 'r' or 'r+' modes.
-        driver: A short format driver name (e.g. "GTiff" or "JPEG") or a list of such names (see GDAL docs at
-            https://gdal.org/drivers/raster/index.html ). In 'w' or 'w+' modes a single name is required. In 'r' or 'r+'
-            modes the driver can usually be omitted. Registered drivers will be tried sequentially until a match is
-            found. When multiple drivers are available for a format such as JPEG2000, one of them can be selected by
-            using this keyword argument.
-
-    Returns:
-        cropped raster with its Affine transform
-    """
-    try:
-        from rasterio.io import MemoryFile
-        from rasterio.mask import mask as rio_mask
-        from shapely.geometry import Polygon
-        from geopandas import GeoSeries
-
-        app_logger.debug(f"raster: type {type(raster)}, raster_ext:{type(raster_bbox)}, {raster_bbox}.")
-        img_to_save, transform = get_transform_raster(raster, raster_bbox)
-        img_height, img_width, number_bands = img_to_save.shape
-        # https://rasterio.readthedocs.io/en/latest/topics/memory-files.html
-        with MemoryFile() as rio_mem_file:
-            app_logger.debug("writing raster in-memory to crop it with rasterio.mask.mask()")
-            with rio_mem_file.open(
-                    driver=driver,
-                    height=img_height,
-                    width=img_width,
-                    count=number_bands,
-                    dtype=str(img_to_save.dtype.name),
-                    crs=crs,
-                    transform=transform,
-            ) as src_raster_rw:
-                for band in range(number_bands):
-                    src_raster_rw.write(img_to_save[:, :, band], band + 1)
-            app_logger.debug("cropping raster in-memory with rasterio.mask.mask()")
-            with rio_mem_file.open() as src_raster_ro:
-                shapes_crop_polygon = Polygon([(n, e), (s, e), (s, w), (n, w), (n, e)])
-                shapes_crop = GeoSeries([shapes_crop_polygon])
-                app_logger.debug(f"cropping with polygon::{shapes_crop_polygon}.")
-                cropped_image, cropped_transform = rio_mask(src_raster_ro, shapes=shapes_crop, crop=True)
-                cropped_image_ndarray = reshape_as_image(cropped_image)
-        app_logger.info(f"cropped image::{cropped_image_ndarray.shape}.")
-        return cropped_image_ndarray, cropped_transform
-    except Exception as e_crop_raster:
-        try:
-            app_logger.error(f"raster type:{type(raster)}.")
-            app_logger.error(f"raster shape:{raster.shape}, dtype:{raster.dtype}.")
-        except Exception as e_shape_dtype:
-            app_logger.exception(f"raster shape or dtype not found:{e_shape_dtype}.", exc_info=True)
-        app_logger.exception(f"e_crop_raster:{e_crop_raster}.", exc_info=True)
-        raise e_crop_raster
-
-
-def get_transform_raster(raster: ndarray, raster_bbox: TupleFloat) -> tuple_ndarray_transform:
-    """
-    Convert the input raster image to RGB and extract the Affine 
-
-    Args:
-        raster: raster image to geo-reference
-        raster_bbox: bounding box of raster to crop
-
-    Returns:
-        rgb raster image and its Affine transform
-    """
-    try:
-        from rasterio.transform import from_origin
-        from numpy import array as np_array, linspace as np_linspace, uint8 as np_uint8
-        from PIL.Image import fromarray
-
-        app_logger.debug(f"raster: type {type(raster)}, raster_ext:{type(raster_bbox)}, {raster_bbox}.")
-        rgb = fromarray(np_uint8(raster)).convert('RGB')
-        np_rgb = np_array(rgb)
-        img_height, img_width, _ = np_rgb.shape
-
-        min_x, max_x, min_y, max_y = raster_bbox
-        app_logger.debug(f"raster rgb shape:{np_rgb.shape}, raster rgb bbox {raster_bbox}.")
-        x = np_linspace(min_x, max_x, img_width)
-        y = np_linspace(min_y, max_y, img_height)
-        res_x = (x[-1] - x[0]) / img_width
-        res_y = (y[-1] - y[0]) / img_height
-        transform = from_origin(x[0] - res_x / 2, y[-1] + res_y / 2, res_x, res_y)
-        return np_rgb, transform
-    except Exception as e_get_transform_raster:
-        app_logger.error(f"arguments raster: {type(raster)}, {raster}.")
-        app_logger.error(f"arguments raster_bbox: {type(raster_bbox)}, {raster_bbox}.")
-        app_logger.exception(f"e_get_transform_raster:{e_get_transform_raster}.", exc_info=True)
-        raise e_get_transform_raster
-
-
-def reshape_as_image(arr):
-    try:
-        from numpy import swapaxes
-
-        return swapaxes(swapaxes(arr, 0, 2), 0, 1)
-    except Exception as e_reshape_as_image:
-        app_logger.error(f"arguments: {type(arr)}, {arr}.")
-        app_logger.exception(f"e_reshape_as_image:{e_reshape_as_image}.", exc_info=True)
-        raise e_reshape_as_image

samgis_lisa_on_cuda/io/wrappers_helpers.py

@@ -1,262 +0,0 @@
-"""lambda helper functions"""
-import logging
-from sys import stdout
-from typing import Dict
-
-import loguru
-from xyzservices import providers, TileProvider
-
-from lisa_on_cuda.utils.app_helpers import get_cleaned_input
-from samgis_lisa_on_cuda import app_logger
-from samgis_lisa_on_cuda.io.coordinates_pixel_conversion import get_latlng_to_pixel_coordinates
-from samgis_lisa_on_cuda.utilities.constants import COMPLETE_URL_TILES_MAPBOX, COMPLETE_URL_TILES_NEXTZEN, CUSTOM_RESPONSE_MESSAGES
-from samgis_lisa_on_cuda.utilities.type_hints import ApiRequestBody, ContentTypes, XYZTerrainProvidersNames, \
-    XYZDefaultProvidersNames, StringPromptApiRequestBody
-from samgis_core.utilities.utilities import base64_decode
-
-
-def get_response(status: int, start_time: float, request_id: str, response_body: Dict = None) -> str:
-    """
-    Response composer
-
-    Args:
-        status: status response
-        start_time: request start time (float)
-        request_id: str
-        response_body: dict we embed into our response
-
-    Returns:
-        json response
-
-    """
-    from json import dumps
-    from time import time
-
-    app_logger.debug(f"response_body:{response_body}.")
-    response_body["duration_run"] = time() - start_time
-    response_body["message"] = CUSTOM_RESPONSE_MESSAGES[status]
-    response_body["request_id"] = request_id
-
-    response = {
-        "statusCode": status,
-        "header": {"Content-Type": ContentTypes.APPLICATION_JSON},
-        "body": dumps(response_body),
-        "isBase64Encoded": False
-    }
-    app_logger.debug(f"response type:{type(response)} => {response}.")
-    return dumps(response)
-
-
-def get_parsed_bbox_points_with_string_prompt(request_input: StringPromptApiRequestBody) -> Dict:
-    """
-        Parse the raw input request into bbox, prompt string and zoom
-
-        Args:
-            request_input: input dict
-
-        Returns:
-            dict with bounding box, prompt string and zoom
-        """
-
-    app_logger.info(f"try to parsing input request {request_input}...")
-
-    bbox = request_input.bbox
-    app_logger.debug(f"request bbox: {type(bbox)}, value:{bbox}.")
-    ne = bbox.ne
-    sw = bbox.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)]
-    new_zoom = int(request_input.zoom)
-    cleaned_prompt = get_cleaned_input(request_input.string_prompt)
-
-    app_logger.debug(f"bbox => {bbox}.")
-    app_logger.debug(f'request_input-prompt cleaned => {cleaned_prompt}.')
-
-    app_logger.info("unpacking elaborated request...")
-    return {
-        "bbox": [ne_latlng, sw_latlng],
-        "prompt": cleaned_prompt,
-        "zoom": new_zoom,
-        "source": get_url_tile(request_input.source_type)
-    }
-
-
-def get_parsed_bbox_points_with_dictlist_prompt(request_input: ApiRequestBody) -> Dict:
-    """
-    Parse the raw input request into bbox, prompt and zoom
-
-    Args:
-        request_input: input dict
-
-    Returns:
-        dict with bounding box, prompt and zoom
-    """
-
-    app_logger.info(f"try to parsing input request {request_input}...")
-
-    bbox = request_input.bbox
-    app_logger.debug(f"request bbox: {type(bbox)}, value:{bbox}.")
-    ne = bbox.ne
-    sw = bbox.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)]
-    new_zoom = int(request_input.zoom)
-    new_prompt_list = _get_parsed_prompt_list(ne, sw, new_zoom, request_input.prompt)
-
-    app_logger.debug(f"bbox => {bbox}.")
-    app_logger.debug(f'request_input-prompt updated => {new_prompt_list}.')
-
-    app_logger.info("unpacking elaborated request...")
-    return {
-        "bbox": [ne_latlng, sw_latlng],
-        "prompt": new_prompt_list,
-        "zoom": new_zoom,
-        "source": get_url_tile(request_input.source_type)
-    }
-
-
-def _get_parsed_prompt_list(bbox_ne, bbox_sw, zoom, prompt_list):
-    new_prompt_list = []
-    for prompt in prompt_list:
-        app_logger.debug(f"current prompt: {type(prompt)}, value:{prompt}.")
-        new_prompt = {"type": prompt.type.value}
-        if prompt.type == "point":
-            new_prompt_data = _get_new_prompt_data_point(bbox_ne, bbox_sw, prompt, zoom)
-            new_prompt["label"] = prompt.label.value
-        elif prompt.type == "rectangle":
-            new_prompt_data = _get_new_prompt_data_rectangle(bbox_ne, bbox_sw, prompt, zoom)
-        else:
-            msg = "Valid prompt type: 'point' or 'rectangle', not '{}'. Check ApiRequestBody parsing/validation."
-            raise TypeError(msg.format(prompt.type))
-        app_logger.debug(f"new_prompt_data: {type(new_prompt_data)}, value:{new_prompt_data}.")
-        new_prompt["data"] = new_prompt_data
-        new_prompt_list.append(new_prompt)
-    return new_prompt_list
-
-
-def _get_new_prompt_data_point(bbox_ne, bbox_sw, prompt, zoom):
-    current_point = get_latlng_to_pixel_coordinates(bbox_ne, bbox_sw, prompt.data, zoom, prompt.type)
-    app_logger.debug(f"current prompt: {type(current_point)}, value:{current_point}, label: {prompt.label}.")
-    return [current_point['x'], current_point['y']]
-
-
-def _get_new_prompt_data_rectangle(bbox_ne, bbox_sw, prompt, zoom):
-    current_point_ne = get_latlng_to_pixel_coordinates(bbox_ne, bbox_sw, prompt.data.ne, zoom, prompt.type)
-    app_logger.debug(
-        f"rectangle:: current_point_ne prompt: {type(current_point_ne)}, value:{current_point_ne}.")
-    current_point_sw = get_latlng_to_pixel_coordinates(bbox_ne, bbox_sw, prompt.data.sw, zoom, prompt.type)
-    app_logger.debug(
-        f"rectangle:: current_point_sw prompt: {type(current_point_sw)}, value:{current_point_sw}.")
-    # correct order for rectangle prompt
-    return [
-        current_point_sw["x"],
-        current_point_ne["y"],
-        current_point_ne["x"],
-        current_point_sw["y"]
-    ]
-
-
-def get_parsed_request_body(event: Dict or str) -> ApiRequestBody:
-    """
-    Validator for the raw input request lambda event
-
-    Args:
-        event: input dict
-
-    Returns:
-        parsed request input
-    """
-    from json import dumps, loads
-    from logging import getLevelName
-
-    def _get_current_log_level(logger: loguru.logger) -> [str, loguru._logger.Level]:
-        levels = logger._core.levels
-        current_log_level = logger._core.min_level
-        level_filt = [l for l in levels.items() if l[1].no == current_log_level]
-        return level_filt[0]
-
-    app_logger.info(f"event:{dumps(event)}...")
-    try:
-        raw_body = event["body"]
-    except Exception as e_constants1:
-        app_logger.error(f"e_constants1:{e_constants1}.")
-        raw_body = event
-    app_logger.debug(f"raw_body, #1: {type(raw_body)}, {raw_body}...")
-    if isinstance(raw_body, str):
-        body_decoded_str = base64_decode(raw_body)
-        app_logger.debug(f"body_decoded_str: {type(body_decoded_str)}, {body_decoded_str}...")
-        raw_body = loads(body_decoded_str)
-    app_logger.info(f"body, #2: {type(raw_body)}, {raw_body}...")
-
-    parsed_body = ApiRequestBody.model_validate(raw_body)
-    log_level = "DEBUG" if parsed_body.debug else "INFO"
-    app_logger.remove()
-    app_logger.add(stdout, level=log_level)
-    try:
-        current_log_level_name, _ = _get_current_log_level(app_logger)
-        app_logger.warning(f"set log level to {getLevelName(current_log_level_name)}.")
-    except Exception as ex:
-        print("failing setting parsing bbox, logger is ok? ex:", ex, "#")
-
-    return parsed_body
-
-
-mapbox_terrain_rgb = TileProvider(
-    name=XYZTerrainProvidersNames.MAPBOX_TERRAIN_TILES_NAME,
-    url=COMPLETE_URL_TILES_MAPBOX,
-    attribution=""
-)
-nextzen_terrain_rgb = TileProvider(
-    name=XYZTerrainProvidersNames.NEXTZEN_TERRAIN_TILES_NAME,
-    url=COMPLETE_URL_TILES_NEXTZEN,
-    attribution=""
-)
-
-
-def get_url_tile(source_type: str):
-    try:
-        match source_type.lower():
-            case XYZDefaultProvidersNames.DEFAULT_TILES_NAME_SHORT:
-                return providers.query_name(XYZDefaultProvidersNames.DEFAULT_TILES_NAME)
-            case XYZTerrainProvidersNames.MAPBOX_TERRAIN_TILES_NAME:
-                return mapbox_terrain_rgb
-            case XYZTerrainProvidersNames.NEXTZEN_TERRAIN_TILES_NAME:
-                app_logger.info("nextzen_terrain_rgb:", nextzen_terrain_rgb)
-                return nextzen_terrain_rgb
-            case _:
-                return providers.query_name(source_type)
-    except ValueError as ve:
-        from pydantic_core import ValidationError
-
-        app_logger.error("ve:", str(ve))
-        raise ValidationError(ve)
-
-
-def check_source_type_is_terrain(source: str | TileProvider):
-    return isinstance(source, TileProvider) and source.name in list(XYZTerrainProvidersNames)
-
-
-def get_source_name(source: str | TileProvider) -> str | bool:
-    try:
-        match source.lower():
-            case XYZDefaultProvidersNames.DEFAULT_TILES_NAME_SHORT:
-                source_output = providers.query_name(XYZDefaultProvidersNames.DEFAULT_TILES_NAME)
-            case _:
-                source_output = providers.query_name(source)
-        if isinstance(source_output, str):
-            return source_output
-        try:
-            source_dict = dict(source_output)
-            app_logger.info(f"source_dict:{type(source_dict)}, {'name' in source_dict}, source_dict:{source_dict}.")
-            return source_dict["name"]
-        except KeyError as ke:
-            app_logger.error(f"ke:{ke}.")
-    except ValueError as ve:
-        app_logger.info(f"source name::{source}, ve:{ve}.")
-    app_logger.info(f"source name::{source}.")
-
-    return False

samgis_lisa_on_cuda/prediction_api/__init__.py

@@ -1 +0,0 @@
-"""functions useful to handle machine learning models"""

samgis_lisa_on_cuda/prediction_api/global_models.py

@@ -1,6 +0,0 @@
-models_dict = {
-    "fastsam": {"instance": None},
-    "lisa": {"inference": None}
-}
-embedding_dict = {}
-

samgis_lisa_on_cuda/prediction_api/lisa.py

@@ -1,77 +0,0 @@
-from datetime import datetime
-
-from lisa_on_cuda.utils import app_helpers
-from samgis_core.utilities.type_hints import LlistFloat, DictStrInt
-from samgis_lisa_on_cuda import app_logger
-from samgis_lisa_on_cuda.io.geo_helpers import get_vectorized_raster_as_geojson
-from samgis_lisa_on_cuda.io.raster_helpers import write_raster_png, write_raster_tiff
-from samgis_lisa_on_cuda.io.tms2geotiff import download_extent
-from samgis_lisa_on_cuda.prediction_api.global_models import models_dict
-from samgis_lisa_on_cuda.utilities.constants import DEFAULT_URL_TILES
-
-msg_write_tmp_on_disk = "found option to write images and geojson output..."
-
-
-def lisa_predict(
-        bbox: LlistFloat,
-        prompt: str,
-        zoom: float,
-        inference_function_name_key: str = "lisa",
-        source: str = DEFAULT_URL_TILES,
-        source_name: str = None
-) -> DictStrInt:
-    """
-    Return predictions as a geojson from a geo-referenced image using the given input prompt.
-
-    1. if necessary instantiate a segment anything machine learning instance model
-    2. download a geo-referenced raster image delimited by the coordinates bounding box (bbox)
-    3. get a prediction image from the segment anything instance model using the input prompt
-    4. get a geo-referenced geojson from the prediction image
-
-    Args:
-        bbox: coordinates bounding box
-        prompt: machine learning input prompt
-        zoom: Level of detail
-        inference_function_name_key: machine learning model name
-        source: xyz
-        source_name: name of tile provider
-
-    Returns:
-        Affine transform
-    """
-    from os import getenv
-
-    app_logger.info("start lisa inference...")
-    if models_dict[inference_function_name_key]["inference"] is None:
-        app_logger.info(f"missing inference function {inference_function_name_key}, instantiating it now!")
-        parsed_args = app_helpers.parse_args([])
-        inference_fn = app_helpers.get_inference_model_by_args(parsed_args)
-        models_dict[inference_function_name_key]["inference"] = inference_fn
-    app_logger.debug(f"using a {inference_function_name_key} instance model...")
-    inference_fn = models_dict[inference_function_name_key]["inference"]
-
-    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)
-    app_logger.info(
-        f"img type {type(img)} with shape/size:{img.size}, transform type: {type(transform)}, transform:{transform}.")
-    folder_write_tmp_on_disk = getenv("WRITE_TMP_ON_DISK", "")
-    prefix = f"w{pt1[1]},s{pt1[0]},e{pt0[1]},n{pt0[0]}_"
-    if bool(folder_write_tmp_on_disk):
-        now = datetime.now().strftime('%Y%m%d_%H%M%S')
-        app_logger.info(msg_write_tmp_on_disk + f"with coords {prefix}, shape:{img.shape}, {len(img.shape)}.")
-        if img.shape and len(img.shape) == 2:
-            write_raster_tiff(img, transform, f"{source_name}_{prefix}_{now}_", f"raw_tiff", folder_write_tmp_on_disk)
-        if img.shape and len(img.shape) == 3 and img.shape[2] == 3:
-            write_raster_png(img, transform, f"{source_name}_{prefix}_{now}_", f"raw_img", folder_write_tmp_on_disk)
-    else:
-        app_logger.info("keep all temp data in memory...")
-
-    app_logger.info(f"source_name:{source_name}, source_name type:{type(source_name)}.")
-    embedding_key = f"{source_name}_z{zoom}_{prefix}"
-    _, mask, output_string = inference_fn(prompt, img, app_logger, embedding_key)
-    # app_logger.info(f"created {n_predictions} masks, preparing conversion to geojson...")
-    return {
-        "output_string": output_string,
-        **get_vectorized_raster_as_geojson(mask, transform)
-    }

samgis_lisa_on_cuda/prediction_api/predictors.py

@@ -1,72 +0,0 @@
-"""functions using machine learning instance model(s)"""
-from samgis_lisa_on_cuda import app_logger, MODEL_FOLDER
-from samgis_lisa_on_cuda.io.geo_helpers import get_vectorized_raster_as_geojson
-from samgis_lisa_on_cuda.io.raster_helpers import get_raster_terrain_rgb_like, get_rgb_prediction_image
-from samgis_lisa_on_cuda.io.tms2geotiff import download_extent
-from samgis_lisa_on_cuda.io.wrappers_helpers import check_source_type_is_terrain
-from samgis_lisa_on_cuda.prediction_api.global_models import models_dict, embedding_dict
-from samgis_lisa_on_cuda.utilities.constants import DEFAULT_URL_TILES, SLOPE_CELLSIZE
-from samgis_core.prediction_api.sam_onnx import SegmentAnythingONNX, get_raster_inference_with_embedding_from_dict
-from samgis_core.utilities.constants import MODEL_ENCODER_NAME, MODEL_DECODER_NAME, DEFAULT_INPUT_SHAPE
-from samgis_core.utilities.type_hints import LlistFloat, DictStrInt, ListDict
-
-
-def samexporter_predict(
-        bbox: LlistFloat,
-        prompt: ListDict,
-        zoom: float,
-        model_name_key: str = "fastsam",
-        source: str = DEFAULT_URL_TILES,
-        source_name: str = None
-) -> DictStrInt:
-    """
-    Return predictions as a geojson from a geo-referenced image using the given input prompt.
-
-    1. if necessary instantiate a segment anything machine learning instance model
-    2. download a geo-referenced raster image delimited by the coordinates bounding box (bbox)
-    3. get a prediction image from the segment anything instance model using the input prompt
-    4. get a geo-referenced geojson from the prediction image
-
-    Args:
-        bbox: coordinates bounding box
-        prompt: machine learning input prompt
-        zoom: Level of detail
-        model_name_key: machine learning model name
-        source: xyz
-        source_name: name of tile provider
-
-    Returns:
-        Affine transform
-    """
-    if models_dict[model_name_key]["instance"] is None:
-        app_logger.info(f"missing instance model {model_name_key}, instantiating it now!")
-        model_instance = SegmentAnythingONNX(
-            encoder_model_path=MODEL_FOLDER / MODEL_ENCODER_NAME,
-            decoder_model_path=MODEL_FOLDER / MODEL_DECODER_NAME
-        )
-        models_dict[model_name_key]["instance"] = model_instance
-    app_logger.debug(f"using a {model_name_key} instance model...")
-    models_instance = models_dict[model_name_key]["instance"]
-
-    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("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}.")
-    app_logger.info(f"source_name:{source_name}, source_name type:{type(source_name)}.")
-    embedding_key = f"{source_name}_z{zoom}_w{pt1[1]},s{pt1[0]},e{pt0[1]},n{pt0[0]}"
-    mask, n_predictions = get_raster_inference_with_embedding_from_dict(
-        img, prompt, models_instance, model_name_key, embedding_key, embedding_dict)
-    app_logger.info(f"created {n_predictions} masks, preparing conversion to geojson...")
-    return {
-        "n_predictions": n_predictions,
-        **get_vectorized_raster_as_geojson(mask, transform)
-    }

samgis_lisa_on_cuda/utilities/__init__.py

@@ -1 +0,0 @@
-"""various helpers functions"""

samgis_lisa_on_cuda/utilities/constants.py

@@ -1,39 +0,0 @@
-"""Project constants"""
-INPUT_CRS_STRING = "EPSG:4326"
-OUTPUT_CRS_STRING = "EPSG:3857"
-DRIVER_RASTERIO_GTIFF = "GTiff"
-CUSTOM_RESPONSE_MESSAGES = {
-    200: "ok",
-    400: "Bad Request",
-    422: "Missing required parameter",
-    500: "Internal server error"
-}
-TILE_SIZE = 256
-EARTH_EQUATORIAL_RADIUS = 6378137.0
-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"]],'
-WKT_3857 += 'PROJECTION["Mercator_1SP"],PARAMETER["central_meridian",0],PARAMETER["scale_factor",1],'
-WKT_3857 += 'PARAMETER["false_easting",0],PARAMETER["false_northing",0],UNIT["metre",1,AUTHORITY["EPSG","9001"]],'
-WKT_3857 += 'AXIS["X",EAST],AXIS["Y",NORTH],EXTENSION["PROJ4","+proj=merc +a=6378137 +b=6378137 +lat_ts=0.0 +lon_0=0.0 '
-WKT_3857 += '+x_0=0.0 +y_0=0 +k=1.0 +units=m +nadgrids=@null +wktext +no_defs"],AUTHORITY["EPSG","3857"]]'
-SERVICE_NAME = "sam-gis"
-DEFAULT_LOG_LEVEL = 'INFO'
-RETRY_DOWNLOAD = 3
-TIMEOUT_DOWNLOAD = 60
-CALLBACK_INTERVAL_DOWNLOAD = 0.05
-BOOL_USE_CACHE = True
-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]
-SLOPE_CELLSIZE = 61

samgis_lisa_on_cuda/utilities/session_logger.py

@@ -1,65 +0,0 @@
-import contextvars
-import logging
-from functools import wraps
-from typing import Callable, Tuple
-
-logging_uuid = contextvars.ContextVar("uuid")
-default_formatter = '%(asctime)s | %(uuid)s [%(pathname)s:%(module)s %(lineno)d] %(levelname)s | %(message)s'
-
-
-loggingType = logging.CRITICAL | logging.ERROR | logging.WARNING | logging.INFO | logging.DEBUG
-
-
-def setup_logging(
-        debug: bool = False, formatter: str = default_formatter, name: str = "logger"
-) -> Tuple[logging, contextvars.ContextVar]:
-    """
-    Create a logging instance with log string formatter.
-
-    Args:
-        debug: logging debug argument
-        formatter: log string formatter
-        name: logger name
-
-    Returns:
-        Logger
-
-    """
-
-    old_factory = logging.getLogRecordFactory()
-
-    def record_factory(*args, **kwargs):
-        record = old_factory(*args, **kwargs)
-        record.uuid = logging_uuid.get("uuid")
-        if isinstance(record.msg, str):
-            record.msg = record.msg.replace("\\", "\\\\").replace("\n", "\\n")
-        return record
-
-    logging.setLogRecordFactory(record_factory)
-    logging.basicConfig(level=logging.DEBUG, format=default_formatter, force=True)
-
-    logger = logging.getLogger(name=name)
-
-    # create a console handler
-    ch = logging.StreamHandler()
-    ch.setLevel(logging.DEBUG)
-
-    # create formatter and add to the console
-    formatter = logging.Formatter(formatter)
-    ch.setFormatter(formatter)
-
-    # add the console handler to logger
-    logger.addHandler(ch)
-    return logger, logging_uuid
-
-
-def set_uuid_logging(func: Callable) -> Callable:
-    @wraps(func)
-    def wrapper(*args, **kwargs):
-        import uuid
-
-        current_uuid = f"{uuid.uuid4()}"
-        logging_uuid.set(current_uuid)
-        return func(*args, **kwargs)
-
-    return wrapper

samgis_lisa_on_cuda/utilities/type_hints.py

@@ -1,111 +0,0 @@
-"""custom type hints"""
-from enum import IntEnum, Enum
-from typing import TypedDict
-
-from affine import Affine
-from numpy import ndarray
-from pydantic import BaseModel
-
-
-tuple_ndarray_transform = tuple[ndarray, Affine]
-
-
-class XYZDefaultProvidersNames(str, Enum):
-    """Default xyz provider names"""
-    DEFAULT_TILES_NAME_SHORT = "openstreetmap"
-    DEFAULT_TILES_NAME = "openstreetmap.mapnik"
-
-
-class XYZTerrainProvidersNames(str, Enum):
-    """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
-    lng: float
-
-
-class ContentTypes(str, Enum):
-    """Segment Anything: validation point prompt type"""
-    APPLICATION_JSON = "application/json"
-    TEXT_PLAIN = "text/plain"
-    TEXT_HTML = "text/html"
-
-
-class PromptPointType(str, Enum):
-    """Segment Anything: validation point prompt type"""
-    point = "point"
-
-
-class PromptRectangleType(str, Enum):
-    """Segment Anything: validation rectangle prompt type"""
-    rectangle = "rectangle"
-
-
-class PromptLabel(IntEnum):
-    """Valid prompt label type"""
-    EXCLUDE = 0
-    INCLUDE = 1
-
-
-class ImagePixelCoordinates(TypedDict):
-    """Image pixel coordinates type"""
-    x: int
-    y: int
-
-
-class RawBBox(BaseModel):
-    """Input lambda bbox request type (not yet parsed)"""
-    ne: LatLngDict
-    sw: LatLngDict
-
-
-class RawPromptPoint(BaseModel):
-    """Input lambda prompt request of type 'PromptPointType' - point (not yet parsed)"""
-    type: PromptPointType
-    data: LatLngDict
-    label: PromptLabel
-
-
-class RawPromptRectangle(BaseModel):
-    """Input lambda prompt request of type 'PromptRectangleType' - rectangle (not yet parsed)"""
-    type: PromptRectangleType
-    data: RawBBox
-
-    def get_type_str(self):
-        return self.type
-
-
-class ApiRequestBody(BaseModel):
-    """Input lambda request validator type (not yet parsed)"""
-    id: str = ""
-    bbox: RawBBox
-    prompt: list[RawPromptPoint | RawPromptRectangle]
-    zoom: int | float
-    source_type: str = "OpenStreetMap.Mapnik"
-    debug: bool = False
-
-
-class StringPromptApiRequestBody(BaseModel):
-    """Input lambda request validator type (not yet parsed)"""
-    id: str = ""
-    bbox: RawBBox
-    string_prompt: str
-    zoom: int | float
-    source_type: str = "OpenStreetMap.Mapnik"
-    debug: bool = False
-
-
-class ApiResponseBodyFailure(BaseModel):
-    duration_run: float
-    message: str
-    request_id: str
-
-
-class ApiResponseBodySuccess(ApiResponseBodyFailure):
-    n_predictions: int
-    geojson: str
-    n_shapes_geojson: int

scripts/create_folders_and_variables_if_not_exists.py

@@ -1,55 +0,0 @@
-import json
-import logging
-import os
-from pathlib import Path
-
-
-def stats_pathname(pathname: Path | str):
-    current_pathname = Path(pathname)
-    return current_pathname.is_dir()
-
-
-def create_folder_if_not_exists(pathname: Path | str):
-    current_pathname = Path(pathname)
-    try:
-        print(f"Pathname exists? {current_pathname.exists()}, That's a folder? {current_pathname.is_dir()}...")
-        logging.info(f"Pathname exists? {current_pathname.exists()}, That's a folder? {current_pathname.is_dir()}...")
-        current_pathname.unlink(missing_ok=True)
-    except PermissionError as pe:
-        print(f"permission denied on removing pathname before folder creation:{pe}.")
-        logging.error(f"permission denied on removing pathname before folder creation:{pe}.")
-    except IsADirectoryError as errdir:
-        print(f"that's a directory:{errdir}.")
-        logging.error(f"that's a directory:{errdir}.")
-
-    print(f"Creating pathname: {current_pathname} ...")
-    logging.info(f"Creating pathname: {current_pathname} ...")
-    current_pathname.mkdir(mode=0o770, parents=True, exist_ok=True)
-
-    print(f"assertion: pathname exists and is a folder: {current_pathname} ...")
-    logging.info(f"assertion: pathname exists and is a folder: {current_pathname} ...")
-    assert current_pathname.is_dir()
-
-
-def run_folder_creation():
-    folders_string = os.getenv("FOLDERS_MAP")
-    try:
-        folders_dict = json.loads(folders_string)
-        for folder_env_ref, folder_env_path in folders_dict.items():
-            print(f"folder_env_ref:{folder_env_ref}, folder_env_path:{folder_env_path}.")
-            logging.info(f"folder_env_ref:{folder_env_ref}, folder_env_path:{folder_env_path}.")
-            create_folder_if_not_exists(folder_env_path)
-            print("========")
-            assert os.getenv(folder_env_ref) == folder_env_path
-    except (json.JSONDecodeError, TypeError) as jde:
-        print(f"jde:{jde}.")
-        logging.error(f"jde:{jde}.")
-        print("double check your variables, e.g. for mispelling like 'FOLDER_MAP'...")
-        logging.info("double check your variables, e.g. for mispelling like 'FOLDER_MAP' instead than 'FOLDERS_MAP'...")
-        for k_env, v_env in dict(os.environ).items():
-            print(f"{k_env}, v_env:{v_env}.")
-            logging.info(f"{k_env}, v_env:{v_env}.")
-
-
-if __name__ == '__main__':
-    run_folder_creation()

scripts/extract-openapi-fastapi.py

@@ -3,12 +3,13 @@ import argparse
 import json
 import logging
 import sys
+from pathlib import Path
 
 import yaml
 from uvicorn.importer import import_from_string
 
-from samgis_lisa_on_cuda import PROJECT_ROOT_FOLDER
 
+PROJECT_ROOT_FOLDER = Path(globals().get("__file__", "./_")).absolute().parent.parent
 parser = argparse.ArgumentParser(prog="extract-openapi-fastapi.py")
 parser.add_argument("app",       help='App import string. Eg. "main:app"', default="main:app")
 parser.add_argument("--app-dir", help="Directory containing the app", default=None)

scripts/extract-openapi-lambda.py

@@ -1,13 +0,0 @@
-import json
-
-from samgis_lisa_on_cuda import PROJECT_ROOT_FOLDER
-
-if __name__ == '__main__':
-    from samgis_lisa_on_cuda.utilities.type_hints import ApiRequestBody, ApiResponseBodyFailure, ApiResponseBodySuccess
-
-    with open(PROJECT_ROOT_FOLDER / "docs" / "specs" / "openapi_lambda_wip.json", "w") as output_json:
-        json.dump({
-            "ApiRequestBody": ApiRequestBody.model_json_schema(),
-            "ApiResponseBodyFailure": ApiResponseBodyFailure.model_json_schema(),
-            "ApiResponseBodySuccess": ApiResponseBodySuccess.model_json_schema()
-        }, output_json)

tests/__init__.py

@@ -1,6 +1,7 @@
-from samgis_lisa_on_cuda import PROJECT_ROOT_FOLDER
+from pathlib import Path
 
 
+PROJECT_ROOT_FOLDER = Path(globals().get("__file__", "./_")).absolute().parent.parent
 TEST_ROOT_FOLDER = PROJECT_ROOT_FOLDER / "tests"
 TEST_EVENTS_FOLDER = TEST_ROOT_FOLDER / "events"
 LOCAL_URL_TILE = "http://localhost:8000/lambda_handler/{z}/{x}/{y}.png"

tests/io/test_coordinates_pixel_conversion.py

@@ -1,27 +0,0 @@
-import json
-
-from samgis_lisa_on_cuda.io.coordinates_pixel_conversion import get_latlng_to_pixel_coordinates
-from samgis_lisa_on_cuda.utilities.type_hints import LatLngDict
-from tests import TEST_EVENTS_FOLDER
-
-
-def test_get_latlng_to_pixel_coordinates():
-    name_fn = "get_latlng_to_pixel_coordinates"
-
-    with open(TEST_EVENTS_FOLDER / f"{name_fn}.json") as tst_json:
-        inputs_outputs = json.load(tst_json)
-        for k, input_output in inputs_outputs.items():
-            print(f"k:{k}")
-            current_input = input_output["input"]
-            zoom = current_input["zoom"]
-            latlng_origin_ne = LatLngDict.model_validate(current_input["latlng_origin_ne"])
-            latlng_origin_sw = LatLngDict.model_validate(current_input["latlng_origin_sw"])
-            latlng_current_point = LatLngDict.model_validate(current_input["latlng_current_point"])
-            output = get_latlng_to_pixel_coordinates(
-                latlng_origin_ne=latlng_origin_ne,
-                latlng_origin_sw=latlng_origin_sw,
-                latlng_current_point=latlng_current_point,
-                zoom=zoom,
-                k=k
-            )
-            assert output == input_output["output"]

tests/io/test_geo_helpers.py

@@ -1,103 +0,0 @@
-import json
-import unittest
-import numpy as np
-import shapely
-
-from samgis_lisa_on_cuda.io.geo_helpers import load_affine_transformation_from_matrix
-from tests import TEST_EVENTS_FOLDER
-
-
-class TestGeoHelpers(unittest.TestCase):
-    def test_load_affine_transformation_from_matrix(self):
-        name_fn = "samexporter_predict"
-
-        expected_output = {
-            'europe': (
-                1524458.6551710723, 0.0, 152.87405657035242, 4713262.318571913, -762229.3275855362, -2356860.470370812
-            ),
-            'north_america': (
-                -13855281.495084189, 0.0, 1222.9924525628194, 6732573.451358326, 6927640.747542094, -3368121.214358007
-            ),
-            'oceania': (
-                7269467.138033403, 0.0, 9783.93962050256, -166326.9735485418, -3634733.5690167015, 68487.57734351706
-            ),
-            'south_america': (
-                -7922544.351904369, 0.0, 305.74811314070394, -5432228.234830927, 3961272.1759521845, 2715655.4952457524
-            )}
-
-        with open(TEST_EVENTS_FOLDER / f"{name_fn}.json") as tst_json:
-            inputs_outputs = json.load(tst_json)
-            for k, input_output in inputs_outputs.items():
-                print(f"k:{k}.")
-
-                output = load_affine_transformation_from_matrix(input_output["input"]["matrix"])
-                assert output.to_shapely() == expected_output[k]
-
-    def test_load_affine_transformation_from_matrix_value_error(self):
-        name_fn = "samexporter_predict"
-        with open(TEST_EVENTS_FOLDER / f"{name_fn}.json") as tst_json:
-            inputs_outputs = json.load(tst_json)
-            with self.assertRaises(ValueError):
-                try:
-                    io_value_error = inputs_outputs["europe"]["input"]["matrix"][:5]
-                    load_affine_transformation_from_matrix(io_value_error)
-                except ValueError as ve:
-                    print(f"ve:{ve}.")
-                    self.assertEqual(str(ve), "Expected 6 coefficients, found 5; argument type: <class 'list'>.")
-                    raise ve
-
-    def test_load_affine_transformation_from_matrix_exception(self):
-        name_fn = "samexporter_predict"
-        with open(TEST_EVENTS_FOLDER / f"{name_fn}.json") as tst_json:
-            inputs_outputs = json.load(tst_json)
-            with self.assertRaises(Exception):
-                try:
-                    io_exception = inputs_outputs["europe"]["input"]["matrix"]
-                    io_exception[0] = "ciao"
-                    load_affine_transformation_from_matrix(io_exception)
-                except Exception as e:
-                    print(f"e:{e}.")
-                    self.assertEqual(str(e), "exception:could not convert string to float: 'ciao', "
-                                             "check https://github.com/rasterio/affine project for updates")
-                    raise e
-
-    def test_get_vectorized_raster_as_geojson_ok(self):
-        from rasterio.transform import Affine
-        from samgis_lisa_on_cuda.io.geo_helpers import get_vectorized_raster_as_geojson
-
-        name_fn = "samexporter_predict"
-
-        with open(TEST_EVENTS_FOLDER / f"{name_fn}.json") as tst_json:
-            inputs_outputs = json.load(tst_json)
-            for k, input_output in inputs_outputs.items():
-                print(f"k:{k}.")
-                mask = np.load(TEST_EVENTS_FOLDER / name_fn / k / "mask.npy")
-
-                transform = Affine.from_gdal(*input_output["input"]["matrix"])
-                output = get_vectorized_raster_as_geojson(mask=mask, transform=transform)
-                assert output["n_shapes_geojson"] == input_output["output"]["n_shapes_geojson"]
-                output_geojson = shapely.from_geojson(output["geojson"])
-                expected_output_geojson = shapely.from_geojson(input_output["output"]["geojson"])
-                assert shapely.equals_exact(output_geojson, expected_output_geojson, tolerance=0.000006)
-
-    def test_get_vectorized_raster_as_geojson_fail(self):
-        from samgis_lisa_on_cuda.io.geo_helpers import get_vectorized_raster_as_geojson
-
-        name_fn = "samexporter_predict"
-
-        with open(TEST_EVENTS_FOLDER / f"{name_fn}.json") as tst_json:
-            inputs_outputs = json.load(tst_json)
-            for k, input_output in inputs_outputs.items():
-                print(f"k:{k}.")
-                mask = np.load(TEST_EVENTS_FOLDER / name_fn / k / "mask.npy")
-
-                # Could be also another generic Exception, here we intercept TypeError caused by wrong matrix input on
-                # rasterio.Affine.from_gdal() wrapped by get_affine_transform_from_gdal()
-                with self.assertRaises(IndexError):
-                    try:
-                        wrong_matrix = 1.0,
-                        get_vectorized_raster_as_geojson(mask=mask, transform=wrong_matrix)
-                    except IndexError as te:
-                        print(f"te:{te}.")
-                        self.assertEqual(str(te), 'tuple index out of range')
-                        raise te

tests/io/test_raster_helpers.py

@@ -1,255 +0,0 @@
-import unittest
-from unittest.mock import patch
-import numpy as np
-
-from samgis_core.utilities.utilities import hash_calculate
-from samgis_lisa_on_cuda.io import raster_helpers
-
-
-def get_three_channels(size=5, param1=1000, param2=3, param3=-88):
-    arr_base = np.arange(size*size).reshape(size, size) / size**2
-    channel_0 = arr_base * param1
-    channel_1 = arr_base * param2
-    channel_2 = arr_base * param3
-    return channel_0, channel_1, channel_2
-
-
-def helper_bell(size=10, param1=0.1, param2=2):
-    x = np.linspace(-size, size, num=size**2)
-    y = np.linspace(-size, size, num=size**2)
-    x, y = np.meshgrid(x, y)
-    return np.exp(-param1 * x ** param2 - param1 * y ** param2)
-
-
-arr_5x5x5 = np.arange(125).reshape((5, 5, 5)) / 25
-arr = np.arange(25).resize((5, 5))
-channel0, channel1, channel2 = get_three_channels()
-z = helper_bell()
-slope_z_cellsize3, curvature_z_cellsize3 = raster_helpers.get_slope_curvature(z, slope_cellsize=3)
-
-
-class Test(unittest.TestCase):
-
-    def test_get_rgb_prediction_image_real(self):
-        output = raster_helpers.get_rgb_prediction_image(z, slope_cellsize=61, invert_image=True)
-        hash_output = hash_calculate(output)
-        assert hash_output == b'QpQ9yxgCLw9cf3klNFKNFXIDHaSkuiZxkbpeQApR8pA='
-        output = raster_helpers.get_rgb_prediction_image(z, slope_cellsize=61, invert_image=False)
-        hash_output = hash_calculate(output)
-        assert hash_output == b'Y+iXO9w/sKzNVOw2rBh2JrVGJUFRqaa8/0F9hpevmLs='
-
-    @patch.object(raster_helpers, "get_slope_curvature")
-    @patch.object(raster_helpers, "normalize_array_list")
-    @patch.object(raster_helpers, "get_rgb_image")
-    def test_get_rgb_prediction_image_mocked(self, get_rgb_image_mocked, normalize_array_list, get_slope_curvature):
-        local_arr = np.array(z * 100, dtype=np.uint8)
-
-        get_slope_curvature.return_value = slope_z_cellsize3, curvature_z_cellsize3
-        normalize_array_list.side_effect = None
-        get_rgb_image_mocked.return_value = np.bitwise_not(local_arr)
-        output = raster_helpers.get_rgb_prediction_image(local_arr, slope_cellsize=61, invert_image=True)
-        hash_output = hash_calculate(output)
-        assert hash_output == b'BPIyVH64RgVunj42EuQAx4/v59Va8ZAjcMnuiGNqTT0='
-        get_rgb_image_mocked.return_value = local_arr
-        output = raster_helpers.get_rgb_prediction_image(local_arr, slope_cellsize=61, invert_image=False)
-        hash_output = hash_calculate(output)
-        assert hash_output == b'XX54sdLQQUrhkUHT6ikQZYSloMYDSfh/AGITDq6jnRM='
-
-    @patch.object(raster_helpers, "get_slope_curvature")
-    def test_get_rgb_prediction_image_value_error(self, get_slope_curvature):
-        msg = "this is a value error"
-        get_slope_curvature.side_effect = ValueError(msg)
-
-        with self.assertRaises(ValueError):
-            try:
-                raster_helpers.get_rgb_prediction_image(arr, slope_cellsize=3)
-            except ValueError as ve:
-                self.assertEqual(str(ve), msg)
-                raise ve
-
-    def test_get_rgb_image(self):
-        output = raster_helpers.get_rgb_image(channel0, channel1, channel2, invert_image=True)
-        hash_output = hash_calculate(output)
-        assert hash_output == b'YVnRWla5Ptfet6reSfM+OEIsGytLkeso6X+CRs34YHk='
-        output = raster_helpers.get_rgb_image(channel0, channel1, channel2, invert_image=False)
-        hash_output = hash_calculate(output)
-        assert hash_output == b'LC/kIZGUZULSrwwSXCeP1My2spTZdW9D7LH+tltwERs='
-
-    def test_get_rgb_image_value_error_1(self):
-        with self.assertRaises(ValueError):
-            try:
-                raster_helpers.get_rgb_image(arr_5x5x5, arr_5x5x5, arr_5x5x5, invert_image=True)
-            except ValueError as ve:
-                self.assertEqual(f"arr_size, wrong type:{type(arr_5x5x5)} or arr_size:{arr_5x5x5.shape}.", str(ve))
-                raise ve
-
-    def test_get_rgb_image_value_error2(self):
-        arr_0 = np.arange(25).reshape((5, 5))
-        arr_1 = np.arange(4).reshape((2, 2))
-        with self.assertRaises(ValueError):
-            try:
-                raster_helpers.get_rgb_image(arr_0, arr_1, channel2, invert_image=True)
-            except ValueError as ve:
-                self.assertEqual('could not broadcast input array from shape (2,2) into shape (5,5)', str(ve))
-                raise ve
-
-    def test_get_slope_curvature(self):
-        slope_output, curvature_output = raster_helpers.get_slope_curvature(z, slope_cellsize=3)
-        hash_curvature = hash_calculate(curvature_output)
-        hash_slope = hash_calculate(slope_output)
-        assert hash_curvature == b'LAL9JFOjJP9D6X4X3fVCpnitx9VPM9drS5YMHwMZ3iE='
-        assert hash_slope == b'IYf6x4G0lmR47j6HRS5kUYWdtmimhLz2nak8py75nwc='
-
-    def test_get_slope_curvature_value_error(self):
-        from samgis_lisa_on_cuda.io import raster_helpers
-
-        with self.assertRaises(ValueError):
-            try:
-                raster_helpers.get_slope_curvature(np.array(1), slope_cellsize=3)
-            except ValueError as ve:
-                self.assertEqual('not enough values to unpack (expected 2, got 0)', str(ve))
-                raise ve
-
-    def test_calculate_slope(self):
-        slope_output = raster_helpers.calculate_slope(z, cell_size=3)
-        hash_output = hash_calculate(slope_output)
-        assert hash_output == b'IYf6x4G0lmR47j6HRS5kUYWdtmimhLz2nak8py75nwc='
-
-    def test_calculate_slope_value_error(self):
-        with self.assertRaises(ValueError):
-            try:
-                raster_helpers.calculate_slope(np.array(1), cell_size=3)
-            except ValueError as ve:
-                self.assertEqual('not enough values to unpack (expected 2, got 0)', str(ve))
-                raise ve
-
-    def test_normalize_array(self):
-        def check_ndarrays_almost_equal(cls, arr1, arr2, places, check_type="float", check_ndiff=1):
-            count_abs_diff = 0
-            for list00, list01 in zip(arr1.tolist(), arr2.tolist()):
-                for el00, el01 in zip(list00, list01):
-                    ndiff = abs(el00 - el01)
-                    if el00 != el01:
-                        count_abs_diff += 1
-                    if check_type == "float":
-                        cls.assertAlmostEqual(el00, el01, places=places)
-                    cls.assertLess(ndiff, check_ndiff)  # cls.assertTrue(ndiff < check_ndiff)
-            print("count_abs_diff:", count_abs_diff)
-
-        normalized_array = raster_helpers.normalize_array(z)
-        hash_output = hash_calculate(normalized_array)
-        assert hash_output == b'MPkQwiiQa5NxL7LDvCS9V143YUEJT/Qh1aNEKc/Ehvo='
-
-        mult_variable = 3.423
-        test_array_input = np.arange(256).reshape((16, 16))
-        test_array_output = raster_helpers.normalize_array(test_array_input * mult_variable)
-        check_ndarrays_almost_equal(self, test_array_output, test_array_input, places=8)
-
-        test_array_output1 = raster_helpers.normalize_array(test_array_input * mult_variable, high=128, norm_type="int")
-        o = np.arange(256).reshape((16, 16)) / 2
-        expected_array_output1 = o.astype(int)
-        check_ndarrays_almost_equal(
-            self, test_array_output1, expected_array_output1, places=2, check_type="int", check_ndiff=2)
-
-    @patch.object(np, "nanmin")
-    @patch.object(np, "nanmax")
-    def test_normalize_array_floating_point_error_mocked(self, nanmax_mocked, nanmin_mocked):
-        nanmax_mocked.return_value = 100
-        nanmin_mocked.return_value = 100
-
-        with self.assertRaises(ValueError):
-            try:
-                raster_helpers.normalize_array(
-                    np.arange(25).reshape((5, 5))
-                )
-            except ValueError as ve:
-                self.assertEqual(
-                    "normalize_array:::h_arr_max:100,h_min_arr:100,fe:divide by zero encountered in divide.",
-                    str(ve)
-                )
-                raise ve
-
-    @patch.object(np, "nanmin")
-    @patch.object(np, "nanmax")
-    def test_normalize_array_exception_error_mocked(self, nanmax_mocked, nanmin_mocked):
-        nanmax_mocked.return_value = 100
-        nanmin_mocked.return_value = np.NaN
-
-        with self.assertRaises(ValueError):
-            try:
-                raster_helpers.normalize_array(
-                    np.arange(25).reshape((5, 5))
-                )
-            except ValueError as ve:
-                self.assertEqual("cannot convert float NaN to integer", str(ve))
-                raise ve
-
-    def test_normalize_array_value_error(self):
-        with self.assertRaises(ValueError):
-            try:
-                raster_helpers.normalize_array(
-                    np.zeros((5, 5))
-                )
-            except ValueError as ve:
-                self.assertEqual(
-                    "normalize_array::empty array '',h_min_arr:0.0,h_arr_max:0.0,h_diff:0.0, " 'dtype:float64.',
-                    str(ve)
-                )
-                raise ve
-
-    def test_normalize_array_list(self):
-        normalized_array = raster_helpers.normalize_array_list([channel0, channel1, channel2])
-        hash_output = hash_calculate(normalized_array)
-        assert hash_output == b'+6IbhIpyb3vPElTgqqPkQdIR0umf4uFP2c7t5IaBVvI='
-
-        test_norm_list_output2 = raster_helpers.normalize_array_list(
-            [channel0, channel1, channel2], exaggerations_list=[2.0, 3.0, 5.0])
-        hash_variable2 = hash_calculate(test_norm_list_output2)
-        assert hash_variable2 == b'yYCYWCKO3i8NYsWk/wgYOzSRRLSLUprEs7mChJkdL+A='
-
-    def test_normalize_array_list_value_error(self):
-        with self.assertRaises(ValueError):
-            try:
-                raster_helpers.normalize_array_list([])
-            except ValueError as ve:
-                self.assertEqual("input list can't be empty:[].", str(ve))
-                raise ve
-
-    def test_check_empty_array(self):
-        a = np.zeros((10, 10))
-        b = np.ones((10, 10))
-        c = np.ones((10, 10)) * 2
-        d = np.zeros((10, 10))
-        d[1, 1] = np.nan
-        e = np.ones((10, 10)) * 3
-        e[1, 1] = np.nan
-
-        self.assertTrue(raster_helpers.check_empty_array(a, 999))
-        self.assertTrue(raster_helpers.check_empty_array(b, 0))
-        self.assertTrue(raster_helpers.check_empty_array(c, 2))
-        self.assertTrue(raster_helpers.check_empty_array(d, 0))
-        self.assertTrue(raster_helpers.check_empty_array(e, 3))
-        self.assertFalse(raster_helpers.check_empty_array(z, 3))
-
-    def test_get_nextzen_terrain_rgb_formula(self):
-        output = raster_helpers.get_nextzen_terrain_rgb_formula(channel0, channel1, channel2)
-        hash_output = hash_calculate(output)
-        assert hash_output == b'3KJ81YKmQRdccRZARbByfwo1iMVLj8xxz9mfsWki/qA='
-
-    def test_get_mapbox__terrain_rgb_formula(self):
-        output = raster_helpers.get_mapbox__terrain_rgb_formula(channel0, channel1, channel2)
-        hash_output = hash_calculate(output)
-        assert hash_output == b'RU7CcoKoR3Fkh5LE+m48DHRVUy/vGq6UgfOFUMXx07M='
-
-    def test_get_raster_terrain_rgb_like(self):
-        from samgis_lisa_on_cuda.utilities.type_hints import XYZTerrainProvidersNames
-
-        arr_input = raster_helpers.get_rgb_image(channel0, channel1, channel2, invert_image=True)
-        output_nextzen = raster_helpers.get_raster_terrain_rgb_like(
-            arr_input, XYZTerrainProvidersNames.NEXTZEN_TERRAIN_TILES_NAME)
-        hash_nextzen = hash_calculate(output_nextzen)
-        assert hash_nextzen == b'+o2OTJliJkkBoqiAIGnhJ4s0xoLQ4MxHOvevLhNxysE='
-        output_mapbox = raster_helpers.get_raster_terrain_rgb_like(
-            arr_input, XYZTerrainProvidersNames.MAPBOX_TERRAIN_TILES_NAME)
-        hash_mapbox = hash_calculate(output_mapbox)
-        assert hash_mapbox == b'zWmekyKrpnmHnuDACnveCJl+o4GuhtHJmGlRDVwsce4='

tests/io/test_tms2geotiff.py

@@ -1,138 +0,0 @@
-import unittest
-
-import numpy as np
-from samgis_core.utilities.utilities import hash_calculate
-
-from samgis_lisa_on_cuda import app_logger
-from samgis_lisa_on_cuda.io.tms2geotiff import download_extent
-from tests import LOCAL_URL_TILE, TEST_EVENTS_FOLDER
-
-
-input_bbox = [[39.036252959636606, 15.040283203125002], [38.302869955150044, 13.634033203125002]]
-
-
-class TestTms2geotiff(unittest.TestCase):
-    # def test_download_extent_simple_source(self):
-    #     from rasterio import Affine
-    #     from xyzservices import TileProvider
-    #     from tests.local_tiles_http_server import LocalTilesHttpServer
-    #
-    #     listen_port = 8000
-    #
-    #     with LocalTilesHttpServer.http_server("localhost", listen_port, directory=TEST_EVENTS_FOLDER):
-    #         pt0, pt1 = input_bbox
-    #         zoom = 10
-    #
-    #         n_lat = pt0[0]
-    #         e_lng = pt0[1]
-    #         s_lat = pt1[0]
-    #         w_lng = pt1[1]
-    #
-    #         source = TileProvider(name="local_tile_provider", url=LOCAL_URL_TILE, attribution="")
-    #         img, matrix = download_extent(w=w_lng, s=s_lat, e=e_lng, n=n_lat, zoom=zoom, source=source)
-    #         app_logger.info(f"# DOWNLOAD ENDED, shape: {img.shape} #")
-    #         np_img = np.ascontiguousarray(img)
-    #         output_hash = hash_calculate(np_img)
-    #         assert output_hash == b'UmbkwbPJpRT1XXcLnLUapUDP320w7YhS/AmT3H7u+b4='
-    #         assert Affine.to_gdal(matrix) == (
-    #             1517657.1966021745, 152.8740565703525, 0.0, 4726942.266183584, 0.0, -152.87405657034955)
-
-    def test_download_extent_source_with_parameter(self):
-        from rasterio import Affine
-        from xyzservices import TileProvider
-        from tests.local_tiles_http_server import LocalTilesHttpServer
-
-        listen_port = 8000
-
-        with LocalTilesHttpServer.http_server("localhost", listen_port, directory=TEST_EVENTS_FOLDER):
-            pt0, pt1 = input_bbox
-            zoom = 10
-
-            n_lat = pt0[0]
-            e_lng = pt0[1]
-            s_lat = pt1[0]
-            w_lng = pt1[1]
-
-            local_url = "http://localhost:8000/{parameter}/{z}/{x}/{y}.png"
-            download_extent_args_no_parameter = {"name": "local_tile_provider", "url": LOCAL_URL_TILE, "attribution": ""}
-            download_extent_args = {
-                "no_parameter": download_extent_args_no_parameter,
-                "with_parameter": {"url": local_url, "parameter": "lambda_handler", **download_extent_args_no_parameter}
-            }
-            for _args_names, _args in download_extent_args.items():
-                app_logger.info(f"args_names:{_args_names}.")
-                source = TileProvider(**_args)
-                img, matrix = download_extent(w=w_lng, s=s_lat, e=e_lng, n=n_lat, zoom=zoom, source=source)
-                app_logger.info(f"# DOWNLOAD ENDED, shape: {img.shape} #")
-                np_img = np.ascontiguousarray(img)
-                output_hash = hash_calculate(np_img)
-                assert output_hash == b'UmbkwbPJpRT1XXcLnLUapUDP320w7YhS/AmT3H7u+b4='
-                assert Affine.to_gdal(matrix) == (
-                    1517657.1966021745, 152.8740565703525, 0.0, 4726942.266183584, 0.0, -152.87405657034955)
-
-    def test_download_extent_source_with_parameter_key_error(self):
-        from xyzservices import TileProvider
-
-        with self.assertRaises(KeyError):
-            try:
-                pt0, pt1 = input_bbox
-                zoom = 10
-
-                n_lat = pt0[0]
-                e_lng = pt0[1]
-                s_lat = pt1[0]
-                w_lng = pt1[1]
-
-                local_url_tile2 = "http://localhost:8000/{parameter}/{z}/{x}/{y}.png"
-                source = TileProvider(name="local_tile_provider", url=local_url_tile2, attribution="")
-                download_extent(w=w_lng, s=s_lat, e=e_lng, n=n_lat, zoom=zoom, source=source)
-            except KeyError as ke:
-                assert str(ke) == "'parameter'"
-                raise ke
-
-    def test_download_extent_io_error1(self):
-
-        with self.assertRaises(Exception):
-            try:
-                pt0, pt1 = input_bbox
-                zoom = 10
-
-                n_lat = pt0[0]
-                e_lng = pt0[1]
-                s_lat = pt1[0]
-                w_lng = pt1[1]
-
-                download_extent(w=w_lng, s=s_lat, e=e_lng, n=n_lat, zoom=zoom, source=f"http://{LOCAL_URL_TILE}")
-                print("exception not raised")
-            except ConnectionError as ioe1:
-                app_logger.error(f"ioe1:{ioe1}.")
-                msg0 = "HTTPConnectionPool(host='localhost', port=8000): Max retries exceeded with url: /lambda_handler"
-                msg1 = "Caused by NewConnectionError"
-                msg2 = ": Failed to establish a new connection: [Errno 61] Connection refused'))"
-                assert msg0 in str(ioe1)
-                assert msg1 in str(ioe1)
-                assert msg2 in str(ioe1)
-                raise ioe1
-
-    def test_download_extent_io_error2(self):
-        from requests import HTTPError
-        from tests.local_tiles_http_server import LocalTilesHttpServer
-
-        listen_port = 8000
-        with LocalTilesHttpServer.http_server("localhost", listen_port, directory=TEST_EVENTS_FOLDER):
-            pt0, pt1 = input_bbox
-            zoom = 10
-
-            with self.assertRaises(HTTPError):
-                try:
-                    n_lat = pt0[0]
-                    e_lng = pt0[1]
-                    s_lat = pt1[0]
-                    w_lng = pt1[1]
-
-                    download_extent(w=w_lng, s=s_lat, e=e_lng, n=n_lat, zoom=zoom,
-                                    source=LOCAL_URL_TILE + "_not_found_raster!")
-                except HTTPError as http_e:
-                    app_logger.error(f"ae:{http_e}.")
-                    assert "Tile URL resulted in a 404 error. Double-check your tile url:" in str(http_e)
-                    raise http_e

tests/io/test_wrappers_helpers.py

@@ -1,135 +0,0 @@
-import json
-import time
-import unittest
-
-from http import HTTPStatus
-from unittest.mock import patch
-
-from samgis_lisa_on_cuda.io import wrappers_helpers
-from samgis_lisa_on_cuda.io.wrappers_helpers import get_parsed_bbox_points_with_dictlist_prompt, get_parsed_request_body, get_response
-from samgis_lisa_on_cuda.utilities.type_hints import ApiRequestBody
-from tests import TEST_EVENTS_FOLDER
-
-
-class WrappersHelpersTest(unittest.TestCase):
-    @patch.object(time, "time")
-    def test_get_response(self, time_mocked):
-        time_diff = 108
-        end_run = 1000
-        time_mocked.return_value = end_run
-        start_time = end_run - time_diff
-        aws_request_id = "test_invoke_id"
-
-        with open(TEST_EVENTS_FOLDER / "get_response.json") as tst_json:
-            inputs_outputs = json.load(tst_json)
-
-        response_type = "200"
-        body_response = inputs_outputs[response_type]["input"]
-        output = get_response(HTTPStatus.OK.value, start_time, aws_request_id, body_response)
-        assert json.loads(output) == inputs_outputs[response_type]["output"]
-
-        response_type = "400"
-        response_400 = get_response(HTTPStatus.BAD_REQUEST.value, start_time, aws_request_id, {})
-        assert response_400 == inputs_outputs[response_type]["output"]
-
-        response_type = "422"
-        response_422 = get_response(HTTPStatus.UNPROCESSABLE_ENTITY.value, start_time, aws_request_id, {})
-        assert response_422 == inputs_outputs[response_type]["output"]
-
-        response_type = "500"
-        response_500 = get_response(HTTPStatus.INTERNAL_SERVER_ERROR.value, start_time, aws_request_id, {})
-        assert response_500 == inputs_outputs[response_type]["output"]
-
-    @staticmethod
-    def test_get_parsed_bbox_points():
-        with open(TEST_EVENTS_FOLDER / "get_parsed_bbox_prompts_single_point.json") as tst_json:
-            inputs_outputs = json.load(tst_json)
-            for k, input_output in inputs_outputs.items():
-                print(f"k:{k}.")
-                raw_body = get_parsed_request_body(**input_output["input"])
-                output = get_parsed_bbox_points_with_dictlist_prompt(raw_body)
-                assert output == input_output["output"]
-
-    @staticmethod
-    def test_get_parsed_bbox_other_inputs():
-        for json_filename in ["single_rectangle", "multi_prompt"]:
-            with open(TEST_EVENTS_FOLDER / f"get_parsed_bbox_prompts_{json_filename}.json") as tst_json:
-                inputs_outputs = json.load(tst_json)
-                parsed_input = ApiRequestBody.model_validate(inputs_outputs["input"])
-                output = get_parsed_bbox_points_with_dictlist_prompt(parsed_input)
-                assert output == inputs_outputs["output"]
-
-    @staticmethod
-    def test_get_parsed_request_body():
-        from samgis_core.utilities.utilities import base64_encode
-
-        input_event = {
-            "event": {
-                "bbox": {
-                    "ne": {"lat": 38.03932961278458, "lng": 15.36808069832851},
-                    "sw": {"lat": 37.455509218936974, "lng": 14.632807441554068}
-                },
-                "prompt": [{"type": "point", "data": {"lat": 37.0, "lng": 15.0}, "label": 0}],
-                "zoom": 10, "source_type": "OpenStreetMap.Mapnik", "debug": True
-            }
-        }
-        expected_output_dict = {
-            "bbox": {
-                "ne": {"lat": 38.03932961278458, "lng": 15.36808069832851},
-                "sw": {"lat": 37.455509218936974, "lng": 14.632807441554068}
-            },
-            "prompt": [{"type": "point", "data": {"lat": 37.0, "lng": 15.0}, "label": 0}],
-            "zoom": 10, "source_type": "OpenStreetMap.Mapnik", "debug": True
-        }
-        output = get_parsed_request_body(input_event["event"])
-        assert output == ApiRequestBody.model_validate(input_event["event"])
-
-        input_event_str = json.dumps(input_event["event"])
-        output = get_parsed_request_body(input_event_str)
-        assert output == ApiRequestBody.model_validate(expected_output_dict)
-
-        event = {"body": base64_encode(input_event_str).decode("utf-8")}
-        output = get_parsed_request_body(event)
-        assert output == ApiRequestBody.model_validate(expected_output_dict)
-
-    @patch.object(wrappers_helpers, "providers")
-    def test_get_url_tile(self, providers_mocked):
-        import xyzservices
-        from samgis_lisa_on_cuda.io.wrappers_helpers import get_url_tile
-
-        from tests import LOCAL_URL_TILE
-
-        local_tile_provider = xyzservices.TileProvider(name="local_tile_provider", url=LOCAL_URL_TILE, attribution="")
-        expected_output = {'name': 'local_tile_provider', 'url': LOCAL_URL_TILE, 'attribution': ''}
-        providers_mocked.query_name.return_value = local_tile_provider
-        assert get_url_tile("OpenStreetMap") == expected_output
-
-        local_url = 'http://localhost:8000/{parameter}/{z}/{x}/{y}.png'
-        local_tile_provider = xyzservices.TileProvider(
-            name="local_tile_provider_param", url=local_url, attribution="", parameter="lamda_handler"
-        )
-        providers_mocked.query_name.return_value = local_tile_provider
-        assert get_url_tile("OpenStreetMap.HOT") == {
-            "parameter": "lamda_handler", 'name': 'local_tile_provider_param', 'url': local_url, 'attribution': ''
-        }
-
-    @staticmethod
-    def test_get_url_tile_real():
-        from samgis_lisa_on_cuda.io.wrappers_helpers import get_url_tile
-
-        assert get_url_tile("OpenStreetMap") == {
-            'url': 'https://tile.openstreetmap.org/{z}/{x}/{y}.png', 'max_zoom': 19,
-            'html_attribution': '&copy; <a href="https://www.openstreetmap.org/copyright">OpenStreetMap</a> contributors',
-            'attribution': '(C) OpenStreetMap contributors',
-            'name': 'OpenStreetMap.Mapnik'}
-
-        html_attribution_hot = '&copy; <a href="https://www.openstreetmap.org/copyright">OpenStreetMap</a> contributors, '
-        html_attribution_hot += 'Tiles style by <a href="https://www.hotosm.org/" target="_blank">Humanitarian '
-        html_attribution_hot += 'OpenStreetMap Team</a> hosted by <a href="https://openstreetmap.fr/" target="_blank">'
-        html_attribution_hot += 'OpenStreetMap France</a>'
-        attribution_hot = '(C) OpenStreetMap contributors, Tiles style by Humanitarian OpenStreetMap Team hosted by '
-        attribution_hot += 'OpenStreetMap France'
-        assert get_url_tile("OpenStreetMap.HOT") == {
-            'url': 'https://{s}.tile.openstreetmap.fr/hot/{z}/{x}/{y}.png', 'max_zoom': 19,
-            'html_attribution': html_attribution_hot, 'attribution': attribution_hot, 'name': 'OpenStreetMap.HOT'
-        }

tests/local_tiles_http_server.py

@@ -1,46 +0,0 @@
-import logging
-import time
-import unittest
-
-
-class LocalTilesHttpServer(unittest.TestCase):
-    from contextlib import contextmanager
-
-    @staticmethod
-    @contextmanager
-    def http_server(host: str, port: int, directory: str):
-        """Function http_server defined within this test class to avoid pytest error "fixture 'host' not found"."""
-        from functools import partial
-        from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer
-        from threading import Thread
-
-        server = ThreadingHTTPServer(
-            (host, port), partial(SimpleHTTPRequestHandler, directory=directory)
-        )
-        print("dir:", directory, "#")
-        server_thread = Thread(target=server.serve_forever, name="http_server")
-        server_thread.start()
-        logging.info(f"listen:: host {host}, port {port}.")
-
-        try:
-            yield
-        finally:
-            server.shutdown()
-            server_thread.join()
-
-
-if __name__ == '__main__':
-    # from tests import TEST_ROOT_FOLDER
-    from pathlib import Path
-
-    PROJECT_ROOT_FOLDER = Path(globals().get("__file__", "./_")).absolute().parent.parent
-
-    TEST_ROOT_FOLDER = PROJECT_ROOT_FOLDER / "tests"
-    TEST_EVENTS_FOLDER = TEST_ROOT_FOLDER / "events"
-
-    main_listen_port = 8000
-    logging.info(f"http_basedir_serve: {TEST_ROOT_FOLDER}.")
-    with LocalTilesHttpServer.http_server("localhost", main_listen_port, directory=str(TEST_ROOT_FOLDER)):
-        time.sleep(1000)
-        logging.info("""import time; time.sleep(10)""")
-    # logging.info("Http server stopped.")

tests/prediction_api/test_predictors.py

@@ -1,64 +0,0 @@
-import json
-from unittest.mock import patch
-
-import numpy as np
-
-from samgis_lisa_on_cuda.prediction_api import predictors
-from samgis_lisa_on_cuda.prediction_api.predictors import get_raster_inference, samexporter_predict
-from tests import TEST_EVENTS_FOLDER
-
-
-@patch.object(predictors, "SegmentAnythingONNX")
-def test_get_raster_inference(segment_anything_onnx_mocked):
-    name_fn = "samexporter_predict"
-
-    with open(TEST_EVENTS_FOLDER / f"{name_fn}.json") as tst_json:
-        inputs_outputs = json.load(tst_json)
-        for k, input_output in inputs_outputs.items():
-            model_mocked = segment_anything_onnx_mocked()
-
-            img = np.load(TEST_EVENTS_FOLDER / f"{name_fn}" / k / "img.npy")
-            inference_out = np.load(TEST_EVENTS_FOLDER / f"{name_fn}" / k / "inference_out.npy")
-            mask = np.load(TEST_EVENTS_FOLDER / f"{name_fn}" / k / "mask.npy")
-            prompt = input_output["input"]["prompt"]
-            model_name = input_output["input"]["model_name"]
-
-            model_mocked.embed.return_value = np.array(img)
-            model_mocked.embed.side_effect = None
-            model_mocked.predict_masks.return_value = inference_out
-            model_mocked.predict_masks.side_effect = None
-            print(f"k:{k}.")
-            output_mask, len_inference_out = get_raster_inference(
-                img=img,
-                prompt=prompt,
-                models_instance=model_mocked,
-                model_name=model_name
-            )
-            assert np.array_equal(output_mask, mask)
-            assert len_inference_out == input_output["output"]["n_predictions"]
-
-
-@patch.object(predictors, "get_raster_inference")
-@patch.object(predictors, "SegmentAnythingONNX")
-@patch.object(predictors, "download_extent")
-@patch.object(predictors, "get_vectorized_raster_as_geojson")
-def test_samexporter_predict(
-        get_vectorized_raster_as_geojson_mocked,
-        download_extent_mocked,
-        segment_anything_onnx_mocked,
-        get_raster_inference_mocked
-):
-    """
-    model_instance = SegmentAnythingONNX()
-    img, matrix = download_extent(DEFAULT_TMS, pt0[0], pt0[1], pt1[0], pt1[1], zoom)
-    transform = get_affine_transform_from_gdal(matrix)
-    mask, n_predictions = get_raster_inference(img, prompt, models_instance, model_name)
-    get_vectorized_raster_as_geojson(mask, matrix)
-    """
-    aff = 1, 2, 3, 4, 5, 6
-    segment_anything_onnx_mocked.return_value = "SegmentAnythingONNX_instance"
-    download_extent_mocked.return_value = np.zeros((10, 10)), aff
-    get_raster_inference_mocked.return_value = np.ones((10, 10)), 1
-    get_vectorized_raster_as_geojson_mocked.return_value = {"geojson": "{}", "n_shapes_geojson": 2}
-    output = samexporter_predict(bbox=[[1, 2], [3, 4]], prompt=[{}], zoom=10, model_name_key="fastsam")
-    assert output == {"n_predictions": 1, "geojson": "{}", "n_shapes_geojson": 2}

tests/{test_fastapi_app.py → test_app.py}

@@ -4,19 +4,15 @@ import unittest
 from unittest.mock import patch
 
 from fastapi.testclient import TestClient
-
-from samgis_lisa_on_cuda import PROJECT_ROOT_FOLDER
-from samgis_lisa_on_cuda.io import wrappers_helpers
+from samgis_web.web import web_helpers
+import app
 from tests import TEST_EVENTS_FOLDER
-from tests.local_tiles_http_server import LocalTilesHttpServer
-from wrappers import fastapi_wrapper
-from wrappers.fastapi_wrapper import app
 
 
 infer_samgis = "/infer_samgis"
 response_status_code = "response.status_code:{}."
 response_body_loaded = "response.body_loaded:{}."
-client = TestClient(app)
+client = TestClient(app.app)
 source = {
     'url': 'https://tile.openstreetmap.org/{z}/{x}/{y}.png', 'max_zoom': 19,
     'html_attribution': '&copy; <a href="https://www.openstreetmap.org/copyright">OpenStreetMap</a> contributors',
@@ -81,18 +77,18 @@ class TestFastapiApp(unittest.TestCase):
         body = response.json()
         assert body == {'msg': 'Error - Unprocessable Entity'}
 
-    def test_index(self):
-        import subprocess
-
-        subprocess.run(["pnpm", "build"], cwd=PROJECT_ROOT_FOLDER / "static")
-        subprocess.run(["pnpm", "tailwindcss", "-i", "./src/input.css", "-o", "./dist/output.css"],
-                       cwd=PROJECT_ROOT_FOLDER / "static")
-        response = client.get("/")
-        assert response.status_code == 200
-        html_body = response.read().decode("utf-8")
-        assert "html" in html_body
-        assert "head" in html_body
-        assert "body" in html_body
+    # def test_index(self):
+    #     import subprocess
+    #
+    #     subprocess.run(["pnpm", "build"], cwd=PROJECT_ROOT_FOLDER / "static")
+    #     subprocess.run(["pnpm", "tailwindcss", "-i", "./src/input.css", "-o", "./dist/output.css"],
+    #                    cwd=PROJECT_ROOT_FOLDER / "static")
+    #     response = client.get("/")
+    #     assert response.status_code == 200
+    #     html_body = response.read().decode("utf-8")
+    #     assert "html" in html_body
+    #     assert "head" in html_body
+    #     assert "body" in html_body
 
     def test_404(self):
         response = client.get("/404")
@@ -119,7 +115,7 @@ class TestFastapiApp(unittest.TestCase):
         assert body_loaded == {'success': False}
 
     @patch.object(time, "time")
-    @patch.object(fastapi_wrapper, "samexporter_predict")
+    @patch.object(app, "samexporter_predict")
     def test_infer_samgis_500(self, samexporter_predict_mocked, time_mocked):
         time_mocked.return_value = 0
         samexporter_predict_mocked.side_effect = ValueError("I raise a value error!")
@@ -131,11 +127,12 @@ class TestFastapiApp(unittest.TestCase):
         print(response_body_loaded.format(body))
         assert body == {'msg': 'Error - Internal Server Error'}
 
-    @patch.object(wrappers_helpers, "get_url_tile")
+    @patch.object(web_helpers, "get_url_tile")
     @patch.object(time, "time")
     def test_infer_samgis_real_200(self, time_mocked, get_url_tile_mocked):
         import shapely
         import xyzservices
+        from samgis_web.utilities.local_tiles_http_server import LocalTilesHttpServer
         from tests import LOCAL_URL_TILE, TEST_EVENTS_FOLDER
 
         time_mocked.return_value = 0
@@ -162,7 +159,7 @@ class TestFastapiApp(unittest.TestCase):
         assert len(output_geojson.geoms) == 3
 
     @patch.object(time, "time")
-    @patch.object(fastapi_wrapper, "samexporter_predict")
+    @patch.object(app, "samexporter_predict")
     def test_infer_samgis_mocked_200(self, samexporter_predict_mocked, time_mocked):
         self.maxDiff = None
 

tests/test_lambda_app.py

@@ -1,232 +0,0 @@
-import json
-import time
-import unittest
-from unittest.mock import patch
-
-from samgis_lisa_on_cuda import IS_AWS_LAMBDA
-
-if IS_AWS_LAMBDA:
-    try:
-        from awslambdaric.lambda_context import LambdaContext
-
-        from samgis_lisa_on_cuda.io import wrappers_helpers
-        from wrappers import lambda_wrapper
-        from tests.local_tiles_http_server import LocalTilesHttpServer
-
-
-        class TestLambdaApp(unittest.TestCase):
-            @patch.object(time, "time")
-            @patch.object(lambda_wrapper, "samexporter_predict")
-            @patch.object(lambda_wrapper, "get_parsed_bbox_points")
-            @patch.object(lambda_wrapper, "get_parsed_request_body")
-            def test_lambda_handler_500(
-                    self,
-                    get_parsed_request_body_mocked,
-                    get_parsed_bbox_points_mocked,
-                    samexporter_predict_mocked,
-                    time_mocked
-            ):
-                from wrappers.lambda_wrapper import lambda_handler
-
-                time_mocked.return_value = 0
-                get_parsed_request_body_mocked.value = {}
-                get_parsed_bbox_points_mocked.return_value = {"bbox": "bbox_object", "prompt": "prompt_object",
-                                                              "zoom": 1}
-                samexporter_predict_mocked.side_effect = ValueError("I raise a value error!")
-
-                event = {"body": {}, "version": 1.0}
-                lambda_context = LambdaContext(
-                    invoke_id="test_invoke_id",
-                    client_context=None,
-                    cognito_identity=None,
-                    epoch_deadline_time_in_ms=time.time()
-                )
-                expected_response_500 = '{"statusCode": 500, "header": {"Content-Type": "application/json"}, '
-                expected_response_500 += '"body": "{\\"duration_run\\": 0, \\"message\\": \\"Internal server error\\", '
-                expected_response_500 += '\\"request_id\\": \\"test_invoke_id\\"}", "isBase64Encoded": false}'
-
-                assert lambda_handler(event, lambda_context) == expected_response_500
-
-
-        @patch.object(time, "time")
-        @patch.object(lambda_wrapper, "get_parsed_request_body")
-        def test_lambda_handler_400(self, get_parsed_request_body_mocked, time_mocked):
-            from wrappers.lambda_wrapper import lambda_handler
-
-            time_mocked.return_value = 0
-            get_parsed_request_body_mocked.return_value = {}
-
-            event = {"body": {}, "version": 1.0}
-            lambda_context = LambdaContext(
-                invoke_id="test_invoke_id",
-                client_context=None,
-                cognito_identity=None,
-                epoch_deadline_time_in_ms=time.time()
-            )
-
-            assert lambda_handler(event, lambda_context) == (
-                '{"statusCode": 400, "header": {"Content-Type": "application/json"}, '
-                '"body": "{\\"duration_run\\": 0, \\"message\\": \\"Bad Request\\", '
-                '\\"request_id\\": \\"test_invoke_id\\"}", "isBase64Encoded": false}')
-
-
-        @patch.object(time, "time")
-        def test_lambda_handler_422(self, time_mocked):
-            from wrappers.lambda_wrapper import lambda_handler
-
-            time_mocked.return_value = 0
-            event = {"body": {}, "version": 1.0}
-            lambda_context = LambdaContext(
-                invoke_id="test_invoke_id",
-                client_context=None,
-                cognito_identity=None,
-                epoch_deadline_time_in_ms=time.time()
-            )
-
-            response_422 = lambda_handler(event, lambda_context)
-            expected_response_422 = '{"statusCode": 422, "header": {"Content-Type": "application/json"}, '
-            expected_response_422 += '"body": "{\\"duration_run\\": 0, \\"message\\": \\"Missing required parameter\\", '
-            expected_response_422 += '\\"request_id\\": \\"test_invoke_id\\"}", "isBase64Encoded": false}'
-
-            assert response_422 == expected_response_422
-
-
-        @patch.object(time, "time")
-        @patch.object(lambda_wrapper, "samexporter_predict")
-        @patch.object(lambda_wrapper, "get_response")
-        @patch.object(lambda_wrapper, "get_parsed_bbox_points")
-        @patch.object(lambda_wrapper, "get_parsed_request_body")
-        def test_lambda_handler_200_mocked(
-                self,
-                get_parsed_request_body_mocked,
-                get_parsed_bbox_points_mocked,
-                get_response_mocked,
-                samexporter_predict_mocked,
-                time_mocked
-        ):
-            from wrappers.lambda_wrapper import lambda_handler
-            from tests import TEST_EVENTS_FOLDER
-
-            time_mocked.return_value = 0
-            get_parsed_request_body_mocked.value = {}
-            get_parsed_bbox_points_mocked.return_value = {"bbox": "bbox_object", "prompt": "prompt_object", "zoom": 1}
-
-            response_type = "200"
-            with open(TEST_EVENTS_FOLDER / "get_response.json") as tst_json_get_response:
-                get_response_io = json.load(tst_json_get_response)
-
-            input_200 = {
-                "bbox": {
-                    "ne": {"lat": 38.03932961278458, "lng": 15.36808069832851},
-                    "sw": {"lat": 37.455509218936974, "lng": 14.632807441554068}
-                },
-                "prompt": [{
-                    "type": "point",
-                    "data": {"lat": 37.0, "lng": 15.0},
-                    "label": 0
-                }],
-                "zoom": 10,
-                "source_type": "OpenStreetMap.Mapnik",
-                "debug": True
-            }
-
-            samexporter_predict_output = get_response_io[response_type]["input"]
-            samexporter_predict_mocked.return_value = samexporter_predict_output
-            samexporter_predict_mocked.side_effect = None
-            get_response_mocked.return_value = get_response_io[response_type]["output"]
-
-            event = {"body": input_200, "version": 1.0}
-
-            lambda_context = LambdaContext(
-                invoke_id="test_invoke_id",
-                client_context=None,
-                cognito_identity=None,
-                epoch_deadline_time_in_ms=time.time()
-            )
-
-            response_200 = lambda_handler(event, lambda_context)
-            expected_response_200 = get_response_io[response_type]["output"]
-            print(f"types: response_200:{type(response_200)}, expected:{type(expected_response_200)}.")
-            assert response_200 == expected_response_200
-
-
-        @patch.object(wrappers_helpers, "get_url_tile")
-        def test_lambda_handler_200_real_single_multi_point(self, get_url_tile_mocked):
-            import xyzservices
-            import shapely
-
-            from wrappers.lambda_wrapper import lambda_handler
-            from tests import LOCAL_URL_TILE, TEST_EVENTS_FOLDER
-
-            local_tile_provider = xyzservices.TileProvider(name="local_tile_provider", url=LOCAL_URL_TILE,
-                                                           attribution="")
-            get_url_tile_mocked.return_value = local_tile_provider
-            fn_name = "lambda_handler"
-            invoke_id = "test_invoke_id"
-
-            for json_filename in [
-                "single_point",
-                "multi_prompt",
-                "single_rectangle"
-            ]:
-                with open(TEST_EVENTS_FOLDER / f"{fn_name}_{json_filename}.json") as tst_json:
-                    inputs_outputs = json.load(tst_json)
-                    lambda_context = LambdaContext(
-                        invoke_id=invoke_id,
-                        client_context=None,
-                        cognito_identity=None,
-                        epoch_deadline_time_in_ms=time.time()
-                    )
-                    expected_response_dict = inputs_outputs["output"]
-                    listen_port = 8000
-                    expected_response_body = json.loads(expected_response_dict["body"])
-
-                    with LocalTilesHttpServer.http_server("localhost", listen_port, directory=TEST_EVENTS_FOLDER):
-                        input_event = inputs_outputs["input"]
-                        input_event_body = json.loads(input_event["body"])
-                        input_event["body"] = json.dumps(input_event_body)
-                        response = lambda_handler(event=input_event, context=lambda_context)
-
-                    response_dict = json.loads(response)
-                    assert response_dict["statusCode"] == 200
-                    body_dict = json.loads(response_dict["body"])
-                    assert body_dict["n_predictions"] == 1
-                    assert body_dict["request_id"] == invoke_id
-                    assert body_dict["message"] == "ok"
-                    assert body_dict["n_shapes_geojson"] == expected_response_body["n_shapes_geojson"]
-
-                    output_geojson = shapely.from_geojson(body_dict["geojson"])
-                    print("output_geojson::", type(output_geojson))
-                    assert isinstance(output_geojson, shapely.GeometryCollection)
-                    assert len(output_geojson.geoms) == expected_response_body["n_shapes_geojson"]
-
-
-        def test_debug(self):
-            from wrappers.lambda_wrapper import lambda_handler
-
-            input_event = {
-                'bbox': {
-                    'ne': {'lat': 46.302592089330524, 'lng': 9.49493408203125},
-                    'sw': {'lat': 46.14011755129237, 'lng': 9.143371582031252}},
-                'prompt': [
-                    {'id': 166, 'type': 'point', 'data': {'lat': 46.18244521829928, 'lng': 9.418544769287111},
-                     'label': 1}
-                ],
-                'zoom': 12, 'source_type': 'OpenStreetMap'
-            }
-            lambda_context = LambdaContext(
-                invoke_id="test_invoke_id",
-                client_context=None,
-                cognito_identity=None,
-                epoch_deadline_time_in_ms=time.time()
-            )
-            response = lambda_handler(event=input_event, context=lambda_context)
-            print(response)
-    except ModuleNotFoundError as mnfe:
-        print("missing awslambdaric...")
-        raise mnfe
-
-
-if __name__ == '__main__':
-    if IS_AWS_LAMBDA:
-        unittest.main()

wrappers/fastapi_wrapper.py

@@ -1,273 +0,0 @@
-import json
-import os
-import pathlib
-import uuid
-
-from fastapi.templating import Jinja2Templates
-import uvicorn
-from fastapi import FastAPI, HTTPException, Request, status
-from fastapi.exceptions import RequestValidationError
-from fastapi.responses import FileResponse, HTMLResponse, JSONResponse
-from fastapi.staticfiles import StaticFiles
-from pydantic import ValidationError
-
-from samgis_lisa_on_cuda import PROJECT_ROOT_FOLDER, WORKDIR
-from samgis_lisa_on_cuda.utilities.type_hints import ApiRequestBody, StringPromptApiRequestBody
-from samgis_core.utilities.fastapi_logger import setup_logging
-
-
-app_logger = setup_logging(debug=True)
-app = FastAPI()
-
-
-@app.middleware("http")
-async def request_middleware(request, call_next):
-    request_id = str(uuid.uuid4())
-    with app_logger.contextualize(request_id=request_id):
-        app_logger.info("Request started")
-
-        try:
-            response = await call_next(request)
-
-        except Exception as ex:
-            app_logger.error(f"Request failed: {ex}")
-            response = JSONResponse(content={"success": False}, status_code=500)
-
-        finally:
-            response.headers["X-Request-ID"] = request_id
-            app_logger.info("Request ended")
-
-        return response
-
-
-@app.post("/post_test_dictlist")
-def post_test_dictlist2(request_input: ApiRequestBody) -> JSONResponse:
-    from samgis_lisa_on_cuda.io.wrappers_helpers import get_parsed_bbox_points_with_dictlist_prompt
-
-    request_body = get_parsed_bbox_points_with_dictlist_prompt(request_input)
-    app_logger.info(f"request_body:{request_body}.")
-    return JSONResponse(
-        status_code=200,
-        content=request_body
-    )
-
-
-@app.get("/health")
-async def health() -> JSONResponse:
-    import importlib.metadata
-    from importlib.metadata import PackageNotFoundError
-
-    try:
-        core_version = importlib.metadata.version('samgis_core')
-        lisa_on_cuda_version = importlib.metadata.version('lisa-on-cuda')
-        samgis_lisa_on_cuda_version = importlib.metadata.version('samgis-lisa-on-cuda')
-    except PackageNotFoundError as pe:
-        app_logger.error(f"pe:{pe}.")
-        samgis_lisa_on_cuda_version = "0.0.0"
-
-    msg = "still alive, "
-    msg += f"""version:{samgis_lisa_on_cuda_version}, core version:{core_version},"""
-    msg += f"""lisa-on-cuda version:{lisa_on_cuda_version},"""
-
-    app_logger.info(msg)
-    return JSONResponse(status_code=200, content={"msg": "still alive..."})
-
-
-@app.post("/post_test_string")
-def post_test_string(request_input: StringPromptApiRequestBody) -> JSONResponse:
-    from lisa_on_cuda.utils import app_helpers
-    from samgis_lisa_on_cuda.io.wrappers_helpers import get_parsed_bbox_points_with_string_prompt
-
-    request_body = get_parsed_bbox_points_with_string_prompt(request_input)
-    app_logger.info(f"request_body:{request_body}.")
-    custom_args = app_helpers.parse_args([])
-    request_body["content"] = {**request_body, "precision": str(custom_args.precision)}
-    return JSONResponse(
-        status_code=200,
-        content=request_body
-    )
-
-
-@app.post("/infer_lisa")
-def infer_lisa(request_input: StringPromptApiRequestBody) -> JSONResponse:
-    from samgis_lisa_on_cuda.prediction_api import lisa
-    from samgis_lisa_on_cuda.io.wrappers_helpers import get_parsed_bbox_points_with_string_prompt, get_source_name
-
-    app_logger.info("starting lisa inference request...")
-
-    try:
-        import time
-
-        time_start_run = time.time()
-        body_request = get_parsed_bbox_points_with_string_prompt(request_input)
-        app_logger.info(f"lisa body_request:{body_request}.")
-        app_logger.info(f"lisa module:{lisa}.")
-        try:
-            source_name = get_source_name(request_input.source_type)
-            app_logger.info(f"source_name = {source_name}.")
-            output = lisa.lisa_predict(
-                bbox=body_request["bbox"], prompt=body_request["prompt"], zoom=body_request["zoom"],
-                source=body_request["source"], source_name=source_name
-            )
-            duration_run = time.time() - time_start_run
-            app_logger.info(f"duration_run:{duration_run}.")
-            body = {
-                "duration_run": duration_run,
-                "output": output
-            }
-            return JSONResponse(status_code=200, content={"body": json.dumps(body)})
-        except Exception as inference_exception:
-            import subprocess
-            project_root_folder_content = subprocess.run(
-                f"ls -l {PROJECT_ROOT_FOLDER}/", shell=True, universal_newlines=True, stdout=subprocess.PIPE
-            )
-            app_logger.error(f"project_root folder 'ls -l' command output: {project_root_folder_content.stdout}.")
-            workdir_folder_content = subprocess.run(
-                f"ls -l {WORKDIR}/", shell=True, universal_newlines=True, stdout=subprocess.PIPE
-            )
-            app_logger.error(f"workdir folder 'ls -l' command output: {workdir_folder_content.stdout}.")
-            app_logger.error(f"inference error:{inference_exception}.")
-            raise HTTPException(
-                status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Internal server error on inference")
-    except ValidationError as va1:
-        app_logger.error(f"validation error: {str(va1)}.")
-        raise ValidationError("Unprocessable Entity")
-
-
-@app.post("/infer_samgis")
-def infer_samgis(request_input: ApiRequestBody) -> JSONResponse:
-    from samgis_lisa_on_cuda.prediction_api import predictors
-    from samgis_lisa_on_cuda.io.wrappers_helpers import get_parsed_bbox_points_with_dictlist_prompt, get_source_name
-
-    app_logger.info("starting plain samgis inference request...")
-
-    try:
-        import time
-
-        time_start_run = time.time()
-        body_request = get_parsed_bbox_points_with_dictlist_prompt(request_input)
-        app_logger.info(f"body_request:{body_request}.")
-        try:
-            source_name = get_source_name(request_input.source_type)
-            app_logger.info(f"source_name = {source_name}.")
-            output = predictors.samexporter_predict(
-                bbox=body_request["bbox"], prompt=body_request["prompt"], zoom=body_request["zoom"],
-                source=body_request["source"], source_name=source_name
-            )
-            duration_run = time.time() - time_start_run
-            app_logger.info(f"duration_run:{duration_run}.")
-            body = {
-                "duration_run": duration_run,
-                "output": output
-            }
-            return JSONResponse(status_code=200, content={"body": json.dumps(body)})
-        except Exception as inference_exception:
-            import subprocess
-            project_root_folder_content = subprocess.run(
-                f"ls -l {PROJECT_ROOT_FOLDER}/", shell=True, universal_newlines=True, stdout=subprocess.PIPE
-            )
-            app_logger.error(f"project_root folder 'ls -l' command output: {project_root_folder_content.stdout}.")
-            workdir_folder_content = subprocess.run(
-                f"ls -l {WORKDIR}/", shell=True, universal_newlines=True, stdout=subprocess.PIPE
-            )
-            app_logger.error(f"workdir folder 'ls -l' command output: {workdir_folder_content.stdout}.")
-            app_logger.error(f"inference error:{inference_exception}.")
-            raise HTTPException(
-                status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Internal server error on inference")
-    except ValidationError as va1:
-        app_logger.error(f"validation error: {str(va1)}.")
-        raise ValidationError("Unprocessable Entity")
-
-
-@app.exception_handler(RequestValidationError)
-async def request_validation_exception_handler(request: Request, exc: RequestValidationError) -> JSONResponse:
-    app_logger.error(f"exception errors: {exc.errors()}.")
-    app_logger.error(f"exception body: {exc.body}.")
-    headers = request.headers.items()
-    app_logger.error(f'request header: {dict(headers)}.')
-    params = request.query_params.items()
-    app_logger.error(f'request query params: {dict(params)}.')
-    return JSONResponse(
-        status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
-        content={"msg": "Error - Unprocessable Entity"}
-    )
-
-
-@app.exception_handler(HTTPException)
-async def http_exception_handler(request: Request, exc: HTTPException) -> JSONResponse:
-    app_logger.error(f"exception: {str(exc)}.")
-    headers = request.headers.items()
-    app_logger.error(f'request header: {dict(headers)}.')
-    params = request.query_params.items()
-    app_logger.error(f'request query params: {dict(params)}.')
-    return JSONResponse(
-        status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
-        content={"msg": "Error - Internal Server Error"}
-    )
-
-
-write_tmp_on_disk = os.getenv("WRITE_TMP_ON_DISK", "")
-app_logger.info(f"write_tmp_on_disk:{write_tmp_on_disk}.")
-if bool(write_tmp_on_disk):
-    try:
-        path_write_tmp_on_disk = pathlib.Path(write_tmp_on_disk)
-        try:
-            pathlib.Path.unlink(path_write_tmp_on_disk, missing_ok=True)
-        except (IsADirectoryError, PermissionError, OSError) as err:
-            app_logger.error(f"{err} while removing old write_tmp_on_disk:{write_tmp_on_disk}.")
-            app_logger.error(f"is file?{path_write_tmp_on_disk.is_file()}.")
-            app_logger.error(f"is symlink?{path_write_tmp_on_disk.is_symlink()}.")
-            app_logger.error(f"is folder?{path_write_tmp_on_disk.is_dir()}.")
-        os.makedirs(write_tmp_on_disk, exist_ok=True)
-        app.mount("/vis_output", StaticFiles(directory=write_tmp_on_disk), name="vis_output")
-    except RuntimeError as rerr:
-        app_logger.error(f"{rerr} while loading the folder write_tmp_on_disk:{write_tmp_on_disk}...")
-        raise rerr
-    templates = Jinja2Templates(directory=WORKDIR / "static")
-
-
-    @app.get("/vis_output", response_class=HTMLResponse)
-    def list_files(request: Request):
-
-        files = os.listdir(write_tmp_on_disk)
-        files_paths = sorted([f"{request.url._url}/{f}" for f in files])
-        print(files_paths)
-        return templates.TemplateResponse(
-            "list_files.html", {"request": request, "files": files_paths}
-        )
-
-
-# important: the index() function and the app.mount MUST be at the end
-# samgis.html
-app.mount("/samgis", StaticFiles(directory=WORKDIR / "static" / "dist", html=True), name="samgis")
-
-
-@app.get("/samgis")
-async def samgis() -> FileResponse:
-    return FileResponse(path=WORKDIR / "static" / "dist" / "samgis.html", media_type="text/html")
-
-
-# lisa.html
-app.mount("/lisa", StaticFiles(directory=WORKDIR / "static" / "dist", html=True), name="lisa")
-
-
-@app.get("/lisa")
-async def lisa() -> FileResponse:
-    return FileResponse(path=WORKDIR / "static" / "dist" / "lisa.html", media_type="text/html")
-
-
-# index.html (lisa.html copy)
-app.mount("/", StaticFiles(directory=WORKDIR / "static" / "dist", html=True), name="index")
-
-
-@app.get("/")
-async def index() -> FileResponse:
-    return FileResponse(path=WORKDIR / "static" / "dist" / "index.html", media_type="text/html")
-
-
-if __name__ == '__main__':
-    try:
-        uvicorn.run(host="0.0.0.0", port=7860, app=app)
-    except Exception as e:
-        app_logger.error("e:", e)
-        raise e

wrappers/lambda_wrapper.py

@@ -1,58 +0,0 @@
-"""Lambda entry point"""
-from http import HTTPStatus
-from typing import Dict
-
-from aws_lambda_powertools.utilities.typing import LambdaContext
-from pydantic import ValidationError
-
-from samgis_lisa_on_cuda import app_logger
-from samgis_lisa_on_cuda.io.wrappers_helpers import get_parsed_request_body, get_parsed_bbox_points_with_dictlist_prompt, get_response
-from samgis_lisa_on_cuda.prediction_api.predictors import samexporter_predict
-
-
-def lambda_handler(event: Dict, context: LambdaContext) -> str:
-    """
-    Handle the request for the serverless backend and return the response
-    (success or a type of error based on the exception raised).
-
-    Args:
-        event: request content
-        context: request context
-
-    Returns:
-        json response from get_response() function
-
-    """
-    from time import time
-    app_logger.info(f"start with aws_request_id:{context.aws_request_id}.")
-    start_time = time()
-
-    if "version" in event:
-        app_logger.info(f"event version: {event['version']}.")
-
-    try:
-        app_logger.info("try get_parsed_event...")
-        request_input = get_parsed_request_body(event)
-        app_logger.info("event parsed: ok")
-        body_request = get_parsed_bbox_points_with_dictlist_prompt(request_input)
-        app_logger.info(f"body_request => {type(body_request)}, {body_request}.")
-
-        try:
-            body_response = samexporter_predict(
-                body_request["bbox"], body_request["prompt"], body_request["zoom"], source=body_request["source"]
-            )
-            app_logger.info(f"output body_response length:{len(body_response)}.")
-            app_logger.debug(f"output body_response:{body_response}.")
-            response = get_response(HTTPStatus.OK.value, start_time, context.aws_request_id, body_response)
-        except Exception as ex2:
-            app_logger.exception(f"exception2:{ex2}.", exc_info=True)
-            response = get_response(HTTPStatus.INTERNAL_SERVER_ERROR.value, start_time, context.aws_request_id, {})
-    except ValidationError as va1:
-        app_logger.exception(f"ValidationError:{va1}.", exc_info=True)
-        response = get_response(HTTPStatus.UNPROCESSABLE_ENTITY.value, start_time, context.aws_request_id, {})
-    except Exception as ex1:
-        app_logger.exception(f"exception1:{ex1}.", exc_info=True)
-        response = get_response(HTTPStatus.BAD_REQUEST.value, start_time, context.aws_request_id, {})
-
-    app_logger.debug(f"response_dumped:{response}...")
-    return response