"""
The central module containing all code dealing with combined heat and power
(CHP) plants.
"""
from pathlib import Path
from geoalchemy2 import Geometry
from shapely.ops import nearest_points
from sqlalchemy import Boolean, Column, Float, Integer, Sequence, String
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
import geopandas as gpd
import pandas as pd
import pypsa
from egon.data import config, db
from egon.data.datasets import Dataset
from egon.data.datasets.chp.match_nep import insert_large_chp
from egon.data.datasets.chp.small_chp import (
assign_use_case,
existing_chp_smaller_10mw,
extension_per_federal_state,
extension_to_areas,
select_target,
)
from egon.data.datasets.mastr import WORKING_DIR_MASTR_OLD
from egon.data.datasets.power_plants import (
assign_bus_id,
assign_voltage_level,
filter_mastr_geometry,
scale_prox2now,
)
Base = declarative_base()
[docs]class EgonChp(Base):
__tablename__ = "egon_chp_plants"
__table_args__ = {"schema": "supply"}
id = Column(Integer, Sequence("chp_seq"), primary_key=True)
sources = Column(JSONB)
source_id = Column(JSONB)
carrier = Column(String)
district_heating = Column(Boolean)
el_capacity = Column(Float)
th_capacity = Column(Float)
electrical_bus_id = Column(Integer)
district_heating_area_id = Column(Integer)
ch4_bus_id = Column(Integer)
voltage_level = Column(Integer)
scenario = Column(String)
geom = Column(Geometry("POINT", 4326))
[docs]class EgonMaStRConventinalWithoutChp(Base):
__tablename__ = "egon_mastr_conventional_without_chp"
__table_args__ = {"schema": "supply"}
id = Column(Integer, Sequence("mastr_conventional_seq"), primary_key=True)
EinheitMastrNummer = Column(String)
carrier = Column(String)
el_capacity = Column(Float)
plz = Column(Integer)
city = Column(String)
federal_state = Column(String)
geometry = Column(Geometry("POINT", 4326))
[docs]def create_tables():
"""Create tables for chp data
Returns
-------
None.
"""
db.execute_sql("CREATE SCHEMA IF NOT EXISTS supply;")
engine = db.engine()
EgonChp.__table__.drop(bind=engine, checkfirst=True)
EgonChp.__table__.create(bind=engine, checkfirst=True)
EgonMaStRConventinalWithoutChp.__table__.drop(bind=engine, checkfirst=True)
EgonMaStRConventinalWithoutChp.__table__.create(
bind=engine, checkfirst=True
)
[docs]def nearest(
row,
df,
centroid=False,
row_geom_col="geometry",
df_geom_col="geometry",
src_column=None,
):
"""
Finds the nearest point and returns the specified column values
Parameters
----------
row : pandas.Series
Data to which the nearest data of df is assigned.
df : pandas.DataFrame
Data which includes all options for the nearest neighbor alogrithm.
centroid : boolean
Use centroid geoemtry. The default is False.
row_geom_col : str, optional
Name of row's geometry column. The default is 'geometry'.
df_geom_col : str, optional
Name of df's geometry column. The default is 'geometry'.
src_column : str, optional
Name of returned df column. The default is None.
Returns
-------
value : pandas.Series
Values of specified column of df
"""
if centroid:
unary_union = df.centroid.unary_union
else:
unary_union = df[df_geom_col].unary_union
# Find the geometry that is closest
nearest = (
df[df_geom_col] == nearest_points(row[row_geom_col], unary_union)[1]
)
# Get the corresponding value from df (matching is based on the geometry)
value = df[nearest][src_column].values[0]
return value
[docs]def assign_heat_bus(scenario="eGon2035"):
"""Selects heat_bus for chps used in district heating.
Parameters
----------
scenario : str, optional
Name of the corresponding scenario. The default is 'eGon2035'.
Returns
-------
None.
"""
sources = config.datasets()["chp_location"]["sources"]
target = config.datasets()["chp_location"]["targets"]["chp_table"]
# Select CHP with use_case = 'district_heating'
chp = db.select_geodataframe(
f"""
SELECT * FROM
{target['schema']}.{target['table']}
WHERE scenario = '{scenario}'
AND district_heating = True
""",
index_col="id",
epsg=4326,
)
# Select district heating areas and their centroid
district_heating = db.select_geodataframe(
f"""
SELECT area_id, ST_Centroid(geom_polygon) as geom
FROM
{sources['district_heating_areas']['schema']}.
{sources['district_heating_areas']['table']}
WHERE scenario = '{scenario}'
""",
epsg=4326,
)
# Assign district heating area_id to district_heating_chp
# According to nearest centroid of district heating area
chp["district_heating_area_id"] = chp.apply(
nearest,
df=district_heating,
row_geom_col="geom",
df_geom_col="geom",
centroid=True,
src_column="area_id",
axis=1,
)
# Drop district heating CHP without heat_bus_id
db.execute_sql(
f"""
DELETE FROM {target['schema']}.{target['table']}
WHERE scenario = '{scenario}'
AND district_heating = True
"""
)
# Insert district heating CHP with heat_bus_id
session = sessionmaker(bind=db.engine())()
for i, row in chp.iterrows():
if row.carrier != "biomass":
entry = EgonChp(
id=i,
sources=row.sources,
source_id=row.source_id,
carrier=row.carrier,
el_capacity=row.el_capacity,
th_capacity=row.th_capacity,
electrical_bus_id=row.electrical_bus_id,
ch4_bus_id=row.ch4_bus_id,
district_heating_area_id=row.district_heating_area_id,
district_heating=row.district_heating,
voltage_level=row.voltage_level,
scenario=scenario,
geom=f"SRID=4326;POINT({row.geom.x} {row.geom.y})",
)
else:
entry = EgonChp(
id=i,
sources=row.sources,
source_id=row.source_id,
carrier=row.carrier,
el_capacity=row.el_capacity,
th_capacity=row.th_capacity,
electrical_bus_id=row.electrical_bus_id,
district_heating_area_id=row.district_heating_area_id,
district_heating=row.district_heating,
voltage_level=row.voltage_level,
scenario=scenario,
geom=f"SRID=4326;POINT({row.geom.x} {row.geom.y})",
)
session.add(entry)
session.commit()
[docs]def insert_biomass_chp(scenario):
"""Insert biomass chp plants of future scenario
Parameters
----------
scenario : str
Name of scenario.
Returns
-------
None.
"""
cfg = config.datasets()["chp_location"]
# import target values from NEP 2021, scneario C 2035
target = select_target("biomass", scenario)
# import data for MaStR
mastr = pd.read_csv(
WORKING_DIR_MASTR_OLD / cfg["sources"]["mastr_biomass"]
).query("EinheitBetriebsstatus=='InBetrieb'")
# Drop entries without federal state or 'AusschließlichWirtschaftszone'
mastr = mastr[
mastr.Bundesland.isin(
pd.read_sql(
f"""SELECT DISTINCT ON (gen)
REPLACE(REPLACE(gen, '-', ''), 'ü', 'ue') as states
FROM {cfg['sources']['vg250_lan']['schema']}.
{cfg['sources']['vg250_lan']['table']}""",
con=db.engine(),
).states.values
)
]
# Scaling will be done per federal state in case of eGon2035 scenario.
if scenario == "eGon2035":
level = "federal_state"
else:
level = "country"
# Choose only entries with valid geometries inside DE/test mode
mastr_loc = filter_mastr_geometry(mastr).set_geometry("geometry")
# Scale capacities to meet target values
mastr_loc = scale_prox2now(mastr_loc, target, level=level)
# Assign bus_id
if len(mastr_loc) > 0:
mastr_loc["voltage_level"] = assign_voltage_level(
mastr_loc, cfg, WORKING_DIR_MASTR_OLD
)
mastr_loc = assign_bus_id(mastr_loc, cfg)
mastr_loc = assign_use_case(mastr_loc, cfg["sources"])
# Insert entries with location
session = sessionmaker(bind=db.engine())()
for i, row in mastr_loc.iterrows():
if row.ThermischeNutzleistung > 0:
entry = EgonChp(
sources={
"chp": "MaStR",
"el_capacity": "MaStR scaled with NEP 2021",
"th_capacity": "MaStR",
},
source_id={"MastrNummer": row.EinheitMastrNummer},
carrier="biomass",
el_capacity=row.Nettonennleistung,
th_capacity=row.ThermischeNutzleistung / 1000,
scenario=scenario,
district_heating=row.district_heating,
electrical_bus_id=row.bus_id,
voltage_level=row.voltage_level,
geom=f"SRID=4326;POINT({row.Laengengrad} {row.Breitengrad})",
)
session.add(entry)
session.commit()
[docs]def insert_chp_egon2035():
"""Insert CHP plants for eGon2035 considering NEP and MaStR data
Returns
-------
None.
"""
sources = config.datasets()["chp_location"]["sources"]
targets = config.datasets()["chp_location"]["targets"]
insert_biomass_chp("eGon2035")
# Insert large CHPs based on NEP's list of conventional power plants
MaStR_konv = insert_large_chp(sources, targets["chp_table"], EgonChp)
# Insert smaller CHPs (< 10MW) based on existing locations from MaStR
existing_chp_smaller_10mw(sources, MaStR_konv, EgonChp)
gpd.GeoDataFrame(
MaStR_konv[
[
"EinheitMastrNummer",
"el_capacity",
"geometry",
"carrier",
"plz",
"city",
"federal_state",
]
]
).to_postgis(
targets["mastr_conventional_without_chp"]["table"],
schema=targets["mastr_conventional_without_chp"]["schema"],
con=db.engine(),
if_exists="replace",
)
[docs]def extension_BW():
extension_per_federal_state("BadenWuerttemberg", EgonChp)
[docs]def extension_BY():
extension_per_federal_state("Bayern", EgonChp)
[docs]def extension_HB():
extension_per_federal_state("Bremen", EgonChp)
[docs]def extension_BB():
extension_per_federal_state("Brandenburg", EgonChp)
[docs]def extension_HH():
extension_per_federal_state("Hamburg", EgonChp)
[docs]def extension_HE():
extension_per_federal_state("Hessen", EgonChp)
[docs]def extension_MV():
extension_per_federal_state("MecklenburgVorpommern", EgonChp)
[docs]def extension_NS():
extension_per_federal_state("Niedersachsen", EgonChp)
[docs]def extension_NW():
extension_per_federal_state("NordrheinWestfalen", EgonChp)
[docs]def extension_SN():
extension_per_federal_state("Sachsen", EgonChp)
[docs]def extension_TH():
extension_per_federal_state("Thueringen", EgonChp)
[docs]def extension_SL():
extension_per_federal_state("Saarland", EgonChp)
[docs]def extension_ST():
extension_per_federal_state("SachsenAnhalt", EgonChp)
[docs]def extension_RP():
extension_per_federal_state("RheinlandPfalz", EgonChp)
[docs]def extension_BE():
extension_per_federal_state("Berlin", EgonChp)
[docs]def extension_SH():
extension_per_federal_state("SchleswigHolstein", EgonChp)
[docs]def insert_chp_egon100re():
"""Insert CHP plants for eGon100RE considering results from pypsa-eur-sec
Returns
-------
None.
"""
sources = config.datasets()["chp_location"]["sources"]
db.execute_sql(
f"""
DELETE FROM {EgonChp.__table__.schema}.{EgonChp.__table__.name}
WHERE scenario = 'eGon100RE'
"""
)
# select target values from pypsa-eur-sec
additional_capacity = db.select_dataframe(
"""
SELECT capacity
FROM supply.egon_scenario_capacities
WHERE scenario_name = 'eGon100RE'
AND carrier = 'urban_central_gas_CHP'
"""
).capacity[0]
if config.settings()["egon-data"]["--dataset-boundary"] != "Everything":
additional_capacity /= 16
target_file = (
Path(".")
/ "data_bundle_egon_data"
/ "pypsa_eur_sec"
/ "2022-07-26-egondata-integration"
/ "postnetworks"
/ "elec_s_37_lv2.0__Co2L0-1H-T-H-B-I-dist1_2050.nc"
)
network = pypsa.Network(str(target_file))
chp_index = "DE0 0 urban central gas CHP"
standard_chp_th = 10
standard_chp_el = (
standard_chp_th
* network.links.loc[chp_index, "efficiency"]
/ network.links.loc[chp_index, "efficiency2"]
)
areas = db.select_geodataframe(
f"""
SELECT
residential_and_service_demand as demand, area_id,
ST_Transform(ST_PointOnSurface(geom_polygon), 4326) as geom
FROM
{sources['district_heating_areas']['schema']}.
{sources['district_heating_areas']['table']}
WHERE scenario = 'eGon100RE'
"""
)
existing_chp = pd.DataFrame(
data={
"el_capacity": standard_chp_el,
"th_capacity": standard_chp_th,
"voltage_level": 5,
},
index=range(1),
)
flh = (
network.links_t.p0[chp_index].sum()
/ network.links.p_nom_opt[chp_index]
)
extension_to_areas(
areas,
additional_capacity,
existing_chp,
flh,
EgonChp,
district_heating=True,
scenario="eGon100RE",
)
# Add one task per federal state for small CHP extension
if (
config.settings()["egon-data"]["--dataset-boundary"]
== "Schleswig-Holstein"
):
extension = extension_SH
else:
extension = {
extension_BW,
extension_BY,
extension_HB,
extension_BB,
extension_HE,
extension_MV,
extension_NS,
extension_NW,
extension_SH,
extension_HH,
extension_RP,
extension_SL,
extension_SN,
extension_ST,
extension_TH,
extension_BE,
}
[docs]class Chp(Dataset):
"""
Extract combined heat and power plants for each scenario
This dataset creates combined heat and power (CHP) plants for each scenario and defines their use case.
The method bases on existing CHP plants from Marktstammdatenregister. For the eGon2035 scenario,
a list of CHP plans from the grid operator is used for new largescale CHP plants. CHP < 10MW are
randomly distributed.
Depending on the distance to a district heating grid, it is decided if the CHP is used to
supply a district heating grid or used by an industrial site.
*Dependencies*
* :py:class:`GasAreaseGon100RE <egon.data.datasets.gas_areas.GasAreaseGon100RE>`
* :py:class:`GasAreaseGon2035 <egon.data.datasets.gas_areas.GasAreaseGon2035>`
* :py:class:`DistrictHeatingAreas <egon.data.datasets.district_heating_areas.DistrictHeatingAreas>`
* :py:class:`IndustrialDemandCurves <egon.data.datasets.industry.IndustrialDemandCurves>`
* :py:class:`OsmLanduse <egon.data.datasets.loadarea.OsmLanduse>`
* :py:func:`download_mastr_data <egon.data.datasets.mastr.download_mastr_data>`
* :py:func:`define_mv_grid_districts <egon.data.datasets.mv_grid_districts.define_mv_grid_districts>`
* :py:class:`ScenarioCapacities <egon.data.datasets.scenario_capacities.ScenarioCapacities>`
*Resulting tables*
* :py:class:`supply.egon_chp_plants <egon.data.datasets.chp.EgonChp>` is created and filled
* :py:class:`supply.egon_mastr_conventional_without_chp <egon.data.datasets.chp.EgonMaStRConventinalWithoutChp>` is created and filled
"""
#:
name: str = "Chp"
#:
version: str = "0.0.6"
def __init__(self, dependencies):
super().__init__(
name=self.name,
version=self.version,
dependencies=dependencies,
tasks=(
create_tables,
{insert_chp_egon2035, insert_chp_egon100re},
assign_heat_bus,
extension,
),
)