Simplify typing

Author: alexamici

xarray_sentinel/sentinel1.py

@@ -14,7 +14,7 @@
 
 import os
 import warnings
-from typing import Any, Dict, List, Optional, Sequence, Tuple, TypeVar
+from typing import Any, Optional, Sequence, TypeVar
 
 import fsspec
 import numpy as np
@@ -35,8 +35,8 @@
 def get_fs_path(
     urlpath_or_path: esa_safe.PathType,
     fs: Optional[fsspec.AbstractFileSystem] = None,
-    storage_options: Optional[Dict[str, Any]] = None,
-) -> Tuple[fsspec.AbstractFileSystem, str]:
+    storage_options: Optional[dict[str, Any]] = None,
+) -> tuple[fsspec.AbstractFileSystem, str]:
     if fs is not None and storage_options is not None:
         raise TypeError("only one of 'fs' and 'storage_options' can be not None")
 
@@ -58,7 +58,7 @@ def get_fs_path(
     return fs, path
 
 
-def normalise_group(group: Optional[str]) -> Tuple[str, Optional[int]]:
+def normalise_group(group: Optional[str]) -> tuple[str, Optional[int]]:
     if group is None:
         group = ""
     if group.startswith("/"):
@@ -72,7 +72,7 @@ def normalise_group(group: Optional[str]) -> Tuple[str, Optional[int]]:
 
 
 def open_calibration_dataset(
-    calibration: esa_safe.PathType, attrs: Dict[str, Any] = {}
+    calibration: esa_safe.PathType, attrs: dict[str, Any] = {}
 ) -> xr.Dataset:
     calibration_vectors = esa_safe.parse_tag_as_list(
         calibration, ".//calibrationVector", "calibration"
@@ -122,7 +122,7 @@ def open_calibration_dataset(
 
 
 def open_reference_replica_dataset(
-    annotation_path: esa_safe.PathType, attrs: Dict[str, Any] = {}
+    annotation_path: esa_safe.PathType, attrs: dict[str, Any] = {}
 ) -> xr.Dataset:
     reference_replica = esa_safe.parse_tag_as_list(
         annotation_path, ".//replicaInformationList/replicaInformation/referenceReplica"
@@ -146,7 +146,7 @@ def open_reference_replica_dataset(
         float(v) for v in reference_replica["phaseCoefficients"]["$"].split()
     ]
 
-    coords: Dict[str, Any] = {
+    coords: dict[str, Any] = {
         "degree": range(len(reference_replica_amplitude_coefficients))
     }
     data_vars = {
@@ -158,7 +158,7 @@ def open_reference_replica_dataset(
 
 
 def open_antenna_pattern(
-    annotation_path: esa_safe.PathType, attrs: Dict[str, Any] = {}
+    annotation_path: esa_safe.PathType, attrs: dict[str, Any] = {}
 ) -> xr.Dataset:
     antenna_pattern_list = esa_safe.parse_tag_as_list(
         annotation_path, ".//antennaPattern/antennaPatternList/antennaPattern"
@@ -227,7 +227,7 @@ def open_antenna_pattern(
 
 
 def open_replica_dataset(
-    annotation_path: esa_safe.PathType, attrs: Dict[str, Any] = {}
+    annotation_path: esa_safe.PathType, attrs: dict[str, Any] = {}
 ) -> xr.Dataset:
     replicaList = esa_safe.parse_tag_as_list(
         annotation_path,
@@ -268,7 +268,7 @@ def open_replica_dataset(
         )
         internal_time_delay_list.append(replica["internalTimeDelay"])
 
-    coords: Dict[str, Any] = {
+    coords: dict[str, Any] = {
         "azimuth_time": [np.datetime64(dt, "ns") for dt in azimuth_time_list],
     }
     data_vars = {
@@ -305,7 +305,7 @@ def open_replica_dataset(
 
 
 def open_noise_range_dataset(
-    noise: esa_safe.PathType, attrs: Dict[str, Any] = {}
+    noise: esa_safe.PathType, attrs: dict[str, Any] = {}
 ) -> xr.Dataset:
     noise_vectors = esa_safe.parse_tag_as_list(noise, ".//noiseRangeVector", "noise")
 
@@ -338,7 +338,7 @@ def open_noise_range_dataset(
 
 
 def open_noise_azimuth_dataset(
-    noise: esa_safe.PathType, attrs: Dict[str, Any] = {}
+    noise: esa_safe.PathType, attrs: dict[str, Any] = {}
 ) -> xr.Dataset:
     noise_vectors = esa_safe.parse_tag_as_list(noise, ".//noiseAzimuthVector", "noise")
 
@@ -365,7 +365,7 @@ def open_noise_azimuth_dataset(
 
 
 def open_coordinate_conversion_dataset(
-    annotation_path: esa_safe.PathType, attrs: Dict[str, Any] = {}
+    annotation_path: esa_safe.PathType, attrs: dict[str, Any] = {}
 ) -> xr.Dataset:
     coordinate_conversion = esa_safe.parse_tag_as_list(
         annotation_path, ".//coordinateConversionList/coordinateConversion"
@@ -377,8 +377,8 @@ def open_coordinate_conversion_dataset(
     sr0 = []
     azimuth_time = []
     slant_range_time = []
-    srgrCoefficients: List[List[float]] = []
-    grsrCoefficients: List[List[float]] = []
+    srgrCoefficients: list[list[float]] = []
+    grsrCoefficients: list[list[float]] = []
     for values in coordinate_conversion:
         sr0.append(values["sr0"])
         gr0.append(values["gr0"])
@@ -391,8 +391,8 @@ def open_coordinate_conversion_dataset(
             [float(v) for v in values["grsrCoefficients"]["$"].split()]
         )
 
-    coords: Dict[str, Any] = {}
-    data_vars: Dict[str, Any] = {}
+    coords: dict[str, Any] = {}
+    data_vars: dict[str, Any] = {}
     coords["azimuth_time"] = [np.datetime64(dt, "ns") for dt in azimuth_time]
     coords["degree"] = list(range(len(srgrCoefficients[0])))
 
@@ -405,13 +405,13 @@ def open_coordinate_conversion_dataset(
     return xr.Dataset(data_vars=data_vars, coords=coords, attrs=attrs)
 
 
-def is_clockwise(poly: List[Tuple[float, float]]) -> bool:
+def is_clockwise(poly: list[tuple[float, float]]) -> bool:
     start = np.array(poly[0])
     return float(np.cross(poly[1] - start, poly[2] - start)) < 0
 
 
 def open_gcp_dataset(
-    annotation: esa_safe.PathOrFileType, attrs: Dict[str, Any] = {}
+    annotation: esa_safe.PathOrFileType, attrs: dict[str, Any] = {}
 ) -> xr.Dataset:
     geolocation_grid_points = esa_safe.parse_tag_as_list(
         annotation, ".//geolocationGridPoint"
@@ -468,8 +468,8 @@ def get_footprint_linestring(
     slant_range_time: xr.DataArray,
     gcp: xr.Dataset,
     method: xr.core.types.InterpOptions = "linear",
-    kwargs: Dict[str, Any] = {"fill_value": "extrapolate"},
-) -> List[Tuple[float, float]]:
+    kwargs: dict[str, Any] = {"fill_value": "extrapolate"},
+) -> list[tuple[float, float]]:
     azimuth_time_mm = [azimuth_time.min(), azimuth_time.max()]
     slant_range_time_mm = [slant_range_time.min(), slant_range_time.max()]
 
@@ -504,8 +504,8 @@ def get_footprint_linestring(
 
 
 def make_geospatial_attributes(
-    footprint: Sequence[Tuple[float, float]],
-) -> Dict[str, Any]:
+    footprint: Sequence[tuple[float, float]],
+) -> dict[str, Any]:
     wkt = "POLYGON((" + ",".join(f"{y} {x}" for y, x in footprint) + "))"
     geospatial_attrs = {
         "geospatial_bounds": wkt,
@@ -518,13 +518,13 @@ def make_geospatial_attributes(
 
 
 def open_attitude_dataset(
-    annotation: esa_safe.PathOrFileType, attrs: Dict[str, Any] = {}
+    annotation: esa_safe.PathOrFileType, attrs: dict[str, Any] = {}
 ) -> xr.Dataset:
     attitudes = esa_safe.parse_tag_as_list(annotation, ".//attitude")
 
     variables = ["q0", "q1", "q2", "q3", "wx", "wy", "wz", "pitch", "roll", "yaw"]
-    azimuth_time: List[Any] = []
-    data_vars: Dict[str, Any]
+    azimuth_time: list[Any] = []
+    data_vars: dict[str, Any]
     data_vars = {var: ("azimuth_time", [], attrs) for var in variables}
     for attitude in attitudes:
         azimuth_time.append(attitude["time"])
@@ -541,10 +541,10 @@ def open_attitude_dataset(
 
 
 def make_orbit(
-    azimuth_time: List[Any],
-    positions: List[List[Any]],
-    velocities: List[List[Any]],
-    attrs: Dict[str, Any] = {},
+    azimuth_time: list[Any],
+    positions: list[list[Any]],
+    velocities: list[list[Any]],
+    attrs: dict[str, Any] = {},
 ) -> xr.Dataset:
     position = xr.Variable(data=positions, dims=("axis", "azimuth_time"))
     velocity = xr.Variable(data=velocities, dims=("axis", "azimuth_time"))
@@ -564,7 +564,7 @@ def make_orbit(
 
 
 def open_orbit_dataset(
-    annotation: esa_safe.PathOrFileType, attrs: Dict[str, Any] = {}
+    annotation: esa_safe.PathOrFileType, attrs: dict[str, Any] = {}
 ) -> xr.Dataset:
     orbits = esa_safe.parse_tag_as_list(annotation, ".//orbit")
 
@@ -573,9 +573,9 @@ def open_orbit_dataset(
     if reference_system is not None:
         attrs.update({"reference_system": reference_system})
 
-    azimuth_times: List[Any] = []
-    positions: List[List[Any]] = [[], [], []]
-    velocities: List[List[Any]] = [[], [], []]
+    azimuth_times: list[Any] = []
+    positions: list[list[Any]] = [[], [], []]
+    velocities: list[list[Any]] = [[], [], []]
     for orbit in orbits:
         azimuth_times.append(orbit["time"])
         positions[0].append(orbit["position"]["x"])
@@ -594,7 +594,7 @@ def open_orbit_dataset(
 
 
 def open_dc_estimate_dataset(
-    annotation: esa_safe.PathOrFileType, attrs: Dict[str, Any] = {}
+    annotation: esa_safe.PathOrFileType, attrs: dict[str, Any] = {}
 ) -> xr.Dataset:
     dc_estimates = esa_safe.parse_tag_as_list(annotation, ".//dcEstimate")
 
@@ -656,7 +656,7 @@ def open_dc_estimate_dataset(
 
 
 def open_azimuth_fm_rate_dataset(
-    annotation: esa_safe.PathOrFileType, attrs: Dict[str, Any] = {}
+    annotation: esa_safe.PathOrFileType, attrs: dict[str, Any] = {}
 ) -> xr.Dataset:
     azimuth_fm_rates = esa_safe.parse_tag_as_list(annotation, ".//azimuthFmRate")
 
@@ -693,13 +693,13 @@ def open_azimuth_fm_rate_dataset(
 
 
 def find_available_groups(
-    product_files: Dict[str, Tuple[str, str, str, str, str]],
+    product_files: dict[str, tuple[str, str, str, str, str]],
     product_path: str,
     product_type: str,
     check_files_exist: bool = False,
     fs: fsspec.AbstractFileSystem = fsspec.filesystem("file"),
-) -> Dict[str, List[str]]:
-    groups: Dict[str, List[str]] = {}
+) -> dict[str, list[str]]:
+    groups: dict[str, list[str]] = {}
     for path, (type, _, swath, polarization, _) in product_files.items():
         swath_pol_group = f"{swath}/{polarization}".upper()
         abspath = os.path.join(product_path, os.path.normpath(path))
@@ -738,7 +738,7 @@ def find_available_groups(
 def open_rasterio_dataarray(
     measurement: esa_safe.PathOrFileType,
     fs: Optional[fsspec.AbstractFileSystem],
-    chunks: Optional[Dict[str, int]],
+    chunks: Optional[dict[str, int]],
 ) -> xr.DataArray:
     # fsspec needs rasterio >= 1.3.0, but we allow earlier rasterio versions for local files
     if fs is None or isinstance(fs, fsspec.implementations.local.LocalFileSystem):
@@ -776,7 +776,7 @@ def open_pol_dataset(
     measurement: esa_safe.PathOrFileType,
     annotation: esa_safe.PathOrFileType,
     fs: Optional[fsspec.AbstractFileSystem] = None,
-    attrs: Dict[str, Any] = {},
+    attrs: dict[str, Any] = {},
     gcp: Optional[xr.Dataset] = None,
 ) -> xr.Dataset:
     product_information = esa_safe.parse_tag(annotation, ".//productInformation")
@@ -811,7 +811,7 @@ def open_pol_dataset(
     )
     encoding = {}
     swap_dims = {}
-    chunks: Optional[Dict[str, int]] = None
+    chunks: Optional[dict[str, int]] = None
 
     azimuth_time = make_azimuth_time(
         product_first_line_utc_time,
@@ -1145,8 +1145,8 @@ def ground_range_to_slant_range_time(
 
 
 def do_override_product_files(
-    template: str, product_files: Dict[str, Tuple[str, str, str, str, str]]
-) -> Dict[str, Tuple[str, str, str, str, str]]:
+    template: str, product_files: dict[str, tuple[str, str, str, str, str]]
+) -> dict[str, tuple[str, str, str, str, str]]:
     overridden_product_files = {}
     for path, description in product_files.items():
         type, prefix, swath, polarization, date = description
@@ -1160,10 +1160,10 @@ def do_override_product_files(
 def open_sentinel1_dataset(
     product_urlpath: esa_safe.PathType,
     *,
-    drop_variables: Optional[Tuple[str]] = None,
+    drop_variables: Optional[tuple[str]] = None,
     group: Optional[str] = None,
     fs: Optional[fsspec.AbstractFileSystem] = None,
-    storage_options: Optional[Dict[str, Any]] = None,
+    storage_options: Optional[dict[str, Any]] = None,
     check_files_exist: bool = False,
     override_product_files: Optional[str] = None,
     parse_geospatial_attrs: bool = True,

xarray_sentinel/xarray_backends.py

@@ -1,5 +1,5 @@
 import os
-from typing import Any, Dict, Optional, Tuple
+from typing import Any, Optional
 
 import fsspec
 import xarray as xr
@@ -11,9 +11,9 @@ class Sentinel1Backend(xr.backends.common.BackendEntrypoint):
     def open_dataset(  # type: ignore
         self,
         filename_or_obj: str,
-        drop_variables: Optional[Tuple[str]] = None,
+        drop_variables: Optional[tuple[str]] = None,
         group: Optional[str] = None,
-        storage_options: Optional[Dict[str, Any]] = None,
+        storage_options: Optional[dict[str, Any]] = None,
         override_product_files: Optional[str] = None,
         fs: Optional[fsspec.AbstractFileSystem] = None,
         check_files_exist: bool = False,