update typing to 3.10

Author: Emma Ai

odc/stats/_algebra.py

@@ -32,7 +32,7 @@ def median_by_ind(xr_da, dim, dtype="float32", name="median_by_ind"):
         meta=np.array((), dtype=dtype),
         drop_axis=0,
     )
-    coords = dict((dim, xr_da.coords[dim]) for dim in xr_da.dims[1:])
+    coords = {dim: xr_da.coords[dim] for dim in xr_da.dims[1:]}
 
     return xr.DataArray(
         res, dims=xr_da.dims[1:], coords=coords, attrs=xr_da.attrs.copy()
@@ -44,5 +44,5 @@ def median_ds(xr_ds, dim, dtype="float32", name="median_ds"):
     for var, data in xr_ds.data_vars.items():
         res[var] = median_by_ind(data, dim, dtype, name)
     # pylint: disable=undefined-loop-variable
-    coords = dict((dim, xr_ds.coords[dim]) for dim in data.dims[1:])
+    coords = {dim: xr_ds.coords[dim] for dim in data.dims[1:]}
     return xr.Dataset(res, coords=coords, attrs=xr_ds.attrs.copy())

odc/stats/_cli_common.py

@@ -2,7 +2,6 @@
 
 import logging
 import sys
-from typing import List, Tuple
 import click
 
 
@@ -12,7 +11,7 @@
 from urllib.parse import urlparse
 
 
-TileIdx_txy = Tuple[str, int, int]  # pylint: disable=invalid-name
+TileIdx_txy = tuple[str, int, int]  # pylint: disable=invalid-name
 
 
 def parse_task(s: str) -> TileIdx_txy:
@@ -27,8 +26,8 @@ def parse_task(s: str) -> TileIdx_txy:
 
 
 def parse_all_tasks(
-    inputs: List[str], all_possible_tasks: List[TileIdx_txy]
-) -> List[TileIdx_txy]:
+    inputs: list[str], all_possible_tasks: list[TileIdx_txy]
+) -> list[TileIdx_txy]:
     """
     Select a subset of all possible tasks given user input on cli.
 
@@ -43,7 +42,7 @@ def parse_all_tasks(
        x+10/y-3/2019--P1Y
     """
 
-    out: List[TileIdx_txy] = []
+    out: list[TileIdx_txy] = []
     full_set = set(all_possible_tasks)
 
     for s in inputs:
@@ -68,7 +67,7 @@ def parse_all_tasks(
     return out
 
 
-def parse_resolution(s: str, separator: str = ",") -> Tuple[float, float]:
+def parse_resolution(s: str, separator: str = ",") -> tuple[float, float]:
     parts = [float(v) for v in split_and_check(s, separator, (1, 2))]
 
     if len(parts) == 1:

odc/stats/_cli_publish_tasks.py

@@ -1,6 +1,5 @@
 import json
 import sys
-from typing import List, Optional
 
 import click
 import fsspec
@@ -28,15 +27,15 @@ def get_geometry(geojson_file: str) -> Geometry:
     )
 
 
-def filter_tasks(tasks: List[TileIdx_txy], geometry: Geometry, grid_name: str):
+def filter_tasks(tasks: list[TileIdx_txy], geometry: Geometry, grid_name: str):
     for task in tasks:
         task_geometry = GRIDS[grid_name].tile_geobox((task[1], task[2])).extent
         if task_geometry.intersects(geometry):
             yield task
 
 
 def publish_tasks(
-    db: str, task_filter: str, geojson_filter: Optional[str], dryrun: bool, queue: str
+    db: str, task_filter: str, geojson_filter: str | None, dryrun: bool, queue: str
 ):
     reader = TaskReader(db)
     if len(task_filter) == 0:
@@ -67,7 +66,7 @@ def publish_tasks(
 
     # We assume the db files are always be the S3 uri. If they are not, there is no need to use SQS queue to process.
     messages = (
-        dict(Id=str(idx), MessageBody=json.dumps(render_sqs(tidx, db)))
+        {"Id": str(idx), "MessageBody": json.dumps(render_sqs(tidx, db))}
         for idx, tidx in enumerate(tasks)
     )
 

odc/stats/_gjson.py

@@ -1,7 +1,7 @@
 import math
 from copy import deepcopy
 import toolz
-from typing import Tuple, Dict, Any
+from typing import Any
 from datetime import timedelta
 
 from odc.geo.gridspec import GridSpec
@@ -11,7 +11,7 @@
 from .model import TileIdx_xy, TileIdx_txy
 
 
-def gs_bounds(gs: GridSpec, tiles: Tuple[Tuple[int, int], Tuple[int, int]]) -> Geometry:
+def gs_bounds(gs: GridSpec, tiles: tuple[tuple[int, int], tuple[int, int]]) -> Geometry:
     """
     Compute Polygon for a selection of tiles.
 
@@ -36,8 +36,8 @@ def timedelta_to_hours(td: timedelta) -> float:
 
 
 def compute_grid_info(
-    cells: Dict[TileIdx_xy, Any], resolution: float = math.inf, title_width: int = 0
-) -> Dict[TileIdx_xy, Any]:
+    cells: dict[TileIdx_xy, Any], resolution: float = math.inf, title_width: int = 0
+) -> dict[TileIdx_xy, Any]:
     """
     Compute geojson feature for every cell in ``cells``.
     Where ``cells`` is produced by ``bin_dataset_stream``
@@ -75,8 +75,8 @@ def compute_grid_info(
 
 
 def gjson_from_tasks(
-    tasks: Dict[TileIdx_txy, Any], grid_info: Dict[TileIdx_xy, Any]
-) -> Dict[str, Dict[str, Any]]:
+    tasks: dict[TileIdx_txy, Any], grid_info: dict[TileIdx_xy, Any]
+) -> dict[str, dict[str, Any]]:
     """
     Group tasks by time period and compute geosjon describing every tile covered by each time period.
 
@@ -96,14 +96,14 @@ def _get(idx):
         dss = tasks[idx]
         utc_offset = timedelta(hours=geo["properties"]["utc_offset"])
 
-        ndays = len(set((ds.time + utc_offset).date() for ds in dss))
+        ndays = len({(ds.time + utc_offset).date() for ds in dss})
         geo["properties"]["total"] = len(dss)
         geo["properties"]["days"] = ndays
 
         return geo
 
     def process(idxs):
-        return dict(type="FeatureCollection", features=[_get(idx) for idx in idxs])
+        return {"type": "FeatureCollection", "features": [_get(idx) for idx in idxs]}
 
     return {
         t: process(idxs)

odc/stats/_grouper.py

@@ -7,7 +7,7 @@
 from __future__ import annotations
 
 from datetime import timedelta
-from typing import TYPE_CHECKING, Any
+from typing import Any, TYPE_CHECKING
 
 import numpy as np
 import pandas as pd

odc/stats/_sqs.py

@@ -2,14 +2,13 @@
 Work token for SQS based job control
 """
 
-from typing import Optional
 from datetime import timedelta, datetime
 import toolz
 from .model import WorkTokenInterface
 
 
 class SQSWorkToken(WorkTokenInterface):
-    def __init__(self, msg, timeout: int, t0: Optional[datetime] = None):
+    def __init__(self, msg, timeout: int, t0: datetime | None = None):
         super().__init__()
         if t0 is None:
             t0 = self.now()

odc/stats/_text.py

@@ -1,28 +1,28 @@
 from pathlib import Path
-from typing import Union, Optional, Tuple, Dict, Any
+from typing import Any
 
-PathLike = Union[str, Path]
+PathLike = str | Path
 
 
 # Copied from odc.io.text
 
 
-def read_int(path: PathLike, default=None, base=10) -> Optional[int]:
+def read_int(path: PathLike, default=None, base=10) -> int | None:
     """
     Read single integer from a text file.
 
     Useful for things like parsing content of /sys/ or /proc.
     """
     try:
-        with open(path, "rt", encoding="utf8") as f:
+        with open(path, encoding="utf8") as f:
             return int(f.read(), base)
     except (FileNotFoundError, ValueError):
         return default
 
 
 def split_and_check(
-    s: str, separator: str, n: Union[int, Tuple[int, ...]]
-) -> Tuple[str, ...]:
+    s: str, separator: str, n: int | tuple[int, ...]
+) -> tuple[str, ...]:
     """Turn string into tuple, checking that there are exactly as many parts as expected.
     :param s: String to parse
     :param separator: Separator character
@@ -44,7 +44,7 @@ def parse_slice(s: str) -> slice:
     Examples "::4", "2:5", "2::10", "3:100:5"
     """
 
-    def parse(part: str) -> Optional[int]:
+    def parse(part: str) -> int | None:
         if part == "":
             return None
         return int(part)
@@ -57,32 +57,32 @@ def parse(part: str) -> Optional[int]:
     return slice(*parts)
 
 
-def parse_yaml(s: str) -> Dict[str, Any]:
+def parse_yaml(s: str) -> dict[str, Any]:
     # pylint: disable=import-outside-toplevel
     import yaml
 
     return yaml.load(s, Loader=getattr(yaml, "CSafeLoader", yaml.SafeLoader))
 
 
-def parse_yaml_file_or_inline(s: str) -> Dict[str, Any]:
+def parse_yaml_file_or_inline(s: str) -> dict[str, Any]:
     """
     Accept on input either a path to yaml file or yaml text, return parsed yaml document.
     """
     try:
         # if file
         path = Path(s)
-        with open(path, "rt", encoding="utf8") as f:
+        with open(path, encoding="utf8") as f:
             txt = f.read()
             assert isinstance(txt, str)
-    except (FileNotFoundError, IOError, ValueError):
+    except (FileNotFoundError, OSError, ValueError):
         txt = s
     result = parse_yaml(txt)
     if isinstance(result, str):
-        raise IOError(f"No such file: {s}")
+        raise OSError(f"No such file: {s}")
     return result
 
 
-def load_yaml_remote(yaml_url: str) -> Dict[str, Any]:
+def load_yaml_remote(yaml_url: str) -> dict[str, Any]:
     """
     Open a yaml file remotely and return the parsed yaml document
     """
@@ -97,7 +97,7 @@ def load_yaml_remote(yaml_url: str) -> Dict[str, Any]:
         raise
 
 
-def parse_range2d_int(s: str) -> Tuple[Tuple[int, int], Tuple[int, int]]:
+def parse_range2d_int(s: str) -> tuple[tuple[int, int], tuple[int, int]]:
     """Parse string like "0:3,4:5" -> ((0,3), (4,5))"""
     try:
         return tuple(