"""The central module containing all code dealing with heat sector in etrago"""
import geopandas as gpd
import pandas as pd
from egon.data import config, db
from egon.data.datasets import Dataset, DatasetSources, DatasetTargets
from egon.data.datasets.etrago_setup import link_geom_from_buses
from egon.data.datasets.heat_etrago.power_to_heat import (
insert_central_power_to_heat,
insert_individual_power_to_heat,
)
from egon.data.datasets.scenario_parameters import get_sector_parameters
[docs]
def insert_buses(carrier, scenario):
"""Insert heat buses to etrago table
Heat buses are divided into central and individual heating
Parameters
----------
carrier : str
Name of the carrier, either 'central_heat' or 'rural_heat'
scenario : str, optional
Name of the scenario.
"""
sources = HeatEtrago.sources
targets = HeatEtrago.targets
# Delete existing heat buses (central or rural)
db.execute_sql(f"""
DELETE FROM {targets.tables["heat_buses"]}
WHERE scn_name = '{scenario}'
AND carrier = '{carrier}'
AND country = 'DE'
""")
# initalize dataframe for heat buses
heat_buses = (
gpd.GeoDataFrame(
columns=["scn_name", "bus_id", "carrier", "x", "y", "geom"]
)
.set_geometry("geom")
.set_crs(epsg=4326)
)
# If central heat, create one bus per district heating area
if carrier == "central_heat":
areas = db.select_geodataframe(
f"""
SELECT area_id, geom_polygon as geom
FROM {sources.tables["district_heating_areas"]}
WHERE scenario = '{scenario}'
""",
index_col="area_id",
)
heat_buses.geom = areas.centroid.to_crs(epsg=4326)
# otherwise create one heat bus per hvmv substation
# which represents aggregated individual heating for etrago
else:
mv_grids = db.select_geodataframe(f"""
SELECT ST_Centroid(geom) AS geom
FROM {sources.tables["mv_grids"]}
WHERE bus_id IN
(SELECT DISTINCT bus_id
FROM {sources.tables["map_zensus_grid_districts"]} a
JOIN {sources.tables["heat_demand"]} b
ON a.zensus_population_id = b.zensus_population_id
WHERE b.scenario = '{scenario}'
AND b.zensus_population_id NOT IN (
SELECT zensus_population_id
FROM {sources.tables["map_district_heating_areas"]}
WHERE scenario = '{scenario}'
)
)
""")
heat_buses.geom = mv_grids.geom.to_crs(epsg=4326)
# Insert values into dataframe
heat_buses.scn_name = scenario
heat_buses.carrier = carrier
heat_buses.x = heat_buses.geom.x
heat_buses.y = heat_buses.geom.y
heat_buses.bus_id = db.next_etrago_id("bus", len(heat_buses.index))
# Insert data into database
heat_buses.to_postgis(
targets.get_table_name("heat_buses"),
schema=targets.get_table_schema("heat_buses"),
if_exists="append",
con=db.engine(),
)
[docs]
def insert_store(scenario, carrier):
sources = HeatEtrago.sources
targets = HeatEtrago.targets
db.execute_sql(f"""
DELETE FROM {targets.tables["heat_buses"]}
WHERE carrier = '{carrier}_store'
AND scn_name = '{scenario}'
AND country = 'DE'
""")
db.execute_sql(f"""
DELETE FROM {targets.tables["heat_links"]}
WHERE carrier LIKE '{carrier}_store%'
AND scn_name = '{scenario}'
AND bus0 IN
(SELECT bus_id
FROM {targets.tables["heat_buses"]}
WHERE scn_name = '{scenario}'
AND country = 'DE')
AND bus1 IN
(SELECT bus_id
FROM {targets.tables["heat_buses"]}
WHERE scn_name = '{scenario}'
AND country = 'DE')
""")
db.execute_sql(f"""
DELETE FROM {targets.tables["heat_stores"]}
WHERE carrier = '{carrier}_store'
AND scn_name = '{scenario}'
AND bus IN
(SELECT bus_id
FROM {targets.tables["heat_buses"]}
WHERE scn_name = '{scenario}'
AND country = 'DE')
""")
dh_bus = db.select_geodataframe(
f"""
SELECT * FROM {targets.tables["heat_buses"]}
WHERE carrier = '{carrier}'
AND scn_name = '{scenario}'
AND country = 'DE'
""",
epsg=4326,
)
water_tank_bus = dh_bus.copy()
water_tank_bus.carrier = carrier + "_store"
water_tank_bus.bus_id = db.next_etrago_id("bus", len(water_tank_bus.index))
water_tank_bus.to_postgis(
targets.get_table_name("heat_buses"),
schema=targets.get_table_schema("heat_buses"),
con=db.engine(),
if_exists="append",
index=False,
)
water_tank_charger = pd.DataFrame(
data={
"scn_name": scenario,
"bus0": dh_bus.bus_id,
"bus1": water_tank_bus.bus_id,
"carrier": carrier + "_store_charger",
"efficiency": get_sector_parameters("heat", scenario)[
"efficiency"
]["water_tank_charger"],
"marginal_cost": get_sector_parameters("heat", scenario)[
"marginal_cost"
]["water_tank_charger"],
"p_nom_extendable": True,
"link_id": db.next_etrago_id("link", len(water_tank_bus.index)),
}
)
water_tank_charger.to_sql(
targets.get_table_name("heat_links"),
schema=targets.get_table_schema("heat_links"),
con=db.engine(),
if_exists="append",
index=False,
)
water_tank_discharger = pd.DataFrame(
data={
"scn_name": scenario,
"bus0": water_tank_bus.bus_id,
"bus1": dh_bus.bus_id,
"carrier": carrier + "_store_discharger",
"efficiency": get_sector_parameters("heat", scenario)[
"efficiency"
]["water_tank_discharger"],
"marginal_cost": get_sector_parameters("heat", scenario)[
"marginal_cost"
]["water_tank_discharger"],
"p_nom_extendable": True,
"link_id": db.next_etrago_id("link", len(water_tank_bus.index)),
}
)
water_tank_discharger.to_sql(
targets.get_table_name("heat_links"),
schema=targets.get_table_schema("heat_links"),
con=db.engine(),
if_exists="append",
index=False,
)
water_tank_store = pd.DataFrame(
data={
"scn_name": scenario,
"bus": water_tank_bus.bus_id,
"carrier": carrier + "_store",
"capital_cost": get_sector_parameters("heat", scenario)[
"capital_cost"
][f"{carrier.split('_')[0]}_water_tank"],
"lifetime": get_sector_parameters("heat", scenario)["lifetime"][
f"{carrier.split('_')[0]}_water_tank"
],
"e_nom_extendable": True,
"store_id": db.next_etrago_id("store", len(water_tank_bus.index)),
}
)
water_tank_store.to_sql(
targets.get_table_name("heat_stores"),
schema=targets.get_table_schema("heat_stores"),
con=db.engine(),
if_exists="append",
index=False,
)
[docs]
def store():
for scenario in config.settings()["egon-data"]["--scenarios"]:
if "status" not in scenario:
insert_store(scenario, "central_heat")
insert_store(scenario, "rural_heat")
[docs]
def insert_rural_direct_heat(scenario):
"""Insert renewable heating technologies (solar thermal)
Parameters
----------
scenario : str
Name of the scenario
Returns
-------
None.
"""
sources = HeatEtrago.sources
targets = HeatEtrago.targets
db.execute_sql(f"""
DELETE FROM {targets.tables["heat_generators"]}
WHERE carrier IN ('rural_solar_thermal')
AND scn_name = '{scenario}'
AND bus IN
(SELECT bus_id
FROM {targets.tables["heat_buses"]}
WHERE scn_name = '{scenario}'
AND country = 'DE')
""")
db.execute_sql(f"""
DELETE FROM {targets.tables["heat_generator_timeseries"]}
WHERE scn_name = '{scenario}'
AND generator_id NOT IN (
SELECT generator_id
FROM {targets.tables["heat_generators"]}
WHERE scn_name = '{scenario}')
""")
rural_solar_thermal = db.select_geodataframe(
f"""
SELECT mv_grid_id as power_bus,
a.carrier, capacity, b.bus_id as heat_bus, geom as geometry
FROM {sources.tables["individual_heating_supply"]} a
JOIN {targets.tables["heat_buses"]} b
ON ST_Intersects(
ST_Buffer(ST_Transform(ST_Centroid(a.geometry), 4326), 0.00000001),
geom)
WHERE scenario = '{scenario}'
AND scn_name = '{scenario}'
AND a.carrier = 'solar_thermal'
AND b.carrier = 'rural_heat'
""",
geom_col="geometry",
)
if rural_solar_thermal.empty:
print(f"No rural solar thermal in scenario {scenario}.")
return
generator = pd.DataFrame(
data={
"scn_name": scenario,
"carrier": "rural_solar_thermal",
"bus": rural_solar_thermal.heat_bus,
"p_nom": rural_solar_thermal.capacity,
"generator_id": db.next_etrago_id(
"generator", len(rural_solar_thermal.index)
),
}
)
weather_cells = db.select_geodataframe(
f"""
SELECT w_id, geom
FROM {sources.tables["weather_cells"]}
""",
index_col="w_id",
)
# Map solar thermal collectors to weather cells
join = gpd.sjoin(weather_cells, rural_solar_thermal)[["index_right"]]
weather_year = get_sector_parameters("global", scenario)["weather_year"]
feedin = db.select_dataframe(
f"""
SELECT w_id, feedin
FROM {sources.tables["feedin_timeseries"]}
WHERE carrier = 'solar_thermal'
AND weather_year = {weather_year}
""",
index_col="w_id",
)
timeseries = pd.DataFrame(
data={
"scn_name": scenario,
"temp_id": 1,
"p_max_pu": feedin.feedin[join.index].values,
"generator_id": generator.generator_id[
generator.carrier == "rural_solar_thermal"
].values,
}
).set_index("generator_id")
generator = generator.set_index("generator_id")
generator.to_sql(
targets.get_table_name("heat_generators"),
schema=targets.get_table_schema("heat_generators"),
if_exists="append",
con=db.engine(),
)
timeseries.to_sql(
targets.get_table_name("heat_generator_timeseries"),
schema=targets.get_table_schema("heat_generator_timeseries"),
if_exists="append",
con=db.engine(),
)
[docs]
def insert_central_direct_heat(scenario):
"""Insert renewable heating technologies (solar and geo thermal)
Parameters
----------
scenario : str
Name of the scenario
Returns
-------
None.
"""
sources = HeatEtrago.sources
targets = HeatEtrago.targets
db.execute_sql(f"""
DELETE FROM {targets.tables["heat_generators"]}
WHERE carrier IN ('solar_thermal_collector', 'geo_thermal')
AND scn_name = '{scenario}'
AND bus IN
(SELECT bus_id
FROM {targets.tables["heat_buses"]}
WHERE scn_name = '{scenario}'
AND country = 'DE')
""")
db.execute_sql(f"""
DELETE FROM {targets.tables["heat_generator_timeseries"]}
WHERE scn_name = '{scenario}'
AND generator_id NOT IN (
SELECT generator_id
FROM {targets.tables["heat_generators"]}
WHERE scn_name = '{scenario}')
""")
central_thermal = db.select_geodataframe(
f"""
SELECT district_heating_id, capacity, geometry, carrier
FROM {sources.tables["district_heating_supply"]}
WHERE scenario = '{scenario}'
AND carrier IN (
'solar_thermal_collector', 'geo_thermal')
""",
geom_col="geometry",
index_col="district_heating_id",
)
map_dh_id_bus_id = db.select_dataframe(
f"""
SELECT bus_id, area_id, id
FROM {targets.tables["heat_buses"]}
JOIN {sources.tables["district_heating_areas"]}
ON ST_Intersects(
ST_Transform(
ST_Buffer(ST_Centroid(geom_polygon),
0.0000001), 4326), geom)
WHERE carrier = 'central_heat'
AND scenario = '{scenario}'
AND scn_name = '{scenario}'
""",
index_col="id",
)
generator = pd.DataFrame(
data={
"scn_name": scenario,
"carrier": central_thermal.carrier,
"bus": map_dh_id_bus_id.bus_id[central_thermal.index],
"p_nom": central_thermal.capacity,
"generator_id": db.next_etrago_id(
"generator", len(central_thermal.index)
),
}
)
solar_thermal = central_thermal[
central_thermal.carrier == "solar_thermal_collector"
]
weather_cells = db.select_geodataframe(
f"""
SELECT w_id, geom
FROM {sources.tables["weather_cells"]}
""",
index_col="w_id",
)
# Map solar thermal collectors to weather cells
join = gpd.sjoin(weather_cells, solar_thermal)[["district_heating_id"]]
weather_year = get_sector_parameters("global", scenario)["weather_year"]
feedin = db.select_dataframe(
f"""
SELECT w_id, feedin
FROM {sources.tables["feedin_timeseries"]}
WHERE carrier = 'solar_thermal'
AND weather_year = {weather_year}
""",
index_col="w_id",
)
timeseries = pd.DataFrame(
data={
"scn_name": scenario,
"temp_id": 1,
"p_max_pu": feedin.feedin[join.index].values,
"generator_id": generator.generator_id[
generator.carrier == "solar_thermal_collector"
].values,
}
).set_index("generator_id")
generator = generator.set_index("generator_id")
generator.to_sql(
targets.get_table_name("heat_generators"),
schema=targets.get_table_schema("heat_generators"),
if_exists="append",
con=db.engine(),
)
timeseries.to_sql(
targets.get_table_name("heat_generator_timeseries"),
schema=targets.get_table_schema("heat_generator_timeseries"),
if_exists="append",
con=db.engine(),
)
[docs]
def insert_central_gas_boilers(scenario):
"""Inserts gas boilers for district heating to eTraGo-table
Parameters
----------
scenario : str
Name of the scenario.
Returns
-------
None.
"""
sources = HeatEtrago.sources
targets = HeatEtrago.targets
db.execute_sql(f"""
DELETE FROM {targets.tables["heat_links"]}
WHERE carrier LIKE '%central_gas_boiler%'
AND scn_name = '{scenario}'
AND link_id IN(
SELECT link_id
FROM {targets.tables["heat_links"]}
WHERE bus0 IN (
SELECT bus_id
FROM {targets.tables["heat_buses"]}
WHERE country = 'DE'
AND scn_name = '{scenario}'
)
AND bus1 IN (
SELECT bus_id
FROM {targets.tables["heat_buses"]}
WHERE country = 'DE'
AND scn_name = '{scenario}'
)
)
""")
central_boilers = db.select_dataframe(f"""
SELECT c.bus_id as bus0, b.bus_id as bus1,
capacity, a.carrier, scenario as scn_name
FROM {sources.tables["district_heating_supply"]} a
JOIN {targets.tables["heat_buses"]} b
ON ST_Transform(ST_Centroid(geometry), 4326) = geom
JOIN {sources.tables["ch4_voronoi"]} c
ON ST_Intersects(ST_Transform(a.geometry, 4326), c.geom)
WHERE scenario = '{scenario}'
AND b.scn_name = '{scenario}'
AND a.carrier = 'gas_boiler'
AND b.carrier='central_heat'
AND c.carrier='CH4'
AND c.scn_name = '{scenario}'
""")
# Add LineString topology
central_boilers = link_geom_from_buses(central_boilers, scenario)
# Add efficiency and marginal costs of gas boilers
central_boilers["efficiency"] = get_sector_parameters("heat", scenario)[
"efficiency"
]["central_gas_boiler"]
central_boilers["marginal_cost"] = get_sector_parameters("heat", scenario)[
"marginal_cost"
]["central_gas_boiler"]
# Transform thermal capacity to CH4 installed capacity
central_boilers["p_nom"] = central_boilers.capacity.div(
central_boilers.efficiency
)
# Drop unused columns
central_boilers.drop(["capacity"], axis=1, inplace=True)
# Set index
central_boilers.index = db.next_etrago_id(
"link", len(central_boilers.index)
)
central_boilers.index.name = "link_id"
# Set carrier name
central_boilers.carrier = "central_gas_boiler"
central_boilers.reset_index().to_postgis(
targets.get_table_name("heat_links"),
schema=targets.get_table_schema("heat_links"),
con=db.engine(),
if_exists="append",
)
[docs]
def insert_rural_gas_boilers(scenario):
"""Inserts gas boilers for individual heating to eTraGo-table
Parameters
----------
scenario : str
Name of the scenario.
Returns
-------
None.
"""
sources = HeatEtrago.sources
targets = HeatEtrago.targets
db.execute_sql(f"""
DELETE FROM {targets.tables["heat_links"]}
WHERE carrier = 'rural_gas_boiler'
AND scn_name = '{scenario}'
AND bus0 IN
(SELECT bus_id
FROM {targets.tables["heat_buses"]}
WHERE scn_name = '{scenario}'
AND country = 'DE')
AND bus1 IN
(SELECT bus_id
FROM {targets.tables["heat_buses"]}
WHERE scn_name = '{scenario}'
AND country = 'DE')
""")
rural_boilers = db.select_dataframe(f"""
SELECT c.bus_id as bus0, b.bus_id as bus1,
capacity, a.carrier, scenario as scn_name
FROM {sources.tables["individual_heating_supply"]} a
JOIN {targets.tables["heat_buses"]} b
ON ST_Transform(ST_Centroid(a.geometry), 4326) = b.geom
JOIN {sources.tables["ch4_voronoi"]} c
ON ST_Intersects(ST_Transform(a.geometry, 4326), c.geom)
WHERE scenario = '{scenario}'
AND b.scn_name = '{scenario}'
AND a.carrier = 'gas_boiler'
AND b.carrier='rural_heat'
AND c.carrier='CH4'
AND c.scn_name = '{scenario}'
""")
if rural_boilers.empty:
print(f"No rural gas boilers in scenario {scenario}.")
return
# Add LineString topology
rural_boilers = link_geom_from_buses(rural_boilers, scenario)
# Add efficiency of gas boilers
rural_boilers["efficiency"] = get_sector_parameters("heat", scenario)[
"efficiency"
]["rural_gas_boiler"]
# Transform thermal capacity to CH4 installed capacity
rural_boilers["p_nom"] = rural_boilers.capacity.div(
rural_boilers.efficiency
)
# Drop unused columns
rural_boilers.drop(["capacity"], axis=1, inplace=True)
# Set index
rural_boilers.index = db.next_etrago_id("link", len(rural_boilers.index))
rural_boilers.index.name = "link_id"
# Set carrier name
rural_boilers.carrier = "rural_gas_boiler"
rural_boilers.reset_index().to_postgis(
targets.get_table_name("heat_links"),
schema=targets.get_table_schema("heat_links"),
con=db.engine(),
if_exists="append",
)
[docs]
def buses():
"""Insert individual and district heat buses into eTraGo-tables
Parameters
----------
Returns
-------
None.
"""
for scenario in config.settings()["egon-data"]["--scenarios"]:
insert_buses("central_heat", scenario)
insert_buses("rural_heat", scenario)
[docs]
def supply():
"""Insert individual and district heat supply into eTraGo-tables
Parameters
----------
Returns
-------
None.
"""
for scenario in config.settings()["egon-data"]["--scenarios"]:
# There is no direct heat in status quo scenario
if "status" not in scenario:
insert_central_direct_heat(scenario)
insert_central_power_to_heat(scenario)
insert_individual_power_to_heat(scenario)
insert_central_gas_boilers(scenario)
insert_rural_gas_boilers(scenario)
insert_rural_direct_heat(scenario)
[docs]
class HeatEtrago(Dataset):
"""
Collect data related to the heat sector for the eTraGo tool
This dataset collects data from the heat sector and puts it into a format that
is needed for the transmission grid optimisation within the tool eTraGo.
It includes the creation of individual and central heat nodes, aggregates the
heat supply technologies (apart from CHP) per medium voltage grid district and
adds extendable heat stores to each bus. This data is then writing into the
corresponding tables that are read by eTraGo.
*Dependencies*
* :py:class:`HeatSupply <egon.data.datasets.heat_supply.HeatSupply>`
* :py:class:`MvGridDistricts <egon.data.datasets.mv_grid_districts.mv_grid_districts_setup>`
* :py:class:`EtragoSetup <egon.data.datasets.etrago_setup.EtragoSetup>`
* :py:class:`RenewableFeedin <egon.data.datasets.renewable_feedin.RenewableFeedin>`
* :py:class:`HeatTimeSeries <egon.data.datasets.heat_demand_timeseries.HeatTimeSeries>`
*Resulting tables*
* :py:class:`grid.egon_etrago_bus <egon.data.datasets.etrago_setup.EgonPfHvBus>` is extended
* :py:class:`grid.egon_etrago_link <egon.data.datasets.etrago_setup.EgonPfHvLink>` is extended
* :py:class:`grid.egon_etrago_link_timeseries <egon.data.datasets.etrago_setup.EgonPfHvLinkTimeseries>` is extended
* :py:class:`grid.egon_etrago_store <egon.data.datasets.etrago_setup.EgonPfHvStore>` is extended
* :py:class:`grid.egon_etrago_generator <egon.data.datasets.etrago_setup.EgonPfHvGenerator>` is extended
"""
#:
name: str = "HeatEtrago"
#:
version: str = "0.0.15"
sources = DatasetSources(
tables={
"scenario_capacities": "supply.egon_scenario_capacities",
"district_heating_areas": "demand.egon_district_heating_areas",
"map_district_heating_areas": "demand.egon_map_zensus_district_heating_areas",
"mv_grids": "grid.egon_mv_grid_district",
"district_heating_supply": "supply.egon_district_heating",
"individual_heating_supply": "supply.egon_individual_heating",
"weather_cells": "supply.egon_era5_weather_cells",
"feedin_timeseries": "supply.egon_era5_renewable_feedin",
"egon_mv_grid_district": "grid.egon_mv_grid_district",
"heat_demand": "demand.egon_peta_heat",
"ch4_voronoi": "grid.egon_gas_voronoi",
"map_zensus_grid_districts": "boundaries.egon_map_zensus_grid_districts",
"zensus_population": "society.destatis_zensus_population_per_ha",
},
)
targets = DatasetTargets(
tables={
"heat_buses": "grid.egon_etrago_bus",
"heat_generators": "grid.egon_etrago_generator",
"heat_generator_timeseries": "grid.egon_etrago_generator_timeseries",
"heat_links": "grid.egon_etrago_link",
"heat_link_timeseries": "grid.egon_etrago_link_timeseries",
"heat_stores": "grid.egon_etrago_store",
},
)
def __init__(self, dependencies):
super().__init__(
name=self.name,
version=self.version,
dependencies=dependencies,
tasks=(buses, supply, store),
)