Add support for intake-esgf

Author: bouweandela

doc/api/esmvalcore.io.intake_esgf.rst

@@ -0,0 +1,5 @@
+Load data using intake-esgf
+===========================
+
+.. automodule:: esmvalcore.io.intake_esgf
+    :no-inherited-members:

doc/api/esmvalcore.io.protocol.rst

@@ -0,0 +1,5 @@
+Protocols for accessing data
+============================
+
+.. automodule:: esmvalcore.io.protocol
+    :no-inherited-members:

doc/api/esmvalcore.io.rst

@@ -0,0 +1,14 @@
+Access data from any source
+===========================
+
+ESMValCore supports a modular system for reading data from various data sources.
+In the future, this module may be extended with support for writing output data.
+
+The interface is defined in the :mod:`esmvalcore.io.protocol` module and
+the other modules here provide an implementation for a particular data source.
+
+.. toctree::
+   :maxdepth: 1
+
+   esmvalcore.io.protocol
+   esmvalcore.io.intake_esgf

doc/api/esmvalcore.rst

@@ -14,6 +14,7 @@ library. This section documents the public API of ESMValCore.
    esmvalcore.dataset
    esmvalcore.esgf
    esmvalcore.exceptions
+   esmvalcore.io
    esmvalcore.iris_helpers
    esmvalcore.local
    esmvalcore.preprocessor

doc/conf.py

@@ -461,6 +461,7 @@
     'dask': ('https://docs.dask.org/en/stable/', None),
     'distributed': ('https://distributed.dask.org/en/stable/', None),
     'iris': ('https://scitools-iris.readthedocs.io/en/stable/', None),
+    'intake_esgf': ('https://intake-esgf.readthedocs.io/en/stable/', None),
     'esmf_regrid': ('https://iris-esmf-regrid.readthedocs.io/en/stable/', None),
     'matplotlib': ('https://matplotlib.org/stable/', None),
     'ncdata': ('https://ncdata.readthedocs.io/en/stable/', None),

doc/configurations

@@ -0,0 +1 @@
+../esmvalcore/config/configurations

environment.yml

@@ -20,6 +20,7 @@ dependencies:
   - fire
   - geopy
   - humanfriendly
+  - intake-esgf
   - intake-esm
   - iris >=3.12.2  # https://github.com/SciTools/iris/issues/6417
   - iris-esmf-regrid >=0.11.0

esmvalcore/_provenance.py

@@ -118,8 +118,8 @@ def __init__(
 
         Arguments
         ---------
-        filename: str
-            Path to the file on disk.
+        filename: :obj:`pathlib.Path` or :obj:`esmvalcore.io.protocol.DataElement`
+            Path or data element containing the data described by the provenance.
         attributes: dict
             Dictionary with facets describing the file. If set to None, this
             will be read from the file when provenance is initialized.
@@ -130,11 +130,15 @@ def __init__(
             differ from `filename` if the file was moved before resuming
             processing.
         """
-        self._filename = filename
+        self._filename = (
+            str(filename) if isinstance(filename, Path) else filename.name
+        )
         if prov_filename is None:
-            self.prov_filename = filename
+            self.prov_filename = self._filename
         else:
             self.prov_filename = prov_filename
+        # TODO: ensure global attributes are recorded for input data if they're
+        # not netcdf files.
         self.attributes = copy.deepcopy(attributes)
 
         self.provenance = None
@@ -167,20 +171,21 @@ def copy_provenance(self):
         if self.provenance is None:
             msg = f"Provenance of {self} not initialized"
             raise ValueError(msg)
-        new = TrackedFile(self.filename, self.attributes)
+        new = TrackedFile(Path(self.filename), self.attributes)
         new.provenance = copy.deepcopy(self.provenance)
         new.entity = new.provenance.get_record(self.entity.identifier)[0]
         new.activity = new.provenance.get_record(self.activity.identifier)[0]
         return new
 
     @property
-    def filename(self):
-        """Filename."""
+    def filename(self) -> str:
+        """Name of data described by this provenance document."""
         return self._filename
 
     @property
     def provenance_file(self):
-        """Filename of provenance."""
+        """Filename of provenance file."""
+        # This may not work well if filename is the instance_id.
         return os.path.splitext(self.filename)[0] + "_provenance.xml"
 
     def initialize_provenance(self, activity):

esmvalcore/_recipe/check.py

@@ -7,6 +7,7 @@
 import os
 import subprocess
 from functools import partial
+from pathlib import Path
 from pprint import pformat
 from shutil import which
 from typing import TYPE_CHECKING, Any
@@ -15,6 +16,7 @@
 import yamale
 
 import esmvalcore.preprocessor
+from esmvalcore.esgf import ESGFFile
 from esmvalcore.exceptions import InputFilesNotFound, RecipeError
 from esmvalcore.local import _get_start_end_year, _parse_period
 from esmvalcore.preprocessor import TIME_PREPROCESSORS, PreprocessingTask
@@ -33,7 +35,6 @@
 
 if TYPE_CHECKING:
     from collections.abc import Iterable, Sequence
-    from pathlib import Path
 
     from esmvalcore._task import TaskSet
     from esmvalcore.dataset import Dataset
@@ -241,6 +242,8 @@ def data_availability(dataset: Dataset, log: bool = True) -> None:
     available_years: set[int] = set()
 
     for file in input_files:
+        if not isinstance(file, Path | ESGFFile):
+            return
         start, end = _get_start_end_year(file)
         available_years.update(range(start, end + 1))
 

esmvalcore/_recipe/recipe.py

@@ -60,6 +60,7 @@
     from collections.abc import Iterable, Sequence
 
     from esmvalcore.config import Session
+    from esmvalcore.io.protocol import DataElement
     from esmvalcore.typing import Facets
 
 logger = logging.getLogger(__name__)
@@ -328,20 +329,12 @@ def _update_weighting_settings(
     _exclude_dataset(settings, facets, "weighting_landsea_fraction")
 
 
-def _add_to_download_list(dataset: Dataset) -> None:
-    """Add the files of `dataset` to `DOWNLOAD_FILES`."""
-    for i, file in enumerate(dataset.files):
-        if isinstance(file, esgf.ESGFFile):
-            DOWNLOAD_FILES.add(file)
-            dataset.files[i] = file.local_file(dataset.session["download_dir"])
-
-
 def _schedule_for_download(datasets: Iterable[Dataset]) -> None:
     """Schedule files for download."""
     for dataset in datasets:
-        _add_to_download_list(dataset)
+        DOWNLOAD_FILES.update(dataset.files)
         for supplementary_ds in dataset.supplementaries:
-            _add_to_download_list(supplementary_ds)
+            DOWNLOAD_FILES.update(supplementary_ds.files)
 
 
 def _log_input_files(datasets: Iterable[Dataset]) -> None:
@@ -367,12 +360,7 @@ def _log_input_files(datasets: Iterable[Dataset]) -> None:
 
 def _get_files_str(dataset: Dataset) -> str:
     """Get nice string representation of all files of a dataset."""
-    return "\n".join(
-        f"  {f}"
-        if f.exists()  # type: ignore
-        else f"  {f} (will be downloaded)"
-        for f in dataset.files
-    )
+    return "\n".join(f"  {f}" for f in dataset.files)
 
 
 def _check_input_files(input_datasets: Iterable[Dataset]) -> set[str]:
@@ -455,10 +443,7 @@ def _get_common_attributes(
 
     # Ensure that attributes start_year and end_year are always available if at
     # least one of the input datasets defines it
-    if "timerange" in attributes:
-        start_year, end_year = _parse_period(attributes["timerange"])
-        attributes["start_year"] = int(str(start_year[0:4]))
-        attributes["end_year"] = int(str(end_year[0:4]))
+    _set_start_end_year(attributes)
 
     return attributes
 
@@ -722,7 +707,7 @@ def _get_preprocessor_products(
     )
 
     for product in products:
-        _set_start_end_year(product)
+        _set_start_end_year(product.attributes)
         product.check()
 
     return products
@@ -782,18 +767,18 @@ def _configure_multi_product_preprocessor(
 
     for product in multimodel_products | ensemble_products:
         product.check()
-        _set_start_end_year(product)
+        _set_start_end_year(product.attributes)
 
 
-def _set_start_end_year(product: PreprocessorFile) -> None:
+def _set_start_end_year(attributes: dict[str, Any]) -> None:
     """Set the attributes `start_year` and `end_year`.
 
     These attributes are used by many diagnostic scripts in ESMValTool.
     """
-    if "timerange" in product.attributes:
-        start_year, end_year = _parse_period(product.attributes["timerange"])
-        product.attributes["start_year"] = int(str(start_year[0:4]))
-        product.attributes["end_year"] = int(str(end_year[0:4]))
+    if "timerange" in attributes:
+        start_year, end_year = _parse_period(attributes["timerange"])
+        attributes["start_year"] = int(str(start_year[0:4]))
+        attributes["end_year"] = int(str(end_year[0:4]))
 
 
 def _update_preproc_functions(
@@ -916,7 +901,7 @@ def __init__(
         # Clear the global variable containing the set of files to download
         DOWNLOAD_FILES.clear()
         USED_DATASETS.clear()
-        self._download_files: set[esgf.ESGFFile] = set()
+        self._download_files: set[DataElement] = set()
         self.session = session
         self.session["write_ncl_interface"] = self._need_ncl(
             raw_recipe["diagnostics"],
@@ -1342,8 +1327,8 @@ def run(self) -> None:
         filled_recipe = self.write_filled_recipe()
 
         # Download required data
-        if self.session["search_esgf"] != "never":
-            esgf.download(self._download_files, self.session["download_dir"])
+        for file in self._download_files:
+            file.prepare()
 
         self.tasks.run(max_parallel_tasks=self.session["max_parallel_tasks"])
         logger.info(