data.datasets.era5.import_cutout() - Code Metrics - Inspection of "Fix/#1180 missing cts buildings" - openego/eGon-data - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — dev (#1181)

unknown

created 2025-02-03 09:46 UTC

data.datasets.era5.import_cutout() B

↳ Parent: data.datasets.era5

Complexity

Conditions

Size

Total Lines	57
Code Lines	33

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	33
dl	0
loc	57
rs	8.6213
c	0
b	0
f	0
cc	5
nop	1

How to fix Long Method

"""
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"
        )

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

            y=ys,

            years=slice(weather_year, weather_year),
        )

        return cutout


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

openego / eGon-data

Pull Request — dev (#1181)

data.datasets.era5.import_cutout() B

Complexity

Size

Duplication

Importance

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like