Functions supporting myopic pathway network creation
add_build_year_to_new_assets(n, baseyear)
add a build year to new assets
Parameters: |
|
---|
Source code in workflow/scripts/add_existing_baseyear.py
def add_build_year_to_new_assets(n: pypsa.Network, baseyear: int):
"""add a build year to new assets
Args:
n (pypsa.Network): the network
baseyear (int): year in which optimized assets are built
"""
# Give assets with lifetimes and no build year the build year baseyear
for c in n.iterate_components(["Link", "Generator", "Store"]):
attr = "e" if c.name == "Store" else "p"
assets = c.df.index[(c.df.lifetime != np.inf) & (c.df[attr + "_nom_extendable"] is True)]
# add -baseyear to name
rename = pd.Series(c.df.index, c.df.index)
rename[assets] += "-" + str(baseyear)
c.df.rename(index=rename, inplace=True)
assets = c.df.index[
(c.df.lifetime != np.inf)
& (c.df[attr + "_nom_extendable"] is True)
& (c.df.build_year == 0)
]
c.df.loc[assets, "build_year"] = baseyear
# rename time-dependent
selection = n.component_attrs[c.name].type.str.contains("series") & n.component_attrs[
c.name
].status.str.contains("Input")
for attr in n.component_attrs[c.name].index[selection]:
c.pnl[attr].rename(columns=rename, inplace=True)
add_power_capacities_installed_before_baseyear(n, grouping_years, costs, baseyear, config)
Parameters
n : pypsa.Network grouping_years : intervals to group existing capacities costs : to read lifetime to estimate YearDecomissioning baseyear : int
Source code in workflow/scripts/add_existing_baseyear.py
def add_power_capacities_installed_before_baseyear(
n: pypsa.Network, grouping_years, costs, baseyear, config
):
"""
Parameters
----------
n : pypsa.Network
grouping_years :
intervals to group existing capacities
costs :
to read lifetime to estimate YearDecomissioning
baseyear : int
"""
logger.info("adding power capacities installed before baseyear")
df_agg = pd.DataFrame()
# include renewables in df_agg
add_existing_capacities(df_agg)
df_agg["grouping_year"] = np.take(
grouping_years, np.digitize(df_agg.DateIn, grouping_years, right=True)
)
df = df_agg.pivot_table(
index=["grouping_year", "Tech"], columns="cluster_bus", values="Capacity", aggfunc="sum"
)
df.fillna(0)
carrier = {
"coal": "coal power plant",
"CHP coal": "CHP coal",
"CHP gas": "CHP gas",
"OCGT": "OCGT gas",
"solar": "solar",
"solar thermal": "solar thermal",
"onwind": "onwind",
"offwind": "offwind",
"coal boiler": "coal boiler",
"ground heat pump": "heat pump",
"nuclear": "nuclear",
}
for grouping_year, generator in df.index:
# capacity is the capacity in MW at each node for this
capacity = df.loc[grouping_year, generator]
capacity = capacity[~capacity.isna()]
capacity = capacity[capacity > config["existing_capacities"]["threshold_capacity"]]
if generator in ["solar", "onwind", "offwind"]:
p_max_pu = n.generators_t.p_max_pu[capacity.index + " " + generator]
n.add(
"Generator",
capacity.index,
suffix=" " + generator + "-" + str(grouping_year),
bus=capacity.index,
carrier=carrier[generator],
p_nom=capacity,
p_nom_min=capacity,
p_nom_extendable=False,
marginal_cost=costs.at[generator, "marginal_cost"],
capital_cost=costs.at[generator, "capital_cost"],
efficiency=costs.at[generator, "efficiency"],
p_max_pu=p_max_pu.rename(columns=n.generators.bus),
build_year=grouping_year,
lifetime=costs.at[generator, "lifetime"],
)
if generator == "coal":
n.add(
"Generator",
capacity.index,
suffix=" " + generator + "-" + str(grouping_year),
bus=capacity.index,
carrier=carrier[generator],
p_nom=capacity,
p_nom_extendable=False,
marginal_cost=costs.at[generator, "marginal_cost"],
capital_cost=costs.at[generator, "capital_cost"],
efficiency=costs.at[generator, "efficiency"],
build_year=grouping_year,
lifetime=costs.at[generator, "lifetime"],
)
if generator == "nuclear":
n.add(
"Generator",
capacity.index,
suffix=" " + generator + "-" + str(grouping_year),
bus=capacity.index,
carrier=carrier[generator],
p_nom=capacity,
p_nom_min=capacity,
p_nom_extendable=False,
p_min_pu=0.7,
marginal_cost=costs.at[generator, "marginal_cost"],
capital_cost=costs.at[generator, "capital_cost"],
efficiency=costs.at[generator, "efficiency"],
build_year=grouping_year,
lifetime=costs.at[generator, "lifetime"],
)
if generator == "solar thermal":
p_max_pu = n.generators_t.p_max_pu[capacity.index + " central " + generator]
p_max_pu.columns = capacity.index
n.add(
"Generator",
capacity.index,
suffix=" central " + generator + "-" + str(grouping_year),
bus=capacity.index + " central heat",
carrier=carrier[generator],
p_nom=capacity,
p_nom_min=capacity,
p_nom_extendable=False,
marginal_cost=costs.at["central " + generator, "marginal_cost"],
capital_cost=costs.at["central " + generator, "capital_cost"],
p_max_pu=p_max_pu,
build_year=grouping_year,
lifetime=costs.at["central " + generator, "lifetime"],
)
if generator == "CHP coal":
bus0 = capacity.index + " coal"
n.add(
"Link",
capacity.index,
suffix=" " + generator + " generator" + "-" + str(grouping_year),
bus0=bus0,
bus1=capacity.index,
carrier=carrier[generator],
marginal_cost=0.37 * costs.at["central coal CHP", "VOM"], # NB: VOM is per MWel
capital_cost=0.37
* costs.at["central coal CHP", "capital_cost"], # NB: fixed cost is per MWel,
p_nom=capacity / 0.37,
p_nom_min=capacity / 0.37,
p_nom_extendable=False,
efficiency=0.37,
p_nom_ratio=1.0,
c_b=0.75,
build_year=grouping_year,
lifetime=costs.at["central coal CHP", "lifetime"],
)
n.add(
"Link",
capacity.index,
suffix=" " + generator + " boiler" + "-" + str(grouping_year),
bus0=bus0,
bus1=capacity.index + " central heat",
carrier=carrier[generator],
marginal_cost=0.37 * costs.at["central coal CHP", "VOM"], # NB: VOM is per MWel
p_nom=capacity / 0.37 * 0.15,
p_nom_min=capacity / 0.37 * 0.15,
p_nom_extendable=False,
efficiency=0.37 / 0.15,
build_year=grouping_year,
lifetime=costs.at["central coal CHP", "lifetime"],
)
if generator == "CHP gas":
bus0 = capacity.index + " gas"
n.add(
"Link",
capacity.index,
suffix=" " + generator + " generator" + "-" + str(grouping_year),
bus0=bus0,
bus1=capacity.index,
carrier=carrier[generator],
marginal_cost=costs.at["central gas CHP", "efficiency"]
* costs.at["central gas CHP", "VOM"], # NB: VOM is per MWel
capital_cost=costs.at["central gas CHP", "efficiency"]
* costs.at["central gas CHP", "capital_cost"], # NB: fixed cost is per MWel,
p_nom=capacity / costs.at["central gas CHP", "efficiency"],
p_nom_min=capacity / costs.at["central gas CHP", "efficiency"],
p_nom_extendable=False,
efficiency=costs.at["central gas CHP", "efficiency"],
p_nom_ratio=1.0,
c_b=costs.at["central gas CHP", "c_b"],
build_year=grouping_year,
lifetime=costs.at["central gas CHP", "lifetime"],
)
n.add(
"Link",
capacity.index,
suffix=" " + generator + " boiler" + "-" + str(grouping_year),
bus0=bus0,
bus1=capacity.index + " central heat",
carrier=carrier[generator],
marginal_cost=costs.at["central gas CHP", "efficiency"]
* costs.at["central gas CHP", "VOM"], # NB: VOM is per MWel
p_nom=capacity
/ costs.at["central gas CHP", "efficiency"]
* costs.at["central gas CHP", "c_v"],
p_nom_min=capacity
/ costs.at["central gas CHP", "efficiency"]
* costs.at["central gas CHP", "c_v"],
p_nom_extendable=False,
efficiency=costs.at["central gas CHP", "efficiency"]
/ costs.at["central gas CHP", "c_v"],
build_year=grouping_year,
lifetime=costs.at["central gas CHP", "lifetime"],
)
if generator == "OCGT":
bus0 = capacity.index + " gas"
n.add(
"Link",
capacity.index,
suffix=" " + generator + "-" + str(grouping_year),
bus0=bus0,
bus1=capacity.index,
carrier=carrier[generator],
marginal_cost=costs.at[generator, "efficiency"]
* costs.at[generator, "VOM"], # NB: VOM is per MWel
capital_cost=costs.at[generator, "efficiency"]
* costs.at[generator, "capital_cost"],
# NB: fixed cost is per MWel
p_nom=capacity / costs.at[generator, "efficiency"],
p_nom_min=capacity / costs.at[generator, "efficiency"],
p_nom_extendable=False,
efficiency=costs.at[generator, "efficiency"],
build_year=grouping_year,
lifetime=costs.at[generator, "lifetime"],
)
if generator == "coal boiler":
bus0 = capacity.index + " coal"
for cat in [" central "]:
n.add(
"Link",
capacity.index,
suffix="" + cat + generator + "-" + str(grouping_year),
bus0=bus0,
bus1=capacity.index + cat + "heat",
carrier=carrier[generator],
marginal_cost=costs.at[cat.lstrip() + generator, "efficiency"]
* costs.at[cat.lstrip() + generator, "VOM"],
capital_cost=costs.at[cat.lstrip() + generator, "efficiency"]
* costs.at[cat.lstrip() + generator, "capital_cost"],
p_nom=capacity / costs.at[cat.lstrip() + generator, "efficiency"],
p_nom_min=capacity / costs.at[cat.lstrip() + generator, "efficiency"],
p_nom_extendable=False,
efficiency=costs.at[cat.lstrip() + generator, "efficiency"],
build_year=grouping_year,
lifetime=costs.at[cat.lstrip() + generator, "lifetime"],
)
# TODO fix centralise
if generator == "ground heat pump":
date_range = pd.date_range(
"2025-01-01 00:00",
"2025-12-31 23:00",
freq=config["snapshots"]["freq"],
tz="Asia/shanghai",
)
date_range = date_range.map(lambda t: t.replace(year=2020))
with pd.HDFStore(snakemake.input.cop_name, mode="r") as store:
gshp_cop = store["gshp_cop_profiles"]
gshp_cop.index = gshp_cop.index.tz_localize("Asia/shanghai")
gshp_cop = gshp_cop.loc[date_range].set_index(n.snapshots)
n.add(
"Link",
capacity.index,
suffix=" " + generator + "-" + str(grouping_year),
bus0=capacity.index,
bus1=capacity.index + " central heat",
carrier="heat pump",
efficiency=(
gshp_cop[capacity.index]
if config["time_dep_hp_cop"]
else costs.at["decentral ground-sourced heat pump", "efficiency"]
),
capital_cost=costs.at["decentral ground-sourced heat pump", "efficiency"]
* costs.at["decentral ground-sourced heat pump", "capital_cost"],
marginal_cost=costs.at["decentral ground-sourced heat pump", "efficiency"]
* costs.at["decentral ground-sourced heat pump", "marginal_cost"],
p_nom=capacity / costs.at["decentral ground-sourced heat pump", "efficiency"],
p_nom_min=capacity / costs.at["decentral ground-sourced heat pump", "efficiency"],
p_nom_extendable=False,
build_year=grouping_year,
lifetime=costs.at["decentral ground-sourced heat pump", "lifetime"],
)