import logging
from copy import deepcopy
from typing import TYPE_CHECKING
import pytest
from pytest import mark, param
from message_ix_models import ScenarioInfo
from message_ix_models.model.transport import build
from message_ix_models.model.transport.report import configure_legacy_reporting
from message_ix_models.model.transport.testing import (
MARK,
built_transport,
make_mark,
simulated_solution,
)
from message_ix_models.report import prepare_reporter, sim
from message_ix_models.testing import GHA
if TYPE_CHECKING:
import message_ix
log = logging.getLogger(__name__)
[docs]
@MARK[7]
@build.get_computer.minimum_version
@pytest.mark.parametrize(
"regions, years",
(
param("R11", "A", marks=make_mark[2](ValueError)),
param("R12", "B", marks=MARK["gh-281"]),
param("R14", "A", marks=MARK[9]),
param("ISR", "A", marks=MARK[3]),
),
)
def test_report_bare_solved(request, test_context, tmp_path, regions, years):
"""Run MESSAGEix-Transport–specific reporting."""
from message_ix_models.model.transport.report import callback
from message_ix_models.report import Config
# Update configuration
# key = "transport all" # All including plots, etc.
key = "transport::iamc+all" # IAMC-structured data stored and written to file
test_context.update(
regions=regions,
years=years,
report=Config("global.yaml", key=key, output_dir=tmp_path),
)
test_context.report.register(callback)
# Built and (optionally) solved scenario. dummy supply data is necessary for the
# scenario to be feasible without any other contents.
scenario = built_transport(
request, test_context, options=dict(dummy_supply=True), solved=True
)
# commented: for debugging
# dump_path = tmp_path / "scenario.xlsx"
# log.info(f"Dump contents to {dump_path}")
# scenario.to_excel(dump_path)
rep, key = prepare_reporter(test_context, scenario)
# Reporting `key` succeeds
rep.get(key)
[docs]
@pytest.fixture
def quiet_genno(caplog):
"""Quiet some log messages from genno via by :func:`.reporting.prepare_reporter`."""
caplog.set_level(logging.WARNING, logger="genno.config")
caplog.set_level(logging.WARNING, logger="genno.compat.pyam")
[docs]
@MARK[7]
@build.get_computer.minimum_version
@mark.usefixtures("quiet_genno")
def test_simulated_solution(request, test_context, regions="R12", years="B"):
""":func:`message_ix_models.report.prepare_reporter` works on the simulated data."""
test_context.update(regions=regions, years=years)
rep = simulated_solution(request, test_context)
# A quantity for a MESSAGEix variable was added and can be retrieved
k = rep.full_key("ACT")
rep.get(k)
# A quantity for MESSAGEix can be computed
k = rep.full_key("out")
rep.get(k)
# A quantity for message_data.model.transport can be computed
k = "transport stock::iamc"
result = rep.get(k)
assert 0 < len(result)
[docs]
@build.get_computer.minimum_version
@mark.usefixtures("quiet_genno")
@pytest.mark.parametrize(
"plot_name",
# # All plots
# list(PLOTS.keys()),
# Only a subset
[
# "energy-by-cmdty",
"stock-ldv",
# "stock-non-ldv",
],
)
def test_plot_simulated(request, test_context, plot_name, regions="R12", years="B"):
"""Plots are generated correctly using simulated data."""
test_context.update(regions=regions, years=years)
log.debug(f"test_plot_simulated: {test_context.regions = }")
rep = simulated_solution(request, test_context)
# print(rep.describe(f"plot {plot_name}")) # DEBUG
# Succeeds
rep.get(f"plot {plot_name}")
[docs]
@pytest.mark.xfail(condition=GHA, reason="Temporary, for #213; fails on GitHub Actions")
@sim.to_simulate.minimum_version
def test_iamc_simulated(
request, tmp_path_factory, test_context, regions="R12", years="B"
) -> None:
test_context.update(regions=regions, years=years)
test_context.report.output_dir = test_context.get_local_path()
rep = simulated_solution(request, test_context)
# Key collecting both file output/scenario update
# NB the trailing colons are necessary because of how genno handles report.yaml
rep.add(
"test",
[
"transport::iamc+file",
"transport::iamc+store",
# DEBUG Other keys:
# "emi:nl-t-yv-ya-m-e-h:transport",
],
)
# print(rep.describe("test")) # DEBUG
result = rep.get("test")
# print(result[-1]) # DEBUG
s: "message_ix.Scenario" = rep.get("scenario")
# File with output was created
path = tmp_path_factory.getbasetemp().joinpath(
"data0", ScenarioInfo(s).path, "transport.csv"
)
assert path.exists(), path
# Retrieve time series data stored on the scenario object
ts = s.timeseries()
# print(ts, ts["variable"].unique(), sep="\n") # DEBUG
# The reported data was stored on the scenario, and has expected variable names
# print("\n".join(sorted(ts["variable"].unique()))) # DEBUG
assert {
"Energy Service|Transportation|Domestic Aviation",
"Final Energy|Transportation|Bus",
"Transport|Stock|Road|Passenger|LDV|BEV",
} <= set(ts["variable"].unique())
del result