data.datasets.gas_stores_germany.GasStores.__init__() - Code Metrics - Inspection of "Add missing gas stores in eGon100RE" - openego/eGon-data - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — dev (#936)

unknown

created 2023-03-31 08:57 UTC

GasStores.init() A

↳ Parent: data.datasets.gas_stores_germany

Complexity

Conditions

Size

Total Lines	6
Code Lines	6

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	6
dl	0
loc	6
rs	10
c	0
b	0
f	0
cc	1
nop	2

# -*- coding: utf-8 -*-
"""
The central module containing all code dealing with importing gas stores

This module contains the functions to import the existing methane stores
in Germany and to insert them into the database. They are modelled as
PyPSA stores and are not extendable.

"""
from pathlib import Path
from telnetlib import GA
import ast

import geopandas
import numpy as np
import pandas as pd

from egon.data import config, db
from egon.data.config import settings
from egon.data.datasets import Dataset
from egon.data.datasets.gas_grid import (
    ch4_nodes_number_G,
    define_gas_nodes_list,
)
from egon.data.datasets.scenario_parameters import get_sector_parameters


class GasStores(Dataset):
    """
    Insert the non extendable gas stores in Germany into the database

    Insert the non extendable gas stores into the database in Germany
    for the scnenarios eGon2035 and eGon100RE using the function
    :py:func:`insert_ch4_storages`.

    *Dependencies*
      * :py:class:`GasAreaseGon2035 <egon.data.datasets.gas_areas.GasAreaseGon2035>`
      * :py:class:`GasAreaseGon2035 <egon.data.datasets.gas_areas.GasAreaseGon100RE>`
      * :py:class:`GasNodesAndPipes <egon.data.datasets.gas_grid.GasNodesAndPipes>`

    *Resulting tables*
      * :py:class:`grid.egon_etrago_store <egon.data.datasets.etrago_setup.EgonPfHvStore>` is extended


    """

    #:
    name: str = "GasStores"
    #:
    version: str = "0.0.3"

    def __init__(self, dependencies):
        super().__init__(
            name=self.name,
            version=self.version,
            dependencies=dependencies,
            tasks=(insert_gas_stores_DE),
        )


def import_installed_ch4_storages(scn_name):
    """
    Define list of CH4 stores from the SciGRID_gas data

    This function reads from the SciGRID_gas dataset the existing CH4
    cavern stores in Germany, adjusts and returns them.
    Caverns reference: SciGRID_gas dataset (datasets/gas_data/data/IGGIELGN_Storages.csv
    downloaded in :func:`download_SciGRID_gas_data <egon.data.datasets.gas_grid.download_SciGRID_gas_data>`).
    For more information on these data, refer to the
    `SciGRID_gas IGGIELGN documentation <https://zenodo.org/record/4767098>`_.

    Parameters
    ----------
    scn_name : str
        Name of the scenario

    Returns
    -------
    Gas_storages_list :
        Dataframe containing the CH4 cavern stores units in Germany

    """
    target_file = (
        Path(".") / "datasets" / "gas_data" / "data" / "IGGIELGN_Storages.csv"
    )

    Gas_storages_list = pd.read_csv(
        target_file,
        delimiter=";",
        decimal=".",
        usecols=["lat", "long", "country_code", "param"],
    )

    Gas_storages_list = Gas_storages_list[
        Gas_storages_list["country_code"].str.match("DE")
    ]

    # Define new columns
    max_workingGas_M_m3 = []
    NUTS1 = []
    end_year = []
    for index, row in Gas_storages_list.iterrows():
        param = ast.literal_eval(row["param"])
        NUTS1.append(param["nuts_id_1"])
        end_year.append(param["end_year"])
        max_workingGas_M_m3.append(param["max_workingGas_M_m3"])

    Gas_storages_list = Gas_storages_list.assign(NUTS1=NUTS1)

    # Calculate e_nom
    conv_factor = 10830  # gross calorific value = 39 MJ/m3 (eurogas.org)
    Gas_storages_list["e_nom"] = [conv_factor * i for i in max_workingGas_M_m3]

    end_year = [float("inf") if x == None else x for x in end_year]
    Gas_storages_list = Gas_storages_list.assign(end_year=end_year)

    # Cut data to federal state if in testmode
    boundary = settings()["egon-data"]["--dataset-boundary"]
    if boundary != "Everything":
        map_states = {
            "Baden-Württemberg": "DE1",
            "Nordrhein-Westfalen": "DEA",
            "Hessen": "DE7",
            "Brandenburg": "DE4",
            "Bremen": "DE5",
            "Rheinland-Pfalz": "DEB",
            "Sachsen-Anhalt": "DEE",
            "Schleswig-Holstein": "DEF",
            "Mecklenburg-Vorpommern": "DE8",
            "Thüringen": "DEG",
            "Niedersachsen": "DE9",
            "Sachsen": "DED",
            "Hamburg": "DE6",
            "Saarland": "DEC",
            "Berlin": "DE3",
            "Bayern": "DE2",
        }

        Gas_storages_list = Gas_storages_list[
            Gas_storages_list["NUTS1"].isin([map_states[boundary], np.nan])
        ]

    # Remove unused storage units
    Gas_storages_list = Gas_storages_list[
        Gas_storages_list["end_year"] >= 2035
    ]

    Gas_storages_list = Gas_storages_list.rename(
        columns={"lat": "y", "long": "x"}
    )
    Gas_storages_list = geopandas.GeoDataFrame(
        Gas_storages_list,
        geometry=geopandas.points_from_xy(
            Gas_storages_list["x"], Gas_storages_list["y"]
        ),
    )
    Gas_storages_list = Gas_storages_list.rename(
        columns={"geometry": "geom"}
    ).set_geometry("geom", crs=4326)

    # Match to associated gas bus
    Gas_storages_list = Gas_storages_list.reset_index(drop=True)
    Gas_storages_list = db.assign_gas_bus_id(
        Gas_storages_list, scn_name, "CH4"
    )

    # Remove useless columns
    Gas_storages_list = Gas_storages_list.drop(
        columns=[
            "x",
            "y",
            "param",
            "country_code",
            "NUTS1",
            "end_year",
            "geom",
            "bus_id",
        ]
    )

    return Gas_storages_list


def import_gas_grid_capacity(scn_name, carrier):
    """
    Define the gas stores modelling the store capacity of the grid

    Define dataframe containing the modelling of the grid storage
    capacity. The whole storage capacity of the grid (130000 MWh,
    estimation of the Bundesnetzagentur) is split uniformly between
    all the German gas nodes of the grid (without consideration of the
    capacities of the pipes).
    In eGon100RE, the storage capacity of the grid is split between H2
    and CH4 stores, with the same share than the pipes capacity (value
    calculated in the p-e-s run).

    Parameters
    ----------
    scn_name : str
        Name of the scenario
    carrier : str
        Name of the carrier

    Returns
    -------
    Gas_storages_list :
        List of gas stores in Germany modelling the gas grid storage capacity

    """
    scn_params = get_sector_parameters("gas", scn_name)
    # Select source from dataset configuration
    source = config.datasets()["gas_stores"]["source"]

    Gas_grid_capacity = 130000  # Storage capacity of the CH4 grid - G.Volk "Die Herauforderung an die Bundesnetzagentur die Energiewende zu meistern" Berlin, Dec 2012
    N_ch4_nodes_G = ch4_nodes_number_G(
        define_gas_nodes_list()
    )  # Number of nodes in Germany
    Store_capacity = (
        Gas_grid_capacity / N_ch4_nodes_G
    )  # Storage capacity associated to each CH4 node of the German grid

    sql_gas = f"""SELECT bus_id, geom
                FROM {source['buses']['schema']}.{source['buses']['table']}
                WHERE carrier = '{carrier}' AND scn_name = '{scn_name}'
                AND country = 'DE';"""
    Gas_storages_list = db.select_geodataframe(sql_gas, epsg=4326)

    # Add missing column
    Gas_storages_list["bus"] = Gas_storages_list["bus_id"]

    if scn_name == "eGon100RE" and carrier == "CH4":
        Gas_storages_list["e_nom"] = Store_capacity * (
            1
            - get_sector_parameters("gas", scn_name)[
                "retrofitted_CH4pipeline-to-H2pipeline_share"
            ]
        )
    elif scn_name == "eGon2035" and carrier == "CH4":
        Gas_storages_list["e_nom"] = Store_capacity
    elif scn_name == "eGon100RE" and carrier == "H2_grid":
        Gas_storages_list["e_nom"] = Store_capacity * (
            scn_params["retrofitted_CH4pipeline-to-H2pipeline_share"]
        )

    # Remove useless columns
    Gas_storages_list = Gas_storages_list.drop(columns=["bus_id", "geom"])

    return Gas_storages_list


def insert_gas_stores_germany(scn_name, carrier):
    """
    Insert gas stores for specific scenario

    Insert non extendable gas stores for specific scenario in Germany
    by executing the following steps:
      * Clean the database
      * For CH4 stores, call the functions
        :py:func:`import_installed_ch4_storages` to receive the CH4
        cavern stores and :py:func:`import_gas_grid_capacity` to
        receive the CH4 stores modelling the storage capacity of the
        grid.
      * For H2 stores, call only the function
        :py:func:`import_gas_grid_capacity` to receive the H2 stores
        modelling the storage capacity of the grid.
      * Aggregate of the stores attached to the same bus
      * Add the missing columns: store_id, scn_name, carrier, e_cyclic
      * Insert the stores in the database

    Parameters
    ----------
    scn_name : str
        Name of the scenario
    carrier: str
        Name of the carrier

    Returns
    -------
    None

    """
    # Connect to local database
    engine = db.engine()

    # Select target from dataset configuration
    source = config.datasets()["gas_stores"]["source"]
    target = config.datasets()["gas_stores"]["target"]

    # Clean table
    db.execute_sql(
        f"""
        DELETE FROM {target['stores']['schema']}.{target['stores']['table']}  
        WHERE "carrier" = '{carrier}'
        AND scn_name = '{scn_name}'
        AND bus IN (
            SELECT bus_id FROM {source['buses']['schema']}.{source['buses']['table']}
            WHERE scn_name = '{scn_name}' 
            AND country = 'DE'
            );
        """
    )

    if carrier == "CH4":
        gas_storages_list = pd.concat(
            [
                import_installed_ch4_storages(scn_name),
                import_gas_grid_capacity(scn_name, carrier),
            ]
        )
    elif carrier == "H2":
        gas_storages_list = import_gas_grid_capacity(scn_name, "H2_grid")

    # Aggregate ch4 stores with same properties at the same bus
    gas_storages_list = (
        gas_storages_list.groupby(["bus"])

        .agg({"e_nom": "sum"})
        .reset_index(drop=False)
    )

    # Add missing columns
    c = {"scn_name": scn_name, "carrier": carrier, "e_cyclic": True}
    gas_storages_list = gas_storages_list.assign(**c)

    new_id = db.next_etrago_id("store")
    gas_storages_list["store_id"] = range(
        new_id, new_id + len(gas_storages_list)
    )

    # Insert data to db
    gas_storages_list.to_sql(
        target["stores"]["table"],
        engine,
        schema=target["stores"]["schema"],
        index=False,
        if_exists="append",
    )


def insert_gas_stores_DE():
    """
    Overall function to import non extendable gas stores in Germany

    This function calls :py:func:`insert_gas_stores_germany` three
    times to insert the corresponding non extendable gas stores in the
    database:
      * The CH4 stores for eGon2035: caverns from SciGRID_gas data and
        modelling of the storage grid capacity
      * The CH4 stores for eGon100RE: caverns from SciGRID_gas data and
        modelling of the storage grid capacity (split with H2)
      * The H2 stores for eGon100RE: caverns from SciGRID_gas data and
        modelling of the storage grid capacity (split with CH4)
    This function has no return.

    """
    insert_gas_stores_germany("eGon2035", "CH4")
    insert_gas_stores_germany("eGon100RE", "CH4")
    insert_gas_stores_germany("eGon100RE", "H2")


1			# -- coding: utf-8 --
2			"""
3			The central module containing all code dealing with importing gas stores
4
5			This module contains the functions to import the existing methane stores
6			in Germany and to insert them into the database. They are modelled as
7			PyPSA stores and are not extendable.
8
9			"""
10			from pathlib import Path
11			from telnetlib import GA
12			import ast
13
14			import geopandas
15			import numpy as np
16			import pandas as pd
17
18			from egon.data import config, db
19			from egon.data.config import settings
20			from egon.data.datasets import Dataset
21			from egon.data.datasets.gas_grid import (
22			ch4_nodes_number_G,
23			define_gas_nodes_list,
24			)
25			from egon.data.datasets.scenario_parameters import get_sector_parameters
26
27
28			class GasStores(Dataset):
29			"""
30			Insert the non extendable gas stores in Germany into the database
31
32			Insert the non extendable gas stores into the database in Germany
33			for the scnenarios eGon2035 and eGon100RE using the function
34			:py:func:`insert_ch4_storages`.
35
36			Dependencies
37			* :py:class:`GasAreaseGon2035 <egon.data.datasets.gas_areas.GasAreaseGon2035>`
38			* :py:class:`GasAreaseGon2035 <egon.data.datasets.gas_areas.GasAreaseGon100RE>`
39			* :py:class:`GasNodesAndPipes <egon.data.datasets.gas_grid.GasNodesAndPipes>`
40
41			Resulting tables
42			* :py:class:`grid.egon_etrago_store <egon.data.datasets.etrago_setup.EgonPfHvStore>` is extended
43
44
45			"""
46
47			#:
48			name: str = "GasStores"
49			#:
50			version: str = "0.0.3"
51
52			def __init__(self, dependencies):
53			super().__init__(
54			name=self.name,
55			version=self.version,
56			dependencies=dependencies,
57			tasks=(insert_gas_stores_DE),
58			)
59
60
61			def import_installed_ch4_storages(scn_name):
62			"""
63			Define list of CH4 stores from the SciGRID_gas data
64
65			This function reads from the SciGRID_gas dataset the existing CH4
66			cavern stores in Germany, adjusts and returns them.
67			Caverns reference: SciGRID_gas dataset (datasets/gas_data/data/IGGIELGN_Storages.csv
68			downloaded in :func:`download_SciGRID_gas_data <egon.data.datasets.gas_grid.download_SciGRID_gas_data>`).
69			For more information on these data, refer to the
70			`SciGRID_gas IGGIELGN documentation <https://zenodo.org/record/4767098>`_.
71
72			Parameters
73			----------
74			scn_name : str
75			Name of the scenario
76
77			Returns
78			-------
79			Gas_storages_list :
80			Dataframe containing the CH4 cavern stores units in Germany
81
82			"""
83			target_file = (
84			Path(".") / "datasets" / "gas_data" / "data" / "IGGIELGN_Storages.csv"
85			)
86
87			Gas_storages_list = pd.read_csv(
88			target_file,
89			delimiter=";",
90			decimal=".",
91			usecols=["lat", "long", "country_code", "param"],
92			)
93
94			Gas_storages_list = Gas_storages_list[
95			Gas_storages_list["country_code"].str.match("DE")
96			]
97
98			# Define new columns
99			max_workingGas_M_m3 = []
100			NUTS1 = []
101			end_year = []
102			for index, row in Gas_storages_list.iterrows():
103			param = ast.literal_eval(row["param"])
104			NUTS1.append(param["nuts_id_1"])
105			end_year.append(param["end_year"])
106			max_workingGas_M_m3.append(param["max_workingGas_M_m3"])
107
108			Gas_storages_list = Gas_storages_list.assign(NUTS1=NUTS1)
109
110			# Calculate e_nom
111			conv_factor = 10830 # gross calorific value = 39 MJ/m3 (eurogas.org)
112			Gas_storages_list["e_nom"] = [conv_factor * i for i in max_workingGas_M_m3]
113
114			end_year = [float("inf") if x == None else x for x in end_year]
115			Gas_storages_list = Gas_storages_list.assign(end_year=end_year)
116
117			# Cut data to federal state if in testmode
118			boundary = settings()["egon-data"]["--dataset-boundary"]
119			if boundary != "Everything":
120			map_states = {
121			"Baden-Württemberg": "DE1",
122			"Nordrhein-Westfalen": "DEA",
123			"Hessen": "DE7",
124			"Brandenburg": "DE4",
125			"Bremen": "DE5",
126			"Rheinland-Pfalz": "DEB",
127			"Sachsen-Anhalt": "DEE",
128			"Schleswig-Holstein": "DEF",
129			"Mecklenburg-Vorpommern": "DE8",
130			"Thüringen": "DEG",
131			"Niedersachsen": "DE9",
132			"Sachsen": "DED",
133			"Hamburg": "DE6",
134			"Saarland": "DEC",
135			"Berlin": "DE3",
136			"Bayern": "DE2",
137			}
138
139			Gas_storages_list = Gas_storages_list[
140			Gas_storages_list["NUTS1"].isin([map_states[boundary], np.nan])
141			]
142
143			# Remove unused storage units
144			Gas_storages_list = Gas_storages_list[
145			Gas_storages_list["end_year"] >= 2035
146			]
147
148			Gas_storages_list = Gas_storages_list.rename(
149			columns={"lat": "y", "long": "x"}
150			)
151			Gas_storages_list = geopandas.GeoDataFrame(
152			Gas_storages_list,
153			geometry=geopandas.points_from_xy(
154			Gas_storages_list["x"], Gas_storages_list["y"]
155			),
156			)
157			Gas_storages_list = Gas_storages_list.rename(
158			columns={"geometry": "geom"}
159			).set_geometry("geom", crs=4326)
160
161			# Match to associated gas bus
162			Gas_storages_list = Gas_storages_list.reset_index(drop=True)
163			Gas_storages_list = db.assign_gas_bus_id(
164			Gas_storages_list, scn_name, "CH4"
165			)
166
167			# Remove useless columns
168			Gas_storages_list = Gas_storages_list.drop(
169			columns=[
170			"x",
171			"y",
172			"param",
173			"country_code",
174			"NUTS1",
175			"end_year",
176			"geom",
177			"bus_id",
178			]
179			)
180
181			return Gas_storages_list
182
183
184			def import_gas_grid_capacity(scn_name, carrier):
185			"""
186			Define the gas stores modelling the store capacity of the grid
187
188			Define dataframe containing the modelling of the grid storage
189			capacity. The whole storage capacity of the grid (130000 MWh,
190			estimation of the Bundesnetzagentur) is split uniformly between
191			all the German gas nodes of the grid (without consideration of the
192			capacities of the pipes).
193			In eGon100RE, the storage capacity of the grid is split between H2
194			and CH4 stores, with the same share than the pipes capacity (value
195			calculated in the p-e-s run).
196
197			Parameters
198			----------
199			scn_name : str
200			Name of the scenario
201			carrier : str
202			Name of the carrier
203
204			Returns
205			-------
206			Gas_storages_list :
207			List of gas stores in Germany modelling the gas grid storage capacity
208
209			"""
210			scn_params = get_sector_parameters("gas", scn_name)
211			# Select source from dataset configuration
212			source = config.datasets()["gas_stores"]["source"]
213
214			Gas_grid_capacity = 130000 # Storage capacity of the CH4 grid - G.Volk "Die Herauforderung an die Bundesnetzagentur die Energiewende zu meistern" Berlin, Dec 2012
215			N_ch4_nodes_G = ch4_nodes_number_G(
216			define_gas_nodes_list()
217			) # Number of nodes in Germany
218			Store_capacity = (
219			Gas_grid_capacity / N_ch4_nodes_G
220			) # Storage capacity associated to each CH4 node of the German grid
221
222			sql_gas = f"""SELECT bus_id, geom
223			FROM {source['buses']['schema']}.{source['buses']['table']}
224			WHERE carrier = '{carrier}' AND scn_name = '{scn_name}'
225			AND country = 'DE';"""
226			Gas_storages_list = db.select_geodataframe(sql_gas, epsg=4326)
227
228			# Add missing column
229			Gas_storages_list["bus"] = Gas_storages_list["bus_id"]
230
231			if scn_name == "eGon100RE" and carrier == "CH4":
232			Gas_storages_list["e_nom"] = Store_capacity * (
233			1
234			- get_sector_parameters("gas", scn_name)[
235			"retrofitted_CH4pipeline-to-H2pipeline_share"
236			]
237			)
238			elif scn_name == "eGon2035" and carrier == "CH4":
239			Gas_storages_list["e_nom"] = Store_capacity
240			elif scn_name == "eGon100RE" and carrier == "H2_grid":
241			Gas_storages_list["e_nom"] = Store_capacity * (
242			scn_params["retrofitted_CH4pipeline-to-H2pipeline_share"]
243			)
244
245			# Remove useless columns
246			Gas_storages_list = Gas_storages_list.drop(columns=["bus_id", "geom"])
247
248			return Gas_storages_list
249
250
251			def insert_gas_stores_germany(scn_name, carrier):
252			"""
253			Insert gas stores for specific scenario
254
255			Insert non extendable gas stores for specific scenario in Germany
256			by executing the following steps:
257			* Clean the database
258			* For CH4 stores, call the functions
259			:py:func:`import_installed_ch4_storages` to receive the CH4
260			cavern stores and :py:func:`import_gas_grid_capacity` to
261			receive the CH4 stores modelling the storage capacity of the
262			grid.
263			* For H2 stores, call only the function
264			:py:func:`import_gas_grid_capacity` to receive the H2 stores
265			modelling the storage capacity of the grid.
266			* Aggregate of the stores attached to the same bus
267			* Add the missing columns: store_id, scn_name, carrier, e_cyclic
268			* Insert the stores in the database
269
270			Parameters
271			----------
272			scn_name : str
273			Name of the scenario
274			carrier: str
275			Name of the carrier
276
277			Returns
278			-------
279			None
280
281			"""
282			# Connect to local database
283			engine = db.engine()
284
285			# Select target from dataset configuration
286			source = config.datasets()["gas_stores"]["source"]
287			target = config.datasets()["gas_stores"]["target"]
288
289			# Clean table
290			db.execute_sql(
291			f"""
292			DELETE FROM {target['stores']['schema']}.{target['stores']['table']}
293			WHERE "carrier" = '{carrier}'
294			AND scn_name = '{scn_name}'
295			AND bus IN (
296			SELECT bus_id FROM {source['buses']['schema']}.{source['buses']['table']}
297			WHERE scn_name = '{scn_name}'
298			AND country = 'DE'
299			);
300			"""
301			)
302
303			if carrier == "CH4":
304			gas_storages_list = pd.concat(
305			[
306			import_installed_ch4_storages(scn_name),
307			import_gas_grid_capacity(scn_name, carrier),
308			]
309			)
310			elif carrier == "H2":
311			gas_storages_list = import_gas_grid_capacity(scn_name, "H2_grid")
312
313			# Aggregate ch4 stores with same properties at the same bus
314			gas_storages_list = (
315			gas_storages_list.groupby(["bus"])
			0 ignored issues – show introduced 2022-09-15 10:00 UTC by Report Bug Copy Issue Report The variable `gas_storages_list` does not seem to be defined for all execution paths. Loading history...
316			.agg({"e_nom": "sum"})
317			.reset_index(drop=False)
318			)
319
320			# Add missing columns
321			c = {"scn_name": scn_name, "carrier": carrier, "e_cyclic": True}
322			gas_storages_list = gas_storages_list.assign(**c)
323
324			new_id = db.next_etrago_id("store")
325			gas_storages_list["store_id"] = range(
326			new_id, new_id + len(gas_storages_list)
327			)
328
329			# Insert data to db
330			gas_storages_list.to_sql(
331			target["stores"]["table"],
332			engine,
333			schema=target["stores"]["schema"],
334			index=False,
335			if_exists="append",
336			)
337
338
339			def insert_gas_stores_DE():
340			"""
341			Overall function to import non extendable gas stores in Germany
342
343			This function calls :py:func:`insert_gas_stores_germany` three
344			times to insert the corresponding non extendable gas stores in the
345			database:
346			* The CH4 stores for eGon2035: caverns from SciGRID_gas data and
347			modelling of the storage grid capacity
348			* The CH4 stores for eGon100RE: caverns from SciGRID_gas data and
349			modelling of the storage grid capacity (split with H2)
350			* The H2 stores for eGon100RE: caverns from SciGRID_gas data and
351			modelling of the storage grid capacity (split with CH4)
352			This function has no return.
353
354			"""
355			insert_gas_stores_germany("eGon2035", "CH4")
356			insert_gas_stores_germany("eGon100RE", "CH4")
357			insert_gas_stores_germany("eGon100RE", "H2")
358

openego / eGon-data

Pull Request — dev (#936)

GasStores.__init__() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

GasStores.init() A