Feature/stransient export

README.md

@@ -68,7 +68,7 @@ PyPSA-DE is a softfork of PyPSA-EUR. As such, large parts of the functionality a
 - Additional constraints that limit maximum capacity of specific technologies
 - Import constraints on Efuels, hydrogen and electricity
 - Renewable build out according to the Wind-an-Land, Wind-auf-See and Solarstrategie laws
-- A comprehensive reporting  module that exports Capacity Expansion, Primary/Secondary/Final Energy, CO2 Emissions per Sector, Trade, Investments, and more.
+- A comprehensive reporting  module that exports Capacity Expansion, Primary/Secondary/Final Energy, CO2 Emissions per Sector, Trade, Investments, and more. Including a new `STRANSIENT` utility script for power systems stability exports.
 - Plotting functionality to compare different scenarios
 - Electricity Network development until 2030 (and for AC beyond) according to the NEP23
 - Offshore development until 2030 according to the Offshore NEP23

config/config.de.yaml

@@ -445,7 +445,10 @@ solving:
         gas pipeline new: 0.3
         H2 pipeline: 0.05
         H2 pipeline retrofitted: 0.05
-      fractional_last_unit_size: true
+    fractional_last_unit_size: true
+  solver:
+    name: highs
+    options: highs-default
   constraints:
     # The default CO2 budget uses the KSG targets, and the non CO2 emissions from the REMIND model in the KN2045_Mix scenario
     co2_budget_national:

scripts/export_stransient.py

@@ -0,0 +1,113 @@
+"""
+Utility script to extract network components from PyPSA-DE exports and format
+them into STRANSIENT-compatible CSV files.
+
+This can be run standalone via the CLI, or automatically as part of a Snakemake workflow.
+"""
+
+import argparse
+from pathlib import Path
+
+import pandas as pd
+
+
+def export_stransient(base: Path, out: Path):
+    """
+    Parses PyPSA network CSV exports and translates them into the STRANSIENT format.
+
+    Parameters
+    ----------
+    base : Path
+        Input directory containing standard PyPSA exports (buses.csv, lines.csv, generators.csv, loads.csv).
+    out : Path
+        Output directory to save the formatted STRANSIENT CSV files.
+    """
+    out.mkdir(parents=True, exist_ok=True)
+    bus_df = pd.read_csv(base / "buses.csv").rename(
+        columns={"name": "bus_id", "v_nom": "vn_kv"}
+    )
+    bus_df["vm_pu"] = bus_df["v_mag_pu_set"]
+    strans_bus = bus_df[["bus_id", "vn_kv", "type", "vm_pu"]].copy()
+    strans_bus["area"] = "DE"
+    strans_bus.to_csv(out / "stransient_bus.csv", index=False)
+    lines = pd.read_csv(base / "lines.csv").rename(columns={"name": "branch_id"})
+    lines["type"] = "AC_line"
+    lines[
+        ["branch_id", "bus0", "bus1", "r_pu", "x_pu", "length", "i_nom", "type"]
+    ].to_csv(out / "stransient_branch.csv", index=False)
+    gens = pd.read_csv(base / "generators.csv")
+    q_source = (
+        "q_nom"
+        if "q_nom" in gens.columns
+        else "q_set"
+        if "q_set" in gens.columns
+        else None
+    )
+    rename_map = {"name": "gen_id", "p_nom": "p_max_mw", "carrier": "type"}
+    if q_source is not None:
+        rename_map[q_source] = "q_max_mvar"
+    else:
+        gens["q_max_mvar"] = 0.0
+    gens = gens.rename(columns=rename_map)
+    required_cols = ["gen_id", "bus", "p_max_mw", "q_max_mvar", "type"]
+    gens[required_cols].to_csv(out / "stransient_gen.csv", index=False)
+    loads = pd.read_csv(base / "loads.csv")
+    p_source = (
+        "p_mw"
+        if "p_mw" in loads.columns
+        else "p_set"
+        if "p_set" in loads.columns
+        else None
+    )
+    q_source = (
+        "q_mvar"
+        if "q_mvar" in loads.columns
+        else "q_set"
+        if "q_set" in loads.columns
+        else None
+    )
+    load_rename = {"name": "load_id"}
+    if p_source:
+        load_rename[p_source] = "p_mw"
+    else:
+        loads["p_mw"] = 0.0
+    if q_source:
+        load_rename[q_source] = "q_mvar"
+    else:
+        loads["q_mvar"] = 0.0
+    loads = loads.rename(columns=load_rename)
+    loads[["load_id", "bus", "p_mw", "q_mvar"]].to_csv(
+        out / "stransient_load.csv", index=False
+    )
+    print("wired exports done")
+
+
+if __name__ == "__main__":
+    if "snakemake" in globals():
+        snakemake = globals()["snakemake"]
+        base_dir = Path(snakemake.input.exports_dir)
+        out_dir = Path(snakemake.output.stransient_dir)
+        export_stransient(base_dir, out_dir)
+    else:
+        parser = argparse.ArgumentParser(
+            description="Export STRANSIENT grids from PyPSA"
+        )
+        parser.add_argument(
+            "--exports-dir",
+            type=Path,
+            default=Path(
+                "results/20260114_limit_cross_border_flows/KN2045_Mix/exports"
+            ),
+            help="Directory containing PyPSA export CSVs",
+        )
+        parser.add_argument(
+            "--out-dir",
+            type=Path,
+            default=None,
+            help="Output directory. Defaults to <exports_dir>/../stransient",
+        )
+        args = parser.parse_args()
+
+        base_dir = args.exports_dir
+        out_dir = args.out_dir if args.out_dir else base_dir.parent / "stransient"
+        export_stransient(base_dir, out_dir)

scripts/pypsa-de/additional_functionality.py

@@ -9,6 +9,33 @@
 logger = logging.getLogger(__name__)
 
 
+def h2_import_limits_enabled(config):
+    return config.get("pypsa-de", {}).get("h2_import_limits", {}).get("enable", True)
+
+
+def safe_add_constraint(model, expr, rhs, sense, name):
+    """Wrap solver call to skip constant-constant constraints."""
+    try:
+        if sense == "<=":
+            model.add_constraints(expr <= rhs, name=name)
+        elif sense == ">=":
+            model.add_constraints(expr >= rhs, name=name)
+        else:
+            raise ValueError(f"Unsupported sense '{sense}'")
+        return True
+    except ValueError as exc:
+        if "Both sides of the constraint are constant" in str(exc):
+            logger.debug(
+                "Skipping constraint %s because both sides are constant (%s %s %s)",
+                name,
+                expr,
+                sense,
+                rhs,
+            )
+            return False
+        raise
+
+
 def add_capacity_limits(n, investment_year, limits_capacity, sense="maximum"):
     for c in n.iterate_components(limits_capacity):
         logger.info(f"Adding {sense} constraints for {c.list_name}")
@@ -208,6 +235,12 @@ def add_pos_neg_aux_variables(n, idx, var_name, infix):
 
 
 def h2_import_limits(n, investment_year, limits_volume_max):
+    if not h2_import_limits_enabled(n.config):
+        logger.info(
+            "Skipping H2 import limit constraints because pypsa-de.h2_import_limits.enable is False."
+        )
+        return
+
     for ct in limits_volume_max["h2_import"]:
         limit = limits_volume_max["h2_import"][ct][investment_year] * 1e6
 
@@ -228,6 +261,18 @@ def h2_import_limits(n, investment_year, limits_volume_max):
             & (n.links.bus1.str[:2] != ct)
         ]
 
+        if incoming.empty and outgoing.empty:
+            logger.warning(
+                f"No hydrogen import/export links found for {ct}; skipping limit enforcement."
+            )
+            continue
+
+        if incoming.empty and outgoing.empty:
+            logger.warning(
+                f"No hydrogen import/export links found for {ct}; skipping limit enforcement."
+            )
+            continue
+
         incoming_p = (
             n.model["Link-p"].loc[:, incoming] * n.snapshot_weightings.generators
         ).sum()
@@ -239,43 +284,57 @@ def h2_import_limits(n, investment_year, limits_volume_max):
 
         cname = f"H2_import_limit-{ct}"
 
-        n.model.add_constraints(lhs <= limit, name=f"GlobalConstraint-{cname}")
+        added = safe_add_constraint(
+            n.model,
+            lhs,
+            limit,
+            "<=",
+            name=f"GlobalConstraint-{cname}",
+        )
 
         if cname in n.global_constraints.index:
             logger.warning(
                 f"Global constraint {cname} already exists. Dropping and adding it again."
             )
             n.global_constraints.drop(cname, inplace=True)
 
-        n.add(
-            "GlobalConstraint",
-            cname,
-            constant=limit,
-            sense="<=",
-            type="",
-            carrier_attribute="",
-        )
+        if added:
+            n.add(
+                "GlobalConstraint",
+                cname,
+                constant=limit,
+                sense="<=",
+                type="",
+                carrier_attribute="",
+            )
 
         logger.info("Adding H2 export ban")
 
         cname = f"H2_export_ban-{ct}"
 
-        n.model.add_constraints(lhs >= 0, name=f"GlobalConstraint-{cname}")
+        added_export = safe_add_constraint(
+            n.model,
+            lhs,
+            0,
+            ">=",
+            name=f"GlobalConstraint-{cname}",
+        )
 
         if cname in n.global_constraints.index:
             logger.warning(
                 f"Global constraint {cname} already exists. Dropping and adding it again."
             )
             n.global_constraints.drop(cname, inplace=True)
 
-        n.add(
-            "GlobalConstraint",
-            cname,
-            constant=0,
-            sense=">=",
-            type="",
-            carrier_attribute="",
-        )
+        if added_export:
+            n.add(
+                "GlobalConstraint",
+                cname,
+                constant=0,
+                sense=">=",
+                type="",
+                carrier_attribute="",
+            )
 
 
 def h2_production_limits(n, investment_year, limits_volume_min, limits_volume_max):