"""Data and parameter generation for the steel sector in MESSAGEix models.
This module provides functions to read, process, and generate parameter data for steel
technologies, demand, recycling, CCS, trade and related constraints.
"""
from collections import defaultdict
from collections.abc import Iterable
from typing import TYPE_CHECKING, Literal
import message_ix
import pandas as pd
from message_ix import make_df
from message_ix_models import ScenarioInfo
from message_ix_models.model.material.data_util import (
calculate_ini_new_cap,
read_rel,
read_sector_data,
read_timeseries,
)
from message_ix_models.model.material.demand import (
read_base_demand,
)
from message_ix_models.model.material.util import (
get_ssp_from_context,
maybe_remove_water_tec,
read_config,
remove_from_list_if_exists,
)
from message_ix_models.util import (
broadcast,
merge_data,
nodes_ex_world,
package_data_path,
same_node,
)
if TYPE_CHECKING:
from message_ix_models.types import ParameterData
[docs]
def gen_mock_demand_steel(scenario: message_ix.Scenario) -> pd.DataFrame:
"""Generate mock steel demand time series for MESSAGEix regions.
** Not used anymore **
Parameters
----------
scenario :
Scenario instance to build steel demand on.
Returns
-------
pd.DataFrame
DataFrame with columns ['node', 'year', 'value'] for steel demand.
"""
s_info = ScenarioInfo(scenario)
nodes = s_info.N
nodes.remove("World")
# The order:
# r = ['R12_AFR', 'R12_RCPA', 'R12_EEU', 'R12_FSU', 'R12_LAM', 'R12_MEA',\
# 'R12_NAM', 'R12_PAO', 'R12_PAS', 'R12_SAS', 'R12_WEU',"R12_CHN"]
# Finished steel demand from:
# https://www.oecd.org/industry/ind/Item_4b_Worldsteel.pdf
# For region definitions:
# https://worldsteel.org/wp-content/uploads/2021-World-Steel-in-Figures.pdf
# For detailed assumptions and calculation see: steel_demand_calculation.xlsx
# under
# https://iiasahub.sharepoint.com/sites/eceprog?cid=75ea8244-8757-44f1-83fd-d34f94ffd06a
if "R12_CHN" in nodes:
nodes.remove("R12_GLB")
sheet_n = "data_R12"
region_set = "R12_"
d = [
20.04,
12.08,
56.55,
56.5,
64.94,
54.26,
97.76,
91.3,
65.2,
164.28,
131.95,
980.1,
]
else:
nodes.remove("R11_GLB")
sheet_n = "data_R11"
region_set = "R11_"
d = [
20.04,
992.18,
56.55,
56.5,
64.94,
54.26,
97.76,
91.3,
65.2,
164.28,
131.95,
]
# MEA change from 39 to 9 to make it feasible (coal supply bound)
# SSP2 R11 baseline GDP projection
gdp_growth = pd.read_excel(
package_data_path("material", "other", "iamc_db ENGAGE baseline GDP PPP.xlsx"),
sheet_name=sheet_n,
)
gdp_growth = gdp_growth.loc[
(gdp_growth["Scenario"] == "baseline") & (gdp_growth["Region"] != "World")
].drop(["Model", "Variable", "Unit", "Notes", 2000, 2005], axis=1)
gdp_growth["Region"] = region_set + gdp_growth["Region"]
demand2020_steel = (
pd.DataFrame({"Region": nodes, "Val": d})
.join(gdp_growth.set_index("Region"), on="Region")
.rename(columns={"Region": "node"})
)
demand2020_steel.iloc[:, 3:] = (
demand2020_steel.iloc[:, 3:]
.div(demand2020_steel[2020], axis=0)
.multiply(demand2020_steel["Val"], axis=0)
)
demand2020_steel = pd.melt(
demand2020_steel.drop(["Val", "Scenario"], axis=1),
id_vars=["node"],
var_name="year",
value_name="value",
)
return demand2020_steel
[docs]
def gen_data_steel_ts(
data_steel_ts: pd.DataFrame, results: dict[str, list], t: str, nodes: list[str]
):
"""Generate time-series parameter data for steel technologies.
Parameters
----------
data_steel_ts :
DataFrame with time-series parameter data.
results :
Dictionary to collect parameter DataFrames.
t :
Technology name.
nodes :
List of model nodes.
"""
common = dict(
time="year",
time_origin="year",
time_dest="year",
)
param_name = data_steel_ts.loc[
(data_steel_ts["technology"] == t), "parameter"
].unique()
for p in set(param_name):
val = data_steel_ts.loc[
(data_steel_ts["technology"] == t) & (data_steel_ts["parameter"] == p),
"value",
]
# units = data_steel_ts.loc[
# (data_steel_ts["technology"] == t)
# & (data_steel_ts["parameter"] == p),
# "units",
# ].values[0]
mod = data_steel_ts.loc[
(data_steel_ts["technology"] == t) & (data_steel_ts["parameter"] == p),
"mode",
]
yr = data_steel_ts.loc[
(data_steel_ts["technology"] == t) & (data_steel_ts["parameter"] == p),
"year",
]
if p == "var_cost":
df = make_df(
p,
technology=t,
value=val,
unit="t",
year_vtg=yr,
year_act=yr,
mode=mod,
**common,
).pipe(broadcast, node_loc=nodes)
if p == "output":
comm = data_steel_ts.loc[
(data_steel_ts["technology"] == t) & (data_steel_ts["parameter"] == p),
"commodity",
]
lev = data_steel_ts.loc[
(data_steel_ts["technology"] == t) & (data_steel_ts["parameter"] == p),
"level",
]
rg = data_steel_ts.loc[
(data_steel_ts["technology"] == t) & (data_steel_ts["parameter"] == p),
"region",
]
df = make_df(
p,
technology=t,
value=val,
unit="t",
year_vtg=yr,
year_act=yr,
mode=mod,
node_loc=rg,
node_dest=rg,
commodity=comm,
level=lev,
**common,
)
else:
rg = data_steel_ts.loc[
(data_steel_ts["technology"] == t) & (data_steel_ts["parameter"] == p),
"region",
]
df = make_df(
p,
technology=t,
value=val,
unit="t",
year_vtg=yr,
year_act=yr,
mode=mod,
node_loc=rg,
**common,
)
results[p].append(df)
return
[docs]
def get_data_steel_const(
data_steel: pd.DataFrame,
results: dict[str, list],
params: Iterable,
t: str,
yv_ya: pd.DataFrame,
nodes: list[str],
global_region: str,
):
"""Generate time-independent (constant) parameter data for steel technologies.
Parameters
----------
data_steel :
DataFrame with constant parameter data.
results :
Dictionary to collect parameter DataFrames.
params :
Iterable of parameter names.
t :
Technology name.
yv_ya :
DataFrame with year vintage and year active combinations.
nodes :
List of model nodes.
global_region :
Name of the global region.
"""
for par in params:
# Obtain the parameter names, commodity,level,emission
split = par.split("|")
param_name = split[0]
# Obtain the scalar value for the parameter
val = data_steel.loc[
((data_steel["technology"] == t) & (data_steel["parameter"] == par)),
"value",
] # .values
regions = data_steel.loc[
((data_steel["technology"] == t) & (data_steel["parameter"] == par)),
"region",
] # .values
common = dict(
year_vtg=yv_ya.year_vtg,
year_act=yv_ya.year_act,
# mode="M1",
time="year",
time_origin="year",
time_dest="year",
)
for rg in regions:
# For the parameters which inlcudes index names
if len(split) > 1:
if (param_name == "input") | (param_name == "output"):
# Assign commodity and level names
com = split[1]
lev = split[2]
mod = split[3]
if (param_name == "input") and (lev == "import"):
df = make_df(
param_name,
technology=t,
commodity=com,
level=lev,
value=val[regions[regions == rg].index[0]],
mode=mod,
unit="t",
node_loc=rg,
node_origin=global_region,
**common,
)
elif (param_name == "output") and (lev == "export"):
df = make_df(
param_name,
technology=t,
commodity=com,
level=lev,
value=val[regions[regions == rg].index[0]],
mode=mod,
unit="t",
node_loc=rg,
node_dest=global_region,
**common,
)
else:
df = make_df(
param_name,
technology=t,
commodity=com,
level=lev,
value=val[regions[regions == rg].index[0]],
mode=mod,
unit="t",
node_loc=rg,
**common,
).pipe(same_node)
# Copy parameters to all regions, when node_loc is not GLB
if (len(regions) == 1) and (rg != global_region):
df["node_loc"] = None
df = df.pipe(broadcast, node_loc=nodes)
# Use same_node only for non-trade technologies
if (lev != "import") and (lev != "export"):
df = df.pipe(same_node)
elif param_name == "emission_factor":
# Assign the emisson type
emi = split[1]
mod = split[2]
df = make_df(
param_name,
technology=t,
value=val[regions[regions == rg].index[0]],
emission=emi,
mode=mod,
unit="t",
node_loc=rg,
**common,
)
else: # time-independent var_cost
mod = split[1]
df = make_df(
param_name,
technology=t,
value=val[regions[regions == rg].index[0]],
mode=mod,
unit="t",
node_loc=rg,
**common,
)
# Parameters with only parameter name
else:
df = make_df(
param_name,
technology=t,
value=val[regions[regions == rg].index[0]],
unit="t",
node_loc=rg,
**common,
)
# Copy parameters to all regions
if (
len(set(df["node_loc"])) == 1
and list(set(df["node_loc"]))[0] != global_region
):
df["node_loc"] = None
df = df.pipe(broadcast, node_loc=nodes)
results[param_name].append(df)
return
[docs]
def gen_data_steel_rel(
data_steel_rel: pd.DataFrame,
results: dict,
regions: set[str],
modelyears: list[int],
):
"""Generate relation parameter data for steel sector.
Parameters
----------
data_steel_rel :
DataFrame with relation parameter data.
results :
Dictionary to collect parameter DataFrames.
regions :
Iterable of region names.
modelyears :
List of model years.
"""
for reg in regions:
for r in data_steel_rel["relation"].unique():
model_years_rel = modelyears.copy()
if r is None:
break
if r == "max_global_recycling_steel":
continue
if r in ["minimum_recycling_steel", "max_regional_recycling_steel"]:
# Do not implement the minimum recycling rate for the year 2020
remove_from_list_if_exists(2020, model_years_rel)
params = set(
data_steel_rel.loc[
(data_steel_rel["relation"] == r), "parameter"
].values
)
common_rel = dict(
year_rel=model_years_rel,
year_act=model_years_rel,
mode="M1",
relation=r,
)
for par_name in params:
if par_name == "relation_activity":
tec_list = data_steel_rel.loc[
(
(data_steel_rel["relation"] == r)
& (data_steel_rel["parameter"] == par_name)
),
"technology",
]
for tec in tec_list.unique():
val = data_steel_rel.loc[
(
(data_steel_rel["relation"] == r)
& (data_steel_rel["parameter"] == par_name)
& (data_steel_rel["technology"] == tec)
& (data_steel_rel["Region"] == reg)
),
"value",
].values[0]
df = make_df(
par_name,
technology=tec,
value=val,
unit="-",
node_loc=reg,
node_rel=reg,
**common_rel,
).pipe(same_node)
results[par_name].append(df)
elif (par_name == "relation_upper") | (par_name == "relation_lower"):
val = data_steel_rel.loc[
(
(data_steel_rel["relation"] == r)
& (data_steel_rel["parameter"] == par_name)
& (data_steel_rel["Region"] == reg)
),
"value",
].values[0]
df = make_df(
par_name, value=val, unit="-", node_rel=reg, **common_rel
)
results[par_name].append(df)
return
[docs]
def gen_data_steel(scenario: message_ix.Scenario, dry_run: bool = False):
"""Generate all MESSAGEix parameter data for the steel sector.
Parameters
----------
scenario :
Scenario instance to build steel model on.
dry_run :
If True, do not perform any file writing or scenario modification.
Returns
-------
dict
Dictionary with MESSAGEix parameters as keys and parametrization as values.
"""
# Load configuration
context = read_config()
config = context["material"]["steel"]
ssp = get_ssp_from_context(context)
# Information about scenario, e.g. node, year
s_info = ScenarioInfo(scenario)
# Techno-economic assumptions
# TEMP: now add cement sector as well
# => Need to separate those since now I have get_data_steel and cement
data_steel = read_sector_data(scenario, "steel", None, "steel_R12.csv")
# Special treatment for time-dependent Parameters
data_steel_ts = read_timeseries(scenario, "steel", None, "timeseries_R12.csv")
data_steel_rel = read_rel(scenario, "steel", None, "relations_R12.csv")
tec_ts = set(data_steel_ts.technology) # set of tecs with var_cost
# List of data frames, to be concatenated together at end
results = defaultdict(list)
modelyears = s_info.Y # s_info.Y is only for modeling years
nodes = nodes_ex_world(s_info.N)
global_region = [i for i in s_info.N if i.endswith("_GLB")][0]
yv_ya = s_info.yv_ya
yv_ya = yv_ya.loc[yv_ya.year_vtg >= 1970]
df = yv_ya.loc[yv_ya.year_act == 2020]
df["year_act"] = None
pre_model = df.pipe(broadcast, year_act=df.year_vtg)
pre_model = pre_model[
(pre_model["year_act"].ge(pre_model["year_vtg"]))
& (pre_model["year_act"].lt(2020))
]
yv_ya = pd.concat([pre_model, yv_ya])
# For each technology there are differnet input and output combinations
# Iterate over technologies
for t in config["technology"]["add"]:
# Retrieve the id if `t` is a Code instance; otherwise use str
t = getattr(t, "id", t)
params = data_steel.loc[(data_steel["technology"] == t), "parameter"].unique()
# Special treatment for time-varying params
if t in tec_ts:
gen_data_steel_ts(data_steel_ts, results, t, nodes)
# Iterate over parameters
get_data_steel_const(
data_steel, results, params, t, yv_ya, nodes, global_region
)
# Add relation for the maximum global scrap use in 2020
for t in [
"scrap_recovery_steel_1",
"scrap_recovery_steel_2",
"scrap_recovery_steel_3",
]:
df_max_recycling = pd.DataFrame(
{
"relation": "max_global_recycling_steel",
"node_rel": "R12_GLB",
"year_rel": 2020,
"year_act": 2020,
"node_loc": nodes,
"technology": t,
"mode": "M1",
"unit": "???",
"value": data_steel_rel.loc[
(
(data_steel_rel["relation"] == "max_global_recycling_steel")
& (data_steel_rel["parameter"] == "relation_activity")
),
"value",
].values[0],
}
)
results["relation_activity"].append(df_max_recycling)
df_max_recycling_upper = pd.DataFrame(
{
"relation": "max_global_recycling_steel",
"node_rel": "R12_GLB",
"year_rel": 2020,
"unit": "???",
"value": data_steel_rel.loc[
(
(data_steel_rel["relation"] == "max_global_recycling_steel")
& (data_steel_rel["parameter"] == "relation_upper")
),
"value",
].values[0],
},
index=[0],
)
df_max_recycling_lower = pd.DataFrame(
{
"relation": "max_global_recycling_steel",
"node_rel": "R12_GLB",
"year_rel": 2020,
"unit": "???",
"value": data_steel_rel.loc[
(
(data_steel_rel["relation"] == "max_global_recycling_steel")
& (data_steel_rel["parameter"] == "relation_lower")
),
"value",
].values[0],
},
index=[0],
)
results["relation_activity"].append(df_max_recycling)
results["relation_upper"].append(df_max_recycling_upper)
results["relation_lower"].append(df_max_recycling_lower)
# Add relations for scrap grades and availability
regions = set(data_steel_rel["Region"].values)
gen_data_steel_rel(data_steel_rel, results, regions, modelyears)
# Create external demand param
df_demand = gen_demand(ssp)
results["demand"].append(df_demand)
new_scrap_ratio = {
"R12_AFR": 0.92,
"R12_CHN": 0.9,
"R12_EEU": 0.9,
"R12_FSU": 0.9,
"R12_LAM": 0.9,
"R12_MEA": 0.92,
"R12_NAM": 0.85,
"R12_PAO": 0.85,
"R12_PAS": 0.9,
"R12_RCPA": 0.92,
"R12_SAS": 0.92,
"R12_WEU": 0.85,
}
scale_fse_demand(df_demand, new_scrap_ratio)
common = dict(
year_vtg=yv_ya.year_vtg,
year_act=yv_ya.year_act,
time="year",
time_origin="year",
time_dest="year",
)
# Add CCS as addon
parname = "addon_conversion"
bf_tec = ["bf_steel"]
df = make_df(
parname, mode="M2", type_addon="bf_ccs_steel_addon", value=1, unit="-", **common
).pipe(broadcast, node=nodes, technology=bf_tec)
results[parname].append(df)
dri_gas_tec = ["dri_gas_steel"]
df = make_df(
parname,
mode="M1",
type_addon="dri_gas_ccs_steel_addon",
value=1,
unit="-",
**common,
).pipe(broadcast, node=nodes, technology=dri_gas_tec)
results[parname].append(df)
dri_tec = ["dri_steel"]
df = make_df(
parname, mode="M1", type_addon="dri_steel_addon", value=1, unit="-", **common
).pipe(broadcast, node=nodes, technology=dri_tec)
results[parname].append(df)
# Concatenate to one data frame per parameter
results = {par_name: pd.concat(dfs) for par_name, dfs in results.items()}
results["initial_new_capacity_up"] = pd.concat(
[
calculate_ini_new_cap(
df_demand=df_demand.copy(deep=True),
technology="dri_gas_ccs_steel",
material="steel",
ssp=ssp,
),
calculate_ini_new_cap(
df_demand=df_demand.copy(deep=True),
technology="bf_ccs_steel",
material="steel",
ssp=ssp,
),
]
)
results["relation_activity"] = pd.concat(
[results["relation_activity"], gen_cokeoven_co2_cc(s_info)]
)
merge_data(
results,
gen_dri_act_bound(),
gen_dri_cap_calibration(),
gen_dri_coal_model(s_info),
get_scrap_prep_cost(s_info, ssp),
gen_max_recycling_rel(s_info, ssp),
gen_grow_cap_up(s_info, ssp),
gen_2020_calibration_relation(s_info, "eaf"),
gen_2020_calibration_relation(s_info, "bof"),
gen_2020_calibration_relation(s_info, "bf"),
gen_bof_pig_input(s_info),
gen_finishing_steel_io(s_info),
gen_manuf_steel_io(new_scrap_ratio, s_info),
gen_iron_ore_cost(s_info, ssp),
gen_charcoal_bf_bound(s_info),
read_hist_cap("eaf"),
read_hist_cap("bof"),
read_hist_cap("bf"),
)
maybe_remove_water_tec(scenario, results)
if ssp == "SSP1":
df_tmp = results["relation_activity"]
df_tmp = df_tmp[
(df_tmp["relation"] == "minimum_recycling_steel")
& (df_tmp["technology"] == "total_EOL_steel")
]
df_tmp = df_tmp[df_tmp["year_rel"] >= 2030]
df_tmp["value"] = -0.7
reduced_pdict = {}
for k, v in results.items():
if set(["year_act", "year_vtg"]).issubset(v.columns):
v = v[(v["year_act"] - v["year_vtg"]) <= 60]
reduced_pdict[k] = v.drop_duplicates().copy(deep=True)
return reduced_pdict
[docs]
def gen_cokeoven_co2_cc(s_info: ScenarioInfo):
"""Generate relation_activity for CO2 emissions from coke ovens.
Returns
-------
pd.DataFrame
DataFrame with relation_activity for CO2.
"""
emi_dict = {
"unit": "-",
"technology": "cokeoven_steel",
"mode": "M1",
"value": 0.814 * 0.2,
# coal coeffient * difference between coal "final" and "dummy_emission" input
"relation": "CO2_cc",
}
df = (
make_df("relation_activity", **emi_dict)
.pipe(broadcast, node_loc=nodes_ex_world(s_info.N), year_act=s_info.Y)
.pipe(same_node)
)
df["year_rel"] = df["year_act"]
return df
[docs]
def gen_dri_act_bound() -> dict[str, pd.DataFrame]:
"""Read historical DRI activity data of R12 regions
for historical_activity and bound activity parameter.
Returns
-------
"""
df_act = pd.read_csv(
package_data_path(
"material", "steel", "baseyear_calibration", "dri_activity_2020.csv"
)
)
df_hist = pd.read_csv(
package_data_path(
"material", "steel", "baseyear_calibration", "dri_activity_hist.csv"
)
)
return {
"bound_activity_up": df_act,
"bound_activity_lo": df_act,
"historical_activity": df_hist,
}
[docs]
def gen_dri_cap_calibration() -> dict[str, pd.DataFrame]:
"""Read historical new DRI capacity data of R12 regions
for historical_activity and bound activity parameter.
Returns
-------
"""
df_cap_2020 = pd.read_csv(
package_data_path(
"material", "steel", "baseyear_calibration", "dri_capacity_2020.csv"
)
)
df_cap_hist = pd.read_csv(
package_data_path(
"material", "steel", "baseyear_calibration", "dri_capacity_hist.csv"
)
)
return {
"bound_new_capacity_up": df_cap_2020,
"bound_new_capacity_lo": df_cap_2020,
"historical_new_capacity": df_cap_hist,
}
[docs]
def gen_dri_coal_model(s_info: ScenarioInfo):
"""Generate techno-economic coal based DRI technology model parameters."""
model_years = s_info.Y
df = pd.read_csv(package_data_path("material", "steel", "dri_coal.csv"))
common = {
"time": "year",
"time_origin": "year",
"time_dest": "year",
}
par_dict = {}
for par in df.parameter.unique():
df_tmp = (
make_df(par, **df[df["parameter"] == par], **common)
.pipe(broadcast, year_act=model_years)
.pipe(same_node)
)
if "year_rel" in df_tmp.columns:
df_tmp["year_rel"] = df_tmp["year_act"]
if "year_vtg" in df_tmp.columns:
df_tmp["year_vtg"] = df_tmp["year_act"]
par_dict[par] = df_tmp
tec_lt_hist = make_df(
"technical_lifetime",
technology="dri_steel",
year_vtg=[i for i in range(1970, 2020, 5)],
value=[i for i in range(60, 30, -5)] + [30] * 4,
unit="???",
**common,
).pipe(
broadcast,
node_loc=nodes_ex_world(s_info.N),
)
tec_lt = make_df(
"technical_lifetime",
technology="dri_steel",
value=30,
year_vtg=model_years,
unit="???",
**common,
).pipe(
broadcast,
node_loc=nodes_ex_world(s_info.N),
)
par_dict["technical_lifetime"] = pd.concat([tec_lt_hist, tec_lt])
return par_dict
[docs]
def gen_2020_calibration_relation(
s_info: ScenarioInfo, tech: Literal["eaf", "bof", "bf"]
) -> dict[str, pd.DataFrame]:
"""Generate generic relation data to calibrate 2020 steel production by process.
Parameters
----------
tech :
Steel technology for which to generate historical activity data.
Returns
-------
"""
modes = {
"eaf": ["M1", "M2", "M3"],
"bof": ["M1", "M2"],
"bf": ["M1", "M2", "M3", "M4"],
}
df = (
make_df(
"relation_activity",
relation=f"{tech}_bound_2020",
year_rel=2020,
value=1,
technology=f"{tech}_steel",
year_act=2020,
unit="???",
)
.pipe(broadcast, mode=modes[tech])
.pipe(broadcast, node_loc=nodes_ex_world(s_info.N))
.pipe(same_node)
.assign(node_rel=lambda x: x["node_loc"])
)
rel_up = pd.read_csv(
package_data_path(
"material", "steel", "baseyear_calibration", f"{tech}_bound_2020.csv"
)
)
return {"relation_activity": df, "relation_upper": rel_up, "relation_lower": rel_up}
[docs]
def get_scrap_prep_cost(s_info: ScenarioInfo, ssp: str) -> "ParameterData":
"""Generate variable cost parameter for steel scrap preparation technologies.
Returns
-------
dict
Dictionary with 'var_cost' parameter DataFrame.
"""
# TODO Retrieve these from `s_info` instead of hard-coding
years1 = [i for i in range(2020, 2065, 5)]
years2 = [i for i in range(2070, 2115, 10)]
years = years1 + years2
ref_tec_ssp = {
"LED": "prep_secondary_steel_1",
"SSP1": "prep_secondary_steel_1",
"SSP3": "prep_secondary_steel_3",
"SSP5": "prep_secondary_steel_3",
}
start_val = {
"prep_secondary_steel_1": 70,
"prep_secondary_steel_2": 100,
"prep_secondary_steel_3": 130,
}
common = {"mode": "M1", "time": "year", "unit": "???", "year_act": years}
dfs = []
if ssp not in ref_tec_ssp.keys():
ref_cost1 = [
start_val[list(start_val.keys())[0]] * 1.025**i
for i, _ in enumerate(years1)
]
ref_cost2 = [ref_cost1[-1] * 1.05 ** (i + 1) for i, _ in enumerate(years2)]
ref_cost = ref_cost1 + ref_cost2
tec2_1 = [
start_val[list(start_val.keys())[1]] * 1.025**i
for i, _ in enumerate(years1)
]
tec2_2 = [tec2_1[-1] * 1.05 ** (i + 1) for i, _ in enumerate(years2)]
tec2 = tec2_1 + tec2_2
tec3_1 = [
start_val[list(start_val.keys())[2]] * 1.025**i
for i, _ in enumerate(years1)
]
tec3_2 = [tec3_1[-1] * 1.05 ** (i + 1) for i, _ in enumerate(years2)]
tec3 = tec3_1 + tec3_2
# Create 3 data frames with different var_cost values
for idx, value in ((0, ref_cost), (1, tec2), (2, tec3)):
t = list(start_val.keys())[idx]
dfs.append(make_df("var_cost", technology=t, value=value, **common))
else:
ref_cost1 = [
start_val[ref_tec_ssp[ssp]] * 1.025**i for i, _ in enumerate(years1)
]
ref_cost2 = [ref_cost1[-1] * 1.05 ** (i + 1) for i, _ in enumerate(years2)]
ref_cost = ref_cost1 + ref_cost2
other_tecs = start_val.keys() - [ref_tec_ssp[ssp]]
tec2 = [
start_val[list(other_tecs)[0]]
+ ((ref_cost[-1] - start_val[list(other_tecs)[0]]) / (len(years) - 1)) * i
for i, _ in enumerate(years)
]
tec3 = [
start_val[list(other_tecs)[1]]
+ ((ref_cost[-1] - start_val[list(other_tecs)[1]]) / (len(years) - 1)) * i
for i, _ in enumerate(years)
]
# Create 3 data frames with different var_cost values
for t, value in (
(ref_tec_ssp[ssp], ref_cost),
(list(other_tecs)[0], tec2),
(list(other_tecs)[1], tec3),
):
dfs.append(make_df("var_cost", technology=t, value=value, **common))
df = (
pd.concat(dfs)
.pipe(broadcast, node_loc=nodes_ex_world(s_info.N))
.assign(year_vtg=lambda x: x.year_act)
)
return {"var_cost": df}
[docs]
def gen_max_recycling_rel(s_info: ScenarioInfo, ssp):
"""Generate maximum recycling relation activity for steel.
Returns
-------
dict
Dictionary with 'relation_activity' DataFrame.
"""
ssp_vals = {
"LED": -0.98,
"SSP1": -0.98,
"SSP2": -0.85,
"SSP3": -0.85,
"SSP4": -0.85,
"SSP5": -0.85,
}
df = (
make_df(
"relation_activity",
technology="total_EOL_steel",
value=ssp_vals[ssp],
mode="M1",
relation="max_regional_recycling_steel",
time="year",
time_origin="year",
unit="???",
)
.pipe(broadcast, node_loc=nodes_ex_world(s_info.N))
.pipe(broadcast, year_act=[i for i in s_info.Y if i > 2020])
.pipe(same_node)
.assign(year_rel=lambda x: x.year_act)
)
return {"relation_activity": df}
[docs]
def gen_grow_cap_up(s_info: ScenarioInfo, ssp):
"""Generate growth constraints for new steel CCS capacity.
Returns
-------
dict
Dictionary with 'growth_new_capacity_up' parameter DataFrame.
"""
ssp_vals = {
"LED": 0.0010,
"SSP1": 0.0010,
"SSP2": 0.0010,
"SSP3": 0.0009,
"SSP4": 0.0020,
"SSP5": 0.0020,
}
df = (
make_df(
"growth_new_capacity_up",
technology=["bf_ccs_steel", "dri_gas_ccs_steel"],
value=ssp_vals[ssp],
unit="???",
)
.pipe(broadcast, node_loc=nodes_ex_world(s_info.N))
.pipe(broadcast, year_vtg=s_info.Y)
)
return {"growth_new_capacity_up": df}
[docs]
def scale_fse_demand(demand: pd.DataFrame, new_scrap_ratio: dict[str, float]):
"""Helper function to convert crude steel demand to
finished steel demand by scaling with new scrap ratio.
Parameters
----------
demand :
Steel demand projection data.
new_scrap_ratio :
Map of new scrap ratios for each region.
"""
demand["value"] = demand.apply(
lambda x: x["value"] * (new_scrap_ratio[x["node"]]), axis=1
)
[docs]
def gen_finishing_steel_io(s_info: ScenarioInfo) -> dict[str, pd.DataFrame]:
"""Generate output parameters for steel finishing process.
The output ratios are derived from worldsteel statistics of 2020.
Returns
-------
dict
Dictionary with 'output' parameter DataFrame.
"""
dimensions = {
"technology": "finishing_steel",
"mode": "M1",
"commodity": "steel",
"time": "year",
"time_dest": "year",
"unit": "???",
}
df = pd.read_csv(package_data_path("material", "steel", "finishing_loss_ratio.csv"))
df_out_steel = (
make_df("output", **df, **dimensions, level="useful_material")
.pipe(same_node)
.pipe(broadcast, year_act=s_info.Y)
.assign(year_vtg=lambda x: x["year_act"])
)
df_out_steel = df_out_steel[df_out_steel["year_act"].le(df_out_steel["year_vtg"])]
df_out_scrap = (
make_df("output", **df, **dimensions, level="new_scrap")
.pipe(same_node)
.pipe(broadcast, year_act=s_info.Y)
.assign(year_vtg=lambda x: x["year_act"])
)
df_out_scrap["value"] = df_out_scrap["value"].sub(1).abs().round(4)
df_out_scrap = df_out_scrap[df_out_scrap["year_act"].le(df_out_scrap["year_vtg"])]
return {"output": pd.concat([df_out_steel, df_out_scrap])}
[docs]
def gen_manuf_steel_io(
ratio: dict[str, float], s_info: ScenarioInfo
) -> dict[str, pd.DataFrame]:
"""Generate output parameters for steel manufacturing process.
Parameters
----------
ratio :
Map of new scrap yield ratio in manufacturing for each region.
Returns
-------
dict
Dictionary with 'output' parameter DataFrame.
"""
dimensions = {
"technology": "manuf_steel",
"mode": "M1",
"commodity": "steel",
"time": "year",
"time_dest": "year",
"unit": "???",
}
df = (
pd.Series(ratio)
.to_frame()
.reset_index()
.rename(columns={"index": "node_loc", 0: "value"})
)
df_out_steel = (
make_df("output", **df, **dimensions, level="product")
.pipe(same_node)
.pipe(broadcast, year_act=s_info.Y)
.assign(year_vtg=lambda x: x["year_act"])
)
df_out_steel = df_out_steel[df_out_steel["year_act"].le(df_out_steel["year_vtg"])]
df_out_scrap = (
make_df("output", **df, **dimensions, level="new_scrap")
.pipe(same_node)
.pipe(broadcast, year_act=s_info.Y)
.assign(year_vtg=lambda x: x["year_act"])
)
df_out_scrap["value"] = df_out_scrap["value"].sub(1).abs().round(4)
df_out_scrap = df_out_scrap[df_out_scrap["year_act"].le(df_out_scrap["year_vtg"])]
return {"output": pd.concat([df_out_steel, df_out_scrap])}
[docs]
def gen_iron_ore_cost(s_info: ScenarioInfo, ssp: str) -> dict[str, pd.DataFrame]:
"""Generate variable cost parameter for iron ore supply based on SSP narrative.
Returns
-------
dict
Dictionary with 'var_cost' parameter DataFrame.
"""
years1 = [i for i in range(2020, 2030, 5)]
years2 = [i for i in range(2030, 2060, 5)] + [i for i in range(2060, 2115, 10)]
start_val = 100
end_val = {
"LED": 130,
"SSP1": 130,
"SSP2": 80,
"SSP3": 100,
"SSP4": 60,
"SSP5": 80,
}
common = {
"technology": "DUMMY_ore_supply",
"mode": "M1",
"time": "year",
"unit": "???",
}
df1 = make_df(
"var_cost",
year_act=years1,
value=start_val,
**common,
)
df2 = make_df(
"var_cost",
year_act=years2,
value=end_val[ssp],
**common,
)
df = (
pd.concat([df1, df2])
.pipe(broadcast, node_loc=nodes_ex_world(s_info.N))
.assign(year_vtg=lambda x: x.year_act)
)
return {"var_cost": df}
[docs]
def gen_charcoal_bf_bound(s_info: ScenarioInfo) -> dict[str, pd.DataFrame]:
"""Generate bound activity for blast furnace modes with
charcoal input of 0 to calibrate to statistics.
Returns
-------
dict
Dictionary with 'bound_activity_up' parameter DataFrame.
"""
dimensions = {
"technology": "bf_steel",
"mode": ["M3", "M4"],
"time": "year",
"unit": "???",
"value": 0,
}
df = (
make_df("bound_activity_up", **dimensions)
.pipe(broadcast, node_loc=nodes_ex_world(s_info.N))
.pipe(broadcast, year_act=[2020, 2025])
)
return {"bound_activity_up": df}
[docs]
def gen_ssp_demand(ssp: str) -> pd.DataFrame:
"""Generate steel demand projections based on SSP scenarios.
Timeseries is currently read from pre-computed data file.
Returns
-------
pd.DataFrame
'demand' parameter DataFrame.
"""
mapping = {
"SSP1": {"phi": 5, "mu": 0.1, "q": 0.01},
"SSP2": {"phi": 5, "mu": 0.1, "q": 0.1},
"SSP3": {"phi": 5, "mu": 0.05, "q": 0.25},
"SSP4": {"phi": 5, "mu": 0.05, "q": 0.1},
"SSP5": {"phi": 5, "mu": 0.1, "q": 0.5},
}
if ssp == "LED":
ssp = "SSP1"
df = pd.read_parquet(
package_data_path("material", "steel", "demand_projections.parquet")
).drop_duplicates()
df = df[
(df["SSP"] == int(ssp.replace("SSP", "")))
& (df["quantile"] == mapping[ssp]["q"])
& (df["mu"] == mapping[ssp]["mu"])
]
df = df.rename(columns={"region": "node", "total_demand": "value"})
df[["commodity", "level", "time", "unit"]] = ["steel", "demand", "year", "t"]
df = make_df("demand", **df).round(2)
return df
[docs]
def gen_demand(ssp: str) -> pd.DataFrame:
"""Generate steel demand DataFrame for the given SSP scenario.
Combines calibrated 2020 and 2025 demand with SSP
dependent demand projection.
Returns
-------
pd.DataFrame
'demand' parameter DataFrame.
"""
df_2025 = pd.read_csv(package_data_path("material", "steel", "demand_2025.csv"))
df_2020 = (
read_base_demand(package_data_path("material", "steel", "demand_steel.yaml"))
.rename(columns={"region": "node"})
.assign(commodity="steel", level="demand", unit="t", time="year")
)
df_demand = gen_ssp_demand(ssp)
df_demand = df_demand[df_demand["year"] != 2025]
df_demand = pd.concat([df_2020, df_2025, df_demand])
return df_demand
[docs]
def read_hist_cap(tec: Literal["eaf", "bof", "bf"]) -> dict[str, pd.DataFrame]:
"""Read historical new capacity data for a given technology.
Parameters
----------
tec :
Steel technology for which to read historical capacity data.
Returns
-------
dict
Dictionary with 'historical_new_capacity' parameter DataFrame.
"""
df = pd.read_csv(
package_data_path(
"material", "steel", "baseyear_calibration", f"{tec}_capacity.csv"
)
)
df = df[df["year_vtg"] < 2020]
return {"historical_new_capacity": df}