Passed
Pull Request — dev (#1008)
by
unknown
01:40
created

allocate_pumped_hydro_eGon100RE()   A

Complexity

Conditions 4

Size

Total Lines 65
Code Lines 29

Duplication

Lines 0
Ratio 0 %

Importance

Changes 0
Metric Value
eloc 29
dl 0
loc 65
rs 9.184
c 0
b 0
f 0
cc 4
nop 0

How to fix   Long Method   

Long Method

Small methods make your code easier to understand, in particular if combined with a good name. Besides, if your method is small, finding a good name is usually much easier.

For example, if you find yourself adding comments to a method's body, this is usually a good sign to extract the commented part to a new method, and use the comment as a starting point when coming up with a good name for this new method.

Commonly applied refactorings include:

1
"""The central module containing all code dealing with power plant data.
2
"""
3
from geoalchemy2 import Geometry
4
from pathlib import Path
5
from sqlalchemy import BigInteger, Column, Float, Integer, Sequence, String
6
from sqlalchemy.dialects.postgresql import JSONB
7
from sqlalchemy.ext.declarative import declarative_base
8
from sqlalchemy.orm import sessionmaker
9
from egon.data.datasets.storages.pumped_hydro import (
10
    select_mastr_pumped_hydro,
11
    select_nep_pumped_hydro,
12
    match_storage_units,
13
    get_location,
14
    apply_voltage_level_thresholds,
15
)
16
from egon.data.datasets.power_plants import assign_voltage_level
17
import geopandas as gpd
18
import pandas as pd
19
20
from egon.data import db, config
21
from egon.data.datasets import Dataset
22
23
24
Base = declarative_base()
25
26
27 View Code Duplication
class EgonStorages(Base):
0 ignored issues
show
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
28
    __tablename__ = "egon_storages"
29
    __table_args__ = {"schema": "supply"}
30
    id = Column(BigInteger, Sequence("storage_seq"), primary_key=True)
31
    sources = Column(JSONB)
32
    source_id = Column(JSONB)
33
    carrier = Column(String)
34
    el_capacity = Column(Float)
35
    bus_id = Column(Integer)
36
    voltage_level = Column(Integer)
37
    scenario = Column(String)
38
    geom = Column(Geometry("POINT", 4326))
39
40
41
class PumpedHydro(Dataset):
42
    def __init__(self, dependencies):
43
        super().__init__(
44
            name="Storages",
45
            version="0.0.2",
46
            dependencies=dependencies,
47
            tasks=(
48
                create_tables,
49
                allocate_pumped_hydro_eGon2035,
50
                allocate_pumped_hydro_eGon100RE,
51
                allocate_pv_home_batteries,
52
            ),
53
        )
54
55
56
def create_tables():
57
    """Create tables for power plant data
58
    Returns
59
    -------
60
    None.
61
    """
62
63
    cfg = config.datasets()["storages"]
64
    db.execute_sql(f"CREATE SCHEMA IF NOT EXISTS {cfg['target']['schema']};")
65
    engine = db.engine()
66
    db.execute_sql(
67
        f"""DROP TABLE IF EXISTS
68
        {cfg['target']['schema']}.{cfg['target']['table']}"""
69
    )
70
71
    db.execute_sql("""DROP SEQUENCE IF EXISTS pp_seq""")
72
    EgonStorages.__table__.create(bind=engine, checkfirst=True)
73
74
75
def allocate_pumped_hydro_eGon2035(export=True):
76
    """Allocates pumped_hydro plants for eGon2035 scenario and either exports
77
    results to data base or returns as a dataframe
78
79
    Parameters
80
    ----------
81
    export : bool
82
        Choose if allocated pumped hydro plants should be exported to the data
83
        base. The default is True.
84
        If export=False a data frame will be returned
85
86
    Returns
87
    -------
88
    power_plants : pandas.DataFrame
89
        List of pumped hydro plants in 'eGon2035' scenario
90
    """
91
92
    carrier = "pumped_hydro"
93
94
    cfg = config.datasets()["power_plants"]
95
96
    nep = select_nep_pumped_hydro()
97
    mastr = select_mastr_pumped_hydro()
98
99
    # Assign voltage level to MaStR
100
    mastr["voltage_level"] = assign_voltage_level(
101
        mastr.rename({"el_capacity": "Nettonennleistung"}, axis=1), cfg
102
    )
103
104
    # Initalize DataFrame for matching power plants
105
    matched = gpd.GeoDataFrame(
106
        columns=[
107
            "carrier",
108
            "el_capacity",
109
            "scenario",
110
            "geometry",
111
            "MaStRNummer",
112
            "source",
113
            "voltage_level",
114
        ]
115
    )
116
117
    # Match pumped_hydro units from NEP list
118
    # using PLZ and capacity
119
    matched, mastr, nep = match_storage_units(
120
        nep, mastr, matched, buffer_capacity=0.1, consider_carrier=False
121
    )
122
123
    # Match plants from NEP list using plz,
124
    # neglecting the capacity
125
    matched, mastr, nep = match_storage_units(
126
        nep,
127
        mastr,
128
        matched,
129
        consider_location="plz",
130
        consider_carrier=False,
131
        consider_capacity=False,
132
    )
133
134
    # Match plants from NEP list using city,
135
    # neglecting the capacity
136
    matched, mastr, nep = match_storage_units(
137
        nep,
138
        mastr,
139
        matched,
140
        consider_location="city",
141
        consider_carrier=False,
142
        consider_capacity=False,
143
    )
144
145
    # Match remaining plants from NEP using the federal state
146
    matched, mastr, nep = match_storage_units(
147
        nep,
148
        mastr,
149
        matched,
150
        buffer_capacity=0.1,
151
        consider_location="federal_state",
152
        consider_carrier=False,
153
    )
154
155
    # Match remaining plants from NEP using the federal state
156
    matched, mastr, nep = match_storage_units(
157
        nep,
158
        mastr,
159
        matched,
160
        buffer_capacity=0.7,
161
        consider_location="federal_state",
162
        consider_carrier=False,
163
    )
164
165
    print(f"{matched.el_capacity.sum()} MW of {carrier} matched")
166
    print(f"{nep.c2035_capacity.sum()} MW of {carrier} not matched")
167
168
    if nep.c2035_capacity.sum() > 0:
169
170
        # Get location using geolocator and city information
171
        located, unmatched = get_location(nep)
172
173
        # Bring both dataframes together
174
        matched = matched.append(
175
            located[
176
                [
177
                    "carrier",
178
                    "el_capacity",
179
                    "scenario",
180
                    "geometry",
181
                    "source",
182
                    "MaStRNummer",
183
                ]
184
            ],
185
            ignore_index=True,
186
        )
187
188
    # Set CRS
189
    matched.crs = "EPSG:4326"
190
191
    # Assign voltage level
192
    matched = apply_voltage_level_thresholds(matched)
193
194
    # Assign bus_id
195
    # Load grid district polygons
196
    mv_grid_districts = db.select_geodataframe(
197
        f"""
198
    SELECT * FROM {cfg['sources']['egon_mv_grid_district']}
199
    """,
200
        epsg=4326,
201
    )
202
203
    ehv_grid_districts = db.select_geodataframe(
204
        f"""
205
    SELECT * FROM {cfg['sources']['ehv_voronoi']}
206
    """,
207
        epsg=4326,
208
    )
209
210
    # Perform spatial joins for plants in ehv and hv level seperately
211
    power_plants_hv = gpd.sjoin(
212
        matched[matched.voltage_level >= 3],
213
        mv_grid_districts[["bus_id", "geom"]],
214
        how="left",
215
    ).drop(columns=["index_right"])
216
    power_plants_ehv = gpd.sjoin(
217
        matched[matched.voltage_level < 3],
218
        ehv_grid_districts[["bus_id", "geom"]],
219
        how="left",
220
    ).drop(columns=["index_right"])
221
222
    # Combine both dataframes
223
    power_plants = pd.concat([power_plants_hv, power_plants_ehv])
224
225
    # Delete existing units in the target table
226
    db.execute_sql(
227
        f""" DELETE FROM {cfg ['target']['schema']}.{cfg ['target']['table']}
228
        WHERE carrier IN ('pumped_hydro')
229
        AND scenario='eGon2035';"""
230
    )
231
232
    # If export = True export pumped_hydro plants to data base
233
234
    if export:
235
        # Insert into target table
236
        session = sessionmaker(bind=db.engine())()
237
        for i, row in power_plants.iterrows():
238
            entry = EgonStorages(
239
                sources={"el_capacity": row.source},
240
                source_id={"MastrNummer": row.MaStRNummer},
241
                carrier=row.carrier,
242
                el_capacity=row.el_capacity,
243
                voltage_level=row.voltage_level,
244
                bus_id=row.bus_id,
245
                scenario=row.scenario,
246
                geom=f"SRID=4326;POINT({row.geometry.x} {row.geometry.y})",
247
            )
248
            session.add(entry)
249
        session.commit()
250
251
    else:
252
        return power_plants
253
254
255
def allocate_pumped_hydro_eGon100RE():
256
    """Allocates pumped_hydro plants for eGon100RE scenario based on a
257
    prox-to-now method applied on allocated pumped-hydro plants in the eGon2035
258
    scenario.
259
260
    Parameters
261
    ----------
262
    None
263
264
    Returns
265
    -------
266
    None
267
    """
268
269
    carrier = "pumped_hydro"
270
    cfg = config.datasets()["power_plants"]
271
    boundary = config.settings()["egon-data"]["--dataset-boundary"]
272
273
    # Select installed capacity for pumped_hydro in eGon100RE scenario from
274
    # scenario capacities table
275
    capacity = db.select_dataframe(
276
        f"""
277
        SELECT capacity
278
        FROM {cfg['sources']['capacities']}
279
        WHERE carrier = '{carrier}'
280
        AND scenario_name = 'eGon100RE';
281
        """
282
    )
283
284
    if boundary == "Schleswig-Holstein":
285
        # Break capacity of pumped hydron plants down SH share in eGon2035
286
        capacity_phes = capacity.iat[0, 0] * 0.0176
287
288
    elif boundary == "Everything":
289
        # Select national capacity for pumped hydro
290
        capacity_phes = capacity.iat[0, 0]
291
292
    else:
293
        raise ValueError(f"'{boundary}' is not a valid dataset boundary.")
294
295
    # Get allocation of pumped_hydro plants in eGon2035 scenario as the reference
296
    # for the distribution in eGon100RE scenario
297
    allocation = allocate_pumped_hydro_eGon2035(export=False)
298
299
    scaling_factor = capacity_phes / allocation.el_capacity.sum()
300
301
    power_plants = allocation.copy()
302
    power_plants["scenario"] = "eGon100RE"
303
    power_plants["el_capacity"] = allocation.el_capacity * scaling_factor
304
305
    # Insert into target table
306
    session = sessionmaker(bind=db.engine())()
307
    for i, row in power_plants.iterrows():
308
        entry = EgonStorages(
309
            sources={"el_capacity": row.source},
310
            source_id={"MastrNummer": row.MaStRNummer},
311
            carrier=row.carrier,
312
            el_capacity=row.el_capacity,
313
            voltage_level=row.voltage_level,
314
            bus_id=row.bus_id,
315
            scenario=row.scenario,
316
            geom=f"SRID=4326;POINT({row.geometry.x} {row.geometry.y})",
317
        )
318
        session.add(entry)
319
    session.commit()
320
321
322
def home_batteries_per_scenario(scenario):
323
    """Allocates home batteries which define a lower boundary for extendable
324
    battery storage units. The overall installed capacity is taken from NEP
325
    for eGon2035 scenario. The spatial distribution of installed battery
326
    capacities is based on the installed pv rooftop capacity.
327
328
    Parameters
329
    ----------
330
    None
331
332
    Returns
333
    -------
334
    None
335
    """
336
337
    cfg = config.datasets()["storages"]
338
    dataset = config.settings()["egon-data"]["--dataset-boundary"]
339
340
    if scenario == "eGon2035":
341
342
        target_file = (
343
            Path(".")
344
            / "data_bundle_egon_data"
345
            / "nep2035_version2021"
346
            / cfg["sources"]["nep_capacities"]
347
        )
348
349
        capacities_nep = pd.read_excel(
350
            target_file,
351
            sheet_name="1.Entwurf_NEP2035_V2021",
352
            index_col="Unnamed: 0",
353
        )
354
355
        target = capacities_nep.Summe["PV-Batteriespeicher"]
356
357
    else:
358
        target = db.select_dataframe(
359
            f"""
360
            SELECT capacity
361
            FROM {cfg['sources']['capacities']}
362
            WHERE scenario_name = '{scenario}'
363
            AND carrier = 'battery';
364
            """
365
        ).capacity[0]
366
367
    pv_rooftop = db.select_dataframe(
368
        f"""
369
        SELECT bus, p_nom, generator_id
370
        FROM {cfg['sources']['generators']}
371
        WHERE scn_name = '{scenario}'
372
        AND carrier = 'solar_rooftop'
373
        AND bus IN
374
            (SELECT bus_id FROM {cfg['sources']['bus']}
375
               WHERE scn_name = '{scenario}' AND country = 'DE' );
376
        """
377
    )
378
379
    if dataset == "Schleswig-Holstein":
380
        target = target / 16
381
382
    battery = pv_rooftop
383
    battery["p_nom_min"] = target * battery["p_nom"] / battery["p_nom"].sum()
384
    battery = battery.drop(columns=["p_nom"])
385
386
    battery["carrier"] = "home_battery"
387
    battery["scenario"] = scenario
388
389
    if scenario == "eGon2035":
390
        source = "NEP"
391
392
    else:
393
        source = "p-e-s"
394
395
    battery[
396
        "source"
397
    ] = f"{source} capacity allocated based in installed PV rooftop capacity"
398
399
    # Insert into target table
400
    session = sessionmaker(bind=db.engine())()
401
    for i, row in battery.iterrows():
402
        entry = EgonStorages(
403
            sources={"el_capacity": row.source},
404
            source_id={"generator_id": row.generator_id},
405
            carrier=row.carrier,
406
            el_capacity=row.p_nom_min,
407
            bus_id=row.bus,
408
            scenario=row.scenario,
409
        )
410
        session.add(entry)
411
    session.commit()
412
413
414
def allocate_pv_home_batteries():
415
416
    home_batteries_per_scenario("eGon2035")
417
    home_batteries_per_scenario("eGon100RE")
418