Merge branch 'develop-1.9' into integrate_1.9

Author: Ariana-B

apps/dc_tools/odc/apps/dc_tools/_stac.py

@@ -10,7 +10,6 @@
 import numpy
 from datacube.model import Dataset
 from odc.geo.geom import Geometry, box
-from eodatasets3.serialise import from_doc
 from eodatasets3.stac import to_stac_item
 from toolz import get_in
 from urllib.parse import urlparse
@@ -231,7 +230,11 @@ def _get_stac_bands(
         # If transform specified here in the asset it should override
         # the properties-specified transform.
         transform = asset.get("proj:transform") or proj_transform
-        grid = f"g{transform[0]:g}m"
+
+        if transform is not None:
+            grid = f"g{transform[0]:g}m"
+        else:
+            grid = default_grid
 
         # As per transform, shape here overrides properties
         shape = asset.get("proj:shape") or proj_shape
@@ -373,6 +376,11 @@ def stac_transform(input_stac: Document) -> Document:
         proj_transform=proj_transform,
     )
 
+    # STAC document may not have top-level proj:shape property
+    # use one of the bands as a default
+    proj_shape = grids.get("default").get("shape")
+    proj_transform = grids.get("default").get("transform")
+
     stac_properties, lineage = _get_stac_properties_lineage(input_stac)
 
     epsg = properties["proj:epsg"]
@@ -444,6 +452,7 @@ def transform_geom_json_coordinates_to_list(geom_json):
 
 def ds_to_stac(ds: Dataset) -> dict:
     """Get STAC document from dataset with eo3 metadata"""
+    from eodatasets3.serialise import from_doc
     if ds.is_eo3:
         if not ds.uris:
             raise ValueError("Can't find dataset location")

apps/dc_tools/odc/apps/dc_tools/stac_api_to_dc.py

@@ -9,9 +9,10 @@
 
 import click
 from datacube import Datacube
-from datacube.index.hl import Doc2Dataset
+from datacube.model import Dataset
 from datacube.ui.click import environment_option, pass_config
-from odc.apps.dc_tools._stac import stac_transform
+from odc.stac.eo3 import stac2ds
+
 from odc.apps.dc_tools.utils import (
     SkippedException,
     allow_unsafe,
@@ -64,8 +65,9 @@ def _parse_options(options: Optional[str]) -> Dict[str, Any]:
 
 def item_to_meta_uri(
     item: Item,
+    dc: Datacube,
     rename_product: Optional[str] = None,
-) -> Generator[Tuple[dict, str, bool], None, None]:
+) -> Generator[Tuple[Dataset, str, bool], None, None]:
     for link in item.links:
         if link.rel == "self":
             uri = link.target
@@ -75,31 +77,44 @@ def item_to_meta_uri(
             uri = link.target
             break
 
-    metadata = item.to_dict()
     if rename_product is not None:
-        metadata["properties"]["odc:product"] = rename_product
-    stac = metadata
-    metadata = stac_transform(metadata)
+        item.properties["odc:product"] = rename_product
+
+    # Try to the Datacube product for the dataset
+    product_name = item.properties.get("odc:product", item.collection_id)
+    product_name_sanitised = product_name.replace("-", "_")
+    product = dc.index.products.get_by_name(product_name_sanitised)
+
+    if product is None:
+        logging.warning(
+            "Couldn't find matching product for product name: %s",
+            product_name_sanitised,
+        )
+        raise SkippedException(f"Couldn't find matching product for product name: {product_name_sanitised}")
+
+    # Convert the STAC Item to a Dataset
+    dataset = next(stac2ds([item]))
+    # And assign the product ID
+    dataset.product = product
 
-    return (metadata, uri, stac)
+    return (dataset, uri, item.to_dict())
 
 
 def process_item(
     item: Item,
     dc: Datacube,
-    doc2ds: Doc2Dataset,
     update_if_exists: bool,
     allow_unsafe: bool,
     rename_product: Optional[str] = None,
     archive_less_mature: int = None,
     publish_action: bool = False,
 ):
-    meta, uri, stac = item_to_meta_uri(item, rename_product)
+    dataset, uri, stac = item_to_meta_uri(item, dc, rename_product)
     index_update_dataset(
-        meta,
+        dataset,
         uri,
         dc,
-        doc2ds,
+        None,
         update_if_exists=update_if_exists,
         allow_unsafe=allow_unsafe,
         archive_less_mature=archive_less_mature,
@@ -118,7 +133,6 @@ def stac_api_to_odc(
     archive_less_mature: int = None,
     publish_action: Optional[str] = None,
 ) -> Tuple[int, int, int]:
-    doc2ds = Doc2Dataset(dc.index)
     client = Client.open(catalog_href)
 
     search = client.search(**config)
@@ -143,14 +157,13 @@ def stac_api_to_odc(
                 process_item,
                 item,
                 dc,
-                doc2ds,
                 update_if_exists=update_if_exists,
                 allow_unsafe=allow_unsafe,
                 rename_product=rename_product,
                 archive_less_mature=archive_less_mature,
                 publish_action=publish_action,
             ): item.id
-            for item in search.get_all_items()
+            for item in search.items()
         }
         for future in concurrent.futures.as_completed(future_to_item):
             item = future_to_item[future]

apps/dc_tools/odc/apps/dc_tools/utils.py

@@ -4,6 +4,7 @@
 import click
 import importlib_resources
 from datacube import Datacube
+from datacube.model import Dataset
 from datacube.index.hl import Doc2Dataset
 from datacube.utils import changes
 from datadog import initialize, statsd
@@ -183,14 +184,15 @@ def get_esri_list():
 
 
 def index_update_dataset(
-    metadata: dict,
+    dataset: dict | Dataset,
     uri: str,
     dc: Datacube,
-    doc2ds: Doc2Dataset,
+    doc2ds: Doc2Dataset | None,
     update: bool = False,
     update_if_exists: bool = False,
     allow_unsafe: bool = False,
     archive_less_mature: Optional[int] = None,
+    auto_add_lineage: Optional[bool] = False,
     publish_action: Optional[str] = None,
     stac_doc: Optional[dict] = None,
 ) -> int:
@@ -215,19 +217,17 @@ def index_update_dataset(
     :param stac_doc: STAC document for publication to SNS topic.
     :return: Returns nothing.  Raises an exception if anything goes wrong.
     """
-    if uri is None:
-        raise IndexingException("Failed to get URI from metadata doc")
     # Make sure we can create a dataset first
-    try:
-        ds, err = doc2ds(metadata, uri)
-    except ValueError as e:
-        raise IndexingException(
-            f"Exception thrown when trying to create dataset: '{e}'\n The URI was {uri}"
-        ) from e
-    if ds is None:
-        raise IndexingException(
-            f"Failed to create dataset with error {err}\n The URI was {uri}"
-        )
+    if not isinstance(dataset, Dataset):
+        print("Not a dataset: ", dataset)
+        try:
+            if doc2ds is None:
+                doc2ds = Doc2Dataset(dc.index)
+            dataset, _ = doc2ds(dataset, uri)
+        except ValueError as e:
+            raise IndexingException(
+                f"Exception thrown when trying to create dataset: '{e}'\n The URI was {uri}"
+            ) from e
 
     with dc.index.transaction():
         # Process in a transaction
@@ -236,13 +236,13 @@ def index_update_dataset(
         updated = False
 
         if isinstance(archive_less_mature, int) and publish_action:
-            dupes = dc.index.datasets.find_less_mature(ds, archive_less_mature)
+            dupes = dc.index.datasets.find_less_mature(dataset, archive_less_mature)
             for dupe in dupes:
                 archive_stacs.append(ds_to_stac(dupe))
 
         # Now do something with the dataset
         # Note that any of the exceptions raised below will rollback any archiving performed above.
-        if dc.index.datasets.has(metadata.get("id")):
+        if dc.index.datasets.has(dataset.id):
             # Update
             if update or update_if_exists:
                 # Set up update fields
@@ -252,7 +252,7 @@ def index_update_dataset(
                 # Do the updating
                 try:
                     dc.index.datasets.update(
-                        ds,
+                        dataset,
                         updates_allowed=updates,
                         archive_less_mature=archive_less_mature,
                     )
@@ -262,9 +262,8 @@ def index_update_dataset(
                         f"Updating the dataset raised an exception: {e}"
                     )
             else:
-                logging.warning("Dataset already exists, not indexing")
                 raise SkippedException(
-                    f"Dataset {metadata.get('id')} already exists, not indexing"
+                    f"Dataset {dataset.id} already exists, not indexing"
                 )
         else:
             if update:
@@ -273,22 +272,26 @@ def index_update_dataset(
                     "Can't update dataset because it doesn't exist."
                 )
             # Everything is working as expected, add the dataset
-            dc.index.datasets.add(ds, archive_less_mature=archive_less_mature)
+            dc.index.datasets.add(
+                dataset,
+                with_lineage=auto_add_lineage,
+                archive_less_mature=archive_less_mature,
+            )
             added = True
 
     if publish_action:
         for arch_stac in archive_stacs:
             publish_to_topic(arn=publish_action, action="ARCHIVED", stac=arch_stac)
 
     if added:
-        logging.info("New Dataset Added: %s", ds.id)
+        logging.info("New Dataset Added: %s", dataset.id)
         if publish_action:
             # if STAC was not provided, generate from dataset
-            stac_doc = stac_doc if stac_doc else ds_to_stac(ds)
+            stac_doc = stac_doc if stac_doc else ds_to_stac(dataset)
             publish_to_topic(arn=publish_action, action="ADDED", stac=stac_doc)
 
     if updated:
-        logging.info("Existing Dataset Updated: %s", ds.id)
+        logging.info("Existing Dataset Updated: %s", dataset.id)
 
 
 def statsd_gauge_reporting(value, tags=None, statsd_setting="localhost:8125"):

apps/dc_tools/setup.cfg

@@ -34,6 +34,7 @@ install_requires =
     odc_io
     odc-cloud[ASYNC]>=0.2.3
     odc-geo
+    odc-stac
     pystac>=1.0.0
     rio-stac
     urlpath