Enable populate_test_platform() with IXMP4Backend

ixmp/backend/base.py

@@ -10,42 +10,14 @@
 import pandas as pd
 
 # TODO Import this from typing when dropping Python 3.11
-from typing_extensions import TypedDict, Unpack
+from typing_extensions import Unpack
 
 from ixmp.backend import ItemType
 from ixmp.backend.io import s_read_excel, s_write_excel, ts_read_file
 from ixmp.core.platform import Platform
 from ixmp.core.scenario import Scenario
 from ixmp.core.timeseries import TimeSeries
-
-
-# These are based on existing calls within ixmp
-class WriteFiltersKwargs(TypedDict, total=False):
-    scenario: Scenario | list[str]
-    model: list[str]
-    variable: list[str]
-    unit: list[str]
-    region: list[str]
-    default: bool
-    export_all_runs: bool
-
-
-class WriteKwargs(TypedDict, total=False):
-    filters: WriteFiltersKwargs
-    record_version_packages: list[str]
-
-
-class ReadKwargs(TypedDict, total=False):
-    filters: WriteFiltersKwargs
-    firstyear: Optional[Any]
-    lastyear: Optional[Any]
-    add_units: bool
-    init_items: bool
-    commit_steps: bool
-    check_solution: bool
-    comment: str
-    equ_list: list[str]
-    var_list: list[str]
+from ixmp.util.ixmp4 import ReadKwargs, WriteFiltersKwargs, WriteKwargs
 
 
 class Backend(ABC):

ixmp/backend/io.py

@@ -2,36 +2,28 @@
 from collections import deque
 from collections.abc import Iterable
 from pathlib import Path
-from typing import TYPE_CHECKING, Optional, Union
+from typing import Optional, TypeVar, Union
 
 import gams.transfer as gt
 import pandas as pd
 
 # from gams import GamsWorkspace
 from ixmp4.core import Run
+from ixmp4.core.optimization.equation import Equation
 from ixmp4.core.optimization.indexset import IndexSet
+from ixmp4.core.optimization.parameter import Parameter
 from ixmp4.core.optimization.scalar import Scalar
+from ixmp4.core.optimization.table import Table
+from ixmp4.core.optimization.variable import Variable
 from ixmp4.data.abstract.optimization.equation import Equation as AbstractEquation
 from ixmp4.data.abstract.optimization.variable import Variable as AbstractVariable
 
 from ixmp.util import as_str_list, maybe_check_out, maybe_commit
 
 from . import ItemType
 
-if TYPE_CHECKING:
-    from typing import TypeVar
-
-    from ixmp4.core.optimization.equation import Equation
-
-    # from ixmp4.core.optimization.indexset import IndexSet
-    from ixmp4.core.optimization.parameter import Parameter
-
-    # from ixmp4.core.optimization.scalar import Scalar
-    from ixmp4.core.optimization.table import Table
-    from ixmp4.core.optimization.variable import Variable
-
-    # Type variable that can be any one of these 6 types, but not a union of 2+ of them
-    Item4 = TypeVar("Item4", Equation, IndexSet, Parameter, Scalar, Table, Variable)
+# Type variable that can be any one of these 6 types, but not a union of 2+ of them
+Item4 = TypeVar("Item4", Equation, IndexSet, Parameter, Scalar, Table, Variable)
 
 
 log = logging.getLogger(__name__)
@@ -393,9 +385,9 @@ def records(
         return result
 
     for item in items:
-        # The gams documentation confuses me: The docstring says `type` is required, the
-        # example says no. It seems to work fine like this, but if we do need a value,
-        # maybe we could guess based on
+        # The gams documentation confuses me: The docstring says `type` is required for
+        # Equations, the example says no. It seems to work fine like this, but if we do
+        # need a value, maybe we could guess based on
         # https://github.com/iiasa/ixmp_source/blob/master/src/main/java/at/ac/iiasa/ixmp/objects/Scenario.java#L1926,
         klass(
             container=container,

ixmp/backend/ixmp4.py

@@ -5,6 +5,7 @@
 from typing import Any, Literal, Optional, Union, cast
 
 import pandas as pd
+from ixmp4 import DataPoint
 from ixmp4 import Platform as ixmp4_platform
 from ixmp4.core import Run
 from ixmp4.core.optimization.equation import Equation
@@ -104,8 +105,36 @@ def get_model_names(self) -> Generator[str, None, None]:
         for model in self._platform.models.list():
             yield model.name
 
-    def get_scenarios(self, default, model, scenario):
-        return self._platform.runs.list()
+    def get_scenarios(
+        self, default: bool, model: Optional[str], scenario: Optional[str]
+    ):
+        runs = self._platform.runs.list(
+            default_only=default,
+            model={"name": model} if model else None,
+            scenario={"name": scenario} if scenario else None,
+        )
+
+        for run in runs:
+            yield [
+                run.model.name,
+                run.scenario.name,
+                # TODO What are we going to use for scheme in ixmp4?
+                "IXMP4Run",
+                run.is_default(),
+                # TODO Change this from being hardcoded
+                False,
+                # TODO Expose the creation, update and lock info in ixmp4
+                # (if we get lock info)
+                "Some user",
+                "Some date",
+                "Some user",
+                "Some date",
+                "Some user",
+                "Some date",
+                # TODO Should Runs get .docs?
+                "Some docs",
+                run.version,
+            ]
 
     def set_unit(self, name: str, comment: str) -> None:
         self._platform.units.create(name=name).docs = comment
@@ -184,12 +213,16 @@ def clone(
         # TODO Is this enough? ixmp4 doesn't support cloning to a different platform at
         # the moment, but maybe we can imitate this here? (Access
         # platform_dest.backend._platform to create a new Run?)
-        cloned_s = Scenario(
-            mp=platform_dest, model=model, scenario=scenario, annotation=annotation
-        )
         cloned_run = self.index[s].clone(
             model=model, scenario=scenario, keep_solution=keep_solution
         )
+        # Instantiate same class as the original object
+        cloned_s = s.__class__(
+            platform_dest,
+            model,
+            scenario,
+            version=cloned_run.version,
+        )
         self._index_and_set_attrs(cloned_run, cloned_s)
         return cloned_s
 
@@ -645,6 +678,68 @@ def cat_list(self, ms: Scenario, name: str) -> list[str]:
         category_indexset = self.index[ms].optimization.indexsets.get(f"type_{name}")
         return cast(list[str], category_indexset.data)
 
+    def set_data(
+        self,
+        ts: TimeSeries,
+        region: str,
+        variable: str,
+        data: dict[int, float],
+        unit: str,
+        subannual: str,
+        meta: bool,
+    ) -> None:
+        log.warning("Parameter `meta` for set_data() currently unused by ixmp4!")
+        log.warning("Parameter `subannual` for set_data() currently unused by ixmp4!")
+
+        # Construct dataframe as ixmp4 expects it
+        years = data.keys()
+        values = data.values()
+        number_of_years = len(years)
+        regions = [region] * number_of_years
+        variables = [variable] * number_of_years
+        units = [unit] * number_of_years
+        _data = list(zip(years, values, regions, variables, units))
+
+        # Add timeseries dataframe
+        # TODO Is this really the only type we're interested in?
+        self.index[ts].iamc.add(
+            pd.DataFrame(
+                _data, columns=["step_year", "value", "region", "variable", "unit"]
+            ),
+            type=DataPoint.Type.ANNUAL,
+        )
+
+    def get_data(
+        self,
+        ts: TimeSeries,
+        region: Sequence[str],
+        variable: Sequence[str],
+        unit: Sequence[str],
+        year: Sequence[str],
+    ) -> Generator[tuple, Any, None]:
+        data = self.index[ts].iamc.tabulate(
+            region={"name__in": region},
+            variable={"name__in": variable},
+            unit={"name__in": unit},
+        )
+
+        # TODO depending on data["type"], a different column name will contain the
+        # "year" values:
+        # step_year if type is ANNUAL or CATEGORICAL
+        # step_category if type is CATEGORICAL
+        # step_datetime if type is DATETIME
+        # We're only adding ANNUAL above for now, so let's assume that
+        data = data.loc[data["step_year"].isin(year)]
+
+        # Select only the columns we're interested in
+        # TODO the ixmp4 docstrings sounds like region, variable, and unit are not part
+        # of the df?
+        data = data[["region", "variable", "unit", "step_year", "value"]]
+
+        # TODO Why would we iterate and yield tuples instead of returning the whole df?
+        for row in data.itertuples(index=False, name=None):
+            yield row
+
     # Handle timeslices
 
     # def set_timeslice(self, name: str, category: str, duration: float) -> None:
@@ -660,8 +755,6 @@ def write_file(
         IXMP4Backend supports writing to:
 
         - ``path='*.gdx', item_type=ItemType.SET | ItemType.PAR``.
-
-        IXMP4Backend does not yet support:
         - ``path='*.csv', item_type=TS``. The `default` keyword argument is
           **required**.
 
@@ -712,6 +805,30 @@ def write_file(
                 file_name=_path,
                 record_version_packages=kwargs["record_version_packages"],
             )
+        elif _path.suffix == ".csv" and item_type is ItemType.TS:
+            models = filters.pop("model")
+            # NOTE this is what we get for not differentiating e.g. scenario vs
+            # scenarios in filters...
+            scenarios = set(cast(list[str], filters.pop("scenario")))
+            variables = filters.pop("variable")
+            units = filters.pop("unit")
+            regions = filters.pop("region")
+            default = filters.pop("default")
+            # TODO (How) Should we include this?
+            # export_all_runs = filters.pop("export_all_runs")
+
+            data = self._backend.runs.tabulate(
+                model={"name__in": models},
+                scenario={"name__in": scenarios},
+                iamc={
+                    "region": {"name__in": regions},
+                    "variable": {"name__in": variables},
+                    "unit": {"name__in": units},
+                },
+                default_only=default,
+            )
+            data.to_csv(path_or_buf=_path)
+
         else:
             raise NotImplementedError
 
@@ -794,14 +911,12 @@ def _ni(self, *args, **kwargs):
     delete = _ni
     delete_geo = _ni
     delete_meta = _ni
-    get_data = _ni
     get_doc = _ni
     get_geo = _ni
     get_meta = _ni
     get_timeslices = _ni
     last_update = _ni
     remove_meta = _ni
-    set_data = _ni
     set_doc = _ni
     set_geo = _ni
     set_meta = _ni

ixmp/core/platform.py

@@ -9,6 +9,7 @@
 from ixmp._config import config
 from ixmp.backend import BACKENDS, FIELDS, ItemType
 from ixmp.util import as_str_list
+from ixmp.util.ixmp4 import WriteFiltersKwargs
 
 if TYPE_CHECKING:
     from ixmp.backend.base import Backend
@@ -236,15 +237,15 @@ def export_timeseries_data(
                 "Invalid arguments: export_all_runs cannot be used when providing a "
                 "model or scenario."
             )
-        filters = {
-            "model": as_str_list(model),
-            "scenario": as_str_list(scenario),
-            "variable": as_str_list(variable),
-            "unit": as_str_list(unit),
-            "region": as_str_list(region),
-            "default": default,
-            "export_all_runs": export_all_runs,
-        }
+        filters = WriteFiltersKwargs(
+            scenario=as_str_list(scenario),
+            model=as_str_list(model),
+            variable=as_str_list(variable),
+            unit=as_str_list(unit),
+            region=as_str_list(region),
+            default=default,
+            export_all_runs=export_all_runs,
+        )
 
         self._backend.write_file(path, ItemType.TS, filters=filters)
 

ixmp/testing/data.py

@@ -11,6 +11,7 @@
 
 from ixmp import Platform, Scenario, TimeSeries
 from ixmp.backend import IAMC_IDX
+from ixmp.backend.ixmp4 import IXMP4Backend
 
 if TYPE_CHECKING:
     from typing import TypedDict
@@ -182,6 +183,11 @@ def add_test_data(scen: Scenario):
     return t, t_foo, t_bar, x
 
 
+def _add_required_units(mp: Platform) -> None:
+    mp.add_unit("USD")
+    mp.add_unit("???")
+
+
 def make_dantzig(
     mp: Platform,
     solve: bool = False,
@@ -209,6 +215,9 @@ def make_dantzig(
     --------
     .DantzigModel
     """
+    if isinstance(mp._backend, IXMP4Backend):
+        _add_required_units(mp=mp)
+
     # Add custom units and region for time series data
     try:
         mp.add_unit("USD/km")

ixmp/util/ixmp4.py

@@ -0,0 +1,36 @@
+# TODO Import this from typing when dropping Python 3.11
+from typing import TYPE_CHECKING, Any, Optional
+
+from typing_extensions import TypedDict
+
+if TYPE_CHECKING:
+    from ixmp.core.scenario import Scenario
+
+
+# These are based on existing calls within ixmp
+class WriteFiltersKwargs(TypedDict, total=False):
+    scenario: "Scenario | list[str]"
+    model: list[str]
+    variable: list[str]
+    unit: list[str]
+    region: list[str]
+    default: bool
+    export_all_runs: bool
+
+
+class WriteKwargs(TypedDict, total=False):
+    filters: WriteFiltersKwargs
+    record_version_packages: list[str]
+
+
+class ReadKwargs(TypedDict, total=False):
+    filters: WriteFiltersKwargs
+    firstyear: Optional[Any]
+    lastyear: Optional[Any]
+    add_units: bool
+    init_items: bool
+    commit_steps: bool
+    check_solution: bool
+    comment: str
+    equ_list: list[str]
+    var_list: list[str]