aletrn commited on
Commit
9c09a5a
·
1 Parent(s): c6a47bd

[doc] update some docstring and typing hints

Browse files
src/app.py CHANGED
@@ -39,7 +39,8 @@ def lambda_handler(event: Dict, context: LambdaContext) -> str:
39
 
40
  try:
41
  body_response = samexporter_predict(body_request["bbox"], body_request["prompt"], body_request["zoom"])
42
- app_logger.info(f"output body_response:{body_response}.")
 
43
  response = get_response(HTTPStatus.OK.value, start_time, context.aws_request_id, body_response)
44
  except Exception as ex2:
45
  app_logger.error(f"exception2:{ex2}.")
@@ -51,5 +52,5 @@ def lambda_handler(event: Dict, context: LambdaContext) -> str:
51
  app_logger.error(f"exception1:{ex1}.")
52
  response = get_response(HTTPStatus.BAD_REQUEST.value, start_time, context.aws_request_id, {})
53
 
54
- app_logger.info(f"response_dumped:{response}...")
55
  return response
 
39
 
40
  try:
41
  body_response = samexporter_predict(body_request["bbox"], body_request["prompt"], body_request["zoom"])
42
+ app_logger.info(f"output body_response length:{len(body_response)}.")
43
+ app_logger.debug(f"output body_response:{body_response}.")
44
  response = get_response(HTTPStatus.OK.value, start_time, context.aws_request_id, body_response)
45
  except Exception as ex2:
46
  app_logger.error(f"exception2:{ex2}.")
 
52
  app_logger.error(f"exception1:{ex1}.")
53
  response = get_response(HTTPStatus.BAD_REQUEST.value, start_time, context.aws_request_id, {})
54
 
55
+ app_logger.debug(f"response_dumped:{response}...")
56
  return response
src/io/geo_helpers.py CHANGED
@@ -1,14 +1,15 @@
1
  """handle geo-referenced raster images"""
2
  from pathlib import Path
3
- from typing import List, Tuple, Dict
4
 
5
  import numpy as np
6
  from affine import Affine
7
 
8
- from src import app_logger, PROJECT_ROOT_FOLDER
 
9
 
10
 
11
- def load_affine_transformation_from_matrix(matrix_source_coefficients: List[float]) -> Affine:
12
  """
13
  Wrapper for rasterio.Affine.from_gdal() method
14
 
@@ -32,7 +33,7 @@ def load_affine_transformation_from_matrix(matrix_source_coefficients: List[floa
32
  raise e
33
 
34
 
35
- def get_affine_transform_from_gdal(matrix_source_coefficients: List[float] or Tuple[float]) -> Affine:
36
  """wrapper for rasterio Affine from_gdal method
37
 
38
  Args:
@@ -44,7 +45,7 @@ def get_affine_transform_from_gdal(matrix_source_coefficients: List[float] or Tu
44
  return Affine.from_gdal(*matrix_source_coefficients)
45
 
46
 
47
- def get_vectorized_raster_as_geojson(mask: np.ndarray, matrix: Tuple[float]) -> Dict[str, int]:
48
  """
49
  Get shapes and values of connected regions in a dataset or array
50
 
@@ -84,10 +85,3 @@ def get_vectorized_raster_as_geojson(mask: np.ndarray, matrix: Tuple[float]) ->
84
  except Exception as e_shape_band:
85
  app_logger.error(f"e_shape_band:{e_shape_band}.")
86
  raise e_shape_band
87
-
88
-
89
- if __name__ == '__main__':
90
- npy_file = "prediction_masks_46.27697017893455_9.616470336914064_46.11441972281433_9.264907836914064.npy"
91
- prediction_masks = np.load(Path(PROJECT_ROOT_FOLDER) / "tmp" / "try_by_steps" / "t0" / npy_file)
92
-
93
- print("#")
 
1
  """handle geo-referenced raster images"""
2
  from pathlib import Path
3
+ from typing import Dict
4
 
5
  import numpy as np
6
  from affine import Affine
7
 
8
+ from src import app_logger
9
+ from src.utilities.type_hints import list_float, tuple_float, dict_str_int
10
 
11
 
12
+ def load_affine_transformation_from_matrix(matrix_source_coefficients: list_float) -> Affine:
13
  """
14
  Wrapper for rasterio.Affine.from_gdal() method
15
 
 
33
  raise e
34
 
35
 
36
+ def get_affine_transform_from_gdal(matrix_source_coefficients: list_float or tuple_float) -> Affine:
37
  """wrapper for rasterio Affine from_gdal method
38
 
39
  Args:
 
45
  return Affine.from_gdal(*matrix_source_coefficients)
46
 
47
 
48
+ def get_vectorized_raster_as_geojson(mask: np.ndarray, matrix: tuple_float) -> dict_str_int:
49
  """
50
  Get shapes and values of connected regions in a dataset or array
51
 
 
85
  except Exception as e_shape_band:
86
  app_logger.error(f"e_shape_band:{e_shape_band}.")
87
  raise e_shape_band
 
 
 
 
 
 
 
src/prediction_api/predictors.py CHANGED
@@ -1,5 +1,4 @@
1
  """functions using machine learning instance model(s)"""
2
- from typing import Dict, Tuple
3
  from PIL.Image import Image
4
  import numpy as np
5
 
@@ -8,7 +7,7 @@ from src.io.geo_helpers import get_vectorized_raster_as_geojson, get_affine_tran
8
  from src.io.tms2geotiff import download_extent
9
  from src.prediction_api.sam_onnx import SegmentAnythingONNX
10
  from src.utilities.constants import MODEL_ENCODER_NAME, MODEL_DECODER_NAME, DEFAULT_TMS
11
- from src.utilities.type_hints import llist_float
12
 
13
 
14
  models_dict = {"fastsam": {"instance": None}}
@@ -20,7 +19,7 @@ def samexporter_predict(
20
  zoom: float,
21
  model_name: str = "fastsam",
22
  url_tile: str = DEFAULT_TMS
23
- ) -> Dict[str, int]:
24
  """
25
  Return predictions as a geojson from a geo-referenced image using the given input prompt.
26
 
@@ -37,7 +36,7 @@ def samexporter_predict(
37
  url_tile: server url tile
38
 
39
  Returns:
40
- dict: Affine transform
41
  """
42
  if models_dict[model_name]["instance"] is None:
43
  app_logger.info(f"missing instance model {model_name}, instantiating it now!")
@@ -67,9 +66,10 @@ def samexporter_predict(
67
 
68
 
69
  def get_raster_inference(
70
- img: Image, prompt: list[dict], models_instance: SegmentAnythingONNX, model_name: str
71
- ) -> Tuple[np.ndarray, int]:
72
- """wrapper for rasterio Affine from_gdal method
 
73
 
74
  Args:
75
  img: input PIL Image
 
1
  """functions using machine learning instance model(s)"""
 
2
  from PIL.Image import Image
3
  import numpy as np
4
 
 
7
  from src.io.tms2geotiff import download_extent
8
  from src.prediction_api.sam_onnx import SegmentAnythingONNX
9
  from src.utilities.constants import MODEL_ENCODER_NAME, MODEL_DECODER_NAME, DEFAULT_TMS
10
+ from src.utilities.type_hints import llist_float, dict_str_int, list_dict, tuple_ndarr_int
11
 
12
 
13
  models_dict = {"fastsam": {"instance": None}}
 
19
  zoom: float,
20
  model_name: str = "fastsam",
21
  url_tile: str = DEFAULT_TMS
22
+ ) -> dict_str_int:
23
  """
24
  Return predictions as a geojson from a geo-referenced image using the given input prompt.
25
 
 
36
  url_tile: server url tile
37
 
38
  Returns:
39
+ Affine transform
40
  """
41
  if models_dict[model_name]["instance"] is None:
42
  app_logger.info(f"missing instance model {model_name}, instantiating it now!")
 
66
 
67
 
68
  def get_raster_inference(
69
+ img: Image, prompt: list_dict, models_instance: SegmentAnythingONNX, model_name: str
70
+ ) -> tuple_ndarr_int:
71
+ """
72
+ Wrapper for rasterio Affine from_gdal method
73
 
74
  Args:
75
  img: input PIL Image
src/utilities/serialize.py CHANGED
@@ -2,7 +2,7 @@
2
  from typing import Mapping
3
 
4
  from src import app_logger
5
- from src.utilities.type_hints import ts_dict_str2, ts_dict_str3
6
 
7
 
8
  def serialize(obj: any, include_none: bool = False):
@@ -14,13 +14,12 @@ def serialize(obj: any, include_none: bool = False):
14
  include_none: bool to indicate if include also keys with None values during dict serialization
15
 
16
  Returns:
17
- any: serialized object
18
-
19
  """
20
  return _serialize(obj, include_none)
21
 
22
 
23
- def _serialize(obj: any, include_none: bool) -> any:
24
  import numpy as np
25
 
26
  primitive = (int, float, str, bool)
@@ -75,11 +74,11 @@ def _serialize_list(ls: list, include_none: bool) -> list:
75
  return [_serialize(elem, include_none) for elem in ls]
76
 
77
 
78
- def _serialize_bytes(b: bytes) -> ts_dict_str2:
79
  import base64
80
  encoded = base64.b64encode(b)
81
  return {"value": encoded.decode('ascii'), "type": "bytes"}
82
 
83
 
84
- def _serialize_exception(e: Exception) -> ts_dict_str3:
85
  return {"msg": str(e), "type": str(type(e)), **e.__dict__}
 
2
  from typing import Mapping
3
 
4
  from src import app_logger
5
+ from src.utilities.type_hints import dict_str, dict_str_any
6
 
7
 
8
  def serialize(obj: any, include_none: bool = False):
 
14
  include_none: bool to indicate if include also keys with None values during dict serialization
15
 
16
  Returns:
17
+ serialized object
 
18
  """
19
  return _serialize(obj, include_none)
20
 
21
 
22
+ def _serialize(obj: any, include_none: bool):
23
  import numpy as np
24
 
25
  primitive = (int, float, str, bool)
 
74
  return [_serialize(elem, include_none) for elem in ls]
75
 
76
 
77
+ def _serialize_bytes(b: bytes) -> dict_str:
78
  import base64
79
  encoded = base64.b64encode(b)
80
  return {"value": encoded.decode('ascii'), "type": "bytes"}
81
 
82
 
83
+ def _serialize_exception(e: Exception) -> dict_str_any:
84
  return {"msg": str(e), "type": str(type(e)), **e.__dict__}
src/utilities/type_hints.py CHANGED
@@ -3,16 +3,23 @@ from enum import Enum
3
  from typing import TypedDict
4
 
5
  from PIL.Image import Image
 
6
  from pydantic import BaseModel
7
 
8
  from src.utilities.constants import DEFAULT_TMS
9
 
10
 
11
- ts_dict_str2 = dict[str, str]
12
- ts_dict_str3 = dict[str, str, any]
13
- ts_ddict1 = dict[str, dict[str, any], dict, dict, any]
 
14
  list_float = list[float]
 
 
 
15
  llist_float = list[list_float]
 
 
16
  PIL_Image = Image
17
 
18
 
 
3
  from typing import TypedDict
4
 
5
  from PIL.Image import Image
6
+ from numpy import ndarray
7
  from pydantic import BaseModel
8
 
9
  from src.utilities.constants import DEFAULT_TMS
10
 
11
 
12
+ dict_str_int = dict[str, int]
13
+ dict_str = dict[str]
14
+ dict_str_any = dict[str, any]
15
+ list_dict = list[dict]
16
  list_float = list[float]
17
+ list_int = list[int]
18
+ tuple_int = tuple[int]
19
+ tuple_ndarr_int = tuple[ndarray, int]
20
  llist_float = list[list_float]
21
+ tuple_float = tuple[float]
22
+ tuple_float_any = tuple[float, any]
23
  PIL_Image = Image
24
 
25
 
src/utilities/utilities.py CHANGED
@@ -56,7 +56,7 @@ def base64_encode(sb: str or bytes) -> bytes:
56
  return base64.b64encode(sb_bytes)
57
 
58
 
59
- def hash_calculate(arr: any) -> str or bytes:
60
  """
61
  Return computed hash from input variable (typically a numpy array).
62
 
@@ -64,7 +64,7 @@ def hash_calculate(arr: any) -> str or bytes:
64
  arr: input variable
65
 
66
  Returns:
67
- str or bytes: computed hash from input variable
68
  """
69
  import hashlib
70
  import numpy as np
 
56
  return base64.b64encode(sb_bytes)
57
 
58
 
59
+ def hash_calculate(arr) -> str or bytes:
60
  """
61
  Return computed hash from input variable (typically a numpy array).
62
 
 
64
  arr: input variable
65
 
66
  Returns:
67
+ computed hash from input variable
68
  """
69
  import hashlib
70
  import numpy as np