From f9061a70469aefe97aadacef182923f0a6dc62b3 Mon Sep 17 00:00:00 2001 From: Sam Vente Date: Mon, 30 Oct 2023 09:50:34 +0100 Subject: [PATCH] user beter typing --- hydromt/data_adapter/geodataframe.py | 16 +++++++++------- hydromt/data_adapter/geodataset.py | 24 ++++++++++++++---------- hydromt/data_adapter/rasterdataset.py | 21 ++++++++++++--------- hydromt/typing.py | 22 ++++++++++++++++++++++ tests/test_data_adapter.py | 6 +++--- 5 files changed, 60 insertions(+), 29 deletions(-) create mode 100644 hydromt/typing.py diff --git a/hydromt/data_adapter/geodataframe.py b/hydromt/data_adapter/geodataframe.py index 1749b0418..1da749d01 100644 --- a/hydromt/data_adapter/geodataframe.py +++ b/hydromt/data_adapter/geodataframe.py @@ -3,8 +3,7 @@ import warnings from datetime import datetime from os.path import basename, join -from pathlib import Path -from typing import Literal, NewType, Optional, Tuple, Union +from typing import Literal, Optional, Union import numpy as np import pyproj @@ -12,6 +11,11 @@ from pystac import Catalog as StacCatalog from pystac import Item as StacItem +from hydromt.typing import ( + GeoDataframeSource, + TotalBounds, +) + from .. import gis_utils, io from .data_adapter import DataAdapter @@ -19,8 +23,6 @@ __all__ = ["GeoDataFrameAdapter", "GeoDataframeSource"] -GeoDataframeSource = NewType("GeoDataframeSource", Union[str, Path]) - class GeoDataFrameAdapter(DataAdapter): @@ -408,7 +410,7 @@ def _set_metadata(self, gdf): return gdf - def get_bbox(self, detect=True) -> Tuple[Tuple[float, float, float, float], int]: + def get_bbox(self, detect=True) -> TotalBounds: """Return the bounding box and espg code of the dataset. if the bounding box is not set and detect is True, @@ -438,7 +440,7 @@ def get_bbox(self, detect=True) -> Tuple[Tuple[float, float, float, float], int] def detect_bbox( self, gdf=None, - ) -> Tuple[Tuple[float, float, float, float], int]: + ) -> TotalBounds: """Detect the bounding box and crs of the dataset. If no dataset is provided, it will be fetched acodring to the settings in the @@ -503,7 +505,7 @@ def to_stac_catalog( bbox, crs = self.get_bbox(detect=True) bbox = list(bbox) props = {**self.meta, "crs": crs} - except Exception as e: + except (IndexError, KeyError, pyproj.exceptions.CRSError) as e: if on_error == "skip": logger.warning( "Skipping {name} during stac conversion because" diff --git a/hydromt/data_adapter/geodataset.py b/hydromt/data_adapter/geodataset.py index 710a40c4f..75879982b 100644 --- a/hydromt/data_adapter/geodataset.py +++ b/hydromt/data_adapter/geodataset.py @@ -4,8 +4,7 @@ import warnings from datetime import datetime from os.path import basename, join -from pathlib import Path -from typing import Literal, NewType, Optional, Tuple, Union +from typing import Literal, Optional, Union import numpy as np import pandas as pd @@ -15,6 +14,12 @@ from pystac import Catalog as StacCatalog from pystac import Item as StacItem +from hydromt.typing import ( + GeoDatasetSource, + TimeRange, + TotalBounds, +) + from .. import gis_utils, io from ..raster import GEO_MAP_COORD from .data_adapter import DataAdapter @@ -23,8 +28,6 @@ __all__ = ["GeoDatasetAdapter", "GeoDatasetSource"] -GeoDatasetSource = NewType("GeoDatasetSource", Union[str, Path]) - class GeoDatasetAdapter(DataAdapter): @@ -475,7 +478,7 @@ def _apply_unit_conversion(self, ds, logger=logger): ds[name].attrs.update(attrs) # set original attributes return ds - def get_bbox(self, detect=True) -> Tuple[Tuple[float, float, float, float], int]: + def get_bbox(self, detect=True) -> TotalBounds: """Return the bounding box and espg code of the dataset. if the bounding box is not set and detect is True, @@ -496,13 +499,14 @@ def get_bbox(self, detect=True) -> Tuple[Tuple[float, float, float, float], int] The ESPG code of the CRS of the coordinates returned in bbox """ bbox = self.extent.get("bbox", None) - crs = self.crs if bbox is None and detect: bbox, crs = self.detect_bbox() + crs = self.crs + return bbox, crs - def get_time_range(self, detect=True): + def get_time_range(self, detect=True) -> TimeRange: """Detect the time range of the dataset. if the time range is not set and detect is True, @@ -531,7 +535,7 @@ def get_time_range(self, detect=True): def detect_bbox( self, ds=None, - ) -> Tuple[Tuple[float, float, float, float], int]: + ) -> TotalBounds: """Detect the bounding box and crs of the dataset. If no dataset is provided, it will be fetched according to the settings in the @@ -562,7 +566,7 @@ def detect_bbox( bounds = ds.vector.bounds return bounds, crs - def detect_time_range(self, ds=None) -> Tuple[np.datetime64, np.datetime64]: + def detect_time_range(self, ds=None) -> TimeRange: """Detect the temporal range of the dataset. If no dataset is provided, it will be fetched according to the settings in the @@ -623,7 +627,7 @@ def to_stac_catalog( start_dt = pd.to_datetime(start_dt) end_dt = pd.to_datetime(end_dt) props = {**self.meta, "crs": crs} - except Exception as e: + except (IndexError, KeyError, pyproj.exceptions.CRSError) as e: if on_error == "skip": logger.warning( "Skipping {name} during stac conversion because" diff --git a/hydromt/data_adapter/rasterdataset.py b/hydromt/data_adapter/rasterdataset.py index b01a63c07..5addc4cac 100644 --- a/hydromt/data_adapter/rasterdataset.py +++ b/hydromt/data_adapter/rasterdataset.py @@ -5,9 +5,8 @@ import os import warnings from datetime import datetime -from os import PathLike from os.path import basename, join -from typing import Dict, Literal, NewType, Optional, Tuple, Union, cast +from typing import Dict, Literal, Optional, Tuple, Union, cast import geopandas as gpd import numpy as np @@ -20,6 +19,12 @@ from pystac import Item as StacItem from rasterio.errors import RasterioIOError +from hydromt.typing import ( + RasterDatasetSource, + TimeRange, + TotalBounds, +) + from .. import gis_utils, io from ..raster import GEO_MAP_COORD from .caching import cache_vrt_tiles @@ -29,8 +34,6 @@ __all__ = ["RasterDatasetAdapter", "RasterDatasetSource"] -RasterDatasetSource = NewType("RasterDatasetSource", Union[str, PathLike]) - class RasterDatasetAdapter(DataAdapter): @@ -674,7 +677,7 @@ def _parse_zoom_level( logger.debug(f"Parsed zoom_level {zl} ({dst_res:.2f})") return zl - def get_bbox(self, detect=True) -> Tuple[Tuple[float, float, float, float], int]: + def get_bbox(self, detect=True) -> TotalBounds: """Return the bounding box and espg code of the dataset. if the bounding box is not set and detect is True, @@ -701,7 +704,7 @@ def get_bbox(self, detect=True) -> Tuple[Tuple[float, float, float, float], int] return bbox, crs - def get_time_range(self, detect=True) -> Tuple[np.datetime64, np.datetime64]: + def get_time_range(self, detect=True) -> TimeRange: """Detect the time range of the dataset. if the time range is not set and detect is True, @@ -730,7 +733,7 @@ def get_time_range(self, detect=True) -> Tuple[np.datetime64, np.datetime64]: def detect_bbox( self, ds=None, - ) -> Tuple[Tuple[float, float, float, float], int]: + ) -> TotalBounds: """Detect the bounding box and crs of the dataset. If no dataset is provided, it will be fetched according to the settings in the @@ -761,7 +764,7 @@ def detect_bbox( return bounds, crs - def detect_time_range(self, ds=None) -> Tuple[np.datetime64, np.datetime64]: + def detect_time_range(self, ds=None) -> TimeRange: """Detect the temporal range of the dataset. If no dataset is provided, it will be fetched accodring to the settings in the @@ -821,7 +824,7 @@ def to_stac_catalog( start_dt = pd.to_datetime(start_dt) end_dt = pd.to_datetime(end_dt) props = {**self.meta, "crs": crs} - except Exception as e: + except (IndexError, KeyError, pyproj.exceptions.CRSError) as e: if on_error == "skip": logger.warning( "Skipping {name} during stac conversion because" diff --git a/hydromt/typing.py b/hydromt/typing.py new file mode 100644 index 000000000..1217df082 --- /dev/null +++ b/hydromt/typing.py @@ -0,0 +1,22 @@ +"""Type aliases used by hydromt.""" + +from datetime import datetime +from enum import Enum +from pathlib import Path +from typing import Tuple, Union + +GeoDataframeSource = Union[str, Path] +GeoDatasetSource = Union[str, Path] +RasterDatasetSource = Union[str, Path] +Bbox = Tuple[float, float, float, float] +Crs = int +TotalBounds = Tuple[Bbox, Crs] +TimeRange = Tuple[datetime, datetime] + + +class ErrorHandleMethod(Enum): + """Strategies for error handling withing hydromt.""" + + RAISE = 1 + SKIP = 2 + COERCE = 3 diff --git a/tests/test_data_adapter.py b/tests/test_data_adapter.py index d325176fa..752bd3518 100644 --- a/tests/test_data_adapter.py +++ b/tests/test_data_adapter.py @@ -501,7 +501,7 @@ def test_to_stac_geodataframe(geodf, tmpdir): gdf_stac_catalog.add_item(gds_stac_item) outcome = cast(StacCatalog, adapter.to_stac_catalog(on_error="coerce")) assert gdf_stac_catalog.to_dict() == outcome.to_dict() # type: ignore - del adapter.crs # manually create an invalid adapter by deleting the crs + adapter.crs = -3.14 # manually create an invalid adapter by deleting the crs assert adapter.to_stac_catalog("skip") is None @@ -535,7 +535,7 @@ def test_to_stac_raster(): outcome = cast(StacCatalog, adapter.to_stac_catalog(on_error="raise")) assert raster_stac_catalog.to_dict() == outcome.to_dict() # type: ignore - del adapter.crs # manually create an invalid adapter by deleting the crs + adapter.crs = -3.14 # manually create an invalid adapter by deleting the crs assert adapter.to_stac_catalog("skip") is None @@ -568,7 +568,7 @@ def test_to_stac_geodataset(geoda, tmpdir): outcome = cast(StacCatalog, adapter.to_stac_catalog(on_error="coerce")) assert gds_stac_catalog.to_dict() == outcome.to_dict() # type: ignore - del adapter.crs # manually create an invalid adapter by deleting the crs + adapter.crs = -3.14 # manually create an invalid adapter by deleting the crs assert adapter.to_stac_catalog("skip") is None