feat: Add entrypoint for anemoi-datasets to PyEarthTools

docs/api/api.md

@@ -24,6 +24,7 @@ data/data_index
 data/data_api
 pipeline/pipeline_index
 pipeline/pipeline_api
+pipeline/pipeline_entrypoints
 training/training_index
 training/training_api
 tutorial/tutorial_index

docs/api/pipeline/pipeline_entrypoints.md

@@ -0,0 +1,51 @@
+# Pipeline Entrypoints
+
+As `PyEarthTools` pipelines propose a generic way to load and prepare various earth system datasets, it is possible to use
+a pipeline as a source for [anemoi-datasets](https://anemoi.readthedocs.io/projects/datasets/en/latest/).
+
+## Example
+
+Below is a minimal example of using a `PyEarthTools` pipeline to load data and prepare it for `anemoi`, please see the `anemoi` docs
+for more information on the `datasets` config.
+
+### Create the Pipeline
+
+.. code-block:: python
+
+    import pyearthtools.data
+    import pyearthtools.pipeline
+
+    pipeline = pyearthtools.pipeline.Pipeline(
+        pyearthtools.data.download.arcoera5.ARCOERA5(['t2m', 'u10', 'v10']),
+        pyearthtools.pipeline.operations.xarray.values.FillNan()
+    )
+    pipeline.save('/PATH/TO/PIPELINE.yaml')
+
+### Create the anemoi-datasets config
+
+.. code-block:: yaml
+
+    name: pyearthtools_to_anemoi
+    description: PyEarthTools Pipeline converted to Anemoi
+    attribution: PyEarthTools
+
+    dates:
+        start: '2025-11-10T00:00:00'
+        end: '2025-11-12T00:00:00'
+        frequency: 1h
+
+    input:
+        pyearthtools: # Use the pyearthtools input object
+            pipeline: /PATH/TO/PIPELINE.yaml
+
+### Run anemoi-datasets
+
+.. code-block:: bash
+
+    anemoi-datasets create /path/to/anemoi/dataset.yaml
+
+## Contract
+
+The expected contract and result from the `PyEarthTools` pipeline is to return an `xarray` object of a single time index.
+
+Both tools provide methods to modify the metadata of the data, and should be used accordingly to prepare for downstream uses.

packages/pipeline/pyproject.toml

@@ -29,6 +29,10 @@ dependencies = [
 
 dynamic = ["version", "readme"]
 
+[project.entry-points]
+# Add PyEarthTools as an anemoi datasets source
+"anemoi.datasets.create.sources".pyearthtools = "pyearthtools.pipeline.entrypoints.anemoi:pyearthtoolsSource"
+
 [project.optional-dependencies]
 distributed = [
   "dask",

packages/pipeline/src/pyearthtools/pipeline/entrypoints/anemoi.py

@@ -0,0 +1,48 @@
+from functools import cached_property
+from pathlib import Path
+
+from pyearthtools.pipeline import load
+from pyearthtools.pipeline import Pipeline
+
+import earthkit.data as ekd
+from anemoi.datasets.create.source import Source
+from anemoi.datasets.create.typing import DateList
+
+
+class pyearthtoolsSource(Source):
+    emoji = "🌏"  # For tracing
+
+    def __init__(self, context, pipeline: str | Path):
+        """Initialise the source.
+
+        Parameters
+        ----------
+        context : Any
+            The context for the data source.
+        pipeline: str
+            The path to the pyearthtools pipeline file.
+        """
+        super().__init__(context)
+        self._pyearthtools_pipeline = pipeline
+
+    @cached_property
+    def pipeline(self) -> Pipeline:
+        return load(self._pyearthtools_pipeline)
+
+    def execute(self, dates: DateList) -> ekd.FieldList:
+        """Execute the source.
+
+        Parameters
+        ----------
+        dates : DateList
+            The input dates.
+
+        Returns
+        -------
+        ekd.FieldList
+            The output data.
+        """
+        fields = []
+        for date in dates:
+            fields.extend(ekd.from_object(self.pipeline[date.isoformat()]))  # type: ignore
+        return ekd.FieldList.from_fields(fields)