Source code for message_ix_models.tests.model.transport.test_emission
import numpy as np
import pandas as pd
import pytest
from message_ix import make_df
from message_ix_models.model.transport import Config, DataSourceConfig, build, testing
from message_ix_models.model.transport.emission import ef_for_input, get_emissions_data
from message_ix_models.util import broadcast, same_node
[docs]
@build.get_computer.minimum_version
def test_ef_for_input(test_context):
# Generate a test "input" data frame
_, info = testing.configure_build(test_context, regions="R11", years="B")
years = info.yv_ya
data = (
make_df(
"input",
year_vtg=years.year_vtg,
year_act=years.year_act,
technology="t",
mode="m",
commodity=None,
level="final",
time="year",
time_origin="year",
value=0.05,
unit="GWa / (Gv km)",
)
.pipe(broadcast, node_loc=info.N)
.pipe(same_node)
)
# Generate random commodity values
c = ("electr", "ethanol", "gas", "hydrogen", "lightoil", "methanol")
splitter = np.random.choice(np.arange(len(c)), len(data))
data = data.assign(
commodity=pd.Categorical.from_codes(splitter, categories=c),
)
assert not data.isna().any().any(), data
# Function runs successfully on these data
result = ef_for_input(test_context, data)
# Returns data for two parameters if transport.Config.emission_relations is True
# (the default)
assert {"emission_factor", "relation_activity"} == set(result)
ef = result["emission_factor"]
# Data is complete
assert not ef.isna().any().any(), ef
# Data have the expected columns
assert sorted(make_df("emission_factor").columns) == sorted(ef.columns)
ra = result["relation_activity"]
assert not ra.isna().any(axis=None), ra
assert ra.dtypes["year_act"] == int # noqa: E721
# print(ra.to_string())
# TODO test specific values
[docs]
@pytest.mark.parametrize("source, rows", (("1", 4717), ("2", 5153)))
@pytest.mark.parametrize("regions", ["R11"])
def test_get_emissions_data(test_context, source, rows, regions):
# Set the value; don't need to read_config()
test_context.model.regions = regions
test_context.transport = Config(data_source=DataSourceConfig(emissions=source))
data = get_emissions_data(test_context)
assert {"emission_factor"} == set(data.keys())
assert rows == len(data["emission_factor"])