data.datasets.chp.insert_chp_egon100re() - Code Metrics - Inspection of "Features/#851 distribute chp egon100re" - openego/eGon-data - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — dev (#894)

unknown

created 2022-09-30 10:44 UTC

data.datasets.chp.insert_chp_egon100re() B

↳ Parent: data.datasets.chp

Complexity

Conditions

Size

Total Lines	83
Code Lines	42

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	42
dl	0
loc	83
rs	8.872
c	0
b	0
f	0
cc	2
nop	0

How to fix Long Method

"""
The central module containing all code dealing with combined heat and power
(CHP) plants.
"""

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

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,
    select_target,
)
from egon.data.datasets.power_plants import (
    assign_bus_id,
    assign_voltage_level,
    filter_mastr_geometry,
    scale_prox2now,
)
import pypsa
from egon.data.datasets.chp.small_chp import extension_to_areas
from pathlib import Path

Base = declarative_base()


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))


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))


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
    )


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


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()


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(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)
        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()


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",
    )


def extension_BW():
    extension_per_federal_state("BadenWuerttemberg", EgonChp)


def extension_BY():
    extension_per_federal_state("Bayern", EgonChp)


def extension_HB():
    extension_per_federal_state("Bremen", EgonChp)


def extension_BB():
    extension_per_federal_state("Brandenburg", EgonChp)


def extension_HH():
    extension_per_federal_state("Hamburg", EgonChp)


def extension_HE():
    extension_per_federal_state("Hessen", EgonChp)


def extension_MV():
    extension_per_federal_state("MecklenburgVorpommern", EgonChp)


def extension_NS():
    extension_per_federal_state("Niedersachsen", EgonChp)


def extension_NW():
    extension_per_federal_state("NordrheinWestfalen", EgonChp)


def extension_SN():
    extension_per_federal_state("Sachsen", EgonChp)


def extension_TH():
    extension_per_federal_state("Thueringen", EgonChp)


def extension_SL():
    extension_per_federal_state("Saarland", EgonChp)


def extension_ST():
    extension_per_federal_state("SachsenAnhalt", EgonChp)


def extension_RP():
    extension_per_federal_state("RheinlandPfalz", EgonChp)


def extension_BE():
    extension_per_federal_state("Berlin", EgonChp)


def extension_SH():
    extension_per_federal_state("SchleswigHolstein", EgonChp)


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,
    }


class Chp(Dataset):
    def __init__(self, dependencies):
        super().__init__(
            name="Chp",
            version="0.0.5",
            dependencies=dependencies,
            tasks=(
                create_tables,
                {insert_chp_egon2035, insert_chp_egon100re},
                assign_heat_bus,
                extension,
            ),
        )


1			"""
2			The central module containing all code dealing with combined heat and power
3			(CHP) plants.
4			"""
5
6			from geoalchemy2 import Geometry
7			from shapely.ops import nearest_points
8			from sqlalchemy import Boolean, Column, Float, Integer, Sequence, String
9			from sqlalchemy.dialects.postgresql import JSONB
10			from sqlalchemy.ext.declarative import declarative_base
11			from sqlalchemy.orm import sessionmaker
12			import geopandas as gpd
13			import pandas as pd
14
15			from egon.data import config, db
16			from egon.data.datasets import Dataset
17			from egon.data.datasets.chp.match_nep import insert_large_chp
18			from egon.data.datasets.chp.small_chp import (
19			assign_use_case,
20			existing_chp_smaller_10mw,
21			extension_per_federal_state,
22			select_target,
23			)
24			from egon.data.datasets.power_plants import (
25			assign_bus_id,
26			assign_voltage_level,
27			filter_mastr_geometry,
28			scale_prox2now,
29			)
30			import pypsa
31			from egon.data.datasets.chp.small_chp import extension_to_areas
32			from pathlib import Path
33
34			Base = declarative_base()
35
36
37			class EgonChp(Base):
38			__tablename__ = "egon_chp_plants"
39			__table_args__ = {"schema": "supply"}
40			id = Column(Integer, Sequence("chp_seq"), primary_key=True)
41			sources = Column(JSONB)
42			source_id = Column(JSONB)
43			carrier = Column(String)
44			district_heating = Column(Boolean)
45			el_capacity = Column(Float)
46			th_capacity = Column(Float)
47			electrical_bus_id = Column(Integer)
48			district_heating_area_id = Column(Integer)
49			ch4_bus_id = Column(Integer)
50			voltage_level = Column(Integer)
51			scenario = Column(String)
52			geom = Column(Geometry("POINT", 4326))
53
54
55			class EgonMaStRConventinalWithoutChp(Base):
56			__tablename__ = "egon_mastr_conventional_without_chp"
57			__table_args__ = {"schema": "supply"}
58			id = Column(Integer, Sequence("mastr_conventional_seq"), primary_key=True)
59			EinheitMastrNummer = Column(String)
60			carrier = Column(String)
61			el_capacity = Column(Float)
62			plz = Column(Integer)
63			city = Column(String)
64			federal_state = Column(String)
65			geometry = Column(Geometry("POINT", 4326))
66
67
68			def create_tables():
69			"""Create tables for chp data
70			Returns
71			-------
72			None.
73			"""
74
75			db.execute_sql("CREATE SCHEMA IF NOT EXISTS supply;")
76			engine = db.engine()
77			EgonChp.__table__.drop(bind=engine, checkfirst=True)
78			EgonChp.__table__.create(bind=engine, checkfirst=True)
79			EgonMaStRConventinalWithoutChp.__table__.drop(bind=engine, checkfirst=True)
80			EgonMaStRConventinalWithoutChp.__table__.create(
81			bind=engine, checkfirst=True
82			)
83
84
85			def nearest(
86			row,
87			df,
88			centroid=False,
89			row_geom_col="geometry",
90			df_geom_col="geometry",
91			src_column=None,
92			):
93			"""
94			Finds the nearest point and returns the specified column values
95
96			Parameters
97			----------
98			row : pandas.Series
99			Data to which the nearest data of df is assigned.
100			df : pandas.DataFrame
101			Data which includes all options for the nearest neighbor alogrithm.
102			centroid : boolean
103			Use centroid geoemtry. The default is False.
104			row_geom_col : str, optional
105			Name of row's geometry column. The default is 'geometry'.
106			df_geom_col : str, optional
107			Name of df's geometry column. The default is 'geometry'.
108			src_column : str, optional
109			Name of returned df column. The default is None.
110
111			Returns
112			-------
113			value : pandas.Series
114			Values of specified column of df
115
116			"""
117
118			if centroid:
119			unary_union = df.centroid.unary_union
120			else:
121			unary_union = df[df_geom_col].unary_union
122			# Find the geometry that is closest
123			nearest = (
124			df[df_geom_col] == nearest_points(row[row_geom_col], unary_union)[1]
125			)
126
127			# Get the corresponding value from df (matching is based on the geometry)
128			value = df[nearest][src_column].values[0]
129
130			return value
131
132
133			def assign_heat_bus(scenario="eGon2035"):
134			"""Selects heat_bus for chps used in district heating.
135
136			Parameters
137			----------
138			scenario : str, optional
139			Name of the corresponding scenario. The default is 'eGon2035'.
140
141			Returns
142			-------
143			None.
144
145			"""
146			sources = config.datasets()["chp_location"]["sources"]
147			target = config.datasets()["chp_location"]["targets"]["chp_table"]
148
149			# Select CHP with use_case = 'district_heating'
150			chp = db.select_geodataframe(
151			f"""
152			SELECT * FROM
153			{target['schema']}.{target['table']}
154			WHERE scenario = '{scenario}'
155			AND district_heating = True
156			""",
157			index_col="id",
158			epsg=4326,
159			)
160
161			# Select district heating areas and their centroid
162			district_heating = db.select_geodataframe(
163			f"""
164			SELECT area_id, ST_Centroid(geom_polygon) as geom
165			FROM
166			{sources['district_heating_areas']['schema']}.
167			{sources['district_heating_areas']['table']}
168			WHERE scenario = '{scenario}'
169			""",
170			epsg=4326,
171			)
172
173			# Assign district heating area_id to district_heating_chp
174			# According to nearest centroid of district heating area
175			chp["district_heating_area_id"] = chp.apply(
176			nearest,
177			df=district_heating,
178			row_geom_col="geom",
179			df_geom_col="geom",
180			centroid=True,
181			src_column="area_id",
182			axis=1,
183			)
184
185			# Drop district heating CHP without heat_bus_id
186			db.execute_sql(
187			f"""
188			DELETE FROM {target['schema']}.{target['table']}
189			WHERE scenario = '{scenario}'
190			AND district_heating = True
191			"""
192			)
193
194			# Insert district heating CHP with heat_bus_id
195			session = sessionmaker(bind=db.engine())()
196			for i, row in chp.iterrows():
197			if row.carrier != "biomass":
198			entry = EgonChp(
199			id=i,
200			sources=row.sources,
201			source_id=row.source_id,
202			carrier=row.carrier,
203			el_capacity=row.el_capacity,
204			th_capacity=row.th_capacity,
205			electrical_bus_id=row.electrical_bus_id,
206			ch4_bus_id=row.ch4_bus_id,
207			district_heating_area_id=row.district_heating_area_id,
208			district_heating=row.district_heating,
209			voltage_level=row.voltage_level,
210			scenario=scenario,
211			geom=f"SRID=4326;POINT({row.geom.x} {row.geom.y})",
212			)
213			else:
214			entry = EgonChp(
215			id=i,
216			sources=row.sources,
217			source_id=row.source_id,
218			carrier=row.carrier,
219			el_capacity=row.el_capacity,
220			th_capacity=row.th_capacity,
221			electrical_bus_id=row.electrical_bus_id,
222			district_heating_area_id=row.district_heating_area_id,
223			district_heating=row.district_heating,
224			voltage_level=row.voltage_level,
225			scenario=scenario,
226			geom=f"SRID=4326;POINT({row.geom.x} {row.geom.y})",
227			)
228			session.add(entry)
229			session.commit()
230
231
232			def insert_biomass_chp(scenario):
233			"""Insert biomass chp plants of future scenario
234
235			Parameters
236			----------
237			scenario : str
238			Name of scenario.
239
240			Returns
241			-------
242			None.
243
244			"""
245			cfg = config.datasets()["chp_location"]
246
247			# import target values from NEP 2021, scneario C 2035
248			target = select_target("biomass", scenario)
249
250			# import data for MaStR
251			mastr = pd.read_csv(cfg["sources"]["mastr_biomass"]).query(
252			"EinheitBetriebsstatus=='InBetrieb'"
253			)
254
255			# Drop entries without federal state or 'AusschließlichWirtschaftszone'
256			mastr = mastr[
257			mastr.Bundesland.isin(
258			pd.read_sql(
259			f"""SELECT DISTINCT ON (gen)
260			REPLACE(REPLACE(gen, '-', ''), 'ü', 'ue') as states
261			FROM {cfg['sources']['vg250_lan']['schema']}.
262			{cfg['sources']['vg250_lan']['table']}""",
263			con=db.engine(),
264			).states.values
265			)
266			]
267
268			# Scaling will be done per federal state in case of eGon2035 scenario.
269			if scenario == "eGon2035":
270			level = "federal_state"
271			else:
272			level = "country"
273			# Choose only entries with valid geometries inside DE/test mode
274			mastr_loc = filter_mastr_geometry(mastr).set_geometry("geometry")
275
276			# Scale capacities to meet target values
277			mastr_loc = scale_prox2now(mastr_loc, target, level=level)
278
279			# Assign bus_id
280			if len(mastr_loc) > 0:
281			mastr_loc["voltage_level"] = assign_voltage_level(mastr_loc, cfg)
282			mastr_loc = assign_bus_id(mastr_loc, cfg)
283			mastr_loc = assign_use_case(mastr_loc, cfg["sources"])
284
285			# Insert entries with location
286			session = sessionmaker(bind=db.engine())()
287			for i, row in mastr_loc.iterrows():
288			if row.ThermischeNutzleistung > 0:
289			entry = EgonChp(
290			sources={
291			"chp": "MaStR",
292			"el_capacity": "MaStR scaled with NEP 2021",
293			"th_capacity": "MaStR",
294			},
295			source_id={"MastrNummer": row.EinheitMastrNummer},
296			carrier="biomass",
297			el_capacity=row.Nettonennleistung,
298			th_capacity=row.ThermischeNutzleistung / 1000,
299			scenario=scenario,
300			district_heating=row.district_heating,
301			electrical_bus_id=row.bus_id,
302			voltage_level=row.voltage_level,
303			geom=f"SRID=4326;POINT({row.Laengengrad} {row.Breitengrad})",
304			)
305			session.add(entry)
306			session.commit()
307
308
309			def insert_chp_egon2035():
310			"""Insert CHP plants for eGon2035 considering NEP and MaStR data
311
312			Returns
313			-------
314			None.
315
316			"""
317
318			sources = config.datasets()["chp_location"]["sources"]
319
320			targets = config.datasets()["chp_location"]["targets"]
321
322			insert_biomass_chp("eGon2035")
323
324			# Insert large CHPs based on NEP's list of conventional power plants
325			MaStR_konv = insert_large_chp(sources, targets["chp_table"], EgonChp)
326
327			# Insert smaller CHPs (< 10MW) based on existing locations from MaStR
328			existing_chp_smaller_10mw(sources, MaStR_konv, EgonChp)
329
330			gpd.GeoDataFrame(
331			MaStR_konv[
332			[
333			"EinheitMastrNummer",
334			"el_capacity",
335			"geometry",
336			"carrier",
337			"plz",
338			"city",
339			"federal_state",
340			]
341			]
342			).to_postgis(
343			targets["mastr_conventional_without_chp"]["table"],
344			schema=targets["mastr_conventional_without_chp"]["schema"],
345			con=db.engine(),
346			if_exists="replace",
347			)
348
349
350			def extension_BW():
351			extension_per_federal_state("BadenWuerttemberg", EgonChp)
352
353
354			def extension_BY():
355			extension_per_federal_state("Bayern", EgonChp)
356
357
358			def extension_HB():
359			extension_per_federal_state("Bremen", EgonChp)
360
361
362			def extension_BB():
363			extension_per_federal_state("Brandenburg", EgonChp)
364
365
366			def extension_HH():
367			extension_per_federal_state("Hamburg", EgonChp)
368
369
370			def extension_HE():
371			extension_per_federal_state("Hessen", EgonChp)
372
373
374			def extension_MV():
375			extension_per_federal_state("MecklenburgVorpommern", EgonChp)
376
377
378			def extension_NS():
379			extension_per_federal_state("Niedersachsen", EgonChp)
380
381
382			def extension_NW():
383			extension_per_federal_state("NordrheinWestfalen", EgonChp)
384
385
386			def extension_SN():
387			extension_per_federal_state("Sachsen", EgonChp)
388
389
390			def extension_TH():
391			extension_per_federal_state("Thueringen", EgonChp)
392
393
394			def extension_SL():
395			extension_per_federal_state("Saarland", EgonChp)
396
397
398			def extension_ST():
399			extension_per_federal_state("SachsenAnhalt", EgonChp)
400
401
402			def extension_RP():
403			extension_per_federal_state("RheinlandPfalz", EgonChp)
404
405
406			def extension_BE():
407			extension_per_federal_state("Berlin", EgonChp)
408
409
410			def extension_SH():
411			extension_per_federal_state("SchleswigHolstein", EgonChp)
412
413
414			def insert_chp_egon100re():
415			"""Insert CHP plants for eGon100RE considering results from pypsa-eur-sec
416
417			Returns
418			-------
419			None.
420
421			"""
422
423			sources = config.datasets()["chp_location"]["sources"]
424
425			db.execute_sql(
426			f"""
427			DELETE FROM {EgonChp.__table__.schema}.{EgonChp.__table__.name}
428			WHERE scenario = 'eGon100RE'
429			"""
430			)
431
432			# select target values from pypsa-eur-sec
433			additional_capacity = db.select_dataframe(
434			"""
435			SELECT capacity
436			FROM supply.egon_scenario_capacities
437			WHERE scenario_name = 'eGon100RE'
438			AND carrier = 'urban_central_gas_CHP'
439			"""
440			).capacity[0]
441
442			if config.settings()["egon-data"]["--dataset-boundary"] != "Everything":
443			additional_capacity /= 16
444			target_file = (
445			Path(".")
446			/ "data_bundle_egon_data"
447			/ "pypsa_eur_sec"
448			/ "2022-07-26-egondata-integration"
449			/ "postnetworks"
450			/ "elec_s_37_lv2.0__Co2L0-1H-T-H-B-I-dist1_2050.nc"
451			)
452
453			network = pypsa.Network(str(target_file))
454			chp_index = "DE0 0 urban central gas CHP"
455
456			standard_chp_th = 10
457			standard_chp_el = (
458			standard_chp_th
459			* network.links.loc[chp_index, "efficiency"]
460			/ network.links.loc[chp_index, "efficiency2"]
461			)
462
463			areas = db.select_geodataframe(
464			f"""
465			SELECT
466			residential_and_service_demand as demand, area_id,
467			ST_Transform(ST_PointOnSurface(geom_polygon), 4326) as geom
468			FROM
469			{sources['district_heating_areas']['schema']}.
470			{sources['district_heating_areas']['table']}
471			WHERE scenario = 'eGon100RE'
472			"""
473			)
474
475			existing_chp = pd.DataFrame(
476			data={
477			"el_capacity": standard_chp_el,
478			"th_capacity": standard_chp_th,
479			"voltage_level": 5,
480			},
481			index=range(1),
482			)
483
484			flh = (
485			network.links_t.p0[chp_index].sum()
486			/ network.links.p_nom_opt[chp_index]
487			)
488
489			extension_to_areas(
490			areas,
491			additional_capacity,
492			existing_chp,
493			flh,
494			EgonChp,
495			district_heating=True,
496			scenario="eGon100RE",
497			)
498
499
500			# Add one task per federal state for small CHP extension
501			if (
502			config.settings()["egon-data"]["--dataset-boundary"]
503			== "Schleswig-Holstein"
504			):
505			extension = extension_SH
506			else:
507			extension = {
508			extension_BW,
509			extension_BY,
510			extension_HB,
511			extension_BB,
512			extension_HE,
513			extension_MV,
514			extension_NS,
515			extension_NW,
516			extension_SH,
517			extension_HH,
518			extension_RP,
519			extension_SL,
520			extension_SN,
521			extension_ST,
522			extension_TH,
523			extension_BE,
524			}
525
526
527			class Chp(Dataset):
528			def __init__(self, dependencies):
529			super().__init__(
530			name="Chp",
531			version="0.0.5",
532			dependencies=dependencies,
533			tasks=(
534			create_tables,
535			{insert_chp_egon2035, insert_chp_egon100re},
536			assign_heat_bus,
537			extension,
538			),
539			)
540

openego / eGon-data

Pull Request — dev (#894)

data.datasets.chp.insert_chp_egon100re() B

Complexity

Size

Duplication

Importance

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like