File size: 3,531 Bytes
243f395 f2a79fa 3dbe011 fa76f5f 43d87b3 150cbc9 c6054f0 f2a79fa 7d6e00c fa76f5f 43d87b3 243f395 7d6e00c fa76f5f f2a79fa 3dbe011 7d6e00c f2a79fa 7d6e00c f2a79fa 26b4f04 ee50e01 7d6e00c 150cbc9 6f1250c 7d6e00c ddb71db 8c1934b 150cbc9 7d6e00c 150cbc9 f2a79fa 43d87b3 3dbe011 7d6e00c f2a79fa 6f1250c 7d6e00c 6f1250c c6054f0 7d6e00c da5737b 43d87b3 fa76f5f 7d6e00c 43d87b3 ee50e01 43d87b3 ee50e01 f2a79fa 7d6e00c f2a79fa |
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 |
import json
import time
from http import HTTPStatus
from typing import Dict
from aws_lambda_powertools.event_handler import content_types
from aws_lambda_powertools.utilities.typing import LambdaContext
from src import app_logger
from src.prediction_api.predictors import samexporter_predict
from src.utilities.constants import CUSTOM_RESPONSE_MESSAGES
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.info(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.info(f"response type:{type(response)} => {response}.")
return json.dumps(response)
def get_parsed_bbox_points(request_input: Dict) -> Dict:
app_logger.info(f"try to parsing input request {request_input}...")
ne = request_input["ne"]
sw = request_input["sw"]
bbox = [
[float(ne["lat"]), float(ne["lng"])],
[float(sw["lat"]), float(sw["lng"])]
]
app_logger.info(f"bbox {bbox}.")
app_logger.info(f"unpacking {request_input}...")
return {
"bbox": bbox,
"prompt": request_input["prompt"],
"zoom": int(request_input["zoom"])
}
def lambda_handler(event: dict, context: LambdaContext):
app_logger.info(f"start with aws_request_id:{context.aws_request_id}.")
start_time = time.time()
if "version" in event:
app_logger.info(f"event version: {event['version']}.")
try:
app_logger.info(f"event:{json.dumps(event)}...")
app_logger.info(f"context:{context}...")
try:
body = event["body"]
except Exception as e_constants1:
app_logger.error(f"e_constants1:{e_constants1}.")
body = event
app_logger.info(f"body: {type(body)}, {body}...")
if isinstance(body, str):
body_decoded_str = base64_decode(body)
app_logger.info(f"body_decoded_str: {type(body_decoded_str)}, {body_decoded_str}...")
body = json.loads(body_decoded_str)
app_logger.info(f"body:{body}...")
try:
body_request = get_parsed_bbox_points(body)
body_response = samexporter_predict(body_request["bbox"], body_request["prompt"], body_request["zoom"])
app_logger.info(f"body_response::output:{body_response}.")
response = get_response(HTTPStatus.OK.value, start_time, context.aws_request_id, body_response)
except Exception as ex2:
app_logger.error(f"exception2:{ex2}.")
response = get_response(HTTPStatus.UNPROCESSABLE_ENTITY.value, start_time, context.aws_request_id, {})
except Exception as ex1:
app_logger.error(f"exception1:{ex1}.")
response = get_response(HTTPStatus.INTERNAL_SERVER_ERROR.value, start_time, context.aws_request_id, {})
app_logger.info(f"response_dumped:{response}...")
return response
|