"""Utilities for handling objects from :mod:`sdmx`."""
import logging
import re
from collections.abc import Iterable, Mapping
from dataclasses import dataclass, fields
from datetime import datetime
from enum import Enum, Flag, auto
# TODO Remove when Python 3.10 is no longer supported
from enum import EnumMeta as EnumType
from functools import cache
from importlib.metadata import version
from pathlib import Path
from typing import TYPE_CHECKING, Generic, Optional, TypeVar, Union, cast
from warnings import warn
import sdmx
import sdmx.message
import sdmx.urn
from genno import Key
from iam_units import registry
from ixmp.report.common import RENAME_DIMS
from sdmx.model import common, v21
from .common import package_data_path
from .context import Context
if TYPE_CHECKING:
from os import PathLike
import pint
from genno import Computer, Key
from sdmx.message import StructureMessage
from message_ix_models.types import KeyLike
# TODO Use "from typing import Self" once Python 3.11 is the minimum supported
Self = TypeVar("Self", bound="AnnotationsMixIn")
log = logging.getLogger(__name__)
CodeLike = Union[str, common.Code]
#: Collection of :class:`.Dataflow` instances.
DATAFLOW: dict[str, "Dataflow"] = {}
#: Common store of SDMX structural artefacts.
STORE = sdmx.message.StructureMessage()
[docs]
@dataclass
class AnnotationsMixIn:
"""Mix-in for dataclasses to allow (de)serializing as SDMX annotations."""
# TODO Type with overrides: list → list
[docs]
def get_annotations(self, _rtype: Union[type[list], type[dict]]):
"""Return a collection of :class:`.Annotation` for the fields of the object.
Returns
-------
list of :class:`Annotation <sdmx.model.common.BaseAnnotation>`
if `_rtype` is :class:`list`.
dict
if `_rtype` is :class:`dict`. The dict has the one key "annotations", mapped
to a :class:`list` of Annotations. This can be used as a keyword argument
to the constructor of a :class:`.AnnotableArtefact` subclass.
"""
result = []
for f in fields(self):
anno_id = f.name.replace("_", "-")
result.append(v21.Annotation(id=anno_id, text=repr(getattr(self, f.name))))
if _rtype is list:
return result
else:
return dict(annotations=result)
[docs]
@classmethod
def from_obj(
cls: type["Self"], obj: common.AnnotableArtefact, globals: Optional[dict] = None
) -> "Self":
"""Return a new instance of `cls` given an AnnotableArtefact `obj`."""
args = []
for f in fields(cls):
anno_id = f.name.replace("_", "-")
args.append(obj.eval_annotation(id=anno_id, globals=globals))
return cls(*args)
[docs]
class Dataflow:
"""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 :attr:`df` directly and ignores other arguments.
- Annotate certain dimensions as optional; expand :meth:`.add_tasks` to
automatically handle insertion of these dimensions.
- Merge with :class:`.ExoDataSource`.
Parameters
----------
module : str
Should be the name of the code module responsible for creating the data flow,
for instance using :py:`__name__`.
id : str
Partial ID of both the DFD and a related DataStructureDefinition. The strings
:py:`"DF_"` and :py:`"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 :class:`.Computer`.
dims : tuple of str, optional
IDs of dimensions of the DSD. These may be short dimension IDs as used in
:mod:`message_ix.report`, for instance :py:`"t"` for the 'technology' dimension.
path : str or tuple of str, optional
Path at which an input data file is located. If not supplied, :attr:`path` is
constructed from `key`, `dims`, and the tag "exo".
description : str, optional
Human-readable description of the data flow, including any notes about required
properties or contents or methods used to handle the data.
required : bool, optional
If :any:`True` (the default), the input data file **must** be present for the
build to succeed.
replace : bool, *optional*
If :any:`True`, replace any existing entry in :data:`DATAFLOW` that with an
equivalent URN to the one implied by `kwargs`. Otherwise (default), raise an
exception.
"""
[docs]
class FLAG(Flag):
"""Flags for :attr:`.intent`."""
#: Input data flow.
IN = 1
#: Output data flow.
OUT = 2
#: :class:`sdmx.Dataflow <sdmx.model.common.BaseDataflowDefinition>` describing the
#: data flow.
#:
#: This instance will always have annotations with the following IDs:
#:
#: - "file-path": see :attr:`.path`.
#: - "genno-key": see :attr:`.key`.
#: - "intent": see :attr:`.intent`.
#: - "preferred-units": see :attr:`.units`.
#: - "required-for-build": see :attr:`.required`.
df: "sdmx.model.common.BaseDataflow"
def __init__(
self,
*,
module: str,
id: Optional[str] = None,
name: str,
units: str,
description: Optional[str] = None,
key: Optional["KeyLike"] = None,
dims: Optional[tuple[str, ...]] = None,
i_o: FLAG = FLAG.IN,
path: Union[str, tuple[str, ...], None] = None,
required: bool = True,
# Used only for creation
replace: bool = False,
cs_urn: list[str] = [],
# Others
**ma_kwargs,
):
import pint
# Ensure CS_MESSAGE_IX_MODELS is available
get_cs()
# Collection of annotations for the data flow
anno = [
v21.Annotation(id="intent", text=str(i_o.name)),
v21.Annotation(id="module", text=module),
v21.Annotation(
id="required-for-build", text=repr((i_o & self.FLAG.IN) and required)
),
]
# Handle `path` argument
if isinstance(path, str):
_path = Path(path) # Convert str to Path
elif path:
_path = Path(*path) # Something else, not None
# Parse and store units
ureg = pint.get_application_registry()
try:
units = ureg.Unit(units)
except Exception as e:
log.info(f"Replace units {units!r} with 'dimensionless' due to {e}")
units = ureg.dimensionless
anno.append(v21.Annotation(id="preferred-units", text=f"{units}"))
if not key:
# Determine from file path
key = Key(" ".join(_path.parts).replace("-", " "), dims or (), "exo")
else:
# Convert to Key object
key = Key(key)
if path is None:
_path = Path(key.name.replace(" ", "-"))
anno.append(v21.Annotation(id="genno-key", text=str(key)))
_path = _path.with_suffix(".csv")
anno.append(v21.Annotation(id="file-path", text=str(_path)))
# Default properties for maintainable artefacts
ma_kwargs.setdefault("maintainer", read("IIASA_ECE:AGENCIES")["IIASA_ECE"])
ma_kwargs.setdefault("is_external_reference", False)
ma_kwargs.setdefault("is_final", False)
ma_kwargs.setdefault("version", get_version())
# IDs for the data flow and data structure
name_for_id = (id or key.name).upper().replace(" ", "_").replace("-", "_")
df_id = f"DF_{name_for_id}"
ds_id = f"DS_{name_for_id}"
# Create a data structure definition
ma_kwargs["name"] = f"Structure of {df_id}"
dsd = v21.DataStructureDefinition(id=ds_id, **ma_kwargs)
dsd.urn = sdmx.urn.make(dsd)
# Add dimensions to `DSD` according to `key.dims`
for order, dim_id in enumerate(key.dims):
# Retrieve the dimension concept and its full ID
concept = get_concept(dim_id, cs_urn=tuple(cs_urn))
# Create a code list for this dimension
cl = get_cl(concept.id)
# Create the dimension
dsd.dimensions.getdefault(
id=dim_id,
concept_identity=concept,
local_representation=common.Representation(enumerated=cl),
order=order,
)
# Add attributes
nsr = v21.NoSpecifiedRelationship()
for attr_id in "MODEL", "SCENARIO", "VERSION", "UNIT_MEASURE":
# Retrieve the attribute concept and its full ID
concept = get_concept(attr_id)
dsd.attributes.getdefault(
id=attr_id, concept_identity=concept, related_to=nsr
)
# Add a measure
dsd.measures.getdefault(id="value")
# Create and store a data flow definition
ma_kwargs["name"] = name
ma_kwargs["description"] = (
f"{description.strip()}\n\n" if description is not None else ""
)
ma_kwargs["annotations"] = anno
self.df = v21.DataflowDefinition(id=df_id, **ma_kwargs, structure=dsd)
self.df.urn = sdmx.urn.make(self.df)
# Update the instance with a docstring
self._update_doc()
# Add or replace an entry in DATAFLOW
if duplicate := list(
filter(
lambda x: x.key == self.key and x.intent == self.intent,
DATAFLOW.values(),
)
):
existing = duplicate[0]
if replace:
log.info(f"Replace existing entry for {existing.df.urn!r}")
DATAFLOW[self.df.urn] = self
else:
raise RuntimeError(
f"Definition of {self} duplicates existing {existing}"
)
else:
# Add to the collection of DATAFLOW and the STORE
DATAFLOW[self.df.urn] = self
STORE.add(self.df)
STORE.add(dsd)
# Does nothing except ensure callable(…) == True for inspection by genno
def __call__(self): ...
def __repr__(self) -> str:
assert self.df.urn is not None
return f"<Dataflow wrapping {sdmx.urn.shorten(self.df.urn)!r}>"
def _update_doc(self) -> None:
"""Update the instance with a docstring."""
lines = [
f"{self.df.name} [{self.units}]",
"",
f"- Key and dimensions: **<{self.key}>**",
]
if self.intent & self.FLAG.IN:
# Lines only relevant for input quantities
lines.append(f"- Input file at path: :file:`{self.path}`")
lines.append(f"- Required for build: {self.required}")
else:
lines.append(
f"- Output data from :mod:`{self.df.eval_annotation('module')}`"
)
lines.append(f"\n{self.df.description}")
self.__doc__ = "\n".join(lines)
# Access to annotations of DFD
@property
def intent(self) -> FLAG:
"""Indicates whether the dataflow is for input or output."""
return self.FLAG[str(self.df.get_annotation(id="intent").text)]
@property
def key(self) -> "Key":
""":class:`genno.Key`, including preferred dimensions.
When :attr:`intent` is :data:`FLAG.IN`, this is the key at which the loaded data
is available. When :attr:`intent` is :data:`FLAG.OUT`, this is the key from
which the output data will be prepared.
"""
return Key(str(self.df.get_annotation(id="genno-key").text))
@property
def module(self) -> str:
"""Module with which the data flow is associated."""
return str(self.df.get_annotation(id="module").text)
@property
def path(self) -> Path:
"""Path fragment for the location of a file containing input data."""
return Path(str(self.df.get_annotation(id="file-path").text))
@property
def required(self) -> bool:
""":any:`True` if the input file :path:`.path` must be present to build a model.
This means that a corresponding function, for instance
:func:`.transport.build.main`, expects the file to be present.
"""
return self.df.eval_annotation(id="required-for-build")
@property
def units(self) -> "pint.Unit":
"""Preferred units for the data."""
import pint
return pint.get_application_registry().Unit(
str(self.df.get_annotation(id="preferred-units").text)
)
# For interaction with genno
[docs]
def add_tasks(
self, c: "Computer", *args, context: "Context"
) -> tuple["KeyLike", ...]:
"""Prepare `c` to read data from a file at :attr:`.path`."""
# TODO Use a package-wide utility or a callback
from message_ix_models.model.transport.util import region_path_fallback
# Identify the path
try:
path = region_path_fallback(context, self.path)
except FileNotFoundError:
if self.required:
raise
else:
return ()
# Use standard RENAME_DIMS from ixmp config
dims = RENAME_DIMS.copy()
values = set(dims.values())
dims.update({d: d for d in self.key.dims if d not in values})
c.add("load_file", path, key=self.key, dims=dims, name=self.key.name)
return (self.key,)
[docs]
def generate_csv_template(self) -> Path:
"""Generate a CSV template file.
Currently not implemented.
"""
raise NotImplementedError
# 1. In the current format.abs
# 2. In SDMX-CSV.
# dm = DataMessage()
# dm.data.append(DataSet(structure))
# template =
T = TypeVar("T", bound=Enum)
# TODO Replace with URNLookupMixin[T] once Python 3.10 is no longer supported
[docs]
class URNLookupMixin(Generic[T]):
""":class:`.Enum` mix-in class for looking up members by URN/retrieving URNs."""
name: str
_member_map_: dict[str, T]
_urn_name: dict[str, str]
[docs]
@classmethod
def by_urn(cls, urn: str) -> T:
"""Return the :class:`.Enum` member given its `urn`."""
return cls._member_map_[cls._urn_name[urn]]
@property
def urn(self) -> str:
"""Return the URN for an Enum member."""
for result, name in self._urn_name.items():
if name == self.name:
break
return result
[docs]
class URNLookupEnum(URNLookupMixin, Enum):
"""Class constructed by ItemSchemeEnumType."""
[docs]
class ItemSchemeEnumType(EnumType):
"""Metaclass for :class:`.Enum` tied to an SDMX :class:`.ItemScheme`.
A class constructed using this metaclass **must** have a method
:py:`_get_item_scheme()` that returns a :class:`sdmx.model.common.ItemScheme`. The
items in the item scheme become the members of the enumeration.
Example
-------
>>> from message_ix_models.util.sdmx import read
>>> class EXAMPLE(URNLookupEnum, metaclass=ItemSchemeEnumType):
...
... def _get_item_scheme(self):
... return read("AGENCY:CODELIST_ID(1.2.3)")
…creates a new subclass of :class:`.Enum` that has the methods and properties of
:class:`.URNLookupMixin`.
"""
@classmethod
def __prepare__(metacls, cls, bases, **kwgs):
return {}
def __init__(cls, *args, **kwds):
super(ItemSchemeEnumType, cls).__init__(*args)
def __new__(metacls, cls, bases, dct, **kwargs) -> type["URNLookupEnum"]:
# Retrieve the item scheme
scheme = dct.pop("_get_item_scheme")(None)
if not isinstance(scheme, common.ItemScheme):
raise RuntimeError(
f"Callback for {cls} returned {scheme}; expected ItemScheme"
)
# Prepare the EnumDict for creating the class
enum_dct = super(ItemSchemeEnumType, metacls).__prepare__(cls, bases, **kwargs)
# Transfer class dct private members
enum_dct.update(dct)
# Populate the class member dictionary and URN → member name mapping
_urn_name = dict()
if any(issubclass(c, Flag) for c in bases):
# Ensure the 0 member is NONE, not any of the codes
enum_dct["NONE"] = 0
for i, item in enumerate(scheme, start=1):
_urn_name[item.urn] = item.id
enum_dct[item.id] = auto()
# Create the class
enum_class = cast(
type["URNLookupEnum"],
super(ItemSchemeEnumType, metacls).__new__(
metacls, cls, bases, enum_dct, **kwargs
),
)
# Store the _urn_name mapping
setattr(enum_class, "_urn_name", _urn_name)
return enum_class
# FIXME Reduce complexity from 13 → ≤11
[docs]
def as_codes( # noqa: C901
data: Union[list[str], dict[str, CodeLike]],
) -> list[common.Code]:
"""Convert `data` to a :class:`list` of :class:`.Code` objects.
Various inputs are accepted:
- :class:`list` of :class:`str`.
- :class:`dict`, in which keys are :attr:`~sdmx.model.common.Code.id` and values are
further :class:`dict` with keys matching other Code attributes.
"""
# Assemble results as a dictionary
result: dict[str, common.Code] = {}
if isinstance(data, list):
# FIXME typing ignored temporarily for PR#9
data = dict(zip(data, data)) # type: ignore [arg-type]
elif not isinstance(data, Mapping):
raise TypeError(data)
for id, info in data.items():
# Pass through Code; convert other types to dict()
if isinstance(info, common.Code):
result[info.id] = info
continue
elif isinstance(info, str):
_info = dict(name=info)
elif isinstance(info, Mapping):
_info = dict(info)
else:
raise TypeError(info)
# Create a Code object
code = common.Code(
id=str(id),
name=_info.pop("name", str(id).title()),
)
# Store the description, if any
try:
code.description = common.InternationalString(
value=_info.pop("description")
)
except KeyError:
pass
# Associate with a parent
try:
parent_id = _info.pop("parent")
except KeyError:
pass # No parent
else:
result[parent_id].append_child(code)
# Associate with any children
for id in _info.pop("child", []):
try:
code.append_child(result[id])
except KeyError:
pass # Not parsed yet
# Convert other dictionary (key, value) pairs to annotations
for id, value in _info.items():
code.annotations.append(
v21.Annotation(
id=id, text=value if isinstance(value, str) else repr(value)
)
)
result[code.id] = code
return list(result.values())
[docs]
def collect_structures(
target: "sdmx.message.StructureMessage", dfd: "sdmx.model.common.BaseDataflow"
) -> None:
"""Update `target` with `dfd` and related structures.
These include:
- The :class:`.DataStructureDefinition` (DSD) that structures `dfd`.
- For each component (dimension, attribute, or measure) in the DSD:
- Any :class:`~sdmx.model.common.ConceptScheme` that provides the concept role.
- Any :class:`~sdmx.model.common.Codelist` that enumerates the component.
"""
target.add(dfd)
target.add(dfd.structure)
for cl in (
dfd.structure.dimensions,
dfd.structure.attributes,
getattr(dfd.structure, "measures", v21.MeasureDescriptor()),
):
for component in cl.components:
if component.concept_identity is not None:
target.add(component.concept_identity.parent)
if component.local_representation is not None:
target.add(component.local_representation.enumerated)
[docs]
def eval_anno(obj: common.AnnotableArtefact, id: str):
"""Retrieve the annotation `id` from `obj`, run :func:`eval` on its contents.
.. deprecated:: 2023.9.12
Use :meth:`sdmx.model.common.AnnotableArtefact.eval_annotation`, which provides
the same functionality.
"""
warn(
"message_ix_models.util.eval_anno; use sdmx.model.common.AnnotableArtefact"
".eval_annotation() instead.",
DeprecationWarning,
stacklevel=2,
)
try:
value = str(obj.get_annotation(id=id).text)
except KeyError: # No such attribute
return None
try:
return eval(value, {"registry": registry})
except Exception as e: # Something that can't be eval()'d, e.g. a plain string
log.debug(f"Could not eval({value!r}): {e}")
return value
[docs]
def get(urn: str) -> Optional["common.MaintainableArtefact"]:
"""Return an object given its URN."""
full_urn = sdmx.urn.expand(urn)
return STORE.get(sdmx.urn.URN(full_urn).id)
[docs]
def get_cl(name: str, context: Optional["Context"] = None) -> "common.Codelist":
"""Return a code list."""
from message_ix_models.model.structure import get_codes
id_ = None
if name == "NODE" and context:
name, id_ = f"node/{context.model.regions}", f"NODE_{context.model.regions}"
elif name == "YEAR" and context:
name, id_ = f"year/{context.model.years}", f"YEAR_{context.model.years}"
name = name or name.lower()
id_ = id_ or name.upper()
as_ = read("IIASA_ECE:AGENCIES")
cl: "common.Codelist" = common.Codelist(
id=f"CL_{id_}",
name=f"Codes for message-ix-models concept {name!r}",
maintainer=as_["IIASA_ECE"],
version=get_version(with_dev=False),
is_external_reference=False,
is_final=True,
)
cl.urn = sdmx.urn.make(cl)
try:
cl.extend(get_codes(name.lower()))
except FileNotFoundError:
pass
return cl
[docs]
@cache
def get_cs() -> "common.ConceptScheme":
"""Return a scheme of common concepts for the MESSAGEix-GLOBIOM model family.
The full artefact contains its own detailed description.
"""
cs = common.ConceptScheme(
id="CS_MESSAGE_IX_MODELS",
name="Concepts for message-ix-models",
description="""These include:
1. Concepts used as dimensions in MESSAGE parameter data (see also :mod:`.structure`).
2. Concepts particular to variants of MESSAGEix-GLOBIOM, such as
:mod:`.model.transport`.
Each concept in the concept scheme has:
- An upper case :py:`.id`, for instance :py:`"TECHNOLOGY"`.
- An annotation with :py:`id="aliases"` which is the :func:`repr` of a :class:`set`
giving alternate labels understood to be equivalent. These include
:data:`ixmp.report.RENAME_DIMS`, for example :py:`"t"` for 'technology'.""",
maintainer=common.Agency(id="IIASA_ECE"),
version="1.0.0",
is_final=True,
is_external_reference=False,
)
# Add concepts for MESSAGE sets/dimensions
for k, v in RENAME_DIMS.items():
# Retrieve or create the Concept for the set (e.g. "year" for k="year_act")
set_name = k.split("_")[0]
concept = cs.setdefault(
id=set_name.upper(),
name=f"{set_name!r} MESSAGEix set",
annotations=[v21.Annotation(id="aliases", text=repr(set()))],
)
# Add `v` to the aliases annotation
anno = concept.get_annotation(id="aliases")
anno.text = repr(eval(str(anno.text)) | {v})
for c_id in "MODEL", "SCENARIO", "VERSION":
cs.setdefault(
id=c_id,
name=f"{c_id.lower()!r} ixmp scenario identifier",
description="""In the ixmp data model, scenario objects are identified by
unique keys including (model name, scenario name, version).""",
)
cs.setdefault(
id="UNIT_MEASURE",
name="Unit of measure",
description="Unit in which data values are expressed",
annotations=[
v21.Annotation(
id="same-as-urn",
text="urn:sdmx:org.sdmx.infomodel.conceptscheme.Concept=SDMX:CROSS_DOMAIN_CONCEPTS(2.0).UNIT_MEASURE",
),
],
)
cs.setdefault(
id="URL",
name="ixmp scenario URL",
description="""URL combining the platform name (~database), model name, scenario
name, and version of an ixmp Scenario. See
https://docs.messageix.org/projects/ixmp/en/stable/api.html#ixmp.TimeSeries.url""",
)
STORE.add(cs)
return cs
[docs]
@cache
def get_concept(string: str, *, cs_urn: tuple[str, ...] = tuple()) -> "common.Concept":
"""Retrieve (or create) a single Concept from a concept scheme.
Items are sought first in "CS_MESSAGE_IX_MODELS" (see :func:`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.
"""
# TODO Would prefer to use StructureMessage.get() here, but does not handle URNs.
# Adjust once supported upstream
cs_all = list(
filter(
None,
cast(
Iterable[Optional["common.ItemScheme"]],
[get("ConceptScheme=IIASA_ECE:CS_MESSAGE_IX_MODELS")]
+ [get(u) for u in cs_urn],
),
)
)
for cs in cs_all:
for concept in cs.items.values():
labels = [concept.id] + list(concept.eval_annotation(id="aliases") or [])
if re.fullmatch("|".join(labels), string, flags=re.IGNORECASE):
return concept
if cs_urn:
return cs_all[-1].setdefault(id=string)
else:
raise ValueError(string)
[docs]
@cache
def get_version(with_dev: Optional[bool] = True) -> str:
"""Return a :class:`sdmx.model.common.Version` for :mod:`message_ix_models`.
.. todo:: Remove :py:`str(...)` once sdmx1 > 2.21.1 can handle Version.
Parameters
----------
with_dev : bool
If :any:`True`, include the dev version part, e.g. "2025.3.12.dev123". If not,
exclude.
"""
tmp, *_ = version(__package__.split(".")[0]).partition("+" if with_dev else ".dev")
return str(common.Version(tmp))
[docs]
def read(urn: str, base_dir: Optional["PathLike"] = None):
"""Read SDMX object from package data given its `urn`."""
# Identify a path that matches `urn`
base_dir = Path(base_dir or package_data_path("sdmx"))
urn = urn.replace(":", "_") # ":" invalid on Windows
paths = sorted(
set(base_dir.glob(f"*{urn}*.xml")) | set(base_dir.glob(f"*{urn.upper()}*.xml"))
)
if len(paths) > 1:
log.info(
f"Match {paths[0].relative_to(base_dir)} for {urn!r}; {len(paths) - 1} "
"other result(s)"
)
try:
with open(paths[0], "rb") as f:
msg = cast("StructureMessage", sdmx.read_sdmx(f))
except IndexError:
raise FileNotFoundError(f"'*{urn}*.xml', '*{urn.upper()}*.xml' or similar")
for _, cls in msg.iter_collections():
try:
return next(iter(msg.objects(cls).values()))
except StopIteration:
pass
[docs]
def write(obj, base_dir: Optional["PathLike"] = None, basename: Optional[str] = None):
"""Store an SDMX object as package data."""
base_dir = Path(base_dir or package_data_path("sdmx"))
if isinstance(obj, sdmx.message.StructureMessage):
msg = obj
assert basename
else:
# Set the URN of the object
obj.urn = sdmx.urn.make(obj)
# Wrap the object in a StructureMessage
msg = sdmx.message.StructureMessage()
msg.add(obj)
# Identify a path to write the file. ":" is invalid on Windows.
basename = basename or obj.urn.split("=")[-1].replace(":", "_")
msg.header = sdmx.message.Header(
source=f"Generated by message_ix_models {version('message_ix_models')}",
prepared=datetime.now(),
)
path = base_dir.joinpath(f"{basename}.xml")
# Write
path.parent.mkdir(parents=True, exist_ok=True)
path.write_bytes(sdmx.to_xml(msg, pretty_print=True))
log.info(f"Wrote {path}")
[docs]
def register_agency(agency: "common.Agency") -> "common.AgencyScheme":
"""Add `agency` to the :class:`.AgencyScheme` "IIASA_ECE:AGENCIES"."""
# Read the existing agency scheme
as_ = read("IIASA_ECE:AGENCIES")
if agency in as_:
log.info(f"Replace or update existing {as_[agency.id]!r}")
as_.items[agency.id] = agency
else:
as_.append(agency)
# Ensure URN is populated
agency.urn = agency.urn or sdmx.urn.make(agency, as_)
log.info(f"Updated {as_!r}")
# Write to file again
write(as_)
return as_