feat(driver): use properties to choose asset harmonisation drive #1998

eodag/api/product/__init__.py

@@ -34,9 +34,6 @@
     from ._assets import Asset, AssetsDict  # type: ignore[assignment]
     from ._product import EOProduct  # type: ignore[assignment]
 
-# exportable content
-__all__ = ["Asset", "AssetsDict", "EOProduct"]
-
 
 def unregistered_product_from_item(
     feature: dict[str, Any], provider: str, plugins_manager: "PluginManager"
@@ -60,3 +57,7 @@ def unregistered_product_from_item(
                     products[0].collection = products[0].properties.get("collection")
                 return products[0]
     return None
+
+
+# exportable content
+__all__ = ["Asset", "AssetsDict", "EOProduct", "unregistered_product_from_item"]

eodag/api/product/_product.py

@@ -260,7 +260,7 @@ def from_geojson(cls, feature: dict[str, Any]) -> EOProduct:
     def __repr__(self) -> str:
         try:
             return "{}(id={}, provider={})".format(
-                self.__class__.__name__, self.properties["id"], self.provider
+                self.__class__.__name__, self.properties.get("id", "?"), self.provider
             )
         except KeyError as e:
             raise MisconfiguredError(
@@ -617,8 +617,16 @@ def format_quicklook_address() -> None:
     def get_driver(self) -> DatasetDriver:
         """Get the most appropriate driver"""
         for driver_conf in DRIVERS:
-            if all([criteria(self) for criteria in driver_conf["criteria"]]):
+
+            # Select a driver if all criterias match
+            match = True
+            for criteria in driver_conf["criteria"]:
+                if not criteria(self):
+                    match = False
+                    break
+            if match:
                 return driver_conf["driver"]
+
         return GenericDriver()
 
     def _repr_html_(self):

eodag/api/product/drivers/__init__.py

@@ -22,10 +22,9 @@
 
 from typing_extensions import TypedDict
 
-from eodag.api.product.drivers.base import DatasetDriver
-from eodag.api.product.drivers.generic import GenericDriver
-from eodag.api.product.drivers.sentinel1 import Sentinel1Driver
-from eodag.api.product.drivers.sentinel2 import Sentinel2Driver
+from .generic import DatasetDriver, GenericDriver
+from .sentinel1 import Sentinel1Driver
+from .sentinel2 import Sentinel2Driver
 
 
 class DriverCriteria(TypedDict):
@@ -40,27 +39,22 @@ class DriverCriteria(TypedDict):
 #: list of drivers and their criteria
 DRIVERS: list[DriverCriteria] = [
     {
-        "criteria": [
-            lambda prod: True
-            if (prod.collection or "").startswith("S2_MSI_")
-            else False
-        ],
+        "criteria": [lambda product: Sentinel2Driver.match(product, by="collection")],
         "driver": Sentinel2Driver(),
     },
     {
-        "criteria": [
-            lambda prod: True
-            if (prod.collection or "").startswith("S1_SAR_")
-            else False
-        ],
+        "criteria": [lambda product: Sentinel1Driver.match(product, by="collection")],
         "driver": Sentinel1Driver(),
     },
     {
-        "criteria": [lambda prod: True],
+        "criteria": [lambda product: Sentinel2Driver.match(product, by="properties")],
+        "driver": Sentinel2Driver(),
+    },
+    {
+        "criteria": [lambda product: GenericDriver.match(product)],
         "driver": GenericDriver(),
     },
 ]
 
-
 # exportable content
 __all__ = ["DRIVERS", "DatasetDriver", "GenericDriver", "Sentinel2Driver"]

eodag/api/product/drivers/base.py

@@ -19,6 +19,7 @@
 
 import logging
 import re
+from abc import abstractmethod
 from typing import TYPE_CHECKING, Optional
 
 from typing_extensions import TypedDict
@@ -52,6 +53,19 @@ class DatasetDriver(metaclass=type):
     #: strip non-alphanumeric characters at the beginning and end of the key
     STRIP_SPECIAL_PATTERN = re.compile(r"^[^A-Z0-9]+|[^A-Z0-9]+$", re.IGNORECASE)
 
+    @staticmethod
+    @abstractmethod
+    def match(product: EOProduct, by: str = "*") -> bool:
+        """
+        Resolve if given product collection match with current driver
+        :paran  EOProduct  product as reference use to extract criteria
+        :paran  str        specific criteria match
+        :return bool       if given product match with current driver
+        """
+        raise TypeError(
+            "Can't call abstract methods match from base class DatasetDriver"
+        )
+
     def _normalize_key(self, key, eo_product):
         # default cleanup
         norm_key = key.replace(eo_product.properties.get("id", ""), "")

eodag/api/product/drivers/generic.py

@@ -19,8 +19,12 @@
 
 import logging
 import re
+from typing import TYPE_CHECKING
 
-from eodag.api.product.drivers.base import AssetPatterns, DatasetDriver
+from .base import AssetPatterns, DatasetDriver
+
+if TYPE_CHECKING:
+    from eodag.api.product._product import EOProduct
 
 logger = logging.getLogger("eodag.driver.generic")
 
@@ -71,3 +75,13 @@ class GenericDriver(DatasetDriver):
             "roles": ["auxiliary"],
         },
     ]
+
+    @staticmethod
+    def match(product: EOProduct, by: str = "*") -> bool:
+        """
+        Resolve if given product collection match with current driver
+        :paran  EOProduct  product as reference use to extract criteria
+        :paran  str        specific criteria match
+        :return bool       if given product match with current driver
+        """
+        return True

eodag/api/product/drivers/sentinel1.py

@@ -20,7 +20,7 @@
 import re
 from typing import TYPE_CHECKING
 
-from eodag.api.product.drivers.base import AssetPatterns, DatasetDriver
+from .base import AssetPatterns, DatasetDriver
 
 if TYPE_CHECKING:
     from eodag.api.product._product import EOProduct
@@ -100,3 +100,22 @@ def _normalize_key(self, key: str, eo_product: EOProduct) -> str:
             key = pattern.sub(replacement, key)
 
         return super()._normalize_key(key, eo_product)
+
+    @staticmethod
+    def match(product: EOProduct, by: str = "*") -> bool:
+        """
+        Resolve if given product collection match with current driver
+        :paran  EOProduct  product as reference use to extract criteria
+        :paran  str        specific criteria match
+        :return bool       if given product match with current driver
+        """
+        if (
+            by == "collection"
+            and hasattr(product, "collection")
+            and isinstance(product.collection, str)
+        ):
+            collection = product.collection.lower()
+            if collection.startswith("s1_sar_"):
+                return True
+
+        return False

eodag/api/product/drivers/sentinel2.py

@@ -20,7 +20,7 @@
 import re
 from typing import TYPE_CHECKING
 
-from eodag.api.product.drivers.base import AssetPatterns, DatasetDriver
+from .base import AssetPatterns, DatasetDriver
 
 if TYPE_CHECKING:
     from eodag.api.product._product import EOProduct
@@ -100,3 +100,41 @@ def _normalize_key(self, key: str, eo_product: EOProduct) -> str:
                         return norm_key
 
         return super()._normalize_key(key, eo_product)
+
+    @staticmethod
+    def match(product: EOProduct, by: str = "*") -> bool:
+        """
+        Resolve if given product collection match with current driver
+        :paran  EOProduct  product as reference use to extract criteria
+        :paran  str        specific criteria match
+        :return bool       if given product match with current driver
+        """
+        if (
+            by == "collection"
+            and hasattr(product, "collection")
+            and isinstance(product.collection, str)
+        ):
+
+            collection = product.collection.lower()
+            if collection.startswith("s2_msi_"):
+                return True
+
+        if (
+            by == "properties"
+            and hasattr(product, "properties")
+            and isinstance(product.properties, dict)
+        ):
+
+            constellation = product.properties.get("constellation", None)
+            if isinstance(constellation, str):
+                constellation = re.sub(r"([^a-z0-9]+)", "", constellation.lower())
+                if constellation == "sentinel2":
+                    return True
+
+            platform = product.properties.get("platform", None)
+            if platform is not None:
+                platform = re.sub(r"([^a-z0-9]+)", "", platform.lower())
+                if re.match("s2[a-z]", platform):
+                    return True
+
+        return False

tests/units/test_eoproduct_driver_base.py

@@ -0,0 +1,51 @@
+# -*- coding: utf-8 -*-
+# Copyright 2021, CS GROUP - France, http://www.c-s.fr
+#
+# This file is part of EODAG project
+#     https://www.github.com/CS-SI/EODAG
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import os
+import sys
+import unittest
+
+if __name__ == "__main__":
+    sys.path.insert(
+        0, os.path.normpath(os.path.join(os.path.dirname(__file__), "..", ".."))
+    )
+
+from tests.context import DatasetDriver, EOProduct
+
+
+class TestEOProductDriverBase(unittest.TestCase):
+    def test_driver_s2_init(self):
+        """Driver base is an abstract"""
+        try:
+            DatasetDriver.match(
+                EOProduct(
+                    "fake_provider",
+                    {"id": "9deb7e78-9341-5530-8fe8-f81fd99c9f0f"},
+                    collection="fake_collection",
+                )
+            )
+            self.fail("Abstract methods are not callable from abstract class")
+        except TypeError:
+            pass
+        except Exception as e:
+            self.fail("Unexpected exception: {}".format(str(e)))
+
+
+if __name__ == "__main__":
+    # Allow direct run test by run file
+    unittest.main()

tests/units/test_eoproduct_driver_generic.py

@@ -17,6 +17,13 @@
 # limitations under the License.
 
 import os
+import sys
+import unittest
+
+if __name__ == "__main__":
+    sys.path.insert(
+        0, os.path.normpath(os.path.join(os.path.dirname(__file__), "..", ".."))
+    )
 
 from tests import TEST_RESOURCES_PATH, EODagTestCase
 from tests.context import EOProduct, GenericDriver
@@ -25,9 +32,15 @@
 class TestEOProductDriverGeneric(EODagTestCase):
     def setUp(self):
         super(TestEOProductDriverGeneric, self).setUp()
+
+        # Drop specific fields could match with sentinel 2 driver
+        self.eoproduct_props["constellation"] = None
+        self.eoproduct_props["platform"] = None
+
         self.product = EOProduct(
             self.provider, self.eoproduct_props, collection="FAKE_COLLECTION"
         )
+        self.product.collection = "fake-collection"
         self.product.properties["title"] = os.path.join(
             TEST_RESOURCES_PATH,
             "products",
@@ -98,3 +111,8 @@ def test_driver_generic_guess_asset_key_and_roles(self):
             ),
             ("quick-look.jpg", ["overview"]),
         )
+
+
+if __name__ == "__main__":
+    # Allow direct run test by run file
+    unittest.main()

tests/units/test_eoproduct_driver_sentinel1.py

@@ -17,6 +17,13 @@
 # limitations under the License.
 
 import os
+import sys
+import unittest
+
+if __name__ == "__main__":
+    sys.path.insert(
+        0, os.path.normpath(os.path.join(os.path.dirname(__file__), "..", ".."))
+    )
 
 from tests import TEST_RESOURCES_PATH, EODagTestCase
 from tests.context import EOProduct, Sentinel1Driver
@@ -110,3 +117,8 @@ def test_driver_s1_guess_asset_key_and_roles(self):
             ),
             ("foo.bar.baz", ["auxiliary"]),
         )
+
+
+if __name__ == "__main__":
+    # Allow direct run test by run file
+    unittest.main()