Add some more type signatures

Author: pjonsson

odc/geo/_rgba.py

@@ -12,7 +12,7 @@
 # pylint: disable=import-outside-toplevel
 
 
-def is_rgb(x: xr.DataArray):
+def is_rgb(x: xr.DataArray) -> bool:
     """
     Check if array is RGB(A).
     """

odc/geo/_xr_interop.py

@@ -112,7 +112,7 @@ def _get_crs_from_attrs(obj: XarrayObject, sdims: Tuple[str, str]) -> Optional[C
     """
     crs_set: Set[CRS] = set()
 
-    def _add_candidate(crs):
+    def _add_candidate(crs) -> None:
         if crs is None:
             return
         if isinstance(crs, str):
@@ -127,11 +127,11 @@ def _add_candidate(crs):
         else:
             warnings.warn(f"Ignoring crs attribute of type: {type(crs)}")
 
-    def process_attrs(attrs):
+    def process_attrs(attrs) -> None:
         _add_candidate(attrs.get("crs", None))
         _add_candidate(attrs.get("crs_wkt", None))
 
-    def process_datavar(x):
+    def process_datavar(x) -> None:
         process_attrs(x.attrs)
         for dim in sdims:
             if dim in x.coords:
@@ -910,7 +910,7 @@ class ODCExtension:
     Common accessors for both Array/Dataset.
     """
 
-    def __init__(self, state: GeoState):
+    def __init__(self, state: GeoState) -> None:
         self._state = state
 
     @property
@@ -987,7 +987,7 @@ class ODCExtensionDa(ODCExtension):
     ODC extension for :py:class:`xarray.DataArray`.
     """
 
-    def __init__(self, xx: xarray.DataArray):
+    def __init__(self, xx: xarray.DataArray) -> None:
         ODCExtension.__init__(self, _locate_geo_info(xx))
         self._xx = xx
 
@@ -1042,7 +1042,7 @@ def nodata(self) -> Nodata:
         return None
 
     @nodata.setter
-    def nodata(self, value: Nodata):
+    def nodata(self, value: Nodata) -> None:
         nodata = resolve_nodata(value, self._xx.dtype)
 
         if nodata is None:
@@ -1069,7 +1069,7 @@ class ODCExtensionDs(ODCExtension):
     ODC extension for :py:class:`xarray.Dataset`.
     """
 
-    def __init__(self, ds: xarray.Dataset):
+    def __init__(self, ds: xarray.Dataset) -> None:
         ODCExtension.__init__(self, _locate_geo_info(ds))
         self._xx = ds
 
@@ -1110,7 +1110,7 @@ def _xarray_geobox(xx: XarrayObject) -> Optional[GeoBox]:
     return None
 
 
-def register_geobox():
+def register_geobox() -> None:
     """
     Backwards compatiblity layer for datacube ``.geobox`` property.
     """

odc/geo/cog/_az.py

@@ -43,7 +43,7 @@ def __init__(
         blob: str,
         credential: Any = None,
         client: Any = None,
-    ):
+    ) -> None:
         """
         Initialise Azure multipart upload.
 
@@ -102,7 +102,7 @@ def finalise(self, parts: list[dict[str, Any]]) -> str:
         self.blob_client.commit_block_list(block_list)
         return self.blob_client.get_blob_properties().etag
 
-    def cancel(self, other: str = ""):
+    def cancel(self, other: str = "") -> None:
         """
         Cancel the upload by clearing the block list.
         """
@@ -143,7 +143,7 @@ class DelayedAzureWriter(AzureLimits):
     Dask-compatible writer for Azure Blob Storage multipart uploads.
     """
 
-    def __init__(self, mpu: AzMultiPartUpload, kw: dict[str, Any]):
+    def __init__(self, mpu: AzMultiPartUpload, kw: dict[str, Any]) -> None:
         """
         Initialise the Azure writer.
 

odc/geo/cog/_mpu.py

@@ -111,7 +111,7 @@ def __repr__(self) -> str:
             s = f"{s} final"
         return s
 
-    def append(self, data: SomeData, chunk_id: Any = None):
+    def append(self, data: SomeData, chunk_id: Any = None) -> None:
         sz = len(data)
         self.observed.append((sz, chunk_id))
         self.data += data

odc/geo/cog/_rio.py

@@ -194,7 +194,7 @@ def _write_cog(
 
     rio_opts.update(extra_rio_opts)
 
-    def _write(pix, band, dst):
+    def _write(pix, band, dst) -> None:
         if not use_windowed_writes:
             dst.write(pix, band)
         else:

odc/geo/cog/_s3.py

@@ -84,7 +84,7 @@ def __init__(
         profile: Optional[str] = None,
         endpoint_url: Optional[str] = None,
         creds: Optional["ReadOnlyCredentials"] = None,
-    ):
+    ) -> None:
         self.bucket = bucket
         self.key = key
         self.uploadId = uploadId
@@ -153,7 +153,7 @@ def started(self) -> bool:
         """Check if the multipart upload has been initiated."""
         return len(self.uploadId) > 0
 
-    def cancel(self, other: str = ""):
+    def cancel(self, other: str = "") -> None:
         """Cancel the multipart upload."""
         uploadId = other if other else self.uploadId
         if not uploadId:
@@ -216,7 +216,7 @@ class DelayedS3Writer(S3Limits):
 
     # pylint: disable=import-outside-toplevel,import-error
 
-    def __init__(self, mpu: S3MultiPartUpload, kw: dict[str, Any]):
+    def __init__(self, mpu: S3MultiPartUpload, kw: dict[str, Any]) -> None:
         self.mpu = mpu
         self.kw = kw  # mostly ContentType= kinda thing
         self._shared_var: Optional["distributed.Variable"] = None

odc/geo/crs.py

@@ -95,7 +95,7 @@ class CRS:
 
     __slots__ = ("_crs", "_epsg", "_str")
 
-    def __init__(self, crs_spec: Any):
+    def __init__(self, crs_spec: Any) -> None:
         """
         Construct CRS object from *something*.
 
@@ -122,8 +122,8 @@ def __init__(self, crs_spec: Any):
     def __getstate__(self):
         return {"crs_str": self._str}
 
-    def __setstate__(self, state):
-        self.__init__(state["crs_str"])
+    def __setstate__(self, state) -> None:
+        self.__init__(state["crs_str"])  # type: ignore[misc]
 
     def to_wkt(self, pretty: bool = False, version: Optional[WktVersion] = None) -> str:
         """

odc/geo/data/__init__.py

@@ -54,7 +54,7 @@ class _CachedGeoDataFrame:
     _lock = threading.Lock()
     _data_url = ""
 
-    def __init__(self):
+    def __init__(self) -> None:
         # Thread safe class-cached dataload
         if self._instance is None:
             with self._lock:

odc/geo/gcp.py

@@ -43,7 +43,7 @@ def __init__(
         pix: SomePointSet,
         wld: SomePointSet,
         crs: MaybeCRS = None,
-    ):
+    ) -> None:
         pix, _ = _points_to_array(pix)
         wld, _crs = _points_to_array(wld)
 
@@ -143,7 +143,7 @@ class GCPGeoBox(GeoBoxBase):
 
     def __init__(
         self, shape: SomeShape, mapping: GCPMapping, affine: Optional[Affine] = None
-    ):
+    ) -> None:
         if affine is None:
             affine = Affine.identity()
         GeoBoxBase.__init__(self, shape, affine, mapping.crs)
@@ -182,7 +182,7 @@ def linear(self) -> bool:
         return False
 
     @property
-    def axis_aligned(self):
+    def axis_aligned(self) -> bool:
         return False
 
     @property
@@ -279,7 +279,7 @@ def zoom_to(
         _shape, _affine = self.compute_zoom_to(shape, resolution=resolution)
         return GCPGeoBox(_shape, self._mapping, _affine)
 
-    def __str__(self):
+    def __str__(self) -> str:
         return self.__repr__()
 
     def __repr__(self) -> str:

odc/geo/geobox.py

@@ -116,7 +116,7 @@ class GeoBoxBase:
 
     __slots__ = ("_shape", "_affine", "_crs", "_extent", "_lazy_ui")
 
-    def __init__(self, shape: SomeShape, affine: Affine, crs: MaybeCRS):
+    def __init__(self, shape: SomeShape, affine: Affine, crs: MaybeCRS) -> None:
         shape = shape_(shape)
 
         self._shape = shape
@@ -511,7 +511,7 @@ class GeoBox(GeoBoxBase):
 
     __slots__ = ()
 
-    def __init__(self, shape: SomeShape, affine: Affine, crs: MaybeCRS):
+    def __init__(self, shape: SomeShape, affine: Affine, crs: MaybeCRS) -> None:
         GeoBoxBase.__init__(self, shape, affine, crs)
 
     @staticmethod
@@ -891,10 +891,10 @@ def to_crs(
             tol=tol,
         )
 
-    def __str__(self):
+    def __str__(self) -> str:
         return self.__repr__()
 
-    def __repr__(self):
+    def __repr__(self) -> str:
         return f"GeoBox({self._shape.yx!r}, {self._affine!r}, {self._crs!r})"
 
     def __eq__(self, other):
@@ -1340,7 +1340,7 @@ def __init__(
         tile_shape: Union[SomeShape, Chunks2d, None],
         *,
         _tiles: Optional[RoiTiles] = None,
-    ):
+    ) -> None:
         """
         Construct from a :py:class:`~odc.geo.GeoBox`.
 
@@ -1546,7 +1546,7 @@ def __dask_tokenize__(self):
             *self._tiles.__dask_tokenize__()[1:],
         )
 
-    def __str__(self):
+    def __str__(self) -> str:
         return str(self.roi)
 
     def __eq__(self, __value: object) -> bool:

odc/geo/geom.py

@@ -49,7 +49,7 @@ class BoundingBox(Sequence[float]):
 
     def __init__(
         self, left: float, bottom: float, right: float, top: float, crs: MaybeCRS = None
-    ):
+    ) -> None:
         self._box = (left, bottom, right, top)
         self._crs = norm_crs(crs)
 
@@ -498,7 +498,7 @@ def __init__(
         self,
         geom: Union[base.BaseGeometry, Dict[str, Any], "Geometry"],
         crs: MaybeCRS = None,
-    ):
+    ) -> None:
         if isinstance(geom, Geometry):
             assert crs is None
             self.crs: Optional[CRS] = geom.crs
@@ -924,10 +924,10 @@ def __eq__(self, other: Any) -> bool:
             and self.geom == other.geom
         )
 
-    def __str__(self):
+    def __str__(self) -> str:
         return f"Geometry({self.__geo_interface__}, {self.crs!r})"
 
-    def __repr__(self):
+    def __repr__(self) -> str:
         return f"Geometry({self.geom}, {self.crs})"
 
     # Implement pickle/unpickle
@@ -936,8 +936,8 @@ def __repr__(self):
     def __getstate__(self):
         return {"geom": self.json, "crs": self.crs}
 
-    def __setstate__(self, state):
-        self.__init__(**state)
+    def __setstate__(self, state) -> None:
+        self.__init__(**state)  # type: ignore[misc]
 
     @property
     def is_multi(self) -> bool:

odc/geo/gridspec.py

@@ -54,7 +54,7 @@ def __init__(
         origin: Optional[XY[float]] = None,
         flipx: bool = False,
         flipy: bool = False,
-    ):
+    ) -> None:
         tile_shape = shape_(tile_shape)
         resolution = res_(resolution)
 

odc/geo/math.py

@@ -653,7 +653,7 @@ class Bin1D:
 
     __slots__ = ("sz", "origin", "direction")
 
-    def __init__(self, sz: float, origin: float = 0.0, direction: Literal[1, -1] = 1):
+    def __init__(self, sz: float, origin: float = 0.0, direction: Literal[1, -1] = 1) -> None:
         """
         Construct :py:class:`~odc.geo.math.Bin1D` object.
 

odc/geo/overlap.py

@@ -53,7 +53,7 @@ class LinearPointTransform:
     Point transform within the same projection.
     """
 
-    def __init__(self, A: Affine, back: Optional["LinearPointTransform"] = None):
+    def __init__(self, A: Affine, back: Optional["LinearPointTransform"] = None) -> None:
         self.A = A
         self._back = back
 
@@ -98,7 +98,7 @@ def __init__(
         src: GeoBoxBase,
         dst: GeoBoxBase,
         back: Optional["GbxPointTransform"] = None,
-    ):
+    ) -> None:
         assert src.crs is not None and dst.crs is not None
         self._src = src
         self._dst = dst

odc/geo/roi.py

@@ -77,7 +77,7 @@ def chunks(self) -> Chunks2d: ...
 
     def locate(self, pix: SomeIndex2d) -> Tuple[int, int]: ...
 
-    def __dask_tokenize__(self): ...
+    def __dask_tokenize__(self) -> None: ...
 
 
 def norm_slice_2d(
@@ -88,7 +88,7 @@ def norm_slice_2d(
     return roi_normalise(iyx_(idx).yx, shape)
 
 
-def _fmt_shape(shape):
+def _fmt_shape(shape) -> str:
     n1, n2 = shape.yx
     if max(n1, n2) > 10_000:
         return f"{n1:_d}x{n2:_d}"

odc/geo/types.py

@@ -482,7 +482,7 @@ class _Func2Map(Mapping[TK, TV]):
 
     __slots__ = ("_func", "_keys")
 
-    def __init__(self, func: Callable[[TK], TV], keys: Sequence[TK]):
+    def __init__(self, func: Callable[[TK], TV], keys: Sequence[TK]) -> None:
         self._func = func
         self._keys = keys
 

odc/geo/ui.py

@@ -84,7 +84,7 @@ def svg_base_map(
     sz=360,
     map_svg_path=None,
     target=None,
-):
+) -> str:
     clip_bbox = None
     if bbox is not None:
         clip_bbox = geom.bbox_intersection(
@@ -140,7 +140,7 @@ def make_svg(
     stroke_width=1,
     bbox=None,
     sz=360,
-):
+) -> str:
     if bbox is None:
         bbox = (0, 0, sz, sz)