data.datasets.heat_etrago.hts_etrago.HtsEtragoTable.__init__() - Code Metrics - Inspection of "Merge pull request #1273 from openego/fixes/#1256-..." - openego/eGon-data - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — dev ( 8582b4...82307e )

unknown

created 2025-06-12 13:09 UTC

HtsEtragoTable.init() A

↳ Parent: data.datasets.heat_etrago.hts_etrago

Complexity

Conditions

Size

Total Lines	6
Code Lines	6

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	6
nop	2
dl	0
loc	6
rs	10
c	0
b	0
f	0

"""
The central module creating heat demand time series for the eTraGo tool
"""
from egon.data import config, db
from egon.data.db import next_etrago_id
from egon.data.datasets import Dataset
from egon.data.datasets.scenario_parameters import get_sector_parameters

import pandas as pd
import numpy as np


def hts_to_etrago(scenario):
    sources = config.datasets()["etrago_heat"]["sources"]
    targets = config.datasets()["etrago_heat"]["targets"]
    carriers = ["central_heat", "rural_heat", "rural_gas_boiler"]

    if "status" in scenario:
        carriers = ["central_heat", "rural_heat"]

    for carrier in carriers:
        if carrier == "central_heat":
            # Map heat buses to district heating id and area_id
            # interlinking bus_id and area_id
            bus_area = db.select_dataframe(
                f"""
                 SELECT bus_id, area_id, id FROM
                 {targets['heat_buses']['schema']}.
                 {targets['heat_buses']['table']}
                 JOIN {sources['district_heating_areas']['schema']}.
                     {sources['district_heating_areas']['table']}
                 ON ST_Transform(ST_Centroid(geom_polygon), 4326) = geom
                 WHERE carrier = '{carrier}'
                 AND scenario='{scenario}'
                 AND scn_name = '{scenario}'
                 """,
                index_col="id",
            )

            # district heating time series time series
            disct_time_series = db.select_dataframe(
                f"""
                                SELECT * FROM 
                                demand.egon_timeseries_district_heating
                                WHERE scenario ='{scenario}'                                
                                """
            )
            # bus_id connected to corresponding time series
            bus_ts = pd.merge(
                bus_area, disct_time_series, on="area_id", how="inner"
            )

        elif carrier == "rural_heat":
            # interlinking heat_bus_id and mv_grid bus_id
            bus_sub = db.select_dataframe(
                f"""
                 SELECT {targets['heat_buses']['schema']}.
                 {targets['heat_buses']['table']}.bus_id as heat_bus_id, 
                 {sources['egon_mv_grid_district']['schema']}.
                             {sources['egon_mv_grid_district']['table']}.bus_id as 
                             bus_id FROM
                 {targets['heat_buses']['schema']}.
                 {targets['heat_buses']['table']}
                 JOIN {sources['egon_mv_grid_district']['schema']}.
                             {sources['egon_mv_grid_district']['table']}
                 ON ST_Transform(ST_Centroid({sources['egon_mv_grid_district']['schema']}.
                             {sources['egon_mv_grid_district']['table']}.geom), 
                                 4326) = {targets['heat_buses']['schema']}.
                                         {targets['heat_buses']['table']}.geom
                 WHERE carrier = '{carrier}'
                 AND scn_name = '{scenario}'
                 """
            )
            ##**scenario name still needs to be adjusted in bus_sub**

            # individual heating time series
            ind_time_series = db.select_dataframe(
                f"""
                SELECT scenario, bus_id, dist_aggregated_mw FROM 
                demand.egon_etrago_timeseries_individual_heating
                WHERE scenario ='{scenario}'
                AND carrier = 'heat_pump'
                """
            )

            # bus_id connected to corresponding time series
            bus_ts = pd.merge(
                bus_sub, ind_time_series, on="bus_id", how="inner"
            )

            # Connect  heat loads to heat buses
            bus_ts.loc[:, "bus_id"] = bus_ts.loc[:, "heat_bus_id"]

        else:
            efficiency_gas_boiler = get_sector_parameters("heat", scenario)[
                "efficiency"
            ]["rural_gas_boiler"]

            # Select rural heat demand coverd by individual gas boilers
            ind_time_series = db.select_dataframe(
                f"""
                SELECT * FROM 
                demand.egon_etrago_timeseries_individual_heating
                WHERE scenario ='{scenario}'
                AND carrier = 'CH4'
                """
            )

            # Select geoetry of medium voltage grid districts
            mvgd_geom = db.select_geodataframe(
                f"""
                SELECT bus_id, ST_CENTROID(geom) as geom FROM 
                {sources['egon_mv_grid_district']['schema']}.
                {sources['egon_mv_grid_district']['table']}
                """
            )

            # Select geometry of gas (CH4) voronoi
            gas_voronoi = db.select_geodataframe(
                f"""
                SELECT bus_id, geom FROM 
                grid.egon_gas_voronoi
                WHERE scn_name = '{scenario}'
                AND carrier = 'CH4'
                """
            )

            # Map centroid of mvgd to gas voronoi
            join = mvgd_geom.sjoin(gas_voronoi, lsuffix="AC", rsuffix="gas")[
                ["bus_id_AC", "bus_id_gas"]
            ].set_index("bus_id_AC")

            # Assign gas bus to each rural heat demand coverd by gas boiler
            ind_time_series["gas_bus"] = join.loc[
                ind_time_series.bus_id
            ].values

            # Initialize dataframe to store final heat demand per gas node
            gas_ts = pd.DataFrame(
                index=ind_time_series["gas_bus"].unique(), columns=range(8760)
            )

            # Group heat demand per hour in the year
            for i in range(8760):
                gas_ts[i] = (
                    ind_time_series.set_index("gas_bus")
                    .dist_aggregated_mw.str[i]
                    .groupby("gas_bus")
                    .sum()
                    .div(efficiency_gas_boiler)
                )

            # Prepare resulting DataFrame
            bus_ts = pd.DataFrame(columns=["dist_aggregated_mw", "bus_id"])

            # Insert values to dataframe
            bus_ts.dist_aggregated_mw = gas_ts.values.tolist()
            bus_ts.bus_id = gas_ts.index

        # Delete existing data from database
        db.execute_sql(
            f"""
            DELETE FROM grid.egon_etrago_load
            WHERE scn_name = '{scenario}'
            AND carrier = '{carrier}'
            AND bus IN (
                SELECT bus_id FROM grid.egon_etrago_bus
                WHERE country = 'DE'
                AND scn_name = '{scenario}'
                )
            """
        )

        db.execute_sql(
            f"""
            DELETE FROM
            grid.egon_etrago_load_timeseries
            WHERE scn_name = '{scenario}'
            AND load_id NOT IN (
            SELECT load_id FROM
            grid.egon_etrago_load
            WHERE scn_name = '{scenario}')
            """
        )

        next_id = next_etrago_id("load")

        bus_ts["load_id"] = np.arange(len(bus_ts)) + next_id

        etrago_load = pd.DataFrame(index=range(len(bus_ts)))
        etrago_load["scn_name"] = scenario
        etrago_load["load_id"] = bus_ts.load_id
        etrago_load["bus"] = bus_ts.bus_id
        etrago_load["carrier"] = carrier
        etrago_load["sign"] = -1

        etrago_load.to_sql(
            "egon_etrago_load",
            schema="grid",
            con=db.engine(),
            if_exists="append",
            index=False,
        )

        etrago_load_timeseries = pd.DataFrame(index=range(len(bus_ts)))
        etrago_load_timeseries["scn_name"] = scenario
        etrago_load_timeseries["load_id"] = bus_ts.load_id
        etrago_load_timeseries["temp_id"] = 1
        etrago_load_timeseries["p_set"] = bus_ts.loc[:, "dist_aggregated_mw"]

        etrago_load_timeseries.to_sql(
            "egon_etrago_load_timeseries",
            schema="grid",
            con=db.engine(),
            if_exists="append",
            index=False,
        )


def demand():
    """Insert demand timeseries for heat into eTraGo tables

    Returns
    -------
    None.

    """
    for scenario in config.settings()["egon-data"]["--scenarios"]:
        hts_to_etrago(scenario)


class HtsEtragoTable(Dataset):
    """
    Collect heat demand time series for the eTraGo tool

    This dataset collects data for individual and district heating demands
    and writes that into the tables that can be read by the eTraGo tool.

    *Dependencies*
      * :py:class:`DistrictHeatingAreas <egon.data.datasets.district_heating_areas.DistrictHeatingAreas>`
      * :py:class:`HeatEtrago <egon.data.datasets.heat_etrago.HeatEtrago>`
      * :py:class:`MvGridDistricts <egon.data.datasets.mv_grid_districts.mv_grid_districts_setup>`
      * :py:class:`HeatPumps2035 <egon.data.datasets.heat_supply.individual_heating.HeatPumps2035>`
      * :py:class:`HeatTimeSeries <egon.data.datasets.heat_demand_timeseries.HeatTimeSeries>`

    *Resulting tables*
      * :py:class:`grid.egon_etrago_load <egon.data.datasets.etrago_setup.EgonPfHvLoad>` is extended
      * :py:class:`grid.egon_etrago_load_timeseries <egon.data.datasets.etrago_setup.EgonPfHvLoadTimeseries>` is extended

    """

    #:
    name: str = "HtsEtragoTable"
    #:
    version: str = "0.0.6"

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


1			"""
2			The central module creating heat demand time series for the eTraGo tool
3			"""
4			from egon.data import config, db
5			from egon.data.db import next_etrago_id
6			from egon.data.datasets import Dataset
7			from egon.data.datasets.scenario_parameters import get_sector_parameters
8
9			import pandas as pd
10			import numpy as np
11
12
13			def hts_to_etrago(scenario):
14			sources = config.datasets()["etrago_heat"]["sources"]
15			targets = config.datasets()["etrago_heat"]["targets"]
16			carriers = ["central_heat", "rural_heat", "rural_gas_boiler"]
17
18			if "status" in scenario:
19			carriers = ["central_heat", "rural_heat"]
20
21			for carrier in carriers:
22			if carrier == "central_heat":
23			# Map heat buses to district heating id and area_id
24			# interlinking bus_id and area_id
25			bus_area = db.select_dataframe(
26			f"""
27			SELECT bus_id, area_id, id FROM
28			{targets['heat_buses']['schema']}.
29			{targets['heat_buses']['table']}
30			JOIN {sources['district_heating_areas']['schema']}.
31			{sources['district_heating_areas']['table']}
32			ON ST_Transform(ST_Centroid(geom_polygon), 4326) = geom
33			WHERE carrier = '{carrier}'
34			AND scenario='{scenario}'
35			AND scn_name = '{scenario}'
36			""",
37			index_col="id",
38			)
39
40			# district heating time series time series
41			disct_time_series = db.select_dataframe(
42			f"""
43			SELECT * FROM
44			demand.egon_timeseries_district_heating
45			WHERE scenario ='{scenario}'
46			"""
47			)
48			# bus_id connected to corresponding time series
49			bus_ts = pd.merge(
50			bus_area, disct_time_series, on="area_id", how="inner"
51			)
52
53			elif carrier == "rural_heat":
54			# interlinking heat_bus_id and mv_grid bus_id
55			bus_sub = db.select_dataframe(
56			f"""
57			SELECT {targets['heat_buses']['schema']}.
58			{targets['heat_buses']['table']}.bus_id as heat_bus_id,
59			{sources['egon_mv_grid_district']['schema']}.
60			{sources['egon_mv_grid_district']['table']}.bus_id as
61			bus_id FROM
62			{targets['heat_buses']['schema']}.
63			{targets['heat_buses']['table']}
64			JOIN {sources['egon_mv_grid_district']['schema']}.
65			{sources['egon_mv_grid_district']['table']}
66			ON ST_Transform(ST_Centroid({sources['egon_mv_grid_district']['schema']}.
67			{sources['egon_mv_grid_district']['table']}.geom),
68			4326) = {targets['heat_buses']['schema']}.
69			{targets['heat_buses']['table']}.geom
70			WHERE carrier = '{carrier}'
71			AND scn_name = '{scenario}'
72			"""
73			)
74			##scenario name still needs to be adjusted in bus_sub
75
76			# individual heating time series
77			ind_time_series = db.select_dataframe(
78			f"""
79			SELECT scenario, bus_id, dist_aggregated_mw FROM
80			demand.egon_etrago_timeseries_individual_heating
81			WHERE scenario ='{scenario}'
82			AND carrier = 'heat_pump'
83			"""
84			)
85
86			# bus_id connected to corresponding time series
87			bus_ts = pd.merge(
88			bus_sub, ind_time_series, on="bus_id", how="inner"
89			)
90
91			# Connect heat loads to heat buses
92			bus_ts.loc[:, "bus_id"] = bus_ts.loc[:, "heat_bus_id"]
93
94			else:
95			efficiency_gas_boiler = get_sector_parameters("heat", scenario)[
96			"efficiency"
97			]["rural_gas_boiler"]
98
99			# Select rural heat demand coverd by individual gas boilers
100			ind_time_series = db.select_dataframe(
101			f"""
102			SELECT * FROM
103			demand.egon_etrago_timeseries_individual_heating
104			WHERE scenario ='{scenario}'
105			AND carrier = 'CH4'
106			"""
107			)
108
109			# Select geoetry of medium voltage grid districts
110			mvgd_geom = db.select_geodataframe(
111			f"""
112			SELECT bus_id, ST_CENTROID(geom) as geom FROM
113			{sources['egon_mv_grid_district']['schema']}.
114			{sources['egon_mv_grid_district']['table']}
115			"""
116			)
117
118			# Select geometry of gas (CH4) voronoi
119			gas_voronoi = db.select_geodataframe(
120			f"""
121			SELECT bus_id, geom FROM
122			grid.egon_gas_voronoi
123			WHERE scn_name = '{scenario}'
124			AND carrier = 'CH4'
125			"""
126			)
127
128			# Map centroid of mvgd to gas voronoi
129			join = mvgd_geom.sjoin(gas_voronoi, lsuffix="AC", rsuffix="gas")[
130			["bus_id_AC", "bus_id_gas"]
131			].set_index("bus_id_AC")
132
133			# Assign gas bus to each rural heat demand coverd by gas boiler
134			ind_time_series["gas_bus"] = join.loc[
135			ind_time_series.bus_id
136			].values
137
138			# Initialize dataframe to store final heat demand per gas node
139			gas_ts = pd.DataFrame(
140			index=ind_time_series["gas_bus"].unique(), columns=range(8760)
141			)
142
143			# Group heat demand per hour in the year
144			for i in range(8760):
145			gas_ts[i] = (
146			ind_time_series.set_index("gas_bus")
147			.dist_aggregated_mw.str[i]
148			.groupby("gas_bus")
149			.sum()
150			.div(efficiency_gas_boiler)
151			)
152
153			# Prepare resulting DataFrame
154			bus_ts = pd.DataFrame(columns=["dist_aggregated_mw", "bus_id"])
155
156			# Insert values to dataframe
157			bus_ts.dist_aggregated_mw = gas_ts.values.tolist()
158			bus_ts.bus_id = gas_ts.index
159
160			# Delete existing data from database
161			db.execute_sql(
162			f"""
163			DELETE FROM grid.egon_etrago_load
164			WHERE scn_name = '{scenario}'
165			AND carrier = '{carrier}'
166			AND bus IN (
167			SELECT bus_id FROM grid.egon_etrago_bus
168			WHERE country = 'DE'
169			AND scn_name = '{scenario}'
170			)
171			"""
172			)
173
174			db.execute_sql(
175			f"""
176			DELETE FROM
177			grid.egon_etrago_load_timeseries
178			WHERE scn_name = '{scenario}'
179			AND load_id NOT IN (
180			SELECT load_id FROM
181			grid.egon_etrago_load
182			WHERE scn_name = '{scenario}')
183			"""
184			)
185
186			next_id = next_etrago_id("load")
187
188			bus_ts["load_id"] = np.arange(len(bus_ts)) + next_id
189
190			etrago_load = pd.DataFrame(index=range(len(bus_ts)))
191			etrago_load["scn_name"] = scenario
192			etrago_load["load_id"] = bus_ts.load_id
193			etrago_load["bus"] = bus_ts.bus_id
194			etrago_load["carrier"] = carrier
195			etrago_load["sign"] = -1
196
197			etrago_load.to_sql(
198			"egon_etrago_load",
199			schema="grid",
200			con=db.engine(),
201			if_exists="append",
202			index=False,
203			)
204
205			etrago_load_timeseries = pd.DataFrame(index=range(len(bus_ts)))
206			etrago_load_timeseries["scn_name"] = scenario
207			etrago_load_timeseries["load_id"] = bus_ts.load_id
208			etrago_load_timeseries["temp_id"] = 1
209			etrago_load_timeseries["p_set"] = bus_ts.loc[:, "dist_aggregated_mw"]
210
211			etrago_load_timeseries.to_sql(
212			"egon_etrago_load_timeseries",
213			schema="grid",
214			con=db.engine(),
215			if_exists="append",
216			index=False,
217			)
218
219
220			def demand():
221			"""Insert demand timeseries for heat into eTraGo tables
222
223			Returns
224			-------
225			None.
226
227			"""
228			for scenario in config.settings()["egon-data"]["--scenarios"]:
229			hts_to_etrago(scenario)
230
231
232			class HtsEtragoTable(Dataset):
233			"""
234			Collect heat demand time series for the eTraGo tool
235
236			This dataset collects data for individual and district heating demands
237			and writes that into the tables that can be read by the eTraGo tool.
238
239			Dependencies
240			* :py:class:`DistrictHeatingAreas <egon.data.datasets.district_heating_areas.DistrictHeatingAreas>`
241			* :py:class:`HeatEtrago <egon.data.datasets.heat_etrago.HeatEtrago>`
242			* :py:class:`MvGridDistricts <egon.data.datasets.mv_grid_districts.mv_grid_districts_setup>`
243			* :py:class:`HeatPumps2035 <egon.data.datasets.heat_supply.individual_heating.HeatPumps2035>`
244			* :py:class:`HeatTimeSeries <egon.data.datasets.heat_demand_timeseries.HeatTimeSeries>`
245
246			Resulting tables
247			* :py:class:`grid.egon_etrago_load <egon.data.datasets.etrago_setup.EgonPfHvLoad>` is extended
248			* :py:class:`grid.egon_etrago_load_timeseries <egon.data.datasets.etrago_setup.EgonPfHvLoadTimeseries>` is extended
249
250			"""
251
252			#:
253			name: str = "HtsEtragoTable"
254			#:
255			version: str = "0.0.6"
256
257			def __init__(self, dependencies):
258			super().__init__(
259			name=self.name,
260			version=self.version,
261			dependencies=dependencies,
262			tasks=(demand,),
263			)
264

openego / eGon-data

Push — dev ( 8582b4...82307e )

HtsEtragoTable.__init__() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

HtsEtragoTable.init() A