feat: add cf extension

CHANGELOG.md

@@ -23,6 +23,7 @@
 * `Asset.get_absolute_href()` now properly resolves root relative hrefs ([#1599](https://github.com/stac-utils/pystac/pull/1599))
 * Clone extra fields on `Item` ([#1601](https://github.com/stac-utils/pystac/pull/1601))
 
+
 ## [v1.14.1] - 2025-09-18
 
 ### Fixed

core/pystac/__init__.py

@@ -91,6 +91,7 @@
 from pystac.utils import HREF
 
 import pystac.extensions.hooks
+import pystac.extensions.cf
 import pystac.extensions.classification
 import pystac.extensions.datacube
 import pystac.extensions.eo
@@ -118,6 +119,7 @@
 
 EXTENSION_HOOKS = pystac.extensions.hooks.RegisteredExtensionHooks(
     [
+        pystac.extensions.cf.CF_EXTENSION_HOOKS,
         pystac.extensions.classification.CLASSIFICATION_EXTENSION_HOOKS,
         pystac.extensions.datacube.DATACUBE_EXTENSION_HOOKS,
         pystac.extensions.eo.EO_EXTENSION_HOOKS,

core/pystac/extensions/ext.py

@@ -12,6 +12,7 @@
     Link,
     STACError,
 )
+from pystac.extensions.cf import CFExtension
 from pystac.extensions.classification import ClassificationExtension
 from pystac.extensions.datacube import DatacubeExtension
 from pystac.extensions.eo import EOExtension
@@ -46,6 +47,7 @@
 U = TypeVar("U", Asset, ItemAssetDefinition)
 
 EXTENSION_NAMES = Literal[
+    "cf",
     "classification",
     "cube",
     "eo",
@@ -70,6 +72,7 @@
 ]
 
 EXTENSION_NAME_MAPPING: dict[EXTENSION_NAMES, Any] = {
+    CFExtension.name: CFExtension,
     ClassificationExtension.name: ClassificationExtension,
     DatacubeExtension.name: DatacubeExtension,
     EOExtension.name: EOExtension,
@@ -152,6 +155,10 @@ class CollectionExt(CatalogExt):
 
     stac_object: Collection
 
+    @property
+    def cf(self) -> CFExtension[Collection]:
+        return CFExtension.ext(self.stac_object)
+
     @property
     def cube(self) -> DatacubeExtension[Collection]:
         return DatacubeExtension.ext(self.stac_object)
@@ -220,6 +227,10 @@ def remove(self, name: EXTENSION_NAMES) -> None:
         """
         _get_class_by_name(name).remove_from(self.stac_object)
 
+    @property
+    def cf(self) -> CFExtension[Item]:
+        return CFExtension.ext(self.stac_object)
+
     @property
     def classification(self) -> ClassificationExtension[Item]:
         return ClassificationExtension.ext(self.stac_object)
@@ -347,6 +358,10 @@ def remove(self, name: EXTENSION_NAMES) -> None:
 class _AssetExt(_AssetsExt[U]):
     stac_object: U
 
+    @property
+    def cf(self) -> CFExtension[U]:
+        return CFExtension.ext(self.stac_object)
+
     @property
     def classification(self) -> ClassificationExtension[U]:
         return ClassificationExtension.ext(self.stac_object)

docs/api/extensions.rst

@@ -13,6 +13,7 @@ pystac.extensions
 
 .. autosummary::
 
+    cf.CFExtension
     classification.ClassificationExtension
     datacube.DatacubeExtension
     eo.EOExtension

docs/api/extensions/cf.rst

@@ -0,0 +1,6 @@
+pystac.extensions.cf
+====================
+
+.. automodule:: pystac.extensions.cf
+   :members:
+   :undoc-members:

extensions/cf/README.md

@@ -0,0 +1,13 @@
+# pystac-ext-cf
+
+[PySTAC](https://pypi.org/project/pystac/) extension package for the [CF (Climate and Forecast) Extension](https://github.com/stac-extensions/cf).
+This extension adds Climate and Forecast (CF) Convention metadata to STAC documents, enabling standardized descriptions of geophysical and climate data variables.
+
+## Supported versions
+
+- [v0.2.0](https://stac-extensions.github.io/cf/v0.2.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. `0.2.0.post1`.

extensions/cf/pyproject.toml

@@ -0,0 +1,36 @@
+[project]
+name = "pystac-ext-cf"
+description = "CF (Climate and Forecast) extension for PySTAC"
+readme = "README.md"
+version = "0.2.0"
+authors = []
+maintainers = []
+keywords = ["pystac", "imagery", "raster", "catalog", "STAC", "cf", "climate", "forecast"]
+license = { text = "Apache-2.0" }
+classifiers = [
+    "Development Status :: 3 - Alpha",
+    "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/cf/pystac/extensions/cf.py

@@ -0,0 +1,239 @@
+"""Implements the CF (Climate and Forecast) Extension.
+
+https://github.com/stac-extensions/cf
+"""
+
+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
+from pystac.extensions.hooks import ExtensionHooks
+from pystac.utils import map_opt
+
+#: Generalized version of :class:`~pystac.Collection`, :class:`~pystac.Item`,
+#: :class:`~pystac.Asset`, or :class:`~pystac.ItemAssetDefinition`
+T = TypeVar(
+    "T", pystac.Collection, pystac.Item, pystac.Asset, pystac.ItemAssetDefinition
+)
+
+SCHEMA_URI: str = "https://stac-extensions.github.io/cf/v0.2.0/schema.json"
+PREFIX: str = "cf:"
+
+# Field names
+PARAMETER_PROP: str = PREFIX + "parameter"
+
+
+class Parameter:
+    """Represents a CF parameter with a name and optional unit."""
+
+    name: str
+    unit: str | None
+
+    def __init__(self, name: str, unit: str | None = None) -> None:
+        self.name = name
+        self.unit = unit
+
+    def __eq__(self, other: Any) -> bool:
+        if not isinstance(other, Parameter):
+            return NotImplemented
+        return self.name == other.name and self.unit == other.unit
+
+    def __repr__(self) -> str:
+        return f"<Parameter name={self.name} unit={self.unit}>"
+
+    def to_dict(self) -> dict[str, str | None]:
+        return {"name": self.name, "unit": self.unit}
+
+    @staticmethod
+    def from_dict(d: dict[str, Any]) -> Parameter:
+        name = d.get("name")
+        if not name:
+            raise ValueError("'name' must be a non-empty string.")
+        return Parameter(name, d.get("unit"))
+
+
+class CFExtension(
+    Generic[T],
+    PropertiesExtension,
+    ExtensionManagementMixin[pystac.Item | pystac.Collection],
+):
+    """An abstract class that can be used to extend the properties of an
+    :class:`~pystac.Asset`, :class:`~pystac.Item`, or a :class:`pystac.Collection`
+    with properties from the :stac-ext:`CF Extension <cf>`.
+    This class is generic over the type of STAC Object to be extended
+    (e.g. :class:`~pystac.Item`, :class:`~pystac.Collection`).
+
+    To create a concrete instance of :class:`CFExtension`, use the
+    :meth:`CFExtension.ext` method. For example:
+
+    .. code-block:: python
+
+       >>> item: pystac.Item = ...
+       >>> cf_ext = CFExtension.ext(item)
+    """
+
+    name: Literal["cf"] = "cf"
+
+    def apply(
+        self,
+        parameters: list[Parameter] | None = None,
+    ) -> None:
+        """Apply CF Extension properties to the extended :class:`~pystac.Asset`,
+        :class:`~pystac.Item`, or :class:`~pystac.Collection`.
+
+        Args:
+            parameters: List of CF parameters describing the physical quantities
+                in this object.
+        """
+        self.parameters = parameters
+
+    @property
+    def parameters(self) -> list[Parameter] | None:
+        """Get or set the list of CF parameters."""
+        return map_opt(
+            lambda params: [Parameter.from_dict(p) for p in params],
+            self._get_property(PARAMETER_PROP, list[dict[str, Any]]),
+        )
+
+    @parameters.setter
+    def parameters(self, v: list[Parameter] | None) -> None:
+        self._set_property(
+            PARAMETER_PROP,
+            map_opt(lambda params: [p.to_dict() for p in params], v),
+        )
+
+    @classmethod
+    def get_schema_uri(cls) -> str:
+        """Return this extension's schema URI."""
+        return SCHEMA_URI
+
+    @classmethod
+    def ext(cls, obj: T, add_if_missing: bool = False) -> CFExtension[T]:
+        """Extend the given STAC Object with properties from the
+        :stac-ext:`CF Extension <cf>`.
+
+        This extension can be applied to instances of :class:`~pystac.Item`,
+        :class:`~pystac.Asset`, :class:`~pystac.Collection`, or
+        :class:`~pystac.ItemAssetDefinition`.
+
+        Raises
+        ------
+            pystac.ExtensionTypeError : If an invalid object type is passed.
+        """
+        if isinstance(obj, pystac.Collection):
+            cls.ensure_has_extension(obj, add_if_missing)
+            return cast(CFExtension[T], CollectionCFExtension(obj))
+        elif isinstance(obj, pystac.Item):
+            cls.ensure_has_extension(obj, add_if_missing)
+            return cast(CFExtension[T], ItemCFExtension(obj))
+        elif isinstance(obj, pystac.Asset):
+            cls.ensure_owner_has_extension(obj, add_if_missing)
+            return cast(CFExtension[T], AssetCFExtension(obj))
+        elif isinstance(obj, pystac.ItemAssetDefinition):
+            cls.ensure_owner_has_extension(obj, add_if_missing)
+            return cast(CFExtension[T], ItemAssetsCFExtension(obj))
+        else:
+            raise pystac.ExtensionTypeError(cls._ext_error_message(obj))
+
+
+class ItemCFExtension(CFExtension[pystac.Item]):
+    """A concrete implementation of :class:`CFExtension` on an
+    :class:`~pystac.Item`.
+
+    This class should generally not be instantiated directly. Instead, call
+    :meth:`CFExtension.ext` on an :class:`~pystac.Item` to extend it.
+    """
+
+    item: pystac.Item
+    properties: dict[str, Any]
+
+    def __init__(self, item: pystac.Item) -> None:
+        self.item = item
+        self.properties = item.properties
+
+    def __repr__(self) -> str:
+        return f"<ItemCFExtension Item id={self.item.id}>"
+
+
+class CollectionCFExtension(CFExtension[pystac.Collection]):
+    """A concrete implementation of :class:`CFExtension` on a
+    :class:`~pystac.Collection`.
+
+    This class should generally not be instantiated directly. Instead, call
+    :meth:`CFExtension.ext` on a :class:`~pystac.Collection` to extend it.
+    """
+
+    collection: pystac.Collection
+    properties: dict[str, Any]
+
+    def __init__(self, collection: pystac.Collection) -> None:
+        self.collection = collection
+        self.properties = collection.extra_fields
+
+    def __repr__(self) -> str:
+        return f"<CollectionCFExtension Collection id={self.collection.id}>"
+
+
+class AssetCFExtension(CFExtension[pystac.Asset]):
+    """A concrete implementation of :class:`CFExtension` on an
+    :class:`~pystac.Asset`.
+
+    This class should generally not be instantiated directly. Instead, call
+    :meth:`CFExtension.ext` on an :class:`~pystac.Asset` to extend it.
+    """
+
+    asset_href: str
+    """The ``href`` value of the :class:`~pystac.Asset` being extended."""
+
+    properties: dict[str, Any]
+    """The :class:`~pystac.Asset` fields, including extension properties."""
+
+    additional_read_properties: Iterable[dict[str, Any]] | None = None
+    """If present, a list containing 1 dictionary representing the properties of
+    the owning :class:`~pystac.Item`."""
+
+    def __init__(self, asset: pystac.Asset) -> None:
+        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]
+
+    def __repr__(self) -> str:
+        return f"<AssetCFExtension Asset href={self.asset_href}>"
+
+
+class ItemAssetsCFExtension(CFExtension[pystac.ItemAssetDefinition]):
+    """A concrete implementation of :class:`CFExtension` on an
+    :class:`~pystac.ItemAssetDefinition`.
+
+    This class should generally not be instantiated directly. Instead, call
+    :meth:`CFExtension.ext` on an :class:`~pystac.ItemAssetDefinition` to extend it.
+    """
+
+    asset_defn: pystac.ItemAssetDefinition
+    properties: dict[str, Any]
+
+    def __init__(self, item_asset: pystac.ItemAssetDefinition) -> None:
+        self.asset_defn = item_asset
+        self.properties = item_asset.properties
+
+
+class CFExtensionHooks(ExtensionHooks):
+    schema_uri: str = SCHEMA_URI
+    prev_extension_ids: set[str] = set()
+    stac_object_types = {
+        pystac.STACObjectType.COLLECTION,
+        pystac.STACObjectType.ITEM,
+    }
+
+
+CF_EXTENSION_HOOKS: ExtensionHooks = CFExtensionHooks()