feat: fetch files from ecfs if path starts with ec: or ectmp: (#585)

## Description

Fetch files from ecfs if path starts with ec: or ectmp: when creating
datasets


## What problem does this change solve?
<!-- Describe if it's a bugfix, new feature, doc update, or breaking
change -->

## What issue or task does this change relate to?
<!-- link to Issue Number -->

##  Additional notes ##
<!-- Include any additional information, caveats, or considerations that
the reviewer should be aware of. -->

***As a contributor to the Anemoi framework, please ensure that your
changes include unit tests, updates to any affected dependencies and
documentation, and have been tested in a parallel setting (i.e., with
multiple GPUs). As a reviewer, you are also responsible for verifying
these aspects and requesting changes if they are not adequately
addressed. For guidelines about those please refer to
https://anemoi.readthedocs.io/en/latest/***

By opening this pull request, I affirm that all authors agree to the
[Contributor License
Agreement.](https://github.com/ecmwf/codex/blob/main/Legal/contributor_license_agreement.md)

---------

Co-authored-by: Florian Pinault <[email protected]>
Co-authored-by: anaprietonem <[email protected]>
Co-authored-by: Ana Prieto Nemesio <[email protected]>

Author: 3 people

src/anemoi/datasets/create/ecfs.py

@@ -0,0 +1,29 @@
+# (C) Copyright 2026- Anemoi contributors.
+#
+# This software is licensed under the terms of the Apache Licence Version 2.0
+# which can be obtained at http://www.apache.org/licenses/LICENSE-2.0.
+#
+# In applying this licence, ECMWF does not waive the privileges and immunities
+# granted to it by virtue of its status as an intergovernmental organisation
+# nor does it submit to any jurisdiction.
+
+import hashlib
+import logging
+import os
+import subprocess
+import tempfile
+
+TMPDIR = None
+LOG = logging.getLogger(__name__)
+
+
+def get_ecfs_file(path: str) -> str:
+    global TMPDIR
+    if TMPDIR is None:
+        TMPDIR = tempfile.mkdtemp()
+
+    _, ext = os.path.splitext(path)
+    local_name = os.path.join(TMPDIR, hashlib.sha1(path.encode()).hexdigest() + ext)
+    LOG.info(f"Calling ecp {path} {local_name}")
+    subprocess.check_call(["ecp", path, local_name])
+    return local_name

src/anemoi/datasets/create/sources/grib.py

@@ -120,6 +120,7 @@ def _execute(
             The loaded dataset.
         """
         given_paths = path if isinstance(path, list) else [path]
+
         if flavour is not None:
             flavour = RuleBasedFlavour(flavour)
 
@@ -145,6 +146,12 @@ def _execute(
 
             for path in _expand(paths):
                 context.trace("📁", "PATH", path)
+
+                if isinstance(path, str) and (path.startswith("ec:") or path.startswith("ectmp:")):
+                    from anemoi.datasets.create.ecfs import get_ecfs_file
+
+                    path = get_ecfs_file(path)
+
                 s = from_source("file", path)
                 if flavour is not None:
                     s = flavour.map(s)

src/anemoi/datasets/create/sources/xarray_support/__init__.py

@@ -96,6 +96,11 @@ def load_one(
 
     context.trace(emoji, dataset, options, kwargs)
 
+    if isinstance(dataset, str) and (dataset.startswith("ec:") or dataset.startswith("ectmp:")):
+        from anemoi.datasets.create.ecfs import get_ecfs_file
+
+        dataset = get_ecfs_file(dataset)
+
     if isinstance(dataset, str) and dataset.endswith(".zarr"):
         # If the dataset is a zarr store, we need to use the zarr engine
         options["engine"] = "zarr"