"""The module containing all code to map MV grid districts to federal states."""
from sqlalchemy import Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
import geopandas as gpd
import pandas as pd
from egon.data import db
Base = declarative_base()
from egon.data.datasets import Dataset, DatasetSources, DatasetTargets
[docs]
class Vg250MvGridDistricts(Dataset):
"""
Maps MV grid districts to federal states and writes it to database.
*Dependencies*
* :py:class:`Vg250 <egon.data.datasets.vg250.Vg250>`
* :py:class:`MvGridDistricts <egon.data.datasets.mv_grid_districts.mv_grid_districts_setup>`
*Resulting tables*
* :py:class:`boundaries.egon_map_mvgriddistrict_vg250 <MapMvgriddistrictsVg250>`
is created and filled
"""
#:
name: str = "Vg250MvGridDistricts"
#:
version: str = "0.0.3"
sources = DatasetSources(
tables={
"egon_mv_grid_district": "grid.egon_mv_grid_district",
"federal_states": "boundaries.vg250_lan_union",
}
)
targets = DatasetTargets(
tables={
"map": "boundaries.egon_map_mvgriddistrict_vg250",
}
)
def __init__(self, dependencies):
super().__init__(
name=self.name,
version=self.version,
dependencies=dependencies,
tasks=(mapping),
)
[docs]
class MapMvgriddistrictsVg250(Base):
"""
Class definition of table boundaries.egon_map_mvgriddistrict_vg250.
"""
__tablename__ = "egon_map_mvgriddistrict_vg250"
__table_args__ = {"schema": "boundaries"}
bus_id = Column(Integer, primary_key=True)
vg250_lan = Column(String)
[docs]
def create_tables():
"""
Create table for mapping grid districts to federal states.
"""
db.execute_sql(
f"CREATE SCHEMA IF NOT EXISTS {Vg250MvGridDistricts.targets.get_table_schema('map')};"
)
engine = db.engine()
MapMvgriddistrictsVg250.__table__.drop(bind=engine, checkfirst=True)
MapMvgriddistrictsVg250.__table__.create(bind=engine, checkfirst=True)
[docs]
def mapping():
"""
Map MV grid districts to federal states and write to database.
Newly creates and fills table boundaries.egon_map_mvgriddistrict_vg250.
"""
# Create table
create_tables()
# Select sources and targets from dataset definition
sources = Vg250MvGridDistricts.sources
targets = Vg250MvGridDistricts.targets
# Delete existing data
db.execute_sql(f"DELETE FROM {targets.tables['map']}")
# Select sources from database
mv_grid_districts = db.select_geodataframe(
f"""
SELECT bus_id as bus_id, ST_Centroid(geom) as geom
FROM {sources.tables['egon_mv_grid_district']}
""",
index_col="bus_id",
)
federal_states = db.select_geodataframe(
f"""
SELECT gen,geometry
FROM {sources.tables['federal_states']}
""",
geom_col="geometry",
index_col="gen",
)
# Join mv grid districts and federal states
df = pd.DataFrame(gpd.sjoin(mv_grid_districts, federal_states)["gen"])
# Rename columns
df.rename({"gen": "vg250_lan"}, axis=1, inplace=True)
# Insert to database
df.to_sql(
targets.get_table_name("map"),
schema=targets.get_table_schema("map"),
if_exists="append",
con=db.engine(),
)