Low-level utilities (`util`)

Submodules:

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

Commonly used:

`Config`(_cache_path, _cache_skip, ...)	Core/top-level settings 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`(...)	Convert units of data.
`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[, context])	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`	alias of `MinimumVersionDecorator`
`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`()	Fill 'node_{,dest,loc,origin,rel,share}' in df from from_col.
`same_time`()	Fill 'time_origin'/'time_dest' in df from 'time'.
`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.

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

class message_ix_models.util.MappingAdapter(maps: Mapping[str, Sequence[tuple[str, str]]], *, on_missing: Literal['log', 'raise', 'warn'] | None = None)[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.
on_missing (Literal[``’log’, ``'raise', 'warn'] | None) –
If provided (default None), perform the given action if maps do not contain all the labels in the respective dimensions of each Quantity passed to adapt():
- ”log”: log a message on level logging.WARNING.
- ”raise”: raise RuntimeError.
- ”warn”: emit RuntimeWarning.

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: TQuantity) → TQuantity[source]: Adapt data.

classmethod from_dicts(*values: dict[str, Sequence[str]], dims: Sequence[str], map_leaves: bool = True, on_missing: Literal['log', 'raise', 'warn'] | None) → MappingAdapter[source]: Construct a MappingAdapter from sequences of dict and dimensions.

class message_ix_models.util.WildcardAdapter(dim: str, coords: Sequence[str])[source]

Adapt data using by broadcasting wildcard (“*”) entries for 1 dimension.

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

message_ix_models.util.add_par_data(scenario: Scenario, data: ParameterData, dry_run: bool = False) → int[source]

Add data to scenario.

Parameters:

data – Any mapping with keys that are valid message_ix parameter names, and values that are pd.DataFrame or other arguments valid for message_ix.Scenario.add_par().
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 Config.cache_path(). On subsequent calls, if the cache exists, it is used instead of calling the (possibly slow) func.

When Config.cache_skip 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(data: Any, **kwargs)[source]

message_ix_models.util.convert_units(data: Series, unit_info: Mapping[str, tuple[float, str, str | None]], store='magnitude') → Series

message_ix_models.util.convert_units(data: DataFrame, info: ScenarioInfo) → DataFrame

message_ix_models.util.convert_units(data: dict[str, DataFrame], info: ScenarioInfo) → dict[str, DataFrame]

Convert units of data.

With pandas.Series: for use with apply().

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

Parameters:

data (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, context: Context | None = None) → 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: tuple[str, str, str], dest: tuple[str, str, str], efficiency: float, on: Literal['input', 'output'] = '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 | DataFrame, **par_value: float | Quantity | dict) → MutableParameterData[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) → MutableParameterData[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: MutableParameterData, *others: ParameterData) → None[source]

Merge dictionaries of DataFrames together into base.

For use with genno, see instead report.operator.merge_data() that returns the merged data rather than updating the first argument.

message_ix_models.util.minimum_version: alias of MinimumVersionDecorator

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] = '', context: Context | None = None) → 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 the following, separated by white space:
  - ”cache”: locate parts in the message_ix_models cache directory. See Config.cache_path.
  - ”local”: locate parts in the user’s local data directory (same as local_data_path()).
  - ”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(data: DataFrame, from_col: str = 'node_loc') → DataFrame[source]
message_ix_models.util.same_node(data: MutableParameterData, from_col: str = 'node_loc') → MutableParameterData: Fill ‘node_{,dest,loc,origin,rel,share}’ in df from from_col.

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

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: MutableParameterData | 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

`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: ~collections.abc.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, for instance "ssp force output_model". The decorated function receives keyword arguments with these names; some are also stored on the

Example

>>> @click.command
... @common_params("ssp force output_model")
... @click.pass_obj
... def mycmd(context, ssp, force, output_model):
...     assert context.force == force

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 Configuration and (meta)data).

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.scenario_param(param_decls: str | list[str], *, values: list[str] | None = None, default: str | None = None)[source]

Add an SSP or scenario option or argument to a click.Command.

The parameter uses store_context() to store the given value (if any) on the Context.

Parameters:

param_decls – "--ssp" (or any other name prefixed by --) to generate a click.Option; "ssp" to generate a click.Argument. Click-style declarations are also supported; see below.
values – Allowable values. If not given, the allowable values are [“LED”, “SSP1”, “SSP2”, “SSP3”, “SSP4”, “SSP5”].
default – Default value.

Raises:

ValueError – if default is given with param_decls that indicate a click.Argument.

Examples

Add a (mandatory, positional) click.Argument. This is nearly the same as using common_params("ssp"), except the decorated function does not receive an ssp argument. The value is still stored on context automatically.

>>> @click.command
... @scenario_param("ssp")
... @click.pass_obj
... def mycmd(context):
...     print(context.ssp)

Add a click.Option with certain, limited values and a default:

>>> @click.command
... @scenario_param("--ssp", values=["SSP1", "SSP2", "SSP3"], default="SSP3")
... @click.pass_obj
... def mycmd(context):
...     print(context.ssp)

An option given by the user as --scenario but stored as Context.ssp:

>>> @click.command
... @scenario_param(["--scenario", "ssp"])
... @click.pass_obj
... def mycmd(context):
...     print(context.ssp)

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.Config(_cache_path: ~pathlib.Path = <factory>, _cache_skip: bool = False, debug_paths: ~collections.abc.Sequence[~pathlib.Path] = <factory>, dest: str | None = None, dest_platform: PlatformArgs = <factory>, dest_scenario: ~collections.abc.MutableMapping[str, str] = <factory>, dry_run: bool = False, local_data: ~pathlib.Path = <factory>, platform_info: PlatformArgs = <factory>, _mp: ixmp.Platform | None = None, scenario_info: ~collections.abc.MutableMapping[str, int | str | None] = <factory>, scenarios: list[~message_ix_models.util.scenarioinfo.ScenarioInfo] = <factory>, url: str | None = None, verbose: bool = False)[source]

Core/top-level settings for message_ix_models and message_data.

property cache_path: Path

Base path for cached data.

By default, the directory message-ix-models within the directory given by platformdirs.user_cache_path(). This may be something like $HOME/.cache/message-ix-models/. The --cache-path CLI option modifies this value.

`util.context`

Context and settings for message_ix_models code.

message_ix_models.util.context.MODULE_WITH_CONFIG_DATACLASS: tuple[tuple[str, str, bool, bool], ...] = (('message_ix_models.util.config', 'core', True, True), ('message_ix_models.model.config', 'model', True, True), ('message_ix_models.report.config', 'report', True, False), ('message_ix_models.transport.config', 'transport', False, False))

Tuples containing:

Full name of module that contains a dataclass named Config.
Context key where an instance of the Config class is stored.
True if such an instance should be created by default for every Context instance.

This should be False if creation of the Config instance is slow or has side effects, as this will occur even where the module (1) is not in use.
True if the dataclass fields of the Config class should be ‘aliased’, or directly available on Context instances.

This should be False for all new modules/Config classes. Aliasing is provided only for backwards-compatible support of earlier code.

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

Context and settings for message_ix_models code.

Context is dict-like, so dict methods including copy() and setdefault() may be used to handle settings. 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.
report: an instance of message_ix_models.report.Config.

Attributes of (1) and (2) may be accessed by shorthand/aliases. For instance, context.regions is an alias for context.model.regions. However, for clarity and to support type checking, explicit reference to the configuration class and its attributes should be used.

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`()	Shorthand for `Config.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 `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) → message_ix.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.

asdict() → dict[str, Any][source]

Return a deepcopy() of the Context’s values as a dict.

Currently this is only used in support of util.cache.

close_db() → None[source]: Shorthand for Config.close_db().

property core: message_ix_models.util.config.Config: An instance of util.config.Config.

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

get(key: str, default: Any | None = None)[source]: Retrieve the value for key.

get_cache_path(*parts) → Path[source]

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

Shorthand for Config.get_cache_path().

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.

Shorthand for Config.get_local_path().

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

Return a Platform from Config.platform_info.

Shorthand for Config.get_platform().

get_scenario() → message_ix.Scenario[source]

Return a Scenario from scenario_info.

Shorthand for Config.get_scenario().

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

Handle command-line arguments.

Shorthand for Config.handle_cli_args().

property model: message_ix_models.model.config.Config: An instance of model.config.Config.

classmethod only() → Context[source]

Return the only Context instance.

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

property report: message_ix_models.report.config.Config: An instance of report.config.Config.

set(key: str, value: Any) → None[source]: Change the stored value for key.

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

Update Config.scenario_info to match an existing scenario.

Shorthand for Config.set_scenario().

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

write_debug_archive() → None[source]

Write an archive containing the files listed in Config.debug_paths.

Shorthand for Config.write_debug_archive().

`util.genno`

Utilities for working with genno.

Most code appearing here should be migrated upstream, to genno itself.

class message_ix_models.util.genno.Collector(target: KeyLike, key_cb: Callable[[KeyLike], KeyLike])[source]

Helper class to collect and merge data at a target key.

Example usage:

# Create a Collector instance
collector = Collector(target="FOO", key_cb="{}::foo".format)

# Associate it with a particular Computer
c = collector.computer = Computer()

# Add a task
collect("bar", func, "k1", "k2", arg1="baz", arg2="qux")

These statements have the following effects:

Add to c a task with the key “FOO” that calls merge_data() on 1 or more inputs.
Construct a key “bar::foo” using the key_cb.
Add a task at “bar::foo” that calls func(k1, k2, arg1="baz", arg2="qux").
Add “bar::foo” (denoting the output of (3)) to the keys merged by (1).

class message_ix_models.util.genno.Keys(**kwargs: Key | str)[source]

A collection of Key.

This is essentially the same as types.SimpleNamespace, except every attribute is a Key.

message_ix_models.util.genno.update_computer(a: Computer, b: Computer) → None[source]

Update a with keys and tasks from b.

For most keys, the task in b is copied to a at the same key. For the key “config”, the contents of the dict in a are updated with the values from the one in b. This overwrites or replaces existing configuration.

Todo

Migrate upstream to a method like genno.Computer.update.

Raises:

RuntimeError –

if any key already exists in a with a task different from the corresponding one in b. - if the key “context” maps to different Context instances in a and b.

`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.

This class allows for future-proof import statements. For example, if there is a module message_data.project.foo.bar, code can be written like:

from message_ix_models.project.foo import bar

The find_spec() method locates the corresponding submodule in message_data and imports it as if it were in message_ix_models. Later, if the module is migrated to message_ix_models.project.foo.bar, the import statement will work directly, without changes or use of MessageDataFinder.

Where the same module names exist within both packages, the submodule of message_ix_models will be found directly and MessageDataFinder will not be invoked.

This behaviour also allows to mix model and project code in the two packages, although this should be avoided where possible.

expr = re.compile('message_ix_models\\.(?P<name>(model|project)\\..*)'): Expression for supported module names.

class message_ix_models.util.importlib.MinimumVersionDecorator(*expr: str, raises: Iterable[type[Exception]] | None = None)[source]

Mark callable objects as requiring minimum version(s) of upstream packages.

If the decorated object is called and any of condition(s) in expr is not met, NotImplementedError is raised with an informative message.

The decorated object gains an attribute .minimum_version, which can be used like pytest.mark.xfail() to decorate test functions. This marks the test as XFAIL, raising NotImplementedError directly; indirectly RuntimeError or AssertionError (for instance, via click test utilities or genno), or any of the classes given by the raises argument.

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

Parameters:

expr –

Zero or more conditions like:

”pkgA 1.2.3.post0; pkgB 2025.2”, specifying minimum version of 1 or more packages.
”python 3.10”, specifying a minimum version of python.
”message_ix_models.foo.bar.baz”, recursively referring to the minimum version required by baz in the module message_ix_models.foo.bar. This object must also have been decorated with MinimumVersionDecorator.

check_versions() → str[source]

Evaluate all the :attr:`conditions.

Return "" if all pass, else a str describing failed conditions.

mark_test(obj)[source]: Apply a pytest XFAIL mark to test function or class obj.

raise_for_version() → None[source]: Raise NotImplementedError if check_versions() fails.

message_ix_models.util.importlib.minimum_version: Alias for MinimumVersionDecorator.

`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: LogRecord) → str[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() → None[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: LogRecord) → 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-Materials': {'pooch_args': {'base_url': 'https://github.com/iiasa/message-ix-models/raw/main/message_ix_models/data/material/', 'registry': {'material.tar.gz': 'sha1:b93f4390cb00749c2316fcdddf901a0eab896774'}}, 'processor': <message_ix_models.util.pooch.Extract object>}, '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>}, '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 exact field values occurring in the ISO 3166-1 database.

Other code may extend this mapping before calling iso_3166_alpha_3().

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: dict[str, list] = <factory>, par: 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 or its data, which can be a performance-limiting step.

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.

substitute_codes() → None[source]

Update the members of set using get_codelist().

Members of the following set(s) that are str are replaced with codes loaded from the corresponding code list(s), including all annotations:

commodity

Todo

Extend for other sets and cases where there are multiple code lists (for instance year, region).

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.

class message_ix_models.util.sdmx.AnnotationsMixIn[source]

Mix-in for dataclasses to allow (de)serializing as SDMX annotations.

classmethod from_obj(obj: AnnotableArtefact) → Self[source]: Return a new instance of cls given an AnnotableArtefact obj.

get_annotations(_rtype: type[list] | type[dict])[source]

Return a collection of Annotation for the fields of the object.

Returns:

list of Annotation – if _rtype is list.
dict – if _rtype is dict. The dict has the one key “annotations”, mapped to a list of Annotations. This can be used as a keyword argument to the constructor of a AnnotableArtefact subclass.

message_ix_models.util.sdmx.DATAFLOW: dict[str, Dataflow] = {'urn:sdmx:org.sdmx.infomodel.datastructure.DataflowDefinition=IIASA_ECE:DF_ACTIVITY(2025.3.11)': <Dataflow wrapping 'DataflowDefinition=IIASA_ECE:DF_ACTIVITY(2025.3.11)'>, 'urn:sdmx:org.sdmx.infomodel.datastructure.DataflowDefinition=IIASA_ECE:DF_ACTIVITY_PASSENGER(2025.3.11)': <Dataflow wrapping 'DataflowDefinition=IIASA_ECE:DF_ACTIVITY_PASSENGER(2025.3.11)'>, 'urn:sdmx:org.sdmx.infomodel.datastructure.DataflowDefinition=IIASA_ECE:DF_ACTIVITY_VEHICLE(2025.3.11)': <Dataflow wrapping 'DataflowDefinition=IIASA_ECE:DF_ACTIVITY_VEHICLE(2025.3.11)'>, 'urn:sdmx:org.sdmx.infomodel.datastructure.DataflowDefinition=IIASA_ECE:DF_AGE(2025.3.11)': <Dataflow wrapping 'DataflowDefinition=IIASA_ECE:DF_AGE(2025.3.11)'>, 'urn:sdmx:org.sdmx.infomodel.datastructure.DataflowDefinition=IIASA_ECE:DF_CAP_NEW(2025.3.11)': <Dataflow wrapping 'DataflowDefinition=IIASA_ECE:DF_CAP_NEW(2025.3.11)'>, 'urn:sdmx:org.sdmx.infomodel.datastructure.DataflowDefinition=IIASA_ECE:DF_CONSTRAINT_DYNAMIC(2025.3.11)': <Dataflow wrapping 'DataflowDefinition=IIASA_ECE:DF_CONSTRAINT_DYNAMIC(2025.3.11)'>, 'urn:sdmx:org.sdmx.infomodel.datastructure.DataflowDefinition=IIASA_ECE:DF_DEMAND_SCALE(2025.3.11)': <Dataflow wrapping 'DataflowDefinition=IIASA_ECE:DF_DEMAND_SCALE(2025.3.11)'>, 'urn:sdmx:org.sdmx.infomodel.datastructure.DataflowDefinition=IIASA_ECE:DF_DISUTILITY(2025.3.11)': <Dataflow wrapping 'DataflowDefinition=IIASA_ECE:DF_DISUTILITY(2025.3.11)'>, 'urn:sdmx:org.sdmx.infomodel.datastructure.DataflowDefinition=IIASA_ECE:DF_ELASTICITY_F(2025.3.11)': <Dataflow wrapping 'DataflowDefinition=IIASA_ECE:DF_ELASTICITY_F(2025.3.11)'>, 'urn:sdmx:org.sdmx.infomodel.datastructure.DataflowDefinition=IIASA_ECE:DF_ELASTICITY_P(2025.3.11)': <Dataflow wrapping 'DataflowDefinition=IIASA_ECE:DF_ELASTICITY_P(2025.3.11)'>, 'urn:sdmx:org.sdmx.infomodel.datastructure.DataflowDefinition=IIASA_ECE:DF_EMISSIONS_INTENSITY(2025.3.11)': <Dataflow wrapping 'DataflowDefinition=IIASA_ECE:DF_EMISSIONS_INTENSITY(2025.3.11)'>, 'urn:sdmx:org.sdmx.infomodel.datastructure.DataflowDefinition=IIASA_ECE:DF_ENERGY(2025.3.11)': <Dataflow wrapping 'DataflowDefinition=IIASA_ECE:DF_ENERGY(2025.3.11)'>, 'urn:sdmx:org.sdmx.infomodel.datastructure.DataflowDefinition=IIASA_ECE:DF_FE_TRANSPORT(2025.3.11)': <Dataflow wrapping 'DataflowDefinition=IIASA_ECE:DF_FE_TRANSPORT(2025.3.11)'>, 'urn:sdmx:org.sdmx.infomodel.datastructure.DataflowDefinition=IIASA_ECE:DF_FREIGHT_ACTIVITY(2025.3.11)': <Dataflow wrapping 'DataflowDefinition=IIASA_ECE:DF_FREIGHT_ACTIVITY(2025.3.11)'>, 'urn:sdmx:org.sdmx.infomodel.datastructure.DataflowDefinition=IIASA_ECE:DF_FREIGHT_MODE_SHARE(2025.3.11)': <Dataflow wrapping 'DataflowDefinition=IIASA_ECE:DF_FREIGHT_MODE_SHARE(2025.3.11)'>, 'urn:sdmx:org.sdmx.infomodel.datastructure.DataflowDefinition=IIASA_ECE:DF_FUEL_ECONOMY(2025.3.11)': <Dataflow wrapping 'DataflowDefinition=IIASA_ECE:DF_FUEL_ECONOMY(2025.3.11)'>, 'urn:sdmx:org.sdmx.infomodel.datastructure.DataflowDefinition=IIASA_ECE:DF_FUEL_EMI_INTENSITY(2025.3.11)': <Dataflow wrapping 'DataflowDefinition=IIASA_ECE:DF_FUEL_EMI_INTENSITY(2025.3.11)'>, 'urn:sdmx:org.sdmx.infomodel.datastructure.DataflowDefinition=IIASA_ECE:DF_GDP_IN(2025.3.11)': <Dataflow wrapping 'DataflowDefinition=IIASA_ECE:DF_GDP_IN(2025.3.11)'>, 'urn:sdmx:org.sdmx.infomodel.datastructure.DataflowDefinition=IIASA_ECE:DF_IKARUS_AVAILABILITY(2025.3.11)': <Dataflow wrapping 'DataflowDefinition=IIASA_ECE:DF_IKARUS_AVAILABILITY(2025.3.11)'>, 'urn:sdmx:org.sdmx.infomodel.datastructure.DataflowDefinition=IIASA_ECE:DF_IKARUS_FIX_COST(2025.3.11)': <Dataflow wrapping 'DataflowDefinition=IIASA_ECE:DF_IKARUS_FIX_COST(2025.3.11)'>, 'urn:sdmx:org.sdmx.infomodel.datastructure.DataflowDefinition=IIASA_ECE:DF_IKARUS_INPUT(2025.3.11)': <Dataflow wrapping 'DataflowDefinition=IIASA_ECE:DF_IKARUS_INPUT(2025.3.11)'>, 'urn:sdmx:org.sdmx.infomodel.datastructure.DataflowDefinition=IIASA_ECE:DF_IKARUS_INV_COST(2025.3.11)': <Dataflow wrapping 'DataflowDefinition=IIASA_ECE:DF_IKARUS_INV_COST(2025.3.11)'>, 'urn:sdmx:org.sdmx.infomodel.datastructure.DataflowDefinition=IIASA_ECE:DF_IKARUS_TECHNICAL_LIFETIME(2025.3.11)': <Dataflow wrapping 'DataflowDefinition=IIASA_ECE:DF_IKARUS_TECHNICAL_LIFETIME(2025.3.11)'>, 'urn:sdmx:org.sdmx.infomodel.datastructure.DataflowDefinition=IIASA_ECE:DF_IKARUS_VAR_COST(2025.3.11)': <Dataflow wrapping 'DataflowDefinition=IIASA_ECE:DF_IKARUS_VAR_COST(2025.3.11)'>, 'urn:sdmx:org.sdmx.infomodel.datastructure.DataflowDefinition=IIASA_ECE:DF_INPUT(2025.3.11)': <Dataflow wrapping 'DataflowDefinition=IIASA_ECE:DF_INPUT(2025.3.11)'>, 'urn:sdmx:org.sdmx.infomodel.datastructure.DataflowDefinition=IIASA_ECE:DF_INPUT_SHARE(2025.3.11)': <Dataflow wrapping 'DataflowDefinition=IIASA_ECE:DF_INPUT_SHARE(2025.3.11)'>, 'urn:sdmx:org.sdmx.infomodel.datastructure.DataflowDefinition=IIASA_ECE:DF_LDV_ACTIVITY(2025.3.11)': <Dataflow wrapping 'DataflowDefinition=IIASA_ECE:DF_LDV_ACTIVITY(2025.3.11)'>, 'urn:sdmx:org.sdmx.infomodel.datastructure.DataflowDefinition=IIASA_ECE:DF_LDV_CLASS(2025.3.11)': <Dataflow wrapping 'DataflowDefinition=IIASA_ECE:DF_LDV_CLASS(2025.3.11)'>, 'urn:sdmx:org.sdmx.infomodel.datastructure.DataflowDefinition=IIASA_ECE:DF_LDV_INPUT_ADJ(2025.3.11)': <Dataflow wrapping 'DataflowDefinition=IIASA_ECE:DF_LDV_INPUT_ADJ(2025.3.11)'>, 'urn:sdmx:org.sdmx.infomodel.datastructure.DataflowDefinition=IIASA_ECE:DF_LIFETIME(2025.3.11)': <Dataflow wrapping 'DataflowDefinition=IIASA_ECE:DF_LIFETIME(2025.3.11)'>, 'urn:sdmx:org.sdmx.infomodel.datastructure.DataflowDefinition=IIASA_ECE:DF_LOAD_FACTOR_LDV(2025.3.11)': <Dataflow wrapping 'DataflowDefinition=IIASA_ECE:DF_LOAD_FACTOR_LDV(2025.3.11)'>, 'urn:sdmx:org.sdmx.infomodel.datastructure.DataflowDefinition=IIASA_ECE:DF_LOAD_FACTOR_NONLDV(2025.3.11)': <Dataflow wrapping 'DataflowDefinition=IIASA_ECE:DF_LOAD_FACTOR_NONLDV(2025.3.11)'>, 'urn:sdmx:org.sdmx.infomodel.datastructure.DataflowDefinition=IIASA_ECE:DF_MA3T_ATTITUDE(2025.3.11)': <Dataflow wrapping 'DataflowDefinition=IIASA_ECE:DF_MA3T_ATTITUDE(2025.3.11)'>, 'urn:sdmx:org.sdmx.infomodel.datastructure.DataflowDefinition=IIASA_ECE:DF_MA3T_DRIVER(2025.3.11)': <Dataflow wrapping 'DataflowDefinition=IIASA_ECE:DF_MA3T_DRIVER(2025.3.11)'>, 'urn:sdmx:org.sdmx.infomodel.datastructure.DataflowDefinition=IIASA_ECE:DF_MA3T_POPULATION(2025.3.11)': <Dataflow wrapping 'DataflowDefinition=IIASA_ECE:DF_MA3T_POPULATION(2025.3.11)'>, 'urn:sdmx:org.sdmx.infomodel.datastructure.DataflowDefinition=IIASA_ECE:DF_MER_TO_PPP(2025.3.11)': <Dataflow wrapping 'DataflowDefinition=IIASA_ECE:DF_MER_TO_PPP(2025.3.11)'>, 'urn:sdmx:org.sdmx.infomodel.datastructure.DataflowDefinition=IIASA_ECE:DF_PDT(2025.3.11)': <Dataflow wrapping 'DataflowDefinition=IIASA_ECE:DF_PDT(2025.3.11)'>, 'urn:sdmx:org.sdmx.infomodel.datastructure.DataflowDefinition=IIASA_ECE:DF_POPULATION_IN(2025.3.11)': <Dataflow wrapping 'DataflowDefinition=IIASA_ECE:DF_POPULATION_IN(2025.3.11)'>, 'urn:sdmx:org.sdmx.infomodel.datastructure.DataflowDefinition=IIASA_ECE:DF_POPULATION_SUBURB_SHARE(2025.3.11)': <Dataflow wrapping 'DataflowDefinition=IIASA_ECE:DF_POPULATION_SUBURB_SHARE(2025.3.11)'>, 'urn:sdmx:org.sdmx.infomodel.datastructure.DataflowDefinition=IIASA_ECE:DF_P_ACTIVITY(2025.3.11)': <Dataflow wrapping 'DataflowDefinition=IIASA_ECE:DF_P_ACTIVITY(2025.3.11)'>, 'urn:sdmx:org.sdmx.infomodel.datastructure.DataflowDefinition=IIASA_ECE:DF_SPEED(2025.3.11)': <Dataflow wrapping 'DataflowDefinition=IIASA_ECE:DF_SPEED(2025.3.11)'>, 'urn:sdmx:org.sdmx.infomodel.datastructure.DataflowDefinition=IIASA_ECE:DF_TECH_SHARE(2025.3.11)': <Dataflow wrapping 'DataflowDefinition=IIASA_ECE:DF_TECH_SHARE(2025.3.11)'>}: Collection of Dataflow instances.

class message_ix_models.util.sdmx.Dataflow(*, module: str, id: str | None = None, name: str, units: str, description: str | None = None, key: KeyLike | None = None, dims: tuple[str, ...] | None = None, i_o: ~message_ix_models.util.sdmx.Dataflow.FLAG = <FLAG.IN: 1>, path: str | tuple[str, ...] | None = None, required: bool = True, replace: bool = False, cs_urn: list[str] = [], **ma_kwargs)[source]

Information about an input or output data flow.

If an input data flow, the data is expected in a file at path.

Todo

Accept an argument that sets df directly and ignores other arguments.
Annotate certain dimensions as optional; expand add_tasks() to automatically handle insertion of these dimensions.
Merge with ExoDataSource.

Parameters:

module (str) – Should be the name of the code module responsible for creating the data flow, for instance using __name__.
id (str) – Partial ID of both the DFD and a related DataStructureDefinition. The strings "DF_" and "DS_" are automatically prepended, and the characters “ ” and “-” replaced with underscores.
name (str) – Human-readable 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 Computer.
dims (tuple of str, optional) – IDs of dimensions of the DSD. These may be short dimension IDs as used in message_ix.report, for instance "t" for the ‘technology’ dimension.
path (str or tuple of str, optional) – Path at which an input data file is located. If not supplied, 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 True (the default), the input data file must be present for the build to succeed.
replace (bool, *optional*) – If True, replace any existing entry in DATAFLOW that with an equivalent URN to the one implied by kwargs. Otherwise (default), raise an exception.

class FLAG(*values)[source]

Flags for intent.

IN = 1: Input data flow.

OUT = 2: Output data flow.

add_tasks(c: Computer, *args, context: Context) → tuple[KeyLike, ...][source]: Prepare c to read data from a file at path.

df: BaseDataflow

sdmx.Dataflow describing the data flow.

This instance will always have annotations with the following IDs:

“file-path”: see path.
“genno-key”: see key.
“intent”: see intent.
“preferred-units”: see units.
“required-for-build”: see required.

generate_csv_template() → Path[source]

Generate a CSV template file.

Currently not implemented.

property intent: FLAG: Indicates whether the dataflow is for input or output.

property key: Key

genno.Key, including preferred dimensions.

When intent is FLAG.IN, this is the key at which the loaded data is available. When intent is FLAG.OUT, this is the key from which the output data will be prepared.

property module: str: Module with which the data flow is associated.

property path: Path: Path fragment for the location of a file containing input data.

property required: bool

True if the input file :path:`.path` must be present to build a model.

This means that a corresponding function, for instance transport.build.main(), expects the file to be present.

property units: pint.Unit: Preferred units for the data.

message_ix_models.util.sdmx.STORE = <sdmx.StructureMessage> <Header> source: test: False ConceptScheme (2): CS_MESSAGE_TRANSPORT CS_MESSAGE_IX_MODELS DataflowDefinition (42): DF_ACTIVITY DF_FREIGHT_ACTIVITY DF_LDV_ACTIV... DataStructureDefinition (42): DS_ACTIVITY DS_FREIGHT_ACTIVITY DS_LDV_...: Common store of SDMX structural artefacts.

class message_ix_models.util.sdmx.URNLookupEnum(new_class_name, /, names, *, module=None, qualname=None, type=None, start=1, boundary=None)[source]

Enum subclass that allows looking up members using a URN.

classmethod by_urn(urn: str)[source]: Return the Enum member given its urn.

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.collect_structures(target: StructureMessage, dfd: BaseDataflow) → None[source]

Update target with dfd and related structures.

These include:

The DataStructureDefinition (DSD) that structures dfd.
For each component (dimension, attribute, or measure) in the DSD:
- Any ConceptScheme that provides the concept role.
- Any Codelist that enumerates the component.

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.get(urn: str) → MaintainableArtefact | None[source]: Return an object given its URN.

message_ix_models.util.sdmx.get_cl(name: str, context: Context | None = None) → Codelist[source]: Return a code list.

message_ix_models.util.sdmx.get_concept(string: str, *, cs_urn: tuple[str, ...] = ()) → Concept[source]

Retrieve (or create) a single Concept from a concept scheme.

Items are sought first in “CS_MESSAGE_IX_MODELS” (see get_cs()) and then in the concept scheme(s) indicated by cs_urn, if any. string is matched against both item IDS and the contents of an “aliases” annotation (if any).

If (and only if) cs_urn is given and the concept is not found, it is created in the last of cs_urn. In other words, CS_MESSAGE_IX_MODELS is never modified.

Raises:: ValueError – if no concept with an ID or alias matching string is found, and cs_urn is not given.

message_ix_models.util.sdmx.get_cs() → ConceptScheme[source]

Return a scheme of common concepts for the MESSAGEix-GLOBIOM model family.

The full artefact contains its own detailed description.

message_ix_models.util.sdmx.get_version(with_dev: bool | None = True) → str[source]

Return a sdmx.model.common.Version for message_ix_models.

Todo

Remove str(...) once sdmx1 > 2.21.1 can handle Version.

Parameters:: with_dev (bool) – If True, include the dev version part, e.g. “2025.3.12.dev123”. If not, exclude.

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

message_ix_models.util.sdmx.read(urn: str, base_dir: TypeAliasForwardRef(':class:`os.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: TypeAliasForwardRef(':class:`os.PathLike`') | None = None, basename: str | None = None)[source]: Store an SDMX object as package data.

`util.slurm`

Utilities for the SLURM Workload Manager.

To use, transform a desired, ordinary invocation of the mix-models CLI:

Listing 4 Command 1

mix-models --opt_a=0 -b 2 command --opt_c=2 subcommand --opt_d=3 arg0 arg1

…into something like:

Listing 5 Command 2

mix-models sbatch --go \
  --username=example_user \
  --venv=/home/example_user/venv/py3.13_demo \
  -- \
  --opt_a=0 -b 2 command --opt_c=2 subcommand --opt_d=3 arg0 arg1

In particular:

The inserted -- separates the command sbatch from the options and arguments to be used to invoke mix-models on the worker node. This command will result in exactly Command 1 being invoked at the end of the script TEMPLATE.
The options --username and --venv are also passed into the template. As the name implies, they are optional. The values are read from the $USER and $VIRTUAL_ENV environment variables, respectively, wherever Command 2 is invoked.
Without the option --go, the batch script is only printed out. Add this option to actually call sbatch.

`types`

Types for hinting.

class message_ix_models.types.MaintainableArtefactArgs[source]: Some keyword arguments to sdmx.model.common.MaintainableArtefact.

message_ix_models.types.MutableParameterData

Mutable collection of message_ix or ixmp parameter data.

alias of MutableMapping[str, DataFrame]

message_ix_models.types.ParameterData

Collection of message_ix or ixmp parameter data. Keys should be parameter names (str), and values should be data frames with the same structure returned by make_df(). See also add_par_data().

alias of Mapping[str, DataFrame]

Low-level utilities (util)

Low-level utilities (`util`)