[pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

Author: pre-commit-ci[bot]

Snakefile

@@ -43,8 +43,8 @@ CDIR = Path(cutout_dir).joinpath("" if run["shared_cutouts"] else RDIR)
 RESULTS = "results/" + RDIR
 
 run_prefix = config["run"]["prefix"]
-regret_scenarios   = ["AriadneDemand", "LowDemand"]
-horizons    = [2025, 2030, 2035]
+regret_scenarios = ["AriadneDemand", "LowDemand"]
+horizons = [2025, 2030, 2035]
 
 
 localrules:
@@ -1081,9 +1081,8 @@ rule regret_all:
             decision=config_provider("run", "name"),
             **config["scenario"],
         ),
-        elec_capa_comp_de_2025 = f"results/{run_prefix}/regret_plots/Ariadne_vs_LowDemand_LT/elec_capa_comp_de_2025.png",
-        elec_price_comp_de = f"results/{run_prefix}/regret_plots/Ariadne_vs_LowDemand/elec_price_comp_de.png",
-
+        elec_capa_comp_de_2025=f"results/{run_prefix}/regret_plots/Ariadne_vs_LowDemand_LT/elec_capa_comp_de_2025.png",
+        elec_price_comp_de=f"results/{run_prefix}/regret_plots/Ariadne_vs_LowDemand/elec_price_comp_de.png",
 
 
 rule regret_all_variables:
@@ -1103,23 +1102,24 @@ rule regret_plots_lt:
         planning_horizons=config_provider("scenario", "planning_horizons"),
         plotting=config_provider("plotting"),
     input:
-        networks = expand(
+        networks=expand(
             "results/{run}/{scenario}/networks/base_s_27__none_{year}.nc",
             run=run_prefix,
             scenario=regret_scenarios,
-            year=horizons
+            year=horizons,
         ),
         regret_variables=expand(
             "results/{run}/{scenario}/regret_variables/regret_variables_{scenario}_full.xlsx",
             run=run_prefix,
             scenario=regret_scenarios,
         ),
     output:
-        elec_capa_comp_de_2025 = f"results/{run_prefix}/regret_plots/Ariadne_vs_LowDemand_LT/elec_capa_comp_de_2025.png",
+        elec_capa_comp_de_2025=f"results/{run_prefix}/regret_plots/Ariadne_vs_LowDemand_LT/elec_capa_comp_de_2025.png",
         dir=directory(f"results/{run_prefix}/regret_plots/Ariadne_vs_LowDemand_LT"),
     script:
         "scripts/pypsa-de/regret_plots_lt.py"
 
+
 rule regret_plots:
     params:
         scenarios=get_scenarios(run),
@@ -1134,7 +1134,7 @@ rule regret_plots:
             **config["scenario"],
         ),
     output:
-        elec_price_comp_de = f"results/{run_prefix}/regret_plots/Ariadne_vs_LowDemand/elec_price_comp_de.png",
+        elec_price_comp_de=f"results/{run_prefix}/regret_plots/Ariadne_vs_LowDemand/elec_price_comp_de.png",
         dir=directory(f"results/{run_prefix}/regret_plots/Ariadne_vs_LowDemand"),
     script:
         "scripts/pypsa-de/regret_plots.py"

scripts/pypsa-de/regret_plots.py

@@ -1,27 +1,26 @@
-import logging
 import os
 import sys
 
-sys.path.append(os.path.abspath(os.path.dirname(__file__))) 
-sys.path.append(
-    os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
-)
+sys.path.append(os.path.abspath(os.path.dirname(__file__)))
+sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
+
+import collections
+import itertools
+import os
+import re
 
 import matplotlib.pyplot as plt
 import numpy as np
 import pandas as pd
 import pypsa
-import os, re, collections
-import itertools
-
 from _helpers import configure_logging, mock_snakemake
 
 groups = {
     "gas": ["gas CHP", "OCGT", "CCGT", "gas"],
     "heat vent": ["heat vent"],
     "water tanks": ["water tank", "water pit"],
-    "heat pump" : ["heat pump"],
-    "resistive heater" : ["resistive heater"],
+    "heat pump": ["heat pump"],
+    "resistive heater": ["resistive heater"],
     "biomass": ["biomass"],
     "lignite": ["lignite"],
     "coal": ["coal"],
@@ -31,18 +30,19 @@
     "offwind": ["offwind"],
 }
 
+
 def aggregate_by_keywords(opex_comp_agg, groups):
     """
     Aggregate rows in opex_comp_agg according to keyword groups.
-    
+
     Parameters
     ----------
     opex_comp_agg : pd.DataFrame
         DataFrame with row index as technology names.
     groups : dict
-        Keys = new aggregated name, 
+        Keys = new aggregated name,
         Values = list of substrings to match in the index.
-    
+
     Returns
     -------
     pd.DataFrame
@@ -57,7 +57,6 @@ def aggregate_by_keywords(opex_comp_agg, groups):
     return df_out
 
 
-
 if __name__ == "__main__":
     if "snakemake" not in globals():
         import os
@@ -72,21 +71,21 @@ def aggregate_by_keywords(opex_comp_agg, groups):
     configure_logging(snakemake)
     config = snakemake.config
     planning_horizons = snakemake.params.planning_horizons
-    scenarios   = ["AriadneDemand", "LowDemand"]
+    scenarios = ["AriadneDemand", "LowDemand"]
     tech_colors = snakemake.params.plotting["tech_colors"]
 
     # Nested dict: networks[year][scenario][decision] = Network
-    networks = collections.defaultdict(
-        lambda: collections.defaultdict(dict)
-    )
+    networks = collections.defaultdict(lambda: collections.defaultdict(dict))
 
     for fn in snakemake.input.regret_networks:
         parts = fn.split(os.sep)
 
         # scenario is the folder name 2 levels up
         scenario = parts[-3]
         if scenario not in scenarios:
-            raise ValueError(f"Unexpected scenario '{scenario}' in {fn}. Allowed: {scenarios}")
+            raise ValueError(
+                f"Unexpected scenario '{scenario}' in {fn}. Allowed: {scenarios}"
+            )
 
         # extract year (4 digits before .nc)
         m = re.search(r"_(\d{4})\.nc$", fn)
@@ -121,17 +120,21 @@ def aggregate_by_keywords(opex_comp_agg, groups):
 
     for i, year in enumerate(years):
         for scenario, decision in itertools.product(scenarios, decisions):
-
-            n = networks[year][scenario][decision]    
-            lmps = n.buses_t.marginal_price.loc[:, 
-                    (n.buses.carrier == "AC") & (n.buses.index.str.startswith("DE"))]
-            lmps_sorted = pd.DataFrame(lmps.values.flatten(), columns=["lmp"]).sort_values(by="lmp", ascending=False)
-            lmps_sorted["percentage"] = np.arange(len(lmps_sorted)) / len(lmps_sorted) * 100
+            n = networks[year][scenario][decision]
+            lmps = n.buses_t.marginal_price.loc[
+                :, (n.buses.carrier == "AC") & (n.buses.index.str.startswith("DE"))
+            ]
+            lmps_sorted = pd.DataFrame(
+                lmps.values.flatten(), columns=["lmp"]
+            ).sort_values(by="lmp", ascending=False)
+            lmps_sorted["percentage"] = (
+                np.arange(len(lmps_sorted)) / len(lmps_sorted) * 100
+            )
 
             ax[i].plot(
-                lmps_sorted["percentage"], 
-                lmps_sorted["lmp"], 
-                label=f"{scenario}_{decision} (avg: {lmps_sorted['lmp'].mean():.2f})"
+                lmps_sorted["percentage"],
+                lmps_sorted["lmp"],
+                label=f"{scenario}_{decision} (avg: {lmps_sorted['lmp'].mean():.2f})",
             )
 
         ax[i].legend()
@@ -143,17 +146,15 @@ def aggregate_by_keywords(opex_comp_agg, groups):
     plt.savefig(snakemake.output.elec_price_comp_de, bbox_inches="tight")
     plt.close()
 
-
     # Print CO2 prices
- 
+
     # for i, year in enumerate(years):
     #     for scenario, decision in itertools.product(scenarios, decisions):
 
-    #         n = networks[year][scenario][decision]   
+    #         n = networks[year][scenario][decision]
 
     #         print(f"CO2 price for {year}, {scenario}, {decision}: {n.global_constraints.loc["CO2Limit", "mu"] + n.global_constraints.loc["co2_limit-DE", "mu"]}")
 
-
     # Plot OPEX
 
     kwargs = {
@@ -166,7 +167,9 @@ def aggregate_by_keywords(opex_comp_agg, groups):
     axes = axes.flatten()
 
     for i, year in enumerate(years):
-        opex_comp = pd.DataFrame(columns=["_".join(tup) for tup in itertools.product(scenarios, decisions)])
+        opex_comp = pd.DataFrame(
+            columns=["_".join(tup) for tup in itertools.product(scenarios, decisions)]
+        )
 
         # Collect OPEX for all scenario-decision combinations
         for scenario, decision in itertools.product(scenarios, decisions):
@@ -175,7 +178,8 @@ def aggregate_by_keywords(opex_comp_agg, groups):
             opex = (
                 n.statistics.opex(**kwargs)
                 .filter(like="DE")
-                .groupby("carrier").sum()
+                .groupby("carrier")
+                .sum()
                 .multiply(1e-9)  # to billion €
             )
             opex_comp[f"{scenario}_{decision}"] = opex
@@ -185,43 +189,59 @@ def aggregate_by_keywords(opex_comp_agg, groups):
         small_rows = opex_comp_agg.abs().max(axis=1) < 0.1
         other_row = opex_comp_agg[small_rows].sum(axis=0)
         opex_comp_agg = opex_comp_agg.loc[~small_rows]
-        opex_comp_agg.loc['Other'] = other_row
+        opex_comp_agg.loc["Other"] = other_row
 
         # Prepare labels with line breaks
-        labels = [col.replace('_', '\n') for col in opex_comp_agg.columns]
+        labels = [col.replace("_", "\n") for col in opex_comp_agg.columns]
 
         # Plot stacked bar
         ax = axes[i]
         bottom = np.zeros(len(opex_comp_agg.columns))
 
         for tech in opex_comp_agg.index:
             values = opex_comp_agg.loc[tech].values
-            ax.bar(labels, values, bottom=bottom, color=tech_colors.get(tech, '#333333'), label=tech)
+            ax.bar(
+                labels,
+                values,
+                bottom=bottom,
+                color=tech_colors.get(tech, "#333333"),
+                label=tech,
+            )
 
             # Add numbers in the middle, except for 'Other'
-            if tech != 'Other':
+            if tech != "Other":
                 for j, val in enumerate(values):
                     if val > 0:  # only if positive
                         ax.text(
-                            j, 
-                            bottom[j] + val/2,  # middle of the segment
-                            f'{val:.2f}',
-                            ha='center', va='center', fontsize=8, color='white'
+                            j,
+                            bottom[j] + val / 2,  # middle of the segment
+                            f"{val:.2f}",
+                            ha="center",
+                            va="center",
+                            fontsize=8,
+                            color="white",
                         )
 
             bottom += values
 
         # Add total sum labels on top of bars
         totals = opex_comp_agg.sum(axis=0)
         for j, total in enumerate(totals):
-            ax.text(j, total + total*0.02, f'{total:.2f}', ha='center', va='bottom', fontsize=10)
+            ax.text(
+                j,
+                total + total * 0.02,
+                f"{total:.2f}",
+                ha="center",
+                va="bottom",
+                fontsize=10,
+            )
 
         # Adjust y-limit
-        ax.set_ylim(0, max(totals)*1.08)
-        ax.set_ylabel('OPEX [billion €]')
-        ax.set_title(f'Stacked OPEX composition by technology, {year}')
+        ax.set_ylim(0, max(totals) * 1.08)
+        ax.set_ylabel("OPEX [billion €]")
+        ax.set_title(f"Stacked OPEX composition by technology, {year}")
 
     # Legend outside
-    axes[-1].legend(loc='upper left', bbox_to_anchor=(1,1))
-    plt.savefig(snakemake.output[-1] + f"/opex_comp_de.png", bbox_inches="tight")
+    axes[-1].legend(loc="upper left", bbox_to_anchor=(1, 1))
+    plt.savefig(snakemake.output[-1] + "/opex_comp_de.png", bbox_inches="tight")
     plt.close()