| """lambda helper functions""" | |
| import json | |
| import logging | |
| import time | |
| 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 | |
| """ | |
| 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: | |
| """ | |
| 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 | |
| """ | |
| 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 | |