Skip to content

common.py

Cross-cutting network helpers: load clipping and resource meta-data attachment.

attach_resources_to_network_meta(n, snakemake)

Attach resource tables to the network meta before the netCDF export.

Embeds energy_totals and co2_totals CSV data directly into n.meta["resources"] so that downstream evaluation rules can access sectoral energy demand and CO₂ totals without relying on a separate post-processing step or extra input files.

The network name is also updated to a human-readable string that includes the planning horizon year.

Parameters:

Name Type Description Default
n pypsa.Network

The solved network whose metadata will be updated in place.

 required
snakemake snakemake.script.Snakemake

The Snakemake workflow object providing inputs, params, config, and wildcards.

 required

Raises:

Type Description
MissingInputException

If the required inputs (energy_totals and co2_totals_name) are not present on snakemake.input.

Returns:

Type Description
None

Updates n.meta and n.name in place.
Source code in mods/network/common.py 
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
def attach_resources_to_network_meta(
    n: pypsa.Network,
    snakemake: Snakemake,
) -> None:
    """
    Attach resource tables to the network meta before the netCDF export.

    Embeds ``energy_totals`` and ``co2_totals`` CSV data directly into
    ``n.meta["resources"]`` so that downstream evaluation rules can access
    sectoral energy demand and CO₂ totals without relying on a separate
    post-processing step or extra input files.

    The network name is also updated to a human-readable string that
    includes the planning horizon year.

    Parameters
    ----------
    n
        The solved network whose metadata will be updated in place.
    snakemake
        The Snakemake workflow object providing inputs, params, config,
        and wildcards.

    Raises
    ------
    MissingInputException
        If the required inputs (``energy_totals`` and ``co2_totals_name``) are
        not present on ``snakemake.input``.

    Returns
    -------
    :
        Updates ``n.meta`` and ``n.name`` in place.
    """
    energy_totals_year = snakemake.params["energy_year"]
    investment_year = snakemake.wildcards.planning_horizons

    energy_totals = pd.read_csv(snakemake.input.energy_totals, index_col=[0, 1]).xs(
        energy_totals_year, level="year"
    )
    co2_totals = pd.read_csv(snakemake.input.co2_totals_name, index_col=0)

    n.meta["resources"] = {
        "energy_totals": energy_totals.to_dict(orient="tight"),
        "co2_totals": co2_totals.to_dict(orient="tight"),
    }
    n.name = f"PyPSA-AT Network {investment_year}"
    logger.info(
        f"Attached energy_totals (year={energy_totals_year}) and co2_totals "
        f"to network meta for planning horizon {investment_year}."
    )

clip_negative_loads_for_edge_cases(n, snakemake)

Clip negative Loads for selected edge cases.

This is neccessary, because some electricity demands are calculated from heuristics. For example, heat for electricity from energy_totals and population share is deducted from regional base load. This can lead to negative Loads if heuristics yield larger values than input data sets. However, there are many examples where this may happen.

Parameters:

Name Type Description Default
n pypsa.Network

The network before solve step.

 required
snakemake snakemake.script.Snakemake

The Snakemake workflow object providing inputs, params, config, and outputs.

 required

Returns:

Type Description
None

Updates network in place.

Raises:

Type Description
RunTimeError

If expected edge cases could not be found.
Source code in mods/network/common.py 
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
def clip_negative_loads_for_edge_cases(n: pypsa.Network, snakemake: Snakemake) -> None:
    """
    Clip negative Loads for selected edge cases.

    This is neccessary, because some electricity demands are calculated
    from heuristics. For example, ``heat for electricity`` from
    ``energy_totals`` and population share is deducted from
    regional ``base load``. This can lead to negative Loads if
    heuristics yield larger values than input data sets. However,
    there are many examples where this may happen.

    Parameters
    ----------
    n
        The network before solve step.
    snakemake
        The Snakemake workflow object providing inputs, params,
        config, and outputs.

    Returns
    -------
    :
        Updates network in place.

    Raises
    ------
    RunTimeError
        If expected edge cases could not be found.

    """
    cfg = snakemake.config

    investment_year = int(snakemake.wildcards.planning_horizons)
    resolution = int(cfg["clustering"]["temporal"]["resolution_sector"].rstrip("H"))
    clustering = cfg["mods"]["modify_nuts3_shapes"]

    def _clip_static(carrier: str):
        idx = n.loads.index[n.loads["carrier"] == carrier]
        negatives = idx[n.loads.loc[idx, "p_set"] < 0]
        if negatives.empty:
            raise RuntimeError(f"Expected negative '{carrier}' Loads.")
        n.loads.loc[negatives, "p_set"] = 0

    def _clip_electricity(location: str):
        negatives = n.loads_t["p_set"][location].lt(0)
        if not any(negatives):
            raise RuntimeError(f"Expected negative electricity Load for {location}.")
        n.loads_t["p_set"].loc[negatives, location] = 0

    # Edge case (CI test config only): in the reduced at10 test network a few
    # "H2 for industry" demands are net-negative (industry produces surplus H2),
    # so the Load injects energy and trips test_no_load_supply. Clip to zero in
    # the test run only; full-resolution production runs are left untouched.
    if cfg["run"]["prefix"] == "test-sector-myopic-at10":
        _clip_static("H2 for industry")
        return  # skip any other clipping

    # Edge case: electricity for heat is larger than base load in AT126
    if resolution == 365 and clustering.startswith("AT35"):
        _clip_electricity("AT126")

    # For 120H runs IT1 always has negative electricity Loads
    if resolution == 120:
        _clip_electricity("IT1")

    # Edge case: runs contain negative H2 for industry Loads until including 2030
    if investment_year <= 2030:
        _clip_static("H2 for industry")

modify_prenetwork(n, snakemake)

Apply all PyPSA-AT specific modifications to the pre-network.

This is the single entry point for all AT-specific modifications during the modify_prenetwork Snakemake step. It orchestrates the individual modification functions and encapsulates the conditional logic for when each modification applies.

Parameters:

Name Type Description Default
n pypsa.Network

The pre-network to be modified in place.

 required
snakemake snakemake.script.Snakemake

The Snakemake workflow object providing inputs, params, config, and wildcards.

 required

Returns:

Type Description
None

Updates the :class:pypsa.Network in place.
Source code in mods/network/common.py 
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
def modify_prenetwork(n: pypsa.Network, snakemake: Snakemake) -> None:
    """
    Apply all PyPSA-AT specific modifications to the pre-network.

    This is the single entry point for all AT-specific modifications during
    the ``modify_prenetwork`` Snakemake step. It orchestrates the individual
    modification functions and encapsulates the conditional logic for when
    each modification applies.

    Parameters
    ----------
    n
        The pre-network to be modified in place.
    snakemake
        The Snakemake workflow object providing inputs, params, config,
        and wildcards.

    Returns
    -------
    :
        Updates the :class:`pypsa.Network` in place.
    """
    from scripts.add_electricity import load_costs

    costs = load_costs(snakemake.input.costs)

    unravel_gas_import_and_production(n, snakemake, costs)
    update_network_to_stop_ukrainian_gas_transit(n, snakemake)
    make_gas_pipelines_unextendable(n, snakemake)

    overwrite_pemmdb_capacities(n, snakemake)
    override_gas_storage_capacities(n, snakemake)
    apply_klien_potential_limits(n, snakemake)
    clip_negative_loads_for_edge_cases(n, snakemake)
    apply_tyndp_transmission_lower_bounds(n, snakemake)
    add_h2_imports(n, snakemake)

prepare_sector_network(n, snakemake, nodes, costs, spatial)

Apply all PyPSA-AT specific modifications during prepare_sector_network.

Parameters:

Name Type Description Default
n

The pre-network to be modified in place.

 required
snakemake

The Snakemake workflow object providing inputs, params, and config.

 required
costs

Processed cost DataFrame for the current planning horizon.

 required
nodes

Clustered node index (pop_layout.index).

 required
spatial

Spatial namespace produced by define_spatial.

 required

Returns:

Type Description

Modifies the network in place.
Source code in mods/network/common.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
def prepare_sector_network(n, snakemake, nodes, costs, spatial):
    """
    Apply all PyPSA-AT specific modifications during ``prepare_sector_network``.

    Parameters
    ----------
    n
        The pre-network to be modified in place.
    snakemake
        The Snakemake workflow object providing inputs, params, and config.
    costs
        Processed cost DataFrame for the current planning horizon.
    nodes
        Clustered node index (``pop_layout.index``).
    spatial
        Spatial namespace produced by ``define_spatial``.

    Returns
    -------
    :
        Modifies the network in place.
    """
    add_h2_for_industry_bus(n, nodes)
    add_methane_pyrolysis_plasma(n, snakemake, costs, nodes, spatial)