Spaces:
Running
Running
| """lambda helper functions""" | |
| import logging | |
| from sys import stdout | |
| from typing import Dict | |
| import loguru | |
| from xyzservices import providers, TileProvider | |
| from lisa_on_cuda.utils.app_helpers import get_cleaned_input | |
| from samgis_lisa_on_cuda import app_logger | |
| from samgis_lisa_on_cuda.io.coordinates_pixel_conversion import get_latlng_to_pixel_coordinates | |
| from samgis_lisa_on_cuda.utilities.constants import COMPLETE_URL_TILES_MAPBOX, COMPLETE_URL_TILES_NEXTZEN, CUSTOM_RESPONSE_MESSAGES | |
| from samgis_lisa_on_cuda.utilities.type_hints import ApiRequestBody, ContentTypes, XYZTerrainProvidersNames, \ | |
| XYZDefaultProvidersNames, StringPromptApiRequestBody | |
| from samgis_core.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": ContentTypes.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_with_string_prompt(request_input: StringPromptApiRequestBody) -> Dict: | |
| """ | |
| Parse the raw input request into bbox, prompt string and zoom | |
| Args: | |
| request_input: input dict | |
| Returns: | |
| dict with bounding box, prompt string 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) | |
| cleaned_prompt = get_cleaned_input(request_input.string_prompt) | |
| app_logger.debug(f"bbox => {bbox}.") | |
| app_logger.debug(f'request_input-prompt cleaned => {cleaned_prompt}.') | |
| app_logger.info("unpacking elaborated request...") | |
| return { | |
| "bbox": [ne_latlng, sw_latlng], | |
| "prompt": cleaned_prompt, | |
| "zoom": new_zoom, | |
| "source": get_url_tile(request_input.source_type) | |
| } | |
| def get_parsed_bbox_points_with_dictlist_prompt(request_input: ApiRequestBody) -> 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 = _get_parsed_prompt_list(ne, sw, new_zoom, request_input.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, | |
| "source": get_url_tile(request_input.source_type) | |
| } | |
| def _get_parsed_prompt_list(bbox_ne, bbox_sw, zoom, prompt_list): | |
| new_prompt_list = [] | |
| for prompt in prompt_list: | |
| app_logger.debug(f"current prompt: {type(prompt)}, value:{prompt}.") | |
| new_prompt = {"type": prompt.type.value} | |
| if prompt.type == "point": | |
| new_prompt_data = _get_new_prompt_data_point(bbox_ne, bbox_sw, prompt, zoom) | |
| new_prompt["label"] = prompt.label.value | |
| elif prompt.type == "rectangle": | |
| new_prompt_data = _get_new_prompt_data_rectangle(bbox_ne, bbox_sw, prompt, zoom) | |
| else: | |
| msg = "Valid prompt type: 'point' or 'rectangle', not '{}'. Check ApiRequestBody parsing/validation." | |
| raise TypeError(msg.format(prompt.type)) | |
| app_logger.debug(f"new_prompt_data: {type(new_prompt_data)}, value:{new_prompt_data}.") | |
| new_prompt["data"] = new_prompt_data | |
| new_prompt_list.append(new_prompt) | |
| return new_prompt_list | |
| def _get_new_prompt_data_point(bbox_ne, bbox_sw, prompt, zoom): | |
| current_point = get_latlng_to_pixel_coordinates(bbox_ne, bbox_sw, prompt.data, zoom, prompt.type) | |
| app_logger.debug(f"current prompt: {type(current_point)}, value:{current_point}, label: {prompt.label}.") | |
| return [current_point['x'], current_point['y']] | |
| def _get_new_prompt_data_rectangle(bbox_ne, bbox_sw, prompt, zoom): | |
| current_point_ne = get_latlng_to_pixel_coordinates(bbox_ne, bbox_sw, prompt.data.ne, zoom, prompt.type) | |
| app_logger.debug( | |
| f"rectangle:: current_point_ne prompt: {type(current_point_ne)}, value:{current_point_ne}.") | |
| current_point_sw = get_latlng_to_pixel_coordinates(bbox_ne, bbox_sw, prompt.data.sw, zoom, prompt.type) | |
| app_logger.debug( | |
| f"rectangle:: current_point_sw prompt: {type(current_point_sw)}, value:{current_point_sw}.") | |
| # correct order for rectangle prompt | |
| return [ | |
| current_point_sw["x"], | |
| current_point_ne["y"], | |
| current_point_ne["x"], | |
| current_point_sw["y"] | |
| ] | |
| def get_parsed_request_body(event: Dict or str) -> ApiRequestBody: | |
| """ | |
| 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 | |
| def _get_current_log_level(logger: loguru.logger) -> [str, loguru._logger.Level]: | |
| levels = logger._core.levels | |
| current_log_level = logger._core.min_level | |
| level_filt = [l for l in levels.items() if l[1].no == current_log_level] | |
| return level_filt[0] | |
| 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 = ApiRequestBody.model_validate(raw_body) | |
| log_level = "DEBUG" if parsed_body.debug else "INFO" | |
| app_logger.remove() | |
| app_logger.add(stdout, level=log_level) | |
| try: | |
| current_log_level_name, _ = _get_current_log_level(app_logger) | |
| app_logger.warning(f"set log level to {getLevelName(current_log_level_name)}.") | |
| except Exception as ex: | |
| print("failing setting parsing bbox, logger is ok? ex:", ex, "#") | |
| return parsed_body | |
| mapbox_terrain_rgb = TileProvider( | |
| name=XYZTerrainProvidersNames.MAPBOX_TERRAIN_TILES_NAME, | |
| url=COMPLETE_URL_TILES_MAPBOX, | |
| attribution="" | |
| ) | |
| nextzen_terrain_rgb = TileProvider( | |
| name=XYZTerrainProvidersNames.NEXTZEN_TERRAIN_TILES_NAME, | |
| url=COMPLETE_URL_TILES_NEXTZEN, | |
| attribution="" | |
| ) | |
| def get_url_tile(source_type: str): | |
| try: | |
| match source_type.lower(): | |
| case XYZDefaultProvidersNames.DEFAULT_TILES_NAME_SHORT: | |
| return providers.query_name(XYZDefaultProvidersNames.DEFAULT_TILES_NAME) | |
| case XYZTerrainProvidersNames.MAPBOX_TERRAIN_TILES_NAME: | |
| return mapbox_terrain_rgb | |
| case XYZTerrainProvidersNames.NEXTZEN_TERRAIN_TILES_NAME: | |
| app_logger.info("nextzen_terrain_rgb:", nextzen_terrain_rgb) | |
| return nextzen_terrain_rgb | |
| case _: | |
| return providers.query_name(source_type) | |
| except ValueError as ve: | |
| from pydantic_core import ValidationError | |
| app_logger.error("ve:", str(ve)) | |
| raise ValidationError(ve) | |
| def check_source_type_is_terrain(source: str | TileProvider): | |
| return isinstance(source, TileProvider) and source.name in list(XYZTerrainProvidersNames) | |
| def get_source_name(source: str | TileProvider) -> str | bool: | |
| try: | |
| match source.lower(): | |
| case XYZDefaultProvidersNames.DEFAULT_TILES_NAME_SHORT: | |
| source_output = providers.query_name(XYZDefaultProvidersNames.DEFAULT_TILES_NAME) | |
| case _: | |
| source_output = providers.query_name(source) | |
| if isinstance(source_output, str): | |
| return source_output | |
| try: | |
| source_dict = dict(source_output) | |
| app_logger.info(f"source_dict:{type(source_dict)}, {'name' in source_dict}, source_dict:{source_dict}.") | |
| return source_dict["name"] | |
| except KeyError as ke: | |
| app_logger.error(f"ke:{ke}.") | |
| except ValueError as ve: | |
| app_logger.info(f"source name::{source}, ve:{ve}.") | |
| app_logger.info(f"source name::{source}.") | |
| return False | |