Low-level utilities (`util`)

Submodules:

`click`	Command-line utilities.
`context`	Context and settings for `message_ix_models` code.
`importlib`	Load model and project code from `message_data`.
`_logging`	Logging utilities.
`node`	Utilities for nodes.
`pooch`	Utilities for using Pooch.
`scenarioinfo`	`ScenarioInfo` class.
`sdmx`	Utilities for handling objects from `sdmx`.

Commonly used:

`Config`(local_data, platform_info, str] =, ...)	Top-level configuration for `message_ix_models` and `message_data`.
`ConfigHelper`()	Mix-in for `dataclass`-based configuration classes.
`Context`(args, *kwargs)	Context and settings for `message_ix_models` code.
`ScenarioInfo`(scenario_obj, empty, ...)	Information about a `Scenario` object.
`Spec`(add, remove, require)	A specification for the structure of a model or variant.
`broadcast`(df[, labels])	Fill missing data in df by broadcasting.
`cached`(func)	Decorator to cache the return value of a function func.
`check_support`(context[, settings, desc])	Check whether a Context is compatible with certain settings.
`convert_units`(s, unit_info[, store])	Convert units of s, for use with `apply()`.
`copy_column`(column_name)	For use with `pandas.DataFrame.assign()`.
`datetime_now_with_tz`()	Current date and time with time zone information.
`ffill`(df, dim, values[, expr])	Forward-fill df on dim to cover values.
`identify_nodes`(scenario)	Return the ID of a node codelist given the contents of scenario.
`iter_keys`(base)	Return an iterator over a sequence of keys starting with base_key.
`load_package_data`(*parts[, suffix])	Load a `message_ix_models` package data file and return its contents.
`load_private_data`(*parts)	Load a private data file from `message_data` and return its contents.
`local_data_path`(*parts)	Construct a path for local data.
`make_io`(src, dest, efficiency[, on])	Return input and output data frames for a 1-to-1 technology.
`make_matched_dfs`(base, **par_value)	Return data frames derived from base for multiple parameters.
`make_source_tech`(info, common, **values)	Return parameter data for a ‘source’ technology.
`maybe_query`(series, query)	Apply `pandas.DataFrame.query()` if the query arg is not `None`.
`merge_data`(base, *others)	Merge dictionaries of DataFrames together into base.
`minimum_version`(expr)	Decorator for functions that require a minimum version of some upstream package.
`nodes_ex_world`(nodes)	Exclude "World" and anything containing "GLB" from nodes.
`package_data_path`(*parts)	Construct a path to a file under `message_ix_models/data/`.
`private_data_path`(*parts)	Construct a path to a file under `data/` in `message_data`.
`same_node`(df[, from_col])	Fill 'node_{,dest,loc,origin,rel,share}' in df from from_col.
`same_time`(df)	Fill 'time_origin'/'time_dest' in df from 'time'.
`series_of_pint_quantity`(args, *kwargs)	Suppress a spurious warning.
`show_versions`()	Output of `ixmp.show_versions()`, as a `str`.

class message_ix_models.util.Adapter[source]

Adapt data.

Adapter is an abstract base class for tools that adapt data in any way, e.g. between different code lists for certain dimensions. An instance of an Adapter can be called with any of the following as data:

genno.Quantity,
pandas.DataFrame, or
dict mapping str parameter names to values (either of the above types).

…and will return data of the same type.

Subclasses can implement different adapter logic by overriding the abstract adapt() method.

abstract adapt(qty: AttrSeries) → AttrSeries[source]: Adapt data.

class message_ix_models.util.MappingAdapter(maps: Mapping[str, Sequence[Tuple[str, str]]])[source]

Adapt data using mappings for 1 or more dimension(s).

Parameters:: maps (dict of Sequence of tuple) – Keys are names of dimensions. Values are sequences of 2-tuples; each tuple consists of an original label and a target label.

Examples

>>> a = MappingAdapter({"foo": [("a", "x"), ("a", "y"), ("b", "z")]})
>>> df = pd.DataFrame(
...     [["a", "m", 1], ["b", "n", 2]], columns=["foo", "bar", "value"]
... )
>>> a(df)
  foo  bar  value
0   x    m      1
1   y    m      1
2   z    n      2

adapt(qty: AttrSeries) → AttrSeries[source]: Adapt data.

message_ix_models.util.add_par_data(scenario: Scenario, data: Mapping[str, DataFrame], dry_run: bool = False)[source]

Add data to scenario.

Parameters:

data – Dict with keys that are parameter names, and values are pd.DataFrame or other arguments
dry_run (optional) – Only show what would be done.

See also

strip_par_data

message_ix_models.util.aggregate_codes(df: DataFrame, dim: str, codes)[source]: Aggregate df along dimension dim according to codes.

message_ix_models.util.broadcast(df: DataFrame, labels: DataFrame | None = None, **kwargs) → DataFrame[source]

Fill missing data in df by broadcasting.

broadcast() is suitable for use with partly-filled data frames returned by message_ix.util.make_df(), with 1 column per dimension, plus a “value” column. It is also usable with pandas.DataFrame.pipe() for chained operations.

labels (if any) are handled first: one copy or duplicate of df is produced for each row (set of labels) in this argument. Then, kwargs are handled; broadcast() returns one copy for each element in the cartesian product of the dimension labels given by kwargs.

Parameters:

labels (pandas.DataFrame) – Each column (dimension) corresponds to one in df. Each row represents one matched set of labels for those dimensions.
kwargs – Keys are dimensions. Values are labels along that dimension to fill.

Returns:

The length is either 1 or an integer multiple of the length of df.

Return type:

pandas.DataFrame

Raises:

ValueError – if any of the columns in labels or kwargs are not present in df, or if those columns are present but not empty.

Examples

>>> from message_ix import make_df
>>> from message_ix_models.util import broadcast
# Create a base data frame with some empty columns
>>> base = make_df("input", technology="t", value=[1.1, 2.2])
# Broadcast (duplicate) the data across 2 dimensions
>>> df = base.pipe(broadcast, node_loc=["node A", "node B"], mode=["m0", "m1"])
# Show part of the result
>>> df.dropna(axis=1)
  mode node_loc technology  value
0   m0   node A          t    1.1
1   m0   node A          t    2.2
2   m0   node B          t    1.1
3   m0   node B          t    2.2
4   m1   node A          t    1.1
5   m1   node A          t    2.2
6   m1   node B          t    1.1
7   m1   node B          t    2.2

message_ix_models.util.cached(func: Callable) → Callable[source]

Decorator to cache the return value of a function func.

On a first call, the data requested is returned and also cached under Context.get_cache_path(). On subsequent calls, if the cache exists, it is used instead of calling the (possibly slow) func.

When SKIP_CACHE is true, func is always called.

See also

Caching in the genno documentation

message_ix_models.util.check_support(context, settings={}, desc: str = '') → None[source]

Check whether a Context is compatible with certain settings.

Raises:

NotImplementedError – if any context value for a key of settings is not among the values in settings.
KeyError – if the key is not set on context at all.

See also

Advertise and check compatibility

message_ix_models.util.convert_units(s: Series, unit_info: Mapping[str, Tuple[float, str, str | None]], store='magnitude') → Series[source]

Convert units of s, for use with apply().

s.name is used to retrieve a tuple of (factor, input_unit, output_unit) from unit_info. The (float) values of s are converted to pint.Quantity with the input_unit and factor; then cast to output_unit, if provided.

Parameters:

s (pandas.Series) –
unit_info (dict (str -> tuple)) – Mapping from quantity name (matched to s.name) to 3-tuples of (factor, input_unit, output_unit). output_unit may be None. For example, see ikarus.UNITS.
store ("magnitude" or "quantity") – If “magnitude”, the values of the returned series are the magnitudes of the results, with no output units. If “quantity”, the values are scalar Quantity objects.

Returns:

Same shape, index, and values as s, with output units.

Return type:

pandas.Series

message_ix_models.util.copy_column(column_name)[source]

For use with pandas.DataFrame.assign().

Examples

Modify df by filling the column ‘baz’ with the value 3, and copying the column ‘bar’ into column ‘foo’.

>>> df.assign(foo=copy_column('bar'), baz=3)

Note that a similar assignment can be achieved with eval():

>>> df.eval("foo = bar")

copy_column() is useful in the context of more complicated calls to assign().

message_ix_models.util.datetime_now_with_tz() → datetime[source]: Current date and time with time zone information.

message_ix_models.util.ffill(df: DataFrame, dim: str, values: Sequence[str | Code], expr: str | None = None) → DataFrame[source]

Forward-fill df on dim to cover values.

Parameters:

df (pandas.DataFrame) – Data to fill forwards.
dim (str) – Dimension to fill along. Must be a column in df.
values (list of str) – Labels along dim that must be present in the returned data frame.
expr (str, optional) – If provided, DataFrame.eval() is called. This can be used to assign one column to another. For instance, if dim == “year_vtg” and expr is “year_act = year_vtg”, then forward filling is performed along the “year_vtg” dimension/ column, and then the filled values are copied to the “year_act” column.

message_ix_models.util.identify_nodes(scenario: Scenario) → str[source]

Return the ID of a node codelist given the contents of scenario.

Returns:: The ID of the Node code lists containing the regions of scenario.
Return type:: str
Raises:: ValueError – if no codelist can be identified, or the nodes in the scenario do not match the children of the “World” node in the codelist.

message_ix_models.util.iter_keys(base: genno.Key) → KeyIterator[source]

Return an iterator over a sequence of keys starting with base_key.

This can be used for shorthand when constructing sequences of genno computations.

Example

>>> base_key = genno.Key("foo:a-b-c")
>>> k = iter_keys(base_key)
>>> k()
<foo:a-b-c:0>
>>> k()
<foo:a-b-c:1>
>>> k()
<foo:a-b-c:2>

message_ix_models.util.load_package_data(*parts: str, suffix: str | None = '.yaml') → Any[source]

Load a message_ix_models package data file and return its contents.

Data is re-used if already loaded.

Example

The single call:

>>> info = load_package_data("node", "R11")

loads the metadata file data/node/R11.yaml, parsing its contents,
stores those values at PACKAGE_DATA["node R11"] for use by other code, and
returns the loaded values.

Parameters:

parts (Iterable of str) – Used to construct a path under message_ix_models/data/.
suffix (str, optional) – File name suffix, including, the “.”, e.g. .yaml.

Returns:

Configuration values that were loaded.

Return type:

dict

message_ix_models.util.load_private_data(*parts: str) → Mapping[source]

Load a private data file from message_data and return its contents.

Analogous to load_package_data(), but for non-public data.

Parameters:: parts (Iterable of str) – Used to construct a path under data/ in the message_data repository.
Returns:: Configuration values that were loaded.
Return type:: dict
Raises:: RuntimeError – if message_data is not installed.

message_ix_models.util.local_data_path(*parts) → Path[source]

Construct a path for local data.

The setting message local data in the user’s ixmp configuration file is used as a base path. If this is not configured, the current working directory is used.

Parameters:: parts (Sequence of str or Path) – Joined to the base path using Path.joinpath().

See also

Choose locations for data

message_ix_models.util.make_io(src, dest, efficiency, on='input', **kwargs)[source]

Return input and output data frames for a 1-to-1 technology.

Parameters:

src (tuple of str) – Input (commodity, level, unit)
dest (tuple of str) – Output (commodity, level, unit)
efficiency (float) – Conversion efficiency.
on ('input' or 'output') – If ‘input’, efficiency applies to the input, and the output, thus the activity level of the technology, is in dest[2] units. If ‘output’, the opposite.
kwargs – Passed to make_df().

Returns:

Keys are ‘input’ and ‘output’; values are data frames.

Return type:

dict (str -> pd.DataFrame)

message_ix_models.util.make_matched_dfs(base: MutableMapping, **par_value: float | Quantity) → Dict[str, DataFrame][source]

Return data frames derived from base for multiple parameters.

Creates one data frame per keyword argument.

Parameters:

base (pandas.DataFrame, dict, etc.) – Used to populate other columns of each data frame. Duplicates—which occur when the target parameter has fewer dimensions than base—are dropped.
par_values – Argument names (e.g. ‘fix_cost’) are passed to make_df(). If the value is float, it overwrites the “value” column; if pint.Quantity, its magnitude overwrites “value” and its units the “units” column, as a formatted string.

Returns:

one for each parameter in par_values.

Return type:

dict of pandas.DataFrame

Examples

>>> input = make_df("input", ...)
>>> cf_tl = make_matched_dfs(
>>>     input,
>>>     capacity_factor=1,
>>>     technical_lifetime=pint.Quantity(8, "year"),
>>> )

message_ix_models.util.make_source_tech(info: Scenario | ScenarioInfo, common, **values) → Dict[str, DataFrame][source]

Return parameter data for a ‘source’ technology.

The technology has no inputs; its output commodity and/or level are determined by common; either single values, or None if the result will be pipe()’d through broadcast().

Parameters:

info (Scenario or ScenarioInfo) –
common (dict) – Passed to make_df().
**values – Values for ‘capacity_factor’ (optional; default 1.0), ‘output’, ‘var_cost’, and optionally ‘technical_lifetime’.

Returns:

Suitable for add_par_data().

Return type:

dict

message_ix_models.util.mark_time(quiet: bool = False) → None[source]: Record and log (if quiet is True) a time mark.

message_ix_models.util.maybe_query(series: Series, query: str | None) → Series[source]

Apply pandas.DataFrame.query() if the query arg is not None.

query() is not chainable (pandas-dev/pandas#37941). Use this function with pandas.Series.pipe(), passing an argument that may be None, to have a chainable query operation that can be a no-op.

message_ix_models.util.merge_data(base: MutableMapping[str, DataFrame], *others: Mapping[str, DataFrame]) → None[source]: Merge dictionaries of DataFrames together into base.

message_ix_models.util.minimum_version(expr: str) → Callable[source]

Decorator for functions that require a minimum version of some upstream package.

See prepare_reporter() / test_prepare_reporter() for a usage example.

Parameters:: expr – Like “example 1.2.3.post0”. The condition for the decorated function is that the installed version must be equal to or greater than this version.

message_ix_models.util.package_data_path(*parts) → Path[source]

Construct a path to a file under message_ix_models/data/.

Use this function to access data packaged and installed with message_ix_models.

Parameters:: parts (Sequence of str or Path) – Joined to the base path using joinpath().

See also

Choose locations for data

message_ix_models.util.path_fallback(*parts: str | Path, where: str | List[str | Path] = '') → Path[source]

Locate a path constructed from parts found in the first of several directories.

This allows to implement ‘fallback’ behaviour in which files or directories in certain locations are used preferentially.

Parameters:

parts – Path parts or fragments such as directory names and a final file name.
where –
Either:
- str containing one or more of:
  - ”cache”: locate parts in the message_ix_models cache directory.
  - ”package”: locate parts in message_ix_models package data (same as package_data_path()).
  - ”private”: locate parts in the message_data /data/ directory (same as private_data_path()).
  - ”test”: locate test data in package_data_path("test", ...)
- list where each element is str (one of the above) or a pathlib.Path.

Returns:

The first of the locations indicated by where in which the file or directory parts exists.

Return type:

pathlib.Path

Raises:

ValueError – If where is empty or parts are not found in any of the indicated locations.

message_ix_models.util.preserve_log_level()[source]: Context manager to preserve the level of the message_ix_models logger.

message_ix_models.util.private_data_path(*parts) → Path[source]

Construct a path to a file under data/ in message_data.

Use this function to access non-public (for instance, embargoed or proprietary) data stored in the message_data repository.

If the repository is not available, the function falls back to Context.get_local_path(), where users may put files obtained through other messages.

Parameters:: parts (Sequence of str or Path) – Joined to the base path using joinpath().

See also

Choose locations for data

Replace data in parameters of scenario.

Parameters:

scenario – Scenario in which to replace data.
parameters (str or Sequence of str) – Name(s) of parameters in which to replace data.
filters – Passed to Scenario.par() argument of the same name.
to_replace – Passed to pandas.DataFrame.replace() argument of the same name.

Examples

Replace data in the “relation_activity” parameter for a particular technology and relation: assign the same values as entries in a different relation name for the same technology.

>>> replace_par_data(
...     scenario,
...     "relation_activity",
...     dict(technology="hp_gas_i", relation="CO2_r_c"),
...     dict(relation={"CO2_r_c": "CO2_ind"}),
... )

message_ix_models.util.same_node(df: DataFrame, from_col='node_loc') → DataFrame[source]: Fill ‘node_{,dest,loc,origin,rel,share}’ in df from from_col.

message_ix_models.util.same_time(df: DataFrame) → DataFrame[source]: Fill ‘time_origin’/’time_dest’ in df from ‘time’.

message_ix_models.util.series_of_pint_quantity(*args, **kwargs) → Series[source]

Suppress a spurious warning.

Creating a pandas.Series with a list of pint.Quantity triggers a warning “The unit of the quantity is stripped when downcasting to ndarray,” even though the entire object is being stored and the unit is not stripped. This function suppresses this warning.

message_ix_models.util.show_versions() → str[source]: Output of ixmp.show_versions(), as a str.

message_ix_models.util.silence_log(names: str | None = None, level: int = 40)[source]

Context manager to temporarily quiet 1 or more loggers.

Parameters:

names (str, optional) – Space-separated names of loggers to quiet.
level (int, optional) – Minimum level of log messages to allow.

Examples

>>> with silence_log():
>>>     log.warning("This message is not recorded.")

message_ix_models.util.strip_par_data(scenario: Scenario, set_name: str, element: str, dry_run: bool = False, dump: Dict[str, DataFrame] | None = None) → int[source]

Remove element from set_name in scenario, optionally dumping to dump.

Parameters:

dry_run (bool, optional) – If True, only show what would be done.
dump (dict, optional) – If provided, stripped data are stored in this dictionary. Otherwise, they are discarded.

Returns:

Total number of rows removed across all parameters.

Return type:

int

See also

add_par_data

message_ix_models.util.cache.SKIP_CACHE = False: Controls whether cached data is returned for functions decorated with cached(). Set to True to force reload.

`util.click`

Command-line utilities.

These are used for building CLIs using click.

PARAMS contains, among others:

--urls-from-file=… Path to a file containing scenario URLs, one per line. These are parsed and stored on Config.scenarios.

class message_ix_models.util.click.CliRunner(cli_cmd: ~click.core.Command, cli_module: str, env: ~typing.Mapping[str, str] = <factory>, charset: str = 'utf-8', method: ~typing.Literal['click', 'subprocess'] = 'click')[source]

Similar to click.testing.CliRunner, with extra features.

assert_exit_0(*args, **kwargs) → Result[source]

Assert a result has exit_code 0, or print its traceback.

If any args or kwargs are given, invoke() is first called. Otherwise, the result from the last call of invoke() is used.

Raises:: AssertionError – if the result exit code is not 0.

cli_cmd: Command: CLI entry point

cli_module: str: CLI module

invoke_subprocess(*args, **kwargs) → Result[source]: Invoke the CLI in a subprocess.

method: Literal['click', 'subprocess'] = 'click': Method for invoking the command

message_ix_models.util.click.PARAMS = {'dest': <Option dest>, 'dry_run': <Option dry_run>, 'force': <Option force>, 'nodes': <Option nodes>, 'output_model': <Option output_model>, 'platform_dest': <Option platform_dest>, 'policy_path': <Option policy_path>, 'quiet': <Option quiet>, 'regions': <Option regions>, 'rep_out_path': <Option rep_out_path>, 'rep_template': <Option rep_template>, 'run_reporting_only': <Option run_reporting_only>, 'ssp': <Argument ssp>, 'urls_from_file': <Option urls_from_file>, 'verbose': <Option verbose>, 'years': <Option years>}: Common command-line parameters (arguments and options). See common_params().

message_ix_models.util.click.common_params(param_names: str)[source]

Decorate a click.command with common parameters param_names.

param_names must be a space-separated string of names appearing in PARAMS, e.g. "ssp force output_model". The decorated function receives keyword arguments with these names:

@click.command()
@common_params("ssp force output_model")
def mycmd(ssp, force, output_model)
    # ...

message_ix_models.util.click.default_path_cb(*default_parts)[source]

Return a callback function for click.Option handling.

If no option value is given, the callback uses Context.get_local_path() and default_parts to provide a path that is relative to local data directory, e.g. the current working directory (see Data, metadata, and configuration).

message_ix_models.util.click.exec_cb(expression: str) → Callable[source]

Return a callback that exec()-utes an expression.

The expression is executed in a limited context that has only two names available:

context: the Context instance.
value: the value passed to the click.Parameter.

Example

>>> @click.command
... @click.option(
...     "--myopt", callback=exec_cb("context.my_mod.my_opt = value + 3")
... )
... def cmd(...):
...     ...

message_ix_models.util.click.format_sys_argv() → str[source]: Format sys.argv in a readable manner.

message_ix_models.util.click.store_context(context: Context | Context, param, value)[source]

Callback that simply stores a value on the Context object.

Use this for parameters that are not used directly in a @click.command() function, but need to be carried by the Context for later use.

message_ix_models.util.click.temporary_command(group: Group, command: Command)[source]: Temporarily attach command command to group.

message_ix_models.util.click.unique_id() → str[source]

Return a unique ID for a CLI invocation.

The return value resembles “mix-models-debug-3332d415ef65bf2a-2023-02-02T162931” and contains:

The CLI name and (sub)commands.
A hash of all the CLI parameters (options and arguments).
The current date and time, in ISO format with Windows-incompatible “:” removed.

message_ix_models.util.click.urls_from_file(context: Context | Context, param, value) → List[ScenarioInfo][source]: Callback to parse scenario URLs from value.

`util.config`

class message_ix_models.util.config.ConfigHelper[source]

Mix-in for dataclass-based configuration classes.

This provides 3 methods—read_file(), replace(), and from_dict()— that help to use dataclass classes for handling message_ix_models configuration.

All 3 methods take advantage of name manipulations: the characters “-” and ” ” are replaced with underscores (“_”). This allows to write the names of attributes in legible ways—e.g. “attribute name” or “attribute-name” instead of “attribute_name”— in configuration files and/or code.

classmethod from_dict(data: Mapping)[source]: Construct an instance from data with name manipulation.

read_file(path: Path, fail='raise') → None[source]

Update configuration from file.

Parameters:

path – to a .yaml file containing a top-level mapping.
fail (str) – if “raise” (the default), any names in path which do not match attributes of the dataclass raise a ValueError. Ottherwise, a message is logged.

replace(**kwargs)[source]: Like dataclasses.replace() with name manipulation.

update(**kwargs)[source]

Update attributes in-place.

Raises:: AttributeError – Any of the kwargs are not fields in the data class.

`util.context`

Context and settings for message_ix_models code.

class message_ix_models.util.context.Context(*args, **kwargs)[source]

Context and settings for message_ix_models code.

Context is a subclass of dict, so common methods like copy() and setdefault() may be used to handle settings. To be forgiving, it also provides attribute access; context.foo is equivalent to context["foo"].

A Context instance always has the following members:

core: an instance of message_ix_models.Config.
model: an instance of message_ix_models.model.Config.

Attributes of these classes may be accessed by shorthand, e.g. context.regions is shorthand for context.model.regions.

Context provides additional methods to do common tasks that depend on configurable settings:

`clone_to_dest`([create])	Return a scenario based on the `--dest` command-line option.
`close_db`()
`delete`()	Hide the current Context from future `get_instance()` calls.
`get_cache_path`(*parts)	Return a path to a local cache file, i.e. within `Config.cache_path`.
`get_local_path`(*parts[, suffix])	Return a path under `Config.local_data`.
`get_platform`([reload])	Return a `Platform` from `Config.platform_info`.
`get_scenario`()	Return a `Scenario` from `scenario_info`.
`handle_cli_args`([url, platform, model_name, ...])	Handle command-line arguments.
`only`()	Return the only `Context` instance.
`use_defaults`(settings)	Update from settings.

clone_to_dest(create=True) → Scenario[source]

Return a scenario based on the --dest command-line option.

Parameters:

create (bool, optional) – If True (the default) and the base scenario does not exist, a bare RES scenario is created. Otherwise, an exception is raised.

Returns:

To prevent the scenario from being garbage collected, keep a reference to its Platform:

Return type:

Scenario

See also

create_res

To use this method, either decorate a command with common_params():

from message_data.tools.cli import common_params

@click.command()
@common_params("dest")
@click.pass_obj
def foo(context, dest):
    scenario, mp = context.clone_to_dest()

or, store the settings Config.dest_scenario and optionally Config.dest_platform on context:

c = Context.get_instance()

c.dest_scenario = dict(model="foo model", scenario="foo scenario")
scenario_mp = context.clone_to_dest()

The resulting scenario has the indicated model- and scenario names.

If --url (or --platform, --model, --scenario, and optionally --version) are given, the identified scenario is used as a ‘base’ scenario, and is cloned. If --url/--platform and --dest refer to different Platform instances, then this is a two-platform clone.

If no base scenario can be loaded, bare.create_res() is called to generate a base scenario.

delete()[source]: Hide the current Context from future get_instance() calls.

get_cache_path(*parts) → Path[source]

Return a path to a local cache file, i.e. within Config.cache_path.

The directory containing the resulting path is created if it does not already exist.

classmethod get_instance(index=0) → Context[source]

Return a Context instance; by default, the first created.

Parameters:: index (int, optional) – Index of the Context instance to return, e.g. -1 for the most recently created.

get_local_path(*parts: str, suffix=None) → Path[source]

Return a path under Config.local_data.

Parameters:

parts – Path fragments, for instance directories, passed to joinpath().
suffix – File name suffix including a “.”—for instance, “.csv”—passed to with_suffix().

get_platform(reload=False) → Platform[source]

Return a Platform from Config.platform_info.

When used through the CLI, Config.platform_info is a ‘base’ platform as indicated by the –url or –platform options.

If a Platform has previously been instantiated with get_platform(), the same object is returned unless reload=True.

get_scenario() → Scenario[source]

Return a Scenario from scenario_info.

When used through the CLI, scenario_info is a ‘base’ scenario for an operation, indicated by the --url or --platform/--model/--scenario options.

handle_cli_args(url=None, platform=None, model_name=None, scenario_name=None, version=None, local_data=None, verbose=False, _store_as=('platform_info', 'scenario_info'))[source]

Handle command-line arguments.

May update the Config.local_data, platform_info, scenario_info, and/or url settings.

classmethod only() → Context[source]

Return the only Context instance.

Raises:: IndexError – If there is more than one instance.

set_scenario(scenario: Scenario) → None[source]

Update Config.scenario_info to match an existing scenario.

Config.url is also updated.

update([E, ]**F) → None. Update D from dict/iterable E and F.[source]: If E is present and has a .keys() method, then does: for k in E: D[k] = E[k] If E is present and lacks a .keys() method, then does: for k, v in E: D[k] = v In either case, this is followed by: for k in F: D[k] = F[k]

use_defaults(settings)[source]: Update from settings.

write_debug_archive() → None[source]

Write an archive containing the files listed in debug_paths.

The archive file name is constructed using unique_id() and appears in a debug subdirectory under the local data path.

The archive also contains a file command.txt that gives the full command-line used to invoke mix-models.

`util.importlib`

Load model and project code from message_data.

class message_ix_models.util.importlib.MessageDataFinder[source]: Load model and project code from message_data.

`util._logging`

Logging utilities.

class message_ix_models.util._logging.Formatter(use_colour: bool = True)[source]

Formatter for log records.

Parameters:: use_color (bool, optional) – If True, colorama is used to colour log messages.

format(record)[source]

Format record.

Records are formatted like:

model.transport.data.add_par_data  220 rows in 'input'
...add_par_data  further messages

…with the calling function name (e.g. ‘add_par_data’) coloured for legibility on first occurrence, then dimmed when repeated.

class message_ix_models.util._logging.QueueListener(queue, *handlers, respect_handler_level=False)[source]

logging.QueueListener with a flush() method.

flush()[source]: Flush the queue: join the listener/monitor thread and then restart.

class message_ix_models.util._logging.SilenceFilter(names: str, level: int)[source]

Log filter that only allows records from names that are at or above level.

filter(record) → bool[source]

Determine if the specified record is to be logged.

Returns True if the record should be logged, or False otherwise. If deemed appropriate, the record may be modified in-place.

class message_ix_models.util._logging.StreamHandler(stream_name: str)[source]

Like logging.StreamHandler, but retrieve the stream on each access.

This avoids the case that click, pytest, or something else adjusts sys.stdout temporarily, but the handler’s stored reference to the original is not updated.

stream_name: str: Name of the sys stream to use, as str rather than a direct reference.

message_ix_models.util._logging.setup(level: str | int = 99, console: bool = True, *, file: bool = False) → None[source]

Initialize logging.

Parameters:

level (str, optional) – Log level for the console log handler.
console (bool, optional) – If False, do not print any messages to console.
file (bool, optional) – If False, do not print any messages to file.

`util.node`

Utilities for nodes.

message_ix_models.util.node.NODE_DIMS = ['n', 'node', 'node_loc', 'node_origin', 'node_dest', 'node_rel', 'node_share']: Names of dimensions indexed by ‘node’.

Todo

to be robust to changes in message_ix, read these names from that package.

message_ix_models.util.node.R11_R12 = (('R11_AFR', 'R12_AFR'), ('R11_CPA', 'R12_CHN'), ('R11_EEU', 'R12_EEU'), ('R11_FSU', 'R12_FSU'), ('R11_LAM', 'R12_LAM'), ('R11_MEA', 'R12_MEA'), ('R11_NAM', 'R12_NAM'), ('R11_PAO', 'R12_PAO'), ('R11_PAS', 'R12_PAS'), ('R11_CPA', 'R12_RCPA'), ('R11_SAS', 'R12_SAS'), ('R11_WEU', 'R12_WEU')): Mapping from R11 to R12 node IDs.

message_ix_models.util.node.R11_R14 = (('R11_AFR', 'R14_AFR'), ('R11_FSU', 'R14_CAS'), ('R11_CPA', 'R14_CPA'), ('R11_EEU', 'R14_EEU'), ('R11_LAM', 'R14_LAM'), ('R11_MEA', 'R14_MEA'), ('R11_NAM', 'R14_NAM'), ('R11_PAO', 'R14_PAO'), ('R11_PAS', 'R14_PAS'), ('R11_FSU', 'R14_RUS'), ('R11_SAS', 'R14_SAS'), ('R11_FSU', 'R14_SCS'), ('R11_FSU', 'R14_UBM'), ('R11_WEU', 'R14_WEU')): Mapping from R11 to R14 node IDs.

message_ix_models.util.node.adapt_R11_R12 = <message_ix_models.util.common.MappingAdapter object>

Adapt data from the R11 to the R14 node list.

The data is adapted using the mappings in R11_R12 for each of the dimensions in NODE_DIMS.

message_ix_models.util.node.adapt_R11_R14 = <message_ix_models.util.common.MappingAdapter object>

Adapt data from the R11 to the R14 node list.

The data is adapted using the mappings in R11_R14 for each of the dimensions in NODE_DIMS.

message_ix_models.util.node.nodes_ex_world(nodes: Sequence[str | Code]) → List[str | Code][source]

Exclude “World” and anything containing “GLB” from nodes.

May also be used as a genno (reporting) operator.

`util.pooch`

Utilities for using Pooch.

class message_ix_models.util.pooch.Extract(members=None, extract_dir=None)[source]

Similar to pooch.Unzip, streamlined using pathlib.

This version supports:

Absolute or relative paths for the extract_dir parameter.
.zip or .tar.xz archives.

message_ix_models.util.pooch.GH_MAIN = 'https://github.com/iiasa/message-ix-models/raw/main/message_ix_models/data': Base URL portion for files stored in the message-ix-models GitHub repository.

message_ix_models.util.pooch.SOURCE: Mapping[str, Mapping[str, Any]] = {'MESSAGEix-Nexus': {'pooch_args': {'base_url': 'https://github.com/iiasa/message-ix-models/raw/main/message_ix_models/data/water/', 'registry': {'water.tar.xz': 'sha1:ec9e0655af90ca844c0158968bb03a194b8fa6c6'}}, 'processor': <message_ix_models.util.pooch.Extract object>}, 'PRIMAP': {'pooch_args': {'base_url': 'ftp://datapub.gfz-potsdam.de/download/10.5880.PIK.2019.001/', 'registry': {'PRIMAP-hist_v2.0_11-Dec-2018.zip': 'md5:f28d58abef4ecfc36fc8ce3e9eef2871'}}, 'processor': <message_ix_models.util.pooch.Extract object>}, 'SSP-Update-3.0': {'pooch_args': {'base_url': 'https://github.com/iiasa/message-ix-models/raw/main/message_ix_models/data/ssp/', 'registry': {'1706548837040-ssp_basic_drivers_release_3.0_full.csv.gz': 'sha1:e2af7a88aeed7d0e44ceaf2dff60f891cf551517'}}}, 'SSP-Update-3.0.1': {'pooch_args': {'base_url': 'https://github.com/iiasa/message-ix-models/raw/main/message_ix_models/data/ssp/', 'registry': {'1710759470883-ssp_basic_drivers_release_3.0.1_full.csv.gz': 'sha1:e5c24c27ee743e79dac5a578235b35a68cd64183'}}}, 'snapshot-0': {'pooch_args': {'base_url': 'doi:10.5281/zenodo.5793870', 'registry': {'MESSAGEix-GLOBIOM_1.1_R11_no-policy_baseline.xlsx': 'md5:222193405c25c3c29cc21cbae5e035f4'}}, 'processor': <message_ix_models.util.pooch.UnpackSnapshot object>}, 'snapshot-1': {'pooch_args': {'base_url': 'doi:10.5281/zenodo.10514052', 'registry': {'MESSAGEix-GLOBIOM_1.1_R11_no-policy_baseline.xlsx': 'md5:e7c0c562843e85c643ad9d84fecef979'}}}}: Supported remote sources of data.

class message_ix_models.util.pooch.UnpackSnapshot[source]: Pooch processor that calls snapshot.unpack().

message_ix_models.util.pooch.fetch(pooch_args: dict, *, extra_cache_path: str | None = None, **fetch_kwargs) → Tuple[Path, ...][source]

Create a Pooch instance and fetch a single file.

Files are stored under the directory identified by Context.get_cache_path(), unless args provides another location.

Parameters:

args – Passed to pooch.create().
kwargs – Passed to pooch.Pooch.fetch().

Returns:

Path to the fetched file.

Return type:

Path

See also

snapshot.load()

`util.pycountry`

message_ix_models.util.pycountry.COUNTRY_NAME = {'Korea': 'Korea, Republic of', 'Republic of Korea': 'Korea, Republic of', 'Russia': 'Russian Federation', 'South Korea': 'Korea, Republic of', 'Turkey': 'Türkiye'}: Mapping from common, non-standard country names to ISO 3166-1 names.

message_ix_models.util.pycountry.iso_3166_alpha_3(name: str) → str | None[source]

Return an ISO 3166 alpha-3 code for a country name.

Parameters:: name (str) – Country name. This is looked up in the pycountry ‘name’, ‘official_name’, or ‘common_name’ field. Values in COUNTRY_NAME are supported.
Return type:: str or None

`util.scenarioinfo`

ScenarioInfo class.

class message_ix_models.util.scenarioinfo.ScenarioInfo(scenario_obj: dataclasses.InitVar[typing.Optional[ForwardRef('Scenario')]] = None, empty: dataclasses.InitVar[bool] = False, platform_name: str | None = None, model: str | None = None, scenario: str | None = None, version: int | None = None, set: ~typing.Dict[str, ~typing.List] = <factory>, par: ~typing.Dict[str, ~pandas.core.frame.DataFrame] = <factory>, y0: int = -1, is_message_macro: bool = False, _yv_ya: ~pandas.core.frame.DataFrame | None = None)[source]

Information about a Scenario object.

Code that prepares data for a target Scenario can accept a ScenarioInfo instance. This avoids the need to create or load an actual Scenario, which can be slow under some conditions.

ScenarioInfo objects can also be used (for instance, by apply_spec()) to describe the contents of a Scenario before it is created.

ScenarioInfo objects have the following convenience attributes:

`set`	Elements of `ixmp`/`message_ix` sets.
`io_units`(technology, commodity[, level])	Return units for the MESSAGE `input` or `output` parameter.
`is_message_macro`	`True` if a MESSAGE-MACRO scenario.
`N`	Elements of the set 'node'.
`units_for`(set_name, id)	Return the units associated with code id in MESSAGE set set_name.
`Y`	Elements of the set 'year' that are >= the first model year.
`y0`	First model year, if set, else `Y[0]`.
`yv_ya`	`pandas.DataFrame` with valid `year_vtg`, `year_act` pairs.

Parameters:: scenario_obj (message_ix.Scenario) – If given, set is initialized from this existing scenario.

Examples

Iterating over an instance gives “model”, “scenario”, “version” and the values of the respective attributes: >>> si = ScenarioInfo.from_url(“model name/scenario name#123”) >>> dict(si) {‘model’: ‘model name’, ‘scenario’: ‘scenario name’, ‘version’: 123}

See also

Spec

property N: Elements of the set ‘node’.

See also

nodes_ex_world

property Y: List[int]: Elements of the set ‘year’ that are >= the first model year.

classmethod from_url(url: str) → ScenarioInfo[source]: Create an instance using only an url.

io_units(technology: str, commodity: str, level: str | None = None) → Unit[source]

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 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.
Raises:: ValueError – if either technology or commodity lack defined units.

is_message_macro: bool = False: True if a MESSAGE-MACRO scenario.

model: str | None = None: Model name; equivalent to TimeSeries.model.

par: Dict[str, DataFrame]: Elements of ixmp/message_ix parameters.

property path: str

A valid file system path name similar to url.

Characters invalid in Windows paths are replaced with “_”.

scenario: str | None = None: Scenario name; equivalent to TimeSeries.scenario.

set: Dict[str, List]: Elements of ixmp/message_ix sets.

units_for(set_name: str, id: str) → Unit[source]

Return the units associated with code id in MESSAGE set set_name.

ixmp (or the sole 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 message_ix_models and 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. Commodities (commodity.yaml)) can be used to carry the standard units for the corresponding quantities. units_for() retrieves these units, for use in model-building and reporting.

Todo

Expand this discussion and transfer to the message_ix docs.

See also

io_units

update(other: ScenarioInfo)[source]: Update with the set elements of other.

property url: str: Identical to TimeSeries.url.

version: int | None = None: Scenario version; equivalent to TimeSeries.version.

y0: int = -1: First model year, if set, else Y[0].

year_from_codes(codes: List[Code])[source]

Update using a list of codes.

The following are updated:

set year
set cat_year, with the first model year.
par duration_period

Any existing values are discarded.

After this, the attributes y0 and 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]

property yv_ya: pandas.DataFrame with valid year_vtg, year_act pairs.

class message_ix_models.util.scenarioinfo.Spec(add: ~message_ix_models.util.scenarioinfo.ScenarioInfo = <factory>, remove: ~message_ix_models.util.scenarioinfo.ScenarioInfo = <factory>, require: ~message_ix_models.util.scenarioinfo.ScenarioInfo = <factory>)[source]

A specification for the structure of a model or variant.

A Spec collects 3 ScenarioInfo instances at the attributes add, remove, and require. This is the type that is accepted by apply_spec(); Building models (model.build) describes how a Spec is used to modify a Scenario. A Spec may also be used to express information about the target structure of data to be prepared; like ScenarioInfo, this can happen before the target 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.
merge(), a helper method.

add: ScenarioInfo: Structure to be added to a base scenario.

static merge(a: Spec, b: Spec) → Spec[source]

Merge Specs a and b together.

Returns a new Spec where each member is a union of the respective members of a and b.

remove: ScenarioInfo: Structure to be removed from a base scenario.

require: ScenarioInfo: Structure that must be present in a base scenario.

`util.sdmx`

Utilities for handling objects from sdmx.

message_ix_models.util.sdmx.as_codes(data: List[str] | Dict[str, str | Code]) → List[Code][source]

Convert data to a list of Code objects.

Various inputs are accepted:

list of str.
dict, in which keys are id and values are further dict with keys matching other Code attributes.

message_ix_models.util.sdmx.eval_anno(obj: AnnotableArtefact, id: str)[source]: Retrieve the annotation id from obj, run eval() on its contents.

Deprecated since version 2023.9.12: Use sdmx.model.common.AnnotableArtefact.eval_annotation(), which provides the same functionality.

message_ix_models.util.sdmx.make_enum(urn, base=<enum 'Enum'>)[source]: Create an enum.Enum (or base) with members from codelist urn.

message_ix_models.util.sdmx.read(urn: str, base_dir: PathLike | None = None)[source]: Read SDMX object from package data given its urn.

message_ix_models.util.sdmx.register_agency(agency: Agency) → AgencyScheme[source]: Add agency to the AgencyScheme “IIASA_ECE:AGENCIES”.

message_ix_models.util.sdmx.write(obj, base_dir: PathLike | None = None, basename: str | None = None)[source]: Store an SDMX object as package data.

Low-level utilities (util)

Low-level utilities (`util`)