"""MESSAGEix-Transport integration with :mod:`.model.material`.
In order to use this code:
1. Set :py:`extra_modules="material"` when constructing
:class:`.transport.config.Config`.
2. Call :func:`.transport.build.main`.
The code expects that an existing/base |Scenario| is available at the :py:`"scenario"`
key, so that existing data for the MESSAGE ``demand`` parameter can be adjusted.
"""
from collections import defaultdict
from typing import TYPE_CHECKING
import genno
from genno import Key, Keys
from genno.core.key import single_key
from message_ix_models.util import minimum_version
from message_ix_models.util.genno import Collector
from . import key, util
if TYPE_CHECKING:
from genno import Computer
from .config import Config
#: Target key that collects all data generated in this module.
TARGET = "transport::MT+ixmp"
collect = Collector(TARGET, "{}::MT+ixmp".format)
# FIXME Do not hard-code this. Instead, use 1 or more of:
# - Labels in input_cap_new.csv that align with .model.materials.
# - A CircEUlar-specific 'commodity' codelist that records correspondence with
# .model.materials
COMMODITY_INFO = {
"automotive steel": "steel",
"cast Al": "aluminum",
"cast iron": "pig_iron", # NB Several other commodities exist
# "co": "", # Missing
# "copper electric grade": "copper", # Commented in material/set.yaml
# "li": "", # Missing
# "mn": "", # Missing
# "nickel": "", # Missing
# "other": "", # Missing
# "p": "", # Missing
# "plastics": "", # Missing
"stainless steel": "steel",
"wrought Al": "aluminum",
# "zinc": "", # Missing
}
DIMS = dict(
commodity="c",
node_loc="n",
node_dest="n",
node_origin="n",
year_vtg="y",
technology="t",
)
#: Portion of the ``input_cap_new`` that is available as ``output_cap_ret`` at the end
#: of lifetime of a technology. Dimensionless.
#:
#: .. todo:: Retrieve from a file.
OUTPUT_SHARE = 0.8
# Keyword arguments for as_message_df() for different parameters
_DEMAND_KW = dict(name="demand", dims=util.DIMS, common=util.COMMON)
_ICN_KW = dict(
name="input_cap_new", dims=DIMS, common=util.COMMON | dict(level="demand")
)
_OCR_KW = dict(
name="output_cap_ret", dims=DIMS, common=util.COMMON | dict(level="end_of_life")
)
[docs]
def get_groups(config: "Config") -> dict[str, dict[str, list[str]]]:
"""Return groups for a :func:`~genno.operator.aggregate` operation on NTNU VMI data.
These include:
- ``c`` (commodity) dimension: 1 or more original (NTNU VMI) commodity IDs
aggregated to :mod:`.model.material` commodity IDs.
- ``t`` (technology) dimension: 1:1 from original (NTNU VMI) technology IDs to
:mod:`.model.transport` technology IDs. This effects rename and broadcast
operations simultaneously.
"""
# Aggregate ≥1 original commodity IDs into .model.material commodity IDs
c_groups = defaultdict(list)
for c_original, c_model in COMMODITY_INFO.items():
c_groups[c_model].append(c_original)
# Retrieve all codes for LDV technologies
t_all = config.spec.add.set["technology"]
t_LDV = t_all[t_all.index("LDV")].child
# Aggregate each original technology ID into 1 or more 'groups' of length 1
# (this is equivalent to a broadcast operation)
t_groups = {}
for t_model in t_LDV:
t_groups[t_model.id] = [
str(t_model.get_annotation(id="ntnu-vmi-technology").text)
]
return dict(c=c_groups, t=t_groups)
[docs]
@minimum_version("message_ix 3.11.2.dev0") # NB Actually 3.12
def prepare_computer(c: "Computer") -> None:
"""Prepare `c` to calculate and add data for materiality of transport."""
# Retrieve transport configuration
config = c.graph["context"].transport
# Collect data in `TARGET` and connect to the "add transport data" key
collect.computer = c
c.add("transport_data", __name__, key=TARGET)
k = Keys(
exo=(key.exo.input_cap_new - "exo") / "scenario",
# Same key as used in .transport.ldv.stock
# TODO Move to .key
sales="sales:n-t-y:LDV",
demand=Key("demand", key.demand_base.dims, "MT"),
)
# From input_cap_new.csv, select:
# - Only a single scenario
# TODO Retrieve the CircEUlar scenario ID from config
indexers = dict(scenario="_CT_C_D_D")
c.add(k.exo[0], "select", key.exo.input_cap_new, indexers=indexers)
# Transform VMI data labels to MESSAGE -MT- labels
c.add(k.exo[1], "aggregate", k.exo[0], groups=get_groups(config), keep=False)
# Convert units: (material commodities [Mt]) / (transport CAP/CAP_NEW [Mvehicle])
c.add(k.exo[2], "convert_units", k.exo[1], units="Mt / Mvehicle")
# Convert to MESSAGE-format data frame
collect("input_cap_new", "as_message_df", k.exo[2], **_ICN_KW)
# Reduce share available for recycling
c.add(k.exo[3], "mul", k.exo[2], OUTPUT_SHARE)
# Convert to MESSAGE-format data frame
collect("output_cap_ret", "as_message_df", k.exo[3], **_OCR_KW)
# Multiply base-period LDV sales by material intensity
tmp = single_key(c.add("demand::MT+0", "mul", k.exo[2], k.sales, sums=True))
# Sum on "t" dimension; expand "l" dimension
c.add(k.demand[0], "expand_dims", tmp / "t", dim={"l": ["demand"]})
# Convert units: material commodities demand [Mt/year]
c.add(k.demand[1], "convert_units", k.demand[0], units="Mt / year")
# Force units for existing model data
# FIXME Adjust to trust the base model's units
c.add(k.demand[2], "apply_units", key.demand_base, units="Mt / year")
# Share of this transport total in existing material demand as of y₀
c.add(k.demand[3], "div", k.demand[1], k.demand[2])
# Clip values to be in (0, 0.8)
c.add(k.demand[4], lambda q: q.clip(0.0, 0.8), k.demand[3])
# Difference with 1.0: should be in range (0.2, 1.0)
c.add(k.demand[5], "sub", genno.Quantity(1.0, units=""), k.demand[4])
# Select only values for y0
c.add(k.demand["share"], "select", k.demand[5], "y0::coord")
# Multiply existing material demand by this share
c.add(k.demand["adj"], "mul", k.demand[2], k.demand["share"])
# Convert data to MESSAGE-format data frame
collect("demand", "as_message_df", k.demand["adj"], **_DEMAND_KW)