Update water to use cost projections from tools.costs

doc/whatsnew.rst

@@ -4,6 +4,7 @@ What's new
 Next release
 ============
 
+- Connect the water module to the cost module for cooling technologies (:pull:`245`).
 - Make setup of constraints for cooling technologies flexible and update solar csp tech. name (:pull:`242`). 
 - Fix the nexus/cooling function and add test for checking some input data (:pull:`236`).
 - Add :doc:`/project/circeular` project code and documentation (:pull:`232`).

message_ix_models/data/costs/cooling/tech_map.csv

@@ -91,3 +91,11 @@ solar_th_ppl__air,energy,solar_th_ppl,220,0,2015
 solar_th_ppl__cl_fresh,energy,solar_th_ppl,100,0,2015
 solar_th_ppl__ot_fresh,energy,solar_th_ppl,0.4,0,2015
 solar_th_ppl__ot_saline,energy,solar_th_ppl,0.3,0,2015
+csp_sm1_ppl__air,energy,csp_sm1_ppl,220,0,2015
+csp_sm1_ppl__cl_fresh,energy,csp_sm1_ppl,100,0,2015
+csp_sm1_ppl__ot_fresh,energy,csp_sm1_ppl,0.4,0,2015
+csp_sm1_ppl__ot_saline,energy,csp_sm1_ppl,0.3,0,2015
+csp_sm3_ppl__air,energy,csp_sm3_ppl,220,0,2015
+csp_sm3_ppl__cl_fresh,energy,csp_sm3_ppl,100,0,2015
+csp_sm3_ppl__ot_fresh,energy,csp_sm3_ppl,0.4,0,2015
+csp_sm3_ppl__ot_saline,energy,csp_sm3_ppl,0.3,0,2015

message_ix_models/model/water/cli.py

@@ -4,14 +4,15 @@
 
 from message_ix_models import Context
 from message_ix_models.model.structure import get_codes
-from message_ix_models.util.click import common_params
+from message_ix_models.util.click import common_params, scenario_param
 
 log = logging.getLogger(__name__)
 
 
 # allows to activate water module
 @click.group("water-ix")
 @common_params("regions")
+@scenario_param("--ssp", default="SSP2")
 @click.option("--time", help="Manually defined time")
 @click.pass_obj
 def cli(context: "Context", regions, time):
@@ -206,6 +207,7 @@ def nexus(context: "Context", regions, rcps, sdgs, rels, macro=False):
 
 @cli.command("cooling")
 @common_params("regions")
+@scenario_param("--ssp")
 @click.option(
     "--rcps",
     default="no_climate",

message_ix_models/model/water/data/water_for_ppl.py

@@ -718,7 +718,7 @@ def cool_tech(context: "Context") -> dict[str, pd.DataFrame]:
     # con4 = cost['technology'].str.endswith("air")
     # con5 = cost.technology.isin(input_cool['technology_name'])
     # inv_cost = cost[(con3) | (con4)]
-    inv_cost = cost.copy()
+
     # Manually removing extra technologies not required
     # TODO make it automatic to not include the names manually
     techs_to_remove = [
@@ -735,23 +735,31 @@ def cool_tech(context: "Context") -> dict[str, pd.DataFrame]:
         "nuc_htemp__cl_fresh",
         "nuc_htemp__air",
     ]
-    inv_cost = inv_cost[~inv_cost["technology"].isin(techs_to_remove)]
-    # Converting the cost to USD/GW
-    inv_cost["investment_USD_per_GW_mid"] = (
-        inv_cost["investment_million_USD_per_MW_mid"] * 1e3
-    )
 
-    inv_cost = (
-        make_df(
-            "inv_cost",
-            technology=inv_cost["technology"],
-            value=inv_cost["investment_USD_per_GW_mid"],
-            unit="USD/GWa",
-        )
-        .pipe(same_node)
-        .pipe(broadcast, node_loc=node_region, year_vtg=info.Y)
+    from message_ix_models.tools.costs.config import Config
+    from message_ix_models.tools.costs.projections import create_cost_projections
+
+    # Set config for cost projections
+    # Using GDP method for cost projections
+    cfg = Config(
+        module="cooling", scenario=context.ssp, method="gdp", node=context.regions
     )
 
+    # Get projected investment and fixed o&m costs
+    cost_proj = create_cost_projections(cfg)
+
+    # Get only the investment costs for cooling technologies
+    inv_cost = cost_proj["inv_cost"][
+        ["year_vtg", "node_loc", "technology", "value", "unit"]
+    ]
+
+    # Remove technologies that are not required
+    inv_cost = inv_cost[~inv_cost["technology"].isin(techs_to_remove)]
+
+    # Only keep cooling module technologies by filtering for technologies with "__"
+    inv_cost = inv_cost[inv_cost["technology"].str.contains("__")]
+
+    # Add the investment costs to the results
     results["inv_cost"] = inv_cost
 
     # Addon conversion

message_ix_models/tests/model/water/data/test_water_for_ppl.py

@@ -94,8 +94,11 @@ def test_cool_tec(request, test_context, RCP):
     test_context.time = "year"
     test_context.nexus_set = "nexus"
     # TODO add
-    test_context.RCP = RCP
-    test_context.REL = "med"
+    test_context.update(
+        RCP=RCP,
+        REL="med",
+        ssp="SSP2",
+    )
 
     # TODO: only leaving this in so you can see which data you might want to assert to
     # be in the result. Please remove after adapting the assertions below:

message_ix_models/tests/util/test_click.py

@@ -1,9 +1,14 @@
 """Basic tests of the command line."""
 
 import click
+import pytest
 
 from message_ix_models.cli import cli_test_group
-from message_ix_models.util.click import common_params, temporary_command
+from message_ix_models.util.click import (
+    common_params,
+    scenario_param,
+    temporary_command,
+)
 
 
 def test_default_path_cb(session_context, mix_models_cli):
@@ -59,6 +64,50 @@ def inner(context, regions):
     assert "ZMB" == result.output.strip()
 
 
+@pytest.mark.parametrize(
+    "args, command, expected",
+    [
+        # As a (required, positional) argument
+        (dict(param_decls="ssp"), ["LED"], "LED"),
+        (dict(param_decls="ssp"), ["FOO"], "'FOO' is not one of 'LED', 'SSP1', "),
+        # As an option
+        # With no default
+        (dict(param_decls="--ssp"), [], "None"),
+        # With a limited of values
+        (
+            dict(param_decls="--ssp", values=["LED", "SSP2"]),
+            ["--ssp=SSP1"],
+            "'SSP1' is not one of 'LED', 'SSP2'",
+        ),
+        # With a default
+        (dict(param_decls="--ssp", default="SSP2"), [], "SSP2"),
+        # With a different name
+        (dict(param_decls=["--scenario", "ssp"]), ["--scenario=SSP5"], "SSP5"),
+    ],
+)
+def test_scenario_param(capsys, mix_models_cli, args, command, expected):
+    """Tests of :func:`scenario_param`."""
+
+    # scenario_param() can be used as a decorator with `args`
+    @click.command
+    @scenario_param(**args)
+    @click.pass_obj
+    def cmd(context):
+        """Temporary click Command: print the direct value and Context attribute."""
+        print(f"{context.ssp}")
+
+    with temporary_command(cli_test_group, cmd):
+        try:
+            result = mix_models_cli.assert_exit_0(["_test", "cmd"] + command)
+        except RuntimeError as e:
+            # `command` raises the expected value or error message
+            assert expected in capsys.readouterr().out, e
+        else:
+            # `command` can be invoked without error, and the function/Context get the
+            # expected value
+            assert expected == result.output.strip()
+
+
 def test_store_context(mix_models_cli):
     """Test :func:`.store_context`."""
 

message_ix_models/util/click.py

@@ -28,22 +28,28 @@
     """Decorate a click.command with common parameters `param_names`.
 
     `param_names` must be a space-separated string of names appearing in :data:`PARAMS`,
-    e.g. ``"ssp force output_model"``. The decorated function receives keyword
-    arguments with these names::
+    for instance :py:`"ssp force output_model"`. The decorated function receives keyword
+    arguments with these names; some are also stored on the
 
-        @click.command()
-        @common_params("ssp force output_model")
-        def mycmd(ssp, force, output_model)
-            # ...
+    Example
+    -------
+
+    >>> @click.command
+    ... @common_params("ssp force output_model")
+    ... @click.pass_obj
+    ... def mycmd(context, ssp, force, output_model):
+    ...     assert context.force == force
     """
 
+    # Create the decorator
     # Simplified from click.decorators._param_memo
     def decorator(f):
-        if not hasattr(f, "__click_params__"):
-            f.__click_params__ = []
-        f.__click_params__.extend(
+        # - Ensure f.__click_params__ exists
+        # - Append each param given in `param_names`
+        f.__dict__.setdefault("__click_params__", []).extend(
             PARAMS[name] for name in reversed(param_names.split())
         )
+
         return f
 
     return decorator
@@ -102,6 +108,101 @@
     return "\n".join(lines)[:-2]
 
 
+def scenario_param(
+    param_decls: Union[str, list[str]],
+    *,
+    values: list[str] = None,
+    default: Optional[str] = None,
+):
+    """Add an SSP or scenario option or argument to a :class:`click.Command`.
+
+    The parameter uses :func:`.store_context` to store the given value (if any) on
+    the :class:`.Context`.
+
+    Parameters
+    ----------
+    param_decls :
+        :py:`"--ssp"` (or any other name prefixed by ``--``) to generate a
+        :class:`click.Option`; :py:`"ssp"` to generate a :class:`click.Argument`.
+        Click-style declarations are also supported; see below.
+    values :
+        Allowable values. If not given, the allowable values are
+        ["LED", "SSP1", "SSP2", "SSP3", "SSP4", "SSP5"].
+    default :
+        Default value.
+
+    Raises
+    ------
+    ValueError
+        if `default` is given with `param_decls` that indicate a
+        :class:`click.Argument`.
+
+    Examples
+    --------
+    Add a (mandatory, positional) :class:`click.Argument`. This is nearly the same as
+    using :py:`common_params("ssp")`, except the decorated function does not receive an
+    :py:`ssp` argument. The value is still stored on :py:`context` automatically.
+
+    >>> @click.command
+    ... @scenario_param("ssp")
+    ... @click.pass_obj
+    ... def mycmd(context):
+    ...     print(context.ssp)
+
+    Add a :class:`click.Option` with certain, limited values and a default:
+
+    >>> @click.command
+    ... @scenario_param("--ssp", values=["SSP1", "SSP2", "SSP3"], default="SSP3")
+    ... @click.pass_obj
+    ... def mycmd(context):
+    ...     print(context.ssp)
+
+    An option given by the user as :command:`--scenario` but stored as
+    :py:`Context.ssp`:
+
+    >>> @click.command
+    ... @scenario_param(["--scenario", "ssp"])
+    ... @click.pass_obj
+    ... def mycmd(context):
+    ...     print(context.ssp)
+    """
+    if values is None:
+        values = ["LED", "SSP1", "SSP2", "SSP3", "SSP4", "SSP5"]
+
+    # Handle param_decls; identify the first string element
+    if isinstance(param_decls, list):
+        decl0 = param_decls[0]
+    else:
+        decl0 = param_decls
+        param_decls = [param_decls]  # Ensure list for use by click
+
+    # Choose either click.Option or click.Argument
+    if decl0.startswith("-"):
+        cls = Option
+    else:
+        cls = Argument
+        if default is not None:
+            raise ValueError(f"{default=} given for {cls}")
+
+    # Create the decorator
+    def decorator(f):
+        # - Ensure f.__click_params__ exists
+        # - Generate and append the parameter
+        f.__dict__.setdefault("__click_params__", []).append(
+            cls(
+                param_decls,
+                callback=store_context,
+                type=Choice(values),
+                default=default,
+                expose_value=False,
+            )
+        )
+
+        return f
+
+    return decorator
+
+
 def store_context(context: Union[click.Context, Context], param, value):
     """Callback that simply stores a value on the :class:`.Context` object.