File size: 4,144 Bytes
6f2f547 3209d49 238719b 85dd36a 238719b 85dd36a 238719b 3209d49 238719b 3209d49 238719b fd9de0f 238719b fd9de0f 238719b fd9de0f 238719b fd9de0f 238719b 6f2f547 3209d49 6f2f547 3209d49 6f2f547 238719b 85dd36a 238719b 85dd36a 238719b 85dd36a 238719b 85dd36a 238719b 85dd36a 238719b ee2ed59 238719b 85dd36a 238719b 3209d49 6f2f547 3209d49 6f2f547 3209d49 6f2f547 fd9de0f 238719b 85dd36a 238719b 85dd36a 238719b fd9de0f 85dd36a ef3d72e 85dd36a fd9de0f 85dd36a |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 |
"""lambda helper functions"""
from typing import Dict
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
from src.utilities.constants import CUSTOM_RESPONSE_MESSAGES
from src.utilities.type_hints import RawRequestInput
from src.utilities.utilities import base64_decode
def get_response(status: int, start_time: float, request_id: str, response_body: Dict = None) -> str:
"""
Response composer
Args:
status: status response
start_time: request start time (float)
request_id: str
response_body: dict we embed into our response
Returns:
json response
"""
from json import dumps
from time import time
app_logger.debug(f"response_body:{response_body}.")
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": dumps(response_body),
"isBase64Encoded": False
}
app_logger.debug(f"response type:{type(response)} => {response}.")
return dumps(response)
def get_parsed_bbox_points(request_input: RawRequestInput) -> Dict:
"""
Parse the raw input request into bbox, prompt and zoom
Args:
request_input: input dict
Returns:
dict with bounding box, prompt and zoom
"""
app_logger.info(f"try to parsing input request {request_input}...")
bbox = request_input.bbox
app_logger.debug(f"request bbox: {type(bbox)}, value:{bbox}.")
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)]
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}.")
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 => {new_prompt_list}.')
app_logger.info("unpacking elaborated request...")
return {
"bbox": [ne_latlng, sw_latlng],
"prompt": new_prompt_list,
"zoom": new_zoom
}
def get_parsed_request_body(event: Dict) -> RawRequestInput:
"""
Validator for the raw input request lambda event
Args:
event: input dict
Returns:
parsed request input
"""
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:
app_logger.error(f"e_constants1:{e_constants1}.")
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}...")
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 {getLevelName(app_logger.log_level)}.")
return parsed_body
|