diff --git a/setup.cfg b/setup.cfg index 40d7452..db95c83 100644 --- a/setup.cfg +++ b/setup.cfg @@ -22,7 +22,7 @@ classifiers = Programming Language :: Python :: 3.9 [options] -python_requires = >=3.8 +python_requires = >=3.8,<3.10 package_dir = =src packages = find_namespace: diff --git a/src/stactools/naip/grid.py b/src/stactools/naip/grid.py new file mode 100644 index 0000000..13bd556 --- /dev/null +++ b/src/stactools/naip/grid.py @@ -0,0 +1,107 @@ +"""Implements the :stac-ext:`Grid Extension `.""" + +import re +from typing import Any, Dict, Optional, Pattern, Set, Union, cast + +import pystac +from pystac.extensions.base import ExtensionManagementMixin, PropertiesExtension +from pystac.extensions.hooks import ExtensionHooks + +SCHEMA_URI: str = "https://stac-extensions.github.io/grid/v1.0.0/schema.json" +PREFIX: str = "grid:" + +# Field names +CODE_PROP: str = PREFIX + "code" # required + +CODE_REGEX: str = r"[A-Z]+-[-_.A-Za-z0-9]+" +CODE_PATTERN: Pattern[str] = re.compile(CODE_REGEX) + + +def validated_code(v: str) -> str: + if not isinstance(v, str): + raise ValueError("Invalid Grid code: must be str") + if not CODE_PATTERN.fullmatch(v): + raise ValueError( + f"Invalid Grid code: {v}" f" does not match the regex {CODE_REGEX}" + ) + return v + + +class GridExtension( + PropertiesExtension, + ExtensionManagementMixin[Union[pystac.Item, pystac.Collection]], +): + """A concrete implementation of :class:`GridExtension` on an :class:`~pystac.Item` + that extends the properties of the Item to include properties defined in the + :stac-ext:`Grid Extension `. + + This class should generally not be instantiated directly. Instead, call + :meth:`GridExtension.ext` on an :class:`~pystac.Item` to extend it. + + .. code-block:: python + + >>> item: pystac.Item = ... + >>> proj_ext = GridExtension.ext(item) + """ + + item: pystac.Item + """The :class:`~pystac.Item` being extended.""" + + properties: Dict[str, Any] + """The :class:`~pystac.Item` properties, including extension properties.""" + + def __init__(self, item: pystac.Item): + self.item = item + self.properties = item.properties + + def __repr__(self) -> str: + return "".format(self.item.id) + + def apply(self, code: str) -> None: + """Applies Grid extension properties to the extended Item. + + Args: + code : REQUIRED. The code of the Item's grid location. + """ + self.code = validated_code(code) + + @property + def code(self) -> Optional[str]: + """Get or sets the latitude band of the datasource.""" + return self._get_property(CODE_PROP, str) + + @code.setter + def code(self, v: str) -> None: + self._set_property(CODE_PROP, validated_code(v), pop_if_none=False) + + @classmethod + def get_schema_uri(cls) -> str: + return SCHEMA_URI + + @classmethod + def ext(cls, obj: pystac.Item, add_if_missing: bool = False) -> "GridExtension": + """Extends the given STAC Object with properties from the :stac-ext:`Grid + Extension `. + + This extension can be applied to instances of :class:`~pystac.Item`. + + Raises: + + pystac.ExtensionTypeError : If an invalid object type is passed. + """ + if isinstance(obj, pystac.Item): + cls.validate_has_extension(obj, add_if_missing) + return cast(GridExtension, GridExtension(obj)) + else: + raise pystac.ExtensionTypeError( + f"Grid Extension does not apply to type '{type(obj).__name__}'" + ) + + +class GridExtensionHooks(ExtensionHooks): + schema_uri: str = SCHEMA_URI + prev_extension_ids: Set[str] = set() + stac_object_types = {pystac.STACObjectType.ITEM} + + +Grid_EXTENSION_HOOKS: ExtensionHooks = GridExtensionHooks() diff --git a/src/stactools/naip/stac.py b/src/stactools/naip/stac.py index 383d47a..4661b2b 100644 --- a/src/stactools/naip/stac.py +++ b/src/stactools/naip/stac.py @@ -1,5 +1,6 @@ import os -from typing import List, Optional +import re +from typing import Final, List, Optional, Pattern import dateutil.parser import pystac @@ -13,8 +14,11 @@ from stactools.core.projection import reproject_geom from stactools.naip import constants +from stactools.naip.grid import GridExtension from stactools.naip.utils import parse_fgdc_metadata +DOQQ_PATTERN: Final[Pattern[str]] = re.compile(r"[A-Za-z]{2}_m_(\d{7})_(ne|se|nw|sw)_") + def naip_item_id(state, resource_name): """Generates a STAC Item ID based on the state and the "Resource Description" @@ -149,16 +153,21 @@ def create_item( item.common_metadata.providers.extend(additional_providers) item.common_metadata.gsd = gsd - # eo, for asset bands + # EO Extension, for asset bands EOExtension.add_to(item) - # proj + # Projection Extension projection = ProjectionExtension.ext(item, add_if_missing=True) projection.epsg = epsg projection.shape = image_shape projection.bbox = original_bbox projection.transform = transform + # Grid Extension + grid = GridExtension.ext(item, add_if_missing=True) + if match := DOQQ_PATTERN.search(item_id): + grid.code = f"DOQQ-{match.group(1)}{match.group(2).upper()}" + # COG item.add_asset( "image",