Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

style: add type hints #45

Merged
merged 1 commit into from
Dec 19, 2023
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
53 changes: 39 additions & 14 deletions eodag_cube/api/product/_product.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,8 +15,11 @@
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from __future__ import annotations

import logging
from contextlib import contextmanager
from typing import TYPE_CHECKING, Any, Dict, Optional, Union

import numpy as np
import rasterio
Expand All @@ -28,7 +31,12 @@
from eodag.utils import get_geometry_from_various
from eodag.utils.exceptions import DownloadError, UnsupportedDatasetAddressScheme

logger = logging.getLogger("eodag.api.product")
if TYPE_CHECKING:
from rasterio.enums import Resampling
from shapely.geometry.base import BaseGeometry
from xarray import DataArray

logger = logging.getLogger("eodag-cube.api.product")


class EOProduct(EOProduct_core):
Expand All @@ -47,6 +55,19 @@ class EOProduct(EOProduct_core):
:type provider: str
:param properties: The metadata of the product
:type properties: dict
:ivar product_type: The product type
:vartype product_type: str
:ivar location: The path to the product, either remote or local if downloaded
:vartype location: str
:ivar remote_location: The remote path to the product
:vartype remote_location: str
:ivar search_kwargs: The search kwargs used by eodag to search for the product
:vartype search_kwargs: Any
:ivar geometry: The geometry of the product
:vartype geometry: :class:`shapely.geometry.base.BaseGeometry`
:ivar search_intersection: The intersection between the product's geometry
and the search area.
:vartype search_intersection: :class:`shapely.geometry.base.BaseGeometry` or None

.. note::
The geojson spec `enforces <https://github.com/geojson/draft-geojson/pull/6>`_
Expand All @@ -56,18 +77,22 @@ class EOProduct(EOProduct_core):
mentioned CRS.
"""

def __init__(self, *args, **kwargs):
super(EOProduct, self).__init__(*args, **kwargs)
def __init__(
self, provider: str, properties: Dict[str, Any], **kwargs: Any
) -> None:
super(EOProduct, self).__init__(
provider=provider, properties=properties, **kwargs
)

def get_data(
self,
band,
crs=None,
resolution=None,
extent=None,
resampling=None,
**rioxr_kwargs,
):
band: str,
crs: Optional[str] = None,
resolution: Optional[float] = None,
extent: Optional[Union[str, Dict[str, float], BaseGeometry]] = None,
resampling: Optional[Resampling] = None,
**rioxr_kwargs: Any,
) -> DataArray:
"""Retrieves all or part of the raster data abstracted by the :class:`EOProduct`

:param band: The band of the dataset to retrieve (e.g.: 'B01')
Expand All @@ -92,7 +117,7 @@ def get_data(
:param resampling: (optional) Warp resampling algorithm passed to :class:`rasterio.vrt.WarpedVRT`
:type resampling: Resampling
:param rioxr_kwargs: kwargs passed to ``rioxarray.open_rasterio()``
:type rioxr_kwargs: dict
:type rioxr_kwargs: Any
:returns: The numeric matrix corresponding to the sub dataset or an empty
array if unable to get the data
:rtype: xarray.DataArray
Expand Down Expand Up @@ -142,7 +167,7 @@ def get_data(
warped_vrt_args["resampling"] = resampling

@contextmanager
def pass_resource(resource, **kwargs):
def pass_resource(resource: Any, **kwargs: Any) -> Any:
yield resource

if warped_vrt_args:
Expand Down Expand Up @@ -181,14 +206,14 @@ def pass_resource(resource, **kwargs):
logger.error(e)
return fail_value

def _get_rio_env(self, dataset_address):
def _get_rio_env(self, dataset_address: str) -> Dict[str, Any]:
"""Get rasterio environement variables needed for data access.

:param dataset_address: address of the data to read
:type dataset_address: str

:return: The rasterio environement variables
:rtype: dict
:rtype: Dict[str, Any]
"""
product_location_scheme = dataset_address.split("://")[0]
if product_location_scheme == "s3" and hasattr(
Expand Down
18 changes: 15 additions & 3 deletions eodag_cube/api/product/drivers/generic.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,23 +15,35 @@
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from __future__ import annotations

from pathlib import Path
from typing import TYPE_CHECKING

import rasterio

from eodag.api.product.drivers.base import DatasetDriver
from eodag.utils import uri_to_path
from eodag.utils.exceptions import AddressNotFound, UnsupportedDatasetAddressScheme

if TYPE_CHECKING:
from eodag.api.product._product import EOProduct


class GenericDriver(DatasetDriver):
"""Generic Driver for products that need to be downloaded"""

def get_data_address(self, eo_product, band):
def get_data_address(self, eo_product: EOProduct, band: str) -> str:
"""Get the address of a product subdataset.

See :func:`~eodag.api.product.drivers.base.DatasetDriver.get_data_address` to get help on the formal
parameters.
:param eo_product: The product whom underlying dataset address is to be retrieved
:type eo_product: :class:`~eodag.api.product._product.EOProduct`
:param band: The band to retrieve (e.g: 'B01')
:type band: str
:returns: An address for the dataset
:rtype: str
:raises: :class:`~eodag.utils.exceptions.AddressNotFound`
:raises: :class:`~eodag.utils.exceptions.UnsupportedDatasetAddressScheme`
"""
product_location_scheme = eo_product.location.split("://")[0]
if product_location_scheme == "file":
Expand Down
20 changes: 16 additions & 4 deletions eodag_cube/api/product/drivers/sentinel2_l1c.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,9 +15,12 @@
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from __future__ import annotations

import os
import re
import warnings
from typing import TYPE_CHECKING

import boto3
import rasterio
Expand All @@ -26,6 +29,9 @@
from eodag.utils import uri_to_path
from eodag.utils.exceptions import AddressNotFound, UnsupportedDatasetAddressScheme

if TYPE_CHECKING:
from eodag.api.product._product import EOProduct


class Sentinel2L1C(DatasetDriver):
"""Driver for Sentinel2 L1C product"""
Expand All @@ -38,7 +44,7 @@ class Sentinel2L1C(DatasetDriver):
"TCI": ("TCI",),
}

def get_data_address(self, eo_product, band):
def get_data_address(self, eo_product: EOProduct, band: str) -> str:
"""Compute the address of a subdataset for a Sentinel2 L1C product.

The algorithm is as follows for a product on the local filesystem:
Expand All @@ -53,8 +59,14 @@ def get_data_address(self, eo_product, band):
filesystem-like address that matches the band file pattern r'^.+_B01\\.jp2$' if band = 'B01' for
example.

See :func:`~eodag.api.product.drivers.base.DatasetDriver.get_data_address` to get help on the formal
parameters.
:param eo_product: The product whom underlying dataset address is to be retrieved
:type eo_product: :class:`~eodag.api.product._product.EOProduct`
:param band: The band to retrieve (e.g: 'B01')
:type band: str
:returns: An address for the dataset
:rtype: str
:raises: :class:`~eodag.utils.exceptions.AddressNotFound`
:raises: :class:`~eodag.utils.exceptions.UnsupportedDatasetAddressScheme`
"""
product_location_scheme = eo_product.location.split("://")[0]
if product_location_scheme == "file":
Expand All @@ -80,7 +92,7 @@ def get_data_address(self, eo_product, band):
lambda f: band_file_pattern.match(f),
subdataset.files,
):
return os.path.normpath(filename)
return str(os.path.normpath(filename))
raise AddressNotFound
if product_location_scheme == "s3":
access_key, access_secret = eo_product.downloader_auth.authenticate()
Expand Down
22 changes: 17 additions & 5 deletions eodag_cube/api/product/drivers/stac_assets.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,20 +15,32 @@
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from __future__ import annotations

import re
from typing import TYPE_CHECKING

from eodag.api.product.drivers.base import DatasetDriver
from eodag.utils.exceptions import AddressNotFound

if TYPE_CHECKING:
from eodag.api.product._product import EOProduct


class StacAssets(DatasetDriver):
"""Driver for Stac Assets"""

def get_data_address(self, eo_product, band):
def get_data_address(self, eo_product: EOProduct, band: str) -> str:
"""Get the address of a subdataset for a STAC provider product.

See :func:`~eodag.api.product.drivers.base.DatasetDriver.get_data_address` to get help on the formal
parameters.
:param eo_product: The product whom underlying dataset address is to be retrieved
:type eo_product: :class:`~eodag.api.product._product.EOProduct`
:param band: The band to retrieve (e.g: 'B01')
:type band: str
:returns: An address for the dataset
:rtype: str
:raises: :class:`~eodag.utils.exceptions.AddressNotFound`
:raises: :class:`~eodag.utils.exceptions.UnsupportedDatasetAddressScheme`
"""
error_message = ""

Expand All @@ -47,7 +59,7 @@ def get_data_address(self, eo_product, band):
and p.match(s)
]
if len(matching_keys) == 1:
return eo_product.assets[matching_keys[0]]["href"]
return str(eo_product.assets[matching_keys[0]]["href"])
else:
error_message += (
rf"{len(matching_keys)} assets keys found matching {p} AND "
Expand All @@ -68,7 +80,7 @@ def get_data_address(self, eo_product, band):
and p.match(s)
]
if len(matching_keys) == 1:
return eo_product.assets[matching_keys[0]]["href"]
return str(eo_product.assets[matching_keys[0]]["href"])
else:
raise AddressNotFound(
rf"Please adapt given band parameter ('{band}') to match only one asset: {error_message}"
Expand Down
Empty file added eodag_cube/py.typed
Empty file.
8 changes: 8 additions & 0 deletions setup.cfg
Original file line number Diff line number Diff line change
Expand Up @@ -37,3 +37,11 @@ select = D1
add_ignore = D107,D100,D105
# Don't require docstrings for tests or setup
match = (?!test|setup).*\.py

[mypy]
[mypy-boto3.*]
ignore_missing_imports = True
[mypy-rasterio.*]
ignore_missing_imports = True
[mypy-shapely.*]
ignore_missing_imports = True
2 changes: 1 addition & 1 deletion setup.py
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,7 @@
url=metadata["__url__"],
license=metadata["__license__"],
packages=find_packages(exclude=("*.tests", "*.tests.*", "tests.*", "tests")),
package_data={"": ["LICENSE"]},
package_data={"": ["LICENSE", "NOTICE"], "eodag_cube": ["py.typed"]},
include_package_data=True,
install_requires=[
"eodag >= 2.3.2",
Expand Down
Loading