Add .transport.operator.indexer_scenario()

- Allow use of scenario="LED" in some cases and scenario="SSP1" in
  others.
- Select LDV load factor data using "LED" label where necessary.

Author: khaeru

message_ix_models/model/transport/build.py

@@ -20,6 +20,7 @@
 from message_ix_models.util.graphviz import HAS_GRAPHVIZ
 
 from . import Config
+from .operator import indexer_scenario
 from .structure import get_technology_groups
 
 if TYPE_CHECKING:
@@ -291,6 +292,8 @@ def add_exogenous_data(c: Computer, info: ScenarioInfo) -> None:
     ("groups::iea to transport", itemgetter(0), "groups::iea eweb"),
     ("groups::transport to iea", itemgetter(1), "groups::iea eweb"),
     ("indexers::iea to transport", itemgetter(2), "groups::iea eweb"),
+    ("indexers:scenario", partial(indexer_scenario, with_LED=False), "config"),
+    ("indexers:scenario:LED", partial(indexer_scenario, with_LED=True), "config"),
     ("n::ex world", "nodes_ex_world", "n"),
     (
         "n:n:ex world",
@@ -395,14 +398,12 @@ def add_structure(c: Computer) -> None:
     # - `Static` tasks
     # - Single 'dynamic' tasks based on config, info, spec, and/or t_groups
     # - Multiple static and dynamic tasks generated in loops etc.
-    tasks = list(STRUCTURE_STATIC) + [
+    tasks: list[tuple] = list(STRUCTURE_STATIC) + [
         ("c::transport", quote(spec.add.set["commodity"])),
         ("c::transport+base", quote(spec.add.set["commodity"] + info.set["commodity"])),
         ("cg", quote(spec.add.set["consumer_group"])),
         ("indexers:cg", spec.add.set["consumer_group indexers"]),
         ("nodes", quote(info.set["node"])),
-        # TODO Use "LED" where appropriate
-        ("indexers:scenario", quote(dict(scenario=repr(config.ssp).split(":")[1]))),
         ("t::transport", quote(spec.add.set["technology"])),
         ("t::transport agg", quote(dict(t=t_groups))),
         ("t::transport all", quote(dict(t=spec.add.set["technology"]))),

message_ix_models/model/transport/ldv.py

@@ -162,7 +162,7 @@ def prepare_computer(c: Computer):
 
     # Select load factor
     k.lf_ny = k.lf_nsy / "scenario"
-    c.add(k.lf_ny[0], "select", k.lf_nsy[0], "indexers:scenario")
+    c.add(k.lf_ny[0], "select", k.lf_nsy[0], "indexers:scenario:LED")
 
     # Insert a scaling factor that varies according to SSP
     c.apply(factor.insert, k.lf_ny[0], name="ldv load factor", target=k.lf_ny)

message_ix_models/model/transport/operator.py

@@ -6,7 +6,7 @@
 from functools import partial, reduce
 from itertools import product
 from operator import gt, le, lt
-from typing import TYPE_CHECKING, Any, Hashable, Optional, Union, cast
+from typing import TYPE_CHECKING, Any, Hashable, Literal, Optional, Union, cast
 
 import genno
 import numpy as np
@@ -64,6 +64,7 @@
     "groups_iea_eweb",
     "groups_y_annual",
     "iea_eei_fv",
+    "indexer_scenario",
     "indexers_n_cd",
     "indexers_usage",
     "logit",
@@ -788,6 +789,30 @@ def iea_eei_fv(name: str, config: dict) -> "AnyQuantity":
     return result.sel(y=ym1, t="Total freight transport", drop=True)
 
 
+def indexer_scenario(config: dict, *, with_LED: bool) -> dict[Literal["scenario"], str]:
+    """Indexer for the ``scenario`` dimension.
+
+    If `with_LED` **and** :py:`config.project["LDV"] = True`, then the single label is
+    "LED". Otherwise it is the short form of the :attr:`.transport.config.Config.ssp`
+    code, e.g. "SSP1". In other words, this treats "LDV" as mutually exclusive with an
+    SSP scenario identifier (instead of orthogonal).
+
+    Parameters
+    ----------
+    config :
+        The genno.Computer "config" dictionary, with a key "transport" mapped to an
+        instance of :class:`.transport.Config`.
+    """
+    # Retrieve the .transport.Config object from the genno.Computer "config" dict
+    c: "Config" = config["transport"]
+
+    return dict(
+        scenario="LED"
+        if (with_LED and c.project.get("LED", False))
+        else repr(c.ssp).split(":")[1]
+    )
+
+
 def indexers_n_cd(config: dict) -> dict[str, xr.DataArray]:
     """Indexers for selecting (`n`, `census_division`) → `n`.