feat: abstracting accumulation

docs/howtos/create/05-create-accumulations.rst

@@ -0,0 +1,129 @@
+.. _create-accumulations:
+
+##########################################
+ Create a dataset with accumulated fields
+##########################################
+
+Many fields come as accumulations over time, e.g `tp` (total
+precipitations), `sd` (now depth) or `ssr` (surface shortwave
+radiation). Given one dataset, one may want to accumulate some of its
+fields on specific periods of time.
+
+This depends on the data's native format. For an accumulated field (say
+`tp` for simplicity), one needs to know:
+
+-  the `accumulation_period` over which to accumulate (e.g 6h).
+-  the desired `validity_time` at which accumulation stops and for which
+   the value is valid.
+-  the `data_accumulation_period`, that is the duration over which the
+   data is already accumulated.
+
+The resulting field is then "`tp` accumulated over `accumulation_period`
+hours up to `validity_time`". In a common case, dataset features, e.g.,
+1h-accumulated `tp` at a 1 hour frequency, and each raw file features
+`tp` as accumulated over the *last* hour. So having 6h-accumulated `tp`
+consists in taking all 6 files before (and including) `validity_time`
+and summing fields in them.
+
+The resulting accumulated field can be treated as a normal anemoi
+`source` in recipes (e.g, filters can be applied to the source).
+
+Note that depending on how your native dataset is built (e.g, your
+native files feature the accumulation on the *next* hour), the
+calculation can be very different. See $Subtleties below with the
+associated recipes.
+
+***************************************
+ Using accumulations in recipes : mars
+***************************************
+
+In the example below we see recipes to create accumulations from MARS
+data. To keep older recipes working, there are two equivalent ways to do
+so. The first one is a generic way working for MARS and grib-index
+sources.
+
+.. literalinclude:: yaml/recipe-accumulate-era.yaml
+
+That recipe will generate the following dataset:
+
+.. code:: bash
+
+   📦 Path       : recipe-accumulate.zarr
+   🔢 Format version: 0.30.0
+
+   📅 Start      : 2021-01-10 18:00
+   📅 End        : 2021-01-12 12:00
+   ⏰ Frequency  : 6h
+   🚫 Missing    : 0
+   🌎 Resolution : 20.0
+   🌎 Field shape: [162]
+
+   📐 Shape      : 8 × 2 × 1 × 162 (10.1 KiB)
+   💽 Size       : 23.2 KiB (23.2 KiB)
+   📁 Files      : 52
+
+      Index │ Variable │ Min │       Max │        Mean │      Stdev
+      ──────┼──────────┼─────┼───────────┼─────────────┼───────────
+         0  │ cp       │   0 │ 0.0110734 │ 0.000244731 │ 0.00103593
+         1  │ tp       │   0 │ 0.0333021 │  0.00058075 │ 0.00210331
+      ──────┴──────────┴─────┴───────────┴─────────────┴───────────
+   🔋 Dataset ready, last update 26 seconds ago.
+   📊 Statistics ready.
+
+The "legacy" way to do is the following (syntax is only slightly
+different)
+
+.. literalinclude:: yaml/recipe-accumulation-era.yaml
+
+The resulting dataset is:
+
+.. code:: bash
+
+   📦 Path       : recipe-accumulation.zarr
+   🔢 Format version: 0.30.0
+
+   📅 Start      : 2021-01-10 18:00
+   📅 End        : 2021-01-12 12:00
+   ⏰ Frequency  : 6h
+   🚫 Missing    : 0
+   🌎 Resolution : 20.0
+   🌎 Field shape: [9, 18]
+
+   📐 Shape      : 8 × 2 × 1 × 162 (10.1 KiB)
+   💽 Size       : 22.2 KiB (22.2 KiB)
+   📁 Files      : 52
+
+      Index │ Variable │ Min │       Max │        Mean │      Stdev
+      ──────┼──────────┼─────┼───────────┼─────────────┼───────────
+         0  │ cp       │   0 │ 0.0110734 │ 0.000244739 │ 0.00103593
+         1  │ tp       │   0 │ 0.0333023 │ 0.000580769 │ 0.00210332
+      ──────┴──────────┴─────┴───────────┴─────────────┴───────────
+   🔋 Dataset ready, last update 3 minutes ago.
+   📊 Statistics ready.
+
+Note that statitics for the two datasets are equal up to `1e-6`, this is
+due to rounding errors that can accumulate. Larger discrepancies are a
+sign something might be wrong.
+
+*********************************************
+ Using accumulations in recipes : grib files
+*********************************************
+
+If your data source is grib files, you can use a grib-index as a source.
+First create a `grib-index
+<https://anemoi.readthedocs.io/projects/datasets/en/latest/howtos/create/01-grib-data.html#using-an-index-file>`_
+
+that creates a database to query fields. Then, say we want to accumulate
+3h-data over 6h.
+
+.. literalinclude:: yaml/recipe-accumulate-gribindex.yaml
+
+Note that we also added a filter at the end of the recipe to rename `tp`
+to `tp_6h`. The frequency of the dataset is `1h`, so the accumulation is
+a moving window.
+
+************
+ Subtleties
+************
+
+Some datasets (such as ERA5) feature

docs/howtos/create/yaml/recipe-accumulate-era.yaml

@@ -0,0 +1,17 @@
+dates:
+  start: 2021-01-10 18:00:00
+  end: 2021-01-12 12:00:00
+  frequency: 6h
+
+input:
+    accumulate:
+      source:
+        mars:
+          expver: "0001"
+          class: ea
+          stream: oper
+          grid: 20./20.
+          levtype: sfc
+          param: [ tp, cp]
+          accumulation_period: 6
+          data_accumulation_period: 6 # this argument will be ignored because of ERA-like accumulations

docs/howtos/create/yaml/recipe-accumulate-gribindex.yaml

@@ -0,0 +1,18 @@
+dates:
+  end: 2020-01-06 12:00:00+00:00
+  start: 2020-01-03 06:00:00+00:00
+  frequency: 1h
+
+input:
+  pipe:
+    - accumulate:
+        source:
+          grib-index:
+            indexdb: /path/to/gribindex.db
+            levtype: sfc
+            param:
+            - tp
+            accumulation_period: 6
+            data_accumulation_period: 3
+    - rename:
+        tp: tp_6h

docs/howtos/create/yaml/recipe-accumulation-era.yaml

@@ -0,0 +1,14 @@
+dates:
+  start: 2021-01-10 18:00:00
+  end: 2021-01-12 12:00:00
+  frequency: 6h
+
+input:
+    accumulations:
+      expver: "0001"
+      class: ea
+      stream: oper
+      grid: 20./20.
+      levtype: sfc
+      param: [ tp, cp]
+      accumulation_period: 6

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

@@ -0,0 +1,126 @@
+# (C) Copyright 2024 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 functools import cached_property
+from typing import Any
+from typing import Dict
+from typing import List
+
+from earthkit.data import FieldList
+
+from ...dates.groups import GroupOfDates
+from ..sources.accumulations2 import accumulations
+from .action import Action
+from .action import action_factory
+from .misc import _tidy
+from .misc import assert_fieldlist
+from .result import Result
+from .template import notify_result
+from .trace import trace_datasource
+from .trace import trace_select
+
+LOG = logging.getLogger(__name__)
+
+
+class AccumulationResult(Result):
+    """Represents a result that accumulates multiple fields.
+
+    Attributes
+    ----------
+    context : object
+        The context object.
+    action_path : list
+        The action path.
+    group_of_dates : GroupOfDates
+        The group of dates.
+    results : List[Result]
+        The list of results.
+    """
+
+    def __init__(
+        self,
+        context: object,
+        action_path: list,
+        group_of_dates: GroupOfDates,
+        source: Any,
+        request: Dict[str, Any],
+        **kwargs: Any,
+    ) -> None:
+        """Initializes a AccumulationResult instance.
+
+        Parameters
+        ----------
+        context : object
+            The context object.
+        action_path : list
+            The action path.
+        group_of_dates : GroupOfDates
+            The group of dates.
+        source: Any
+            The original source used to perform accumulation (as action_factory)
+        request: Dict[str,Any]
+            The description of the accumulate request
+        """
+        super().__init__(context, action_path, group_of_dates)
+        self.source: Any = source
+        self.request: Dict[str, Any] = request
+
+    @cached_property
+    @assert_fieldlist
+    @notify_result
+    @trace_datasource
+    def datasource(self) -> FieldList:
+        """Returns the combined datasource from all results."""
+        ds = accumulations(self.context, self.group_of_dates, self.source, **self.request)
+
+        return _tidy(ds)
+
+    def __repr__(self) -> str:
+        """Returns a string representation of the AccumulationResult instance."""
+        content: str = f"AccumulationRsult({self.source})"
+        return self._repr(content)
+
+
+class AccumulationAction(Action):
+    """An Action implementation that selects and transforms a group of dates."""
+
+    def __init__(self, context: Any, action_path: List[str], source: Dict[str, Any]) -> None:
+        """Initialize AccumulationAction.
+
+        Parameters
+        ----------
+            context: Any
+                The context needed to initialize the action
+            action_path: List[str]
+                The action path to initialize the action.
+            source: Dict[str, Any]
+                The configuration describing the data source.
+        """
+        super().__init__(context, action_path, source)
+
+        self.source: Any = action_factory(source, context, action_path + ["source"])
+        self.request = source[list(source.keys())[0]]
+
+    @trace_select
+    def select(self, group_of_dates: GroupOfDates) -> AccumulationResult:
+        """Select and transform the group of dates.
+
+        Parameters
+        ----------
+            group_of_dates: GroupOfDates
+                The group of dates to select.
+
+        Returns
+        -------
+        AccumulationResult
+            The result of the accumulate operation.
+        """
+
+        return AccumulationResult(self.context, self.action_path, group_of_dates, self.source, self.request)

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

@@ -17,6 +17,7 @@
 from ...dates.groups import GroupOfDates
 from .context import Context
 from .template import substitute
+from .trace import trace
 
 LOG = logging.getLogger(__name__)
 
@@ -193,6 +194,18 @@ def __init__(self, /, order_by: str, flatten_grid: bool, remapping: dict[str, An
         self.remapping = build_remapping(remapping)
         self.use_grib_paramid = use_grib_paramid
 
+    def trace(self, emoji: str, *args: Any) -> None:
+        """Traces the given arguments with an emoji.
+
+        Parameters
+        ----------
+        emoji : str
+            The emoji to use.
+        *args : Any
+            The arguments to trace.
+        """
+        trace(emoji, *args)
+
 
 def action_factory(config: dict[str, Any], context: ActionContext, action_path: list[str]) -> Action:
     """Factory function to create an Action instance based on the configuration.
@@ -211,15 +224,14 @@ def action_factory(config: dict[str, Any], context: ActionContext, action_path:
     Action
         The created Action instance.
     """
+    from .accumulate import AccumulationAction
     from .concat import ConcatAction
     from .data_sources import DataSourcesAction
     from .function import FunctionAction
     from .join import JoinAction
     from .pipe import PipeAction
     from .repeated_dates import RepeatedDatesAction
 
-    # from .data_sources import DataSourcesAction
-
     assert isinstance(context, Context), (type, context)
     if not isinstance(config, dict):
         raise ValueError(f"Invalid input config {config}")
@@ -246,6 +258,7 @@ def action_factory(config: dict[str, Any], context: ActionContext, action_path:
         "function": FunctionAction,
         "repeated_dates": RepeatedDatesAction,
         "repeated-dates": RepeatedDatesAction,
+        "accumulate": AccumulationAction,
     }.get(key)
 
     if cls is None:

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

@@ -349,8 +349,7 @@ def get_cube(self) -> Any:
             LOG.debug(f"Sorting done in {seconds_to_human(time.time()-start)}.")
         except ValueError:
             self.explain(ds, order_by, remapping=remapping, patches=patches)
-            # raise ValueError(f"Error in {self}")
-            exit(1)
+            raise ValueError(f"Error in {self}")
 
         if LOG.isEnabledFor(logging.DEBUG):
             LOG.debug("Cube shape: %s", cube)
@@ -517,7 +516,6 @@ def explain(self, ds: Any, *args: Any, remapping: Any, patches: Any) -> None:
         print()
         print("❌" * 40)
         print()
-        exit(1)
 
     def _repr(self, *args: Any, _indent_: str = "\n", **kwargs: Any) -> str:
         """Return the string representation of the Result instance.