[refactor] don't import complete modules, use only "from" syntax

src/__init__.py

@@ -1,6 +1,5 @@
 """Get machine learning predictions from geodata raster images"""
 from aws_lambda_powertools import Logger
-import os
 from pathlib import Path
 
 from src.utilities.constants import SERVICE_NAME

src/app.py

@@ -1,5 +1,4 @@
 """Lambda entry point"""
-import time
 from http import HTTPStatus
 from typing import Dict
 
@@ -24,8 +23,9 @@ def lambda_handler(event: Dict, context: LambdaContext) -> str:
         json response from get_response() function
 
     """
+    from time import time
     app_logger.info(f"start with aws_request_id:{context.aws_request_id}.")
-    start_time = time.time()
+    start_time = time()
 
     if "version" in event:
         app_logger.info(f"event version: {event['version']}.")

src/io/coordinates_pixel_conversion.py

@@ -1,6 +1,4 @@
 """functions useful to convert to/from latitude-longitude coordinates to pixel image coordinates"""
-import math
-
 from src import app_logger
 from src.utilities.constants import TILE_SIZE
 from src.utilities.type_hints import ImagePixelCoordinates
@@ -8,17 +6,19 @@ from src.utilities.type_hints import LatLngDict
 
 
 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 = math.sin(latlng.lat * math.pi / 180)
+        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 - math.log((1 + sin_y) / (1 - sin_y)) / (4 * math.pi))
+        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}
@@ -28,14 +28,16 @@ def _get_latlng2pixel_projection(latlng: LatLngDict) -> ImagePixelCoordinates:
 
 
 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=math.floor(world_coordinate["x"] * scale),
-            y=math.floor(world_coordinate["y"] * scale)
+            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'format_latlng_to_pixel_coordinates:{e_format_latlng_to_pixel_coordinates}.')

src/io/geo_helpers.py

@@ -1,9 +1,6 @@
 """handle geo-referenced raster images"""
-from pathlib import Path
-from typing import Dict
-
-import numpy as np
 from affine import Affine
+from numpy import ndarray as np_ndarray
 
 from src import app_logger
 from src.utilities.type_hints import list_float, tuple_float, dict_str_int
@@ -45,7 +42,7 @@ def get_affine_transform_from_gdal(matrix_source_coefficients: list_float or tup
     return Affine.from_gdal(*matrix_source_coefficients)
 
 
-def get_vectorized_raster_as_geojson(mask: np.ndarray, matrix: tuple_float) -> dict_str_int:
+def get_vectorized_raster_as_geojson(mask: np_ndarray, matrix: tuple_float) -> dict_str_int:
     """
         Get shapes and values of connected regions in a dataset or array
 

src/io/lambda_helpers.py

@@ -1,7 +1,4 @@
 """lambda helper functions"""
-import json
-import logging
-import time
 from typing import Dict
 from aws_lambda_powertools.event_handler import content_types
 
@@ -26,19 +23,22 @@ def get_response(status: int, start_time: float, request_id: str, response_body:
         json response
 
     """
+    from json import dumps
+    from time import time
+
     app_logger.debug(f"response_body:{response_body}.")
-    response_body["duration_run"] = time.time() - start_time
+    response_body["duration_run"] = time() - start_time
     response_body["message"] = CUSTOM_RESPONSE_MESSAGES[status]
     response_body["request_id"] = request_id
 
     response = {
         "statusCode": status,
         "header": {"Content-Type": content_types.APPLICATION_JSON},
-        "body": json.dumps(response_body),
+        "body": dumps(response_body),
         "isBase64Encoded": False
     }
     app_logger.debug(f"response type:{type(response)} => {response}.")
-    return json.dumps(response)
+    return dumps(response)
 
 
 def get_parsed_bbox_points(request_input: RawRequestInput) -> Dict:
@@ -98,7 +98,10 @@ def get_parsed_request_body(event: Dict) -> RawRequestInput:
     Returns:
         parsed request input
     """
-    app_logger.info(f"event:{json.dumps(event)}...")
+    from json import dumps, loads
+    from logging import getLevelName
+
+    app_logger.info(f"event:{dumps(event)}...")
     try:
         raw_body = event["body"]
     except Exception as e_constants1:
@@ -108,12 +111,12 @@ def get_parsed_request_body(event: Dict) -> RawRequestInput:
     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}...")
-        raw_body = json.loads(body_decoded_str)
+        raw_body = loads(body_decoded_str)
     app_logger.info(f"body, #2: {type(raw_body)}, {raw_body}...")
 
     parsed_body = RawRequestInput.model_validate(raw_body)
     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)}.")
+    app_logger.warning(f"set log level to {getLevelName(app_logger.log_level)}.")
 
     return parsed_body

src/prediction_api/predictors.py

@@ -1,6 +1,6 @@
 """functions using machine learning instance model(s)"""
 from PIL.Image import Image
-import numpy as np
+from numpy import array as np_array, uint8, zeros
 
 from src import app_logger, MODEL_FOLDER
 from src.io.geo_helpers import get_vectorized_raster_as_geojson, get_affine_transform_from_gdal
@@ -80,7 +80,7 @@ def get_raster_inference(
     Returns:
         raster prediction mask, prediction number
     """
-    np_img = np.array(img)
+    np_img = np_array(img)
     app_logger.info(f"img type {type(np_img)}, prompt:{prompt}.")
     app_logger.debug(f"onnxruntime input shape/size (shape if PIL) {np_img.size}.")
     try:
@@ -95,7 +95,7 @@ def get_raster_inference(
     len_inference_out = len(inference_out[0, :, :, :])
     app_logger.info(f"Created {len_inference_out} prediction_masks,"
                     f"shape:{inference_out.shape}, dtype:{inference_out.dtype}.")
-    mask = np.zeros((inference_out.shape[2], inference_out.shape[3]), dtype=np.uint8)
+    mask = zeros((inference_out.shape[2], inference_out.shape[3]), dtype=uint8)
     for n, m in enumerate(inference_out[0, :, :, :]):
         app_logger.debug(f"{n}th of prediction_masks shape {inference_out.shape}"
                          f" => mask shape:{mask.shape}, {mask.dtype}.")

src/prediction_api/sam_onnx.py

@@ -1,5 +1,5 @@
 """
-Define a machine learning model executed by ONNX Runtime (https://onnxruntime.ai/)
+Define a machine learning model executed by ONNX Runtime (https://ai/)
 for Segment Anything (https://segment-anything.com).
 Modified from https://github.com/vietanhdev/samexporter/
 
@@ -24,10 +24,9 @@ OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 SOFTWARE.
 """
 from copy import deepcopy
-
-import cv2
-import numpy as np
-import onnxruntime
+from numpy import array as np_array, concatenate, float32, linalg, matmul, ndarray, ones, zeros
+from cv2 import INTER_LINEAR, warpAffine
+from onnxruntime import get_available_providers, InferenceSession
 
 from src import app_logger
 
@@ -40,7 +39,7 @@ class SegmentAnythingONNX:
         self.input_size = (684, 1024)
 
         # Load models
-        providers = onnxruntime.get_available_providers()
+        providers = get_available_providers()
 
         # Pop TensorRT Runtime due to crashing issues
         # TODO: Add back when TensorRT backend is stable
@@ -52,11 +51,11 @@ class SegmentAnythingONNX:
             )
         else:
             app_logger.warning("No available providers for ONNXRuntime")
-        self.encoder_session = onnxruntime.InferenceSession(
+        self.encoder_session = InferenceSession(
             encoder_model_path, providers=providers
         )
         self.encoder_input_name = self.encoder_session.get_inputs()[0].name
-        self.decoder_session = onnxruntime.InferenceSession(
+        self.decoder_session = InferenceSession(
             decoder_model_path, providers=providers
         )
 
@@ -76,7 +75,7 @@ class SegmentAnythingONNX:
                 )  # bottom right
                 labels.append(2)
                 labels.append(3)
-        points, labels = np.array(points), np.array(labels)
+        points, labels = np_array(points), np_array(labels)
         return points, labels
 
     def run_encoder(self, encoder_inputs):
@@ -96,9 +95,9 @@ class SegmentAnythingONNX:
         new_h = int(new_h + 0.5)
         return new_h, new_w
 
-    def apply_coords(self, coords: np.ndarray, original_size, target_length):
+    def apply_coords(self, coords: ndarray, original_size, target_length):
         """
-        Expects a numpy array of length 2 in the final dimension. Requires the
+        Expects a numpy np_array of length 2 in the final dimension. Requires the
         original image size in (H, W) format.
         """
         old_h, old_w = original_size
@@ -117,30 +116,30 @@ class SegmentAnythingONNX:
         input_points, input_labels = self.get_input_points(prompt)
 
         # Add a batch index, concatenate a padding point, and transform.
-        onnx_coord = np.concatenate(
-            [input_points, np.array([[0.0, 0.0]])], axis=0
+        onnx_coord = concatenate(
+            [input_points, np_array([[0.0, 0.0]])], axis=0
         )[None, :, :]
-        onnx_label = np.concatenate([input_labels, np.array([-1])], axis=0)[
+        onnx_label = concatenate([input_labels, np_array([-1])], axis=0)[
             None, :
-        ].astype(np.float32)
+        ].astype(float32)
         onnx_coord = self.apply_coords(
             onnx_coord, self.input_size, self.target_size
-        ).astype(np.float32)
+        ).astype(float32)
 
         # Apply the transformation matrix to the coordinates.
-        onnx_coord = np.concatenate(
+        onnx_coord = concatenate(
             [
                 onnx_coord,
-                np.ones((1, onnx_coord.shape[1], 1), dtype=np.float32),
+                ones((1, onnx_coord.shape[1], 1), dtype=float32),
             ],
             axis=2,
         )
-        onnx_coord = np.matmul(onnx_coord, transform_matrix.T)
-        onnx_coord = onnx_coord[:, :, :2].astype(np.float32)
+        onnx_coord = matmul(onnx_coord, transform_matrix.T)
+        onnx_coord = onnx_coord[:, :, :2].astype(float32)
 
         # Create an empty mask input and an indicator for no mask.
-        onnx_mask_input = np.zeros((1, 1, 256, 256), dtype=np.float32)
-        onnx_has_mask_input = np.zeros(1, dtype=np.float32)
+        onnx_mask_input = zeros((1, 1, 256, 256), dtype=float32)
+        onnx_has_mask_input = zeros(1, dtype=float32)
 
         decoder_inputs = {
             "image_embeddings": image_embedding,
@@ -148,12 +147,12 @@ class SegmentAnythingONNX:
             "point_labels": onnx_label,
             "mask_input": onnx_mask_input,
             "has_mask_input": onnx_has_mask_input,
-            "orig_im_size": np.array(self.input_size, dtype=np.float32),
+            "orig_im_size": np_array(self.input_size, dtype=float32),
         }
         masks, _, _ = self.decoder_session.run(None, decoder_inputs)
 
         # Transform the masks back to the original image size.
-        inv_transform_matrix = np.linalg.inv(transform_matrix)
+        inv_transform_matrix = linalg.inv(transform_matrix)
         transformed_masks = self.transform_masks(
             masks, original_size, inv_transform_matrix
         )
@@ -175,11 +174,11 @@ class SegmentAnythingONNX:
                         app_logger.debug(f"mask_shape transform_masks:{mask.shape}, dtype:{mask.dtype}.")
                     except Exception as e_mask_shape_transform_masks:
                         app_logger.error(f"e_mask_shape_transform_masks:{e_mask_shape_transform_masks}.")
-                    mask = cv2.warpAffine(
+                    mask = warpAffine(
                         mask,
                         transform_matrix[:2],
                         (original_size[1], original_size[0]),
-                        flags=cv2.INTER_LINEAR,
+                        flags=INTER_LINEAR,
                     )
                 except Exception as e_warp_affine1:
                     app_logger.error(f"e_warp_affine1 mask shape:{mask.shape}, dtype:{mask.dtype}.")
@@ -188,7 +187,7 @@ class SegmentAnythingONNX:
                     raise e_warp_affine1
                 batch_masks.append(mask)
             output_masks.append(batch_masks)
-        return np.array(output_masks)
+        return np_array(output_masks)
 
     def encode(self, cv_image):
         """
@@ -200,7 +199,7 @@ class SegmentAnythingONNX:
         scale_x = self.input_size[1] / cv_image.shape[1]
         scale_y = self.input_size[0] / cv_image.shape[0]
         scale = min(scale_x, scale_y)
-        transform_matrix = np.array(
+        transform_matrix = np_array(
             [
                 [scale, 0, 0],
                 [0, scale, 0],
@@ -208,22 +207,22 @@ class SegmentAnythingONNX:
             ]
         )
         try:
-            cv_image = cv2.warpAffine(
+            cv_image = warpAffine(
                 cv_image,
                 transform_matrix[:2],
                 (self.input_size[1], self.input_size[0]),
-                flags=cv2.INTER_LINEAR,
+                flags=INTER_LINEAR,
             )
         except Exception as e_warp_affine2:
             app_logger.error(f"e_warp_affine2:{e_warp_affine2}.")
-            np_cv_image = np.array(cv_image)
+            np_cv_image = np_array(cv_image)
             app_logger.error(f"e_warp_affine2 cv_image shape:{np_cv_image.shape}, dtype:{np_cv_image.dtype}.")
             app_logger.error(f"e_warp_affine2 transform_matrix:{transform_matrix}, [:2] {transform_matrix[:2]}")
             app_logger.error(f"e_warp_affine2 self.input_size:{self.input_size}.")
             raise e_warp_affine2
 
         encoder_inputs = {
-            self.encoder_input_name: cv_image.astype(np.float32),
+            self.encoder_input_name: cv_image.astype(float32),
         }
         image_embedding = self.run_encoder(encoder_inputs)
         return {

src/utilities/serialize.py

@@ -20,18 +20,18 @@ def serialize(obj: any, include_none: bool = False):
 
 
 def _serialize(obj: any, include_none: bool):
-    import numpy as np
+    from numpy import ndarray as np_ndarray, floating as np_floating, integer as np_integer
 
     primitive = (int, float, str, bool)
     # print(type(obj))
     try:
         if obj is None:
             return None
-        elif isinstance(obj, np.integer):
+        elif isinstance(obj, np_integer):
             return int(obj)
-        elif isinstance(obj, np.floating):
+        elif isinstance(obj, np_floating):
             return float(obj)
-        elif isinstance(obj, np.ndarray):
+        elif isinstance(obj, np_ndarray):
             return obj.tolist()
         elif isinstance(obj, primitive):
             return obj

src/utilities/utilities.py

@@ -66,26 +66,26 @@ def hash_calculate(arr) -> str or bytes:
     Returns:
         computed hash from input variable
     """
-    import hashlib
-    import numpy as np
+    from hashlib import sha256
     from base64 import b64encode
+    from numpy import ndarray as np_ndarray
 
-    if isinstance(arr, np.ndarray):
-        hash_fn = hashlib.sha256(arr.data)
+    if isinstance(arr, np_ndarray):
+        hash_fn = sha256(arr.data)
     elif isinstance(arr, dict):
         import json
 
         serialized = serialize(arr)
         variable_to_hash = json.dumps(serialized, sort_keys=True).encode('utf-8')
-        hash_fn = hashlib.sha256(variable_to_hash)
+        hash_fn = sha256(variable_to_hash)
     elif isinstance(arr, str):
         try:
-            hash_fn = hashlib.sha256(arr)
+            hash_fn = sha256(arr)
         except TypeError:
             app_logger.warning(f"TypeError, re-try encoding arg:{arr},type:{type(arr)}.")
-            hash_fn = hashlib.sha256(arr.encode('utf-8'))
+            hash_fn = sha256(arr.encode('utf-8'))
     elif isinstance(arr, bytes):
-        hash_fn = hashlib.sha256(arr)
+        hash_fn = sha256(arr)
     else:
         raise ValueError(f"variable 'arr':{arr} not yet handled.")
     return b64encode(hash_fn.digest())