feat: track origins of variable from dataset creation to inference

docs/cli/grib-index.rst

@@ -1,7 +1,7 @@
 .. _grib-index_command:
 
 Grib-index Command
-============
+==================
 
 The `grib-index` command is used to create an index file for GRIB files. The index file is then used
 by the `grib-index` :ref:`source <grib-index_source>`.

docs/datasets/building/filters.rst

@@ -0,0 +1,14 @@
+.. _filters:
+
+#########
+ Filters
+#########
+
+.. warning::
+
+   This is still a work-in-progress. Some of the filters may be renamed
+   later.
+
+Filters are used to modify the data or metadata in a dataset.
+
+See :ref:`install <anemoi-transform:filters>` for more information.

docs/datasets/building/sources/repeated-dates.rst

@@ -10,7 +10,7 @@ dates of the dataset.
 
 The general format of the `repeated-dates` source is:
 
-.. literalinclude:: yaml/repeated_dates1.yaml
+.. literalinclude:: yaml/repeated-dates1.yaml
    :language: yaml
 
 where ``source`` is any of the :ref:`operations <operations>` or

pyproject.toml

@@ -52,7 +52,6 @@ dependencies = [
   "anemoi-utils[provenance]>=0.4.32",
   "cfunits",
   "glom",
-  "jsonschema",
   "numcodecs<0.16",                   # Until we move to zarr3
   "numpy",
   "pyyaml",

src/anemoi/datasets/create/__init__.py

@@ -743,6 +743,7 @@ def _run(self) -> int:
         metadata["end_date"] = dates[-1].isoformat()
         metadata["frequency"] = frequency
         metadata["missing_dates"] = [_.isoformat() for _ in missing]
+        metadata["origins"] = self.minimal_input.origins
 
         metadata["version"] = VERSION
 

src/anemoi/datasets/create/input/action.py

@@ -8,20 +8,15 @@
 # nor does it submit to any jurisdiction.
 
 import logging
+from abc import ABC
+from abc import abstractmethod
 
 from anemoi.datasets.dates import DatesProvider
 
 LOG = logging.getLogger(__name__)
 
 
-class Action:
-    """An "Action" represents a single operation described in the yaml configuration, e.g. a source, a filter,
-    pipe, join, etc.
-
-    See :ref:`operations` for more details.
-
-    """
-
+class Action(ABC):
     def __init__(self, config, *path):
         self.config = config
         self.path = path
@@ -30,6 +25,13 @@ def __init__(self, config, *path):
             "data_sources",
         ), f"{self.__class__.__name__}: path must start with 'input' or 'data_sources': {path}"
 
+    @abstractmethod
+    def __call__(self, context, argument):
+        pass
+
+    def __repr__(self):
+        return f"{self.__class__.__name__}({'.'.join(str(x) for x in self.path)}, {self.config})"
+
 
 class Concat(Action):
     """The Concat contruct is used to concat different actions that are responsible
@@ -65,6 +67,7 @@ def __init__(self, config, *path):
 
         for i, item in enumerate(config):
 
+            assert "dates" in item, f"Value must contain the key 'dates' {item}"
             dates = item["dates"]
             filtering_dates = DatesProvider.from_config(**dates)
             action = action_factory({k: v for k, v in item.items() if k != "dates"}, *self.path, str(i))
@@ -186,7 +189,13 @@ def create_object(self, context, config):
         return create_datasets_source(context, config)
 
     def call_object(self, context, source, argument):
-        return source.execute(context.source_argument(argument))
+        result = source.execute(context.source_argument(argument))
+        return context.origin(result, self, argument)
+
+    def origin(self):
+        from .origin import Source
+
+        return Source(self.path[-1], self.config)
 
 
 class TransformSourceMixin:
@@ -197,6 +206,15 @@ def create_object(self, context, config):
 
         return create_transform_source(context, config)
 
+    def combine_origins(self, current, previous):
+        assert previous is None, f"Cannot combine origins, previous already exists: {previous}"
+        return current
+
+    def origin(self):
+        from .origin import Source
+
+        return Source(self.path[-1], self.config)
+
 
 class TransformFilterMixin:
     """Mixin class for filters defined in anemoi-transform"""
@@ -207,14 +225,16 @@ def create_object(self, context, config):
         return create_transform_filter(context, config)
 
     def call_object(self, context, filter, argument):
-        return filter.forward(context.filter_argument(argument))
+        result = filter.forward(context.filter_argument(argument))
+        return context.origin(result, self, argument)
 
+    def origin(self):
+        from .origin import Filter
 
-class FilterFunction(Function):
-    """Action to call a filter on the argument (e.g. rename, regrid, etc.)."""
+        return Filter(self.path[-1], self.config)
 
-    def __call__(self, context, argument):
-        return self.call(context, argument, context.filter_argument)
+    def combine_origins(self, current, previous):
+        return {"_apply": current, **(previous or {})}
 
 
 def _make_name(name, what):

src/anemoi/datasets/create/input/context/__init__.py

@@ -34,7 +34,7 @@ def register(self, data: Any, path: list[str]) -> Any:
 
         assert path[0] in ("input", "data_sources"), path
 
-        LOG.info(f"Registering data at path: {path}")
+        LOG.info(f"Registering data at path: {'.'.join(str(x) for x in path)}")
         self.results[tuple(path)] = data
         return data
 

src/anemoi/datasets/create/input/context/field.py

@@ -10,6 +10,8 @@
 
 from typing import Any
 
+from anemoi.transform.fields import new_field_with_metadata
+from anemoi.transform.fields import new_fieldlist_from_list
 from earthkit.data.core.order import build_remapping
 
 from ..result.field import FieldResult
@@ -52,3 +54,20 @@ def matching_dates(self, filtering_dates, group_of_dates: Any) -> Any:
         from anemoi.datasets.dates.groups import GroupOfDates
 
         return GroupOfDates(sorted(set(group_of_dates) & set(filtering_dates)), group_of_dates.provider)
+
+    def origin(self, data: Any, action: Any, action_arguments: Any) -> Any:
+
+        origin = action.origin()
+
+        result = []
+        for fs in data:
+            previous = fs.metadata("anemoi_origin", default=None)
+            fall_through = fs.metadata("anemoi_fall_through", default=False)
+            if fall_through:
+                # The field has pass unchanges in a filter
+                result.append(fs)
+            else:
+                anemoi_origin = origin.combine(previous, action, action_arguments)
+                result.append(new_field_with_metadata(fs, anemoi_origin=anemoi_origin))
+
+        return new_fieldlist_from_list(result)

src/anemoi/datasets/create/input/origin.py

@@ -0,0 +1,159 @@
+# (C) Copyright 2025 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 logging
+from abc import ABC
+
+LOG = logging.getLogger(__name__)
+
+
+class Origin(ABC):
+
+    def __init__(self, when="dataset-create"):
+        self.when = when
+
+    def __eq__(self, other):
+        if not isinstance(other, Origin):
+            return False
+        return self is other
+
+    def __hash__(self):
+        return id(self)
+
+
+def _un_dotdict(x):
+    if isinstance(x, dict):
+        return {k: _un_dotdict(v) for k, v in x.items()}
+
+    if isinstance(x, (list, tuple, set)):
+        return [_un_dotdict(a) for a in x]
+
+    return x
+
+
+class Pipe(Origin):
+    def __init__(self, s1, s2, when="dataset-create"):
+        super().__init__(when)
+        self.steps = [s1, s2]
+
+        assert s1 is not None, (s1, s2)
+        assert s2 is not None, (s1, s2)
+
+        if isinstance(s1, Pipe):
+            assert not isinstance(s2, Pipe), (s1, s2)
+            self.steps = s1.steps + [s2]
+
+    def combine(self, previous, action, action_arguments):
+        assert False, (self, previous)
+
+    def as_dict(self):
+        return {
+            "type": "pipe",
+            "steps": [s.as_dict() for s in self.steps],
+            "when": self.when,
+        }
+
+    def __repr__(self):
+        return " | ".join(repr(s) for s in self.steps)
+
+
+class Join(Origin):
+    def __init__(self, origins, when="dataset-create"):
+        assert isinstance(origins, (list, tuple, set)), origins
+        super().__init__(when)
+        self.steps = list(origins)
+
+        assert all(o is not None for o in origins), origins
+
+    def combine(self, previous, action, action_arguments):
+        assert False, (self, previous)
+
+    def as_dict(self):
+        return {
+            "type": "join",
+            "steps": [s.as_dict() for s in self.steps],
+            "when": self.when,
+        }
+
+    def __repr__(self):
+        return " & ".join(repr(s) for s in self.steps)
+
+
+class Source(Origin):
+    def __init__(self, name, config, when="dataset-create"):
+        super().__init__(when)
+        assert isinstance(config, dict), f"Config must be a dictionary {config}"
+        self.name = name
+        self.config = _un_dotdict(config)
+
+    def combine(self, previous, action, action_arguments):
+        assert previous is None, f"Cannot combine origins, previous already exists: {previous}"
+        return self
+
+    def as_dict(self):
+        return {
+            "type": "source",
+            "name": self.name,
+            "config": self.config,
+            "when": self.when,
+        }
+
+    def __repr__(self):
+        return f"{self.name}({id(self)})"
+
+
+class Filter(Origin):
+    def __init__(self, name, config, when="dataset-create"):
+        super().__init__(when)
+        assert isinstance(config, dict), f"Config must be a dictionary {config}"
+        self.name = name
+        self.config = _un_dotdict(config)
+        self._cache = {}
+
+    def combine(self, previous, action, action_arguments):
+
+        if previous is None:
+            # This can happen if the filter does not tag its output with an origin
+            # (e.g. a user plugin). In that case we try to get the origin from the action arguments
+            key = (id(action), id(action_arguments))
+            if key not in self._cache:
+
+                LOG.warning(f"No previous origin to combine with: {self}. Action: {action}")
+                LOG.warning(f"Connecting to action arguments {action_arguments}")
+                origins = set()
+                for k in action_arguments:
+                    o = k.metadata("anemoi_origin", default=None)
+                    if o is None:
+                        raise ValueError(
+                            f"Cannot combine origins, previous is None and action_arguments {action_arguments} has no origin"
+                        )
+                    origins.add(o)
+                if len(origins) == 1:
+                    self._cache[key] = origins.pop()
+                else:
+                    self._cache[key] = Join(origins)
+            previous = self._cache[key]
+
+        if previous in self._cache:
+            # We use a cache to avoid recomputing the same combination
+            return self._cache[previous]
+
+        self._cache[previous] = Pipe(previous, self)
+        return self._cache[previous]
+
+    def as_dict(self):
+        return {
+            "type": "filter",
+            "name": self.name,
+            "config": self.config,
+            "when": self.when,
+        }
+
+    def __repr__(self):
+        return f"{self.name}({id(self)})"