data.datasets.gas_areas.create_voronoi() - Code Metrics - Inspection of "Features/#888 move gas voronoi class definition" - openego/eGon-data - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — dev (#891)

unknown

created 2022-09-29 12:11 UTC

data.datasets.gas_areas.create_voronoi() A

↳ Parent: data.datasets.gas_areas

Complexity

Conditions

Size

Total Lines	59
Code Lines	27

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	27
dl	0
loc	59
rs	9.232
c	0
b	0
f	0
cc	2
nop	2

How to fix Long Method

"""The central module containing code to create CH4 and H2 voronoi polygons

"""
from geoalchemy2.types import Geometry
from sqlalchemy import BigInteger, Column, Text
from sqlalchemy.ext.declarative import declarative_base

from egon.data import db
from egon.data.datasets import Dataset
from egon.data.datasets.generate_voronoi import get_voronoi_geodataframe


class GasAreaseGon2035(Dataset):
    def __init__(self, dependencies):
        super().__init__(
            name="GasAreaseGon2035",
            version="0.0.2",
            dependencies=dependencies,
            tasks=(create_gas_voronoi_table, voronoi_egon2035),
        )


class GasAreaseGon100RE(Dataset):
    def __init__(self, dependencies):
        super().__init__(
            name="GasAreaseGon100RE",
            version="0.0.1",
            dependencies=dependencies,
            tasks=(voronoi_egon100RE),
        )


Base = declarative_base()


class EgonPfHvGasVoronoi(Base):
    __tablename__ = "egon_gas_voronoi"
    __table_args__ = {"schema": "grid"}

    scn_name = Column(Text, primary_key=True, nullable=False)
    bus_id = Column(BigInteger, primary_key=True, nullable=False)
    carrier = Column(Text)
    geom = Column(Geometry("GEOMETRY", 4326))


def create_gas_voronoi_table():
    engine = db.engine()
    EgonPfHvGasVoronoi.__table__.drop(bind=engine, checkfirst=True)
    EgonPfHvGasVoronoi.__table__.create(bind=engine, checkfirst=True)


def voronoi_egon2035():
    """
    Create voronoi polygons for all gas carriers in eGon2035 scenario
    """
    for carrier in ["CH4", "H2_grid", "H2_saltcavern"]:
        create_voronoi("eGon2035", carrier)


def voronoi_egon100RE():
    """
    Create voronoi polygons for all gas carriers in eGon100RE scenario
    """
    for carrier in ["CH4", "H2_grid", "H2_saltcavern"]:
        create_voronoi("eGon100RE", carrier)


def create_voronoi(scn_name, carrier):
    """
    Create voronoi polygons for specified carrier in specified scenario.

    Parameters
    ----------
    scn_name : str
        Name of the scenario
    carrier : str
        Name of the carrier
    """
    boundary = db.select_geodataframe(
        f"""
            SELECT id, geometry
            FROM boundaries.vg250_sta_union;
        """,
        geom_col="geometry",
    ).to_crs(epsg=4326)

    engine = db.engine()

    db.execute_sql(
        f"""
        DELETE FROM grid.egon_gas_voronoi
        WHERE "carrier" = '{carrier}' and "scn_name" = '{scn_name}';
        """
    )

    buses = db.select_geodataframe(
        f"""
            SELECT bus_id, geom
            FROM grid.egon_etrago_bus
            WHERE scn_name = '{scn_name}'
            AND country = 'DE'
            AND carrier = '{carrier}';
        """,
    ).to_crs(epsg=4326)

    if len(buses) == 0:
        return

    buses["x"] = buses.geometry.x
    buses["y"] = buses.geometry.y
    # generate voronois
    gdf = get_voronoi_geodataframe(buses, boundary.geometry.iloc[0])
    # set scn_name
    gdf["scn_name"] = scn_name
    gdf["carrier"] = carrier

    gdf.rename_geometry("geom", inplace=True)
    gdf.drop(columns=["id"], inplace=True)
    # Insert data to db
    gdf.set_crs(epsg=4326).to_postgis(
        f"egon_gas_voronoi",
        engine,
        schema="grid",
        index=False,
        if_exists="append",
        dtype={"geom": Geometry},
    )


1			"""The central module containing code to create CH4 and H2 voronoi polygons
2
3			"""
4			from geoalchemy2.types import Geometry
5			from sqlalchemy import BigInteger, Column, Text
6			from sqlalchemy.ext.declarative import declarative_base
7
8			from egon.data import db
9			from egon.data.datasets import Dataset
10			from egon.data.datasets.generate_voronoi import get_voronoi_geodataframe
11
12
13			class GasAreaseGon2035(Dataset):
14			def __init__(self, dependencies):
15			super().__init__(
16			name="GasAreaseGon2035",
17			version="0.0.2",
18			dependencies=dependencies,
19			tasks=(create_gas_voronoi_table, voronoi_egon2035),
20			)
21
22
23			class GasAreaseGon100RE(Dataset):
24			def __init__(self, dependencies):
25			super().__init__(
26			name="GasAreaseGon100RE",
27			version="0.0.1",
28			dependencies=dependencies,
29			tasks=(voronoi_egon100RE),
30			)
31
32
33			Base = declarative_base()
34
35
36			class EgonPfHvGasVoronoi(Base):
37			__tablename__ = "egon_gas_voronoi"
38			__table_args__ = {"schema": "grid"}
39
40			scn_name = Column(Text, primary_key=True, nullable=False)
41			bus_id = Column(BigInteger, primary_key=True, nullable=False)
42			carrier = Column(Text)
43			geom = Column(Geometry("GEOMETRY", 4326))
44
45
46			def create_gas_voronoi_table():
47			engine = db.engine()
48			EgonPfHvGasVoronoi.__table__.drop(bind=engine, checkfirst=True)
49			EgonPfHvGasVoronoi.__table__.create(bind=engine, checkfirst=True)
50
51
52			def voronoi_egon2035():
53			"""
54			Create voronoi polygons for all gas carriers in eGon2035 scenario
55			"""
56			for carrier in ["CH4", "H2_grid", "H2_saltcavern"]:
57			create_voronoi("eGon2035", carrier)
58
59
60			def voronoi_egon100RE():
61			"""
62			Create voronoi polygons for all gas carriers in eGon100RE scenario
63			"""
64			for carrier in ["CH4", "H2_grid", "H2_saltcavern"]:
65			create_voronoi("eGon100RE", carrier)
66
67
68			def create_voronoi(scn_name, carrier):
69			"""
70			Create voronoi polygons for specified carrier in specified scenario.
71
72			Parameters
73			----------
74			scn_name : str
75			Name of the scenario
76			carrier : str
77			Name of the carrier
78			"""
79			boundary = db.select_geodataframe(
80			f"""
81			SELECT id, geometry
82			FROM boundaries.vg250_sta_union;
83			""",
84			geom_col="geometry",
85			).to_crs(epsg=4326)
86
87			engine = db.engine()
88
89			db.execute_sql(
90			f"""
91			DELETE FROM grid.egon_gas_voronoi
92			WHERE "carrier" = '{carrier}' and "scn_name" = '{scn_name}';
93			"""
94			)
95
96			buses = db.select_geodataframe(
97			f"""
98			SELECT bus_id, geom
99			FROM grid.egon_etrago_bus
100			WHERE scn_name = '{scn_name}'
101			AND country = 'DE'
102			AND carrier = '{carrier}';
103			""",
104			).to_crs(epsg=4326)
105
106			if len(buses) == 0:
107			return
108
109			buses["x"] = buses.geometry.x
110			buses["y"] = buses.geometry.y
111			# generate voronois
112			gdf = get_voronoi_geodataframe(buses, boundary.geometry.iloc[0])
113			# set scn_name
114			gdf["scn_name"] = scn_name
115			gdf["carrier"] = carrier
116
117			gdf.rename_geometry("geom", inplace=True)
118			gdf.drop(columns=["id"], inplace=True)
119			# Insert data to db
120			gdf.set_crs(epsg=4326).to_postgis(
121			f"egon_gas_voronoi",
122			engine,
123			schema="grid",
124			index=False,
125			if_exists="append",
126			dtype={"geom": Geometry},
127			)
128

openego / eGon-data

Pull Request — dev (#891)

data.datasets.gas_areas.create_voronoi() A

Complexity

Size

Duplication

Importance

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like