Improving logic and extensibility

eo3/__init__.py

@@ -1,7 +1,6 @@
 from ._version import get_versions
-from .assemble import IncompleteDatasetError
 from .images import GridSpec, ValidDataMethod
-from .model import Eo3DatasetDocBase
+from .model import DatasetDocBase
 from .properties import Eo3DictBase
 
 REPO_URL = "https://github.com/GeoscienceAustralia/eo-datasets.git"
@@ -10,10 +9,9 @@
 del get_versions
 
 __all__ = (
-    "Eo3DatasetDocBase",
+    "DatasetDocBase",
     "Eo3DictBase",
     "GridSpec",
-    "IncompleteDatasetError",
     "REPO_URL",
     "ValidDataMethod",
     "__version__",

eo3/eo3_core.py

@@ -16,7 +16,7 @@
     polygon,
 )
 
-EO3_SCHEMA = "https://schemas.opendatacube.org/dataset"
+from eo3.model import ODC_DATASET_SCHEMA_URL
 
 
 # This is should become eo3.models.GridDoc
@@ -174,7 +174,7 @@ def is_doc_eo3(doc: Dict[str, Any]) -> bool:
     if schema is None:
         return False
 
-    if schema == EO3_SCHEMA:
+    if schema == ODC_DATASET_SCHEMA_URL:
         return True
 
     # Otherwise it has an unknown schema.
@@ -204,46 +204,46 @@ def is_doc_geo(doc: Dict[str, Any], check_eo3: bool = True) -> bool:
     return "extent" in doc or "grid_spatial" in doc
 
 
-def prep_eo3(
-    doc: Dict[str, Any], auto_skip: bool = False, resolution: Optional[float] = None
-) -> Dict[str, Any]:
-    """Modify spatial and lineage sections of eo3 metadata
+def prep_eo3(doc: Dict[str, Any],
+             resolution: Optional[float] = None,  # can we remove this?
+             remap_lineage=True) -> Dict[str, Any]:
+    """ Modify spatial and lineage sections of eo3 metadata
     :param doc: input document
-    :param auto_skip: If true check if dataset is EO3 and if not
-                      silently return input dataset without modifications
+    :param remap_lineage: If True (default) disambiguate lineage classifiers so that
+                          source_id and classifier form a unique index (for indexes that DON'T
+                          support external_lineage).
+                          If False, leave lineage in the same format.
     """
     if doc is None:
         return None
 
-    if auto_skip:
-        if not is_doc_eo3(doc):
-            return doc
-
     def stringify(u: Optional[Union[str, UUID]]) -> Optional[str]:
         return u if isinstance(u, str) else str(u) if u else None
 
-    doc["id"] = stringify(doc.get("id", None))
+    doc['id'] = stringify(doc.get('id', None))
 
     doc = add_eo3_parts(doc, resolution=resolution)
-    lineage = doc.pop("lineage", {})
-
-    def remap_lineage(name, uuids) -> Dict[str, Any]:
-        """Turn name, [uuid] -> {name: {id: uuid}}"""
-        if len(uuids) == 0:
-            return {}
-        if isinstance(uuids, dict) or isinstance(uuids[0], dict):
-            raise ValueError("Embedded lineage not supported for eo3 metadata types")
-        if len(uuids) == 1:
-            return {name: {"id": stringify(uuids[0])}}
-
-        out = {}
-        for idx, uuid in enumerate(uuids, start=1):
-            out[name + str(idx)] = {"id": stringify(uuid)}
-        return out
-
-    sources = {}
-    for name, uuids in lineage.items():
-        sources.update(remap_lineage(name, uuids))
-
-    doc["lineage"] = dict(source_datasets=sources)
+    if remap_lineage:
+        lineage = doc.pop('lineage', {})
+
+        def lineage_remap(name, uuids) -> Dict[str, Any]:
+            """ Turn name, [uuid] -> {name: {id: uuid}}
+            """
+            if len(uuids) == 0:
+                return {}
+            if isinstance(uuids, dict) or isinstance(uuids[0], dict):
+                raise ValueError("Embedded lineage not supported for eo3 metadata types")
+            if len(uuids) == 1:
+                return {name: {'id': stringify(uuids[0])}}
+
+            out = {}
+            for idx, uuid in enumerate(uuids, start=1):
+                out[name+str(idx)] = {'id': stringify(uuid)}
+            return out
+
+        sources = {}
+        for name, uuids in lineage.items():
+            sources.update(lineage_remap(name, uuids))
+
+        doc['lineage'] = dict(source_datasets=sources)
     return doc