import logging
from functools import lru_cache
from pathlib import Path
from typing import TYPE_CHECKING, List, Optional, Tuple, Union
from genno import Key
from message_ix_models.util import package_data_path
from message_ix_models.util.ixmp import get_reversed_rename_dims
from .key import pdt_cap
if TYPE_CHECKING:
import genno
import pint
import sdmx.message
import sdmx.model.common
from genno.core.key import KeyLike
from sdmx.model.common import ConceptScheme
from message_ix_models import Context
log = logging.getLogger(__name__)
#: List of all :class:`.ExogenousDataFile`.
FILES: List["ExogenousDataFile"] = []
[docs]class ExogenousDataFile:
"""Exogenous/input data for MESSAGEix-Transport expected in a file.
.. todo::
- Generate documentation (docstrings or snippets) in reStructuredText format.
- Accept an argument that sets :attr:`dfd` directly; skip handling other
arguments.
- Annotate certain dimensions as optional; expand :meth:`.add_tasks` to
automatically handle insertion of these dimensions.
- Merge with, or make a subclass of, :class:`.ExoData`.
Parameters
----------
name : str
Name of the data flow.
units : str
Units for observations in the data flow.
key : KeyLike, optional
Key at which the data from the file will be present in a :class:`.Computer`.
dims : tuple of str, optional
Dimensions of the data.
path : str or tuple of str, optional
Path at which the file is located. If not supplied, :attr:`path` is constructed
from `key`, `dims`, and the tag "exo".
description : str, optional
Human-readable description of the data flow, including any notes about required
properties or contents or methods used to handle the data.
required : bool, optional
If :any:`True` (the default), the file **must** be present for the build to
succeed.
"""
#: :class:`sdmx.Dataflow <sdmx.model.common.BaseDataflowDefinition>` describing the
#: input data flow.
df: "sdmx.model.common.BaseDataflowDefinition"
# Access to annotations of DFD
@property
def key(self) -> Key:
""":class:`genno.Key`, including preferred dimensions."""
return Key(str(self.df.get_annotation(id="genno-key").text))
@property
def path(self) -> Path:
"""Path fragment for the location of a file containing the data."""
return Path(str(self.df.get_annotation(id="file-path").text))
@property
def required(self) -> bool:
""":any:`True` if the data must be present for :func:`.transport.build.main`."""
return self.df.eval_annotation(id="required-for-build")
@property
def units(self) -> "pint.Unit":
"""Preferred units."""
import pint
return pint.get_application_registry().Unit(
self.df.eval_annotation(id="preferred-units")
)
[docs] def __init__(
self,
*,
name: str,
units: str,
key: Optional["KeyLike"] = None,
dims: Optional[Tuple[str, ...]] = None,
path: Union[str, Tuple[str, ...], None] = None,
description: Optional[str] = None,
required: bool = True,
):
import pint
from sdmx.model.common import Annotation
from sdmx.model.v21 import DataflowDefinition, DataStructureDefinition
# Collection of annotations for the data flow
anno = [Annotation(id="required-for-build", text=repr(required))]
# Handle `path` argument
if isinstance(path, str):
path = Path(path)
elif path:
path = Path(*path)
# Parse and store units
ureg = pint.get_application_registry()
try:
units = ureg.Unit(units)
except Exception as e:
log.info(f"Replace units {units!r} with 'dimensionless' due to {e}")
units = ureg.dimensionless
anno.append(Annotation(id="preferred-units", text=f"{units}"))
if not key:
# Determine from file path
key = Key(" ".join(path.parts).replace("-", " "), dims or (), "exo")
else:
# Convert to Key object
key = Key(key)
if path is None:
path = Path(key.name.replace(" ", "-"))
anno.append(Annotation(id="genno-key", text=str(key)))
path = path.with_suffix(".csv")
anno.append(Annotation(id="file-path", text=str(path)))
# Retrieve the shared concept scheme
common = common_structures()
cs: "ConceptScheme" = common.concept_scheme["CS_MESSAGE_TRANSPORT"]
# Reuse its properties for maintainable artefacts
kw = dict(
maintainer=cs.maintainer,
version=cs.version,
is_final=cs.is_final,
is_external_reference=cs.is_external_reference,
)
# SDMX IDs for the data flow and data structure
name_for_id = key.name.upper().replace(" ", "_")
df_id = f"DF_{name_for_id}"
ds_id = f"DS_{name_for_id}"
# Create a data structure definition
dsd = DataStructureDefinition(id=ds_id, **kw, name=f"Structure of {df_id}")
# Add dimensions
dims = get_reversed_rename_dims()
for dim in key.dims:
# Symbol ('n') → Dimension ID ('node') → upper case
dim_id = dims.get(dim, dim).upper()
# Add to the concept scheme
concept = cs.setdefault(id=dim_id)
# Add the dimension to the DSD
dsd.dimensions.getdefault(id=dim_id, concept_identity=concept)
if description is not None:
desc = f"{description.strip()}\n\n"
else:
desc = ""
desc += "Input data for MESSAGEix-Transport."
# Create and store a data flow definition
self.df = DataflowDefinition(
id=df_id, **kw, name=name, description=desc, structure=dsd, annotations=anno
)
# Does nothing except ensure callable(…) == True for inspection by genno
def __call__(self): ...
def __repr__(self) -> str:
return f"<ExogenousDataFile {self.path} → {self.key}>"
[docs] def add_tasks(
self, c: "genno.Computer", *args, context: "Context"
) -> Tuple["KeyLike", ...]:
"""Prepare `c` to read data from a file like :attr:`.path`."""
from message_ix_models.util.ixmp import rename_dims
from .util import path_fallback
# Identify the path
try:
path = path_fallback(context, self.path)
except FileNotFoundError:
if self.required:
raise
else:
return ()
# Use standard RENAME_DIMS from ixmp config
dims = rename_dims().copy()
values = set(dims.values())
dims.update({d: d for d in self.key.dims if d not in values})
c.add("load_file", path, key=self.key, dims=dims, name=self.key.name)
return (self.key,)
[docs] def generate_csv_template(self) -> Path:
"""Generate a CSV template file."""
raise NotImplementedError
# 1. In the current format.abs
# 2. In SDMX-CSV.
# dm = DataMessage()
# dm.data.append(DataSet(structure))
# template =
[docs]def add(*, replace: bool = False, **kwargs):
"""Add or replace an entry in :data:`FILES`.
Parameters
----------
replace : bool, *optional*
If :any:`True`, replace any existing entry in :data:`FILES` that targets an
equivalent key to the one implied by `kwargs`. Otherwise (default), raise an
exception.
kwargs :
Passed on to :class:`ExogenousDataFile`.
"""
edf = ExogenousDataFile(**kwargs)
if duplicate := list(filter(lambda x: x[1].key == edf.key, enumerate(FILES))):
i, existing = duplicate[0]
if replace:
log.info(f"Replace existing entry for {existing.key} at index {i}")
FILES[i] = edf
else:
raise RuntimeError(f"Definition of {edf} duplicates existing {existing}")
else:
# Add to the list of FILES
FILES.append(edf)
return edf.key
[docs]@lru_cache()
def common_structures() -> "sdmx.message.StructureMessage":
"""Return common structures for use in the current module."""
from importlib.metadata import version
from packaging.version import parse
from sdmx.message import StructureMessage
from sdmx.model.common import ConceptScheme
from message_ix_models.util.sdmx import read
# Create a shared concept scheme with…
# - Same maintainer "IIASA_ECE" as in "IIASA_ECE:AGENCIES".
# - Version based on the current version of message_ix_models.
# - Final and not an external reference
cs = ConceptScheme(
id="CS_MESSAGE_TRANSPORT",
maintainer=read("IIASA_ECE:AGENCIES")["IIASA_ECE"],
version=parse(version("message_ix_models")).base_version,
is_final=False,
is_external_reference=False,
)
# Return encapsulated in a StructureMessage
sm = StructureMessage()
sm.add(cs)
return sm
[docs]def collect_structures() -> "sdmx.message.StructureMessage":
"""Collect all SDMX data structures from :data:`FILES` and store.
The structural metadata are written to :file:`transport-in.xml`.
"""
from message_ix_models.util.sdmx import write
sm = common_structures()
for file in FILES:
# Add both the DFD and DSD
sm.add(file.df)
sm.add(file.df.structure)
write(sm, basename="transport-in")
return sm
[docs]def read_structures() -> "sdmx.message.StructureMessage":
"""Read structural metadata from :file:`transport-in.xml`."""
import sdmx
with open(package_data_path("sdmx", "transport-in.xml"), "rb") as f:
return sdmx.read_sdmx(f)
activity_freight = add(
key="freight activity:n:exo",
name="Freight transport activity",
units="Gt / km",
)
activity_ldv = add(
key="ldv activity:n:exo",
name="Activity (driving distance) per light duty vehicle",
units="km / year",
)
pdt_cap_ref = add(
key=(pdt_cap / "y") + "ref",
path="pdt-cap-ref",
name="Reference (historical) passenger distance travelled per capita",
units="km / year",
)
disutility = add(
key="disutility:n-cg-t-y:per vehicle",
name="Disutility cost of LDV usage",
units="kUSD / vehicle",
)
demand_scale = add(
key="demand-scale:n-y",
name="Scaling of total demand relative to base year levels",
units="dimensionless",
)
energy_other = add(
key="energy:c-n:transport other",
path="energy-other",
name="2020 demand for other transport energy",
units="TJ",
required=False,
)
fuel_emi_intensity = add(
key="fuel-emi-intensity:c-e",
name="Carbon emissions intensity of fuel use",
description="""Values are in GWP-equivalent mass of carbon, not in mass of the
emissions species.""",
units="tonne / kWa",
)
lifetime_ldv = add(
key="lifetime:nl-yv:ldv+exo",
path="lifetime-ldv",
name="Technical lifetime (maximum age) of LDVs",
description="""Values are filled forwards. In MESSAGE(V)-Transport, this quantity
had the additional dimension of driver_type, and values were 20 years for
driver_type='average', 15 y for 'moderate', and 10 y for 'frequent'.""",
units="year",
)
mode_share_freight = add(
key="freight mode share:n-t:exo",
path="freight-mode-share-ref",
name="Mode shares of freight activity in the model base period",
units="dimensionless",
)
ikarus_availability = add(
path=("ikarus", "availability"),
dims=("source", "t", "c", "y"),
name="Availability of non-LDV transport technologies",
description="""- 'source' is either "IKARUS" or "Krey/Linßen".""",
units="km / a",
)
ikarus_fix_cost = add(
path=("ikarus", "fix_cost"),
dims=("source", "t", "c", "y"),
name="Fixed cost of non-LDV transport technologies",
description="Costs are per vehicle.",
units="kEUR_2000",
)
ikarus_input = add(
path=("ikarus", "input"),
dims=("source", "t", "c", "y"),
name="Input energy intensity of non-LDV transport technologies",
units="GJ / (100 vehicle km)",
)
ikarus_inv_cost = add(
path=("ikarus", "inv_cost"),
dims=("source", "t", "c", "y"),
name="Investment/capital cost of non-LDV transport technologies",
units="MEUR_2000",
)
ikarus_technical_lifetime = add(
path=("ikarus", "technical_lifetime"),
dims=("source", "t", "c", "y"),
name="Technical lifetime of non-LDV transport technologies",
units="year",
)
ikarus_var_cost = add(
path=("ikarus", "var_cost"),
dims=("source", "t", "c", "y"),
name="Variable cost of non-LDV transport technologies",
units="EUR_2000 / (100 vehicle km)",
)
input_base = add(
path="input-base",
key="input:t-c-h:base",
name="Base model input efficiency",
units="GWa",
)
age_ldv = add(
path="ldv-age",
key="age:n-t-y:ldv+exo",
name="Mean age of LDVs as of the model base period",
units="years",
)
class_ldv = add(
path="ldv-class",
dims=("n", "vehicle_class"),
name="Share of light-duty vehicles by class",
units="dimensionless",
required=False,
)
input_adj_ldv = add(
key="ldv input adj:n-scenario:exo",
name="Calibration factor for LDV fuel economy",
units="dimensionless",
)
input_ref_ldv = add(
path="ldv-input-ref",
key="fuel economy:nl-m:ldv+ref",
name="Reference fuel economy for LDVs",
units="GWa / (Gvehicle km)",
required=False,
)
cap_new_ldv = add(
path="ldv-new-capacity",
key="cap_new:nl-t-yv:ldv+exo",
name="New capacity values for LDVs",
description="""Applied as historical_new_capacity and bound_new_capacity_{lo,up}
values for LDVs""",
units="MVehicle",
required=False,
)
t_share_ldv = add(
path="ldv-t-share",
key="tech share:n-t:ldv+exo",
name="Share of total stock for LDV technologies",
description="""
- Values must sum to 1 across the 't' dimension.
- Technology codes annotated "historical-only: True" (e.g. ICE_L_ptrp) must be omitted
or have zero values. If not, incompatible/infeasible constraint values are created.
""",
units="dimensionless",
)
act_non_ldv = add(
path="act-non_ldv.csv",
key="activity:n-t-y:non-ldv+exo",
name="Fixed activity of non-LDV technologies.",
units="dimensionless",
)
load_factor_ldv = add(
key="load factor ldv:scenario-n-y:exo",
name="Load factor (occupancy) of LDVs",
description="""Units are implicity passengers per vehicle.""",
units="dimensionless",
)
load_factor_nonldv = add(
key="load factor nonldv:t:exo",
name="Load factor (occupancy) of non-LDV passenger vehicles",
units="passenger / vehicle",
)
pop_share_attitude = add(
path=("ma3t", "attitude"),
dims=("attitude",),
name="Share of population by technology propensity/attitude",
units="dimensionless",
)
pop_share_driver = add(
path=("ma3t", "driver"),
dims=("census_division", "area_type", "driver_type"),
name="Share of population by driver type, census_division, and area_type",
description="""Values sum to roughly 1 across 'area_type' for each combination of
other dimensions.""",
units="dimensionless",
)
pop_share_cd_at = add(
path=("ma3t", "population"),
dims=("census_division", "area_type"),
name="Share of population by census division and area type",
description="Values sum to roughly 1 across 'area_type' for each census_division.",
units="dimensionless",
)
mer_to_ppp = add(
key="mer to ppp:n-y",
name="Convertion from market exchange rate (MER) to PPP",
units="dimensionless",
required=False,
)
pdt_elasticity = add(
key="pdt elasticity:scenario-n:exo",
name="“Elasticity” of PDT-capita with respect to GDP",
units="dimensionless",
)
population_suburb_share = add(
key="population suburb share:n-y:exo",
name="Share of MSA population that is suburban",
units="dimensionless",
required=False,
)
speed = add(
key="speed:scenario-n-t-y:exo",
name="Vehicle speed",
description="""
- This is the mean value for all vehicles of all technologies for the given mode.
- The code that handles this file interpolates on the ‘year’ dimension.""",
units="km / hour",
)