refactor: complete refactor because of 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"

scripts/create_folders_and_variables_if_not_exists.py

@@ -1,51 +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()
-
-
-if __name__ == '__main__':
-    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}.")

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_zero 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_zero import PROJECT_ROOT_FOLDER
-
-if __name__ == '__main__':
-    from samgis_lisa_on_zero.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_zero 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_package/test_coordinates_pixel_conversion.py

@@ -1,27 +0,0 @@
-import json
-
-from samgis_lisa_on_zero.io_package.coordinates_pixel_conversion import get_latlng_to_pixel_coordinates
-from samgis_lisa_on_zero.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_package/test_geo_helpers.py

@@ -1,103 +0,0 @@
-import json
-import unittest
-import numpy as np
-import shapely
-
-from samgis_lisa_on_zero.io_package.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_zero.io_package.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_zero.io_package.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_package/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_zero.io_package 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_zero.io_package 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_zero.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_package/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_zero import app_logger
-from samgis_lisa_on_zero.io_package.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_package/test_wrappers_helpers.py

@@ -1,78 +0,0 @@
-import json
-import unittest
-from unittest.mock import patch
-
-from samgis_lisa_on_zero.io_package import wrappers_helpers
-from samgis_lisa_on_zero.io_package.wrappers_helpers import get_parsed_bbox_points_with_dictlist_prompt
-from samgis_lisa_on_zero.utilities.type_hints import ApiRequestBody
-from tests import TEST_EVENTS_FOLDER
-
-
-class WrappersHelpersTest(unittest.TestCase):
-    @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"]
-
-    def test_get_parsed_bbox_points_with_string_prompt(self):
-        from samgis_lisa_on_zero.io_package.wrappers_helpers import get_parsed_bbox_points_with_string_prompt
-        req = {
-            "bbox":{
-                "ne":{"lat":46.17271333276639,"lng":10.079505443573},"sw":{"lat":46.1677724417049,"lng":10.068830251693727}
-            },
-            "string_prompt":"You are a ...",
-            "zoom":17,
-            "source_type":"Esri.WorldImagery"
-        }
-        print(req)
-        out = get_parsed_bbox_points_with_string_prompt(json.dumps(req))
-        assert isinstance(out, dict)
-        for out_k, req_k in zip(out.keys(), req.keys()):
-            assert isinstance(out_k, str)
-            assert isinstance(req_k, str)
-
-    @patch.object(wrappers_helpers, "providers")
-    def test_get_url_tile(self, providers_mocked):
-        import xyzservices
-        from samgis_lisa_on_zero.io_package.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_zero.io_package.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_zero.prediction_api import predictors
-from samgis_lisa_on_zero.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_zero import PROJECT_ROOT_FOLDER
-from samgis_lisa_on_zero.io_package import wrappers_helpers
-from tests import TEST_EVENTS_FOLDER
-from tests.local_tiles_http_server import LocalTilesHttpServer
+from samgis_web.web import web_helpers
 import app
-from app import app
-
+from tests import TEST_EVENTS_FOLDER
+from samgis_core.utilities import frontend_builder
 
 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_zero import IS_AWS_LAMBDA
-
-if IS_AWS_LAMBDA:
-    try:
-        from awslambdaric.lambda_context import LambdaContext
-
-        from samgis_lisa_on_zero.io_package 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()