Add an option to skip existing intermediate variables when aggregating recursivly

RELEASE_NOTES.md

@@ -1,5 +1,6 @@
 # Next Release
 
+- [#532](https://github.com/IAMconsortium/pyam/pull/532) Add an option to skip existing intermediate variables when aggregating recursivly
 - [#527](https://github.com/IAMconsortium/pyam/pull/527) Add an in-dataframe basic mathematical operations `subtract`, `add`, `multiply`, `divide`
 - [#519](https://github.com/IAMconsortium/pyam/pull/519) Enable explicit `label` and fix for non-string items in plot legend 
 

pyam/_aggregate.py

@@ -57,21 +57,36 @@ def _aggregate(df, variable, components=None, method=np.sum):
     return _group_and_agg(_df, [], method)
 
 
-def _aggregate_recursive(df, variable):
+def _aggregate_recursive(df, variable, recursive):
     """Recursive aggregation along the variable tree"""
 
     # downselect to components of `variable`, initialize list for aggregated (new) data
     _df = df.filter(variable=f"{variable}|*")
+    # keep variable at highest level if it exists
+    _df.append(df.filter(variable=variable), inplace=True)
     data_list = []
 
     # iterate over variables (bottom-up) and aggregate all components up to `variable`
     for d in reversed(range(find_depth(variable), max(find_depth(_df.variable)))):
         components = compress(_df.variable, find_depth(_df.variable, level=d + 1))
         var_list = set([reduce_hierarchy(v, -1) for v in components])
 
+        # collect already existing variables
+        intermediate_var = var_list.intersection(set(_df.variable))
+
         # a temporary dataframe allows to distinguish between full data and new data
         temp_df = _df.aggregate(variable=var_list)
+        # if intermediate variables exist, delete already existing entries in _data
+        if intermediate_var:
+            # index which doesn't exist in _df-index yet
+            _index = temp_df._data.index.difference(_df._data.index)
+            temp_df._data = temp_df._data[_index]
         _df.append(temp_df, inplace=True)
+        # check consistency of intermediate variable
+        if not recursive == "skip-validate":
+            if intermediate_var:
+                if _df.check_aggregate(intermediate_var):
+                    raise ValueError("Aggregated data is inconsistent.")
         data_list.append(temp_df._data)
 
     return pd.concat(data_list)

pyam/core.py

@@ -1163,7 +1163,12 @@ def normalize(self, inplace=False, **kwargs):
             return ret
 
     def aggregate(
-        self, variable, components=None, method="sum", recursive=False, append=False
+        self,
+        variable,
+        components=None,
+        method="sum",
+        recursive=False,
+        append=False,
     ):
         """Aggregate timeseries by components or subcategories within each region
 
@@ -1175,8 +1180,11 @@ def aggregate(
             Components to be aggregate, defaults to all subcategories of `variable`.
         method : func or str, optional
             Aggregation method, e.g. :func:`numpy.mean`, :func:`numpy.sum`, 'min', 'max'
-        recursive : bool, optional
+        recursive : bool or str, optional
             Iterate recursively (bottom-up) over all subcategories of `variable`.
+            If there are existing intermediate variables, it validates the aggregated
+            value.
+            If recursive='skip-validate', it skips the validation.
         append : bool, optional
             Whether to append aggregated timeseries data to this instance.
 
@@ -1191,15 +1199,17 @@ def aggregate(
         for individual components as 0.
         """
 
-        if recursive is True:
+        if recursive:
             if components is not None:
                 raise ValueError("Recursive aggregation cannot take `components`!")
             if method != "sum":
                 raise ValueError(
                     "Recursive aggregation only supported with `method='sum'`!"
                 )
 
-            _df = IamDataFrame(_aggregate_recursive(self, variable), meta=self.meta)
+            _df = IamDataFrame(
+                _aggregate_recursive(self, variable, recursive), meta=self.meta
+            )
         else:
             _df = _aggregate(self, variable, components=components, method=method)
 

tests/test_feature_aggregate.py

@@ -143,7 +143,7 @@ def test_aggregate_recursive(time_col):
     )
     df = IamDataFrame(data, model="model_a", scenario="scen_a", region="World")
     df2 = df.rename(scenario={"scen_a": "scen_b"})
-    df2.data.value *= 2
+    df2._data *= 2
     df.append(df2, inplace=True)
 
     # create object without variables to be aggregated
@@ -160,6 +160,29 @@ def test_aggregate_recursive(time_col):
     assert_iamframe_equal(df_minimal, df)
 
 
+@pytest.mark.parametrize("time_col", (("year"), ("time")))
+def test_aggregate_skip_intermediate(time_col):
+    # use the feature `recursive=skip-validate`
+    data = (
+        RECURSIVE_DF
+        if time_col == "year"
+        else RECURSIVE_DF.rename(DTS_MAPPING, axis="columns")
+    )
+    df = IamDataFrame(data, model="model_a", scenario="scen_a", region="World")
+    df2 = df.rename(scenario={"scen_a": "scen_b"})
+    df2._data *= 2
+    df.append(df2, inplace=True)
+
+    # create object without variables to be aggregated, but with intermediate variables
+    v = "Secondary Energy|Electricity"
+    agg_vars = [f"{v}{i}" for i in [""]]
+    df_minimal = df.filter(variable=agg_vars, scenario="scen_a", keep=False)
+
+    # append to `self`
+    df_minimal.aggregate(variable=v, recursive="skip-validate", append=True)
+    assert_iamframe_equal(df_minimal, df)
+
+
 @pytest.mark.parametrize(
     "variable, append", (("Primary Energy|Coal", "foo"), (False, True))
 )