253 lines
7.4 KiB
Python
253 lines
7.4 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
# SPDX-FileCopyrightText: : 2017-2023 The PyPSA-Eur Authors
|
|
#
|
|
# SPDX-License-Identifier: MIT
|
|
|
|
import matplotlib.pyplot as plt
|
|
import pandas as pd
|
|
import pypsa
|
|
import country_converter as coco
|
|
import seaborn as sns
|
|
from _helpers import configure_logging
|
|
|
|
sns.set_theme("paper", style="whitegrid")
|
|
|
|
cc = coco.CountryConverter()
|
|
|
|
color_country = {
|
|
"AL": "#440154",
|
|
"AT": "#482677",
|
|
"BA": "#43398e",
|
|
"BE": "#3953a4",
|
|
"BG": "#2c728e",
|
|
"CH": "#228b8d",
|
|
"CZ": "#1f9d8a",
|
|
"DE": "#29af7f",
|
|
"DK": "#3fbc73",
|
|
"EE": "#5ec962",
|
|
"ES": "#84d44b",
|
|
"FI": "#addc30",
|
|
"FR": "#d8e219",
|
|
"GB": "#fde725",
|
|
"GR": "#f0f921",
|
|
"HR": "#f1c25e",
|
|
"HU": "#f4a784",
|
|
"IE": "#f78f98",
|
|
"IT": "#f87ea0",
|
|
"LT": "#f87a9a",
|
|
"LU": "#f57694",
|
|
"LV": "#f3758d",
|
|
"ME": "#f37685",
|
|
"MK": "#f37b7c",
|
|
"NL": "#f28774",
|
|
"NO": "#f1976b",
|
|
"PL": "#efaa63",
|
|
"PT": "#ebb160",
|
|
"RO": "#e6c260",
|
|
"RS": "#e2d75e",
|
|
"SE": "#dedc5b",
|
|
"SI": "#d9e35a",
|
|
"SK": "#d3e75a",
|
|
}
|
|
|
|
def sort_one_country(country, df):
|
|
indices = [link for link in df.columns if country in link]
|
|
df_country = df[indices].copy()
|
|
for link in df_country.columns:
|
|
if country in link[5:]:
|
|
df_country[link] = -df_country[link]
|
|
link_reverse = str(link[5:] + " - " + link[:2])
|
|
df_country = df_country.rename(columns={link: link_reverse})
|
|
|
|
return df_country.reindex(sorted(df_country.columns), axis=1)
|
|
|
|
|
|
def cross_border_time_series(countries, data):
|
|
|
|
fig, ax = plt.subplots(2*len(countries), 1, figsize=(15, 10*len(countries)))
|
|
axis = 0
|
|
|
|
for country in countries:
|
|
ymin = 0
|
|
ymax = 0
|
|
for df in data:
|
|
df_country = sort_one_country(country, df)
|
|
df_neg, df_pos = df_country.clip(upper=0), df_country.clip(lower=0)
|
|
|
|
color = [color_country[link[5:]] for link in df_country.columns]
|
|
|
|
df_pos.plot.area(
|
|
ax=ax[axis], stacked=True, linewidth=0.0, color=color, ylim=[-1, 1]
|
|
)
|
|
|
|
df_neg.plot.area(
|
|
ax=ax[axis], stacked=True, linewidth=0.0, color=color, ylim=[-1, 1]
|
|
)
|
|
if (axis % 2) == 0:
|
|
title = "Historic"
|
|
else:
|
|
title = "Optimized"
|
|
|
|
ax[axis].set_title(
|
|
title + " Import / Export for " + cc.convert(country, to="name_short")
|
|
)
|
|
|
|
# Custom legend elements
|
|
legend_elements = []
|
|
|
|
for link in df_country.columns:
|
|
legend_elements = legend_elements + [
|
|
plt.fill_between(
|
|
[],
|
|
[],
|
|
color=color_country[link[5:]],
|
|
label=cc.convert(link[5:], to="name_short"),
|
|
)
|
|
]
|
|
|
|
# Create the legend
|
|
ax[axis].legend(handles=legend_elements, loc="upper right")
|
|
|
|
# rescale the y axis
|
|
neg_min = df_neg.sum(axis=1).min() * 1.2
|
|
if neg_min < ymin:
|
|
ymin = neg_min
|
|
|
|
pos_max = df_pos.sum(axis=1).max() * 1.2
|
|
if pos_max < ymax:
|
|
ymax = pos_max
|
|
|
|
axis = axis + 1
|
|
|
|
for x in range(axis-2,axis):
|
|
ax[x].set_ylim([neg_min, pos_max])
|
|
|
|
fig.savefig(snakemake.output.trade_time_series, bbox_inches="tight")
|
|
|
|
|
|
def cross_border_bar(countries, data):
|
|
df_positive = pd.DataFrame()
|
|
df_negative = pd.DataFrame()
|
|
color = []
|
|
|
|
for country in countries:
|
|
order = 0
|
|
for df in data:
|
|
df_country = sort_one_country(country, df)
|
|
df_neg, df_pos = df_country.clip(upper=0), df_country.clip(lower=0)
|
|
|
|
if (order % 2) == 0:
|
|
title = "Historic"
|
|
else:
|
|
title = "Optimized"
|
|
|
|
color = [color_country[link[5:]] for link in df_country.columns] + color
|
|
df_positive_new = pd.DataFrame(data=df_pos.sum()).T.rename(
|
|
{0: title + " " + cc.convert(country, to="name_short")}
|
|
)
|
|
df_negative_new = pd.DataFrame(data=df_neg.sum()).T.rename(
|
|
{0: title + " " + cc.convert(country, to="name_short")}
|
|
)
|
|
|
|
df_positive = pd.concat([df_positive_new, df_positive])
|
|
df_negative = pd.concat([df_negative_new, df_negative])
|
|
|
|
order = order + 1
|
|
|
|
fig, ax = plt.subplots(figsize=(15, 60))
|
|
|
|
df_positive.plot.barh(ax=ax, stacked=True, color=color, zorder=2)
|
|
df_negative.plot.barh(ax=ax, stacked=True, color=color, zorder=2)
|
|
|
|
plt.grid(axis="x", zorder=0)
|
|
plt.grid(axis="y", zorder=0)
|
|
|
|
# Custom legend elements
|
|
legend_elements = []
|
|
|
|
for country in list(color_country.keys()):
|
|
legend_elements = legend_elements + [
|
|
plt.fill_between(
|
|
[],
|
|
[],
|
|
color=color_country[country],
|
|
label=cc.convert(country, to="name_short"),
|
|
)
|
|
]
|
|
|
|
# Create the legend
|
|
plt.legend(handles=legend_elements, loc="upper right")
|
|
|
|
fig.savefig(snakemake.output.cross_border_bar, bbox_inches="tight")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
if "snakemake" not in globals():
|
|
from _helpers import mock_snakemake
|
|
|
|
snakemake = mock_snakemake(
|
|
"plot_electricity_prices",
|
|
simpl="",
|
|
opts="Ept-12h",
|
|
clusters="37",
|
|
ll="v1.0",
|
|
)
|
|
configure_logging(snakemake)
|
|
|
|
countries = snakemake.params.countries
|
|
|
|
n = pypsa.Network(snakemake.input.network)
|
|
n.loads.carrier = "load"
|
|
|
|
historic = pd.read_csv(
|
|
snakemake.input.cross_border_flows,
|
|
index_col=0,
|
|
header=0,
|
|
parse_dates=True,
|
|
)
|
|
|
|
if len(historic.index) > len(n.snapshots):
|
|
historic = historic.resample(n.snapshots.inferred_freq).mean().loc[n.snapshots]
|
|
|
|
# Preparing network data to be shaped similar to ENTSOE datastructure
|
|
optimized_links = n.links_t.p0.rename(
|
|
columns=dict(n.links.bus0.str[:2] + " - " + n.links.bus1.str[:2])
|
|
)
|
|
optimized_lines = n.lines_t.p0.rename(
|
|
columns=dict(n.lines.bus0.str[:2] + " - " + n.lines.bus1.str[:2])
|
|
)
|
|
optimized = pd.concat([optimized_links, optimized_lines], axis=1)
|
|
|
|
# Drop internal country connection
|
|
optimized.drop([c for c in optimized.columns if c[:2] == c[5:]], axis=1, inplace=True)
|
|
|
|
# align columns name
|
|
for c1 in optimized.columns:
|
|
for c2 in optimized.columns:
|
|
if c1[:2] == c2[5:] and c2[:2] == c1[5:]:
|
|
optimized = optimized.rename(columns={c1: c2})
|
|
|
|
optimized = optimized.groupby(lambda x: x, axis=1).sum()
|
|
|
|
cross_border_bar(countries,[historic, optimized])
|
|
|
|
cross_border_time_series(countries, [historic, optimized])
|
|
|
|
# optimized = n.buses_t.marginal_price.groupby(n.buses.country, axis=1).mean()
|
|
|
|
# data = pd.concat([historic, optimized], keys=["Historic", "Optimized"], axis=1)
|
|
# data.columns.names = ["Kind", "Country"]
|
|
|
|
# # %% total production per carrier
|
|
# fig, ax = plt.subplots(figsize=(6, 6))
|
|
|
|
# df = data.mean().unstack().T
|
|
# df.plot.barh(ax=ax, xlabel="Electricity Price [€/MWh]", ylabel="")
|
|
# ax.grid(axis="y")
|
|
# fig.savefig(snakemake.output.price_bar, bbox_inches="tight")
|
|
|
|
# touch file
|
|
with open(snakemake.output.plots_touch, "a"):
|
|
pass
|