Add Point Cloud Extension summaries

Author: duckontheweb

pystac/extensions/pointcloud.py

@@ -9,8 +9,10 @@
 from pystac.extensions.base import (
     ExtensionManagementMixin,
     PropertiesExtension,
+    SummariesExtension,
 )
 from pystac.extensions.hooks import ExtensionHooks
+from pystac.summaries import RangeSummary
 from pystac.utils import map_opt, get_required
 
 T = TypeVar("T", pystac.Item, pystac.Asset)
@@ -320,9 +322,16 @@ def to_dict(self) -> Dict[str, Any]:
         """Returns a JSON-like dictionary representing this ``Statistic``."""
         return self.properties
 
+    def __eq__(self, o: object) -> bool:
+        if not isinstance(o, Statistic):
+            return NotImplemented
+        return self.to_dict() == o.to_dict()
+
 
 class PointcloudExtension(
-    Generic[T], PropertiesExtension, ExtensionManagementMixin[pystac.Item]
+    Generic[T],
+    PropertiesExtension,
+    ExtensionManagementMixin[Union[pystac.Item, pystac.Collection]],
 ):
     """An abstract class that can be used to extend the properties of an
     :class:`~pystac.Item` or :class:`~pystac.Asset` with properties from the
@@ -473,6 +482,16 @@ def ext(cls, obj: T, add_if_missing: bool = False) -> "PointcloudExtension[T]":
                 f"Pointcloud extension does not apply to type '{type(obj).__name__}'"
             )
 
+    @classmethod
+    def summaries(
+        cls, obj: pystac.Collection, add_if_missing: bool = False
+    ) -> "SummariesPointcloudExtension":
+        if not add_if_missing:
+            cls.validate_has_extension(obj)
+        else:
+            cls.add_to(obj)
+        return SummariesPointcloudExtension(obj)
+
 
 class ItemPointcloudExtension(PointcloudExtension[pystac.Item]):
     """A concrete implementation of :class:`PointcloudExtension` on an :class:`~pystac.Item`
@@ -513,6 +532,59 @@ def __repr__(self) -> str:
         return f"<AssetPointcloudExtension Asset {self.repr_id}>"
 
 
+class SummariesPointcloudExtension(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:`Point Cloud Extension <pointcloud>`.
+    """
+
+    @property
+    def count(self) -> Optional[RangeSummary[int]]:
+        return self.summaries.get_range(COUNT_PROP)
+
+    @count.setter
+    def count(self, v: Optional[RangeSummary[int]]) -> None:
+        self._set_summary(COUNT_PROP, v)
+
+    @property
+    def type(self) -> Optional[List[Union[PhenomenologyType, str]]]:
+        return self.summaries.get_list(TYPE_PROP)
+
+    @type.setter
+    def type(self, v: Optional[List[Union[PhenomenologyType, str]]]) -> None:
+        self._set_summary(TYPE_PROP, v)
+
+    @property
+    def encoding(self) -> Optional[List[str]]:
+        return self.summaries.get_list(ENCODING_PROP)
+
+    @encoding.setter
+    def encoding(self, v: Optional[List[str]]) -> None:
+        self._set_summary(ENCODING_PROP, v)
+
+    @property
+    def density(self) -> Optional[RangeSummary[float]]:
+        return self.summaries.get_range(DENSITY_PROP)
+
+    @density.setter
+    def density(self, v: Optional[RangeSummary[float]]) -> None:
+        self._set_summary(DENSITY_PROP, v)
+
+    @property
+    def statistics(self) -> Optional[List[Statistic]]:
+        return map_opt(
+            lambda stats: [Statistic(d) for d in stats],
+            self.summaries.get_list(STATISTICS_PROP),
+        )
+
+    @statistics.setter
+    def statistics(self, v: Optional[List[Statistic]]) -> None:
+        self._set_summary(
+            STATISTICS_PROP,
+            map_opt(lambda stats: [s.to_dict() for s in stats], v),
+        )
+
+
 class PointcloudExtensionHooks(ExtensionHooks):
     schema_uri: str = SCHEMA_URI
     prev_extension_ids = {"pointcloud"}

tests/extensions/test_pointcloud.py

@@ -15,6 +15,7 @@
     SchemaType,
     Statistic,
 )
+from pystac.summaries import RangeSummary
 from tests.utils import TestCases, assert_to_from_dict
 
 
@@ -336,3 +337,119 @@ def test_asset_ext_add_to(self) -> None:
         _ = PointcloudExtension.ext(asset, add_if_missing=True)
 
         self.assertIn(PointcloudExtension.get_schema_uri(), item.stac_extensions)
+
+
+class PointcloudSummariesTest(unittest.TestCase):
+    def setUp(self) -> None:
+        self.maxDiff = None
+        self.collection = pystac.Collection.from_file(
+            TestCases.get_path("data-files/collections/multi-extent.json")
+        )
+
+    def test_count(self) -> None:
+        collection = self.collection.clone()
+        summaries_ext = PointcloudExtension.summaries(collection, True)
+        count_range = RangeSummary(1000, 10000)
+
+        summaries_ext.count = count_range
+
+        self.assertEqual(
+            summaries_ext.count,
+            count_range,
+        )
+
+        summaries_dict = collection.to_dict()["summaries"]
+
+        self.assertEqual(
+            summaries_dict["pc:count"],
+            count_range.to_dict(),
+        )
+
+    def test_type(self) -> None:
+        collection = self.collection.clone()
+        summaries_ext = PointcloudExtension.summaries(collection, True)
+        type_list = [PhenomenologyType.LIDAR, "something"]
+
+        summaries_ext.type = type_list
+
+        self.assertEqual(
+            summaries_ext.type,
+            type_list,
+        )
+
+        summaries_dict = collection.to_dict()["summaries"]
+
+        self.assertEqual(
+            summaries_dict["pc:type"],
+            type_list,
+        )
+
+    def test_encoding(self) -> None:
+        collection = self.collection.clone()
+        summaries_ext = PointcloudExtension.summaries(collection, True)
+        encoding_list = ["LASzip"]
+
+        summaries_ext.encoding = encoding_list
+
+        self.assertEqual(
+            summaries_ext.encoding,
+            encoding_list,
+        )
+
+        summaries_dict = collection.to_dict()["summaries"]
+
+        self.assertEqual(
+            summaries_dict["pc:encoding"],
+            encoding_list,
+        )
+
+    def test_density(self) -> None:
+        collection = self.collection.clone()
+        summaries_ext = PointcloudExtension.summaries(collection, True)
+        density_range = RangeSummary(500.0, 1000.0)
+
+        summaries_ext.density = density_range
+
+        self.assertEqual(
+            summaries_ext.density,
+            density_range,
+        )
+
+        summaries_dict = collection.to_dict()["summaries"]
+
+        self.assertEqual(
+            summaries_dict["pc:density"],
+            density_range.to_dict(),
+        )
+
+    def test_statistics(self) -> None:
+        collection = self.collection.clone()
+        summaries_ext = PointcloudExtension.summaries(collection, True)
+        statistics_list = [
+            Statistic(
+                {
+                    "average": 637294.1783,
+                    "count": 10653336,
+                    "maximum": 639003.73,
+                    "minimum": 635577.79,
+                    "name": "X",
+                    "position": 0,
+                    "stddev": 967.9329805,
+                    "variance": 936894.2548,
+                }
+            )
+        ]
+
+        summaries_ext.statistics = statistics_list
+
+        self.assertEqual(
+            summaries_ext.statistics,
+            statistics_list,
+        )
+
+        summaries_dict = collection.to_dict()["summaries"]
+
+        self.assertEqual(
+            summaries_dict["pc:statistics"],
+            [s.to_dict() for s in statistics_list],
+        )