import logging
from dataclasses import dataclass
from typing import TYPE_CHECKING
from message_ix_models.tools.exo_data import BaseOptions, ExoDataSource, register_source
from message_ix_models.tools.iamc import iamc_like_data_for_query
from message_ix_models.util import path_fallback
if TYPE_CHECKING:
from genno.types import AnyQuantity
__all__ = [
"ADVANCE",
]
log = logging.getLogger(__name__)
#: Expected location of the ADVANCE WP2 data snapshot.
LOCATION = "advance", "advance_compare_20171018-134445.csv.xz"
#: Name of the data file within the archive.
NAME = "advance_compare_20171018-134445.csv"
[docs]
@register_source
class ADVANCE(ExoDataSource):
"""Provider of exogenous data from the ADVANCE project database.
Example
-------
Load the metadata packaged with :mod:`message_ix_models` to identify usable
`source_kw`:
>>> import sdmx
>>> from message_ix_models.util import package_data_path
>>>
>>> msg = sdmx.read_sdmx(package_data_path("sdmx", "ADVANCE.xml"))
>>> msg
<sdmx.StructureMessage>
<Header>
prepared: '2023-11-03T21:51:46.052879'
source: en: Generated by message_ix_models 2023.9.13.dev57+ga558c0b4.d20231011
test: False
Codelist (5): MODEL SCENARIO REGION VARIABLE UNIT
>>> msg.codelist["MODEL].items
{'AIM/CGE': <Code AIM/CGE>,
'DNE21+': <Code DNE21+>,
'GCAM': <Code GCAM>,
'GEM-E3': <Code GEM-E3>,
'IMACLIM V1.1': <Code IMACLIM V1.1>,
'IMAGE': <Code IMAGE>,
'MESSAGE': <Code MESSAGE>,
'POLES ADVANCE': <Code POLES ADVANCE>,
'REMIND': <Code REMIND>,
'TIAM-UCL': <Code TIAM-UCL>,
'WITCH': <Code WITCH>,
'iPETS V.1.5': <Code iPETS V.1.5>}
"""
[docs]
@dataclass
class Options(BaseOptions):
#: If :any:`True`, aggregate data from the ADVANCE native regions using
#: ``n::groups`` (same behaviour as the base class). Otherwise, do not
#: aggregate.
aggregate: bool = True
#: One of 3080 codes for the "VARIABLE" dimension.
measure: str = ""
#: One of 12 codes including "MESSAGE".
model: str = ""
#: One of 51 codes including "ADV3TRAr2_Base".
scenario: str = ""
#: :py:`where` argument to :func:`path_fallback`, used to identify the directory
#: containing :data:`LOCATION`. Currently data is stored in message-static-data,
#: cloned and linked from within the user's 'local' data directory.
where = ["local"]
def __init__(self, *args, **kwargs) -> None:
opt = self.options = self.Options.from_args(self, *args, **kwargs)
# Map the `measure` keyword to a string appearing in the data
opt.name = variable = {"GDP": "GDP|PPP", "POP": "Population"}.get(
opt.measure, opt.measure
)
# Assemble a query string
self.query = (
f"SCENARIO == {opt.scenario!r} and VARIABLE == {variable!r} and "
+ (f"MODEL == {opt.model!r}" if opt.model else "True")
)
log.debug(self.query)
# Expected location of the ADVANCE WP2 data snapshot.
self.path = path_fallback(*LOCATION, where=self._where())
super().__init__() # Create .key
[docs]
def get(self) -> "AnyQuantity":
return iamc_like_data_for_query(
self.path, self.query, archive_member=NAME, non_iso_3166="keep"
)