Skip to content

Commit

Permalink
Use new-style type annotations (#670)
Browse files Browse the repository at this point in the history 
* reorder pyproject (disable fixing)

* ignore mccabe warnings

* import future annotations where necessary

* autofix new-style unions

* use new-style builtin-types annotations

* remove unnecessary quotes in types and misc fixes
  • Loading branch information
@zigaLuksic
zigaLuksic committed May 24, 2023
1 parent 82181b0 commit 96b4e7a
Show file tree
Hide file tree
Showing 41 changed files with 590 additions and 591 deletions.
32 changes: 15 additions & 17 deletions core/eolearn/core/core_tasks.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,7 @@

import copy
from abc import ABCMeta
from typing import Any, Callable, Dict, Iterable, List, Optional, Tuple, Union, cast
from typing import Any, Callable, Iterable, Tuple, Union, cast

import fs
import numpy as np
Expand Down Expand Up @@ -52,9 +52,7 @@ def execute(self, eopatch: EOPatch) -> EOPatch:
class IOTask(EOTask, metaclass=ABCMeta):
"""An abstract Input/Output task that can handle a path and a filesystem object."""

def __init__(
self, path: str, filesystem: Optional[FS] = None, create: bool = False, config: Optional[SHConfig] = None
):
def __init__(self, path: str, filesystem: FS | None = None, create: bool = False, config: SHConfig | None = None):
"""
:param path: root path where all EOPatches are saved
:param filesystem: An existing filesystem object. If not given it will be initialized according to the EOPatch
Expand Down Expand Up @@ -84,7 +82,7 @@ def filesystem(self) -> FS:
class SaveTask(IOTask):
"""Saves the given EOPatch to a filesystem."""

def __init__(self, path: str, filesystem: Optional[FS] = None, config: Optional[SHConfig] = None, **kwargs: Any):
def __init__(self, path: str, filesystem: FS | None = None, config: SHConfig | None = None, **kwargs: Any):
"""
:param path: root path where all EOPatches are saved
:param filesystem: An existing filesystem object. If not given it will be initialized according to the EOPatch
Expand Down Expand Up @@ -115,7 +113,7 @@ def execute(self, eopatch: EOPatch, *, eopatch_folder: str = "") -> EOPatch:
class LoadTask(IOTask):
"""Loads an EOPatch from a filesystem."""

def __init__(self, path: str, filesystem: Optional[FS] = None, config: Optional[SHConfig] = None, **kwargs: Any):
def __init__(self, path: str, filesystem: FS | None = None, config: SHConfig | None = None, **kwargs: Any):
"""
:param path: root directory where all EOPatches are saved
:param filesystem: An existing filesystem object. If not given it will be initialized according to the EOPatch
Expand All @@ -128,7 +126,7 @@ def __init__(self, path: str, filesystem: Optional[FS] = None, config: Optional[
self.kwargs = kwargs
super().__init__(path, filesystem=filesystem, create=False, config=config)

def execute(self, eopatch: Optional[EOPatch] = None, *, eopatch_folder: str = "") -> EOPatch:
def execute(self, eopatch: EOPatch | None = None, *, eopatch_folder: str = "") -> EOPatch:
"""Loads the EOPatch from disk: `folder/eopatch_folder`.
:param eopatch: Optional input EOPatch. If given the loaded features are merged onto it, otherwise a new EOPatch
Expand Down Expand Up @@ -256,9 +254,9 @@ class InitializeFeatureTask(EOTask):
def __init__(
self,
features: FeaturesSpecification,
shape: Union[Tuple[int, ...], FeatureSpec],
shape: tuple[int, ...] | FeatureSpec,
init_value: int = 0,
dtype: Union[np.dtype, type] = np.uint8,
dtype: np.dtype | type = np.uint8,
):
"""
:param features: A collection of features to initialize.
Expand All @@ -269,8 +267,8 @@ def __init__(
"""

self.features = self.parse_features(features)
self.shape_feature: Optional[Tuple[FeatureType, Optional[str]]]
self.shape: Union[None, Tuple[int, int, int], Tuple[int, int, int, int]]
self.shape_feature: tuple[FeatureType, str | None] | None
self.shape: None | tuple[int, int, int] | tuple[int, int, int, int]

try:
self.shape_feature = self.parse_feature(shape) # type: ignore[arg-type]
Expand Down Expand Up @@ -372,7 +370,7 @@ def __init__(
self,
input_features: FeaturesSpecification,
output_features: FeaturesSpecification,
map_function: Optional[Callable] = None,
map_function: Callable | None = None,
**kwargs: Any,
):
"""
Expand Down Expand Up @@ -452,7 +450,7 @@ def __init__(
self,
input_features: FeaturesSpecification,
output_feature: SingleFeatureSpec,
zip_function: Optional[Callable] = None,
zip_function: Callable | None = None,
**kwargs: Any,
):
"""
Expand Down Expand Up @@ -489,15 +487,15 @@ def zip_method(self, *features: Any) -> Any:
class MergeFeatureTask(ZipFeatureTask):
"""Merges multiple features together by concatenating their data along the specified axis."""

def zip_method(self, *f: np.ndarray, dtype: Union[None, np.dtype, type] = None, axis: int = -1) -> np.ndarray:
def zip_method(self, *f: np.ndarray, dtype: None | np.dtype | type = None, axis: int = -1) -> np.ndarray:
"""Concatenates the data of features along the specified axis."""
return np.concatenate(f, axis=axis, dtype=dtype) # pylint: disable=unexpected-keyword-arg


class ExtractBandsTask(MapFeatureTask):
"""Moves a subset of bands from one feature to a new one."""

def __init__(self, input_feature: FeaturesSpecification, output_feature: FeaturesSpecification, bands: List[int]):
def __init__(self, input_feature: FeaturesSpecification, output_feature: FeaturesSpecification, bands: list[int]):
"""
:param input_feature: A source feature from which to take the subset of bands.
:param output_feature: An output feature to which to write the bands.
Expand All @@ -518,8 +516,8 @@ class ExplodeBandsTask(EOTask):

def __init__(
self,
input_feature: Tuple[FeatureType, str],
output_mapping: Dict[Tuple[FeatureType, str], Union[int, Iterable[int]]],
input_feature: tuple[FeatureType, str],
output_mapping: dict[tuple[FeatureType, str], int | Iterable[int]],
):
"""
:param input_feature: A source feature from which to take the subset of bands.
Expand Down
101 changes: 47 additions & 54 deletions core/eolearn/core/eodata.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -20,15 +20,10 @@
Callable,
Iterable,
Iterator,
List,
Literal,
Mapping,
MutableMapping,
Optional,
Set,
Tuple,
TypeVar,
Union,
cast,
overload,
)
Expand Down Expand Up @@ -75,7 +70,7 @@ class _FeatureDict(MutableMapping[str, T], metaclass=ABCMeta):

FORBIDDEN_CHARS = {".", "/", "\\", "|", ";", ":", "\n", "\t"}

def __init__(self, feature_dict: Mapping[str, Union[T, FeatureIO[T]]], feature_type: FeatureType):
def __init__(self, feature_dict: Mapping[str, T | FeatureIO[T]], feature_type: FeatureType):
"""
:param feature_dict: A dictionary of feature names and values
:param feature_type: Type of features
Expand All @@ -85,7 +80,7 @@ def __init__(self, feature_dict: Mapping[str, Union[T, FeatureIO[T]]], feature_t
self.feature_type = feature_type
self._content = dict(feature_dict.items())

def __setitem__(self, feature_name: str, value: Union[T, FeatureIO[T]]) -> None:
def __setitem__(self, feature_name: str, value: T | FeatureIO[T]) -> None:
"""Before setting value to the dictionary it checks that value is of correct type and dimension and tries to
transform value in correct form.
"""
Expand Down Expand Up @@ -117,7 +112,7 @@ def __getitem__(self, feature_name: str) -> T:

return value

def _get_unloaded(self, feature_name: str) -> Union[T, FeatureIO[T]]:
def _get_unloaded(self, feature_name: str) -> T | FeatureIO[T]:
"""Returns the value, bypassing lazy-loading mechanisms."""
return self._content[feature_name]

Expand Down Expand Up @@ -227,26 +222,26 @@ class EOPatch:
"""

# establish types of property value holders
_timestamps: List[dt.datetime]
_bbox: Optional[BBox]
_meta_info: Union[FeatureIOJson, _FeatureDictJson]
_timestamps: list[dt.datetime]
_bbox: BBox | None
_meta_info: FeatureIOJson | _FeatureDictJson

def __init__(
self,
*,
data: Optional[Mapping[str, np.ndarray]] = None,
mask: Optional[Mapping[str, np.ndarray]] = None,
scalar: Optional[Mapping[str, np.ndarray]] = None,
label: Optional[Mapping[str, np.ndarray]] = None,
vector: Optional[Mapping[str, gpd.GeoDataFrame]] = None,
data_timeless: Optional[Mapping[str, np.ndarray]] = None,
mask_timeless: Optional[Mapping[str, np.ndarray]] = None,
scalar_timeless: Optional[Mapping[str, np.ndarray]] = None,
label_timeless: Optional[Mapping[str, np.ndarray]] = None,
vector_timeless: Optional[Mapping[str, gpd.GeoDataFrame]] = None,
meta_info: Optional[Mapping[str, Any]] = None,
bbox: Optional[BBox] = None,
timestamps: Optional[List[dt.datetime]] = None,
data: Mapping[str, np.ndarray] | None = None,
mask: Mapping[str, np.ndarray] | None = None,
scalar: Mapping[str, np.ndarray] | None = None,
label: Mapping[str, np.ndarray] | None = None,
vector: Mapping[str, gpd.GeoDataFrame] | None = None,
data_timeless: Mapping[str, np.ndarray] | None = None,
mask_timeless: Mapping[str, np.ndarray] | None = None,
scalar_timeless: Mapping[str, np.ndarray] | None = None,
label_timeless: Mapping[str, np.ndarray] | None = None,
vector_timeless: Mapping[str, gpd.GeoDataFrame] | None = None,
meta_info: Mapping[str, Any] | None = None,
bbox: BBox | None = None,
timestamps: list[dt.datetime] | None = None,
):
self.data: MutableMapping[str, np.ndarray] = _FeatureDictNumpy(data or {}, FeatureType.DATA)
self.mask: MutableMapping[str, np.ndarray] = _FeatureDictNumpy(mask or {}, FeatureType.MASK)
Expand All @@ -273,18 +268,18 @@ def __init__(
self.timestamps = timestamps or []

@property
def timestamp(self) -> List[dt.datetime]:
def timestamp(self) -> list[dt.datetime]:
"""A property for handling the deprecated timestamp attribute."""
warn(TIMESTAMP_RENAME_WARNING, category=EODeprecationWarning, stacklevel=2)
return self.timestamps

@timestamp.setter
def timestamp(self, value: List[dt.datetime]) -> None:
def timestamp(self, value: list[dt.datetime]) -> None:
warn(TIMESTAMP_RENAME_WARNING, category=EODeprecationWarning, stacklevel=2)
self.timestamps = value

@property
def timestamps(self) -> List[dt.datetime]:
def timestamps(self) -> list[dt.datetime]:
"""A property for handling the `timestamps` attribute."""
return self._timestamps

Expand All @@ -296,12 +291,12 @@ def timestamps(self, value: Iterable[dt.datetime]) -> None:
raise TypeError(f"Cannot assign {value} as timestamps. Should be a sequence of datetime.datetime objects.")

@property
def bbox(self) -> Optional[BBox]:
def bbox(self) -> BBox | None:
"""A property for handling the `bbox` attribute."""
return self._bbox

@bbox.setter
def bbox(self, value: Optional[BBox]) -> None:
def bbox(self, value: BBox | None) -> None:
if not (isinstance(value, BBox) or value is None):
raise TypeError(f"Cannot assign {value} as bbox. Should be a `BBox` object.")
if value is None:
Expand All @@ -317,7 +312,7 @@ def meta_info(self) -> MutableMapping[str, Any]:
return self._meta_info # type: ignore[return-value] # mypy cannot verify due to mutations

@meta_info.setter
def meta_info(self, value: Union[Mapping[str, Any], FeatureIOJson]) -> None:
def meta_info(self, value: Mapping[str, Any] | FeatureIOJson) -> None:
self._meta_info = value if isinstance(value, FeatureIOJson) else _FeatureDictJson(value, FeatureType.META_INFO)

def __setattr__(self, key: str, value: object) -> None:
Expand All @@ -331,20 +326,20 @@ def __setattr__(self, key: str, value: object) -> None:
super().__setattr__(key, value)

@overload
def __getitem__(self, key: Union[Literal[FeatureType.BBOX], Tuple[Literal[FeatureType.BBOX], Any]]) -> BBox:
def __getitem__(self, key: Literal[FeatureType.BBOX] | tuple[Literal[FeatureType.BBOX], Any]) -> BBox:
...

@overload
def __getitem__(
self, key: Union[Literal[FeatureType.TIMESTAMPS], Tuple[Literal[FeatureType.TIMESTAMPS], Any]]
) -> List[dt.datetime]:
self, key: Literal[FeatureType.TIMESTAMPS] | tuple[Literal[FeatureType.TIMESTAMPS], Any]
) -> list[dt.datetime]:
...

@overload
def __getitem__(self, key: Union[FeatureType, Tuple[FeatureType, Union[str, None, EllipsisType]]]) -> Any:
def __getitem__(self, key: FeatureType | tuple[FeatureType, str | None | EllipsisType]) -> Any:
...

def __getitem__(self, key: Union[FeatureType, Tuple[FeatureType, Union[str, None, EllipsisType]]]) -> Any:
def __getitem__(self, key: FeatureType | tuple[FeatureType, str | None | EllipsisType]) -> Any:
"""Provides features of requested feature type. It can also accept a tuple of (feature_type, feature_name).
:param key: Feature type or a (feature_type, feature_name) pair.
Expand All @@ -356,9 +351,7 @@ def __getitem__(self, key: Union[FeatureType, Tuple[FeatureType, Union[str, None
return value[feature_name]
return value

def __setitem__(
self, key: Union[FeatureType, Tuple[FeatureType, Union[str, None, EllipsisType]]], value: Any
) -> None:
def __setitem__(self, key: FeatureType | tuple[FeatureType, str | None | EllipsisType], value: Any) -> None:
"""Sets a new value to the given FeatureType or tuple of (feature_type, feature_name)."""
feature_type, feature_name = key if isinstance(key, tuple) else (key, None)
ftype_attr = FeatureType(feature_type).value
Expand All @@ -368,7 +361,7 @@ def __setitem__(
else:
setattr(self, ftype_attr, value)

def __delitem__(self, feature: Union[FeatureType, FeatureSpec]) -> None:
def __delitem__(self, feature: FeatureType | FeatureSpec) -> None:
"""Deletes the selected feature type or feature."""
if isinstance(feature, tuple):
feature_type, feature_name = feature
Expand Down Expand Up @@ -490,7 +483,7 @@ def __copy__(self, features: FeaturesSpecification = ...) -> EOPatch:
new_eopatch[feature_type][feature_name] = self[feature_type]._get_unloaded(feature_name) # noqa: SLF001
return new_eopatch

def __deepcopy__(self, memo: Optional[dict] = None, features: FeaturesSpecification = ...) -> EOPatch:
def __deepcopy__(self, memo: dict | None = None, features: FeaturesSpecification = ...) -> EOPatch:
"""Returns a new EOPatch with deep copies of given features.
:param memo: built-in parameter for memoization
Expand Down Expand Up @@ -544,7 +537,7 @@ def reset_feature_type(self, feature_type: FeatureType) -> None:
else:
self[feature_type] = {}

def get_spatial_dimension(self, feature_type: FeatureType, feature_name: str) -> Tuple[int, int]:
def get_spatial_dimension(self, feature_type: FeatureType, feature_name: str) -> tuple[int, int]:
"""
Returns a tuple of spatial dimensions (height, width) of a feature.
Expand All @@ -557,12 +550,12 @@ def get_spatial_dimension(self, feature_type: FeatureType, feature_name: str) ->

raise ValueError(f"Features of type {feature_type} do not have a spatial dimension or are not arrays.")

def get_features(self) -> List[FeatureSpec]:
def get_features(self) -> list[FeatureSpec]:
"""Returns a list of all non-empty features of EOPatch.
:return: List of non-empty features
"""
feature_list: List[FeatureSpec] = []
feature_list: list[FeatureSpec] = []
for feature_type in FeatureType:
if feature_type is FeatureType.BBOX or feature_type is FeatureType.TIMESTAMPS:
if feature_type in self:
Expand All @@ -578,7 +571,7 @@ def save(
features: FeaturesSpecification = ...,
overwrite_permission: OverwritePermission = OverwritePermission.ADD_ONLY,
compress_level: int = 0,
filesystem: Optional[FS] = None,
filesystem: FS | None = None,
) -> None:
"""Method to save an EOPatch from memory to a storage.
Expand Down Expand Up @@ -606,7 +599,7 @@ def save(

@staticmethod
def load(
path: str, features: FeaturesSpecification = ..., lazy_loading: bool = False, filesystem: Optional[FS] = None
path: str, features: FeaturesSpecification = ..., lazy_loading: bool = False, filesystem: FS | None = None
) -> EOPatch:
"""Method to load an EOPatch from a storage into memory.
Expand Down Expand Up @@ -640,8 +633,8 @@ def merge(
self,
*eopatches: EOPatch,
features: FeaturesSpecification = ...,
time_dependent_op: Union[Literal[None, "concatenate", "min", "max", "mean", "median"], Callable] = None,
timeless_op: Union[Literal[None, "concatenate", "min", "max", "mean", "median"], Callable] = None,
time_dependent_op: Literal[None, "concatenate", "min", "max", "mean", "median"] | Callable = None,
timeless_op: Literal[None, "concatenate", "min", "max", "mean", "median"] | Callable = None,
) -> EOPatch:
"""Merge features of given EOPatches into a new EOPatch.
Expand Down Expand Up @@ -674,7 +667,7 @@ def merge(
self, *eopatches, features=features, time_dependent_op=time_dependent_op, timeless_op=timeless_op
)

def consolidate_timestamps(self, timestamps: List[dt.datetime]) -> Set[dt.datetime]:
def consolidate_timestamps(self, timestamps: list[dt.datetime]) -> set[dt.datetime]:
"""Removes all frames from the EOPatch with a date not found in the provided timestamps list.
:param timestamps: keep frames with date found in this list
Expand All @@ -697,12 +690,12 @@ def plot(
self,
feature: FeatureSpec,
*,
times: Union[List[int], slice, None] = None,
channels: Union[List[int], slice, None] = None,
channel_names: Optional[List[str]] = None,
rgb: Optional[Tuple[int, int, int]] = None,
backend: Union[str, PlotBackend] = "matplotlib",
config: Optional[PlotConfig] = None,
times: list[int] | slice | None = None,
channels: list[int] | slice | None = None,
channel_names: list[str] | None = None,
rgb: tuple[int, int, int] | None = None,
backend: str | PlotBackend = "matplotlib",
config: PlotConfig | None = None,
**kwargs: Any,
) -> object:
"""Plots an `EOPatch` feature.
Expand Down

0 comments on commit 96b4e7a

Please sign in to comment.