[wip]

Author: jpolchlo

pystac/__init__.py

@@ -84,6 +84,7 @@
 import pystac.validation
 
 import pystac.extensions.hooks
+import pystac.extensions.classification
 import pystac.extensions.datacube
 import pystac.extensions.eo
 import pystac.extensions.file
@@ -105,6 +106,7 @@
 
 EXTENSION_HOOKS = pystac.extensions.hooks.RegisteredExtensionHooks(
     [
+        pystac.extensions.classification.CLASSIFICATION_EXTENSION_HOOKS,
         pystac.extensions.datacube.DATACUBE_EXTENSION_HOOKS,
         pystac.extensions.eo.EO_EXTENSION_HOOKS,
         pystac.extensions.file.FILE_EXTENSION_HOOKS,

pystac/extensions/classification.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import re
 from typing import (
     Any,
     Dict,
@@ -32,7 +33,7 @@
 # Field names
 BITFIELDS_PROP: str = PREFIX + "bitfields"
 CLASSES_PROP: str = PREFIX + "classes"
-
+RASTER_BANDS_PROP: str = "raster:bands"
 
 class Classification:
     properties: Dict[str, Any]
@@ -44,12 +45,12 @@ def apply(
         self,
         value: int,
         name: str,
-        description: Optional[str],
-        title: Optional[str],
+        description: Optional[str] = None,
+        title: Optional[str] = None,
         color_hint: Optional[
             str
-        ],  # TODO: convert to an RGB string with syntax checking
-        nodata: Optional[bool],
+        ] = None,
+        nodata: Optional[bool] = None,
     ) -> None:
         self.value = value
         self.name = name
@@ -58,6 +59,10 @@ def apply(
         self.color_hint = color_hint
         self.nodata = nodata
 
+        color_hint_pattern = re.compile("^([0-9A-F]{6})$")
+        match = color_hint_pattern.match(color_hint)
+        assert color_hint is None or match is not None and match.group() == color_hint, "Must format color hints as '^([0-9A-F]{6})$'"
+
     @classmethod
     def create(
         cls,
@@ -170,9 +175,12 @@ def apply(
 
         assert offset >= 0, "Non-negative offsets only"
         assert length >= 1, "Positive field lengths only"
+        assert len(classes) > 0, "Must specify at least one class"
+        assert roles is None or len(roles) > 0, "When set, roles must contain at least one item"
 
-        class_coverage = set([c.value for c in classes])
-        assert set(range(0, 2**length)) - class_coverage == set(), "Classes must cover the complete range of values"
+        ## Ensure complete coverage of bit fields in classes
+        # class_coverage = set([c.value for c in classes])
+        # assert set(range(0, 2**length)) - class_coverage == set(), "Classes must cover the complete range of values"
 
     @classmethod
     def create(
@@ -213,11 +221,11 @@ def length(self, v: int) -> None:
 
     @property
     def classes(self) -> List[Classification]:
-        return get_required(self.properties.get("classes"), self, "classes")
+        return [Classification(d) for d in get_required(self.properties.get("classes"), self, "classes")]
 
     @classes.setter
     def classes(self, v: List[Classification]) -> None:
-        self.properties["classes"] = v
+        self.properties["classes"] = [c.to_dict() for c in v]
 
     @property
     def roles(self) -> Optional[List[str]]:
@@ -259,6 +267,27 @@ def to_dict(self) -> Dict[str, Any]:
         return self.properties
 
 
+def __fetch_from_dict(d: Dict[str, Any]) -> Optional[Union[Classification, Bitfield]]:
+    assert not (CLASSES_PROP in d and BITFIELDS_PROP in d), f"Cannot define both {CLASSES_PROP} and {BITFIELDS:PROP} in the same entity"
+
+    if CLASSES_PROP in d:
+        return list(
+            map(
+                lambda item: Classification(item),
+                d[CLASSES_PROP]
+            )
+        )
+    elif BITFIELDS_PROP in d:
+        return list(
+            map(
+                lambda item: Bitfield(item),
+                d[BITFIELDS_PROP]
+            )
+        )
+    else:
+        return None
+
+
 class ClassificationExtension(
     Generic[T],
     PropertiesExtension,
@@ -269,11 +298,12 @@ def apply(
         classes: Optional[List[Classification]] = None,
         bitfields: Optional[List[Bitfield]] = None,
     ) -> None:
+        # assert classes is None and bitfields is not None or bitfields is None and classes is not None, "Must set exactly one of `classes` or `bitfields`"
         self.classes = classes
         self.bitfields = bitfields
 
     @property
-    def classes(self) -> Optional[List[Classification]]:
+    def classes(self) -> Optional[List[Optional[Classification]]]:
         return self._get_classes()
 
     @classes.setter
@@ -282,7 +312,7 @@ def classes(self, v: Optional[List[Classification]]) -> None:
             CLASSES_PROP, map_opt(lambda classes: [c.to_dict() for c in classes], v)
         )
 
-    def _get_classes(self) -> Optional[List[Classification]]:
+    def _get_classes(self) -> Optional[List[Optional[Classification]]]:
         return map_opt(
             lambda classes: [Classification(c) for c in classes],
             self._get_property(CLASSES_PROP, List[Dict[str, Any]]),
@@ -317,6 +347,9 @@ def ext(cls, obj: T, add_if_missing: bool = False) -> ClassificationExtension[T]
         elif isinstance(obj, pystac.Asset):
             cls.validate_owner_has_extension(obj, add_if_missing)
             return cast(ClassificationExtension[T], AssetClassificationExtension(obj))
+        elif isinstance(obj, pystac.Collection):
+            cls.validate_has_extension(obj, add_if_missing)
+            return cast(ClassificationExtension[T], CollectionClassificationExtension(obj))
         else:
             raise pystac.ExtensionTypeError(
                 f"Classification extension does not apply to type '{type(obj).__name__}'"
@@ -342,41 +375,61 @@ def __init__(self, item: pystac.Item):
     def _get_classes(self) -> Optional[List[Classification]]:
         classes = self._get_property(CLASSES_PROP, List[Dict[str, Any]])
 
-        if classes is None:
-            asset_classes: List[Dict[str, Any]] = []
-            for _, value in self.item.get_assets().items():
-                if CLASSES_PROP in value.extra_fields:
-                    asset_classes.extend(
-                        cast(List[Dict[str, Any]], value.extra_fields.get(CLASSES_PROP))
-                    )
-            if any(asset_classes):
-                classes = asset_classes
-
         if classes is not None:
             return [Classification(c) for c in classes]
 
         return None
 
+    def _get_bitfields(self) -> Optional[List[Bitfields]]:
+        bitfields = self._get_property(BITFIELDS_PROP, List[Dict[str, Any]])
+
+        if bitfields is not None:
+            return [Bitfields(b) for b in bitfields]
+
+        return None
+
     def __repr__(self) -> str:
         return f"<ItemClassificationExtension Item id={self.item.id}>"
 
 
 class AssetClassificationExtension(ClassificationExtension[pystac.Asset]):
+    asset: pystac.Asset
     asset_href: str
     properties: Dict[str, Any]
     additional_red_properties: Optional[Iterable[Dict[str, Any]]] = None
 
     def _get_classes(self) -> Optional[List[Classification]]:
         if CLASSES_PROP not in self.properties:
-            return None
-        return list(
-            map(
-                lambda cls: Classification(cls),
-                cast(List[Dict[str, Any]], self.properties.get(CLASSES_PROP)),
-            )
-        )
+            if (self.asset.owner and RasterExtension.has_extension(self.asset.owner)):
+                bnds = RasterExtension.ext(self.asset).bands
+                return list(
+                    map(
+                        lambda b: Classification(b.properties.get(CLASSES_PROP)),
+                        bnds
+                    )
+                )
+            else:
+                return None
+        else:
+            return [Classification(self.properties.get(CLASSES_PROP))]
+
+    def _get_bitfields(self) -> Optional[List[Bitfield]]:
+        if BITFIELDS_PROP not in self.properties:
+            if (self.asset.owner and RasterExtension.has_extension(self.asset.owner)):
+                bnds = RasterExtension.ext(self.asset).bands
+                return list(
+                    map(
+                        lambda b: Bitfield(b.properties.get(BITFIELDS_PROP)),
+                        bnds
+                    )
+                )
+            else:
+                return None
+        else:
+            return [Classification(self.properties.get(CLASSES_PROP))]
 
     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):
@@ -386,6 +439,28 @@ def __repr__(self) -> str:
         return f"<AssetClassificationExtension Asset href={self.asset_href}>"
 
 
+class CollectionClassificationExtension(ClassificationExtension[pystac.Collection]):
+    """A concrete implementation of :class:`ClassificationExtension` on a
+    :class:`~pystac.Collection` that extends the properties of a Collection to
+    include properties defined in the :stac_ext:`Classification Extension
+    <classification>`.
+    """
+    collection: pystac.Collection
+    properties: Dict[str, Any]
+
+    def __init__(self, collection: pystac.Collection):
+        self.collection = collection
+        self.properties = collection.assets
+
+    def _get_classes(self) -> Optional[List[Optional[Classification]]]:
+        classes = __fetch_from_dict(self.properties)
+        if classes is None and ItemAssetsExtension.has_extension(self.collection):
+            classes = __fetch_from_dict(ItemAssetsExtension.ext(self.collection).item_assets)
+        return list(filter(lambda x: isinstance(x, Classification), classes))
+
+    def _get_bitfields(self) -> Optional[List[Bitfield]]:
+        pass
+
 class SummariesClassificationExtension(SummariesExtension):
     @property
     def classes(self) -> Optional[List[Classification]]:

tests/data-files/classification/classification-landsat-example.json

@@ -1216,6 +1216,6 @@
     "https://stac-extensions.github.io/projection/v1.0.0/schema.json",
     "https://landsat.usgs.gov/stac/landsat-extension/v1.1.1/schema.json",
     "https://stac-extensions.github.io/scientific/v1.0.0/schema.json",
-    "https://stac-extensions.github.io/classification/v1.1.0/schema.json"
+    "https://stac-extensions.github.io/classification/v1.0.0/schema.json"
   ]
 }