Plot statistics

add_second_xaxis(data, ax, label, **kwargs)

Add a secondary X-axis to the plot.

Parameters:

Name	Type	Description	Default
data	Series	The data to plot. Its values will be plotted on the secondary X-axis.	required
ax	Axes	The main matplotlib Axes object.	required
label	str	The label for the secondary X-axis.	required
**kwargs		Optional keyword arguments for plot styling.	{}

Source code in workflow/scripts/plot_statistics.py

def add_second_xaxis(data: pd.Series, ax, label, **kwargs):
    """
    Add a secondary X-axis to the plot.

    Args:
        data (pd.Series): The data to plot. Its values will be plotted on the secondary X-axis.
        ax (matplotlib.axes.Axes): The main matplotlib Axes object.
        label (str): The label for the secondary X-axis.
        **kwargs: Optional keyword arguments for plot styling.
    """
    defaults = {"color": "red", "text_offset": 0.5, "markersize": 8, "fontsize": 9}
    kwargs.update(defaults)

    ax2 = ax.twiny()
    # # y_pos creates a sequence of integers (e.g., [0, 1, 2, 3]) to serve as distinct vertical positions
    # for each data point on the shared Y-axis. This is necessary because data.values are plotted
    # horizontally on the secondary X-axis (ax2), requiring vertical separation for clarity.
    y_pos = range(len(data))

    ax2.plot(
        data.values,
        y_pos,
        marker="o",
        linestyle="",
        color=kwargs["color"],
        markersize=kwargs["markersize"],
        label="Generation Share (%)",
    )

    for i, val in enumerate(data.values):
        ax2.text(
            val + kwargs["text_offset"],
            i,
            f"{val:.1f}%",
            color=kwargs["color"],
            va="center",
            ha="left",
            fontsize=kwargs["fontsize"],
        )

    ax2.set_xlim(left=0)
    ax2.set_xlabel(label)
    ax2.grid(False)
    ax2.tick_params(axis="x", labelsize=kwargs["fontsize"])  # Remove color setting for ticks

    return ax2

filter_small_caps(n, threshold=100)

Drop small capacities for plotting (eliminate numerical zeroes) -> this would be more robust based on the objective cost tolerance

Parameters:

Name	Type	Description	Default
n	Network	the pypsa network to remove small comps from	required
threshold	int	the removal threshold. Defaults to 100.	100

Source code in workflow/scripts/plot_statistics.py

def filter_small_caps(n: pypsa.Network, threshold=100):
    """Drop small capacities for plotting (eliminate numerical zeroes)
    -> this would be more robust based on the objective cost tolerance

    Args:
        n (pypsa.Network): the pypsa network to remove small comps from
        threshold (int, optional): the removal threshold. Defaults to 100.
    """
    for c in ["links", "generators", "stores", "storage_units"]:
        attr = "e_nom_opt" if c == "stores" else "p_nom_opt"
        comp = getattr(n, c)
        mask = comp[attr] > threshold
        comp = comp.loc[mask]
        setattr(n, c, comp)

fix_load_carriers(n, config)

Set unspecified load carriers to load

Parameters:

Name	Type	Description	Default
n	Network	The PyPSA network instance.	required
config	dict	the plotting config	required

Source code in workflow/scripts/plot_statistics.py

def fix_load_carriers(n: pypsa.Network, config: dict):
    """Set unspecified load carriers to load

    Args:
        n (pypsa.Network): The PyPSA network instance.
        config (dict): the plotting config
    """
    mask = n.loads.query("carrier==''").index
    n.loads.loc[mask, "carrier"] = "Load"
    n.carriers.loc["Load", ["nice_name", "color"]] = (
        "Load",
        config["tech_colors"]["electric load"],
    )

plot_capacity_factor(cf_filtered, theo_cf_filtered, ax, colors, **kwargs)

Plot actual and theoretical capacity factors for each technology.

Parameters:

Name	Type	Description	Default
cf_filtered	Series	Actual capacity factors indexed by technology.	required
theo_cf_filtered	Series	Theoretical capacity factors indexed by technology.	required
ax	Axes	The axis to plot on.	required
colors	dict	Color mapping for technologies.	required

Returns:

Type	Description
	matplotlib.axes.Axes: The axis with the plot.

Source code in workflow/scripts/plot_statistics.py

def plot_capacity_factor(
    cf_filtered: pd.Series, theo_cf_filtered: pd.Series, ax: axes.Axes, colors: dict, **kwargs
):
    """
    Plot actual and theoretical capacity factors for each technology.

    Args:
        cf_filtered (pd.Series): Actual capacity factors indexed by technology.
        theo_cf_filtered (pd.Series): Theoretical capacity factors indexed by technology.
        ax (matplotlib.axes.Axes): The axis to plot on.
        colors (dict): Color mapping for technologies.

    Returns:
        matplotlib.axes.Axes: The axis with the plot.
    """
    x_pos = range(len(cf_filtered))
    width = 0.35

    ax.barh(
        [i - width / 2 for i in x_pos],
        cf_filtered.values,
        width,
        color=[colors.get(tech, "lightgrey") for tech in cf_filtered.index],
        alpha=0.8,
        label="Actual CF",
    )
    ax.barh(
        [i + width / 2 for i in x_pos],
        theo_cf_filtered.values,
        width,
        color=[colors.get(tech, "lightgrey") for tech in theo_cf_filtered.index],
        alpha=0.4,
        label="Theoretical CF",
    )

    for i, (tech, cf_val) in enumerate(cf_filtered.items()):
        ax.text(
            cf_val + 0.01,
            i - width / 2,
            f"{cf_val:.2f}",
            va="center",
            ha="left",
            fontsize=8,
            bbox=dict(boxstyle="round,pad=0.2", facecolor="white", alpha=0.8),
        )
        theo_val = theo_cf_filtered.get(tech, 0)
        ax.text(
            theo_val + 0.01,
            i + width / 2,
            f"{theo_val:.2f}",
            va="center",
            ha="left",
            fontsize=8,
            bbox=dict(boxstyle="round,pad=0.2", facecolor="white", alpha=0.5),
        )

    ax.set_yticks(list(x_pos))
    ax.set_yticklabels(cf_filtered.index)
    ax.set_xlabel("Capacity Factor")
    ax.set_xlim(0, max(cf_filtered.max(), theo_cf_filtered.max()) * 1.1)
    ax.grid(False)
    ax.legend()

    return ax

plot_province_peakload_capacity(df_plot, bar_cols, color_list, outp_dir)

Plot province peak load vs installed capacity by technology.

Parameters:

Name	Type	Description	Default
df_plot		DataFrame with provinces as index, columns as technologies and 'Peak Load'.	required
bar_cols		List of technology columns to plot as bars.	required
color_list		List of colors for each technology.	required
outp_dir		Output directory for saving the figure.	required

Source code in workflow/scripts/plot_statistics.py

def plot_province_peakload_capacity(df_plot, bar_cols, color_list, outp_dir):
    """
    Plot province peak load vs installed capacity by technology.

    Args:
        df_plot: DataFrame with provinces as index, columns as technologies and 'Peak Load'.
        bar_cols: List of technology columns to plot as bars.
        color_list: List of colors for each technology.
        outp_dir: Output directory for saving the figure.
    """
    fig, ax = plt.subplots(figsize=(14, 8))

    df_plot[bar_cols].plot(kind="barh", stacked=True, ax=ax, color=color_list, alpha=0.8)
    # Plot peak load as red vertical line
    for i, prov in enumerate(df_plot.index):
        ax.plot(
            df_plot.loc[prov, "Peak Load"],
            i,
            "r|",
            markersize=18,
            label="Peak Load" if i == 0 else "",
        )
    ax.set_xlabel("Capacity [GW]")
    ax.set_ylabel("Province")
    ax.set_title("Peak Load vs Installed Capacity by Province")
    ax.grid(False)
    # Only keep one Peak Load legend
    handles, labels = ax.get_legend_handles_labels()
    seen = set()
    new_handles, new_labels = [], []
    for h, l in zip(handles, labels):
        if l not in seen:
            new_handles.append(h)
            new_labels.append(l)
            seen.add(l)
    ax.legend(new_handles, new_labels, loc="best")
    fig.tight_layout()
    fig.savefig(os.path.join(outp_dir, "province_peakload_capacity.png"))

plot_static_per_carrier(ds, ax, colors, drop_zero_vals=True, add_labels=True)

Generic function to plot different statics

Parameters:

Name	Type	Description	Default
ds	Series	the data to plot	required
ax	Axes	plotting axes	required
colors	Series	colors for the carriers	required
drop_zero_vals	bool	Drop zeroes from data. Defaults to True.	True
add_labels	bool	Add value labels on bars. If None, reads from config. Defaults to None.	True

Source code in workflow/scripts/plot_statistics.py

def plot_static_per_carrier(
    ds: pd.Series, ax: axes.Axes, colors: pd.Series, drop_zero_vals=True, add_labels=True
):
    """Generic function to plot different statics

    Args:
        ds (pd.Series): the data to plot
        ax (matplotlib.axes.Axes): plotting axes
        colors (pd.Series): colors for the carriers
        drop_zero_vals (bool, optional): Drop zeroes from data. Defaults to True.
        add_labels (bool, optional): Add value labels on bars. If None, reads from config. Defaults to None.
    """
    if not ax:
        fig, ax = plt.subplots()
    else:
        fig = ax.get_figure()

    if drop_zero_vals:
        ds = ds[ds != 0]
    ds = ds.dropna()
    c = colors[ds.index.get_level_values("carrier")]
    ds = ds.pipe(rename_index)
    label = f"{ds.attrs['name']} [{ds.attrs['unit']}]"
    ds.plot.barh(color=c.values, xlabel=label, ax=ax)
    if add_labels:
        ymax = ax.get_xlim()[1] * 1.05
        for i, (index, value) in enumerate(ds.items()):
            align = "left"
            txt = f"{value:.2f}" if value <= 100 else f"{value:.1e}"
            ax.text(ymax, i, txt, va="center", ha=align, fontsize=8)
        # # Add outer y-ticks at the right y-axis frame
        # ax.tick_params(axis="y", direction="out", right=True, left=False)
    ax.grid(axis="y")
    fig.tight_layout()

    return fig

prepare_capacity_factor_data(n, carrier)

Prepare Series for actual and theoretical capacity factors per technology. Args: n (pypsa.Network): The PyPSA network instance. carrier (str): The carrier for which to prepare the data. Returns: cf_filtered: Series of actual capacity factors (index: nice_name) theo_cf_filtered: Series of theoretical capacity factors (index: nice_name)

Source code in workflow/scripts/plot_statistics.py

def prepare_capacity_factor_data(n: pypsa.Network, carrier: str):
    """
    Prepare Series for actual and theoretical capacity factors per technology.
    Args:
        n (pypsa.Network): The PyPSA network instance.
        carrier (str): The carrier for which to prepare the data.
    Returns:
        cf_filtered: Series of actual capacity factors (index: nice_name)
        theo_cf_filtered: Series of theoretical capacity factors (index: nice_name)
    """
    cf_data = n.statistics.capacity_factor(groupby=["carrier"]).dropna()
    if ("Link", "battery") in cf_data.index:
        cf_data.loc[("Link", "battery charger")] = cf_data.loc[("Link", "battery")]
        cf_data.drop(index=("Link", "battery"), inplace=True)
    cf_data = cf_data.groupby(level=1).mean()

    # Theoretical capacity factor
    gen = n.generators.copy()
    p_max_pu = n.generators_t.p_max_pu
    gen["p_nom_used"] = gen["p_nom_opt"].fillna(gen["p_nom"])
    weighted_energy_per_gen = (p_max_pu * gen["p_nom_used"]).sum()
    gen["weighted_energy"] = weighted_energy_per_gen

    gen["nice_name"] = gen["carrier"].map(
        lambda x: n.carriers.loc[x, "nice_name"] if x in n.carriers.index else x
    )
    grouped_energy = gen.groupby("nice_name")["weighted_energy"].sum()
    grouped_capacity = gen.groupby("nice_name")["p_nom_used"].sum()
    theoretical_cf_weighted = grouped_energy / grouped_capacity / len(n.snapshots)

    # Only keep technologies present in both actual and theoretical CF
    common_techs = cf_data.index.intersection(theoretical_cf_weighted.index)
    cf_filtered = cf_data.loc[common_techs]
    theo_cf_filtered = theoretical_cf_weighted.loc[cf_filtered.index]
    # Todo: use config nondispatchable_techs
    non_zero_mask = (cf_filtered != 0) & (theo_cf_filtered != 0)
    cf_filtered = cf_filtered[non_zero_mask]
    theo_cf_filtered = theo_cf_filtered[non_zero_mask]
    cf_filtered = cf_filtered.sort_values(ascending=True)
    theo_cf_filtered = theo_cf_filtered.loc[cf_filtered.index]

    return cf_filtered, theo_cf_filtered

prepare_province_peakload_capacity_data(n, attached_carriers=None)

Prepare DataFrame for province peak load and installed capacity by technology.

Returns:

Name	Type	Description
df_plot		DataFrame with provinces as index, columns as technologies and 'Peak Load'.
bar_cols		List of technology columns to plot as bars.
color_list		List of colors for each technology.

Source code in workflow/scripts/plot_statistics.py

def prepare_province_peakload_capacity_data(n, attached_carriers=None):
    """
    Prepare DataFrame for province peak load and installed capacity by technology.

    Returns:
        df_plot: DataFrame with provinces as index, columns as technologies and 'Peak Load'.
        bar_cols: List of technology columns to plot as bars.
        color_list: List of colors for each technology.
    """
    # Calculate peak load per province
    load = n.loads.copy()
    load["province"] = load["bus"].map(n.buses["location"])
    peak_load = n.loads_t.p_set.groupby(load["province"], axis=1).sum().max()
    peak_load = peak_load / PLOT_CAP_UNITS  # ensure peak load is in GW

    # Calculate installed capacity per province and technology using optimal_capacity
    ds = n.statistics.optimal_capacity(groupby=["location", "carrier"]).dropna()
    valid_components = ["Generator", "StorageUnit", "Link"]
    ds = ds.loc[ds.index.get_level_values(0).isin(valid_components)]
    if ("Link", "battery") in ds.index:
        ds.loc[("Link", "battery charger")] = ds.loc[("Link", "battery")]
        ds = ds.drop(index=("Link", "battery"))
    if "stations" in ds.index.get_level_values(2):
        ds = ds.drop("stations", level=2)
    if "load shedding" in ds.index.get_level_values(2):
        ds = ds.drop("load shedding", level=2)
    ds = ds.groupby(level=[1, 2]).sum()
    ds.index = pd.MultiIndex.from_tuples(
        [
            (prov, n.carriers.loc[carrier, "nice_name"] if carrier in n.carriers.index else carrier)
            for prov, carrier in ds.index
        ],
        names=["province", "nice_name"],
    )
    cap_by_prov_tech = ds.unstack(level=-1).fillna(0)
    cap_by_prov_tech = cap_by_prov_tech.abs() / PLOT_CAP_UNITS

    if "Battery Discharger" in cap_by_prov_tech.columns:
        cap_by_prov_tech = cap_by_prov_tech.drop(columns="Battery Discharger")
    if "AC" in cap_by_prov_tech.columns:
        cap_by_prov_tech = cap_by_prov_tech.drop(columns="AC")
    # Only keep columns in attached_carriers if provided
    if attached_carriers is not None:
        # Ensure nice_name mapping for attached_carriers
        attached_nice_names = [
            n.carriers.loc[c, "nice_name"] if c in n.carriers.index else c
            for c in attached_carriers
        ]
        cap_by_prov_tech = cap_by_prov_tech[
            [c for c in cap_by_prov_tech.columns if c in attached_nice_names]
        ]

    # Merge peak load and capacity
    df_plot = cap_by_prov_tech.copy()
    df_plot["Peak Load"] = peak_load

    # Bar columns: exclude Peak Load, only keep nonzero
    bar_cols = [c for c in df_plot.columns if c != "Peak Load"]
    bar_cols = [c for c in bar_cols if df_plot[c].sum() > 0]
    color_list = [
        n.carriers.set_index("nice_name").color.get(tech, "lightgrey") for tech in bar_cols
    ]
    return df_plot, bar_cols, color_list

set_link_output_capacities(n, carriers)

Set link capacity to output and not input. PyPSA uses input link capacities but typically want to report output capacities (e.g MWel)

Parameters:

Name	Type	Description	Default
n	Network	The PyPSA network instance.	required
carriers	list	List of carrier names to adjust.	required

Returns: pd.DataFrame: the original link capacities.

Source code in workflow/scripts/plot_statistics.py

def set_link_output_capacities(n: pypsa.Network, carriers: list) -> pd.DataFrame:
    """Set link capacity to output and not input.
    PyPSA uses input link capacities but typically want to report output capacities (e.g MWel)

    Args:
        n (pypsa.Network): The PyPSA network instance.
        carriers (list): List of carrier names to adjust.
    Returns:
        pd.DataFrame: the original link capacities.
    """
    # Temporarily save original link capacities
    original_p_nom_opt = n.links.p_nom_opt.copy()

    # For links where bus1 is AC, multiply capacity by efficiency coefficient to get AC side capacity
    ac_links = n.links[n.links.bus1.map(n.buses.carrier).isin(carriers)].index
    n.links.loc[ac_links, "p_nom_opt"] *= n.links.loc[ac_links, "efficiency"]

    # ignore lossy link dummies
    pseudo_links = n.links.query("Link.str.contains('reversed') & capital_cost ==0 ").index
    n.links.loc[pseudo_links, "p_nom_opt"] = 0

    return original_p_nom_opt