Source code for message_ix_models.model.transport.config

import logging
import re
from copy import deepcopy
from dataclasses import InitVar, dataclass, field, replace
from typing import TYPE_CHECKING, Any, Literal, Optional, Union

from genno import Quantity
from genno.operator import as_quantity

from message_ix_models import Context, ScenarioInfo, Spec
from message_ix_models.project.navigate import T35_POLICY as NAVIGATE_SCENARIO
from message_ix_models.project.ssp import SSP_2024, ssp_field
from message_ix_models.project.transport_futures import SCENARIO as FUTURES_SCENARIO
from message_ix_models.util import package_data_path
from message_ix_models.util.config import ConfigHelper
from message_ix_models.util.sdmx import AnnotationsMixIn

from .policy import ExogenousEmissionPrice, TaxEmission

if TYPE_CHECKING:
    from sdmx.model import common

    from message_ix_models.tools.policy import Policy


log = logging.getLogger(__name__)

#: All files in :file:`data/transport/R12/price-emission/`.
PRICE_EMISSION_URL = {
    # "LED-SSP2": "SSP_LED_v5.3.1/baseline_1000f_v1",
    # "LED-SSP2": "SSP_LED_v5.3.1/INDC2030i_SSP2 - Very Low Emissions_v1",
    "LED-SSP2": "SSP_LED_v5.3.1/SSP2 - Very Low Emissions_v2",
    # "SSP1": "SSP_SSP1_v5.3.1/baseline_1000f_v1",
    # "SSP1": "SSP_SSP1_v5.3.1/INDC2030i_SSP1 - Low Emissions_a_v1",
    # "SSP1": "SSP_SSP1_v5.3.1/INDC2030i_SSP1 - Low Emissions_v1",
    # "SSP1": "SSP_SSP1_v5.3.1/INDC2030i_SSP1 - Very Low Emissions_v1",
    # "SSP1": "SSP_SSP1_v5.3.1/SSP1 - Low Emissions_a_v2",
    "SSP1": "SSP_SSP1_v5.3.1/SSP1 - Low Emissions_v2",
    # "SSP1": "SSP_SSP1_v5.3.1/SSP1 - Very Low Emissions_v2",
    # "SSP2": "SSP_SSP2_v5.3.1/baseline_1000f_v2",
    # "SSP2": "SSP_SSP2_v5.3.1/baselineS_10_v1",
    # "SSP2": "SSP_SSP2_v5.3.1/baselineS_110_v1",
    # "SSP2": "SSP_SSP2_v5.3.1/baselineS_15_v1",
    # "SSP2": "SSP_SSP2_v5.3.1/baselineS_20_v1",
    # "SSP2": "SSP_SSP2_v5.3.1/baselineS_25_v1",
    # "SSP2": "SSP_SSP2_v5.3.1/baselineS_50_v1",
    # "SSP2": "SSP_SSP2_v5.3.1/baselineS_5_v3",
    # "SSP2": "SSP_SSP2_v5.3.1/INDC2030i_SSP2 - Low Emissions_a_v1",
    # "SSP2": "SSP_SSP2_v5.3.1/INDC2030i_SSP2 - Low Emissions_v1",
    # "SSP2": "SSP_SSP2_v5.3.1/npiref2035_low_dem_scen2_v1",
    # "SSP2": "SSP_SSP2_v5.3.1/NPIREF_price_cap_5$_bkp_v1",
    # "SSP2": "SSP_SSP2_v5.3.1/NPiREF_SSP2 - Low Overshootf_price_cap_5$_bkp_v1",
    # "SSP2": "SSP_SSP2_v5.3.1/NPiREF_SSP2 - Low Overshootf_v3",
    # "SSP2": "SSP_SSP2_v5.3.1/NPiREF_SSP2 - Medium-Low Emissionsf_v1",
    # "SSP2": "SSP_SSP2_v5.3.1/NPiREF_v10",
    # "SSP2": "SSP_SSP2_v5.3.1/SSP2 - Low Emissions_a_v2",
    "SSP2": "SSP_SSP2_v5.3.1/SSP2 - Low Emissions_v2",
    # "SSP2": "SSP_SSP2_v5.3.1/SSP2 - Low Overshoot_v2",
    # "SSP2": "SSP_SSP2_v5.3.1/SSP2 - Medium Emissions_a_v2",
    # "SSP2": "SSP_SSP2_v5.3.1/SSP2 - Medium Emissions_v2",
    # "SSP2": "SSP_SSP2_v5.3.1/SSP2 - Medium-Low Emissions_v2",
    "SSP3": "SSP_SSP3_v5.3.1/baseline_1000f_v1",
    # "SSP4": "SSP_SSP4_v5.3.1/baseline_1000f_v1",
    # "SSP4": "SSP_SSP4_v5.3.1/NPi2030_v1",
    # "SSP4": "SSP_SSP4_v5.3.1/NPiREF_SSP4 - Low Overshootf_v1",
    # "SSP4": "SSP_SSP4_v5.3.1/NPiREF_v1",
    "SSP4": "SSP_SSP4_v5.3.1/SSP4 - Low Overshoot_v2",
    # "SSP5": "SSP_SSP5_v5.3.1/baseline_1000f_v2",
    # "SSP5": "SSP_SSP5_v5.3.1/baseline2055_low_dem_scen_v1",
    # "SSP5": "SSP_SSP5_v5.3.1/baseline2060_low_dem_scen_v2",
    # "SSP5": "SSP_SSP5_v5.3.1/NPi2030_v1",
    # "SSP5": "SSP_SSP5_v5.3.1/NPiREF_SSP5 - Low Overshootf_v1",
    # "SSP5": "SSP_SSP5_v5.3.1/NPiREF_v1",
    "SSP5": "SSP_SSP5_v5.3.1/SSP5 - Low Overshoot_v2",
}


[docs] @dataclass class DataSourceConfig(ConfigHelper): """Sources for input data.""" #: Emissions: ID of a dump from a base scenario. emissions: str = "1" #: Non-passenger and non-light-duty vehicles. non_LDV: str = "IKARUS"
[docs] def quantity_field(value): """Field with a mutable default value that is a :class:`.Quantity`.""" return field(default_factory=lambda: as_quantity(value))
[docs] @dataclass class Config(ConfigHelper): """Configuration for MESSAGEix-Transport. This dataclass stores and documents all configuration settings required and used by :mod:`~message_ix_models.model.transport`. It also handles (via :meth:`.from_context`) loading configuration and values from files like :file:`config.yaml`, while respecting higher-level configuration, for instance :attr:`.model.Config.regions`. """ #: Information about the base model. base_model_info: ScenarioInfo = field(default_factory=ScenarioInfo) #: Scaling factors for costs. #: #: ``ldv nga`` #: Scaling factor to reduce the cost of NGA vehicles. #: #: .. note:: DLM: “applied to the original US-TIMES cost data. That original data #: simply seems too high - much higher than conventional gasoline vehicles in #: the base-year and in future, which is strange. #: #: ``bus inv`` #: Investment costs of bus technologies, relative to the cost of ``ICG_bus``. #: Dictionary with 1 key per ``BUS`` technology. #: #: - Used in ikarus.py #: - This is from the IKARUS data in GEAM_TRP_Technologies.xlsx; sheet #: 'updateTRPdata', with the comment "Original data from Sei (PAO)." #: - This probably refers to some source that gave relative costs of different #: buses, in PAO, for this year; it is applied across all years. cost: dict = field( default_factory=lambda: { # "ldv nga": 0.85, "bus inv": { "ICH_bus": 1.153, # ie. 150,000 / 130,000 "PHEV_bus": 1.153, "FC_bus": 1.538, # ie. 200,000 / 130,000 "FCg_bus": 1.538, "FCm_bus": 1.538, }, } ) #: Sources for input data. data_source: DataSourceConfig = field(default_factory=DataSourceConfig) #: Set of modes handled by demand projection. This list must correspond to groups #: specified in the corresponding technology.yaml file. #: #: .. todo:: Read directly from technology.yaml demand_modes: list[str] = field( default_factory=lambda: ["LDV", "2W", "AIR", "BUS", "RAIL"] ) #: Include dummy ``demand`` data for testing and debugging. dummy_demand: bool = False #: Include dummy data for LDV technologies. dummy_LDV: bool = False #: Include dummy technologies supplying commodities required by transport, for #: testing and debugging. dummy_supply: bool = False #: Various efficiency factors. efficiency: dict = field( default_factory=lambda: { "*": 0.2, "hev": 0.2, "phev": 0.2, "fcev": 0.2, # Similar to 'cost/bus inv' above, except for output efficiency. "bus output": { "ICH_bus": 1.424, # ie. 47.6 / 33.42 "PHEV_bus": 1.424, "FC_bus": 1.563, # ie. 52.25 / 33.42 "FCg_bus": 1.563, "FCm_bus": 1.563, }, } ) #: Generate relation entries for emissions. emission_relations: bool = True #: Various other factors. factor: dict = field(default_factory=dict) #: If :obj:`True` (the default), do not record/preserve parameter data when removing #: set elements from the base model. fast: bool = True #: Fixed future point for total passenger activity. fixed_GDP: Quantity = quantity_field("1500 kUSD_2005 / passenger / year") #: Fixed future point for total passenger activity. #: #: AJ: Assuming mean speed of the high-speed transport is 330 km/h leads to 132495 #: passenger km / capita / year (Schafer & Victor 2000). #: Original comment (DLM): “Assume only half the speed (330 km/h) and not as steep a #: curve.” fixed_pdt: Quantity = quantity_field("132495 km / year") #: Load factors for vehicles [tonne km per vehicle km]. #: #: ``F ROAD``: similar to IEA “Future of Trucks” (2017) values; see #: .transport.freight. Alternately use 5.0, similar to Roadmap 2017 values. load_factor: dict = field( default_factory=lambda: { "F ROAD": 10.0, "F RAIL": 10.0, } ) #: Logit share exponents or cost distribution parameters [0] lamda: float = -2.0 #: Period in which LDV costs match those of a reference region. #: Dimensions: (node,). ldv_cost_catch_up_year: dict = field(default_factory=dict) #: Method for calibrating LDV stock and sales: #: #: - :py:`"A"`: use data from :file:`ldv-new-capacity.csv`, if it exists. #: - :py:`"B"`: use func:`.ldv.stock`; see the function documentation. ldv_stock_method: Literal["A", "B"] = "B" #: Tuples of (node, technology (transport mode), commodity) for which minimum #: activity should be enforced. See :func:`.non_ldv.bound_activity_lo`. minimum_activity: dict[tuple[str, tuple[str, ...], str], float] = field( default_factory=dict ) #: Base year shares of activity by mode. This should be the stem of a CSV file in #: the directory :file:`data/transport/{regions}/mode-share/`. mode_share: str = "default" #: List of modules containing model-building calculations. modules: list[str] = field( default_factory=lambda: ( "groups demand constraint freight ikarus ldv disutility other passenger " "plot data" ).split() ) #: Used by :func:`.get_USTIMES_MA3T` to map MESSAGE regions to U.S. census divisions #: appearing in MA³T. node_to_census_division: dict = field(default_factory=dict) #: Instances of :class:`.Policy` subclasses applicable in a workflow or to a #: scenario. policy: set["Policy"] = field(default_factory=set) #: Flags for distinct scenario features according to projects. In addition to #: providing values directly, this can be set by passing :attr:`futures_scenario` or #: :attr:`navigate_scenario` to the constructor, or by calling #: :meth:`set_futures_scenario` or :meth:`set_navigate_scenario` on an existing #: Config instance. #: #: :mod:`.transport.build` and :mod:`.transport.report` code will respond to these #: settings in documented ways. project: dict[str, Any] = field( default_factory=lambda: dict( futures=FUTURES_SCENARIO.BASE, navigate=NAVIGATE_SCENARIO.REF ) ) #: Scaling factors for production function [0] scaling: float = 1.0 #: Mapping from nodes to other nodes towards which share weights should converge. share_weight_convergence: dict = field(default_factory=dict) #: Specification for the structure of MESSAGEix-Transport, processed from contents #: of :file:`set.yaml` and :file:`technology.yaml`. spec: Spec = field(default_factory=Spec) #: Enum member indicating a Shared Socioeconomic Pathway, if any, to use for #: exogenous data. ssp: ssp_field = ssp_field(default=SSP_2024["2"]) #: :any:`True` if a base model or MESSAGEix-Transport scenario (possibly with #: solution data) is available. with_scenario: bool = False #: :any:`True` if solution data is available. with_solution: bool = False #: Work hours per year, used to compute the value of time. work_hours: Quantity = quantity_field("1600 hours / passenger / year") #: Year for share convergence. year_convergence: int = 2110 # Init-only variables #: Extra entries for :attr:`modules`, supplied to the constructor. May be either a #: space-delimited string (:py:`"module_a -module_b"`) or sequence of strings. #: Values prefixed with a hyphen (:py:`"-module_b"`) are *removed* from #: :attr:`.modules`. extra_modules: InitVar[Union[str, list[str]]] = None #: Identifier of a Transport Futures scenario, used to update :attr:`project` via #: :meth:`.ScenarioFlags.parse_futures`. futures_scenario: InitVar[str] = None #: Identifiers of NAVIGATE T3.5 demand-side scenarios, used to update #: :attr:`project` via :meth:`.ScenarioFlags.parse_navigate`. navigate_scenario: InitVar[str] = None def __post_init__(self, extra_modules, futures_scenario, navigate_scenario): # Handle extra_modules em = extra_modules or [] for m in em.split() if isinstance(em, str) else em: if m.startswith("-"): try: idx = self.modules.index(m[1:]) except ValueError: pass else: self.modules.pop(idx) else: self.modules.append(m) # Handle values for :attr:`futures_scenario` and :attr:`navigate_scenario` self.set_futures_scenario(futures_scenario) self.set_navigate_scenario(navigate_scenario)
[docs] @classmethod def from_context(cls, context: Context, options: Optional[dict] = None) -> "Config": """Configure `context` for building MESSAGEix-Transport. :py:`context.transport` is set to an instance of :class:`Config`. Configuration files and metadata are read and override the class defaults. The files listed in :data:`.METADATA` are stored in the respective attributes for instance :attr:`set` corresponding to :file:`data/transport/set.yaml`. If a subdirectory of :file:`data/transport/` exists corresponding to :py:`context.model.regions` (:attr:`.model.Config.regions`), then the files are loaded from that subdirectory, for instance :file:`data/transport/ISR/set.yaml` is preferred to :file:`data/transport/set.yaml`. .. note:: This method previously had behaviour similar to :meth:`.model.Config.regions_from_scenario`. Calling code should call that method if it is needed to ensure that :attr:`.model.Config.regions` has the desired value. """ from .structure import make_spec # Handle arguments options = options or dict() # Default configuration config = cls() try: # Update with region-specific configuration config.read_file( package_data_path("transport", context.model.regions, "config.yaml") ) except FileNotFoundError as e: log.warning(e) # Data structure that cannot be stored in YAML if isinstance(config.minimum_activity, list): config.minimum_activity = { tuple(row[:-1]): row[-1] for row in config.minimum_activity } # Separate data source options ds_options = options.pop("data source", {}) # Update values, store on context result = context["transport"] = replace( config, **options, data_source=config.data_source.replace(**ds_options) ) # Create the structural spec result.spec = make_spec(context.model.regions) return result
[docs] def check(self): """Check consistency of :attr:`project`.""" s1 = self.project["futures"] s2 = self.project["navigate"] if all(map(lambda s: s.value > 0, [s1, s2])): raise ValueError(f"Scenario settings {s1} and {s2} are not compatible")
[docs] def set_futures_scenario(self, value: Optional[str]) -> None: """Update :attr:`project` from a string indicating a Transport Futures scenario. See :meth:`ScenarioFlags.parse_futures`. This method alters :attr:`mode_share` and :attr:`fixed_demand` according to the `value` (if any). """ if value is None: return s = FUTURES_SCENARIO.parse(value) self.project.update(futures=s) self.check() self.mode_share = s.id() if self.mode_share == "A---": log.info(f"Set fixed demand for TF scenario {value!r}") self.fixed_demand = as_quantity("275000 km / year")
[docs] def set_navigate_scenario(self, value: Optional[str]) -> None: """Update :attr:`project` from a string representing a NAVIGATE scenario. See :meth:`ScenarioFlags.parse_navigate`. """ if value is None: return s = NAVIGATE_SCENARIO.parse(value) self.project.update(navigate=s) self.check()
[docs] def use_scenario_code(self, code: "common.Code") -> tuple[str, str]: """Update settings given a `code` with :class:`ScenarioCodeAnnotations`. Returns ------- tuple of str The entries are: 1. A short label suitable for a :class:`.Workflow` step name, for instance "SSP3 policy" or "SSP5", where the first part is :py:`code.id`. See :func:`.transport.workflow.generate`. 2. A longer, more explicity label suitable for (part of) a :attr:`message_ix.Scenario.scenario` name in an :mod:`ixmp` database, for instance "SSP_2024.3". """ from message_ix_models.project.digsy.structure import SCENARIO as DIGSY from message_ix_models.project.edits.structure import SCENARIO as EDITS sca = ScenarioCodeAnnotations.from_obj(code) # Look up the SSP_2024 Enum self.ssp = SSP_2024.by_urn(sca.SSP_URN) # Store settings on the Config instance self.base_scenario_url = sca.base_scenario_URL if sca.policy: self.policy.add(sca.policy) self.project["LED"] = sca.is_LED_scenario self.project["DIGSY"] = DIGSY.by_urn(sca.DIGSY_scenario_URN) self.project["EDITS"] = EDITS.by_urn(sca.EDITS_scenario_URN) # Construct labels including the SSP code and policy identifier # 1. ‘Short’ label used for workflow steps # 2. ‘Full’ label used in the scenario name return code.id, re.sub("^SSP", "SSP_2024.", code.id)
[docs] @dataclass class ScenarioCodeAnnotations(AnnotationsMixIn): """Set of annotations appearing on each Code in ``CL_TRANSPORT_SCENARIO``.""" SSP_URN: str is_LED_scenario: bool DIGSY_scenario_URN: str EDITS_scenario_URN: str base_scenario_URL: str policy: Optional["Policy"]
[docs] @classmethod def from_obj(cls, obj, globals=None): globals = (globals or {}) | dict( TaxEmission=TaxEmission, ExogenousEmissionPrice=ExogenousEmissionPrice, ) return super().from_obj(obj, globals=globals)
[docs] def get_cl_scenario() -> "common.Codelist": """Retrieve ``Codelist=IIASA_ECE:CL_TRANSPORT_SCENARIO``. This code lists contains unique IDs for scenarios supported by the MESSAGEix-Transport workflow (:mod:`.transport.workflow`), plus the annotations: - ``SSP-URN``: the URN of a code identifying the SSP scenario to be used for sociodemographic data, for instance "urn:sdmx:org.sdmx.infomodel.codelist.Code=ICONICS:SSP(2024).1". - ``is-LED-scenario``: either "True" or "False". - ``EDITS-activity-id``: either "None", "'CA'", or "'HA'". """ from sdmx.model import common from message_ix_models.util.sdmx import read IIASA_ECE = read("IIASA_ECE:AGENCIES")["IIASA_ECE"] return refresh_cl_scenario( common.Codelist( id="CL_TRANSPORT_SCENARIO", maintainer=IIASA_ECE, version="1.0.0" ) )
[docs] def refresh_cl_scenario( existing: Optional["common.Codelist"] = None, ) -> "common.Codelist": """Refresh ``Codelist=IIASA_ECE:CL_TRANSPORT_SCENARIO``. The code list is entirely regenerated. If it is different from `cl`, the new version is returned. Otherwise, `cl` is returned unaltered. """ from sdmx.model import common import message_ix_models.project.digsy.structure import message_ix_models.project.edits.structure from message_ix_models.util.sdmx import read, write # Other data structures IIASA_ECE = read("IIASA_ECE:AGENCIES")["IIASA_ECE"] cl_ssp_2024 = read("ICONICS:SSP(2024)") cl_edits = message_ix_models.project.edits.structure.get_cl_scenario() cl_digsy = message_ix_models.project.digsy.structure.get_cl_scenario() cl: "common.Codelist" = common.Codelist( id="CL_TRANSPORT_SCENARIO", maintainer=IIASA_ECE, version="1.2.0", is_external_reference=False, is_final=False, ) # - Model name: # - 2024-11-25: use _v1.1 per a Microsoft Teams message. # - 2025-02-20: update to _v2.1 per discussion with OF. At this point _v2.3 is the # latest appearing in the database. # - 2025-05-05: update to _v5.0. # - 2025-06-24: update to _v6.1. # - The scenario names appear to form a sequence from "baseline_DEFAULT" to # "baseline_DEFAULT_step_15" and finally "baseline". The one used below is the # latest in this sequence for which y₀=2020, rather than 2030. base_url = "ixmp://ixmp-dev/SSP_SSP{}_v6.1/baseline_DEFAULT_step_13" def _code( id: str, name: str, c_ssp: "common.Code", led: bool, edits: str, digsy: str ) -> "common.Code": """Shorthand for creating a code.""" assert c_ssp.urn sca = ScenarioCodeAnnotations( c_ssp.urn, led, cl_digsy[digsy].urn, cl_edits[edits].urn, base_url.format(c_ssp.id), None, ) return common.Code(id=id, name=name, **sca.get_annotations(dict)) # SSP baselines and policies for c_ssp in cl_ssp_2024: c_base = _code(f"SSP{c_ssp.id}", "", c_ssp, False, "_Z", "_Z") cl.append(c_base) # Simple carbon tax c = deepcopy(c_base) c.get_annotation(id="policy").text = repr(TaxEmission(1000.0)) c.id += " tax" cl.append(c) # PRICE_EMISSION from exogenous data file c = deepcopy(c_base) c.get_annotation(id="policy").text = repr( ExogenousEmissionPrice("ixmp://ixmp-dev/" + PRICE_EMISSION_URL[c.id]) ) c.id += " exo price" cl.append(c) # LED name = "Low Energy Demand/High-with-Low scenario with SSP{} demographics" for ssp_id in ("1", "2"): c_ssp = cl_ssp_2024[ssp_id] cl.append( _code(f"LED-SSP{ssp_id}", name.format(ssp_id), c_ssp, True, "_Z", "_Z") ) # DIGSY c_ssp, name = cl_ssp_2024["2"], "DIGSY {!r} scenario with SSP2" for id_ in ("BEST", "WORST"): cl.append(_code(f"DIGSY-{id_}", name.format(id_), c_ssp, False, "_Z", id_)) # EDITS c_ssp, name = cl_ssp_2024["2"], "EDITS scenario with ITF PASTA {!r} activity" for id_ in ("CA", "HA"): cl.append(_code(f"EDITS-{id_}", name.format(id_), c_ssp, False, id_, "_Z")) # FIXME This condition may appear to be always False, because the date/time differs. # Adjust upstream (in sdmx1) to ignore this difference. if existing is None or not cl.compare(existing, strict=True): # No existing code list or new code list differs from existing write(cl) return cl else: return existing