data.datasets.loadarea.OsmLanduse.__init__() - Code Metrics - Inspection of "Features/#1014 load areas" - openego/eGon-data - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — dev (#1020)

unknown

created 2022-11-29 11:14 UTC

data.datasets.loadarea.OsmLanduse.init() A

↳ Parent: data.datasets.loadarea

Complexity

Conditions

Size

Total Lines	14
Code Lines	13

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	13
dl	0
loc	14
rs	9.75
c	0
b	0
f	0
cc	1
nop	2

"""
OSM landuse extraction and load areas creation.

**Landuse**

* Landuse data is extracted from OpenStreetMap: residential, retail,
  industrial, Agricultural
* Data is cut with German borders (VG 250), data outside is dropped
* Invalid geometries are fixed
* Results are stored in table `openstreetmap.osm_landuse`

**Load Areas**

TBD
"""

import os

from airflow.operators.postgres_operator import PostgresOperator
from geoalchemy2.types import Geometry
from sqlalchemy import Column, Float, Integer, String
from sqlalchemy.dialects.postgresql import HSTORE
from sqlalchemy.ext.declarative import declarative_base
import importlib_resources as resources

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

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


class OsmPolygonUrban(Base):
    __tablename__ = "osm_landuse"
    __table_args__ = {"schema": "openstreetmap"}
    id = Column(Integer, primary_key=True)
    osm_id = Column(Integer)
    name = Column(String)
    sector = Column(Integer)
    sector_name = Column(String(20))
    area_ha = Column(Float)
    tags = Column(HSTORE)
    vg250 = Column(String(10))
    geom = Column(Geometry("MultiPolygon", 3035))


class OsmLanduse(Dataset):
    def __init__(self, dependencies):
        super().__init__(
            name="OsmLanduse",
            version="0.0.0",
            dependencies=dependencies,
            tasks=(
                create_landuse_table,
                PostgresOperator(
                    task_id="osm_landuse_extraction",
                    sql=resources.read_text(
                        __name__, "osm_landuse_extraction.sql"
                    ),
                    postgres_conn_id="egon_data",
                    autocommit=True,
                ),
            ),
        )


class LoadArea(Dataset):
    def __init__(self, dependencies):
        super().__init__(
            name="LoadArea",
            version="0.0.1",
            dependencies=dependencies,
            tasks=(
                osm_landuse_melt,
                census_cells_melt,
                osm_landuse_census_cells_melt,
                loadareas_create,
                {
                    loadareas_add_demand_hh,
                    loadareas_add_demand_cts,
                    loadareas_add_demand_ind,
                },
                drop_temp_tables,
            ),
        )


def create_landuse_table():
    """Create tables for landuse data
    Returns
    -------
    None.
    """
    cfg = egon.data.config.datasets()["landuse"]["target"]

    # Create schema if not exists
    db.execute_sql(f"""CREATE SCHEMA IF NOT EXISTS {cfg['schema']};""")

    # Drop tables
    db.execute_sql(
        f"""DROP TABLE IF EXISTS
            {cfg['schema']}.{cfg['table']} CASCADE;"""
    )

    engine = db.engine()
    OsmPolygonUrban.__table__.create(bind=engine, checkfirst=True)


def execute_sql_script(script):
    """Execute SQL script

    Parameters
    ----------
    script : str
        Filename of script
    """
    db.execute_sql_script(os.path.join(os.path.dirname(__file__), script))


def osm_landuse_melt():
    """Melt all OSM landuse areas by: buffer, union, unbuffer"""
    print("Melting OSM landuse areas from openstreetmap.osm_landuse...")
    execute_sql_script("osm_landuse_melt.sql")


def census_cells_melt():
    """Melt all census cells: buffer, union, unbuffer"""
    print(
        "Melting census cells from "
        "society.destatis_zensus_population_per_ha_inside_germany..."
    )
    execute_sql_script("census_cells_melt.sql")


def osm_landuse_census_cells_melt():
    """Melt OSM landuse areas and census cells"""
    print(
        "Melting OSM landuse areas from openstreetmap.osm_landuse_melted and "
        "census cells from "
        "society.egon_destatis_zensus_cells_melted_cluster..."
    )
    execute_sql_script("osm_landuse_census_cells_melt.sql")


def loadareas_create():
    """Create load areas from merged OSM landuse and census cells:

    * Cut Loadarea with MV Griddistrict
    * Identify and exclude Loadarea smaller than 100m².
    * Generate Centre of Loadareas with Centroid and PointOnSurface.
    * Calculate population from Census 2011.
    * Cut all 4 OSM sectors with MV Griddistricts.
    * Calculate statistics like NUTS and AGS code.
    * Check for Loadareas without AGS code.
    """
    print("Create initial load areas and add some sector stats...")
    execute_sql_script("loadareas_create.sql")


def loadareas_add_demand_hh():
    """Adds consumption and peak load to load areas for households"""
    print("Add consumption and peak loads to load areas for households...")
    execute_sql_script("loadareas_add_demand_hh.sql")


def loadareas_add_demand_cts():
    """Adds consumption and peak load to load areas for CTS"""
    print("Add consumption and peak loads to load areas for CTS...")
    execute_sql_script("loadareas_add_demand_cts.sql")


def loadareas_add_demand_ind():
    """Adds consumption and peak load to load areas for industry"""
    print("Add consumption and peak loads to load areas for industry...")
    execute_sql_script("loadareas_add_demand_ind.sql")


def drop_temp_tables():
    print("Dropping temp tables, views and sequences...")
    execute_sql_script("drop_temp_tables.sql")


1			"""
2			OSM landuse extraction and load areas creation.
3
4			Landuse
5
6			* Landuse data is extracted from OpenStreetMap: residential, retail,
7			industrial, Agricultural
8			* Data is cut with German borders (VG 250), data outside is dropped
9			* Invalid geometries are fixed
10			* Results are stored in table `openstreetmap.osm_landuse`
11
12			Load Areas
13
14			TBD
15			"""
16
17			import os
18
19			from airflow.operators.postgres_operator import PostgresOperator
20			from geoalchemy2.types import Geometry
21			from sqlalchemy import Column, Float, Integer, String
22			from sqlalchemy.dialects.postgresql import HSTORE
23			from sqlalchemy.ext.declarative import declarative_base
24			import importlib_resources as resources
25
26			from egon.data import db
27			from egon.data.datasets import Dataset
28			import egon.data.config
29
30			# will be later imported from another file ###
31			Base = declarative_base()
32
33
34			class OsmPolygonUrban(Base):
35			__tablename__ = "osm_landuse"
36			__table_args__ = {"schema": "openstreetmap"}
37			id = Column(Integer, primary_key=True)
38			osm_id = Column(Integer)
39			name = Column(String)
40			sector = Column(Integer)
41			sector_name = Column(String(20))
42			area_ha = Column(Float)
43			tags = Column(HSTORE)
44			vg250 = Column(String(10))
45			geom = Column(Geometry("MultiPolygon", 3035))
46
47
48			class OsmLanduse(Dataset):
49			def __init__(self, dependencies):
50			super().__init__(
51			name="OsmLanduse",
52			version="0.0.0",
53			dependencies=dependencies,
54			tasks=(
55			create_landuse_table,
56			PostgresOperator(
57			task_id="osm_landuse_extraction",
58			sql=resources.read_text(
59			__name__, "osm_landuse_extraction.sql"
60			),
61			postgres_conn_id="egon_data",
62			autocommit=True,
63			),
64			),
65			)
66
67
68			class LoadArea(Dataset):
69			def __init__(self, dependencies):
70			super().__init__(
71			name="LoadArea",
72			version="0.0.1",
73			dependencies=dependencies,
74			tasks=(
75			osm_landuse_melt,
76			census_cells_melt,
77			osm_landuse_census_cells_melt,
78			loadareas_create,
79			{
80			loadareas_add_demand_hh,
81			loadareas_add_demand_cts,
82			loadareas_add_demand_ind,
83			},
84			drop_temp_tables,
85			),
86			)
87
88
89			def create_landuse_table():
90			"""Create tables for landuse data
91			Returns
92			-------
93			None.
94			"""
95			cfg = egon.data.config.datasets()["landuse"]["target"]
96
97			# Create schema if not exists
98			db.execute_sql(f"""CREATE SCHEMA IF NOT EXISTS {cfg['schema']};""")
99
100			# Drop tables
101			db.execute_sql(
102			f"""DROP TABLE IF EXISTS
103			{cfg['schema']}.{cfg['table']} CASCADE;"""
104			)
105
106			engine = db.engine()
107			OsmPolygonUrban.__table__.create(bind=engine, checkfirst=True)
108
109
110			def execute_sql_script(script):
111			"""Execute SQL script
112
113			Parameters
114			----------
115			script : str
116			Filename of script
117			"""
118			db.execute_sql_script(os.path.join(os.path.dirname(__file__), script))
119
120
121			def osm_landuse_melt():
122			"""Melt all OSM landuse areas by: buffer, union, unbuffer"""
123			print("Melting OSM landuse areas from openstreetmap.osm_landuse...")
124			execute_sql_script("osm_landuse_melt.sql")
125
126
127			def census_cells_melt():
128			"""Melt all census cells: buffer, union, unbuffer"""
129			print(
130			"Melting census cells from "
131			"society.destatis_zensus_population_per_ha_inside_germany..."
132			)
133			execute_sql_script("census_cells_melt.sql")
134
135
136			def osm_landuse_census_cells_melt():
137			"""Melt OSM landuse areas and census cells"""
138			print(
139			"Melting OSM landuse areas from openstreetmap.osm_landuse_melted and "
140			"census cells from "
141			"society.egon_destatis_zensus_cells_melted_cluster..."
142			)
143			execute_sql_script("osm_landuse_census_cells_melt.sql")
144
145
146			def loadareas_create():
147			"""Create load areas from merged OSM landuse and census cells:
148
149			* Cut Loadarea with MV Griddistrict
150			* Identify and exclude Loadarea smaller than 100m².
151			* Generate Centre of Loadareas with Centroid and PointOnSurface.
152			* Calculate population from Census 2011.
153			* Cut all 4 OSM sectors with MV Griddistricts.
154			* Calculate statistics like NUTS and AGS code.
155			* Check for Loadareas without AGS code.
156			"""
157			print("Create initial load areas and add some sector stats...")
158			execute_sql_script("loadareas_create.sql")
159
160
161			def loadareas_add_demand_hh():
162			"""Adds consumption and peak load to load areas for households"""
163			print("Add consumption and peak loads to load areas for households...")
164			execute_sql_script("loadareas_add_demand_hh.sql")
165
166
167			def loadareas_add_demand_cts():
168			"""Adds consumption and peak load to load areas for CTS"""
169			print("Add consumption and peak loads to load areas for CTS...")
170			execute_sql_script("loadareas_add_demand_cts.sql")
171
172
173			def loadareas_add_demand_ind():
174			"""Adds consumption and peak load to load areas for industry"""
175			print("Add consumption and peak loads to load areas for industry...")
176			execute_sql_script("loadareas_add_demand_ind.sql")
177
178
179			def drop_temp_tables():
180			print("Dropping temp tables, views and sequences...")
181			execute_sql_script("drop_temp_tables.sql")
182

openego / eGon-data

Pull Request — dev (#1020)

data.datasets.loadarea.OsmLanduse.__init__() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

data.datasets.loadarea.OsmLanduse.init() A