Add a function to initialize an IamDataFrame form an ixmp4 Run

Author: danielhuppmann

pyam/__init__.py

@@ -13,7 +13,7 @@
     read_datapackage,
 )
 from pyam.iiasa import lazy_read_iiasa, read_iiasa
-from pyam.ixmp4 import read_ixmp4
+from pyam.ixmp4 import read_ixmp4, read_ixmp4_run
 from pyam.netcdf import read_netcdf
 from pyam.run_control import run_control
 from pyam.statistics import Statistics

pyam/ixmp4.py

@@ -70,6 +70,48 @@ def read_ixmp4(
     return IamDataFrame(data, meta=meta, index=index)
 
 
+def read_ixmp4_run(
+    run: ixmp4.Run,
+    region: str | list[str] | None = None,
+    variable: str | list[str] | None = None,
+    unit: str | list[str] | None = None,
+    year: int | list[int] | None = None,
+):
+    """Return timeseries data and meta indicatos from an ixmp4 run as IamDataFrame
+
+    Parameters
+    ----------
+    run : :class:`ixmp4.Run`
+        An ixmp4 run instance from which to read timeseries data and meta indicators
+    region, variable, unit : str or list of str, optional
+        Filter timeseries data by these dimensions.
+    year : int or list of int, optional
+        Filter timeseries data by time domain.
+    """
+    from pyam import IamDataFrame
+
+    if year is not None:
+        raise NotImplementedError("Filter by 'year' not implemented in ixmp4.")
+
+    meta = pd.DataFrame.from_dict(run.meta, orient="index").T
+    meta.index = pd.MultiIndex.from_tuples(
+        [(run.model.name, run.scenario.name)],
+        name=["model", "scenario"]
+    )
+    meta["version"] = run.version
+
+    return IamDataFrame(
+        data=run.iamc.tabulate(
+            region=region,
+            variable=variable,
+            unit=unit,
+        ),
+        meta=meta,
+        model=run.model.name,
+        scenario=run.scenario.name,
+    )
+
+
 def write_to_ixmp4(platform: ixmp4.Platform | str, df, checkpoint_message: str):
     """Save all scenarios as new default runs in an ixmp4 platform database instance
 

tests/test_ixmp4.py

@@ -3,7 +3,7 @@
 from ixmp4.core.unit import UnitModel
 
 import pyam
-from pyam import read_ixmp4
+from pyam import read_ixmp4, read_ixmp4_run
 from pyam.testing import assert_iamframe_equal
 
 
@@ -109,3 +109,19 @@ def test_ixmp4_reserved_columns(test_platform, test_df_year, drop_meta):
 
     # version is included when reading again from the platform
     assert_iamframe_equal(test_df_year, pyam.read_ixmp4(test_platform))
+
+
+def test_ixmp4_read_run(test_platform, test_df):
+    """Initialize an IamDataFrame from an ixmp4 run"""
+
+    # write to platform
+    test_df.to_ixmp4(platform=test_platform)
+
+    # get a run and cast to an IamDataFrame
+    run = test_platform.runs.get("model_a", "scen_a")
+    obs = read_ixmp4_run(run)
+
+    # assert that the returned object and the original IamDataFrame are equal
+    exp = test_df.filter(scenario="scen_a")
+    exp.set_meta(1, "version")  # the IamDataFrame read from ixmp4 includes the version
+    pyam.assert_iamframe_equal(exp, obs)