data.datasets.electricity_demand.distribute_cts_demands() - 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

distribute_cts_demands() A

↳ Parent: data.datasets.electricity_demand

Complexity

Conditions

Size

Total Lines	82
Code Lines	28

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	28
dl	0
loc	82
rs	9.208
c	0
b	0
f	0
cc	3
nop	0

How to fix Long Method

"""The central module containing all code dealing with processing
 data from demandRegio

"""
from sqlalchemy import Column, Float, ForeignKey, Integer, String
from sqlalchemy.ext.declarative import declarative_base
import pandas as pd

from egon.data import db
from egon.data.datasets import Dataset
from egon.data.datasets.electricity_demand.temporal import insert_cts_load
from egon.data.datasets.electricity_demand_timeseries.hh_buildings import (
    HouseholdElectricityProfilesOfBuildings,
    get_iee_hh_demand_profiles_raw,
)
from egon.data.datasets.electricity_demand_timeseries.hh_profiles import (
    HouseholdElectricityProfilesInCensusCells,
)
from egon.data.datasets.zensus_vg250 import DestatisZensusPopulationPerHa
import egon.data.config

# will be later imported from another file ###
Base = declarative_base()
engine = db.engine()


class HouseholdElectricityDemand(Dataset):
    """
    Create table and store data on household electricity demands per census cell

    Create a table to store the annual electricity demand for households on census cell level.
    In a next step the annual electricity demand per cell is determined by
    executing function :py:func:`get_annual_household_el_demand_cells`

    *Dependencies*
      * :py:class:`DemandRegio <egon.data.datasets.demandregio>`
      * :py:class:`DataBundle <egon.data.datasets.data_bundle.DataBundle>`

    *Resulting tables*
      * :py:class:`demand.egon_demandregio_zensus_electricity <egon.data.datasets.electricity_demand.EgonDemandRegioZensusElectricity>` is created and filled

    """

    #:
    name: str = "HouseholdElectricityDemand"
    #:
    version: str = "0.0.5"

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


class CtsElectricityDemand(Dataset):
    """
    Create table and store data on cts electricity demands per census cell

    Creates a table to store data on electricity demands from the cts sector on
    census cell level. For a spatial distribution of electricity demands data
    from DemandRegio, which provides the data on NUT3-level, is used and
    distributed to census cells according to heat demand data from Peta.
    Annual demands are then aggregated per MV grid district and a corresponding
    time series is created taking the shares of different cts branches and their
    specific standard load profiles into account.


    *Dependencies*
      * :py:class:`MapZensusGridDistricts <egon.data.datasets.zensus_mv_grid_districts.MapZensusGridDistricts>`
      * :py:class:`DemandRegio <egon.data.datasets.demandregio.DemandRegio>`
      * :py:class:`HeatDemandImport <egon.data.datasets.heat_demand.HeatDemandImport>`
      * :py:class:`HouseholdElectricityDemand <egon.data.datasets.electricity_demand.HouseholdElectricityDemand>`
      * :py:class:`EtragoSetup <egon.data.datasets.etrago_setup.EtragoSetup>`
      * :py:class:`ZensusMvGridDistricts <egon.data.datasets.zensus_mv_grid_districts.ZensusMvGridDistricts>`
      * :py:class:`ZensusVg250 <egon.data.datasets.zensus_vg250.ZensusVg250>`


    *Resulting tables*
      * :py:class:`demand.egon_etrago_electricity_cts <egon.data.datasets.electricity_demand.temporal.EgonEtragoElectricityCts>` is created and filled


    """

    #:
    name: str = "CtsElectricityDemand"
    #:
    version: str = "0.0.2"

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


class EgonDemandRegioZensusElectricity(Base):
    __tablename__ = "egon_demandregio_zensus_electricity"
    __table_args__ = {"schema": "demand", "extend_existing": True}
    zensus_population_id = Column(
        Integer, ForeignKey(DestatisZensusPopulationPerHa.id), primary_key=True
    )
    scenario = Column(String(50), primary_key=True)
    sector = Column(String, primary_key=True)
    demand = Column(Float)


def create_tables():
    """Create tables for demandregio data
    Returns
    -------
    None.
    """
    db.execute_sql("CREATE SCHEMA IF NOT EXISTS demand;")
    db.execute_sql("CREATE SCHEMA IF NOT EXISTS society;")
    engine = db.engine()
    EgonDemandRegioZensusElectricity.__table__.drop(
        bind=engine, checkfirst=True
    )
    EgonDemandRegioZensusElectricity.__table__.create(
        bind=engine, checkfirst=True
    )


def get_annual_household_el_demand_cells():
    """
    Annual electricity demand per cell is determined

    Timeseries for every cell are accumulated, the maximum value
    determined and with the respective nuts3 factor scaled for 2035 and 2050
    scenario.

    Note
    ----------
    In test-mode 'SH' the iteration takes place by 'cell_id' to avoid
    intensive RAM usage. For whole Germany 'nuts3' are taken and
    RAM > 32GB is necessary.
    """

    with db.session_scope() as session:
        cells_query = (
            session.query(
                HouseholdElectricityProfilesOfBuildings,
                HouseholdElectricityProfilesInCensusCells.nuts3,
                HouseholdElectricityProfilesInCensusCells.factor_2019,
                HouseholdElectricityProfilesInCensusCells.factor_2023,
                HouseholdElectricityProfilesInCensusCells.factor_2035,
                HouseholdElectricityProfilesInCensusCells.factor_2050,
            )
            .filter(
                HouseholdElectricityProfilesOfBuildings.cell_id
                == HouseholdElectricityProfilesInCensusCells.cell_id
            )
            .order_by(HouseholdElectricityProfilesOfBuildings.id)
        )

    df_buildings_and_profiles = pd.read_sql(
        cells_query.statement, cells_query.session.bind, index_col="id"
    )

    # Read demand profiles from egon-data-bundle
    df_profiles = get_iee_hh_demand_profiles_raw()

    def ve(s):
        raise (ValueError(s))

    dataset = egon.data.config.settings()["egon-data"]["--dataset-boundary"]
    scenarios = egon.data.config.settings()["egon-data"]["--scenarios"]

    iterate_over = (
        "nuts3"
        if dataset == "Everything"
        else "cell_id"
        if dataset == "Schleswig-Holstein"
        else ve(f"'{dataset}' is not a valid dataset boundary.")
    )

    df_annual_demand = pd.DataFrame(
        columns=scenarios + ["zensus_population_id"]
    )

    for _, df in df_buildings_and_profiles.groupby(by=iterate_over):
        df_annual_demand_iter = pd.DataFrame(
            columns=scenarios + ["zensus_population_id"]
        )

        if "eGon2035" in scenarios:
            df_annual_demand_iter["eGon2035"] = (
                df_profiles.loc[:, df["profile_id"]].sum(axis=0)
                * df["factor_2035"].values
            )
        if "eGon100RE" in scenarios:
            df_annual_demand_iter["eGon100RE"] = (
                df_profiles.loc[:, df["profile_id"]].sum(axis=0)
                * df["factor_2050"].values
            )
        if "status2019" in scenarios:
            df_annual_demand_iter["status2019"] = (
                df_profiles.loc[:, df["profile_id"]].sum(axis=0)
                * df["factor_2019"].values
            )

        if "status2023" in scenarios:
            df_annual_demand_iter["status2023"] = (
                df_profiles.loc[:, df["profile_id"]].sum(axis=0)
                * df["factor_2023"].values
            )
        df_annual_demand_iter["zensus_population_id"] = df["cell_id"].values
        df_annual_demand = pd.concat([df_annual_demand, df_annual_demand_iter])

    df_annual_demand = (
        df_annual_demand.groupby("zensus_population_id").sum().reset_index()
    )
    df_annual_demand["sector"] = "residential"
    df_annual_demand = df_annual_demand.melt(
        id_vars=["zensus_population_id", "sector"],
        var_name="scenario",
        value_name="demand",
    )
    # convert from Wh to MWh
    df_annual_demand["demand"] = df_annual_demand["demand"] / 1e6

    # delete all cells for residentials
    with db.session_scope() as session:
        session.query(EgonDemandRegioZensusElectricity).filter(
            EgonDemandRegioZensusElectricity.sector == "residential"
        ).delete()

    # Insert data to target table
    df_annual_demand.to_sql(
        name=EgonDemandRegioZensusElectricity.__table__.name,
        schema=EgonDemandRegioZensusElectricity.__table__.schema,
        con=db.engine(),
        index=False,
        if_exists="append",
    )


def distribute_cts_demands():
    """Distribute electrical demands for cts to zensus cells.

    The demands on nuts3-level from demandregio are linear distributed
    to the heat demand of cts in each zensus cell.

    Returns
    -------
    None.

    """

    sources = egon.data.config.datasets()["electrical_demands_cts"]["sources"]

    target = egon.data.config.datasets()["electrical_demands_cts"]["targets"][
        "cts_demands_zensus"
    ]

    db.execute_sql(
        f"""DELETE FROM {target['schema']}.{target['table']}
                   WHERE sector = 'service'"""
    )

    # Select match between zensus cells and nuts3 regions of vg250
    map_nuts3 = db.select_dataframe(
        f"""SELECT zensus_population_id, vg250_nuts3 as nuts3 FROM
        {sources['map_zensus_vg250']['schema']}.
        {sources['map_zensus_vg250']['table']}""",
        index_col="zensus_population_id",
    )

    # Insert data per scenario
    for scn in egon.data.config.settings()["egon-data"]["--scenarios"]:
        # Select heat_demand per zensus cell
        peta = db.select_dataframe(
            f"""SELECT zensus_population_id, demand as heat_demand,
            sector, scenario FROM
            {sources['heat_demand_cts']['schema']}.
            {sources['heat_demand_cts']['table']}
            WHERE scenario = '{scn}'
            AND sector = 'service'""",
            index_col="zensus_population_id",
        )

        # Add nuts3 key to zensus cells
        peta["nuts3"] = map_nuts3.nuts3

        # Calculate share of nuts3 heat demand per zensus cell
        for nuts3, df in peta.groupby("nuts3"):
            peta.loc[df.index, "share"] = (
                df["heat_demand"] / df["heat_demand"].sum()
            )

        # Select forecasted electrical demands from demandregio table
        demand_nuts3 = db.select_dataframe(
            f"""SELECT nuts3, SUM(demand) as demand FROM
            {sources['demandregio']['schema']}.
            {sources['demandregio']['table']}
            WHERE scenario = '{scn}'
            AND wz IN (
                SELECT wz FROM
                {sources['demandregio_wz']['schema']}.
                {sources['demandregio_wz']['table']}
                WHERE sector = 'CTS')
            GROUP BY nuts3""",
            index_col="nuts3",
        )

        # Scale demands on nuts3 level linear to heat demand share
        peta["demand"] = peta["share"].mul(
            demand_nuts3.demand[peta["nuts3"]].values
        )

        # Rename index
        peta.index = peta.index.rename("zensus_population_id")

        # Insert data to target table
        peta[["scenario", "demand", "sector"]].to_sql(
            target["table"],
            schema=target["schema"],
            con=db.engine(),
            if_exists="append",
        )


1			"""The central module containing all code dealing with processing
2			data from demandRegio
3
4			"""
5			from sqlalchemy import Column, Float, ForeignKey, Integer, String
6			from sqlalchemy.ext.declarative import declarative_base
7			import pandas as pd
8
9			from egon.data import db
10			from egon.data.datasets import Dataset
11			from egon.data.datasets.electricity_demand.temporal import insert_cts_load
12			from egon.data.datasets.electricity_demand_timeseries.hh_buildings import (
13			HouseholdElectricityProfilesOfBuildings,
14			get_iee_hh_demand_profiles_raw,
15			)
16			from egon.data.datasets.electricity_demand_timeseries.hh_profiles import (
17			HouseholdElectricityProfilesInCensusCells,
18			)
19			from egon.data.datasets.zensus_vg250 import DestatisZensusPopulationPerHa
20			import egon.data.config
21
22			# will be later imported from another file ###
23			Base = declarative_base()
24			engine = db.engine()
25
26
27			class HouseholdElectricityDemand(Dataset):
28			"""
29			Create table and store data on household electricity demands per census cell
30
31			Create a table to store the annual electricity demand for households on census cell level.
32			In a next step the annual electricity demand per cell is determined by
33			executing function :py:func:`get_annual_household_el_demand_cells`
34
35			Dependencies
36			* :py:class:`DemandRegio <egon.data.datasets.demandregio>`
37			* :py:class:`DataBundle <egon.data.datasets.data_bundle.DataBundle>`
38
39			Resulting tables
40			* :py:class:`demand.egon_demandregio_zensus_electricity <egon.data.datasets.electricity_demand.EgonDemandRegioZensusElectricity>` is created and filled
41
42			"""
43
44			#:
45			name: str = "HouseholdElectricityDemand"
46			#:
47			version: str = "0.0.5"
48
49			def __init__(self, dependencies):
50			super().__init__(
51			name=self.name,
52			version=self.version,
53			dependencies=dependencies,
54			tasks=(create_tables, get_annual_household_el_demand_cells),
55			)
56
57
58			class CtsElectricityDemand(Dataset):
59			"""
60			Create table and store data on cts electricity demands per census cell
61
62			Creates a table to store data on electricity demands from the cts sector on
63			census cell level. For a spatial distribution of electricity demands data
64			from DemandRegio, which provides the data on NUT3-level, is used and
65			distributed to census cells according to heat demand data from Peta.
66			Annual demands are then aggregated per MV grid district and a corresponding
67			time series is created taking the shares of different cts branches and their
68			specific standard load profiles into account.
69
70
71			Dependencies
72			* :py:class:`MapZensusGridDistricts <egon.data.datasets.zensus_mv_grid_districts.MapZensusGridDistricts>`
73			* :py:class:`DemandRegio <egon.data.datasets.demandregio.DemandRegio>`
74			* :py:class:`HeatDemandImport <egon.data.datasets.heat_demand.HeatDemandImport>`
75			* :py:class:`HouseholdElectricityDemand <egon.data.datasets.electricity_demand.HouseholdElectricityDemand>`
76			* :py:class:`EtragoSetup <egon.data.datasets.etrago_setup.EtragoSetup>`
77			* :py:class:`ZensusMvGridDistricts <egon.data.datasets.zensus_mv_grid_districts.ZensusMvGridDistricts>`
78			* :py:class:`ZensusVg250 <egon.data.datasets.zensus_vg250.ZensusVg250>`
79
80
81			Resulting tables
82			* :py:class:`demand.egon_etrago_electricity_cts <egon.data.datasets.electricity_demand.temporal.EgonEtragoElectricityCts>` is created and filled
83
84
85			"""
86
87			#:
88			name: str = "CtsElectricityDemand"
89			#:
90			version: str = "0.0.2"
91
92			def __init__(self, dependencies):
93			super().__init__(
94			name=self.name,
95			version=self.version,
96			dependencies=dependencies,
97			tasks=(distribute_cts_demands, insert_cts_load),
98			)
99
100
101			class EgonDemandRegioZensusElectricity(Base):
102			__tablename__ = "egon_demandregio_zensus_electricity"
103			__table_args__ = {"schema": "demand", "extend_existing": True}
104			zensus_population_id = Column(
105			Integer, ForeignKey(DestatisZensusPopulationPerHa.id), primary_key=True
106			)
107			scenario = Column(String(50), primary_key=True)
108			sector = Column(String, primary_key=True)
109			demand = Column(Float)
110
111
112			def create_tables():
113			"""Create tables for demandregio data
114			Returns
115			-------
116			None.
117			"""
118			db.execute_sql("CREATE SCHEMA IF NOT EXISTS demand;")
119			db.execute_sql("CREATE SCHEMA IF NOT EXISTS society;")
120			engine = db.engine()
121			EgonDemandRegioZensusElectricity.__table__.drop(
122			bind=engine, checkfirst=True
123			)
124			EgonDemandRegioZensusElectricity.__table__.create(
125			bind=engine, checkfirst=True
126			)
127
128
129			def get_annual_household_el_demand_cells():
130			"""
131			Annual electricity demand per cell is determined
132
133			Timeseries for every cell are accumulated, the maximum value
134			determined and with the respective nuts3 factor scaled for 2035 and 2050
135			scenario.
136
137			Note
138			----------
139			In test-mode 'SH' the iteration takes place by 'cell_id' to avoid
140			intensive RAM usage. For whole Germany 'nuts3' are taken and
141			RAM > 32GB is necessary.
142			"""
143
144			with db.session_scope() as session:
145			cells_query = (
146			session.query(
147			HouseholdElectricityProfilesOfBuildings,
148			HouseholdElectricityProfilesInCensusCells.nuts3,
149			HouseholdElectricityProfilesInCensusCells.factor_2019,
150			HouseholdElectricityProfilesInCensusCells.factor_2023,
151			HouseholdElectricityProfilesInCensusCells.factor_2035,
152			HouseholdElectricityProfilesInCensusCells.factor_2050,
153			)
154			.filter(
155			HouseholdElectricityProfilesOfBuildings.cell_id
156			== HouseholdElectricityProfilesInCensusCells.cell_id
157			)
158			.order_by(HouseholdElectricityProfilesOfBuildings.id)
159			)
160
161			df_buildings_and_profiles = pd.read_sql(
162			cells_query.statement, cells_query.session.bind, index_col="id"
163			)
164
165			# Read demand profiles from egon-data-bundle
166			df_profiles = get_iee_hh_demand_profiles_raw()
167
168			def ve(s):
169			raise (ValueError(s))
170
171			dataset = egon.data.config.settings()["egon-data"]["--dataset-boundary"]
172			scenarios = egon.data.config.settings()["egon-data"]["--scenarios"]
173
174			iterate_over = (
175			"nuts3"
176			if dataset == "Everything"
177			else "cell_id"
178			if dataset == "Schleswig-Holstein"
179			else ve(f"'{dataset}' is not a valid dataset boundary.")
180			)
181
182			df_annual_demand = pd.DataFrame(
183			columns=scenarios + ["zensus_population_id"]
184			)
185
186			for _, df in df_buildings_and_profiles.groupby(by=iterate_over):
187			df_annual_demand_iter = pd.DataFrame(
188			columns=scenarios + ["zensus_population_id"]
189			)
190
191			if "eGon2035" in scenarios:
192			df_annual_demand_iter["eGon2035"] = (
193			df_profiles.loc[:, df["profile_id"]].sum(axis=0)
194			* df["factor_2035"].values
195			)
196			if "eGon100RE" in scenarios:
197			df_annual_demand_iter["eGon100RE"] = (
198			df_profiles.loc[:, df["profile_id"]].sum(axis=0)
199			* df["factor_2050"].values
200			)
201			if "status2019" in scenarios:
202			df_annual_demand_iter["status2019"] = (
203			df_profiles.loc[:, df["profile_id"]].sum(axis=0)
204			* df["factor_2019"].values
205			)
206
207			if "status2023" in scenarios:
208			df_annual_demand_iter["status2023"] = (
209			df_profiles.loc[:, df["profile_id"]].sum(axis=0)
210			* df["factor_2023"].values
211			)
212			df_annual_demand_iter["zensus_population_id"] = df["cell_id"].values
213			df_annual_demand = pd.concat([df_annual_demand, df_annual_demand_iter])
214
215			df_annual_demand = (
216			df_annual_demand.groupby("zensus_population_id").sum().reset_index()
217			)
218			df_annual_demand["sector"] = "residential"
219			df_annual_demand = df_annual_demand.melt(
220			id_vars=["zensus_population_id", "sector"],
221			var_name="scenario",
222			value_name="demand",
223			)
224			# convert from Wh to MWh
225			df_annual_demand["demand"] = df_annual_demand["demand"] / 1e6
226
227			# delete all cells for residentials
228			with db.session_scope() as session:
229			session.query(EgonDemandRegioZensusElectricity).filter(
230			EgonDemandRegioZensusElectricity.sector == "residential"
231			).delete()
232
233			# Insert data to target table
234			df_annual_demand.to_sql(
235			name=EgonDemandRegioZensusElectricity.__table__.name,
236			schema=EgonDemandRegioZensusElectricity.__table__.schema,
237			con=db.engine(),
238			index=False,
239			if_exists="append",
240			)
241
242
243			def distribute_cts_demands():
244			"""Distribute electrical demands for cts to zensus cells.
245
246			The demands on nuts3-level from demandregio are linear distributed
247			to the heat demand of cts in each zensus cell.
248
249			Returns
250			-------
251			None.
252
253			"""
254
255			sources = egon.data.config.datasets()["electrical_demands_cts"]["sources"]
256
257			target = egon.data.config.datasets()["electrical_demands_cts"]["targets"][
258			"cts_demands_zensus"
259			]
260
261			db.execute_sql(
262			f"""DELETE FROM {target['schema']}.{target['table']}
263			WHERE sector = 'service'"""
264			)
265
266			# Select match between zensus cells and nuts3 regions of vg250
267			map_nuts3 = db.select_dataframe(
268			f"""SELECT zensus_population_id, vg250_nuts3 as nuts3 FROM
269			{sources['map_zensus_vg250']['schema']}.
270			{sources['map_zensus_vg250']['table']}""",
271			index_col="zensus_population_id",
272			)
273
274			# Insert data per scenario
275			for scn in egon.data.config.settings()["egon-data"]["--scenarios"]:
276			# Select heat_demand per zensus cell
277			peta = db.select_dataframe(
278			f"""SELECT zensus_population_id, demand as heat_demand,
279			sector, scenario FROM
280			{sources['heat_demand_cts']['schema']}.
281			{sources['heat_demand_cts']['table']}
282			WHERE scenario = '{scn}'
283			AND sector = 'service'""",
284			index_col="zensus_population_id",
285			)
286
287			# Add nuts3 key to zensus cells
288			peta["nuts3"] = map_nuts3.nuts3
289
290			# Calculate share of nuts3 heat demand per zensus cell
291			for nuts3, df in peta.groupby("nuts3"):
292			peta.loc[df.index, "share"] = (
293			df["heat_demand"] / df["heat_demand"].sum()
294			)
295
296			# Select forecasted electrical demands from demandregio table
297			demand_nuts3 = db.select_dataframe(
298			f"""SELECT nuts3, SUM(demand) as demand FROM
299			{sources['demandregio']['schema']}.
300			{sources['demandregio']['table']}
301			WHERE scenario = '{scn}'
302			AND wz IN (
303			SELECT wz FROM
304			{sources['demandregio_wz']['schema']}.
305			{sources['demandregio_wz']['table']}
306			WHERE sector = 'CTS')
307			GROUP BY nuts3""",
308			index_col="nuts3",
309			)
310
311			# Scale demands on nuts3 level linear to heat demand share
312			peta["demand"] = peta["share"].mul(
313			demand_nuts3.demand[peta["nuts3"]].values
314			)
315
316			# Rename index
317			peta.index = peta.index.rename("zensus_population_id")
318
319			# Insert data to target table
320			peta[["scenario", "demand", "sector"]].to_sql(
321			target["table"],
322			schema=target["schema"],
323			con=db.engine(),
324			if_exists="append",
325			)
326

openego / eGon-data

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

distribute_cts_demands() A

Complexity

Size

Duplication

Importance

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like