[fix] handle source types with additional parameters other than x,y,z; set default source_type = OpenStreetMap.Mapnik

src/app.py

@@ -39,7 +39,7 @@ def lambda_handler(event: Dict, context: LambdaContext) -> str:
 
         try:
             body_response = samexporter_predict(
-                body_request["bbox"], body_request["prompt"], body_request["zoom"], url_tile=body_request["url_tile"]
+                body_request["bbox"], body_request["prompt"], body_request["zoom"], source=body_request["source"]
             )
             app_logger.info(f"output body_response length:{len(body_response)}.")
             app_logger.debug(f"output body_response:{body_response}.")

src/io/lambda_helpers.py

@@ -1,7 +1,7 @@
 """lambda helper functions"""
 from typing import Dict
-from aws_lambda_powertools.event_handler import content_types
 from xyzservices import providers
+from aws_lambda_powertools.event_handler import content_types
 
 from src import app_logger
 from src.io.coordinates_pixel_conversion import get_latlng_to_pixel_coordinates
@@ -52,6 +52,7 @@ def get_parsed_bbox_points(request_input: ApiRequestBody) -> Dict:
     Returns:
         dict with bounding box, prompt and zoom
     """
+
     app_logger.info(f"try to parsing input request {request_input}...")
 
     bbox = request_input.bbox
@@ -73,7 +74,7 @@ def get_parsed_bbox_points(request_input: ApiRequestBody) -> Dict:
         "bbox": [ne_latlng, sw_latlng],
         "prompt": new_prompt_list,
         "zoom": new_zoom,
-        "url_tile": get_url_tile(request_input.source_type)
+        "source": get_url_tile(request_input.source_type)
     }
 
 
@@ -153,9 +154,8 @@ def get_parsed_request_body(event: Dict or str) -> ApiRequestBody:
 
 
 def get_url_tile(source_type: str):
-    from src.utilities.constants import DEFAULT_TMS_NAME, DEFAULT_TMS
+    from src.utilities.constants import DEFAULT_TMS_NAME, DEFAULT_TMS_NAME_SHORT
 
-    if source_type.lower() == DEFAULT_TMS_NAME:
-        return DEFAULT_TMS
-    providers_type = providers.query_name(source_type)
-    return providers_type.url
+    if source_type.lower() == DEFAULT_TMS_NAME_SHORT:
+        return providers.query_name(DEFAULT_TMS_NAME)
+    return providers.query_name(source_type)

src/io/tms2geotiff.py

@@ -1,12 +1,12 @@
 import os
 from numpy import ndarray
+from xyzservices import TileProvider
 
 from src import app_logger
 from src.utilities.constants import (OUTPUT_CRS_STRING, DRIVER_RASTERIO_GTIFF, N_MAX_RETRIES, N_CONNECTION, N_WAIT,
                                      ZOOM_AUTO, BOOL_USE_CACHE)
 from src.utilities.type_hints import tuple_ndarray_transform, tuple_float
 
-
 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))
@@ -14,7 +14,8 @@ n_wait = int(os.getenv("N_WAIT", N_WAIT))
 zoom_auto_string = os.getenv("ZOOM_AUTO", ZOOM_AUTO)
 
 
-def download_extent(w: float, s: float, e: float, n: float, zoom: int or str = zoom_auto_string, source: str = None,
+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:
     """
@@ -26,15 +27,11 @@ def download_extent(w: float, s: float, e: float, n: float, zoom: int or str = z
         e: East edge
         n: North edge
         zoom: Level of detail
-        source: xyzservices.TileProvider object or str
-            [Optional. Default: OpenStreetMap Humanitarian web tiles]
-            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.
+        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
@@ -59,8 +56,8 @@ def download_extent(w: float, s: float, e: float, n: float, zoom: int or str = z
         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)
+            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)

src/prediction_api/predictors.py

@@ -16,7 +16,7 @@ def samexporter_predict(
         prompt: list_dict,
         zoom: float,
         model_name: str = "fastsam",
-        url_tile: str = DEFAULT_TMS
+        source: str = DEFAULT_TMS
 ) -> dict_str_int:
     """
     Return predictions as a geojson from a geo-referenced image using the given input prompt.
@@ -31,7 +31,7 @@ def samexporter_predict(
         prompt: machine learning input prompt
         zoom: Level of detail
         model_name: machine learning model name
-        url_tile: server url tile
+        source: xyz
 
     Returns:
         Affine transform
@@ -47,8 +47,8 @@ def samexporter_predict(
     models_instance = models_dict[model_name]["instance"]
 
     pt0, pt1 = bbox
-    app_logger.info(f"tile_source: {url_tile}: 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=url_tile)
+    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}.")
 

src/utilities/constants.py

@@ -13,7 +13,8 @@ MODEL_ENCODER_NAME = "mobile_sam.encoder.onnx"
 MODEL_DECODER_NAME = "sam_vit_h_4b8939.decoder.onnx"
 TILE_SIZE = 256
 EARTH_EQUATORIAL_RADIUS = 6378137.0
-DEFAULT_TMS_NAME = "openstreetmap"
+DEFAULT_TMS_NAME_SHORT = "openstreetmap"
+DEFAULT_TMS_NAME = "OpenStreetMap.Mapnik"
 DEFAULT_TMS = 'https://tile.openstreetmap.org/{z}/{x}/{y}.png'
 WKT_3857 = 'PROJCS["WGS 84 / Pseudo-Mercator",GEOGCS["WGS 84",DATUM["WGS_1984",SPHEROID["WGS 84",6378137,298.257223563,'
 WKT_3857 += 'AUTHORITY["EPSG","7030"]],AUTHORITY["EPSG","6326"]],PRIMEM["Greenwich",0,AUTHORITY["EPSG","8901"]],'

src/utilities/type_hints.py

@@ -81,9 +81,8 @@ class ApiRequestBody(BaseModel):
     bbox: RawBBox
     prompt: list[RawPromptPoint | RawPromptRectangle]
     zoom: int | float
-    source_type: str = "Satellite"
+    source_type: str = "OpenStreetMap.Mapnik"
     debug: bool = False
-    url_tile: str = DEFAULT_TMS
 
 
 class ApiResponseBodyFailure(BaseModel):