File size: 4,063 Bytes
6f2f547 238719b 85dd36a 238719b 6f2f547 238719b 85dd36a 238719b 85dd36a 238719b 6f2f547 238719b 85dd36a 238719b 85dd36a 238719b 85dd36a 238719b 85dd36a 238719b 85dd36a 238719b 85dd36a 238719b 85dd36a 238719b 6f2f547 238719b 85dd36a 238719b 85dd36a 238719b 85dd36a ef3d72e 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 |
"""lambda helper functions"""
import json
import logging
import time
from typing import Dict, List
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:
"""
Return a response for frontend clients.
Args:
status: status response
start_time: request start time (float)
request_id: str
response_body: dict we embed into our response
Returns:
str: json response
"""
app_logger.debug(f"response_body:{response_body}.")
response_body["duration_run"] = time.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),
"isBase64Encoded": False
}
app_logger.debug(f"response type:{type(response)} => {response}.")
return json.dumps(response)
def get_parsed_bbox_points(request_input: RawRequestInput) -> Dict:
"""
Format the bbox and prompt request input
Args:
request_input: input dict
Returns:
Dict:
"""
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(f"unpacking elaborated request...")
return {
"bbox": [ne_latlng, sw_latlng],
"prompt": new_prompt_list,
"zoom": new_zoom
}
def get_parsed_request_body(event: Dict):
"""
Parse the input request lambda event.
Args:
event: input dict
Returns:
RawRequestInput: parsed request input
"""
app_logger.info(f"event:{json.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 = json.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)}.")
return parsed_body
|