Add summaries for Timestamps Extension

Author: duckontheweb

pystac/extensions/timestamps.py

@@ -4,12 +4,14 @@
 """
 
 from datetime import datetime as Datetime
+from pystac.summaries import RangeSummary
 from typing import Dict, Any, Generic, Iterable, Optional, Set, TypeVar, cast
 
 import pystac
 from pystac.extensions.base import (
     ExtensionManagementMixin,
     PropertiesExtension,
+    SummariesExtension,
 )
 from pystac.extensions.hooks import ExtensionHooks
 from pystac.utils import datetime_to_str, map_opt, str_to_datetime
@@ -139,6 +141,11 @@ def ext(cls, obj: T, add_if_missing: bool = False) -> "TimestampsExtension[T]":
                 f"Timestamps extension does not apply to type '{type(obj).__name__}'"
             )
 
+    @staticmethod
+    def summaries(obj: pystac.Collection) -> "SummariesTimestampsExtension":
+        """Returns the extended summaries object for the given collection."""
+        return SummariesTimestampsExtension(obj)
+
 
 class ItemTimestampsExtension(TimestampsExtension[pystac.Item]):
     """A concrete implementation of :class:`TimestampsExtension` on an
@@ -192,6 +199,88 @@ def __repr__(self) -> str:
         return "<AssetTimestampsExtension Asset href={}>".format(self.asset_href)
 
 
+class SummariesTimestampsExtension(SummariesExtension):
+    """A concrete implementation of :class:`~SummariesExtension` that extends
+    the ``summaries`` field of a :class:`~pystac.Collection` to include properties
+    defined in the :stac-ext:`Timestamps Extension <timestamps>`.
+    """
+
+    @property
+    def published(self) -> Optional[RangeSummary[Datetime]]:
+        """Get or sets the summary of :attr:`TimestampsExtension.published` values
+        for this Collection.
+        """
+
+        return map_opt(
+            lambda s: RangeSummary(
+                str_to_datetime(s.minimum), str_to_datetime(s.maximum)
+            ),
+            self.summaries.get_range(PUBLISHED_PROP),
+        )
+
+    @published.setter
+    def published(self, v: Optional[RangeSummary[Datetime]]) -> None:
+        self._set_summary(
+            PUBLISHED_PROP,
+            map_opt(
+                lambda s: RangeSummary(
+                    datetime_to_str(s.minimum), datetime_to_str(s.maximum)
+                ),
+                v,
+            ),
+        )
+
+    @property
+    def expires(self) -> Optional[RangeSummary[Datetime]]:
+        """Get or sets the summary of :attr:`TimestampsExtension.expires` values
+        for this Collection.
+        """
+
+        return map_opt(
+            lambda s: RangeSummary(
+                str_to_datetime(s.minimum), str_to_datetime(s.maximum)
+            ),
+            self.summaries.get_range(EXPIRES_PROP),
+        )
+
+    @expires.setter
+    def expires(self, v: Optional[RangeSummary[Datetime]]) -> None:
+        self._set_summary(
+            EXPIRES_PROP,
+            map_opt(
+                lambda s: RangeSummary(
+                    datetime_to_str(s.minimum), datetime_to_str(s.maximum)
+                ),
+                v,
+            ),
+        )
+
+    @property
+    def unpublished(self) -> Optional[RangeSummary[Datetime]]:
+        """Get or sets the summary of :attr:`TimestampsExtension.unpublished` values
+        for this Collection.
+        """
+
+        return map_opt(
+            lambda s: RangeSummary(
+                str_to_datetime(s.minimum), str_to_datetime(s.maximum)
+            ),
+            self.summaries.get_range(UNPUBLISHED_PROP),
+        )
+
+    @unpublished.setter
+    def unpublished(self, v: Optional[RangeSummary[Datetime]]) -> None:
+        self._set_summary(
+            UNPUBLISHED_PROP,
+            map_opt(
+                lambda s: RangeSummary(
+                    datetime_to_str(s.minimum), datetime_to_str(s.maximum)
+                ),
+                v,
+            ),
+        )
+
+
 class TimestampsExtensionHooks(ExtensionHooks):
     schema_uri: str = SCHEMA_URI
     prev_extension_ids: Set[str] = set(["timestamps"])

tests/extensions/test_timestamps.py

@@ -4,6 +4,7 @@
 
 import pystac
 from pystac import ExtensionTypeError
+from pystac.summaries import RangeSummary
 from pystac.extensions.timestamps import TimestampsExtension
 from pystac.utils import get_opt, str_to_datetime, datetime_to_str
 from tests.utils import TestCases, assert_to_from_dict
@@ -233,3 +234,106 @@ def test_should_raise_exception_when_passing_invalid_extension_object(
             TimestampsExtension.ext,
             object(),
         )
+
+    def test_item_repr(self) -> None:
+        item = pystac.Item.from_file(self.example_uri)
+
+        self.assertEqual(
+            TimestampsExtension.ext(item).__repr__(),
+            f"<ItemTimestampsExtension Item id={item.id}>",
+        )
+
+    def test_asset_repr(self) -> None:
+        item = pystac.Item.from_file(self.example_uri)
+        asset = item.assets["blue"]
+
+        self.assertEqual(
+            TimestampsExtension.ext(asset).__repr__(),
+            f"<AssetTimestampsExtension Asset href={asset.href}>",
+        )
+
+
+class TimestampsSummariesTest(unittest.TestCase):
+    def setUp(self) -> None:
+        self.maxDiff = None
+
+    @staticmethod
+    def collection() -> pystac.Collection:
+        return pystac.Collection.from_file(
+            TestCases.get_path("data-files/collections/multi-extent.json")
+        )
+
+    def test_published(self) -> None:
+        collection = self.collection()
+        summaries_ext = TimestampsExtension.summaries(collection)
+        published_range = RangeSummary(
+            str_to_datetime("2020-01-01T00:00:00.000Z"),
+            str_to_datetime("2020-01-02T00:00:00.000Z"),
+        )
+
+        summaries_ext.published = published_range
+
+        self.assertEqual(
+            summaries_ext.published,
+            published_range,
+        )
+
+        summaries_dict = collection.to_dict()["summaries"]
+
+        self.assertDictEqual(
+            summaries_dict["published"],
+            {
+                "minimum": datetime_to_str(published_range.minimum),
+                "maximum": datetime_to_str(published_range.maximum),
+            },
+        )
+
+    def test_expires(self) -> None:
+        collection = self.collection()
+        summaries_ext = TimestampsExtension.summaries(collection)
+        expires_range = RangeSummary(
+            str_to_datetime("2020-01-01T00:00:00.000Z"),
+            str_to_datetime("2020-01-02T00:00:00.000Z"),
+        )
+
+        summaries_ext.expires = expires_range
+
+        self.assertEqual(
+            summaries_ext.expires,
+            expires_range,
+        )
+
+        summaries_dict = collection.to_dict()["summaries"]
+
+        self.assertDictEqual(
+            summaries_dict["expires"],
+            {
+                "minimum": datetime_to_str(expires_range.minimum),
+                "maximum": datetime_to_str(expires_range.maximum),
+            },
+        )
+
+    def test_unpublished(self) -> None:
+        collection = self.collection()
+        summaries_ext = TimestampsExtension.summaries(collection)
+        unpublished_range = RangeSummary(
+            str_to_datetime("2020-01-01T00:00:00.000Z"),
+            str_to_datetime("2020-01-02T00:00:00.000Z"),
+        )
+
+        summaries_ext.unpublished = unpublished_range
+
+        self.assertEqual(
+            summaries_ext.unpublished,
+            unpublished_range,
+        )
+
+        summaries_dict = collection.to_dict()["summaries"]
+
+        self.assertDictEqual(
+            summaries_dict["unpublished"],
+            {
+                "minimum": datetime_to_str(unpublished_range.minimum),
+                "maximum": datetime_to_str(unpublished_range.maximum),
+            },
+        )