refactor: remove unused files, update samgis-lisa to version 1.0.1 also in pyproject.toml

pyproject.toml

@@ -13,7 +13,7 @@ version = "1.5.0"
 
 [tool.poetry.dependencies]
 python = ">=3.10,<3.11"
-samgis-lisa = "1.0.0"
+samgis-lisa = "1.0.1"
 
 [tool.poetry.group.test]
 optional = true

samgis_lisa_on_zero/__init__.py

@@ -1,21 +0,0 @@
-"""Get machine learning predictions from geodata raster images"""
-import logging
-import os
-
-# not used here but contextily_tile is imported in samgis_lisa_on_zero.io_package.tms2geotiff
-from contextily import tile as contextily_tile
-from pathlib import Path
-
-from lisa_on_cuda.utils import session_logger
-from samgis_lisa_on_zero.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))
-
-loglevel = os.getenv('LOGLEVEL', 'INFO').upper()
-session_logger.change_logging(loglevel)
-app_logger = logging.getLogger(__name__)

samgis_lisa_on_zero/__version__.py

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

samgis_lisa_on_zero/io_package/__init__.py

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

samgis_lisa_on_zero/io_package/coordinates_pixel_conversion.py

@@ -1,102 +0,0 @@
-"""functions useful to convert to/from latitude-longitude coordinates to pixel image coordinates"""
-from lisa_on_cuda.utils import session_logger
-from samgis_core.utilities.type_hints import TupleFloat, TupleFloatAny
-from samgis_lisa_on_zero import app_logger
-from samgis_lisa_on_zero.utilities.constants import TILE_SIZE, EARTH_EQUATORIAL_RADIUS
-from samgis_lisa_on_zero.utilities.type_hints import ImagePixelCoordinates
-from samgis_lisa_on_zero.utilities.type_hints import LatLngDict
-
-
-@session_logger.set_uuid_logging
-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
-
-
-@session_logger.set_uuid_logging
-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
-
-
-@session_logger.set_uuid_logging
-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_zero/io_package/geo_helpers.py

@@ -1,95 +0,0 @@
-"""handle geo-referenced raster images"""
-
-from affine import Affine
-from lisa_on_cuda.utils import session_logger
-from numpy import ndarray as np_ndarray
-
-from samgis_core.utilities.type_hints import ListFloat, TupleFloat, DictStrInt
-from samgis_lisa_on_zero import app_logger
-
-
-@session_logger.set_uuid_logging
-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)
-
-
-@session_logger.set_uuid_logging
-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_zero/io_package/raster_helpers.py

@@ -1,343 +0,0 @@
-"""helpers for computer vision duties"""
-import logging
-
-import numpy as np
-from numpy import ndarray, bitwise_not
-from rasterio import open as rasterio_open
-
-from lisa_on_cuda.utils import session_logger
-from samgis_lisa_on_zero.utilities.constants import OUTPUT_CRS_STRING
-from samgis_lisa_on_zero.utilities.type_hints import XYZTerrainProvidersNames
-
-from samgis_lisa_on_zero import app_logger
-
-
-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
-
-
-@session_logger.set_uuid_logging
-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_zero.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
-
-
-@session_logger.set_uuid_logging
-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
-
-
-@session_logger.set_uuid_logging
-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
-
-
-@session_logger.set_uuid_logging
-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
-
-
-@session_logger.set_uuid_logging
-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)
-
-
-@session_logger.set_uuid_logging
-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)
-
-
-@session_logger.set_uuid_logging
-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
-
-
-@session_logger.set_uuid_logging
-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.")
-
-
-@session_logger.set_uuid_logging
-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_zero/io_package/tms2geotiff.py

@@ -1,187 +0,0 @@
-import os
-
-from lisa_on_cuda.utils import session_logger
-from numpy import ndarray
-from samgis_core.utilities.type_hints import TupleFloat
-from xyzservices import TileProvider
-
-from samgis_lisa_on_zero import app_logger
-from samgis_lisa_on_zero.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_zero.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)
-
-
-@session_logger.set_uuid_logging
-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_zero import contextily_tile
-        from samgis_lisa_on_zero.io_package.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
-
-
-@session_logger.set_uuid_logging
-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
-
-
-@session_logger.set_uuid_logging
-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
-
-
-@session_logger.set_uuid_logging
-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_zero/io_package/wrappers_helpers.py

@@ -1,195 +0,0 @@
-"""lambda helper functions"""
-from typing import Dict
-
-from lisa_on_cuda.utils import session_logger
-from lisa_on_cuda.utils.app_helpers import get_cleaned_input
-from xyzservices import providers, TileProvider
-
-from samgis_lisa_on_zero import app_logger
-from samgis_lisa_on_zero.io_package.coordinates_pixel_conversion import get_latlng_to_pixel_coordinates
-from samgis_lisa_on_zero.utilities.constants import COMPLETE_URL_TILES_MAPBOX, COMPLETE_URL_TILES_NEXTZEN
-from samgis_lisa_on_zero.utilities.type_hints import (
-    ApiRequestBody, XYZTerrainProvidersNames, XYZDefaultProvidersNames, StringPromptApiRequestBody)
-
-
-@session_logger.set_uuid_logging
-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: {type(request_input)}, {request_input}...")
-    if isinstance(request_input, str):
-        app_logger.info(f"string/json input, parsing it to {type(StringPromptApiRequestBody)}...")
-        request_input = StringPromptApiRequestBody.model_validate_json(request_input)
-        app_logger.info(f"parsed input, now of type {type(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),
-        "source_name": get_source_name(request_input.source_type)
-    }
-
-
-@session_logger.set_uuid_logging
-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),
-        "source_name": get_source_name(request_input.source_type)
-    }
-
-
-@session_logger.set_uuid_logging
-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
-
-
-@session_logger.set_uuid_logging
-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']]
-
-
-@session_logger.set_uuid_logging
-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"]
-    ]
-
-
-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=""
-)
-
-
-@session_logger.set_uuid_logging
-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)
-
-
-@session_logger.set_uuid_logging
-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_zero/prediction_api/__init__.py

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

samgis_lisa_on_zero/prediction_api/global_models.py

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

samgis_lisa_on_zero/prediction_api/lisa.py

@@ -1,107 +0,0 @@
-from datetime import datetime
-
-from lisa_on_cuda.utils import session_logger
-from samgis_core.utilities.type_hints import LlistFloat, DictStrInt
-from spaces import GPU as SPACES_GPU
-
-from samgis_lisa_on_zero import app_logger
-from samgis_lisa_on_zero.io_package.geo_helpers import get_vectorized_raster_as_geojson
-from samgis_lisa_on_zero.io_package.raster_helpers import write_raster_png, write_raster_tiff
-from samgis_lisa_on_zero.io_package.tms2geotiff import download_extent
-from samgis_lisa_on_zero.utilities.constants import DEFAULT_URL_TILES, LISA_INFERENCE_FN
-
-msg_write_tmp_on_disk = "found option to write images and geojson output..."
-
-
-@session_logger.set_uuid_logging
-def load_model_and_inference_fn(inference_function_name_key: str):
-    from lisa_on_cuda.utils import app_helpers
-    from samgis_lisa_on_zero.prediction_api.global_models import models_dict
-
-    if models_dict[inference_function_name_key]["inference"] is None:
-        msg = f"missing inference function {inference_function_name_key}, "
-        msg += f"instantiating it now using inference_decorator {SPACES_GPU}!"
-        app_logger.info(msg)
-        parsed_args = app_helpers.parse_args([])
-        inference_fn = app_helpers.get_inference_model_by_args(
-            parsed_args,
-            internal_logger0=app_logger,
-            inference_decorator=SPACES_GPU
-        )
-        models_dict[inference_function_name_key]["inference"] = inference_fn
-
-
-@session_logger.set_uuid_logging
-def lisa_predict(
-        bbox: LlistFloat,
-        prompt: str,
-        zoom: float,
-        inference_function_name_key: str = LISA_INFERENCE_FN,
-        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
-    from samgis_lisa_on_zero.prediction_api.global_models import models_dict
-
-    if source_name is None:
-        source_name = str(source)
-
-    app_logger.info("start lisa inference...")
-    app_logger.debug(f"type(source):{type(source)}, source:{source},")
-    app_logger.debug(f"type(source_name):{type(source_name)}, source_name:{source_name}.")
-
-    load_model_and_inference_fn(inference_function_name_key)
-    app_logger.debug(f"using a '{inference_function_name_key}' instance model...")
-    inference_fn = models_dict[inference_function_name_key]["inference"]
-    app_logger.info(f"loaded inference function '{inference_fn.__name__}'.")
-
-    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"lisa_zero, source_name:{source_name}, source_name type:{type(source_name)}.")
-    app_logger.info(f"lisa_zero, prompt type:{type(prompt)}.")
-    app_logger.info(f"lisa_zero, prompt:{prompt}.")
-    prompt_str = str(prompt)
-    app_logger.info(f"lisa_zero, img type:{type(img)}.")
-    embedding_key = f"{source_name}_z{zoom}_{prefix}"
-    _, mask, output_string = inference_fn(input_str=prompt_str, input_image=img, embedding_key=embedding_key)
-    app_logger.info(f"lisa_zero, output_string type:{type(output_string)}.")
-    app_logger.info(f"lisa_zero, mask_output type:{type(mask)}.")
-    app_logger.info(f"created output_string '{output_string}', preparing conversion to geojson...")
-    return {
-        "output_string": output_string,
-        **get_vectorized_raster_as_geojson(mask, transform)
-    }

samgis_lisa_on_zero/prediction_api/predictors.py

@@ -1,77 +0,0 @@
-"""functions using machine learning instance model(s)"""
-
-from lisa_on_cuda.utils import session_logger
-from samgis_core.prediction_api import sam_onnx2, sam_onnx_inference
-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
-
-from samgis_lisa_on_zero import MODEL_FOLDER
-from samgis_lisa_on_zero import app_logger
-from samgis_lisa_on_zero.io_package.geo_helpers import get_vectorized_raster_as_geojson
-from samgis_lisa_on_zero.io_package.raster_helpers import get_raster_terrain_rgb_like, get_rgb_prediction_image
-from samgis_lisa_on_zero.io_package.tms2geotiff import download_extent
-from samgis_lisa_on_zero.io_package.wrappers_helpers import check_source_type_is_terrain
-from samgis_lisa_on_zero.prediction_api.global_models import models_dict, embedding_dict
-from samgis_lisa_on_zero.utilities.constants import DEFAULT_URL_TILES, SLOPE_CELLSIZE
-
-
-@session_logger.set_uuid_logging
-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 = sam_onnx2.SegmentAnythingONNX2(
-            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 = sam_onnx_inference.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_zero/utilities/__init__.py

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

samgis_lisa_on_zero/utilities/constants.py

@@ -1,64 +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
-LISA_INFERENCE_FN = "lisa"
-GRADIO_EXAMPLE_BODY = {
-    "bbox":{
-        "ne":{"lat":46.17271333276639,"lng":10.079505443573},
-        "sw":{"lat":46.1677724417049,"lng":10.068830251693727}
-    },
-    "string_prompt":"",
-    "zoom":17,
-    "source_type":"Esri.WorldImagery"
-}
-GRADIO_EXAMPLES_TEXT_LIST = [
-    """You need to identify the areas with trees in this photogrammetric image. Please output segmentation mask.""",
-    """You need to identify the areas with streets in this photogrammetric image. Please output segmentation mask.""",
-    """You need to identify the houses in this photogrammetric image. Please output segmentation mask and explain why.""",
-    """Describe what do you see in this image.""",
-]
-GRADIO_MARKDOWN = """# [LISA](https://github.com/dvlab-research/LISA) + [SamGIS](https://github.com/trincadev/samgis-be) on Zero GPU!
-
-This project aims to permit use of [LISA](https://github.com/dvlab-research/LISA) (Reasoning Segmentation via Large Language Model) applied to geospatial data thanks to [SamGIS](https://github.com/trincadev/samgis-be). In this space I adapted LISA to HuggingFace [lisa-on-cuda](https://huggingface.co/spaces/aletrn/lisa-on-cuda) ZeroGPU space.
-
-This [home page project](https://huggingface.co/spaces/aletrn/samgis-lisa-on-zero) is a plane Gradio interface that take a json in input to translate it to a geojson. More information about these API implementation [here](
-https://aletrn-samgis-lisa-on-zero.hf.space/docs). On this [blog page](https://trinca.tornidor.com/projects/lisa-adapted-for-samgis) you can find more details, including some request and response examples with the geojson map representations.
-
-You can also find the alternative map interface [here](https://aletrn-samgis-lisa-on-zero.hf.space/lisa/) useful to create on the fly the payload requests and to represent the geojson response.
-"""

samgis_lisa_on_zero/utilities/session_logger.py

@@ -1,63 +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(log_level: loggingType, formatter: str = default_formatter, name: str = "logger"
-) -> Tuple[logging, contextvars.ContextVar]:
-    """
-    Create a logging instance with log string formatter.
-
-    Args:
-        log_level: logging level
-        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="DEBUG", format=default_formatter, force=True)
-
-    logger = logging.getLogger(name=name)
-
-    # create a console handler
-    ch = logging.StreamHandler()
-    ch.setLevel("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_zero/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