data.datasets.era5.create_tables() - Code Metrics - Inspection of "Fixes/#1190 profile building allocation wrong cell..." - openego/eGon-data - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — dev (#1197)

unknown

created 2025-03-13 22:39 UTC

data.datasets.era5.create_tables() A

↳ Parent: data.datasets.era5

Complexity

Conditions

Size

Total Lines	11
Code Lines	8

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	8
dl	0
loc	11
rs	10
c	0
b	0
f	0
cc	1
nop	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):
    __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):
    __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			__tablename__ = "egon_era5_weather_cells"
58			__table_args__ = {"schema": "supply"}
59			w_id = Column(Integer, primary_key=True)
60			geom = Column(Geometry("POLYGON", 4326))
61			geom_point = Column(Geometry("POINT", 4326))
62
63
64			class EgonRenewableFeedIn(Base):
65			__tablename__ = "egon_era5_renewable_feedin"
66			__table_args__ = {"schema": "supply"}
67			w_id = Column(Integer, primary_key=True)
68			weather_year = Column(Integer, primary_key=True)
69			carrier = Column(String, primary_key=True)
70			feedin = Column(ARRAY(Float()))
71
72
73			def create_tables():
74			db.execute_sql("CREATE SCHEMA IF NOT EXISTS supply;")
75			engine = db.engine()
76			db.execute_sql(
77			f"""
78			DROP TABLE IF EXISTS {EgonEra5Cells.__table__.schema}.{EgonEra5Cells.__table__.name} CASCADE;
79			"""
80			)
81			EgonEra5Cells.__table__.create(bind=engine, checkfirst=True)
82			EgonRenewableFeedIn.__table__.drop(bind=engine, checkfirst=True)
83			EgonRenewableFeedIn.__table__.create(bind=engine, checkfirst=True)
84
85
86			def import_cutout(boundary="Europe"):
87			"""Import weather data from cutout
88
89			Returns
90			-------
91			cutout : atlite.cutout.Cutout
92			Weather data stored in cutout
93
94			"""
95			for scn in set(egon.data.config.settings()["egon-data"]["--scenarios"]):
96			weather_year = get_sector_parameters("global", scn)["weather_year"]
97
98			if boundary == "Europe":
99			xs = slice(-12.0, 35.1)
100			ys = slice(72.0, 33.0)
101
102			elif boundary == "Germany":
103			geom_de = (
104			gpd.read_postgis(
105			"SELECT geometry as geom FROM boundaries.vg250_sta_bbox",
106			db.engine(),
107			)
108			.to_crs(4326)
109			.geom
110			)
111			xs = slice(geom_de.bounds.minx[0], geom_de.bounds.maxx[0])
112			ys = slice(geom_de.bounds.miny[0], geom_de.bounds.maxy[0])
113
114			elif boundary == "Germany-offshore":
115			xs = slice(5.5, 14.5)
116			ys = slice(55.5, 53.5)
117
118			else:
119			print(
120			f"Boundary {boundary} not defined. "
121			"Choose either 'Europe' or 'Germany'"
122			)
123
124			directory = (
125			Path(".")
126			/ (
127			egon.data.config.datasets()["era5_weather_data"]["targets"][
128			"weather_data"
129			]["path"]
130			)
131			/ f"{boundary.lower()}-{str(weather_year)}-era5.nc"
132			)
133
134			return atlite.Cutout(
135			path=directory.absolute(),
136			module="era5",
137			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...
138			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...
139			years=slice(weather_year, weather_year),
140			)
141
142
143			def download_era5():
144			"""Download weather data from era5
145
146			Returns
147			-------
148			None.
149
150			"""
151
152			directory = Path(".") / (
153			egon.data.config.datasets()["era5_weather_data"]["targets"][
154			"weather_data"
155			]["path"]
156			)
157
158			if not os.path.exists(directory):
159			os.mkdir(directory)
160
161			cutout = import_cutout()
162
163			if not cutout.prepared:
164			cutout.prepare()
165
166			cutout = import_cutout("Germany")
167
168			if not cutout.prepared:
169			cutout.prepare()
170
171			cutout = import_cutout("Germany-offshore")
172
173			if not cutout.prepared:
174			cutout.prepare()
175
176
177			def insert_weather_cells():
178			"""Insert weather cells from era5 into database table
179
180			Returns
181			-------
182			None.
183
184			"""
185			cfg = egon.data.config.datasets()["era5_weather_data"]
186
187			db.execute_sql(
188			f"""
189			DELETE FROM {cfg['targets']['weather_cells']['schema']}.
190			{cfg['targets']['weather_cells']['table']}
191			"""
192			)
193
194			cutout = import_cutout()
195
196			df = gpd.GeoDataFrame(
197			{"geom": cutout.grid.geometry}, geometry="geom", crs=4326
198			)
199
200			df.to_postgis(
201			cfg["targets"]["weather_cells"]["table"],
202			schema=cfg["targets"]["weather_cells"]["schema"],
203			con=db.engine(),
204			if_exists="append",
205			)
206
207			db.execute_sql(
208			f"""UPDATE {cfg['targets']['weather_cells']['schema']}.
209			{cfg['targets']['weather_cells']['table']}
210			SET geom_point=ST_Centroid(geom);"""
211			)
212

openego / eGon-data

Pull Request — dev (#1197)

data.datasets.era5.create_tables() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like