Source code for egon.data.datasets.re_potential_areas

"""The central module containing all code dealing with importing data on
potential areas for wind onshore and ground-mounted PV.
"""

from pathlib import Path

from geoalchemy2 import Geometry
from sqlalchemy import Column, Integer
from sqlalchemy.ext.declarative import declarative_base
import geopandas as gpd

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

Base = declarative_base()


class EgonRePotentialAreaPvAgriculture(Base):
    """
    Class definition of table supply.egon_re_potential_area_pv_agriculture.

    """
    __tablename__ = "egon_re_potential_area_pv_agriculture"
    __table_args__ = {"schema": "supply"}
    id = Column(Integer, primary_key=True)
    geom = Column(Geometry("MULTIPOLYGON", 3035))


class EgonRePotentialAreaPvRoadRailway(Base):
    """
    Class definition of table supply.egon_re_potential_area_pv_road_railway.

    """
    __tablename__ = "egon_re_potential_area_pv_road_railway"
    __table_args__ = {"schema": "supply"}
    id = Column(Integer, primary_key=True)
    geom = Column(Geometry("MULTIPOLYGON", 3035))


class EgonRePotentialAreaWind(Base):
    """
    Class definition of table supply.egon_re_potential_area_wind.

    """
    __tablename__ = "egon_re_potential_area_wind"
    __table_args__ = {"schema": "supply"}
    id = Column(Integer, primary_key=True)
    geom = Column(Geometry("MULTIPOLYGON", 3035))


def create_tables():
    """Create tables for RE potential areas"""

    data_config = egon.data.config.datasets()
    schema = data_config["re_potential_areas"]["target"].get(
        "schema", "supply"
    )

    db.execute_sql(f"CREATE SCHEMA IF NOT EXISTS {schema};")
    engine = db.engine()

    # drop tables
    EgonRePotentialAreaPvAgriculture.__table__.drop(engine, checkfirst=True)
    EgonRePotentialAreaPvRoadRailway.__table__.drop(engine, checkfirst=True)
    EgonRePotentialAreaWind.__table__.drop(engine, checkfirst=True)

    # create tables
    EgonRePotentialAreaPvAgriculture.__table__.create(
        bind=engine, checkfirst=True
    )
    EgonRePotentialAreaPvRoadRailway.__table__.create(
        bind=engine, checkfirst=True
    )
    EgonRePotentialAreaWind.__table__.create(bind=engine, checkfirst=True)


def insert_data():
    """Insert data into DB"""

    data_bundle_dir = Path(
        ".",
        "data_bundle_egon_data",
        "re_potential_areas",
    )

    dataset = egon.data.config.settings()["egon-data"]["--dataset-boundary"]
    if dataset == "Everything":
        map_section = "path_table_map"
    elif dataset == "Schleswig-Holstein":
        map_section = "path_table_map_testmode"
    else:
        raise ValueError(f"'{dataset}' is not a valid dataset boundary.")

    data_config = egon.data.config.datasets()
    pa_config = data_config["re_potential_areas"]

    file_table_map = {
        data_bundle_dir / Path(file).name: table
        for file, table in pa_config["target"][map_section].items()
    }

    engine = db.engine()

    for file, table in file_table_map.items():
        data = gpd.read_file(file).to_crs("EPSG:3035")
        data.rename(columns={"geometry": "geom"}, inplace=True)
        data.set_geometry("geom", inplace=True)

        schema = pa_config["target"].get("schema", "supply")

        # create database table from geopandas dataframe
        data[["id", "geom"]].to_postgis(
            table,
            engine,
            schema=schema,
            index=False,
            if_exists="append",
            dtype={"geom": Geometry()},
        )


class re_potential_area_setup(Dataset):
    """
    Downloads potential areas for PV and wind power plants from data bundle and
    writes them to the database.

    *Dependencies*
      * :py:func:`Setup <egon.data.datasets.database.setup>`
      * :py:class:`DataBundle <egon.data.datasets.data_bundle.DataBundle>`

    *Resulting Tables*
      * :py:class:`EgonRePotentialAreaPvAgriculture <egon.data.datasets.re_potential_areas.EgonRePotentialAreaPvAgriculture>`
      * :py:class:`EgonRePotentialAreaPvRoadRailway <egon.data.datasets.re_potential_areas.EgonRePotentialAreaPvRoadRailway>`
      * :py:class:`EgonRePotentialAreaWind <egon.data.datasets.re_potential_areas.EgonRePotentialAreaWind>`

    """
    #:
    name: str = "RePotentialAreas"
    #:
    version: str = "0.0.1"
    #:
    tasks = (create_tables, insert_data)

    def __init__(self, dependencies):
        super().__init__(
            name=self.name,
            version=self.version,
            dependencies=dependencies,
            tasks=self.tasks,
        )