"""
Data and parameter generation for the methanol sector in
MESSAGEix-Materials model.
This module provides functions to read, process, and generate parameter data
for methanol technologies, demand, trade, and related constraints.
"""
from ast import literal_eval
from typing import TYPE_CHECKING
import pandas as pd
import yaml
from message_ix import make_df
from message_ix_models import ScenarioInfo
from message_ix_models.model.material.data_util import (
gen_chemicals_co2_ind_factors,
gen_plastics_emission_factors,
)
from message_ix_models.model.material.demand import gen_demand_petro
from message_ix_models.model.material.util import (
get_ssp_from_context,
read_config,
)
from message_ix_models.util import (
broadcast,
merge_data,
nodes_ex_world,
package_data_path,
same_node,
)
if TYPE_CHECKING:
from message_ix import Scenario
from message_ix_models.types import ParameterData
ssp_mode_map = {
"SSP1": "CTS core",
"SSP2": "RTS core",
"SSP3": "RTS high",
"SSP4": "CTS high",
"SSP5": "RTS high",
"LED": "CTS core", # TODO: move to even lower projection
}
iea_elasticity_map = {
"CTS core": (1.2, 0.25),
"CTS high": (1.3, 0.48),
"RTS core": (1.25, 0.35),
"RTS high": (1.4, 0.54),
}
[docs]
def gen_data_methanol(scenario: "Scenario") -> "ParameterData":
"""Generate parameter data for methanol industry model."""
context = read_config()
ssp = get_ssp_from_context(context)
df_pars = pd.read_excel(
package_data_path("material", "methanol", "methanol_sensitivity_pars.xlsx"),
sheet_name="Sheet1",
dtype=object,
)
pars = df_pars.set_index("par").to_dict()["value"]
if pars["mtbe_scenario"] == "phase-out":
pars_dict = pd.read_excel(
package_data_path("material", "methanol", "methanol_techno_economic.xlsx"),
sheet_name=None,
dtype=object,
)
else:
pars_dict = pd.read_excel(
package_data_path(
"material", "methanol", "methanol_techno_economic_high_demand.xlsx"
),
sheet_name=None,
dtype=object,
)
for i in pars_dict.keys():
pars_dict[i] = unpivot_input_data(pars_dict[i], i)
# TODO: only temporary hack to ensure SSP_dev compatibility
if "SSP_dev" in scenario.model:
file_path = package_data_path("material", "methanol", "missing_rels.yaml")
with open(file_path, "r") as file:
missing_rels = yaml.safe_load(file)
df = pars_dict["relation_activity"]
pars_dict["relation_activity"] = df[~df["relation"].isin(missing_rels)]
default_gdp_elasticity_2020, default_gdp_elasticity_2030 = iea_elasticity_map[
ssp_mode_map[ssp]
]
df_2025 = pd.read_csv(package_data_path("material", "methanol", "demand_2025.csv"))
df_demand = gen_demand_petro(
scenario, "methanol", default_gdp_elasticity_2020, default_gdp_elasticity_2030
)
df_demand["value"] = df_demand["value"].apply(
lambda x: x * pars["methanol_resid_demand_share"]
)
df_demand = df_demand[df_demand["year"] != 2025]
df_demand = pd.concat([df_2025, df_demand])
pars_dict["demand"] = df_demand
s_info = ScenarioInfo(scenario)
downstream_tec_pars = gen_meth_fs_downstream(s_info)
meth_downstream_emi_top_down = gen_plastics_emission_factors(s_info, "methanol")
meth_downstream_emi_bot_up = gen_chemicals_co2_ind_factors(s_info, "methanol")
merge_data(
pars_dict,
downstream_tec_pars,
meth_downstream_emi_top_down,
meth_downstream_emi_bot_up,
)
scen_rel_set = scenario.set("relation")
for par in ["activity", "upper", "lower"]:
df_rel = pars_dict[f"relation_{par}"]
df_rel = df_rel[df_rel["relation"].isin(scen_rel_set.values)]
exc_rels = [
i for i in df_rel["relation"].unique() if i not in scen_rel_set.values
]
print(
f"following relations are dropped from relation_{par} of methanol input "
f"data because they are not compatible with the scenario: {exc_rels}"
)
pars_dict[f"relation_{par}"] = df_rel
return pars_dict
[docs]
def broadcast_nodes(
df_bc_node: pd.DataFrame,
df_final: pd.DataFrame,
node_cols: list[str],
node_cols_codes: dict[str, pd.Series],
i: int,
) -> pd.DataFrame:
"""Broadcast nodes that were stored in pivoted row.
Parameters
----------
df_bc_node :
DataFrame containing parameter data with node values pivoted
df_final :
Full list of pivoted parameter data
node_cols :
list of node columns of the respective parameter
node_cols_codes :
list of node codes used for broadcasting
i :
index of the row in df_final to be broadcasted
"""
if len(node_cols) == 1:
if "node_loc" in node_cols:
df_bc_node = df_bc_node.pipe(
broadcast, node_loc=node_cols_codes["node_loc"]
)
if "node_vtg" in node_cols:
df_bc_node = df_bc_node.pipe(
broadcast, node_vtg=node_cols_codes["node_vtg"]
)
if "node_rel" in node_cols:
df_bc_node = df_bc_node.pipe(
broadcast, node_rel=node_cols_codes["node_rel"]
)
if "node" in node_cols:
df_bc_node = df_bc_node.pipe(broadcast, node=node_cols_codes["node"])
if "node_share" in node_cols:
df_bc_node = df_bc_node.pipe(
broadcast, node_share=node_cols_codes["node_share"]
)
else:
df_bc_node = df_bc_node.pipe(broadcast, node_loc=node_cols_codes["node_loc"])
if len(df_final.loc[i][node_cols].T.unique()) == 1:
# df_bc_node["node_rel"] = df_bc_node["node_loc"]
df_bc_node = df_bc_node.pipe(
same_node
) # not working for node_rel in installed message_ix_models version
else:
if "node_rel" in list(df_bc_node.columns):
df_bc_node = df_bc_node.pipe(
broadcast, node_rel=node_cols_codes["node_rel"]
)
if "node_origin" in list(df_bc_node.columns):
df_bc_node = df_bc_node.pipe(
broadcast, node_origin=node_cols_codes["node_origin"]
)
if "node_dest" in list(df_bc_node.columns):
df_bc_node = df_bc_node.pipe(
broadcast, node_dest=node_cols_codes["node_dest"]
)
return df_bc_node
[docs]
def broadcast_years(
df_bc_node: pd.DataFrame,
yr_col_out: list[str],
yr_cols_codes: dict[str, list[str]],
col: str,
) -> pd.DataFrame:
"""Broadcast years that were stored in pivoted row.
Parameters
----------
df_bc_node :
DataFrame containing parameter data with year values pivoted
yr_col_out :
list of year columns to be broadcasted
yr_cols_codes :
list of year codes used for broadcasting
col :
name of year column to be broadcasted
"""
if len(yr_col_out) == 1:
yr_list = [i[0] for i in yr_cols_codes[col]]
# print(yr_list)
if "year_act" in yr_col_out:
df_bc_node = df_bc_node.pipe(broadcast, year_act=yr_list)
if "year_vtg" in yr_col_out:
df_bc_node = df_bc_node.pipe(broadcast, year_vtg=yr_list)
if "year_rel" in yr_col_out:
df_bc_node = df_bc_node.pipe(broadcast, year_rel=yr_list)
if "year" in yr_col_out:
df_bc_node = df_bc_node.pipe(broadcast, year=yr_list)
df_bc_node[yr_col_out] = df_bc_node[yr_col_out].astype(int)
else:
if "year_vtg" in yr_col_out:
y_v = [str(i) for i in yr_cols_codes[col]]
df_bc_node = df_bc_node.pipe(broadcast, year_vtg=y_v)
df_bc_node["year_act"] = [
literal_eval(i)[1] for i in df_bc_node["year_vtg"]
]
df_bc_node["year_vtg"] = [
literal_eval(i)[0] for i in df_bc_node["year_vtg"]
]
if "year_rel" in yr_col_out:
if "year_act" in yr_col_out:
df_bc_node = df_bc_node.pipe(
broadcast, year_act=[i[0] for i in yr_cols_codes[col]]
)
df_bc_node["year_rel"] = df_bc_node["year_act"]
return df_bc_node
[docs]
def gen_meth_fs_downstream(s_info: "ScenarioInfo") -> "ParameterData":
"""Generate input and output parametrization for methanol downstream use technology.
Parameters
----------
s_info:
Scenario info object to infer regions and years from.
"""
# input parameter
yv_ya = s_info.yv_ya
year_all = yv_ya["year_act"].unique()
tec_name = "meth_ind_fs"
cols = {
"technology": tec_name,
"commodity": "methanol",
"mode": "M1",
"level": "final_material",
"time": "year",
"time_origin": "year",
"value": 1,
"unit": "Mt",
}
df_in = (
make_df("input", **cols)
.pipe(broadcast, node_loc=nodes_ex_world(s_info.N), year_act=year_all)
.pipe(same_node)
)
df_in["year_vtg"] = df_in["year_act"]
# output parameter
tec_name = "meth_ind_fs"
cols = {
"technology": tec_name,
"commodity": "methanol",
"mode": "M1",
"level": "demand",
"time": "year",
"time_dest": "year",
"value": 1,
"unit": "Mt",
}
df_out = (
make_df("output", **cols)
.pipe(broadcast, node_loc=nodes_ex_world(s_info.N), year_act=year_all)
.pipe(same_node)
)
df_out["year_vtg"] = df_out["year_act"]
return dict(input=df_in, output=df_out)