data.datasets.scenario_parameters.ScenarioParameters.__init__() - Code Metrics - Inspection of "Merge pull request #1273 from openego/fixes/#1256-..." - openego/eGon-data - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — dev ( 8582b4...82307e )

unknown

created 2025-06-12 13:09 UTC

ScenarioParameters.init() A

↳ Parent: data.datasets.scenario_parameters

Complexity

Conditions

Size

Total Lines	9
Code Lines	9

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	9
dl	0
loc	9
rs	9.95
c	0
b	0
f	0
cc	1
nop	2

"""The central module containing all code dealing with scenario table.
"""

from pathlib import Path
from urllib.request import urlretrieve
import shutil
import zipfile

from sqlalchemy import VARCHAR, Column, String
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
import pandas as pd

from egon.data import db
from egon.data.datasets import Dataset
import egon.data.config
import egon.data.datasets.scenario_parameters.parameters as parameters

Base = declarative_base()


class EgonScenario(Base):
    __tablename__ = "egon_scenario_parameters"
    __table_args__ = {"schema": "scenario"}
    name = Column(String, primary_key=True)
    global_parameters = Column(JSONB)
    electricity_parameters = Column(JSONB)
    gas_parameters = Column(JSONB)
    heat_parameters = Column(JSONB)
    mobility_parameters = Column(JSONB)
    description = Column(VARCHAR)


def create_table():
    """Create table for scenarios
    Returns
    -------
    None.
    """
    engine = db.engine()
    db.execute_sql("CREATE SCHEMA IF NOT EXISTS scenario;")
    db.execute_sql(
        "DROP TABLE IF EXISTS scenario.egon_scenario_parameters CASCADE;"
    )
    EgonScenario.__table__.create(bind=engine, checkfirst=True)


def get_scenario_year(scenario_name):
    """Derives scenarios year from scenario name. Scenario
    eGon100RE is an exception as year is not in the name."""
    try:
        year = int(scenario_name[-4:])
    except ValueError as e:
        if e.args[0] == "invalid literal for int() with base 10: '00RE'":
            year = 2050  # eGon100RE
        else:
            raise ValueError("The names of the scenarios do not end with the year!")
    return year


def insert_scenarios():
    """Insert scenarios and their parameters to scenario table

    Returns
    -------
    None.

    """

    db.execute_sql("DELETE FROM scenario.egon_scenario_parameters CASCADE;")

    session = sessionmaker(bind=db.engine())()

    # Scenario eGon2035
    egon2035 = EgonScenario(name="eGon2035")

    egon2035.description = """
        The mid-term scenario eGon2035 is based on scenario C 2035 of the
        Netzentwicklungsplan Strom 2035, Version 2021.
        Scenario C 2035 is characretized by an ambitious expansion of
        renewable energies and a higher share of sector coupling.
        Analogous to the Netzentwicklungsplan, the countries bordering germany
        are modeled based on Ten-Year Network Development Plan, Version 2020.
        """
    egon2035.global_parameters = parameters.global_settings(egon2035.name)

    egon2035.electricity_parameters = parameters.electricity(egon2035.name)

    egon2035.gas_parameters = parameters.gas(egon2035.name)

    egon2035.heat_parameters = parameters.heat(egon2035.name)

    egon2035.mobility_parameters = parameters.mobility(egon2035.name)

    session.add(egon2035)

    session.commit()

    # Scenario eGon100RE
    egon100re = EgonScenario(name="eGon100RE")

    egon100re.description = """
        The long-term scenario eGon100RE represents a 100% renewable
        energy secor in Germany.
        """
    egon100re.global_parameters = parameters.global_settings(egon100re.name)

    egon100re.electricity_parameters = parameters.electricity(egon100re.name)

    egon100re.gas_parameters = parameters.gas(egon100re.name)

    egon100re.heat_parameters = parameters.heat(egon100re.name)

    egon100re.mobility_parameters = parameters.mobility(egon100re.name)

    session.add(egon100re)

    session.commit()

    # Scenario eGon2021
    eGon2021 = EgonScenario(name="eGon2021")

    eGon2021.description = """
        Status quo scenario for 2021. Note: This is NOT A COMPLETE SCENARIO
        and covers only some sector data required by ding0, such as demand
        on NUTS 3 level and generation units .
        """
    eGon2021.global_parameters = parameters.global_settings(eGon2021.name)

    eGon2021.electricity_parameters = parameters.electricity(eGon2021.name)

    eGon2021.gas_parameters = parameters.gas(eGon2021.name)

    eGon2021.heat_parameters = parameters.heat(eGon2021.name)

    eGon2021.mobility_parameters = parameters.mobility(eGon2021.name)

    session.add(eGon2021)

    session.commit()

    # Scenario status2019
    status2019 = EgonScenario(name="status2019")

    status2019.description = """
        Status quo ante scenario for 2019 for validation use within the project PoWerD.
        """
    status2019.global_parameters = parameters.global_settings(status2019.name)

    status2019.electricity_parameters = parameters.electricity(status2019.name)

    status2019.gas_parameters = parameters.gas(status2019.name)

    status2019.heat_parameters = parameters.heat(status2019.name)

    status2019.mobility_parameters = parameters.mobility(status2019.name)

    session.add(status2019)

    session.commit()

    # Scenario status2023
    status2023 = EgonScenario(name="status2023")

    status2023.description = """
        Status quo ante scenario for 2023.
        """
    # TODO status2023 all settings from 2019 are used
    status2023.global_parameters = parameters.global_settings(status2023.name)

    status2023.electricity_parameters = parameters.electricity(status2019.name)

    status2023.gas_parameters = parameters.gas(status2019.name)

    status2023.heat_parameters = parameters.heat(status2019.name)

    status2023.mobility_parameters = parameters.mobility(status2023.name)

    session.add(status2023)

    session.commit()


def get_sector_parameters(sector, scenario=None):
    """Returns parameters for each sector as dictionary.

    If scenario=None data for all scenarios is returned as pandas.DataFrame.
    Otherwise the parameters of the specific scenario are returned as a dict.

    Parameters
    ----------
    sector : str
        Name of the sector.
        Options are: ['global', 'electricity', 'heat', 'gas', 'mobility']
    scenario : str, optional
        Name of the scenario. The default is None.

    Returns
    -------
    values : dict or pandas.DataFrane
        List or table of parameters for the selected sector

    """

    if scenario:
        if (
            scenario
            in db.select_dataframe(
                "SELECT name FROM scenario.egon_scenario_parameters"
            ).name.values
        ):
            values = db.select_dataframe(
                f"""
                    SELECT {sector}_parameters as val
                    FROM scenario.egon_scenario_parameters
                    WHERE name = '{scenario}';"""
            ).val[0]
        else:
            print(f"Scenario name {scenario} is not valid.")
    else:
        values = pd.concat(
            [
                pd.DataFrame(
                    db.select_dataframe(
                        f"""
                    SELECT {sector}_parameters as val
                    FROM scenario.egon_scenario_parameters
                    WHERE name='eGon2035'"""
                    ).val[0],
                    index=["eGon2035"],
                ),
                pd.DataFrame(
                    db.select_dataframe(
                        f"""
                        SELECT {sector}_parameters as val
                        FROM scenario.egon_scenario_parameters
                        WHERE name='eGon100RE'"""
                    ).val[0],
                    index=["eGon100RE"],
                ),
                pd.DataFrame(
                    db.select_dataframe(
                        f"""
                        SELECT {sector}_parameters as val
                        FROM scenario.egon_scenario_parameters
                        WHERE name='eGon2021'"""
                    ).val[0],
                    index=["eGon2021"],
                ),
            ],
            ignore_index=True,
        )

    return values



def download_pypsa_technology_data():
    """Downlad PyPSA technology data results."""
    data_path = Path(".") / "pypsa_technology_data"
    # Delete folder if it already exists
    if data_path.exists() and data_path.is_dir():
        shutil.rmtree(data_path)
    # Get parameters from config and set download URL
    sources = egon.data.config.datasets()["pypsa-technology-data"]["sources"][
        "zenodo"
    ]
    url = f"""https://zenodo.org/record/{sources['deposit_id']}/files/{sources['file']}"""
    target_file = egon.data.config.datasets()["pypsa-technology-data"][
        "targets"
    ]["file"]

    # Retrieve files
    urlretrieve(url, target_file)

    with zipfile.ZipFile(target_file, "r") as zip_ref:
        zip_ref.extractall(".")


class ScenarioParameters(Dataset):
    """
    Create and fill table with central parameters for each scenario

    This dataset creates and fills a table in the database that includes central parameters
    for each scenarios. These parameters are mostly from extrernal sources, they are defined
    and referenced within this dataset.
    The table is acced by various datasets to access the parameters for all sectors.


    *Dependencies*
      * :py:func:`Setup <egon.data.datasets.database.setup>`


    *Resulting tables*
      * :py:class:`scenario.egon_scenario_parameters <egon.data.datasets.scenario_parameters.EgonScenario>` is created and filled


    """

    #:
    name: str = "ScenarioParameters"
    #:
    version: str = "0.0.18"

    def __init__(self, dependencies):
        super().__init__(
            name=self.name,
            version=self.version,
            dependencies=dependencies,
            tasks=(
                create_table,
                download_pypsa_technology_data,
                insert_scenarios,
            ),
        )


1			"""The central module containing all code dealing with scenario table.
2			"""
3
4			from pathlib import Path
5			from urllib.request import urlretrieve
6			import shutil
7			import zipfile
8
9			from sqlalchemy import VARCHAR, Column, String
10			from sqlalchemy.dialects.postgresql import JSONB
11			from sqlalchemy.ext.declarative import declarative_base
12			from sqlalchemy.orm import sessionmaker
13			import pandas as pd
14
15			from egon.data import db
16			from egon.data.datasets import Dataset
17			import egon.data.config
18			import egon.data.datasets.scenario_parameters.parameters as parameters
19
20			Base = declarative_base()
21
22
23			class EgonScenario(Base):
24			__tablename__ = "egon_scenario_parameters"
25			__table_args__ = {"schema": "scenario"}
26			name = Column(String, primary_key=True)
27			global_parameters = Column(JSONB)
28			electricity_parameters = Column(JSONB)
29			gas_parameters = Column(JSONB)
30			heat_parameters = Column(JSONB)
31			mobility_parameters = Column(JSONB)
32			description = Column(VARCHAR)
33
34
35			def create_table():
36			"""Create table for scenarios
37			Returns
38			-------
39			None.
40			"""
41			engine = db.engine()
42			db.execute_sql("CREATE SCHEMA IF NOT EXISTS scenario;")
43			db.execute_sql(
44			"DROP TABLE IF EXISTS scenario.egon_scenario_parameters CASCADE;"
45			)
46			EgonScenario.__table__.create(bind=engine, checkfirst=True)
47
48
49			def get_scenario_year(scenario_name):
50			"""Derives scenarios year from scenario name. Scenario
51			eGon100RE is an exception as year is not in the name."""
52			try:
53			year = int(scenario_name[-4:])
54			except ValueError as e:
55			if e.args[0] == "invalid literal for int() with base 10: '00RE'":
56			year = 2050 # eGon100RE
57			else:
58			raise ValueError("The names of the scenarios do not end with the year!")
59			return year
60
61
62			def insert_scenarios():
63			"""Insert scenarios and their parameters to scenario table
64
65			Returns
66			-------
67			None.
68
69			"""
70
71			db.execute_sql("DELETE FROM scenario.egon_scenario_parameters CASCADE;")
72
73			session = sessionmaker(bind=db.engine())()
74
75			# Scenario eGon2035
76			egon2035 = EgonScenario(name="eGon2035")
77
78			egon2035.description = """
79			The mid-term scenario eGon2035 is based on scenario C 2035 of the
80			Netzentwicklungsplan Strom 2035, Version 2021.
81			Scenario C 2035 is characretized by an ambitious expansion of
82			renewable energies and a higher share of sector coupling.
83			Analogous to the Netzentwicklungsplan, the countries bordering germany
84			are modeled based on Ten-Year Network Development Plan, Version 2020.
85			"""
86			egon2035.global_parameters = parameters.global_settings(egon2035.name)
87
88			egon2035.electricity_parameters = parameters.electricity(egon2035.name)
89
90			egon2035.gas_parameters = parameters.gas(egon2035.name)
91
92			egon2035.heat_parameters = parameters.heat(egon2035.name)
93
94			egon2035.mobility_parameters = parameters.mobility(egon2035.name)
95
96			session.add(egon2035)
97
98			session.commit()
99
100			# Scenario eGon100RE
101			egon100re = EgonScenario(name="eGon100RE")
102
103			egon100re.description = """
104			The long-term scenario eGon100RE represents a 100% renewable
105			energy secor in Germany.
106			"""
107			egon100re.global_parameters = parameters.global_settings(egon100re.name)
108
109			egon100re.electricity_parameters = parameters.electricity(egon100re.name)
110
111			egon100re.gas_parameters = parameters.gas(egon100re.name)
112
113			egon100re.heat_parameters = parameters.heat(egon100re.name)
114
115			egon100re.mobility_parameters = parameters.mobility(egon100re.name)
116
117			session.add(egon100re)
118
119			session.commit()
120
121			# Scenario eGon2021
122			eGon2021 = EgonScenario(name="eGon2021")
123
124			eGon2021.description = """
125			Status quo scenario for 2021. Note: This is NOT A COMPLETE SCENARIO
126			and covers only some sector data required by ding0, such as demand
127			on NUTS 3 level and generation units .
128			"""
129			eGon2021.global_parameters = parameters.global_settings(eGon2021.name)
130
131			eGon2021.electricity_parameters = parameters.electricity(eGon2021.name)
132
133			eGon2021.gas_parameters = parameters.gas(eGon2021.name)
134
135			eGon2021.heat_parameters = parameters.heat(eGon2021.name)
136
137			eGon2021.mobility_parameters = parameters.mobility(eGon2021.name)
138
139			session.add(eGon2021)
140
141			session.commit()
142
143			# Scenario status2019
144			status2019 = EgonScenario(name="status2019")
145
146			status2019.description = """
147			Status quo ante scenario for 2019 for validation use within the project PoWerD.
148			"""
149			status2019.global_parameters = parameters.global_settings(status2019.name)
150
151			status2019.electricity_parameters = parameters.electricity(status2019.name)
152
153			status2019.gas_parameters = parameters.gas(status2019.name)
154
155			status2019.heat_parameters = parameters.heat(status2019.name)
156
157			status2019.mobility_parameters = parameters.mobility(status2019.name)
158
159			session.add(status2019)
160
161			session.commit()
162
163			# Scenario status2023
164			status2023 = EgonScenario(name="status2023")
165
166			status2023.description = """
167			Status quo ante scenario for 2023.
168			"""
169			# TODO status2023 all settings from 2019 are used
170			status2023.global_parameters = parameters.global_settings(status2023.name)
171
172			status2023.electricity_parameters = parameters.electricity(status2019.name)
173
174			status2023.gas_parameters = parameters.gas(status2019.name)
175
176			status2023.heat_parameters = parameters.heat(status2019.name)
177
178			status2023.mobility_parameters = parameters.mobility(status2023.name)
179
180			session.add(status2023)
181
182			session.commit()
183
184
185			def get_sector_parameters(sector, scenario=None):
186			"""Returns parameters for each sector as dictionary.
187
188			If scenario=None data for all scenarios is returned as pandas.DataFrame.
189			Otherwise the parameters of the specific scenario are returned as a dict.
190
191			Parameters
192			----------
193			sector : str
194			Name of the sector.
195			Options are: ['global', 'electricity', 'heat', 'gas', 'mobility']
196			scenario : str, optional
197			Name of the scenario. The default is None.
198
199			Returns
200			-------
201			values : dict or pandas.DataFrane
202			List or table of parameters for the selected sector
203
204			"""
205
206			if scenario:
207			if (
208			scenario
209			in db.select_dataframe(
210			"SELECT name FROM scenario.egon_scenario_parameters"
211			).name.values
212			):
213			values = db.select_dataframe(
214			f"""
215			SELECT {sector}_parameters as val
216			FROM scenario.egon_scenario_parameters
217			WHERE name = '{scenario}';"""
218			).val[0]
219			else:
220			print(f"Scenario name {scenario} is not valid.")
221			else:
222			values = pd.concat(
223			[
224			pd.DataFrame(
225			db.select_dataframe(
226			f"""
227			SELECT {sector}_parameters as val
228			FROM scenario.egon_scenario_parameters
229			WHERE name='eGon2035'"""
230			).val[0],
231			index=["eGon2035"],
232			),
233			pd.DataFrame(
234			db.select_dataframe(
235			f"""
236			SELECT {sector}_parameters as val
237			FROM scenario.egon_scenario_parameters
238			WHERE name='eGon100RE'"""
239			).val[0],
240			index=["eGon100RE"],
241			),
242			pd.DataFrame(
243			db.select_dataframe(
244			f"""
245			SELECT {sector}_parameters as val
246			FROM scenario.egon_scenario_parameters
247			WHERE name='eGon2021'"""
248			).val[0],
249			index=["eGon2021"],
250			),
251			],
252			ignore_index=True,
253			)
254
255			return values
			0 ignored issues – show introduced 2021-06-30 13:10 UTC by Report Bug Copy Issue Report The variable `values` does not seem to be defined for all execution paths. Loading history...
256
257
258			def download_pypsa_technology_data():
259			"""Downlad PyPSA technology data results."""
260			data_path = Path(".") / "pypsa_technology_data"
261			# Delete folder if it already exists
262			if data_path.exists() and data_path.is_dir():
263			shutil.rmtree(data_path)
264			# Get parameters from config and set download URL
265			sources = egon.data.config.datasets()["pypsa-technology-data"]["sources"][
266			"zenodo"
267			]
268			url = f"""https://zenodo.org/record/{sources['deposit_id']}/files/{sources['file']}"""
269			target_file = egon.data.config.datasets()["pypsa-technology-data"][
270			"targets"
271			]["file"]
272
273			# Retrieve files
274			urlretrieve(url, target_file)
275
276			with zipfile.ZipFile(target_file, "r") as zip_ref:
277			zip_ref.extractall(".")
278
279
280			class ScenarioParameters(Dataset):
281			"""
282			Create and fill table with central parameters for each scenario
283
284			This dataset creates and fills a table in the database that includes central parameters
285			for each scenarios. These parameters are mostly from extrernal sources, they are defined
286			and referenced within this dataset.
287			The table is acced by various datasets to access the parameters for all sectors.
288
289
290			Dependencies
291			* :py:func:`Setup <egon.data.datasets.database.setup>`
292
293
294			Resulting tables
295			* :py:class:`scenario.egon_scenario_parameters <egon.data.datasets.scenario_parameters.EgonScenario>` is created and filled
296
297
298			"""
299
300			#:
301			name: str = "ScenarioParameters"
302			#:
303			version: str = "0.0.18"
304
305			def __init__(self, dependencies):
306			super().__init__(
307			name=self.name,
308			version=self.version,
309			dependencies=dependencies,
310			tasks=(
311			create_table,
312			download_pypsa_technology_data,
313			insert_scenarios,
314			),
315			)
316

openego / eGon-data

Push — dev ( 8582b4...82307e )

ScenarioParameters.__init__() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

ScenarioParameters.init() A