data.datasets.era5.WeatherData.__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

data.datasets.era5.WeatherData.init() A

↳ Parent: data.datasets.era5

Complexity

Conditions

Size

Total Lines	6
Code Lines	6

Duplication

Lines	0
Ratio	0 %

Importance

Changes

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

"""
Central module containing all code dealing with importing era5 weather data.
"""
from pathlib import Path
import os

from geoalchemy2 import Geometry
from sqlalchemy import ARRAY, Column, Float, Integer, String
from sqlalchemy.ext.declarative import declarative_base
import atlite
import geopandas as gpd

from egon.data import db
from egon.data.datasets import Dataset
from egon.data.datasets.scenario_parameters import get_sector_parameters
import egon.data.config

# will be later imported from another file ###
Base = declarative_base()


class WeatherData(Dataset):
    """
    Download weather data from ERA5 using atlite

    This dataset downloads weather data for the selected representative weather year.
    This is done by applying functions from the atlite-tool.The downloaded wetaher data is stored into
    files within the subfolder 'cutouts'.


    *Dependencies*
      * :py:class:`ScenarioParameters <egon.data.datasets.scenario_parameters.ScenarioParameters>`
      * :py:class:`Vg250 <egon.data.datasets.vg250.Vg250>`
      * :py:func:`Setup <egon.data.datasets.database.setup>`

    *Resulting tables*
      * :py:class:`supply.egon_era5_weather_cells <egon.data.datasets.era5.EgonEra5Cells>` is created and filled
      * :py:class:`supply.egon_era5_renewable_feedin <egon.data.datasets.era5.EgonRenewableFeedIn>` is created

    """

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

    def __init__(self, dependencies):
        super().__init__(
            name=self.name,
            version=self.version,
            dependencies=dependencies,
            tasks=({create_tables, download_era5}, insert_weather_cells),
        )


class EgonEra5Cells(Base):
    """
    Class definition of table supply.egon_era5_weather_cells.

    """
    __tablename__ = "egon_era5_weather_cells"
    __table_args__ = {"schema": "supply"}
    w_id = Column(Integer, primary_key=True)
    geom = Column(Geometry("POLYGON", 4326))
    geom_point = Column(Geometry("POINT", 4326))


class EgonRenewableFeedIn(Base):
    """
    Class definition of table supply.egon_era5_renewable_feedin.

    """
    __tablename__ = "egon_era5_renewable_feedin"
    __table_args__ = {"schema": "supply"}
    w_id = Column(Integer, primary_key=True)
    weather_year = Column(Integer, primary_key=True)
    carrier = Column(String, primary_key=True)
    feedin = Column(ARRAY(Float()))


def create_tables():
    db.execute_sql("CREATE SCHEMA IF NOT EXISTS supply;")
    engine = db.engine()
    db.execute_sql(
        f"""
        DROP TABLE IF EXISTS {EgonEra5Cells.__table__.schema}.{EgonEra5Cells.__table__.name} CASCADE;
        """
    )
    EgonEra5Cells.__table__.create(bind=engine, checkfirst=True)
    EgonRenewableFeedIn.__table__.drop(bind=engine, checkfirst=True)
    EgonRenewableFeedIn.__table__.create(bind=engine, checkfirst=True)


def import_cutout(boundary="Europe"):
    """Import weather data from cutout

    Returns
    -------
    cutout : atlite.cutout.Cutout
        Weather data stored in cutout

    """
    for scn in set(egon.data.config.settings()["egon-data"]["--scenarios"]):
        weather_year = get_sector_parameters("global", scn)["weather_year"]

        if boundary == "Europe":
            xs = slice(-12.0, 35.1)
            ys = slice(72.0, 33.0)

        elif boundary == "Germany":
            geom_de = (
                gpd.read_postgis(
                    "SELECT geometry as geom FROM boundaries.vg250_sta_bbox",
                    db.engine(),
                )
                .to_crs(4326)
                .geom
            )
            xs = slice(geom_de.bounds.minx[0], geom_de.bounds.maxx[0])
            ys = slice(geom_de.bounds.miny[0], geom_de.bounds.maxy[0])

        elif boundary == "Germany-offshore":
            xs = slice(5.5, 14.5)
            ys = slice(55.5, 53.5)

        else:
            print(
                f"Boundary {boundary} not defined. "
                "Choose either 'Europe' or 'Germany'"
            )

        directory = (
            Path(".")
            / (
                egon.data.config.datasets()["era5_weather_data"]["targets"][
                    "weather_data"
                ]["path"]
            )
            / f"{boundary.lower()}-{str(weather_year)}-era5.nc"
        )

        return atlite.Cutout(
            path=directory.absolute(),
            module="era5",
            x=xs,

            y=ys,

            years=slice(weather_year, weather_year),
        )


def download_era5():
    """Download weather data from era5

    Returns
    -------
    None.

    """

    directory = Path(".") / (
        egon.data.config.datasets()["era5_weather_data"]["targets"][
            "weather_data"
        ]["path"]
    )

    if not os.path.exists(directory):
        os.mkdir(directory)

    cutout = import_cutout()

    if not cutout.prepared:
        cutout.prepare()

    cutout = import_cutout("Germany")

    if not cutout.prepared:
        cutout.prepare()

    cutout = import_cutout("Germany-offshore")

    if not cutout.prepared:
        cutout.prepare()


def insert_weather_cells():
    """Insert weather cells from era5 into database table

    Returns
    -------
    None.

    """
    cfg = egon.data.config.datasets()["era5_weather_data"]

    db.execute_sql(
        f"""
        DELETE FROM {cfg['targets']['weather_cells']['schema']}.
        {cfg['targets']['weather_cells']['table']}
        """
    )

    cutout = import_cutout()

    df = gpd.GeoDataFrame(
        {"geom": cutout.grid.geometry}, geometry="geom", crs=4326
    )

    df.to_postgis(
        cfg["targets"]["weather_cells"]["table"],
        schema=cfg["targets"]["weather_cells"]["schema"],
        con=db.engine(),
        if_exists="append",
    )

    db.execute_sql(
        f"""UPDATE {cfg['targets']['weather_cells']['schema']}.
        {cfg['targets']['weather_cells']['table']}
        SET geom_point=ST_Centroid(geom);"""
    )


1			"""
2			Central module containing all code dealing with importing era5 weather data.
3			"""
4			from pathlib import Path
5			import os
6
7			from geoalchemy2 import Geometry
8			from sqlalchemy import ARRAY, Column, Float, Integer, String
9			from sqlalchemy.ext.declarative import declarative_base
10			import atlite
11			import geopandas as gpd
12
13			from egon.data import db
14			from egon.data.datasets import Dataset
15			from egon.data.datasets.scenario_parameters import get_sector_parameters
16			import egon.data.config
17
18			# will be later imported from another file ###
19			Base = declarative_base()
20
21
22			class WeatherData(Dataset):
23			"""
24			Download weather data from ERA5 using atlite
25
26			This dataset downloads weather data for the selected representative weather year.
27			This is done by applying functions from the atlite-tool.The downloaded wetaher data is stored into
28			files within the subfolder 'cutouts'.
29
30
31			Dependencies
32			* :py:class:`ScenarioParameters <egon.data.datasets.scenario_parameters.ScenarioParameters>`
33			* :py:class:`Vg250 <egon.data.datasets.vg250.Vg250>`
34			* :py:func:`Setup <egon.data.datasets.database.setup>`
35
36			Resulting tables
37			* :py:class:`supply.egon_era5_weather_cells <egon.data.datasets.era5.EgonEra5Cells>` is created and filled
38			* :py:class:`supply.egon_era5_renewable_feedin <egon.data.datasets.era5.EgonRenewableFeedIn>` is created
39
40			"""
41
42			#:
43			name: str = "Era5"
44			#:
45			version: str = "0.0.3"
46
47			def __init__(self, dependencies):
48			super().__init__(
49			name=self.name,
50			version=self.version,
51			dependencies=dependencies,
52			tasks=({create_tables, download_era5}, insert_weather_cells),
53			)
54
55
56			class EgonEra5Cells(Base):
57			"""
58			Class definition of table supply.egon_era5_weather_cells.
59
60			"""
61			__tablename__ = "egon_era5_weather_cells"
62			__table_args__ = {"schema": "supply"}
63			w_id = Column(Integer, primary_key=True)
64			geom = Column(Geometry("POLYGON", 4326))
65			geom_point = Column(Geometry("POINT", 4326))
66
67
68			class EgonRenewableFeedIn(Base):
69			"""
70			Class definition of table supply.egon_era5_renewable_feedin.
71
72			"""
73			__tablename__ = "egon_era5_renewable_feedin"
74			__table_args__ = {"schema": "supply"}
75			w_id = Column(Integer, primary_key=True)
76			weather_year = Column(Integer, primary_key=True)
77			carrier = Column(String, primary_key=True)
78			feedin = Column(ARRAY(Float()))
79
80
81			def create_tables():
82			db.execute_sql("CREATE SCHEMA IF NOT EXISTS supply;")
83			engine = db.engine()
84			db.execute_sql(
85			f"""
86			DROP TABLE IF EXISTS {EgonEra5Cells.__table__.schema}.{EgonEra5Cells.__table__.name} CASCADE;
87			"""
88			)
89			EgonEra5Cells.__table__.create(bind=engine, checkfirst=True)
90			EgonRenewableFeedIn.__table__.drop(bind=engine, checkfirst=True)
91			EgonRenewableFeedIn.__table__.create(bind=engine, checkfirst=True)
92
93
94			def import_cutout(boundary="Europe"):
95			"""Import weather data from cutout
96
97			Returns
98			-------
99			cutout : atlite.cutout.Cutout
100			Weather data stored in cutout
101
102			"""
103			for scn in set(egon.data.config.settings()["egon-data"]["--scenarios"]):
104			weather_year = get_sector_parameters("global", scn)["weather_year"]
105
106			if boundary == "Europe":
107			xs = slice(-12.0, 35.1)
108			ys = slice(72.0, 33.0)
109
110			elif boundary == "Germany":
111			geom_de = (
112			gpd.read_postgis(
113			"SELECT geometry as geom FROM boundaries.vg250_sta_bbox",
114			db.engine(),
115			)
116			.to_crs(4326)
117			.geom
118			)
119			xs = slice(geom_de.bounds.minx[0], geom_de.bounds.maxx[0])
120			ys = slice(geom_de.bounds.miny[0], geom_de.bounds.maxy[0])
121
122			elif boundary == "Germany-offshore":
123			xs = slice(5.5, 14.5)
124			ys = slice(55.5, 53.5)
125
126			else:
127			print(
128			f"Boundary {boundary} not defined. "
129			"Choose either 'Europe' or 'Germany'"
130			)
131
132			directory = (
133			Path(".")
134			/ (
135			egon.data.config.datasets()["era5_weather_data"]["targets"][
136			"weather_data"
137			]["path"]
138			)
139			/ f"{boundary.lower()}-{str(weather_year)}-era5.nc"
140			)
141
142			return atlite.Cutout(
143			path=directory.absolute(),
144			module="era5",
145			x=xs,
			0 ignored issues – show introduced 2021-04-23 11:49 UTC by Report Bug Copy Issue Report The variable `xs` does not seem to be defined for all execution paths. Loading history...
146			y=ys,
			0 ignored issues – show introduced 2021-04-23 11:49 UTC by Report Bug Copy Issue Report The variable `ys` does not seem to be defined for all execution paths. Loading history...
147			years=slice(weather_year, weather_year),
148			)
149
150
151			def download_era5():
152			"""Download weather data from era5
153
154			Returns
155			-------
156			None.
157
158			"""
159
160			directory = Path(".") / (
161			egon.data.config.datasets()["era5_weather_data"]["targets"][
162			"weather_data"
163			]["path"]
164			)
165
166			if not os.path.exists(directory):
167			os.mkdir(directory)
168
169			cutout = import_cutout()
170
171			if not cutout.prepared:
172			cutout.prepare()
173
174			cutout = import_cutout("Germany")
175
176			if not cutout.prepared:
177			cutout.prepare()
178
179			cutout = import_cutout("Germany-offshore")
180
181			if not cutout.prepared:
182			cutout.prepare()
183
184
185			def insert_weather_cells():
186			"""Insert weather cells from era5 into database table
187
188			Returns
189			-------
190			None.
191
192			"""
193			cfg = egon.data.config.datasets()["era5_weather_data"]
194
195			db.execute_sql(
196			f"""
197			DELETE FROM {cfg['targets']['weather_cells']['schema']}.
198			{cfg['targets']['weather_cells']['table']}
199			"""
200			)
201
202			cutout = import_cutout()
203
204			df = gpd.GeoDataFrame(
205			{"geom": cutout.grid.geometry}, geometry="geom", crs=4326
206			)
207
208			df.to_postgis(
209			cfg["targets"]["weather_cells"]["table"],
210			schema=cfg["targets"]["weather_cells"]["schema"],
211			con=db.engine(),
212			if_exists="append",
213			)
214
215			db.execute_sql(
216			f"""UPDATE {cfg['targets']['weather_cells']['schema']}.
217			{cfg['targets']['weather_cells']['table']}
218			SET geom_point=ST_Centroid(geom);"""
219			)
220

openego / eGon-data

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

data.datasets.era5.WeatherData.__init__() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

data.datasets.era5.WeatherData.init() A