import plotly.graph_objects as go
import plotly.express as px
from typing import Dict
import pandas as pd
from pathlib import Path
import pypsa # type: ignore
from . import utils
import numpy as np
import matplotlib.pyplot as plt
import seaborn as sns
from pypsa_bc import utils as pypsabc_utils
import matplotlib.cm as cm
font_family='serif'
strategy_scenario_colors = {
'Uncoordinated': "#db990b", #Oranges
'Hybrid': "#0536A1", # Blues
'Coordinated': "#1aa71a", # Greens
'V2g': "#7413cf" #Purples
}
strategy_scenario_cmaps = {
'Uncoordinated': cm.Oranges,
'Hybrid': cm.Blues,
'Coordinated': cm.Greens,
'V2g': cm.Purples
}
resample_labels = {
'h': 'Hourly',
'd': 'Daily',
'w': 'Weekly',
'm': 'Monthly',
'me': 'Month-End',
'q': 'Quarterly',
'qe': 'Quarter-End'
}
aggregation_method_annotations={
'max':'Peaks',
'mean':'Average',
'mode':'Modes',
'median':'Medians',
'sum':'Total',
'min':'Minimums'
}
[docs]
def get_resampled_data(data:pd.DataFrame,
resample_freq:str='h',
aggregation_method:str= 'mean') -> pd.DataFrame:
"""
Resample the data based on the specified frequency and aggregation method.
Args:
data (pd.DataFrame): The input data to be resampled.
resample_freq (str): The frequency string for resampling (e.g., 'H', 'D', 'W', 'M', 'Q').
aggregation_method (str): The aggregation method to apply (mean, median, sum, max, min).
"""
resample_freq = resample_freq.lower()
if aggregation_method == 'mean':
data_resampled = data.resample(resample_freq).mean()
elif aggregation_method == 'median':
data_resampled = data.resample(resample_freq).median()
elif aggregation_method == 'sum':
data_resampled = data.resample(resample_freq).sum()
elif aggregation_method == 'max':
data_resampled = data.resample(resample_freq).max()
elif aggregation_method == 'min':
data_resampled = data.resample(resample_freq).min()
elif aggregation_method == 'mode':
data_resampled = data.resample(resample_freq).apply(lambda x: x.mode().iloc[0] if not x.mode().empty else np.nan)
else:
raise ValueError(f"Invalid aggregation method: {aggregation_method}. Defaulting to median.")
return data_resampled
[docs]
def get_annotation_text(resample_freq: str,
aggregation_method :str=None):
resample_freq=resample_freq.lower()
resample_freq_suffix = resample_labels.get(resample_freq, '')
aggregation_method_annotation=aggregation_method_annotations.get(aggregation_method, '')
# Construct annotation text (skip aggregation if resampling is hourly)
annotation_text = ''
if resample_freq != 'h': # Exclude hourly resampling
if aggregation_method:
annotation_text += f'{resample_freq_suffix} | {aggregation_method_annotation}'
else:
annotation_text += f'{resample_freq_suffix} | Average' # Default aggregation method if resampling is applied
return annotation_text
[docs]
def plot_ev_loads_interactive(ev_population: int,
ev_load_results: dict,
resample_freq: str,
aggregation_method:str='mean'):
"""
Plots interactive EV load profiles for different scenarios using Plotly.
Parameters:
- scale (int): The penetration scale of EVs (e.g., 50, 100).
- ev_load_results (dict): Dictionary containing EV load time series data for different networks.
- resample_freq (str, optional): Frequency string to resample the time series data (e.g., 'h' for hourly, 'D' for daily, 'M' for Monthly, 'Q' for Quarterly).
If None, no resampling is applied.
- aggregation_method (str, optional): Aggregation method (mean, median, sum, max, min). Default is 'median'.
The function creates an interactive plot with different colors for each coordination type (uncoordinated, coordinated, v2g).
"""
fig = go.Figure()
ev_population=int(ev_population)
for network_name, ev_load in ev_load_results.items():
name_parts = network_name.split('_')
if "hybrid" in name_parts:
coord_type= name_parts[3] +'_'+ name_parts[4] + '_'+name_parts[5]
ev_pop= name_parts[6]
run_tag = name_parts[7]+'_'+name_parts[8]
plot_title=f'EV Load Profile 100% Penetration [{str(ev_population)}% Coordinated, {str(100-ev_population)}% Uncoordinated Charging])'
else:
coord_type = name_parts[3]
ev_pop=name_parts[4]
run_tag = name_parts[5]
plot_title=f'EV Load Profile ({str(ev_population)}% Penetration)'
if ev_population == int(ev_pop):
label = f"{coord_type.capitalize()} ({run_tag})"
if resample_freq:
resample_freq = resample_freq.lower()
resample_freq_suffix = resample_labels.get(resample_freq, '')
ev_load_resampled=get_resampled_data(ev_load,resample_freq,aggregation_method)
fig.add_trace(go.Scatter(
x=ev_load_resampled.index,
y=ev_load_resampled.values,
mode='lines',
name=label,
line=dict(color=strategy_scenario_colors.get(coord_type.capitalize(), "blue"), width=2),
opacity=0.5
))
# Construct annotation text (skip aggregation if resampling is hourly)
annotation_text = ''
if resample_freq and resample_freq != 'h': # Exclude hourly resampling
annotation_text += f'{resample_freq_suffix}'
if aggregation_method:
annotation_text += f' | {aggregation_method.capitalize()}'
else:
annotation_text += ' | Median' # Default aggregation method if resampling is applied
# Update layout for better visualization
fig.update_layout(
title=plot_title,
# xaxis_title='Time',
yaxis_title='EV Load (MW)',
legend_title='Scenario, Run Dates',
template='plotly_white',
annotations=[
dict(
text=annotation_text,
xref='paper', yref='paper',
x=-.02, y=1.16,
showarrow=False,
bgcolor='lightyellow',
font=dict(size=12)
)
] if annotation_text else []
)
fig.show()
[docs]
def get_elbow_curve(df: pd.DataFrame, scenario: str, ax=None,strategy_scenario_colors:dict=None):
"""
Plot elbow curve for a given scenario on the specified matplotlib axis.
"""
scenario = scenario.capitalize()
scenario_filters = {
'Uncoordinated': df.columns.str.contains('uncoordinated'),
'Hybrid': df.columns.str.contains('hybrid'),
'Coordinated': df.columns.str.contains('coordinated') & ~df.columns.str.contains('uncoordinated'),
'V2g': df.columns.str.contains('v2g')
}
if strategy_scenario_colors is None:
colors = {
'Uncoordinated': '#2a9d8f',
'Hybrid': "#BE0A79",
'Coordinated': '#e76f51',
'V2g': "#122eac"
}
else:
colors=strategy_scenario_colors
filtered_df = df.loc[:, scenario_filters[scenario]]
scenario_totals = filtered_df.sum(axis=0)
scenario_labels = scenario_totals.index.str.extract(r'_(\d+)')[0].astype(int)
elbow_df = pd.DataFrame({
f'{scenario} %': scenario_labels,
'Total New Capacity (MW)': scenario_totals.values
}).sort_values(f'{scenario} %')
if ax is None:
fig, ax = plt.subplots(figsize=(10, 6))
sns.lineplot(
data=elbow_df,
x=f'{scenario} %',
y='Total New Capacity (MW)',
marker='o',
color=colors[scenario],
linewidth=2.5,
ax=ax,
label=scenario
)
xlabel = 'EV Charging Share (%)' if scenario != 'Hybrid' else 'Coordinated Charging Share (%)'
ax.set_xlabel(xlabel, fontsize=12)
ax.set_ylabel('Total New VRE Capacity (MW)', fontsize=12)
ax.tick_params(axis='both', labelsize=10)
ax.grid(True, linestyle='--', linewidth=0.5, alpha=0.7)
if scenario == 'Hybrid':
ax2 = ax.twiny()
ax2.set_xlim(ax.get_xlim())
x_vals = elbow_df[f'{scenario} %'].values
ax2.set_xticks(x_vals)
ax2.set_xticklabels([100 - x for x in x_vals])
ax2.set_xlabel('Uncoordinated Charging Share (%)', fontsize=12)
ax2.tick_params(axis='x', labelsize=10)
sns.despine()
[docs]
def get_combo_elbow_plot(new_capacity_df: pd.DataFrame,
font_family:str='serif',
strategy_scenario_colors:dict=None):
"""
Generate a combo plot of elbow curves comparing EV charging strategies,
with shared y-axis scaled to show all curves clearly.
"""
sns.set_theme(style="whitegrid")
# Create subplots
fig, (ax_combined, ax_v2g,ax_hybrid) = plt.subplots(1, 3, figsize=(21,6), sharey=True)
# Plot shared-axis curves
for scenario in ['Uncoordinated', 'Coordinated']:
get_elbow_curve(new_capacity_df, scenario=scenario, ax=ax_combined,strategy_scenario_colors=strategy_scenario_colors)
ax_combined.set_title('UnCoordinated vs Coordinated Charging', fontsize=14, weight='bold')
# Plot V2g separately
get_elbow_curve(new_capacity_df, scenario='V2g', ax=ax_v2g,strategy_scenario_colors=strategy_scenario_colors)
ax_v2g.set_title('Vehicle to Grid Charging Strategy', fontsize=14, weight='bold')
# Plot Hybrid separately
get_elbow_curve(new_capacity_df, scenario='Hybrid', ax=ax_hybrid,strategy_scenario_colors=strategy_scenario_colors)
ax_hybrid.set_title('Hybrid Charging Strategy', fontsize=14, weight='bold')
ax_hybrid.text(
0.75, 0.95,
"Hybrid charging shows\n100% EV penetration\nfor all data points",
transform=ax_hybrid.transAxes,
fontsize=10,
verticalalignment='top',
bbox={'boxstyle': 'round,pad=0.9', 'fc': 'lightgrey', 'ec': 'lightgrey', 'alpha': 0.5},
fontfamily=font_family
)
# Remove y-axis ticks on right subplot for cleanliness
ax_hybrid.set_ylabel('')
ax_hybrid.yaxis.set_tick_params(labelleft=False)
# Set font family for matplotlib figure
for ax in [ax_combined, ax_v2g, ax_hybrid]:
for label in (ax.get_xticklabels() + ax.get_yticklabels()):
label.set_fontfamily(font_family)
ax.title.set_fontfamily(font_family)
ax.xaxis.label.set_fontfamily(font_family)
ax.yaxis.label.set_fontfamily(font_family)
plt.tight_layout()
plt.show()
return fig
[docs]
def plot_load_profiles(network_names: list,
network_dict: Dict[str, pypsa.Network],
resample_freq:str,
aggregation_method:str='mean',
save_to="vis",
font_family='serif'):
"""
Plot load profiles for multiple networks on the same figure. Ideally give a list of network with same runtag.
Args:
network_names (list): List of network names. Ideally give a list of network with same runtag.
network_dict (dict): Dictionary containing PyPSA network objects.
resample_freq (str): Resampling frequency (e.g., 'H', 'D', 'W').
Aggregation_method (str): Aggregation method (mean, median, sum, max, min).
save_to (str): Directory to save the plots.
"""
network_names = sorted(network_names)
save_to = Path(save_to)
save_to.mkdir(parents=True, exist_ok=True)
annotation_text=get_annotation_text(resample_freq, aggregation_method)
# Create a single figure for all networks
fig = go.Figure()
# Extract relevant information from the network name
network_name=network_names[0]
runtag = network_name.split('_')[5]
# Build title prefix dynamically
title_prefix = f"runs for {runtag}"
for network_name in network_names:
if "hybrid" in network_name.split('_')[3]:
charging_strategy = network_name.split('_')[3]+ network_name.split('_')[4]+network_name.split('_')[5]
ev_penetration = network_name.split('_')[6]
else:
charging_strategy = network_name.split('_')[3]
ev_penetration = network_name.split('_')[4]
# Extract load data
load_data = network_dict[network_name].loads_t.p_set
# Select a specific load bus (replace 'your_load_bus' with the actual load bus name)
load_bus = 'BC ELC Load'
bc_load = load_data[load_bus]
# Sum all load profiles to create a combined load profile
charging_scenario_total_load = load_data.sum(axis=1)
# Resample the data
charging_scenario_total_load = get_resampled_data(charging_scenario_total_load,resample_freq,aggregation_method)
bc_load = bc_load.resample(resample_freq).mean()
# Add trace for each network
fig.add_trace(go.Scatter(
x=charging_scenario_total_load.index,
y=charging_scenario_total_load,
mode='lines',
name=charging_strategy+'-'+ev_penetration, # Retaining detailed title info in legend
line=dict(width=2)
))
# Add base load (only once)
fig.add_trace(go.Scatter(
x=bc_load.index,
y=bc_load,
mode='lines',
name='BC Base Load',
line=dict(color='black', width=2, dash='dot')
))
# Update layout
fig.update_layout(
title=f"Load Profiles | {title_prefix}",
xaxis_title='Time',
yaxis_title='Load (MW)',
legend=dict(title='Load Profiles'),
template='plotly_white',
annotations=[
dict(
text=annotation_text,
xref='paper', yref='paper',
x=0.002, y=1.16,
showarrow=False,
bgcolor='lightyellow',
font=dict(size=12)
)
] if annotation_text else []
)
fig.update_layout(font=dict(family=font_family))
# Save the figure
fig.write_html(save_to / 'Load_profiles_comparison.html')
utils.print_update(level=1,message=f"plot saved to {save_to / 'Load_profiles_comparison.html'} ")
# Show the figure
fig.show()
[docs]
def plot_generators(unique_tag: str,
network_names: list,
network_dict: Dict[str, pypsa.Network],
resample_freq: str = 'D',
save_to:str="vis",
font_family:str='serif'):
"""
Plot generator power output for different networks. Networks provided are ideally with same runtag.
Args:
unique_tag (str): Unique tag to filter generators.
network_names (list): List of network names. Ideally with same runtag
network_dict (dict): Dictionary of networks.
resample_freq (str): Resampling frequency for the data. "D" for daily, "h" for hourly, "W" for weekly, "M" for monthly, "Qe" for quarterly.
"""
save_to = Path(save_to)
save_to.mkdir(parents=True, exist_ok=True)
resample_freq=resample_freq.lower()
generator_power_output_daily_peak = pd.DataFrame()
if unique_tag.upper() == "NEW":
unique_tag = "New"
title_prefix = "| New Resource Options (Expansion)"
elif unique_tag.upper() == "CFP":
unique_tag = "CFP"
title_prefix = "| Call for Power Resources (Committed)"
elif unique_tag.upper() == "BACKSTOP":
unique_tag = "Backstop"
title_prefix = "| Backstop "
else:
title_prefix = f"with keywords - '{unique_tag}'"
for network_name in network_names:
# charging_strategy = network_name.split('_')[3]
# ev_penetration = network_name.split('_')[4]
filtered_gens_with_unique_tag_ts_df = network_dict[network_name].generators_t.p.loc[:, network_dict[network_name].generators_t.p.columns.str.contains(unique_tag).tolist()]
# Resample and sum the daily peaks
daily_peaks = filtered_gens_with_unique_tag_ts_df.resample(resample_freq).max().sum(axis=1)
generator_power_output_daily_peak[network_name] = daily_peaks
# Remove the column if all values are zero
if (daily_peaks == 0).all():
generator_power_output_daily_peak.drop(columns=[network_name], inplace=True)
pypsabc_utils.print_update(level=2,message=f"Backstop usage null for {network_name} ")
# Create a Plotly figure
# Construct annotation text (skip aggregation if resampling is hourly)
annotation_text = ''
resample_freq_suffix = resample_labels.get(resample_freq, '')
annotation_text += f'{resample_freq_suffix} Peak served'
fig = px.line(generator_power_output_daily_peak, title=f"Generators {title_prefix}") # Retaining detailed title info in legend)
# Update legend names by stripping the prefix 'pypsa_n_2021_'
fig.for_each_trace(lambda trace: trace.update(name=trace.name[len('pypsa_n_2021_'):]))
# Update the layout
fig.update_layout(
xaxis_title='Time',
yaxis_title='Power (MW)',
# showlegend=True,
annotations=[
dict(
text=annotation_text,
xref='paper', yref='paper',
x=0.005, y=1.16,
showarrow=False,
bgcolor='lightyellow',
font=dict(size=12)
)
] if annotation_text else []
)
fig.update_layout(font=dict(family=font_family))
# Save the figure
fig.write_html(save_to / f'Generations_for_{unique_tag}.html')
utils.print_update(level=1,message=f"plot saved to {save_to / f'Generations_for_{unique_tag}.html'} ")
# Show the figure
fig.show()
[docs]
def calculate_net_load(network:pypsa.Network,coord_type:str)->list:
ror_gen = pypsabc_utils.get_generators(network, 'RoR')
pv_gen = pypsabc_utils.get_generators(network, 'PV') + pypsabc_utils.get_generators(network, 'Solar')
wind_gen = pypsabc_utils.get_generators(network, 'Wind')
vre_gen = ror_gen + pv_gen + wind_gen
if coord_type == 'uncoordinated':
load = network.loads_t['p_set'].sum(axis=1)
elif coord_type == 'coordinated':
charger_cols = network.links.index[network.links.index.str.contains('Charger')].to_list()
load = network.loads_t.p_set['BC ELC Load'] + network.links_t.p0[charger_cols].sum(axis=1)
elif coord_type =='hybrid':
charger_cols = network.links.index[network.links.index.str.contains('Charger')].to_list()
load = network.loads_t.p_set['BC ELC Load'] + network.links_t.p0[charger_cols].sum(axis=1)+network.loads_t['p_set'].sum(axis=1)
elif coord_type == 'v2g':
charger_cols = network.links.index[network.links.index.str.contains('Charger')].to_list()
discharger_cols = network.links.index[network.links.index.str.contains('Discharger')].to_list()
load = network.loads_t.p_set['BC ELC Load'] + network.links_t.p0[charger_cols].sum(axis=1) - network.links_t.p0[discharger_cols].sum(axis=1)
net_load:list = load - vre_gen
return net_load
[docs]
def get_investment_comparison_plot(filtered_gens: pd.DataFrame,
gen_tag: str = '',
save_to: str = "vis",
font_family:str='serif'):
from pathlib import Path
import plotly.graph_objects as go
import plotly.express as px
save_to = Path(save_to)
save_to.mkdir(parents=True, exist_ok=True)
# Reset index to turn Generator into a column
filtered_gens_df = filtered_gens.reset_index()
# Sort by total capacity (sum across scenarios)
filtered_gens_df['total'] = filtered_gens_df.iloc[:, 1:].sum(axis=1)
filtered_gens_df = filtered_gens_df.sort_values(by='total', ascending=True)
filtered_gens_df = filtered_gens_df.drop(columns='total') # optional
# Define color map
color_map = px.colors.qualitative.Set1
# Create figure
fig = go.Figure()
for i, scenario in enumerate(filtered_gens_df.columns[1:]): # Skip 'Generator'
fig.add_trace(go.Bar(
y=filtered_gens_df['Generator'],
x=filtered_gens_df[scenario],
name=scenario,
orientation='h',
marker=dict(color=color_map[i % len(color_map)], line=dict(width=1)),
opacity=0.85
))
fig.update_layout(
# title='Filtered Generators Capacity (p_nom_opt)',
barmode='group',
bargap=0.2,
bargroupgap=0.1,
xaxis=dict(
title='Capacity (MW)',
showgrid=True,
gridcolor='lightgrey',
zeroline=False,
showline=True,
),
yaxis=dict(
title='',
automargin=True,
tickfont=dict(size=12),
showgrid=False,
),
margin=dict(l=180, r=40, t=80, b=80),
hovermode='y unified',
legend=dict(
title='Scenario',
traceorder='normal',
orientation='h',
y=1.1,
yanchor='bottom',
x=0.5,
xanchor='center'
),
template='plotly_white',
)
output_path = save_to / f"investment_comparison_{gen_tag}.html"
fig.update_layout(font=dict(family=font_family))
fig.write_html(output_path)
print(f"Plot saved to: {output_path}")
fig.show() # Comment this out if running headlessly
[docs]
def plot_selected_generators_power_output(search_keyword:str,
network_name: str,
network_dict: Dict[str,pypsa.Network],
save_to:str='vis',
font_family:str='serif'
):
"""
Plots the power output of 'New' generators for a given network.
Parameters:
- network_dict (dict): Dictionary containing PyPSA networks.
- run_tag (int): The run tag to identify the network.
- network_key (str): The key of the network to analyze (e.g., 'pypsa_n_2021_uncoordinated_100_{run_tag}').
"""
known_keywords={
'new':'New',
'cfp':'CFP',
'cfp24':'CFP24',
'backstop':'Backstop'}
save_to = Path(save_to)
save_to.mkdir(parents=True, exist_ok=True)
if search_keyword.lower() in known_keywords:
search_keyword = known_keywords.get(search_keyword.lower(), search_keyword)
# Get all generators that contain 'New' in their name
generator_names = network_dict[network_name].generators[
network_dict[network_name].generators.index.str.contains(search_keyword)
].index
charging_strat=network_name.split('_')[3]
penetration=network_name.split('_')[4]
run_tag=network_name.split('_')[5]
# Create an interactive figure
fig = go.Figure()
# Loop through the generators and add a trace for each
for generator_name in generator_names:
# Extract the time series data for the current generator
time_series = network_dict[network_name].generators_t.p[generator_name]
# Add the time series as a line trace to the figure
fig.add_trace(go.Scatter(
x=time_series.index, # Time index
y=time_series.values, # Power output values
mode='lines', # Line plot
name=generator_name, # Name the trace by generator name
line=dict(width=2), # Set line width
))
# Update layout for better appearance
fig.update_layout(
title=f"Power Output of Generators| {search_keyword} | {charging_strat} {penetration}% | Run: {run_tag}",
xaxis_title="Time",
yaxis_title="Power Output (MW)",
template="plotly_white", # Clean background
hovermode="x unified", # Show hover information for all data points in a vertical line
xaxis=dict(showgrid=True, gridcolor='lightgrey'),
yaxis=dict(showgrid=True, gridcolor='lightgrey'),
margin=dict(l=40, r=40, t=60, b=60), # Adjust margins for readability
legend=dict(
title=search_keyword.capitalize() if isinstance(search_keyword, str) else str(search_keyword),
orientation='h', # Horizontal layout
y=-0.3, # Move below the plot
yanchor='top',
x=0.5,
xanchor='center'
),
)
fig.update_layout(font=dict(family=font_family))
# Show the interactive plot
fig.show()
fig.write_html(f'vis/Generators_profile_{search_keyword}_{run_tag}.html')
[docs]
def get_network_optimized_installed_capacity_comparison_plot(run_tag:str,
ev_scenario_tag:str,
ev_penetration:int,
network_features:dict,
font_family:str='serif'):
"""
Dictionary of snapshots of different generator profiles.
"""
# Extract network names and their total new installed capacities
network_names = list(network_features.keys())
network_features[f'pypsa_n_2021_{ev_scenario_tag}_{str(ev_penetration)}_{run_tag}']['total_new_installed_capacity']
new_installed_capacities = [features['total_new_installed_capacity'] / 1E3 for features in network_features.values()] # Convert to GW
# Remove 'pypsa_n_2021_' from the network names
network_names_cleaned1 = [name.replace('pypsa_n_2021_', '') for name in network_names]
network_names_cleaned2 = [name.replace('_'+str(run_tag), '') for name in network_names_cleaned1]
network_names_cleaned2 = sorted(network_names_cleaned2)
# network_names_cleaned2.sort()
# Create a DataFrame
df = pd.DataFrame({
'Network Name': network_names_cleaned2, # Use cleaned names
'New Installed Capacity (GW)': new_installed_capacities
})
# Dynamically adjust figure height based on number of scenarios
base_height = 400
per_scenario_height = 40
min_height = 400
max_height = 1200
fig_height = min(max(base_height + per_scenario_height * len(network_names_cleaned2), min_height), max_height)
# Create an interactive bar chart with swapped axes using Plotly Express
fig = px.bar(
df,
y='Network Name', # Network names are on the y-axis
x='New Installed Capacity (GW)', # Capacity on the x-axis
title='Optimized New Installed Capacity for Each Network',
labels={'New Installed Capacity (GW)': 'Capacity (GW)', 'Network Name': 'Network Names'},
color='Network Name', # Optional: Add color differentiation per network
template='plotly_white', # Use a clean template
height=fig_height
)
# Improve layout for better readability with swapped axes
fig.update_layout(
xaxis=dict(
title='Capacity (GW)',
showgrid=True, # Show gridlines for the x-axis
gridcolor='lightgrey', # Use light gridlines for subtle separation
),
yaxis=dict(
title='Scenarios',
tickangle=0, # Make y-axis labels horizontal
showgrid=False, # Hide gridlines on the y-axis
),
margin=dict(l=100, r=40, t=60, b=100), # Adjust margins for better use of space
hovermode='y unified', # Show hover info for all bars in a vertical line
showlegend=False, # Optional: Disable legend for a cleaner look
)
fig.update_layout(font=dict(family=font_family))
# Show the interactive plot
fig.show()
fig.write_html('vis/Installed_capacity_scenarios.html')
utils.print_update(level=1,message= "Plot save dto 'vis/Installed_capacity_scenarios.html' ")
[docs]
def get_net_load_curves(network_dict:dict,
scenario_keyword:str,
run_tag:str,
font_family:str='serif'):
"""_summary_
Args:
scenario_keyword (str): 'hybrid_coord_uncoord','coordinated','uncoordinated','v2g'
"""
import numpy as np
# Automatically extract available scenario names from network_dict
# If scenario_keyword is 'coordinated', also include 'hybrid_coord_uncoord'
if scenario_keyword == 'coordinated':
scenarios = [name for name in network_dict if ('coordinated' in name and 'uncoordinated' not in name)]
else:
scenarios = [name for name in network_dict if scenario_keyword in name]
# Helper to extract penetration value
def get_penetration(name):
parts = name.split("_")
if "hybrid" in parts:
# For hybrid_coord_uncoord_X, penetration is at index after 'hybrid_coord_uncoord'
idx = parts.index("hybrid") # should be "hybrid"
# The next part is "coord", then "uncoord", then the penetration value
return 'hybrid',int(parts[idx + 3])
else:
# fallback: try to find the first integer after 'coordinated', 'uncoordinated', or 'v2g'
for key in ["coordinated", "uncoordinated", "v2g"]:
if key in parts:
idx = parts.index(key)
return key,int(parts[idx + 1])
scenarios = sorted(scenarios, key=get_penetration)
# Calculate net loads for each hybrid scenario
net_loads = {}
for name in scenarios:
net_loads[name] = calculate_net_load(network_dict[name], "hybrid")
# Plot all hybrid scenarios found
fig, ax = plt.subplots(figsize=(9, 5))
colors = plt.cm.inferno(np.linspace(0, 1, len(scenarios)))
for color, name in zip(colors, scenarios):
scenario,penetration = get_penetration(name)
if scenario.lower()=='hybrid':
label = f"Coord {penetration}%, Uncoord {100-penetration}%"
else:
label = f"{scenario} {penetration}%"
sorted_net_load = pd.Series(net_loads[name]).sort_values(ascending=False).values
ax.plot(sorted_net_load, label=label, color=color, linewidth=2)
ax.set_title(f'Net Load Duration Curves for {scenario.capitalize()} Scenarios', fontsize=14)
ax.set_ylabel('Net Load (MW)', fontsize=12)
ax.set_xlabel('Hours', fontsize=12)
ax.grid(True, axis='y', linestyle='--', linewidth=0.3, alpha=0.7)
ax.grid(True, axis='x', linestyle='--', linewidth=0.3, alpha=0.7)
# ax.set_xlim([0, len(next(iter(net_loads.values())))])
legend = ax.legend(frameon=False, loc='upper right', fontsize=10, title_fontsize=11)
legend.get_frame().set_facecolor('white')
legend.get_frame().set_alpha(0.9)
plt.rcParams.update({'font.family': font_family})
#
plt.tight_layout()
plt.savefig(f'vis/NLC_{run_tag}_{scenario_keyword}_scenarios.png', dpi=300)
plt.show()
[docs]
def plot_curtailment_profile(curtailment_profile:pd.DataFrame,
show:bool=True,
font_family:str='serif'):
# Assuming 'curtailment' is a DataFrame indexed by time, columns = generator names
fig = go.Figure()
# Add a line for each generator
for gen in curtailment_profile.columns:
fig.add_trace(go.Scatter(
x=curtailment_profile.index,
y=curtailment_profile[gen],
mode='lines',
name=gen
))
fig.update_layout(
title="Wind Curtailment Over Time for New VRE Generators",
xaxis_title="Time",
yaxis_title="Curtailment (MW)",
hovermode="x unified",
legend_title="Generators",
template="plotly_white",
font=dict(family=font_family)
)
if show:
fig.show()
[docs]
def get_generators_curtailment(n:pypsa.Network,
generators_df:pd.DataFrame,
show:bool=True,
font_family:str='serif'):
# maximum possible output = nominal capacity × availability profile
potential_output = (
n.generators_t.p_max_pu[generators_df.index]
* n.generators.p_nom_opt[generators_df.index]
)
actual_output = n.generators_t.p[generators_df.index]
curtailment_generators_df = potential_output - actual_output
curtailment_generators_df[curtailment_generators_df < 0] = 0 # remove numerical noise
total_curtailment = curtailment_generators_df.sum(axis=0)
# Create a DataFrame for curtailment in GWh
curtailment_gwh_df = pd.DataFrame({
"Curtailment [GWh]": total_curtailment / 1e3
})
# plot total curtailment
curtailment_gwh_df.plot(kind='barh', legend=False)
plt.xlabel("Curtailment [GWh]")
plt.title("Total Curtailment by New VRE Generator")
plt.tight_layout()
plt.rcParams.update({'font.family': font_family})
if show:
plt.show()
plot_curtailment_profile(curtailment_generators_df,show)
return curtailment_gwh_df,curtailment_generators_df