diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 0ff7a8b..285c86f 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -7,7 +7,7 @@ ci: repos: - repo: https://github.com/pre-commit/pre-commit-hooks.git - rev: v5.0.0 + rev: v6.0.0 hooks: - id: check-json - id: check-yaml @@ -18,7 +18,7 @@ repos: - repo: https://github.com/astral-sh/ruff-pre-commit # Ruff version. - rev: v0.8.3 + rev: v0.13.3 hooks: # Run the linter. - id: ruff diff --git a/ci/test_lsi.py b/ci/test_lsi.py index 0223454..f644759 100644 --- a/ci/test_lsi.py +++ b/ci/test_lsi.py @@ -26,6 +26,7 @@ # def set_env(): # os.environ["NUMBA_DEBUG"] = "0" + def get_ci_path() -> AnyPathType: """ Get the path to the CI folder. diff --git a/lsi/lsi_core.py b/lsi/lsi_core.py index 0687392..6997f3b 100644 --- a/lsi/lsi_core.py +++ b/lsi/lsi_core.py @@ -193,12 +193,12 @@ def check_parameters(input_dict: dir) -> None: # Check if other_dem_path is needed if (dem_name == DemType.OTHER.value) and (other_dem_path is None): - raise ValueError(f'{"Dem path is needed !"}') + raise ValueError(f"{'Dem path is needed !'}") if (location == LocationType.GLOBAL.value) and ( landcover_name is LandcoverType.CLC.value ): raise ValueError( - f'{"Corine Land Cover can not be used for GLOBAL calculations, only in Europe !"}' + f"{'Corine Land Cover can not be used for GLOBAL calculations, only in Europe !'}" ) return @@ -224,7 +224,7 @@ def lsi_core(input_dict: dict, ftep) -> None: LOGGER.info(f"*** Tool version: {__version__} ***") # --- Check parameters --- # Catch FTEP-S3 errors (S3 access instability most likely related to networking issues, timeouts, or rate limits) - try: + try: check_parameters(input_dict) # --- Extract parameters --- @@ -289,7 +289,9 @@ def lsi_core(input_dict: dict, ftep) -> None: # Read GADM layer and overlay with AOI aoi_gadm = aoi.copy() aoi_gadm.geometry = aoi_gadm.geometry.buffer(GADM_BUFFER) - with warnings.catch_warnings(): # For cases of polygons with more than 100 parts + with ( + warnings.catch_warnings() + ): # For cases of polygons with more than 100 parts warnings.simplefilter("ignore") gadm = vectors.read(gadm_path, window=aoi_gadm) gadm = gadm.to_crs(proj_crs) @@ -393,7 +395,9 @@ def lsi_core(input_dict: dict, ftep) -> None: lsi_tif = rasters.crop(elsus_path, aoi) lsi_tif = lsi_tif.rio.write_crs(proj_crs) lsi_tif = lsi_tif.rio.reproject( - proj_crs, resolution=output_resolution, resampling=Resampling.bilinear + proj_crs, + resolution=output_resolution, + resampling=Resampling.bilinear, ) lsi_tif.rasters.crop(lsi_tif, aoi) lsi_tif.name = "Value" @@ -437,7 +441,9 @@ def lsi_core(input_dict: dict, ftep) -> None: physio_zones = physio_zones.to_crs(proj_crs) physio_zones_aoi = gpd.clip(physio_zones, aoi) - vectors.write(physio_zones_aoi, os.path.join(tmp_dir, "physio_zone_AOI.shp")) + vectors.write( + physio_zones_aoi, os.path.join(tmp_dir, "physio_zone_AOI.shp") + ) # Define a mapping of Zones to the Zone_Weights database file zone_to_dbf = { 0: "Zone0", @@ -503,7 +509,9 @@ def lsi_core(input_dict: dict, ftep) -> None: else: # the weights are set to 0 for Zone0 # A random file is chosen (in this case Zone1). which file i is, is not important because at the end # the weights are set to 0 for Zone0 - landcover_w_path = str(elsus_weights_path / "Zone1/Landcover_Z1.dbf") + landcover_w_path = str( + elsus_weights_path / "Zone1/Landcover_Z1.dbf" + ) # -- Compute the Rasters landcover_dir = landcover_raster_eu( @@ -523,7 +531,8 @@ def lsi_core(input_dict: dict, ftep) -> None: # ---- SLOPE CASE ---- # -- Define weights path slope_w_path = str( - elsus_weights_path / str(str(db_file) + "/Slope_Z" + str(zone) + ".dbf") + elsus_weights_path + / str(str(db_file) + "/Slope_Z" + str(zone) + ".dbf") ) # -- Compute the Rasters slope_dir = slope_raster_eu( @@ -561,7 +570,9 @@ def lsi_core(input_dict: dict, ftep) -> None: warnings.simplefilter("ignore") fw_dbf = gpd.read_file(global_final_weights_dbf_path) geology_weights = fw_dbf[fw_dbf.Factors == "Geology"].Weights.iloc[0] - lsi_tif = slope_tif + geology_tif * float(geology_weights) + landcover_tif + lsi_tif = ( + slope_tif + geology_tif * float(geology_weights) + landcover_tif + ) # lsi_tif = slope_tif + lithology_tif + landcover_tif elif location == LocationType.GLOBAL.value: @@ -602,7 +613,9 @@ def lsi_core(input_dict: dict, ftep) -> None: # Compute Geology layer with warnings.catch_warnings(): warnings.simplefilter("ignore") - geology_layer = geology_raster(lithology_path, dem, aoi, proj_crs, tmp_dir) + geology_layer = geology_raster( + lithology_path, dem, aoi, proj_crs, tmp_dir + ) # -- 2. Slope slope_layer = slope_raster(dem_b, aoi, tmp_dir) @@ -651,7 +664,9 @@ def lsi_core(input_dict: dict, ftep) -> None: geology_weights = fw_dbf[fw_dbf.Factors == "Geology"].Weights.iloc[0] aspect_weights = fw_dbf[fw_dbf.Factors == "Slope aspect"].Weights.iloc[0] elevation_weights = fw_dbf[fw_dbf.Factors == "Elevation"].Weights.iloc[0] - hydro_weights = fw_dbf[fw_dbf.Factors == "Distance from river"].Weights.iloc[0] + hydro_weights = fw_dbf[ + fw_dbf.Factors == "Distance from river" + ].Weights.iloc[0] landuse_weights = fw_dbf[fw_dbf.Factors == "Land use"].Weights.iloc[0] # Final weight @@ -660,7 +675,9 @@ def lsi_core(input_dict: dict, ftep) -> None: # Read all layers from temporal directory slope_layer = rasters.read(os.path.join(tmp_dir, "slope_weight.tif")) geology_layer = rasters.read(os.path.join(tmp_dir, "geology_weight.tif")) - elevation_layer = rasters.read(os.path.join(tmp_dir, "elevation_weight.tif")) + elevation_layer = rasters.read( + os.path.join(tmp_dir, "elevation_weight.tif") + ) aspect_layer = rasters.read(os.path.join(tmp_dir, "aspect_weight.tif")) landuse_layer = rasters.read(os.path.join(tmp_dir, "landcover_weight.tif")) hydro_layer = rasters.read(os.path.join(tmp_dir, "hydro_weight.tif")) @@ -670,12 +687,16 @@ def lsi_core(input_dict: dict, ftep) -> None: geology_layer = rasters.collocate( slope_layer, geology_layer, Resampling.bilinear ) - aspect_layer = rasters.collocate(slope_layer, aspect_layer, Resampling.bilinear) + aspect_layer = rasters.collocate( + slope_layer, aspect_layer, Resampling.bilinear + ) aspect_layer = rasters.crop(aspect_layer, aoi) landuse_layer = rasters.collocate( slope_layer, landuse_layer, Resampling.bilinear ) - hydro_layer = rasters.collocate(slope_layer, hydro_layer, Resampling.bilinear) + hydro_layer = rasters.collocate( + slope_layer, hydro_layer, Resampling.bilinear + ) # Calculate lsi lsi_tif = ( @@ -732,7 +753,9 @@ def lsi_core(input_dict: dict, ftep) -> None: vectors.write(lsi_vector, os.path.join(output_path, "LandslideRisk.shp")) # Write in memory - rasters.write(lsi_tif_sieved, os.path.join(output_path, "LandslideRisk.tif")) + rasters.write( + lsi_tif_sieved, os.path.join(output_path, "LandslideRisk.tif") + ) LOGGER.info("-- Computing LSI statistics (FER_LR_av)") @@ -742,12 +765,16 @@ def lsi_core(input_dict: dict, ftep) -> None: ) except RasterioIOError as e: - LOGGER.error("Could not open or read the raster. Check if your data is available at your path or try relaunching RUSLE now or later") + LOGGER.error( + "Could not open or read the raster. Check if your data is available at your path or try relaunching RUSLE now or later" + ) print(str(e)) sys.exit(1) except DataSourceError as e: - LOGGER.error("Could not open or read the vector file. Check if your data is available at your path or try relaunching RUSLE now or later") + LOGGER.error( + "Could not open or read the vector file. Check if your data is available at your path or try relaunching RUSLE now or later" + ) print(str(e)) sys.exit(1) diff --git a/lsi/src/lsi_calculator.py b/lsi/src/lsi_calculator.py index 16901b2..e3df4ed 100644 --- a/lsi/src/lsi_calculator.py +++ b/lsi/src/lsi_calculator.py @@ -315,7 +315,7 @@ def hydro_raster_wbw( # When computing in the FTEP we use pysheds for fill_depressions to avoid panicking issue # identified here: gitlab.unistra.fr/sertit/arcgis-pro/lsi/-/issues/2 - if ftep: + if ftep: LOGGER.info("-- -- Preparing the DEM: Filling Pits") # -- Compute D8 flow directions grid = Grid.from_raster(dem_b_path) @@ -342,7 +342,9 @@ def hydro_raster_wbw( else: LOGGER.info("-- -- Preparing the DEM: Filling Pits") # -- Fill pits - filled_pits = wbe.fill_pits(wbe.read_raster(os.path.join(tmp_dir, "dem_d.tif"))) + filled_pits = wbe.fill_pits( + wbe.read_raster(os.path.join(tmp_dir, "dem_d.tif")) + ) LOGGER.info("-- -- Preparing the DEM: Filling Depressions") # Write and read using WbW @@ -396,8 +398,9 @@ def hydro_raster_wbw( # No value_field defined as it is already in binary flowacc_thresh_lines = rasters.rasterize( - dem_b, flowacc_thresh_lines, - ) + dem_b, + flowacc_thresh_lines, + ) rasters.write( flowacc_thresh_lines,