feat: implement Product STAC Extension

docs/api/extensions.rst

@@ -24,6 +24,7 @@ pystac.extensions
     mlm.AssetGeneralMLMExtension
     mlm.AssetDetailedMLMExtension
     pointcloud.PointcloudExtension
+    product.ProductExtension
     projection.ProjectionExtension
     raster.RasterExtension
     render.RenderExtension

docs/api/extensions/product.rst

@@ -0,0 +1,12 @@
+pystac.extensions.product
+=========================
+
+The Product Extension exposes product packaging, timeliness, acquisition type,
+and lifecycle status metadata on Items, Collections, Assets, and Item Asset
+Definitions. Lifecycle status values are represented by
+:class:`~pystac.extensions.product.ProductStatus`; Collection summaries accept
+lists of those values.
+
+.. automodule:: pystac.extensions.product
+   :members:
+   :undoc-members:

docs/conf.py

@@ -256,3 +256,7 @@
     ("py:class", "HREF"),  # this one partially works
     ("py:class", "jsonschema.validators.Draft7Validator"),
 ]
+
+nitpick_ignore_regex = [
+    ("py:class", r"pystac\.extensions\.[^.]+\.T"),
+]

extensions/product/README.md

@@ -0,0 +1,13 @@
+# pystac-ext-product
+
+[PySTAC](https://pypi.org/project/pystac/) extension package for the [Product Extension](https://github.com/stac-extensions/product).
+This extension provides fields for describing product type, timeliness, timeliness category, and acquisition type.
+
+## Supported versions
+
+- [v1.0.0](https://stac-extensions.github.io/product/v1.0.0/schema.json)
+
+## Versioning
+
+This package's version corresponds to the version of the extension specification it targets.
+When we release updates to the package code without changing the target extension version, we use [post releases](https://packaging.python.org/en/latest/discussions/versioning/#post-releases), e.g. `1.0.0.post1`.

extensions/product/pyproject.toml

@@ -0,0 +1,36 @@
+[project]
+name = "pystac-ext-product"
+description = "Product extension for PySTAC"
+readme = "README.md"
+version = "1.0.0"
+authors = []
+maintainers = []
+keywords = ["pystac", "imagery", "raster", "catalog", "STAC", "product"]
+license = { text = "Apache-2.0" }
+classifiers = [
+    "Development Status :: 5 - Production/Stable",
+    "Intended Audience :: Developers",
+    "License :: OSI Approved :: Apache Software License",
+    "Natural Language :: English",
+    "Programming Language :: Python :: 3",
+    "Programming Language :: Python :: 3.10",
+    "Programming Language :: Python :: 3.11",
+    "Programming Language :: Python :: 3.12",
+    "Programming Language :: Python :: 3.13",
+]
+requires-python = ">=3.10"
+dependencies = ["pystac-core"]
+
+[project.urls]
+Documentation = "https://pystac.readthedocs.io"
+Repository = "https://github.com/stac-utils/pystac"
+Issues = "https://github.com/stac-utils/pystac/issues"
+Changelog = "https://github.com/stac-utils/pystac/blob/main/CHANGELOG.md"
+Discussions = "https://github.com/radiantearth/stac-spec/discussions/categories/stac-software"
+
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"
+
+[tool.hatch.build.targets.wheel]
+packages = ["pystac"]

extensions/product/pystac/extensions/product.py

@@ -0,0 +1,283 @@
+"""Implements the :stac-ext:`Product Extension <product>`."""
+
+from __future__ import annotations
+
+from collections.abc import Iterable
+from typing import Any, Generic, Literal, TypeVar, cast
+
+import pystac
+from pystac.extensions.base import (
+    ExtensionManagementMixin,
+    PropertiesExtension,
+    SummariesExtension,
+)
+from pystac.extensions.hooks import ExtensionHooks
+from pystac.utils import StringEnum
+
+T = TypeVar(
+    "T",
+    pystac.Collection,
+    pystac.Item,
+    pystac.Asset,
+    pystac.ItemAssetDefinition,
+)
+
+SCHEMA_URI: str = "https://stac-extensions.github.io/product/v1.0.0/schema.json"
+
+PREFIX: str = "product:"
+TYPE_PROP: str = PREFIX + "type"
+TIMELINESS_PROP: str = PREFIX + "timeliness"
+TIMELINESS_CATEGORY_PROP: str = PREFIX + "timeliness_category"
+ACQUISITION_TYPE_PROP: str = PREFIX + "acquisition_type"
+STATUS_PROP: str = PREFIX + "status"
+
+
+class AcquisitionType(StringEnum):
+    """Allowed values for product:acquisition_type."""
+
+    NOMINAL = "nominal"
+    CALIBRATION = "calibration"
+    OTHER = "other"
+
+
+class ProductStatus(StringEnum):
+    """Allowed values for :data:`product:status <STATUS_PROP>`."""
+
+    ARCHIVED = "archived"
+    ACQUIRED = "acquired"
+    CANCELLED = "cancelled"
+    FAILED = "failed"
+    PLANNED = "planned"
+    POTENTIAL = "potential"
+    REJECTED = "rejected"
+    QUALITY_DEGRADED = "qualitydegraded"
+
+
+class ProductExtension(
+    Generic[T],
+    PropertiesExtension,
+    ExtensionManagementMixin[pystac.Collection | pystac.Item],
+):
+    name: Literal["product"] = "product"
+
+    def apply(
+        self,
+        *,
+        product_type: str | None = None,
+        timeliness: str | None = None,
+        timeliness_category: str | None = None,
+        acquisition_type: AcquisitionType | str | None = None,
+        status: ProductStatus | None = None,
+    ) -> None:
+        if (
+            timeliness_category is not None
+            and timeliness is None
+            and self.timeliness is None
+        ):
+            raise ValueError(
+                f"'{TIMELINESS_CATEGORY_PROP}' requires '{TIMELINESS_PROP}' to be set."
+            )
+
+        self.product_type = product_type
+        if timeliness is not None:
+            self.timeliness = timeliness
+        if timeliness_category is not None:
+            self.timeliness_category = timeliness_category
+        self.acquisition_type = acquisition_type
+        self.status = status
+
+    @property
+    def product_type(self) -> str | None:
+        return self._get_property(TYPE_PROP, str)
+
+    @product_type.setter
+    def product_type(self, v: str | None) -> None:
+        self._set_property(TYPE_PROP, v)
+
+    @property
+    def timeliness(self) -> str | None:
+        return self._get_property(TIMELINESS_PROP, str)
+
+    @timeliness.setter
+    def timeliness(self, v: str | None) -> None:
+        self._set_property(TIMELINESS_PROP, v)
+
+    @property
+    def timeliness_category(self) -> str | None:
+        return self._get_property(TIMELINESS_CATEGORY_PROP, str)
+
+    @timeliness_category.setter
+    def timeliness_category(self, v: str | None) -> None:
+        if v is not None and self.timeliness is None:
+            raise ValueError(
+                f"'{TIMELINESS_CATEGORY_PROP}' requires '{TIMELINESS_PROP}' to be set."
+            )
+        self._set_property(TIMELINESS_CATEGORY_PROP, v)
+
+    @property
+    def acquisition_type(self) -> AcquisitionType | str | None:
+        raw = self._get_property(ACQUISITION_TYPE_PROP, str)
+        if raw is None:
+            return None
+        try:
+            return AcquisitionType(raw)
+        except Exception:
+            return raw
+
+    @acquisition_type.setter
+    def acquisition_type(self, v: AcquisitionType | str | None) -> None:
+        self._set_property(ACQUISITION_TYPE_PROP, None if v is None else str(v))
+
+    @property
+    def status(self) -> ProductStatus | None:
+        """Gets the product lifecycle status."""
+        raw = self._get_property(STATUS_PROP, str)
+        return None if raw is None else ProductStatus(raw)
+
+    @status.setter
+    def status(self, v: ProductStatus | None) -> None:
+        """Sets the product lifecycle status.
+
+        Args:
+            v: The product lifecycle status.
+        """
+        self._set_property(STATUS_PROP, None if v is None else v.value)
+
+    @classmethod
+    def get_schema_uri(cls) -> str:
+        return SCHEMA_URI
+
+    @classmethod
+    def ext(cls, obj: T, add_if_missing: bool = False) -> ProductExtension[T]:
+        if isinstance(obj, pystac.Collection):
+            cls.ensure_has_extension(obj, add_if_missing)
+            return cast(ProductExtension[T], CollectionProductExtension(obj))
+        if isinstance(obj, pystac.Item):
+            cls.ensure_has_extension(obj, add_if_missing)
+            return cast(ProductExtension[T], ItemProductExtension(obj))
+        if isinstance(obj, pystac.Asset):
+            cls.ensure_owner_has_extension(obj, add_if_missing)
+            return cast(ProductExtension[T], AssetProductExtension(obj))
+        if isinstance(obj, pystac.ItemAssetDefinition):
+            cls.ensure_owner_has_extension(obj, add_if_missing)
+            return cast(ProductExtension[T], ItemAssetsProductExtension(obj))
+        raise pystac.ExtensionTypeError(cls._ext_error_message(obj))
+
+    @classmethod
+    def summaries(
+        cls, obj: pystac.Collection, add_if_missing: bool = False
+    ) -> SummariesProductExtension:
+        cls.ensure_has_extension(obj, add_if_missing)
+        return SummariesProductExtension(obj)
+
+
+class CollectionProductExtension(ProductExtension[pystac.Collection]):
+    collection: pystac.Collection
+    properties: dict[str, Any]
+
+    def __init__(self, collection: pystac.Collection):
+        self.collection = collection
+        self.properties = collection.extra_fields
+
+    def __repr__(self) -> str:
+        return f"<CollectionProductExtension Collection id={self.collection.id}>"
+
+
+class ItemProductExtension(ProductExtension[pystac.Item]):
+    item: pystac.Item
+    properties: dict[str, Any]
+
+    def __init__(self, item: pystac.Item):
+        self.item = item
+        self.properties = item.properties
+
+    def __repr__(self) -> str:
+        return f"<ItemProductExtension Item id={self.item.id}>"
+
+
+class AssetProductExtension(ProductExtension[pystac.Asset]):
+    asset: pystac.Asset
+    asset_href: str
+    properties: dict[str, Any]
+    additional_read_properties: Iterable[dict[str, Any]] | None
+
+    def __init__(self, asset: pystac.Asset):
+        self.asset = asset
+        self.asset_href = asset.href
+        self.properties = asset.extra_fields
+        if asset.owner and isinstance(asset.owner, pystac.Item):
+            self.additional_read_properties = [asset.owner.properties]
+        else:
+            self.additional_read_properties = None
+
+    def __repr__(self) -> str:
+        return f"<AssetProductExtension Asset href={self.asset_href}>"
+
+
+class ItemAssetsProductExtension(ProductExtension[pystac.ItemAssetDefinition]):
+    asset_defn: pystac.ItemAssetDefinition
+    properties: dict[str, Any]
+
+    def __init__(self, item_asset: pystac.ItemAssetDefinition):
+        self.asset_defn = item_asset
+        self.properties = item_asset.properties
+
+    def __repr__(self) -> str:
+        return "<ItemAssetsProductExtension ItemAssetDefinition>"
+
+
+class SummariesProductExtension(SummariesExtension):
+    @property
+    def product_type(self) -> list[str] | None:
+        return self.summaries.get_list(TYPE_PROP)
+
+    @product_type.setter
+    def product_type(self, v: list[str] | None) -> None:
+        self._set_summary(TYPE_PROP, v)
+
+    @property
+    def timeliness(self) -> list[str] | None:
+        return self.summaries.get_list(TIMELINESS_PROP)
+
+    @timeliness.setter
+    def timeliness(self, v: list[str] | None) -> None:
+        self._set_summary(TIMELINESS_PROP, v)
+
+    @property
+    def timeliness_category(self) -> list[str] | None:
+        return self.summaries.get_list(TIMELINESS_CATEGORY_PROP)
+
+    @timeliness_category.setter
+    def timeliness_category(self, v: list[str] | None) -> None:
+        self._set_summary(TIMELINESS_CATEGORY_PROP, v)
+
+    @property
+    def acquisition_type(self) -> list[str] | None:
+        return self.summaries.get_list(ACQUISITION_TYPE_PROP)
+
+    @acquisition_type.setter
+    def acquisition_type(self, v: list[str] | None) -> None:
+        self._set_summary(ACQUISITION_TYPE_PROP, v)
+
+    @property
+    def status(self) -> list[ProductStatus] | None:
+        """Get the product lifecycle statuses from the summaries."""
+        return self.summaries.get_list(STATUS_PROP)
+
+    @status.setter
+    def status(self, v: list[ProductStatus] | None) -> None:
+        """Set the product lifecycle statuses in the summaries.
+
+        Args:
+            v: The product lifecycle statuses.
+        """
+        self._set_summary(STATUS_PROP, v)
+
+
+class ProductExtensionHooks(ExtensionHooks):
+    schema_uri: str = SCHEMA_URI
+    prev_extension_ids = {"product"}
+    stac_object_types = {pystac.STACObjectType.COLLECTION, pystac.STACObjectType.ITEM}
+
+
+PRODUCT_EXTENSION_HOOKS: ExtensionHooks = ProductExtensionHooks()

extensions/product/pystac/extensions/py.typed

@@ -0,0 +1 @@
+