update ruff config

pyproject.toml

@@ -76,3 +76,30 @@ where = ["src"]
 markers = [
     "integration: long-running integration tests",
 ]
+
+[tool.ruff]
+line-length = 120
+src = ["src", "tests"]
+exclude = ["prototype"]
+
+[tool.ruff.format]
+indent-style = "space"
+quote-style = "single"
+
+[tool.ruff.lint]
+extend-select = [
+    "I",   # isort: https://docs.astral.sh/ruff/rules/#isort-i
+    "UP",  # pyupgrade: https://docs.astral.sh/ruff/rules/#pyupgrade-up
+
+    # TODO: uncomment the following extensions and address their warnings:
+    #"D",   # pydocstyle: https://docs.astral.sh/ruff/rules/#pydocstyle-d
+    #"ANN", # annotations: https://docs.astral.sh/ruff/rules/#flake8-annotations-ann
+    #"PTH", # use-pathlib-pth: https://docs.astral.sh/ruff/rules/#flake8-use-pathlib-pth
+]
+
+[tool.ruff.lint.pydocstyle]
+convention = "google"
+
+[tool.ruff.lint.isort]
+case-sensitive = true
+lines-after-imports = 2

src/asf_tools/__init__.py

@@ -2,6 +2,7 @@
 
 from importlib.metadata import version
 
+
 __version__ = version(__name__)
 
 __all__ = [

src/asf_tools/aws.py

@@ -1,10 +1,10 @@
 import logging
 from mimetypes import guess_type
 from pathlib import Path
-from typing import Union
 
 import boto3
 
+
 S3_CLIENT = boto3.client('s3')
 log = logging.getLogger(__name__)
 
@@ -14,14 +14,14 @@ def get_tag_set() -> dict:
     return tag_set
 
 
-def get_content_type(file_location: Union[Path, str]) -> str:
+def get_content_type(file_location: Path | str) -> str:
     content_type = guess_type(file_location)[0]
     if not content_type:
         content_type = 'application/octet-stream'
     return content_type
 
 
-def upload_file_to_s3(path_to_file: Union[str, Path], bucket: str, prefix: str = ''):
+def upload_file_to_s3(path_to_file: str | Path, bucket: str, prefix: str = ''):
     path_to_file = Path(path_to_file)
     key = str(Path(prefix) / path_to_file.name)
     extra_args = {'ContentType': get_content_type(key)}

src/asf_tools/composite.py

@@ -17,19 +17,19 @@
 import sys
 from statistics import multimode
 from tempfile import TemporaryDirectory
-from typing import List
 
 import numpy as np
 from osgeo import gdal
 
 from asf_tools.raster import read_as_array, write_cog
 from asf_tools.util import get_epsg_code
 
+
 gdal.UseExceptions()
 log = logging.getLogger(__name__)
 
 
-def get_target_epsg_code(codes: List[int]) -> int:
+def get_target_epsg_code(codes: list[int]) -> int:
     """Determine the target UTM EPSG projection for the output composite
 
     Args:
@@ -151,7 +151,7 @@ def reproject_to_target(raster_info: dict, target_epsg_code: int, target_resolut
     return target_raster_info
 
 
-def make_composite(out_name: str, rasters: List[str], resolution: float = None):
+def make_composite(out_name: str, rasters: list[str], resolution: float = None):
     """Creates a local-resolution-weighted composite from Sentinel-1 RTC products
 
     Args:

src/asf_tools/dem.py

@@ -1,21 +1,21 @@
 """Prepare a Copernicus GLO-30 DEM virtual raster (VRT) covering a given geometry"""
 
 from pathlib import Path
-from typing import Union
 
 from osgeo import gdal, ogr
 from shapely.geometry.base import BaseGeometry
 
 from asf_tools import vector
 from asf_tools.util import GDALConfigManager
 
+
 DEM_GEOJSON = '/vsicurl/https://asf-dem-west.s3.amazonaws.com/v2/cop30-2021.geojson'
 
 gdal.UseExceptions()
 ogr.UseExceptions()
 
 
-def prepare_dem_vrt(vrt: Union[str, Path], geometry: Union[ogr.Geometry, BaseGeometry]):
+def prepare_dem_vrt(vrt: str | Path, geometry: ogr.Geometry | BaseGeometry):
     """Create a DEM mosaic VRT covering a given geometry
 
     The DEM mosaic is assembled from the Copernicus GLO-30 DEM tiles that intersect the geometry.

src/asf_tools/hydrosar/__init__.py

@@ -1,5 +1,6 @@
 from warnings import warn
 
+
 HYDROSAR_MOVE_WARNING = """
     ---------------------------------------------------------------------------
     The HydroSAR codes (`flood_map`, `water_map` and `hand` modules) are being

src/asf_tools/hydrosar/flood_map.py

@@ -13,9 +13,10 @@
 import sys
 import tempfile
 import warnings
+from collections.abc import Callable
 from pathlib import Path
 from shutil import make_archive
-from typing import Callable, Literal, Optional, Tuple, Union
+from typing import Literal
 
 import numpy as np
 from osgeo import gdal
@@ -26,6 +27,7 @@
 from asf_tools.raster import read_as_masked_array, write_cog
 from asf_tools.util import get_coordinates, get_epsg_code
 
+
 log = logging.getLogger(__name__)
 
 
@@ -36,7 +38,7 @@ def get_pw_threshold(water_array: np.array) -> float:
     return round(ths_orig) + 1
 
 
-def get_waterbody(input_info: dict, threshold: Optional[float] = None) -> np.array:
+def get_waterbody(input_info: dict, threshold: float | None = None) -> np.array:
     epsg = get_epsg_code(input_info)
 
     west, south, east, north = get_coordinates(input_info)
@@ -86,7 +88,7 @@ def _goal_fmi(w):
         tp, _, fp, fn = get_confusion_matrix(w)
         return 1 - np.sqrt((tp / (tp + fp)) * (tp / (tp + fn)))
 
-    class MyBounds(object):
+    class MyBounds:
         def __init__(self, xmax=max(water_levels), xmin=min(water_levels)):
             self.xmax = np.array(xmax)
             self.xmin = np.array(xmin)
@@ -119,7 +121,7 @@ def __call__(self, **kwargs):
         return np.inf  # set as inf to mark unstable solution
 
 
-def logstat(data: np.ndarray, func: Callable = np.nanstd) -> Union[np.ndarray, float]:
+def logstat(data: np.ndarray, func: Callable = np.nanstd) -> np.ndarray | float:
     """Calculate a function in logarithmic scale and return in linear scale.
     INF values inside the data array are set to nan.
 
@@ -141,7 +143,7 @@ def estimate_flood_depth(
     flood_labels: np.ndarray,
     estimator: str = 'iterative',
     water_level_sigma: float = 3.0,
-    iterative_bounds: Tuple[int, int] = (0, 15),
+    iterative_bounds: tuple[int, int] = (0, 15),
     iterative_min_size: int = 0,
     minimization_metric: str = 'ts',
 ) -> float:
@@ -179,14 +181,14 @@ def estimate_flood_depth(
 
 
 def make_flood_map(
-    out_raster: Union[str, Path],
-    vv_raster: Union[str, Path],
-    water_raster: Union[str, Path],
-    hand_raster: Union[str, Path],
+    out_raster: str | Path,
+    vv_raster: str | Path,
+    water_raster: str | Path,
+    hand_raster: str | Path,
     estimator: str = 'iterative',
     water_level_sigma: float = 3.0,
-    known_water_threshold: Optional[float] = None,
-    iterative_bounds: Tuple[int, int] = (0, 15),
+    known_water_threshold: float | None = None,
+    iterative_bounds: tuple[int, int] = (0, 15),
     iterative_min_size: int = 0,
     minimization_metric: str = 'ts',
 ):
@@ -233,7 +235,6 @@ def make_flood_map(
     References:
         Jean-Francios Pekel, Andrew Cottam, Noel Gorelik, Alan S. Belward. 2016. <https://doi:10.1038/nature20584>
     """
-
     info = gdal.Info(str(water_raster), format='json')
     epsg = get_epsg_code(info)
     geotransform = info['geoTransform']
@@ -328,13 +329,13 @@ def make_flood_map(
     )
 
 
-def optional_str(value: str) -> Optional[str]:
+def optional_str(value: str) -> str | None:
     if value.lower() == 'none':
         return None
     return value
 
 
-def optional_float(value: str) -> Optional[float]:
+def optional_float(value: str) -> float | None:
     if value.lower() == 'none':
         return None
     return float(value)

src/asf_tools/hydrosar/hand/__init__.py

@@ -4,6 +4,7 @@
 )
 from asf_tools.hydrosar.hand.prepare import prepare_hand_vrt
 
+
 __all__ = [
     'calculate_hand_for_basins',
     'make_copernicus_hand',

src/asf_tools/hydrosar/hand/calculate.py

@@ -6,7 +6,6 @@
 import warnings
 from pathlib import Path
 from tempfile import NamedTemporaryFile
-from typing import Optional, Union
 
 import astropy.convolution
 import fiona
@@ -19,6 +18,7 @@
 from asf_tools.dem import prepare_dem_vrt
 from asf_tools.raster import write_cog
 
+
 log = logging.getLogger(__name__)
 
 
@@ -59,7 +59,7 @@ def calculate_hand(
     dem_affine: rasterio.Affine,
     dem_crs: rasterio.crs.CRS,
     basin_mask,
-    acc_thresh: Optional[int] = 100,
+    acc_thresh: int | None = 100,
 ):
     """Calculate the Height Above Nearest Drainage (HAND)
 
@@ -143,10 +143,10 @@ def calculate_hand(
 
 
 def calculate_hand_for_basins(
-    out_raster: Union[str, Path],
+    out_raster: str | Path,
     geometries: GeometryCollection,
-    dem_file: Union[str, Path],
-    acc_thresh: Optional[int] = 100,
+    dem_file: str | Path,
+    acc_thresh: int | None = 100,
 ):
     """Calculate the Height Above Nearest Drainage (HAND) for watershed boundaries (hydrobasins).
 
@@ -177,9 +177,9 @@ def calculate_hand_for_basins(
 
 
 def make_copernicus_hand(
-    out_raster: Union[str, Path],
-    vector_file: Union[str, Path],
-    acc_thresh: Optional[int] = 100,
+    out_raster: str | Path,
+    vector_file: str | Path,
+    acc_thresh: int | None = 100,
 ):
     """Copernicus GLO-30 Height Above Nearest Drainage (HAND)
 

src/asf_tools/hydrosar/hand/prepare.py

@@ -2,7 +2,6 @@
 
 from pathlib import Path
 from tempfile import NamedTemporaryFile
-from typing import Union
 
 from osgeo import gdal, ogr
 from rasterio.enums import Resampling
@@ -12,13 +11,14 @@
 from asf_tools import vector
 from asf_tools.util import GDALConfigManager, get_epsg_code
 
+
 HAND_GEOJSON = '/vsicurl/https://glo-30-hand.s3.amazonaws.com/v1/2021/glo-30-hand.geojson'
 
 gdal.UseExceptions()
 ogr.UseExceptions()
 
 
-def prepare_hand_vrt(vrt: Union[str, Path], geometry: Union[ogr.Geometry, BaseGeometry]):
+def prepare_hand_vrt(vrt: str | Path, geometry: ogr.Geometry | BaseGeometry):
     """Prepare a HAND mosaic VRT covering a given geometry
 
     Prepare a Height Above Nearest Drainage (HAND) virtual raster (VRT) covering a given geometry.
@@ -50,8 +50,8 @@ def prepare_hand_vrt(vrt: Union[str, Path], geometry: Union[ogr.Geometry, BaseGe
 
 
 def prepare_hand_for_raster(
-    hand_raster: Union[str, Path],
-    source_raster: Union[str, Path],
+    hand_raster: str | Path,
+    source_raster: str | Path,
     resampling_method: str = 'lanczos',
 ):
     """Create a HAND raster pixel-aligned to a source raster

src/asf_tools/hydrosar/threshold.py

@@ -54,7 +54,6 @@ def expectation_maximization_threshold(tile: np.ndarray, number_of_classes: int
     Returns:
         threshold: threshold value that can be used to create a water extent map
     """
-
     image_copy = tile.copy()
     image_copy2 = np.ma.filled(tile.astype(float), np.nan)  # needed for valid posterior_lookup keys
     image = tile.flatten()