data.datasets.electricity_demand.CtsElectricityDemand.__init__() - Code Metrics - Inspection of "feature/validation_integration" - openego/eGon-data - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — dev (#1375)

unknown

created 2026-01-19 10:52 UTC

CtsElectricityDemand.init() A

↳ Parent: data.datasets.electricity_demand

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 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.validation.rules.custom.sanity import (
    ResidentialElectricityAnnualSum,
    ResidentialElectricityHhRefinement,
)
from egon_validation import (
    RowCountValidation,
    DataTypeValidation,
    WholeTableNotNullAndNotNaNValidation,
    ValueSetValidation
)

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),
            validation={
                "data_quality": [
                    ResidentialElectricityAnnualSum(
                        table="demand.egon_demandregio_zensus_electricity",
                        rule_id="SANITY_RESIDENTIAL_ELECTRICITY_ANNUAL_SUM",
                        rtol=0.005
                    ),
                    ResidentialElectricityHhRefinement(
                        table="society.egon_destatis_zensus_household_per_ha_refined",
                        rule_id="SANITY_RESIDENTIAL_HH_REFINEMENT",
                        rtol=1e-5
                    ),
                    RowCountValidation(
                        table=" demand.egon_demandregio_zensus_electricity",
                        rule_id="ROW_COUNT.egon_demandregio_zensus_electricity",
                        expected_count={"Schleswig-Holstein": 154527, "Everything": 7355160}
                    ),
                    DataTypeValidation(
                        table="demand.egon_demandregio_zensus_electricity",
                        rule_id="DATA_MULTIPLE_TYPES.egon_demandregio_zensus_electricity",
                        column_types={"zensus_population_id": "integer", "scenario": "character varying", "sector": "character varying", "demand": "double precision"}
                    ),
                    WholeTableNotNullAndNotNaNValidation(
                        table="demand.egon_demandregio_zensus_electricity",
                        rule_id="WHOLE_TABLE_NOT_NAN.egon_demandregio_zensus_electricity"
                    ),
                    ValueSetValidation(
                        table="demand.egon_demandregio_zensus_electricity",
                        rule_id="VALUE_SET_VALIDATION_SCENARIO.egon_demandregio_zensus_electricity",
                        column="scenario",
                        expected_values=["eGon2035", "eGon100RE"]
                    ),
                    ValueSetValidation(
                        table="demand.egon_demandregio_zensus_electricity",
                        rule_id="VALUE_SET_VALIDATION_SECTOR.egon_demandregio_zensus_electricity",
                        column="sector",
                        expected_values=["residential", "service"]
                    ),
                ]
            },
            on_validation_failure="continue"
        )


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

openego / eGon-data

Pull Request — dev (#1375)

CtsElectricityDemand.__init__() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

CtsElectricityDemand.init() A