Skip to main content
`Ctrl`+`K`

# Source code for pystac.collection

    from __future__ import annotations

    import warnings
    from collections.abc import Iterable
    from copy import deepcopy
    from datetime import datetime, timezone
    from typing import (
        TYPE_CHECKING,
        Any,
        Optional,
        TypeVar,
        cast,
    )

    from dateutil import tz

    import pystac
    from pystac import CatalogType, STACObjectType
    from pystac.asset import Asset, Assets
    from pystac.catalog import Catalog
    from pystac.errors import DeprecatedWarning, ExtensionNotImplemented, STACTypeError
    from pystac.item_assets import ItemAssetDefinition, _ItemAssets
    from pystac.layout import HrefLayoutStrategy
    from pystac.link import Link
    from pystac.provider import Provider
    from pystac.serialization import (
        identify_stac_object,
        identify_stac_object_type,
        migrate_to_latest,
    )
    from pystac.summaries import Summaries
    from pystac.utils import (
        datetime_to_str,
        str_to_datetime,
    )

    if TYPE_CHECKING:
        from pystac.extensions.ext import CollectionExt
        from pystac.item import Item

    #: Generalized version of :class:`Collection`
    C = TypeVar("C", bound="Collection")

    Bboxes = list[list[float | int]]
    TemporalIntervals = list[list[datetime]] | list[list[Optional[datetime]]]
    TemporalIntervalsLike = TemporalIntervals | list[datetime] | list[Optional[datetime]]

    class SpatialExtent:
        """Describes the spatial extent of a Collection.

        Args:
            bboxes : A list of bboxes that represent the spatial
                extent of the collection. Each bbox can be 2D or 3D. The length of the bbox
                array must be 2*n where n is the number of dimensions. For example, a
                2D Collection with only one bbox would be [[xmin, ymin, xmax, ymax]]

            extra_fields : Dictionary containing additional top-level fields defined on the
                Spatial Extent object.
        """

        bboxes: Bboxes
        """A list of bboxes that represent the spatial
        extent of the collection. Each bbox can be 2D or 3D. The length of the bbox
        array must be 2*n where n is the number of dimensions. For example, a
        2D Collection with only one bbox would be [[xmin, ymin, xmax, ymax]]"""

        extra_fields: dict[str, Any]
        """Dictionary containing additional top-level fields defined on the Spatial
        Extent object."""

        def __init__(
            self,
            bboxes: Bboxes | list[float | int],
            extra_fields: dict[str, Any] | None = None,
        ) -> None:
            if not isinstance(bboxes, list):
                raise TypeError("bboxes must be a list")

            # A common mistake is to pass in a single bbox instead of a list of bboxes.
            # Account for this by transforming the input in that case.
            if isinstance(bboxes[0], (float, int)):
                self.bboxes = [cast(list[float | int], bboxes)]
            else:
                self.bboxes = cast(Bboxes, bboxes)

            self.extra_fields = extra_fields or {}

        def to_dict(self) -> dict[str, Any]:
            """Returns this spatial extent as a dictionary.

            Returns:
                dict: A serialization of the SpatialExtent.
            """
            d = {"bbox": self.bboxes, **self.extra_fields}
            return d

        def clone(self) -> SpatialExtent:
            """Clones this object.

            Returns:
                SpatialExtent: The clone of this object.
            """
            cls = self.__class__
            return cls(
                bboxes=deepcopy(self.bboxes), extra_fields=deepcopy(self.extra_fields)
            )

        @staticmethod
        def from_dict(d: dict[str, Any]) -> SpatialExtent:
            """Constructs a SpatialExtent from a dict.

            Returns:
                SpatialExtent: The SpatialExtent deserialized from the JSON dict.
            """
            return SpatialExtent(
                bboxes=d["bbox"], extra_fields={k: v for k, v in d.items() if k != "bbox"}
            )

        @staticmethod
        def from_coordinates(
            coordinates: list[Any], extra_fields: dict[str, Any] | None = None
        ) -> SpatialExtent:
            """Constructs a SpatialExtent from a set of coordinates.

            This method will only produce a single bbox that covers all points
            in the coordinate set.

            Args:
                coordinates : Coordinates to derive the bbox from.
                extra_fields : Dictionary containing additional top-level fields defined on
                    the SpatialExtent object.

            Returns:
                SpatialExtent: A SpatialExtent with a single bbox that covers the
                given coordinates.
            """

            def process_coords(
                coord_lists: list[Any],
                xmin: float | None = None,
                ymin: float | None = None,
                xmax: float | None = None,
                ymax: float | None = None,
            ) -> tuple[float | None, float | None, float | None, float | None]:
                for coord in coord_lists:
                    if isinstance(coord[0], list):
                        xmin, ymin, xmax, ymax = process_coords(
                            coord, xmin, ymin, xmax, ymax
                        )
                    else:
                        x, y = coord
                        if xmin is None or x < xmin:
                            xmin = x
                        elif xmax is None or xmax < x:
                            xmax = x
                        if ymin is None or y < ymin:
                            ymin = y
                        elif ymax is None or ymax < y:
                            ymax = y
                return xmin, ymin, xmax, ymax

            xmin, ymin, xmax, ymax = process_coords(coordinates)

            if xmin is None or ymin is None or xmax is None or ymax is None:
                raise ValueError(
                    f"Could not determine bounds from coordinate sequence {coordinates}"
                )

            return SpatialExtent(
                bboxes=[[xmin, ymin, xmax, ymax]], extra_fields=extra_fields
            )

    class TemporalExtent:
        """Describes the temporal extent of a Collection.

        Args:
            intervals :  A list of two datetimes wrapped in a list,
                representing the temporal extent of a Collection. Open date ranges are
                supported by setting either the start (the first element of the interval)
                or the end (the second element of the interval) to None.

            extra_fields : Dictionary containing additional top-level fields defined on the
                Temporal Extent object.
        Note:
            Datetimes are required to be in UTC.
        """

        intervals: TemporalIntervals
        """A list of two datetimes wrapped in a list,
        representing the temporal extent of a Collection. Open date ranges are
        represented by either the start (the first element of the interval) or the
        end (the second element of the interval) being None."""

        extra_fields: dict[str, Any]
        """Dictionary containing additional top-level fields defined on the Temporal
        Extent object."""

        def __init__(
            self,
            intervals: TemporalIntervals | list[datetime | None],
            extra_fields: dict[str, Any] | None = None,
        ):
            if not isinstance(intervals, list):
                raise TypeError("intervals must be a list")
            # A common mistake is to pass in a single interval instead of a
            # list of intervals. Account for this by transforming the input
            # in that case.
            if isinstance(intervals[0], datetime) or intervals[0] is None:
                self.intervals = [cast(list[Optional[datetime]], intervals)]
            else:
                self.intervals = cast(TemporalIntervals, intervals)

            self.extra_fields = extra_fields or {}

        def to_dict(self) -> dict[str, Any]:
            """Returns this temporal extent as a dictionary.

            Returns:
                dict: A serialization of the TemporalExtent.
            """
            encoded_intervals: list[list[str | None]] = []
            for i in self.intervals:
                start = None
                end = None

                if i[0] is not None:
                    start = datetime_to_str(i[0])

                if i[1] is not None:
                    end = datetime_to_str(i[1])

                encoded_intervals.append([start, end])

            d = {"interval": encoded_intervals, **self.extra_fields}
            return d

        def clone(self) -> TemporalExtent:
            """Clones this object.

            Returns:
                TemporalExtent: The clone of this object.
            """
            cls = self.__class__
            return cls(
                intervals=deepcopy(self.intervals), extra_fields=deepcopy(self.extra_fields)
            )

        @staticmethod
        def from_dict(d: dict[str, Any]) -> TemporalExtent:
            """Constructs an TemporalExtent from a dict.

            Returns:
                TemporalExtent: The TemporalExtent deserialized from the JSON dict.
            """
            parsed_intervals: list[list[datetime | None]] = []
            for i in d["interval"]:
                if isinstance(i, str):
                    # d["interval"] is a list of strings, so we correct the list and
                    # try again
                    # https://github.com/stac-utils/pystac/issues/1221
                    warnings.warn(
                        "A collection's temporal extent should be a list of lists, but "
                        "is instead a "
                        "list of strings. pystac is fixing this issue and continuing "
                        "deserialization, but note that the source "
                        "collection is invalid STAC.",
                        UserWarning,
                    )
                    d["interval"] = [d["interval"]]
                    return TemporalExtent.from_dict(d)
                start = None
                end = None

                if i[0]:
                    start = str_to_datetime(i[0])
                if i[1]:
                    end = str_to_datetime(i[1])
                parsed_intervals.append([start, end])

            return TemporalExtent(
                intervals=parsed_intervals,
                extra_fields={k: v for k, v in d.items() if k != "interval"},
            )

        @staticmethod
        def from_now() -> TemporalExtent:
            """Constructs an TemporalExtent with a single open interval that has
            the start time as the current time.

            Returns:
                TemporalExtent: The resulting TemporalExtent.
            """
            return TemporalExtent(
                intervals=[[datetime.now(timezone.utc).replace(microsecond=0), None]]
            )

    class Extent:
        """Describes the spatiotemporal extents of a Collection.

        Args:
            spatial : Potential spatial extent covered by the collection.
            temporal : Potential temporal extent covered by the collection.
            extra_fields : Dictionary containing additional top-level fields defined on the
                Extent object.
        """

        spatial: SpatialExtent
        """Potential spatial extent covered by the collection."""

        temporal: TemporalExtent
        """Potential temporal extent covered by the collection."""

        extra_fields: dict[str, Any]
        """Dictionary containing additional top-level fields defined on the Extent
        object."""

        def __init__(
            self,
            spatial: SpatialExtent,
            temporal: TemporalExtent,
            extra_fields: dict[str, Any] | None = None,
        ):
            self.spatial = spatial
            self.temporal = temporal
            self.extra_fields = extra_fields or {}

        def to_dict(self) -> dict[str, Any]:
            """Returns this extent as a dictionary.

            Returns:
                dict: A serialization of the Extent.
            """
            d = {
                "spatial": self.spatial.to_dict(),
                "temporal": self.temporal.to_dict(),
                **self.extra_fields,
            }

            return d

        def clone(self) -> Extent:
            """Clones this object.

            Returns:
                Extent: The clone of this extent.
            """
            cls = self.__class__
            return cls(
                spatial=self.spatial.clone(),
                temporal=self.temporal.clone(),
                extra_fields=deepcopy(self.extra_fields),
            )

        @staticmethod
        def from_dict(d: dict[str, Any]) -> Extent:
            """Constructs an Extent from a dict.

            Returns:
                Extent: The Extent deserialized from the JSON dict.
            """
            return Extent(
                spatial=SpatialExtent.from_dict(d["spatial"]),
                temporal=TemporalExtent.from_dict(d["temporal"]),
                extra_fields={
                    k: v for k, v in d.items() if k not in {"spatial", "temporal"}
                },
            )

        @staticmethod
        def from_items(
            items: Iterable[Item], extra_fields: dict[str, Any] | None = None
        ) -> Extent:
            """Create an Extent based on the datetimes and bboxes of a list of items.

            Args:
                items : A list of items to derive the extent from.
                extra_fields : Optional dictionary containing additional top-level fields
                    defined on the Extent object.

            Returns:
                Extent: An Extent that spatially and temporally covers all of the
                given items.
            """
            bounds_values: list[list[float]] = [
                [float("inf")],
                [float("inf")],
                [float("-inf")],
                [float("-inf")],
            ]
            datetimes: list[datetime] = []
            starts: list[datetime] = []
            ends: list[datetime] = []

            for item in items:
                if item.bbox is not None:
                    for i in range(0, 4):
                        bounds_values[i].append(item.bbox[i])
                if item.datetime is not None:
                    datetimes.append(item.datetime)
                if item.common_metadata.start_datetime is not None:
                    starts.append(item.common_metadata.start_datetime)
                if item.common_metadata.end_datetime is not None:
                    ends.append(item.common_metadata.end_datetime)

            if not any(datetimes + starts):
                start_timestamp = None
            else:
                start_timestamp = min(
                    [
                        dt if dt.tzinfo else dt.replace(tzinfo=tz.UTC)
                        for dt in datetimes + starts
                    ]
                )
            if not any(datetimes + ends):
                end_timestamp = None
            else:
                end_timestamp = max(
                    [
                        dt if dt.tzinfo else dt.replace(tzinfo=tz.UTC)
                        for dt in datetimes + ends
                    ]
                )

            spatial = SpatialExtent(
                [
                    [
                        min(bounds_values[0]),
                        min(bounds_values[1]),
                        max(bounds_values[2]),
                        max(bounds_values[3]),
                    ]
                ]
            )
            temporal = TemporalExtent([[start_timestamp, end_timestamp]])

            return Extent(spatial=spatial, temporal=temporal, extra_fields=extra_fields)

    class Collection(Catalog, Assets):
        """A Collection extends the Catalog spec with additional metadata that helps
        enable discovery.

        Args:
            id : Identifier for the collection. Must be unique within the STAC.
            description : Detailed multi-line description to fully explain the
                collection. `CommonMark 0.29 syntax <https://commonmark.org/>`_ MAY
                be used for rich text representation.
            extent : Spatial and temporal extents that describe the bounds of
                all items contained within this Collection.
            title : Optional short descriptive one-line title for the
                collection.
            stac_extensions : Optional list of extensions the Collection
                implements.
            href : Optional HREF for this collection, which be set as the
                collection's self link's HREF.
            catalog_type : Optional catalog type for this catalog. Must
                be one of the values in :class`~pystac.CatalogType`.
            license :  Collection's license(s) as a
                `SPDX License identifier <https://spdx.org/licenses/>`_,
                or `other`. If collection includes data with multiple
                different licenses, use `other` and add a link for
                each. The licenses `various` and `proprietary` are deprecated.
                Defaults to 'other'.
            keywords : Optional list of keywords describing the collection.
            providers : Optional list of providers of this Collection.
            summaries : An optional map of property summaries,
                either a set of values or statistics such as a range.
            extra_fields : Extra fields that are part of the top-level
                JSON properties of the Collection.
            assets : A dictionary mapping string keys to :class:`~pystac.Asset` objects. All
                :class:`~pystac.Asset` values in the dictionary will have their
                :attr:`~pystac.Asset.owner` attribute set to the created Collection.
            strategy : The layout strategy to use for setting the
                HREFs of the catalog child objections and items.
                If not provided, it will default to strategy of the parent and fallback to
                :class:`~pystac.layout.BestPracticesLayoutStrategy`.
        """

        description: str
        """Detailed multi-line description to fully explain the collection."""

        extent: Extent
        """Spatial and temporal extents that describe the bounds of all items contained
        within this Collection."""

        id: str
        """Identifier for the collection."""

        stac_extensions: list[str]
        """List of extensions the Collection implements."""

        title: str | None
        """Optional short descriptive one-line title for the collection."""

        keywords: list[str] | None
        """Optional list of keywords describing the collection."""

        providers: list[Provider] | None
        """Optional list of providers of this Collection."""

        summaries: Summaries
        """A map of property summaries, either a set of values or statistics such as a
        range."""

        links: list[Link]
        """A list of :class:`~pystac.Link` objects representing all links associated with
        this Collection."""

        extra_fields: dict[str, Any]
        """Extra fields that are part of the top-level JSON properties of the Collection."""

        STAC_OBJECT_TYPE = STACObjectType.COLLECTION

        DEFAULT_FILE_NAME = "collection.json"
        """Default file name that will be given to this STAC object
        in a canonical format."""

        def __init__(
            self,
            id: str,
            description: str,
            extent: Extent,
            title: str | None = None,
            stac_extensions: list[str] | None = None,
            href: str | None = None,
            extra_fields: dict[str, Any] | None = None,
            catalog_type: CatalogType | None = None,
            license: str = "other",
            keywords: list[str] | None = None,
            providers: list[Provider] | None = None,
            summaries: Summaries | None = None,
            assets: dict[str, Asset] | None = None,
            strategy: HrefLayoutStrategy | None = None,
        ):
            super().__init__(
                id,
                description,
                title,
                stac_extensions,
                extra_fields,
                href,
                catalog_type or CatalogType.ABSOLUTE_PUBLISHED,
                strategy,
            )
            self.extent = extent
            self.license = license

            self.stac_extensions: list[str] = stac_extensions or []
            self.keywords = keywords
            self.providers = providers
            self.summaries = summaries or Summaries.empty()
            self._item_assets: _ItemAssets | None = None

            self.assets = {}
            if assets is not None:
                for k, asset in assets.items():
                    self.add_asset(k, asset)

        def __repr__(self) -> str:
            return f"<Collection id={self.id}>"

        def add_item(
            self,
            item: Item,
            title: str | None = None,
            strategy: HrefLayoutStrategy | None = None,
            set_parent: bool = True,
        ) -> Link:
            link = super().add_item(item, title, strategy, set_parent)
            item.set_collection(self)
            return link

        def to_dict(
            self, include_self_link: bool = True, transform_hrefs: bool = True
        ) -> dict[str, Any]:
            d = super().to_dict(
                include_self_link=include_self_link, transform_hrefs=transform_hrefs
            )
            d["extent"] = self.extent.to_dict()
            d["license"] = self.license
            if self.stac_extensions:
                d["stac_extensions"] = self.stac_extensions
            if self.keywords:
                d["keywords"] = self.keywords
            if self.providers:
                d["providers"] = list(map(lambda x: x.to_dict(), self.providers))
            if not self.summaries.is_empty():
                d["summaries"] = self.summaries.to_dict()
            if any(self.assets):
                d["assets"] = {k: v.to_dict() for k, v in self.assets.items()}

            return d

        def clone(self) -> Collection:
            cls = self.__class__
            clone = cls(
                id=self.id,
                description=self.description,
                extent=self.extent.clone(),
                title=self.title,
                stac_extensions=self.stac_extensions.copy(),
                extra_fields=deepcopy(self.extra_fields),
                catalog_type=self.catalog_type,
                license=self.license,
                keywords=self.keywords.copy() if self.keywords is not None else None,
                providers=deepcopy(self.providers),
                summaries=self.summaries.clone(),
                assets={k: asset.clone() for k, asset in self.assets.items()},
            )

            clone._resolved_objects.cache(clone)

            for link in self.links:
                if link.rel == pystac.RelType.ROOT:
                    # Collection __init__ sets correct root to clone; don't reset
                    # if the root link points to self
                    root_is_self = link.is_resolved() and link.target is self
                    if not root_is_self:
                        clone.set_root(None)
                        clone.add_link(link.clone())
                else:
                    clone.add_link(link.clone())

            return clone

        @classmethod
        def from_dict(
            cls: type[C],
            d: dict[str, Any],
            href: str | None = None,
            root: Catalog | None = None,
            migrate: bool = True,
            preserve_dict: bool = True,
        ) -> C:
            from pystac.extensions.version import CollectionVersionExtension

            if migrate:
                info = identify_stac_object(d)
                d = migrate_to_latest(d, info)

            if not cls.matches_object_type(d):
                raise STACTypeError(d, cls)

            catalog_type = CatalogType.determine_type(d)

            if preserve_dict:
                d = deepcopy(d)

            id = d.pop("id")
            description = d.pop("description")
            license = d.pop("license")
            extent = Extent.from_dict(d.pop("extent"))
            title = d.pop("title", None)
            stac_extensions = d.pop("stac_extensions", None)
            keywords = d.pop("keywords", None)
            providers = d.pop("providers", None)
            if providers is not None:
                providers = list(map(lambda x: pystac.Provider.from_dict(x), providers))
            summaries = d.pop("summaries", None)
            if summaries is not None:
                summaries = Summaries(summaries)

            assets = d.pop("assets", None)
            if assets:
                assets = {k: Asset.from_dict(v) for k, v in assets.items()}
            links = d.pop("links")

            d.pop("stac_version")

            collection = cls(
                id=id,
                description=description,
                extent=extent,
                title=title,
                stac_extensions=stac_extensions,
                extra_fields=d,
                license=license,
                keywords=keywords,
                providers=providers,
                summaries=summaries,
                href=href,
                catalog_type=catalog_type,
                assets=assets,
            )

            for link in links:
                if link["rel"] == pystac.RelType.ROOT:
                    # Remove the link that's generated in Catalog's constructor.
                    collection.remove_links(pystac.RelType.ROOT)

                if link["rel"] != pystac.RelType.SELF or href is None:
                    collection.add_link(Link.from_dict(link))

            if root:
                collection.set_root(root)

            try:
                version = CollectionVersionExtension.ext(collection)
                if version.deprecated:
                    warnings.warn(
                        f"The collection '{collection.id}' is deprecated.",
                        DeprecatedWarning,
                    )
                # Collection asset deprecation checks pending version extension support
            except ExtensionNotImplemented:
                pass

            return collection

        def get_item(self, id: str, recursive: bool = False) -> Item | None:
            """Returns an item with a given ID.

            Args:
                id : The ID of the item to find.
                recursive : If True, search this collection and all children for the
                    item; otherwise, only search the items of this collection. Defaults
                    to False.

            Return:
                Item or None: The item with the given ID, or None if not found.
            """
            try:
                return next(self.get_items(id, recursive=recursive), None)
            except TypeError as e:
                if any("recursive" in arg for arg in e.args):
                    # For inherited classes that do not yet support recursive
                    # See https://github.com/stac-utils/pystac-client/issues/485
                    return super().get_item(id, recursive=recursive)
                raise e

        @property
        def item_assets(self) -> dict[str, ItemAssetDefinition]:
            """Accessor for `item_assets
            <https://github.com/radiantearth/stac-spec/blob/v1.1.0/collection-spec/collection-spec.md#item_assets>`__
            on this collection.

            Example::

            .. code-block:: python

               >>> print(collection.item_assets)
               {'thumbnail': <pystac.item_assets.ItemAssetDefinition at 0x72aea0420750>,
                'metadata': <pystac.item_assets.ItemAssetDefinition at 0x72aea017dc90>,
                'B5': <pystac.item_assets.ItemAssetDefinition at 0x72aea017efd0>,
                'B6': <pystac.item_assets.ItemAssetDefinition at 0x72aea016d5d0>,
                'B7': <pystac.item_assets.ItemAssetDefinition at 0x72aea016e050>,
                'B8': <pystac.item_assets.ItemAssetDefinition at 0x72aea016da90>}
               >>> collection.item_assets["thumbnail"].title
               'Thumbnail'

            Set attributes on :class:`~pystac.ItemAssetDefinition` objects

            .. code-block:: python

               >>> collection.item_assets["thumbnail"].title = "New Title"

            Add to the ``item_assets`` dict:

            .. code-block:: python

               >>> collection.item_assets["B4"] = {
                   'type': 'image/tiff; application=geotiff; profile=cloud-optimized',
                   'eo:bands': [{'name': 'B4', 'common_name': 'red'}]
               }
               >>> collection.item_assets["B4"].owner == collection
               True
            """
            if self._item_assets is None:
                self._item_assets = _ItemAssets(self)
            return self._item_assets

        @item_assets.setter
        def item_assets(
            self, item_assets: dict[str, ItemAssetDefinition | dict[str, Any]] | None
        ) -> None:
            # clear out the cached value
            self._item_assets = None

            if item_assets is None:
                self.extra_fields.pop("item_assets")
            else:
                self.extra_fields["item_assets"] = {
                    k: v if isinstance(v, dict) else v.to_dict()
                    for k, v in item_assets.items()
                }

        def update_extent_from_items(self) -> None:
            """
            Update datetime and bbox based on all items to a single bbox and time window.
            """
            self.extent = Extent.from_items(self.get_items(recursive=True))

        def full_copy(
            self, root: Catalog | None = None, parent: Catalog | None = None
        ) -> Collection:
            return cast(Collection, super().full_copy(root, parent))

        @classmethod
        def matches_object_type(cls, d: dict[str, Any]) -> bool:
            return identify_stac_object_type(d) == STACObjectType.COLLECTION

        @property
        def ext(self) -> CollectionExt:
            """Accessor for extension classes on this collection

            Example::

                print(collection.ext.xarray)
            """
            from pystac.extensions.ext import CollectionExt

            return CollectionExt(stac_object=self)