Source code for message_ix_models.project.advance.data

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" )