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

@@ -458,6 +458,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

@@ -11,6 +11,8 @@
 from PIL.PngImagePlugin import PngInfo
 from prov.model import ProvDerivation, ProvDocument
 
+from esmvalcore.io.protocol import DataElement
+
 from ._version import __version__
 
 logger = logging.getLogger(__name__)
@@ -109,7 +111,7 @@ class TrackedFile:
 
     def __init__(
         self,
-        filename,
+        filename: Path | DataElement,
         attributes=None,
         ancestors=None,
         prov_filename=None,
@@ -118,7 +120,7 @@ def __init__(
 
         Arguments
         ---------
-        filename: str
+        filename:
             Path to the file on disk.
         attributes: dict
             Dictionary with facets describing the file. If set to None, this
@@ -130,9 +132,11 @@ 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
         self.attributes = copy.deepcopy(attributes)

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]:
@@ -916,7 +904,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 +1330,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(