Adding intake asset plugin.

Author: rhysrevans3

extraction_methods/plugins/intake_esm_assets.py

@@ -51,6 +51,7 @@
 # Python imports
 import logging
 from datetime import datetime
+from pathlib import Path
 
 # Thirdparty imports
 import intake
@@ -87,15 +88,17 @@ def open_catalog(self):
         return catalog
 
     def update_search_kwargs(self, body:dict):
-      for search_kwarg_key, search_kwarg_value in search_kwargs.items():
-        if self.search_kwarg_value[0] == self.exists_key:
-                self.search_kwargs[search_kwarg_key] = body[self.search_kwarg_value[1:]]
+      for search_kwarg_key, search_kwarg_value in self.search_kwargs.items():
+        if search_kwarg_value[0] == self.exists_key:
+                self.search_kwargs[search_kwarg_key] = body[search_kwarg_value[1:]]
 
     def run(self, body: dict, **kwargs) -> dict:
 
         self.update_search_kwargs(body)
         catalog = self.open_catalog()
 
+        assets = body.get("assets", {})
+
         for _, row in catalog.df.iterrows():
             href = getattr(row, self.object_attr)
             asset = {
@@ -106,7 +109,7 @@ def run(self, body: dict, **kwargs) -> dict:
                 for extraction_method in self.extraction_methods:
                     asset = extraction_method.run(asset)
 
-            assets[Path(path).name] = asset
+            assets[Path(href).name] = asset
 
         body["assets"] = assets
 

extraction_methods/plugins/json_file.py

@@ -14,12 +14,12 @@
 
 # Python imports
 from collections import defaultdict
+import os
 import json
 import logging
 from typing import Optional
 
 from extraction_methods.core.extraction_method import ExtractionMethod
-from extraction_methods.core.types import SpatialExtent, TemporalExtent
 
 LOGGER = logging.getLogger(__name__)
 
@@ -51,27 +51,33 @@ class JsonFileExtract(ExtractionMethod):
     """
 
     def get_facet_values(self) -> list:
-        output = defaultdict(list)
+        output = defaultdict(set)
 
         for filepath in os.listdir(self.dirpath):
 
-            with open(filepath, "r") as file:
+            with open(os.path.join(self.dirpath, filepath), "r") as file:
                 item = json.load(file)
 
-                item_properties = item["body"]["properties"]
+                item_properties = item["properties"]
 
                 for facet in self.terms:
                     if facet in item_properties:
-                        output[facet].extend(item_properties[facet])
+                        if isinstance(item_properties[facet], list):
+                            output[facet].update(item_properties[facet])
+                        else:
+                            output[facet].add(item_properties[facet])
+
+        for facet in self.terms:
+            output[facet] = list(output[facet])
 
         return output
 
     @staticmethod
-    def get_spatial_extent(item_list: list) -> Optional[SpatialExtent]:
+    def get_spatial_extent(item_list: list) -> dict:
         ...
 
     @staticmethod
-    def get_temporal_extent(item_list: list) -> Optional[TemporalExtent]:
+    def get_temporal_extent(item_list: list) -> dict:
         start_datetime = []
         end_datetime = []
         datetime = []
@@ -91,7 +97,7 @@ def get_extent(self, file_id: str) -> dict:
             file_data = json.load(file)
 
             for item in file_data:
-                if item["body"]["collection_id"] == file_id:
+                if item["collection_id"] == file_id:
                     item_list.append(item)
 
         # spatial_extent = self.get_spatial_extent(item_list)
@@ -100,7 +106,7 @@ def get_extent(self, file_id: str) -> dict:
     def run(self, body: dict, **kwargs) -> dict:
         output = self.get_facet_values()
 
-        if values:
+        if output:
             body |= output
 
         # No need to include extents since the example scanner has none.

requirements.txt

@@ -14,6 +14,7 @@ elasticsearch==7.13.1
 fsspec==2021.10.0
 gcsfs==2021.10.0
 idna==2.10
+intake-esm==2024.2.6
 imagesize==1.2.0
 Jinja2==3.0.1
 jmespath==0.10.0

setup.py

@@ -71,7 +71,7 @@
             "lambda = extraction_methods.plugins.lambda:LambdaExtract",
             "netcdf = extraction_methods.plugins.netcdf:NetCDFfExtract",
             "asset_aggregator = extraction_methods.plugins.asset_aggregator:AssetAggregatorExtract",
-            "intake_asset = extraction_methods.plugins.intake_esm_assets:IntakeESMAssetExtract",
+            "intake_assets = extraction_methods.plugins.intake_esm_assets:IntakeESMAssetExtract",
         ],
         "extraction_methods.header.backends": [
             "ncml = extraction_methods.plugins.header.backends.ncml:NcMLBackend",