[feat] add input parsing and validation with pydantic, refactor code

events/payload_point.json

@@ -9,5 +9,6 @@
     "label": 0
   }],
   "zoom": 6,
-  "source_type": "Satellite"
-}
+  "source_type": "Satellite",
+  "debug": true
+}

src/app.py

@@ -2,6 +2,7 @@ import time
 from http import HTTPStatus
 
 from aws_lambda_powertools.utilities.typing import LambdaContext
+from pydantic import ValidationError
 
 from src import app_logger
 from src.io.lambda_helpers import get_parsed_request_body, get_parsed_bbox_points, get_response
@@ -16,19 +17,22 @@ def lambda_handler(event: dict, context: LambdaContext):
         app_logger.info(f"event version: {event['version']}.")
 
     try:
-        body = get_parsed_request_body(event, context)
+        app_logger.info(f"try get_parsed_event...")
+        request_input = get_parsed_request_body(event)
+        app_logger.info(f"event parsed: ok")
+        body_request = get_parsed_bbox_points(request_input)
 
         try:
-            prompt_latlng = body["prompt"]
-            app_logger.debug(f"prompt_latlng:{prompt_latlng}.")
-            body_request = get_parsed_bbox_points(body)
-            app_logger.info(f"body_request=> {type(body_request)}, {body_request}.")
+            app_logger.info(f"body_request => {type(body_request)}, {body_request}.")
             body_response = samexporter_predict(body_request["bbox"], body_request["prompt"], body_request["zoom"])
             app_logger.info(f"output body_response:{body_response}.")
             response = get_response(HTTPStatus.OK.value, start_time, context.aws_request_id, body_response)
         except Exception as ex2:
             app_logger.error(f"exception2:{ex2}.")
-            response = get_response(HTTPStatus.UNPROCESSABLE_ENTITY.value, start_time, context.aws_request_id, {})
+            response = get_response(HTTPStatus.BAD_REQUEST.value, start_time, context.aws_request_id, {})
+    except ValidationError as va1:
+        app_logger.error(f"ValidationError:{va1}.")
+        response = get_response(HTTPStatus.UNPROCESSABLE_ENTITY.value, start_time, context.aws_request_id, {})
     except Exception as ex1:
         app_logger.error(f"exception1:{ex1}.")
         response = get_response(HTTPStatus.INTERNAL_SERVER_ERROR.value, start_time, context.aws_request_id, {})

src/io/coordinates_pixel_conversion.py

@@ -3,6 +3,7 @@ from typing import TypedDict
 
 from src import app_logger
 from src.utilities.constants import TILE_SIZE
+from src.utilities.type_hints import LatLngDict
 
 
 class PixelCoordinate(TypedDict):
@@ -10,16 +11,16 @@ class PixelCoordinate(TypedDict):
     y: int
 
 
-def get_latlng2pixel_projection(latlng) -> PixelCoordinate:
+def get_latlng2pixel_projection(latlng: LatLngDict) -> PixelCoordinate:
     app_logger.info(f"latlng: {type(latlng)}, value:{latlng}.")
-    app_logger.info(f'latlng lat: {type(latlng["lat"])}, value:{latlng["lat"]}.')
-    app_logger.info(f'latlng lng: {type(latlng["lng"])}, value:{latlng["lng"]}.')
+    app_logger.info(f'latlng lat: {type(latlng.lat)}, value:{latlng.lat}.')
+    app_logger.info(f'latlng lng: {type(latlng.lng)}, value:{latlng.lng}.')
     try:
-        sin_y: float = math.sin(latlng["lat"] * math.pi / 180)
+        sin_y: float = math.sin(latlng.lat * math.pi / 180)
         app_logger.info(f"sin_y, #1:{sin_y}.")
         sin_y = min(max(sin_y, -0.9999), 0.9999)
         app_logger.info(f"sin_y, #2:{sin_y}.")
-        x = TILE_SIZE * (0.5 + latlng["lng"] / 360)
+        x = TILE_SIZE * (0.5 + latlng.lng / 360)
         app_logger.info(f"x:{x}.")
         y = TILE_SIZE * (0.5 - math.log((1 + sin_y) / (1 - sin_y)) / (4 * math.pi))
         app_logger.info(f"y:{y}.")
@@ -30,7 +31,7 @@ def get_latlng2pixel_projection(latlng) -> PixelCoordinate:
         raise e_get_latlng2pixel_projection
 
 
-def get_point_latlng_to_pixel_coordinates(latlng, zoom: int) -> PixelCoordinate:
+def get_point_latlng_to_pixel_coordinates(latlng: LatLngDict, zoom: int | float) -> PixelCoordinate:
     try:
         world_coordinate: PixelCoordinate = get_latlng2pixel_projection(latlng)
         app_logger.info(f"world_coordinate:{world_coordinate}.")
@@ -45,7 +46,13 @@ def get_point_latlng_to_pixel_coordinates(latlng, zoom: int) -> PixelCoordinate:
         raise e_format_latlng_to_pixel_coordinates
 
 
-def get_latlng_to_pixel_coordinates(latlng_origin_ne, latlng_origin_sw, latlng_current_point, zoom, k: str):
+def get_latlng_to_pixel_coordinates(
+        latlng_origin_ne: LatLngDict,
+        latlng_origin_sw: LatLngDict,
+        latlng_current_point: LatLngDict,
+        zoom: int | float,
+        k: str
+):
     app_logger.info(f"latlng_origin - {k}: {type(latlng_origin_ne)}, value:{latlng_origin_ne}.")
     app_logger.info(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)

src/io/lambda_helpers.py

@@ -1,14 +1,16 @@
 import json
+import logging
 import time
 from typing import Dict
 
 from aws_lambda_powertools.event_handler import content_types
+from aws_lambda_powertools.utilities.parser import parse
 
 from src import app_logger
 from src.io.coordinates_pixel_conversion import get_latlng_to_pixel_coordinates
-from src.utilities.constants import CUSTOM_RESPONSE_MESSAGES, DEFAULT_LOG_LEVEL
+from src.utilities.constants import CUSTOM_RESPONSE_MESSAGES
 from src.utilities.type_hints import RawRequestInput
-from src.utilities.utilities import base64_decode, LogArgumentsDecorator
+from src.utilities.utilities import base64_decode
 
 
 def get_response(status: int, start_time: float, request_id: str, response_body: Dict = None) -> str:
@@ -42,59 +44,59 @@ def get_response(status: int, start_time: float, request_id: str, response_body:
 
 def get_parsed_bbox_points(request_input: RawRequestInput) -> Dict:
     app_logger.info(f"try to parsing input request {request_input}...")
-    bbox = request_input["bbox"]
+
+    bbox = request_input.bbox
     app_logger.debug(f"request bbox: {type(bbox)}, value:{bbox}.")
-    ne = bbox["ne"]
-    sw = bbox["sw"]
+    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"])]
-    bbox = [ne_latlng, sw_latlng]
-    zoom = int(request_input["zoom"])
-    for prompt in request_input["prompt"]:
+    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 = []
+    for prompt in request_input.prompt:
         app_logger.debug(f"current prompt: {type(prompt)}, value:{prompt}.")
-        data = prompt["data"]
-        if prompt["type"] == "point":
-            current_point = get_latlng_to_pixel_coordinates(ne, sw, data, zoom, "point")
-            app_logger.debug(f"current prompt: {type(current_point)}, value:{current_point}.")
-            new_prompt_data = [current_point['x'], current_point['y']]
-            app_logger.debug(f"new_prompt_data: {type(new_prompt_data)}, value:{new_prompt_data}.")
-            prompt["data"] = new_prompt_data
-        else:
-            raise ValueError("valid prompt type is only 'point'")
+        current_point = get_latlng_to_pixel_coordinates(ne, sw, prompt.data, new_zoom, prompt.type)
+        app_logger.debug(f"current prompt: {type(current_point)}, value:{current_point}.")
+        new_prompt_data = [current_point['x'], current_point['y']]
+        app_logger.debug(f"new_prompt_data: {type(new_prompt_data)}, value:{new_prompt_data}.")
+        new_prompt = {
+            "type": prompt.type,
+            "data": new_prompt_data
+        }
+        if prompt.label is not None:
+            new_prompt["label"] = prompt.label
+        new_prompt_list.append(new_prompt)
 
     app_logger.debug(f"bbox => {bbox}.")
-    app_logger.debug(f'request_input-prompt updated => {request_input["prompt"]}.')
+    app_logger.debug(f'request_input-prompt updated => {new_prompt_list}.')
 
-    app_logger.info(f"unpacking elaborated {request_input}...")
+    app_logger.info(f"unpacking elaborated request...")
     return {
-        "bbox": bbox,
-        "prompt": request_input["prompt"],
-        "zoom": zoom
+        "bbox": [ne_latlng, sw_latlng],
+        "prompt": new_prompt_list,
+        "zoom": new_zoom
     }
 
 
-def get_parsed_request_body(event, context):
+def get_parsed_request_body(event):
     app_logger.info(f"event:{json.dumps(event)}...")
-    app_logger.info(f"context:{context}...")
     try:
-        body = event["body"]
+        raw_body = event["body"]
     except Exception as e_constants1:
         app_logger.error(f"e_constants1:{e_constants1}.")
-        body = event
-    app_logger.debug(f"body, #1: {type(body)}, {body}...")
-    if isinstance(body, str):
-        body_decoded_str = base64_decode(body)
+        raw_body = event
+    app_logger.debug(f"raw_body, #1: {type(raw_body)}, {raw_body}...")
+    if isinstance(raw_body, str):
+        body_decoded_str = base64_decode(raw_body)
         app_logger.debug(f"body_decoded_str: {type(body_decoded_str)}, {body_decoded_str}...")
-        body = json.loads(body_decoded_str)
-    app_logger.info(f"body, #2: {type(body)}, {body}...")
-    try:
-        log_level = 'DEBUG' if body['debug'] else DEFAULT_LOG_LEVEL
-        app_logger.warning(f"set logger level to DEBUG")
-        app_logger.setLevel(log_level)
-    except KeyError:
-        app_logger.warning(f"can't set log level, reset it...")
-        app_logger.setLevel(DEFAULT_LOG_LEVEL)
-    app_logger.warning(f"logger level is {app_logger.log_level}.")
-    return body
+        raw_body = json.loads(body_decoded_str)
+    app_logger.info(f"body, #2: {type(raw_body)}, {raw_body}...")
+
+    parsed_body = parse(event=raw_body, model=RawRequestInput)
+    log_level = "DEBUG" if parsed_body.debug else "INFO"
+    app_logger.setLevel(log_level)
+    app_logger.warning(f"set log level to {logging.getLevelName(app_logger.log_level)}.")
+
+    return parsed_body

src/io/tms2geotiff.py

@@ -128,7 +128,7 @@ def get_tile(url):
 
 def print_progress(progress, total, done=False):
     if done:
-        print('Downloaded image %d/%d, %.2f%%' % (progress, total, progress * 100 / total))
+        app_logger.info('Downloaded image %d/%d, %.2f%%' % (progress, total, progress * 100 / total))
 
 
 class ProgressBar:
@@ -160,7 +160,7 @@ class ProgressBar:
         if self.tqdm_bar:
             self.tqdm_bar.close()
         else:
-            print('\nDone.')
+            app_logger.info('\nDone.')
 
 
 def mbtiles_save(db, img_data, xy, zoom, img_format):
@@ -643,8 +643,8 @@ def downloader(input_args, input_parser):
             coords1 = tuple(map(float, getattr(input_args, 'to').split(',')))
             print("coords1:", coords1, "#")
             download_args.extend((coords0[0], coords0[1], coords1[0], coords1[1]))
-    except Exception as e:
-        print(f"e:", e, "#")
+    except Exception as e_downloader:
+        print(f"e_downloader:", e_downloader, "#")
         input_parser.print_help()
         return 1
     download_args.append(input_args.zoom)

src/prediction_api/predictors.py

@@ -29,7 +29,7 @@ def samexporter_predict(bbox, prompt: list[dict], zoom: float, model_name: str =
         with tempfile.TemporaryDirectory() as input_tmp_dir:
             app_logger.info(f'tile_source: {DEFAULT_TMS}!')
             pt0, pt1 = bbox
-            app_logger.info("downloading...")
+            app_logger.info(f"downloading geo-referenced raster with bbox {bbox}, zoom {zoom}.")
             img, matrix = download_extent(DEFAULT_TMS, pt0[0], pt0[1], pt1[0], pt1[1], zoom)
             app_logger.debug(f"img type {type(img)} with shape/size:{img.size}, matrix:{matrix}.")
 

src/utilities/type_hints.py

@@ -1,4 +1,6 @@
 """custom type hints"""
+from enum import Enum
+
 from pydantic import BaseModel
 from typing import TypedDict
 
@@ -15,26 +17,13 @@ class LatLngDict(BaseModel):
     lng: float
 
 
-class RawBBox(BaseModel):
-    ne: LatLngDict
-    sw: LatLngDict
-
-
-class RawPrompt(BaseModel):
-    type: str
-    data: LatLngDict
-    label: int = 0
-
-
-class RawRequestInput(BaseModel):
-    bbox: RawBBox
-    prompt: RawPrompt
-    zoom: int | float
-    source_type: str = "Satellite"
+class PromptType(str, Enum):
+    point = "point"
+    # rectangle = "rectangle"
 
 
 class ParsedPrompt(BaseModel):
-    type: str
+    type: PromptType
     data: llist_float
     label: int = 0
 
@@ -48,3 +37,22 @@ class ParsedRequestInput(BaseModel):
 class PixelCoordinate(TypedDict):
     x: int
     y: int
+
+
+class RawBBox(BaseModel):
+    ne: LatLngDict
+    sw: LatLngDict
+
+
+class RawPrompt(BaseModel):
+    type: PromptType
+    data: LatLngDict
+    label: int = 0
+
+
+class RawRequestInput(BaseModel):
+    bbox: RawBBox
+    prompt: list[RawPrompt]
+    zoom: int | float
+    source_type: str = "Satellite"
+    debug: bool = False

tests/events/get_parsed_bbox_points.json

@@ -1,7 +1,7 @@
 {
   "europe": {
     "input": {
-      "request_input": {
+      "event": {
         "bbox": {
           "ne": {
             "lat": 38.03932961278458,
@@ -53,7 +53,7 @@
   },
   "north_america": {
     "input": {
-      "request_input": {
+      "event": {
         "bbox": {
           "ne": {
             "lat": 44.918201144476456,
@@ -104,7 +104,7 @@
   },
   "oceania": {
     "input": {
-      "request_input": {
+      "event": {
         "bbox": {
           "ne": {
             "lat": -1.4939713066293112,
@@ -155,7 +155,7 @@
   },
   "south_america": {
     "input": {
-      "request_input": {
+      "event": {
         "bbox": {
           "ne": {
             "lat": -24.80663308621806,

tests/io/test_lambda_helpers.py

@@ -1,5 +1,14 @@
-from tests.io.test_utilities import fn_reading_json_inputs_outputs__test
+import json
+
+from src.io.lambda_helpers import get_parsed_bbox_points, get_parsed_request_body
+from tests import TEST_EVENTS_FOLDER
 
 
 def test_get_parsed_bbox_points():
-    fn_reading_json_inputs_outputs__test(name_fn="get_parsed_bbox_points")
+    with open(TEST_EVENTS_FOLDER / "get_parsed_bbox_points.json") as tst_json:
+        inputs_outputs = json.load(tst_json)
+        for k, input_output in inputs_outputs.items():
+            print(f"k:{k}.")
+            raw_body = get_parsed_request_body(**input_output["input"])
+            output = get_parsed_bbox_points(raw_body)
+            assert output == input_output["output"]