| """Async download raster tiles""" | |
| import os | |
| from pathlib import Path | |
| import numpy as np | |
| from src import app_logger, PROJECT_ROOT_FOLDER | |
| from src.io.helpers import get_lat_lon_coords, merge_tiles, get_geojson_square_angles, crop_raster | |
| from src.io.tms2geotiff import download_extent, save_geotiff_gdal | |
| from src.utilities.constants import COMPLETE_URL_TILES, DEFAULT_TMS | |
| from src.utilities.type_hints import ts_llist2 | |
| COOKIE_SESSION = { | |
| "Accept": "*/*", | |
| "Accept-Encoding": "gzip, deflate", | |
| "User-Agent": "Mozilla/5.0 (Windows NT 10.0; rv:91.0) Gecko/20100101 Firefox/91.0", | |
| } | |
| def load_affine_transformation_from_matrix(matrix_source_coeffs): | |
| from affine import Affine | |
| if len(matrix_source_coeffs) != 6: | |
| raise ValueError(f"Expected 6 coefficients, found {len(matrix_source_coeffs)};" | |
| f"argument type: {type(matrix_source_coeffs)}.") | |
| try: | |
| a, d, b, e, c, f = (float(x) for x in matrix_source_coeffs) | |
| center = tuple.__new__(Affine, [a, b, c, d, e, f, 0.0, 0.0, 1.0]) | |
| return center * Affine.translation(-0.5, -0.5) | |
| except Exception as e: | |
| app_logger.error(f"exception:{e}, check https://github.com/rasterio/affine project for updates") | |
| # @timing_decorator | |
| def convert(bounding_box: ts_llist2, zoom: int) -> tuple: | |
| """ | |
| Starting from a bounding box of two couples of latitude and longitude coordinate values, recognize a stratovolcano from an RGB image. The algorithm | |
| create the image composing three channels as slope, DEM (Digital Elevation Model) and curvature. In more detail: | |
| - download a series of terrain DEM (Digital Elevation Model) raster tiles enclosed within that bounding box | |
| - merge all the downloaded rasters | |
| - crop the merged raster | |
| - process the cropped raster to extract slope and curvature (1st and 2nd degree derivative) | |
| - produce three raster channels (DEM, slope and curvature rasters) to produce an RGB raster image | |
| - submit the RGB image to a remote machine learning service to try to recognize a polygon representing a stratovolcano | |
| - the output of the machine learning service is a json, so we need to georeferencing it | |
| - finally we return a dict as response containing | |
| - uploaded_file_name | |
| - bucket_name | |
| - prediction georeferenced geojson-like dict | |
| Args: | |
| bounding_box: float latitude/longitude bounding box | |
| zoom: integer zoom value | |
| Returns: | |
| dict: uploaded_file_name (str), bucket_name (str), prediction_georef (dict), n_total_obj_prediction (str) | |
| """ | |
| import tempfile | |
| # from src.surferdtm_prediction_api.utilities.constants import NODATA_VALUES | |
| # from src.surferdtm_prediction_api.utilities.utilities import setup_logging | |
| # from src.surferdtm_prediction_api.raster.elaborate_images import elaborate_images.get_rgb_prediction_image | |
| # from src.surferdtm_prediction_api.raster.prediction import model_prediction | |
| # from src.surferdtm_prediction_api.geo.helpers import get_lat_lon_coords, merge_tiles, crop_raster, get_prediction_georeferenced, \ | |
| # get_geojson_square_angles, get_perc | |
| # app_logger = setup_logging(debug) | |
| ext = "tif" | |
| debug = False | |
| tile_source = COMPLETE_URL_TILES | |
| app_logger.info(f"start_args: tile_source:{tile_source},bounding_box:{bounding_box},zoom:{zoom}.") | |
| try: | |
| import rasterio | |
| lon_min, lat_min, lon_max, lat_max = get_lat_lon_coords(bounding_box) | |
| with tempfile.TemporaryDirectory() as input_tmp_dir: | |
| # with tempfile.TemporaryDirectory() as output_tmp_dir: | |
| output_tmp_dir = input_tmp_dir | |
| app_logger.info(f'tile_source: {tile_source}!') | |
| app_logger.info(f'created temporary input/output directory: {input_tmp_dir} => {output_tmp_dir}!') | |
| pt0, pt1 = bounding_box | |
| app_logger.info("downloading...") | |
| img, matrix = download_extent(DEFAULT_TMS, pt0[0], pt0[1], pt1[0], pt1[1], zoom) | |
| app_logger.info(f'img: type {type(img)}, len_matrix:{len(matrix)}, matrix {matrix}.') | |
| app_logger.info(f'img: size (shape if PIL) {img.size}.') | |
| try: | |
| np_img = np.array(img) | |
| app_logger.info(f'img: shape (numpy) {np_img.shape}.') | |
| except Exception as e_shape: | |
| app_logger.info(f'e_shape {e_shape}.') | |
| raise e_shape | |
| img.save(f"/tmp/downloaded_{pt0[0]}_{pt0[1]}_{pt1[0]}_{pt1[1]}.png") | |
| app_logger.info("saved PIL image") | |
| return img, matrix | |
| # app_logger.info("prepare writing...") | |
| # app_logger.info(f'img: type {type(img)}, len_matrix:{len(matrix)}, matrix {matrix}.') | |
| # | |
| # rio_output = str(Path(output_tmp_dir) / "downloaded_rio.tif") | |
| # app_logger.info(f'writing to disk img, output file {rio_output}.') | |
| # save_geotiff_gdal(img, rio_output, matrix) | |
| # app_logger.info(f'img written to output file {rio_output}.') | |
| # | |
| # source_tiles = os.path.join(input_tmp_dir, f"*.{ext}") | |
| # suffix_raster_filename = f"{lon_min},{lat_min},{lon_max},{lat_max}_{zoom}" | |
| # merged_raster_filename = f"merged_{suffix_raster_filename}.{ext}" | |
| # masked_raster_filename = f"masked_{suffix_raster_filename}.{ext}" | |
| # output_merged_path = os.path.join(output_tmp_dir, merged_raster_filename) | |
| # | |
| # app_logger.info(f"try merging tiles to:{output_merged_path}.") | |
| # merge_tiles(source_tiles, output_merged_path, input_tmp_dir) | |
| # app_logger.info(f"Merge complete, try crop...") | |
| # geojson = get_geojson_square_angles(bounding_box, name=suffix_raster_filename, debug=debug) | |
| # app_logger.info(f"geojson to convert:{geojson}.") | |
| # | |
| # crop_raster_output = crop_raster(output_merged_path, geojson, debug=False) | |
| # masked_raster = crop_raster_output["masked_raster"] | |
| # masked_meta = crop_raster_output["masked_meta"] | |
| # masked_transform = crop_raster_output["masked_transform"] | |
| # | |
| # return masked_raster, masked_transform | |
| # app_logger.info(f"resampling -32768 values as NaN for file:{masked_raster_filename}.") | |
| # masked_raster = masked_raster[0].astype(float) | |
| # masked_raster[masked_raster == NODATA_VALUES] = 0 | |
| # # info | |
| # nan_count = np.count_nonzero(~np.isnan(masked_raster)) | |
| # total_count = masked_raster.shape[-1] * masked_raster.shape[-2] | |
| # perc = get_perc(nan_count, total_count) | |
| # msg = f"img:{masked_raster_filename}, shape:{masked_raster.shape}: found {nan_count} not-NaN values / {total_count} total, %:{perc}." | |
| # app_logger.info(msg) | |
| # | |
| # app_logger.info(f"crop complete, shape:{masked_raster.shape}, dtype:{masked_raster.dtype}. Create RGB image...") | |
| # # rgb_filename, rgb_path = elaborate_images.get_rgb_prediction_image(masked_raster, slope_cellsize, suffix_raster_filename, output_tmp_dir, debug=debug) | |
| # # prediction = model_prediction(rgb_path, project_name=model_project_name, version=model_version, api_key=model_api_key, debug=False) | |
| # | |
| # mask_vectorizing = np.ones(masked_raster.shape).astype(rasterio.uint8) | |
| # app_logger.info(f"prediction success, try to geo-referencing it with transform:{masked_transform}.") | |
| # | |
| # app_logger.info( | |
| # f"image/geojson origin matrix:, masked_transform:{masked_transform}: create shapes_generator...") | |
| # app_logger.info(f"raster mask to vectorize, type:{type(mask_vectorizing)}.") | |
| # app_logger.info(f"raster mask to vectorize: shape:{mask_vectorizing.shape}, {mask_vectorizing.dtype}.") | |
| # | |
| # shapes_generator = ({ | |
| # 'properties': {'raster_val': v}, 'geometry': s} | |
| # for i, (s, v) | |
| # in enumerate(shapes(mask_vectorizing, mask=mask_vectorizing, transform=masked_transform)) | |
| # ) | |
| # shapes_list = list(shapes_generator) | |
| # app_logger.info(f"created {len(shapes_list)} polygons.") | |
| # gpd_polygonized_raster = GeoDataFrame.from_features(shapes_list, crs="EPSG:3857") | |
| # app_logger.info(f"created a GeoDataFrame: type {type(gpd_polygonized_raster)}.") | |
| # geojson = gpd_polygonized_raster.to_json(to_wgs84=True) | |
| # app_logger.info(f"created geojson: type {type(geojson)}, len:{len(geojson)}.") | |
| # serialized_geojson = serialize.serialize(geojson) | |
| # app_logger.info(f"created serialized_geojson: type {type(serialized_geojson)}, len:{len(serialized_geojson)}.") | |
| # loaded_geojson = json.loads(geojson) | |
| # app_logger.info(f"loaded_geojson: type {type(loaded_geojson)}, loaded_geojson:{loaded_geojson}.") | |
| # n_feats = len(loaded_geojson["features"]) | |
| # app_logger.info(f"created geojson: n_feats {n_feats}.") | |
| # | |
| # output_geojson = str(Path(ROOT) / "geojson_output.json") | |
| # with open(output_geojson, "w") as jj_out: | |
| # app_logger.info(f"writing geojson file to {output_geojson}.") | |
| # json.dump(loaded_geojson, jj_out) | |
| # app_logger.info(f"geojson file written to {output_geojson}.") | |
| # | |
| # # prediction_georef = helpers.get_prediction_georeferenced(prediction, masked_transform, skip_conditions_list, debug=debug) | |
| # app_logger.info(f"success on geo-referencing prediction.") | |
| # # app_logger.info(f"success on creating file {rgb_filename}, now try upload it to bucket_name {bucket_name}...") | |
| # return { | |
| # # "uploaded_file_name": rgb_filename, | |
| # "geojson": loaded_geojson, | |
| # # "prediction_georef": prediction_georef, | |
| # "n_total_obj_prediction": n_feats | |
| # } | |
| except ImportError as e_import_convert: | |
| app_logger.error(f"e0:{e_import_convert}.") | |
| raise e_import_convert | |
| if __name__ == '__main__': | |
| from PIL import Image | |
| npy_file = "prediction_masks_46.27697017893455_9.616470336914064_46.11441972281433_9.264907836914064.npy" | |
| prediction_masks = np.load(Path(PROJECT_ROOT_FOLDER) / "tmp" / "try_by_steps" / "t0" / npy_file) | |
| print("#") | |