Source code for message_ix_models.util.scenarioinfo

""":class:`ScenarioInfo` class."""
import logging
from collections import defaultdict
from dataclasses import dataclass, field
from itertools import product
from typing import Dict, List, Optional

import pandas as pd
import pint
import sdmx.model.v21 as sdmx_model

from .sdmx import eval_anno

log = logging.getLogger(__name__)


[docs]class ScenarioInfo: """Information about a :class:`~message_ix.Scenario` object. Code that prepares data for a target Scenario can accept a ScenarioInfo instance. This avoids the need to load a Scenario, which can be slow under some conditions. ScenarioInfo objects can also be used (e.g. by :func:`.apply_spec`) to describe the contents of a Scenario *before* it is created. ScenarioInfo objects have the following convenience attributes: .. autosummary:: set io_units is_message_macro N units_for Y y0 yv_ya Parameters ---------- scenario : message_ix.Scenario If given, :attr:`.set` is initialized from this existing scenario. See also -------- .Spec """ #: Elements of :mod:`ixmp`/:mod:`message_ix` sets. set: Dict[str, List] = {} #: Elements of :mod:`ixmp`/:mod:`message_ix` parameters. par: Dict[str, pd.DataFrame] = {} #: First model year, if set, else ``Y[0]``. y0: int = -1 #: :obj:`True` if a MESSAGE-MACRO scenario. is_message_macro: bool = False _yv_ya: pd.DataFrame = None def __init__(self, scenario=None): self.set = defaultdict(list) self.par = dict() if not scenario: return for name in scenario.set_list(): try: self.set[name] = scenario.set(name).tolist() except AttributeError: continue # pd.DataFrame for ≥2-D set; don't convert for name in ("duration_period",): self.par[name] = scenario.par(name) self.is_message_macro = "PRICE_COMMODITY" in scenario.par_list() # Computed once fmy = scenario.cat("year", "firstmodelyear") self.y0 = int(fmy[0]) if len(fmy) else self.set["year"][0] self._yv_ya = scenario.vintage_and_active_years() @property def yv_ya(self): """:class:`pandas.DataFrame` with valid ``year_vtg``, ``year_act`` pairs.""" if self._yv_ya is None: first = self.y0 # Product of all years yv = ya = self.set["year"] # Predicate for filtering years def _valid(elem): yv, ya = elem return first <= yv <= ya # - Cartesian product of all yv and ya. # - Filter only valid years. # - Convert to data frame. self._yv_ya = pd.DataFrame( filter(_valid, product(yv, ya)), columns=["year_vtg", "year_act"] ) return self._yv_ya @property def N(self): """Elements of the set 'node'. See also -------- .nodes_ex_world """ return list(map(str, self.set["node"])) @property def Y(self): """Elements of the set 'year' that are >= the first model year.""" return list(filter(lambda y: y >= self.y0, self.set["year"]))
[docs] def update(self, other: "ScenarioInfo"): """Update with the set elements of `other`.""" for name, data in other.set.items(): self.set[name].extend(filter(lambda id: id not in self.set[name], data)) for name, data in other.par.items(): raise NotImplementedError("Merging parameter data")
def __repr__(self): return ( f"<ScenarioInfo: {sum(len(v) for v in self.set.values())} code(s) in " f"{len(self.set)} set(s)>" )
[docs] def units_for(self, set_name: str, id: str) -> pint.Unit: """Return the units associated with code `id` in MESSAGE set `set_name`. :mod:`ixmp` (or the sole :class:`.JDBCBackend`, as of v3.5.0) does not handle unit information for variables and equations (unlike parameter values), such as MESSAGE decision variables ``ACT``, ``CAP``, etc. In :mod:`message_ix_models` and :mod:`message_data`, the following conventions are (generally) followed: - The units of ``ACT`` and others are consistent for each ``technology``. - The units of ``COMMODITY_BALANCE``, ``STOCK``, ``commodity_stock``, etc. are consistent for each ``commodity``. Thus, codes for elements of these sets (e.g. :ref:`commodity-yaml`) can be used to carry the standard units for the corresponding quantities. :func:`units_for` retrieves these units, for use in model-building and reporting. .. todo:: Expand this discussion and transfer to the :mod:`message_ix` docs. See also -------- io_units """ try: idx = self.set[set_name].index(id) except ValueError: print(self.set[set_name]) raise return eval_anno(self.set[set_name][idx], "units")
[docs] def io_units( self, technology: str, commodity: str, level: Optional[str] = None ) -> pint.Unit: """Return units for the MESSAGE ``input`` or ``output`` parameter. These are implicitly determined as the ratio of: - The units for the origin (for ``input``) or destination `commodity`, per :meth:`.units_for`. - The units of activity for the `technology`. Parameters ---------- level : str Placeholder for future functionality, i.e. to use different units per (commodity, level). Currently ignored. If given, a debug message is logged. """ if level is not None: log.debug(f"{level = } ignored") return self.units_for("commodity", commodity) / self.units_for( "technology", technology )
[docs] def year_from_codes(self, codes: List[sdmx_model.Code]): """Update using a list of `codes`. The following are updated: - :attr:`.set` ``year`` - :attr:`.set` ``cat_year``, with the first model year. - :attr:`.par` ``duration_period`` Any existing values are discarded. After this, the attributes :attr:`.y0` and :attr:`.Y` give the first model year and model years, respectively. Examples -------- Get a particular code list, create a ScenarioInfo instance, and update using the codes: >>> years = get_codes("year/A") >>> info = ScenarioInfo() >>> info.year_from_codes(years) Use populated values: >>> info.y0 2020 >>> info.Y[:3] [2020, 2030, 2040] >>> info.Y[-3:] [2090, 2100, 2110] """ from message_ix_models.util import eval_anno # Clear existing values if len(self.set["year"]): log.debug(f"Discard existing 'year' elements: {repr(self.set['year'])}") self.set["year"] = [] if len(self.set["cat_year"]): log.debug( f"Discard existing 'cat_year' elements: {repr(self.set['cat_year'])}" ) self.set["cat_year"] = [] if "duration_period" in self.par: log.debug("Discard existing 'duration_period' elements") fmy_set = False duration_period: List[Dict] = [] # TODO use sorted() here once supported by sdmx for code in codes: year = int(code.id) # Store the year self.set["year"].append(year) # Check for an annotation 'firstmodelyear: true' if eval_anno(code, "firstmodelyear"): if fmy_set: # No coverage: data that triggers this should not be committed raise ValueError( # pragma: no cover "≥2 periods are annotated firstmodelyear: true" ) self.y0 = year self.set["cat_year"].append(("firstmodelyear", year)) fmy_set = True # Store the duration_period: either from an annotation, or computed vs. the # prior period duration_period.append( dict( year=year, value=eval_anno(code, "duration_period") or (year - duration_period[-1]["year"]), unit="y", ) ) # Store self.par["duration_period"] = pd.DataFrame(duration_period)
[docs]@dataclass class Spec: """A specification for the structure of a model or variant. A Spec collects 3 :class:`.ScenarioInfo` instances at the attributes :attr:`.add`, :attr:`.remove`, and :attr:`.require`. This is the type that is accepted by :func:`.apply_spec`; :doc:`model-build` describes how a Spec is used to modify a :class:`Scenario`. A Spec may also be used to express information about the target structure of data to be prepared; like :class:`.ScenarioInfo`, this can happen before the target :class:`.Scenario` exists. Spec also provides: - Dictionary-style access, e.g. ``s["add"]`` is equivalent to ``s.add.``. - Error checking; setting keys other than add/remove/require results in an error. - :meth:`.merge`, a helper method. """ #: Structure to be added to a base scenario. add: ScenarioInfo = field(default_factory=ScenarioInfo) #: Structure to be removed from a base scenario. remove: ScenarioInfo = field(default_factory=ScenarioInfo) #: Structure that must be present in a base scenario. require: ScenarioInfo = field(default_factory=ScenarioInfo) # Dict-like features def __getitem__(self, key): try: return getattr(self, key) except AttributeError: raise KeyError(key) def __setitem__(self, key, value: ScenarioInfo): if not hasattr(self, key): raise KeyError(key) setattr(self, key, value) def values(self): yield self.add yield self.remove yield self.require # Static methods
[docs] @staticmethod def merge(a: "Spec", b: "Spec") -> "Spec": """Merge Specs `a` and `b` together. Returns a new Spec where each member is a union of the respective members of `a` and `b`. """ result = Spec() for key in {"add", "remove", "require"}: result[key].update(a[key]) result[key].update(b[key]) return result