[doc] update some docstrings

src/io/coordinates_pixel_conversion.py

@@ -3,8 +3,8 @@ import math
 
 from src import app_logger
 from src.utilities.constants import TILE_SIZE
-from src.utilities.type_hints import LatLngDict
 from src.utilities.type_hints import ImagePixelCoordinates
+from src.utilities.type_hints import LatLngDict
 
 
 def _get_latlng2pixel_projection(latlng: LatLngDict) -> ImagePixelCoordinates:
@@ -50,7 +50,7 @@ def get_latlng_to_pixel_coordinates(
         k: str
 ) -> ImagePixelCoordinates:
     """
-    Parse the input request lambda event.
+    Parse the input request lambda event
 
     Args:
         latlng_origin_ne: NE latitude-longitude origin point

src/io/geo_helpers.py

@@ -2,28 +2,29 @@
 from pathlib import Path
 from typing import List, Tuple, Dict
 
-from affine import Affine
 import numpy as np
+from affine import Affine
 
 from src import app_logger, PROJECT_ROOT_FOLDER
 
 
-def load_affine_transformation_from_matrix(matrix_source_coeffs: List[float]) -> Affine:
-    """wrapper for rasterio Affine from_gdal method
+def load_affine_transformation_from_matrix(matrix_source_coefficients: List[float]) -> Affine:
+    """
+    Wrapper for rasterio.Affine.from_gdal() method
 
     Args:
-        matrix_source_coeffs: 6 floats ordered by GDAL.
+        matrix_source_coefficients: 6 floats ordered by GDAL.
 
     Returns:
-        Affine: Affine transform
+        Affine transform
     """
 
-    if len(matrix_source_coeffs) != 6:
-        raise ValueError(f"Expected 6 coefficients, found {len(matrix_source_coeffs)}; "
-                         f"argument type: {type(matrix_source_coeffs)}.")
+    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_coeffs)
+        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:
@@ -31,28 +32,28 @@ def load_affine_transformation_from_matrix(matrix_source_coeffs: List[float]) ->
         raise e
 
 
-def get_affine_transform_from_gdal(matrix_source_coeffs: List[float] or Tuple[float]) -> Affine:
+def get_affine_transform_from_gdal(matrix_source_coefficients: List[float] or Tuple[float]) -> Affine:
     """wrapper for rasterio Affine from_gdal method
 
     Args:
-        matrix_source_coeffs: 6 floats ordered by GDAL.
+        matrix_source_coefficients: 6 floats ordered by GDAL.
 
     Returns:
-        Affine: Affine transform
+        Affine transform
     """
-    return Affine.from_gdal(*matrix_source_coeffs)
+    return Affine.from_gdal(*matrix_source_coefficients)
 
 
 def get_vectorized_raster_as_geojson(mask: np.ndarray, matrix: Tuple[float]) -> Dict[str, int]:
     """
-        Parse the input request lambda event.
+        Get shapes and values of connected regions in a dataset or array
 
         Args:
             mask: numpy mask
             matrix: tuple of float to transform into an Affine transform
 
         Returns:
-            Dict: dict containing the output geojson and the predictions number
+            dict containing the output geojson and the predictions number
     """
     try:
         from rasterio.features import shapes
@@ -65,7 +66,7 @@ def get_vectorized_raster_as_geojson(mask: np.ndarray, matrix: Tuple[float]) ->
         shapes_generator = ({
             'properties': {'raster_val': v}, 'geometry': s}
             for i, (s, v)
-            # in enumerate(shapes(mask, mask=(band != 0), transform=rio_src.transform))
+            # 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))
         )

src/io/lambda_helpers.py

@@ -2,7 +2,7 @@
 import json
 import logging
 import time
-from typing import Dict, List
+from typing import Dict
 from aws_lambda_powertools.event_handler import content_types
 
 from src import app_logger
@@ -14,7 +14,7 @@ from src.utilities.utilities import base64_decode
 
 def get_response(status: int, start_time: float, request_id: str, response_body: Dict = None) -> str:
     """
-    Return a response for frontend clients.
+    Response composer
 
     Args:
         status: status response
@@ -23,7 +23,7 @@ def get_response(status: int, start_time: float, request_id: str, response_body:
         response_body: dict we embed into our response
 
     Returns:
-        str: json response
+        json response
 
     """
     app_logger.debug(f"response_body:{response_body}.")
@@ -43,13 +43,13 @@ def get_response(status: int, start_time: float, request_id: str, response_body:
 
 def get_parsed_bbox_points(request_input: RawRequestInput) -> Dict:
     """
-    Format the bbox and prompt request input
+    Parse the raw input request into bbox, prompt and zoom
 
     Args:
         request_input: input dict
 
     Returns:
-        Dict:
+        dict with bounding box, prompt and zoom
     """
     app_logger.info(f"try to parsing input request {request_input}...")
 
@@ -88,15 +88,15 @@ def get_parsed_bbox_points(request_input: RawRequestInput) -> Dict:
     }
 
 
-def get_parsed_request_body(event: Dict):
+def get_parsed_request_body(event: Dict) -> RawRequestInput:
     """
-    Parse the input request lambda event.
+    Validator for the raw input request lambda event
 
     Args:
         event: input dict
 
     Returns:
-        RawRequestInput: parsed request input
+        parsed request input
     """
     app_logger.info(f"event:{json.dumps(event)}...")
     try:

src/io/tms2geotiff.py

@@ -28,19 +28,20 @@ CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
 OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
 OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 """
-
 import concurrent.futures
 import io
 import itertools
 import math
 import re
 import time
-
+from typing import Tuple, Callable
+import PIL
 from PIL import Image
 
 from src import app_logger
 from src.utilities.constants import EARTH_EQUATORIAL_RADIUS, RETRY_DOWNLOAD, TIMEOUT_DOWNLOAD, TILE_SIZE, \
     CALLBACK_INTERVAL_DOWNLOAD
+from src.utilities.type_hints import PIL_Image
 
 Image.MAX_IMAGE_PIXELS = None
 
@@ -140,10 +141,27 @@ def print_progress(progress, total, done=False):
 
 
 def download_extent(
-        source, lat0, lon0, lat1, lon1, zoom,
-        save_image=True, progress_callback=print_progress,
-        callback_interval=CALLBACK_INTERVAL_DOWNLOAD
-):
+        source: str, lat0: float, lon0: float, lat1: float, lon1: float, zoom: int,
+        save_image: bool = True, progress_callback: Callable = print_progress,
+        callback_interval: float = CALLBACK_INTERVAL_DOWNLOAD
+) -> Tuple[PIL_Image, Tuple[float]] or Tuple[None]:
+    """
+    Download, merge and crop a list of tiles into a single geo-referenced image or a raster geodata
+
+    Args:
+        source: remote url tile
+        lat0: point0 bounding box latitude
+        lat1: point0 bounding box longitude
+        lon0: point1 bounding box latitude
+        lon1: point1 bounding box longitude
+        zoom: bounding box zoom
+        save_image: boolean to choose if save the image
+        progress_callback: callback function
+        callback_interval: process callback interval time
+
+    Returns:
+        parsed request input
+    """
     x0, y0 = deg2num(lat0, lon0, zoom)
     x1, y1 = deg2num(lat1, lon1, zoom)
     if x0 > x1:
@@ -185,16 +203,16 @@ def download_extent(
     y2 = round(base_size[1] * y_frac)
     img_w = round(base_size[0] * (x1 - x0))
     img_h = round(base_size[1] * (y1 - y0))
-    ret_im = big_im.crop((x2, y2, x2 + img_w, y2 + img_h))
-    if ret_im.mode == 'RGBA' and ret_im.getextrema()[3] == (255, 255):
-        ret_im = ret_im.convert('RGB')
+    final_image = big_im.crop((x2, y2, x2 + img_w, y2 + img_h))
+    if final_image.mode == 'RGBA' and final_image.getextrema()[3] == (255, 255):
+        final_image = final_image.convert('RGB')
     big_im.close()
     xp0, yp0 = from4326_to3857(lat0, lon0)
     xp1, yp1 = from4326_to3857(lat1, lon1)
-    p_width = abs(xp1 - xp0) / ret_im.size[0]
-    p_height = abs(yp1 - yp0) / ret_im.size[1]
-    matrix = (min(xp0, xp1), p_width, 0, max(yp0, yp1), 0, -p_height)
-    return ret_im, matrix
+    p_width = abs(xp1 - xp0) / final_image.size[0]
+    p_height = abs(yp1 - yp0) / final_image.size[1]
+    matrix = min(xp0, xp1), p_width, 0, max(yp0, yp1), 0, -p_height
+    return final_image, matrix
 
 
 def run_future_tile_download(

src/utilities/type_hints.py

@@ -1,50 +1,54 @@
 """custom type hints"""
 from enum import Enum
+from typing import TypedDict
 
+from PIL.Image import Image
 from pydantic import BaseModel
-from typing import TypedDict
 
 from src.utilities.constants import DEFAULT_TMS
 
+
 ts_dict_str2 = dict[str, str]
 ts_dict_str3 = dict[str, str, any]
 ts_ddict1 = dict[str, dict[str, any], dict, dict, any]
 list_float = list[float]
 llist_float = list[list_float]
+PIL_Image = Image
 
 
 class LatLngDict(BaseModel):
-    """A latitude-longitude type"""
+    """Generic geographic latitude-longitude type"""
     lat: float
     lng: float
 
 
 class PromptType(str, Enum):
     """Segment Anything enumeration prompt type"""
-    point = "point"
     # rectangle = "rectangle"
+    point = "point"
 
 
 class ImagePixelCoordinates(TypedDict):
+    """Image pixel coordinates type"""
     x: int
     y: int
 
 
 class RawBBox(BaseModel):
-    """Input lambda bbox request - not parsed"""
+    """Input lambda bbox request type (not parsed)"""
     ne: LatLngDict
     sw: LatLngDict
 
 
 class RawPrompt(BaseModel):
-    """Input lambda prompt request - not parsed"""
+    """Input lambda prompt request type (not parsed)"""
     type: PromptType
     data: LatLngDict
     label: int = 0
 
 
 class RawRequestInput(BaseModel):
-    """Input lambda request - not parsed"""
+    """Input lambda request validator type (not parsed)"""
     bbox: RawBBox
     prompt: list[RawPrompt]
     zoom: int | float

tests/test_app.py

@@ -14,7 +14,7 @@ class TestAppFailures(unittest.TestCase):
     @patch.object(app, "samexporter_predict")
     @patch.object(app, "get_parsed_bbox_points")
     @patch.object(app, "get_parsed_request_body")
-    def test_lambda_handler_400(
+    def test_lambda_handler_500(
             self,
             get_parsed_request_body_mocked,
             get_parsed_bbox_points_mocked,
@@ -35,16 +35,15 @@ class TestAppFailures(unittest.TestCase):
             cognito_identity=None,
             epoch_deadline_time_in_ms=time.time()
         )
-        expected_response_400 = '{"statusCode": 400, "header": {"Content-Type": "application/json"}, '
-        expected_response_400 += '"body": "{\\"duration_run\\": 0, \\"message\\": \\"Bad Request\\", '
-        expected_response_400 += '\\"request_id\\": \\"test_invoke_id\\"}", "isBase64Encoded": false}'
+        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}'
 
-        response_400 = lambda_handler(event, lambda_context)
-        assert response_400 == expected_response_400
+        assert lambda_handler(event, lambda_context) == expected_response_500
 
     @patch.object(time, "time")
     @patch.object(app, "get_parsed_request_body")
-    def test_lambda_handler_500(self, get_parsed_request_body_mocked, time_mocked):
+    def test_lambda_handler_400(self, get_parsed_request_body_mocked, time_mocked):
         from src.app import lambda_handler
 
         time_mocked.return_value = 0
@@ -58,14 +57,10 @@ class TestAppFailures(unittest.TestCase):
             epoch_deadline_time_in_ms=time.time()
         )
 
-        response_500 = lambda_handler(event, lambda_context)
-        check_500 = response_500 == (
-            '{"statusCode": 500, "header": {"Content-Type": "application/json"}, '
-            '"body": "{\\"duration_run\\": 0, \\"message\\": \\"Internal server error\\", '
+        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}')
-        print(f"test_lambda_handler_422:{check_500}.")
-        assert check_500
-        print("check_500")
 
     @patch.object(time, "time")
     def test_lambda_handler_422(self, time_mocked):