data.datasets.chp.insert_biomass_chp() - Code Metrics - Inspection of "Features/#450 add metadata" - openego/eGon-data - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — dev (#568)

unknown

created 2025-04-16 13:02 UTC

data.datasets.chp.insert_biomass_chp() B

↳ Parent: data.datasets.chp

Complexity

Conditions

Size

Total Lines	77
Code Lines	41

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	41
dl	0
loc	77
rs	8.4293
c	0
b	0
f	0
cc	5
nop	1

How to fix Long Method

"""
The central module containing all code dealing with combined heat and power
(CHP) plants.
"""
from pathlib import Path
import datetime
import json
import time

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
import pypsa

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,
    extension_to_areas,
    select_target,
)
from egon.data.datasets.mastr import WORKING_DIR_MASTR_OLD
from egon.data.datasets.power_plants import (
    assign_bus_id,
    assign_voltage_level,
    filter_mastr_geometry,
    scale_prox2now,
)
from egon.data.metadata import (
    context,
    generate_resource_fields_from_sqla_model,
    license_egon_data_odbl,
    meta_metadata,
    sources,
)

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 metadata():
    """Write metadata for heat supply tables

    Returns
    -------
    None.

    """

    fields = generate_resource_fields_from_sqla_model(EgonChp)

    fields_df = pd.DataFrame(data=fields).set_index("name")
    fields_df.loc["id", "description"] = "Unique identifyer"
    fields_df.loc["sources", "description"] = "List of sources"
    fields_df.loc[
        "source_id", "description"
    ] = "Names of sources, e.g. MaStr_id"
    fields_df.loc["carrier", "description"] = "Energy carrier"
    fields_df.loc[
        "district_heating", "description"
    ] = "Used in district heating or not"
    fields_df.loc[
        "el_capacity", "description"
    ] = "Installed electrical capacity"
    fields_df.loc["th_capacity", "description"] = "Installed thermal capacity"
    fields_df.loc[
        "electrical_bus_id", "description"
    ] = "Index of corresponding electricity bus"
    fields_df.loc[
        "district_heating_area_id", "description"
    ] = "Index of corresponding district heating bus"
    fields_df.loc[
        "ch4_bus_id", "description"
    ] = "Index of corresponding methane bus"
    fields_df.loc["voltage_level", "description"] = "Voltage level"
    fields_df.loc["scenario", "description"] = "Name of scenario"
    fields_df.loc["geom", "description"] = "Location of CHP plant"

    fields_df.loc["el_capacity", "unit"] = "MW_el"
    fields_df.loc["th_capacity", "unit"] = "MW_th"
    fields_df.unit.fillna("none", inplace=True)

    fields = fields_df.reset_index().to_dict(orient="records")

    meta_district = {
        "name": "supply.egon_chp_plants",
        "title": "eGon combined heat and power plants",
        "id": "WILL_BE_SET_AT_PUBLICATION",
        "description": "Combined heat and power plants",
        "language": ["EN"],
        "publicationDate": datetime.date.today().isoformat(),
        "context": context(),
        "spatial": {
            "location": None,
            "extent": "Germany",
            "resolution": None,
        },
        "sources": [
            sources()["vg250"],
            sources()["egon-data"],
            sources()["egon-data_bundle"],
            sources()["openstreetmap"],
            sources()["mastr"],
        ],
        "licenses": [license_egon_data_odbl()],
        "contributors": [
            {
                "title": "Clara Büttner",
                "email": "http://github.com/ClaraBuettner",
                "date": time.strftime("%Y-%m-%d"),
                "object": None,
                "comment": "Imported data",
            },
        ],
        "resources": [
            {
                "profile": "tabular-data-resource",
                "name": "supply.egon_chp_plants",
                "path": None,
                "format": "PostgreSQL",
                "encoding": "UTF-8",
                "schema": {
                    "fields": fields,
                    "primaryKey": ["index"],
                    "foreignKeys": [],
                },
                "dialect": {"delimiter": None, "decimalSeparator": "."},
            }
        ],
        "metaMetadata": meta_metadata(),
    }

    # Add metadata as a comment to the table
    db.submit_comment(
        "'" + json.dumps(meta_district) + "'",
        EgonChp.__table__.schema,
        EgonChp.__table__.name,
    )


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(
        WORKING_DIR_MASTR_OLD / 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, WORKING_DIR_MASTR_OLD
        )
        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):
    """
    Extract combined heat and power plants for each scenario

    This dataset creates combined heat and power (CHP) plants for each scenario and defines their use case.
    The method bases on existing CHP plants from Marktstammdatenregister. For the eGon2035 scenario,
    a list of CHP plans from the grid operator is used for new largescale CHP plants. CHP < 10MW are
    randomly distributed.
    Depending on the distance to a district heating grid, it is decided if the CHP is used to
    supply a district heating grid or used by an industrial site.


    *Dependencies*
      * :py:class:`GasAreaseGon100RE <egon.data.datasets.gas_areas.GasAreaseGon100RE>`
      * :py:class:`GasAreaseGon2035 <egon.data.datasets.gas_areas.GasAreaseGon2035>`
      * :py:class:`DistrictHeatingAreas <egon.data.datasets.district_heating_areas.DistrictHeatingAreas>`
      * :py:class:`IndustrialDemandCurves <egon.data.datasets.industry.IndustrialDemandCurves>`
      * :py:class:`OsmLanduse <egon.data.datasets.loadarea.OsmLanduse>`
      * :py:func:`download_mastr_data <egon.data.datasets.mastr.download_mastr_data>`
      * :py:func:`define_mv_grid_districts <egon.data.datasets.mv_grid_districts.define_mv_grid_districts>`
      * :py:class:`ScenarioCapacities <egon.data.datasets.scenario_capacities.ScenarioCapacities>`


    *Resulting tables*
      * :py:class:`supply.egon_chp_plants <egon.data.datasets.chp.EgonChp>` is created and filled
      * :py:class:`supply.egon_mastr_conventional_without_chp <egon.data.datasets.chp.EgonMaStRConventinalWithoutChp>` is created and filled

    """

    #:
    name: str = "Chp"
    #:
    version: str = "0.0.7"

    def __init__(self, dependencies):
        super().__init__(
            name=self.name,
            version=self.version,
            dependencies=dependencies,
            tasks=(
                create_tables,
                {insert_chp_egon2035, insert_chp_egon100re},
                assign_heat_bus,
                extension,
                metadata,
            ),
        )


1			"""
2			The central module containing all code dealing with combined heat and power
3			(CHP) plants.
4			"""
5			from pathlib import Path
6			import datetime
7			import json
8			import time
9
10			from geoalchemy2 import Geometry
11			from shapely.ops import nearest_points
12			from sqlalchemy import Boolean, Column, Float, Integer, Sequence, String
13			from sqlalchemy.dialects.postgresql import JSONB
14			from sqlalchemy.ext.declarative import declarative_base
15			from sqlalchemy.orm import sessionmaker
16			import geopandas as gpd
17			import pandas as pd
18			import pypsa
19
20			from egon.data import config, db
21			from egon.data.datasets import Dataset
22			from egon.data.datasets.chp.match_nep import insert_large_chp
23			from egon.data.datasets.chp.small_chp import (
24			assign_use_case,
25			existing_chp_smaller_10mw,
26			extension_per_federal_state,
27			extension_to_areas,
28			select_target,
29			)
30			from egon.data.datasets.mastr import WORKING_DIR_MASTR_OLD
31			from egon.data.datasets.power_plants import (
32			assign_bus_id,
33			assign_voltage_level,
34			filter_mastr_geometry,
35			scale_prox2now,
36			)
37			from egon.data.metadata import (
38			context,
39			generate_resource_fields_from_sqla_model,
40			license_egon_data_odbl,
41			meta_metadata,
42			sources,
43			)
44
45			Base = declarative_base()
46
47
48			class EgonChp(Base):
49			__tablename__ = "egon_chp_plants"
50			__table_args__ = {"schema": "supply"}
51			id = Column(Integer, Sequence("chp_seq"), primary_key=True)
52			sources = Column(JSONB)
53			source_id = Column(JSONB)
54			carrier = Column(String)
55			district_heating = Column(Boolean)
56			el_capacity = Column(Float)
57			th_capacity = Column(Float)
58			electrical_bus_id = Column(Integer)
59			district_heating_area_id = Column(Integer)
60			ch4_bus_id = Column(Integer)
61			voltage_level = Column(Integer)
62			scenario = Column(String)
63			geom = Column(Geometry("POINT", 4326))
64
65
66			class EgonMaStRConventinalWithoutChp(Base):
67			__tablename__ = "egon_mastr_conventional_without_chp"
68			__table_args__ = {"schema": "supply"}
69			id = Column(Integer, Sequence("mastr_conventional_seq"), primary_key=True)
70			EinheitMastrNummer = Column(String)
71			carrier = Column(String)
72			el_capacity = Column(Float)
73			plz = Column(Integer)
74			city = Column(String)
75			federal_state = Column(String)
76			geometry = Column(Geometry("POINT", 4326))
77
78
79			def metadata():
80			"""Write metadata for heat supply tables
81
82			Returns
83			-------
84			None.
85
86			"""
87
88			fields = generate_resource_fields_from_sqla_model(EgonChp)
89
90			fields_df = pd.DataFrame(data=fields).set_index("name")
91			fields_df.loc["id", "description"] = "Unique identifyer"
92			fields_df.loc["sources", "description"] = "List of sources"
93			fields_df.loc[
94			"source_id", "description"
95			] = "Names of sources, e.g. MaStr_id"
96			fields_df.loc["carrier", "description"] = "Energy carrier"
97			fields_df.loc[
98			"district_heating", "description"
99			] = "Used in district heating or not"
100			fields_df.loc[
101			"el_capacity", "description"
102			] = "Installed electrical capacity"
103			fields_df.loc["th_capacity", "description"] = "Installed thermal capacity"
104			fields_df.loc[
105			"electrical_bus_id", "description"
106			] = "Index of corresponding electricity bus"
107			fields_df.loc[
108			"district_heating_area_id", "description"
109			] = "Index of corresponding district heating bus"
110			fields_df.loc[
111			"ch4_bus_id", "description"
112			] = "Index of corresponding methane bus"
113			fields_df.loc["voltage_level", "description"] = "Voltage level"
114			fields_df.loc["scenario", "description"] = "Name of scenario"
115			fields_df.loc["geom", "description"] = "Location of CHP plant"
116
117			fields_df.loc["el_capacity", "unit"] = "MW_el"
118			fields_df.loc["th_capacity", "unit"] = "MW_th"
119			fields_df.unit.fillna("none", inplace=True)
120
121			fields = fields_df.reset_index().to_dict(orient="records")
122
123			meta_district = {
124			"name": "supply.egon_chp_plants",
125			"title": "eGon combined heat and power plants",
126			"id": "WILL_BE_SET_AT_PUBLICATION",
127			"description": "Combined heat and power plants",
128			"language": ["EN"],
129			"publicationDate": datetime.date.today().isoformat(),
130			"context": context(),
131			"spatial": {
132			"location": None,
133			"extent": "Germany",
134			"resolution": None,
135			},
136			"sources": [
137			sources()["vg250"],
138			sources()["egon-data"],
139			sources()["egon-data_bundle"],
140			sources()["openstreetmap"],
141			sources()["mastr"],
142			],
143			"licenses": [license_egon_data_odbl()],
144			"contributors": [
145			{
146			"title": "Clara Büttner",
147			"email": "http://github.com/ClaraBuettner",
148			"date": time.strftime("%Y-%m-%d"),
149			"object": None,
150			"comment": "Imported data",
151			},
152			],
153			"resources": [
154			{
155			"profile": "tabular-data-resource",
156			"name": "supply.egon_chp_plants",
157			"path": None,
158			"format": "PostgreSQL",
159			"encoding": "UTF-8",
160			"schema": {
161			"fields": fields,
162			"primaryKey": ["index"],
163			"foreignKeys": [],
164			},
165			"dialect": {"delimiter": None, "decimalSeparator": "."},
166			}
167			],
168			"metaMetadata": meta_metadata(),
169			}
170
171			# Add metadata as a comment to the table
172			db.submit_comment(
173			"'" + json.dumps(meta_district) + "'",
174			EgonChp.__table__.schema,
175			EgonChp.__table__.name,
176			)
177
178
179			def create_tables():
180			"""Create tables for chp data
181			Returns
182			-------
183			None.
184			"""
185
186			db.execute_sql("CREATE SCHEMA IF NOT EXISTS supply;")
187			engine = db.engine()
188			EgonChp.__table__.drop(bind=engine, checkfirst=True)
189			EgonChp.__table__.create(bind=engine, checkfirst=True)
190			EgonMaStRConventinalWithoutChp.__table__.drop(bind=engine, checkfirst=True)
191			EgonMaStRConventinalWithoutChp.__table__.create(
192			bind=engine, checkfirst=True
193			)
194
195
196			def nearest(
197			row,
198			df,
199			centroid=False,
200			row_geom_col="geometry",
201			df_geom_col="geometry",
202			src_column=None,
203			):
204			"""
205			Finds the nearest point and returns the specified column values
206
207			Parameters
208			----------
209			row : pandas.Series
210			Data to which the nearest data of df is assigned.
211			df : pandas.DataFrame
212			Data which includes all options for the nearest neighbor alogrithm.
213			centroid : boolean
214			Use centroid geoemtry. The default is False.
215			row_geom_col : str, optional
216			Name of row's geometry column. The default is 'geometry'.
217			df_geom_col : str, optional
218			Name of df's geometry column. The default is 'geometry'.
219			src_column : str, optional
220			Name of returned df column. The default is None.
221
222			Returns
223			-------
224			value : pandas.Series
225			Values of specified column of df
226
227			"""
228
229			if centroid:
230			unary_union = df.centroid.unary_union
231			else:
232			unary_union = df[df_geom_col].unary_union
233			# Find the geometry that is closest
234			nearest = (
235			df[df_geom_col] == nearest_points(row[row_geom_col], unary_union)[1]
236			)
237
238			# Get the corresponding value from df (matching is based on the geometry)
239			value = df[nearest][src_column].values[0]
240
241			return value
242
243
244			def assign_heat_bus(scenario="eGon2035"):
245			"""Selects heat_bus for chps used in district heating.
246
247			Parameters
248			----------
249			scenario : str, optional
250			Name of the corresponding scenario. The default is 'eGon2035'.
251
252			Returns
253			-------
254			None.
255
256			"""
257			sources = config.datasets()["chp_location"]["sources"]
258			target = config.datasets()["chp_location"]["targets"]["chp_table"]
259
260			# Select CHP with use_case = 'district_heating'
261			chp = db.select_geodataframe(
262			f"""
263			SELECT * FROM
264			{target['schema']}.{target['table']}
265			WHERE scenario = '{scenario}'
266			AND district_heating = True
267			""",
268			index_col="id",
269			epsg=4326,
270			)
271
272			# Select district heating areas and their centroid
273			district_heating = db.select_geodataframe(
274			f"""
275			SELECT area_id, ST_Centroid(geom_polygon) as geom
276			FROM
277			{sources['district_heating_areas']['schema']}.
278			{sources['district_heating_areas']['table']}
279			WHERE scenario = '{scenario}'
280			""",
281			epsg=4326,
282			)
283
284			# Assign district heating area_id to district_heating_chp
285			# According to nearest centroid of district heating area
286			chp["district_heating_area_id"] = chp.apply(
287			nearest,
288			df=district_heating,
289			row_geom_col="geom",
290			df_geom_col="geom",
291			centroid=True,
292			src_column="area_id",
293			axis=1,
294			)
295
296			# Drop district heating CHP without heat_bus_id
297			db.execute_sql(
298			f"""
299			DELETE FROM {target['schema']}.{target['table']}
300			WHERE scenario = '{scenario}'
301			AND district_heating = True
302			"""
303			)
304
305			# Insert district heating CHP with heat_bus_id
306			session = sessionmaker(bind=db.engine())()
307			for i, row in chp.iterrows():
308			if row.carrier != "biomass":
309			entry = EgonChp(
310			id=i,
311			sources=row.sources,
312			source_id=row.source_id,
313			carrier=row.carrier,
314			el_capacity=row.el_capacity,
315			th_capacity=row.th_capacity,
316			electrical_bus_id=row.electrical_bus_id,
317			ch4_bus_id=row.ch4_bus_id,
318			district_heating_area_id=row.district_heating_area_id,
319			district_heating=row.district_heating,
320			voltage_level=row.voltage_level,
321			scenario=scenario,
322			geom=f"SRID=4326;POINT({row.geom.x} {row.geom.y})",
323			)
324			else:
325			entry = EgonChp(
326			id=i,
327			sources=row.sources,
328			source_id=row.source_id,
329			carrier=row.carrier,
330			el_capacity=row.el_capacity,
331			th_capacity=row.th_capacity,
332			electrical_bus_id=row.electrical_bus_id,
333			district_heating_area_id=row.district_heating_area_id,
334			district_heating=row.district_heating,
335			voltage_level=row.voltage_level,
336			scenario=scenario,
337			geom=f"SRID=4326;POINT({row.geom.x} {row.geom.y})",
338			)
339			session.add(entry)
340			session.commit()
341
342
343			def insert_biomass_chp(scenario):
344			"""Insert biomass chp plants of future scenario
345
346			Parameters
347			----------
348			scenario : str
349			Name of scenario.
350
351			Returns
352			-------
353			None.
354
355			"""
356			cfg = config.datasets()["chp_location"]
357
358			# import target values from NEP 2021, scneario C 2035
359			target = select_target("biomass", scenario)
360
361			# import data for MaStR
362			mastr = pd.read_csv(
363			WORKING_DIR_MASTR_OLD / cfg["sources"]["mastr_biomass"]
364			).query("EinheitBetriebsstatus=='InBetrieb'")
365
366			# Drop entries without federal state or 'AusschließlichWirtschaftszone'
367			mastr = mastr[
368			mastr.Bundesland.isin(
369			pd.read_sql(
370			f"""SELECT DISTINCT ON (gen)
371			REPLACE(REPLACE(gen, '-', ''), 'ü', 'ue') as states
372			FROM {cfg['sources']['vg250_lan']['schema']}.
373			{cfg['sources']['vg250_lan']['table']}""",
374			con=db.engine(),
375			).states.values
376			)
377			]
378
379			# Scaling will be done per federal state in case of eGon2035 scenario.
380			if scenario == "eGon2035":
381			level = "federal_state"
382			else:
383			level = "country"
384			# Choose only entries with valid geometries inside DE/test mode
385			mastr_loc = filter_mastr_geometry(mastr).set_geometry("geometry")
386
387			# Scale capacities to meet target values
388			mastr_loc = scale_prox2now(mastr_loc, target, level=level)
389
390			# Assign bus_id
391			if len(mastr_loc) > 0:
392			mastr_loc["voltage_level"] = assign_voltage_level(
393			mastr_loc, cfg, WORKING_DIR_MASTR_OLD
394			)
395			mastr_loc = assign_bus_id(mastr_loc, cfg)
396			mastr_loc = assign_use_case(mastr_loc, cfg["sources"])
397
398			# Insert entries with location
399			session = sessionmaker(bind=db.engine())()
400			for i, row in mastr_loc.iterrows():
401			if row.ThermischeNutzleistung > 0:
402			entry = EgonChp(
403			sources={
404			"chp": "MaStR",
405			"el_capacity": "MaStR scaled with NEP 2021",
406			"th_capacity": "MaStR",
407			},
408			source_id={"MastrNummer": row.EinheitMastrNummer},
409			carrier="biomass",
410			el_capacity=row.Nettonennleistung,
411			th_capacity=row.ThermischeNutzleistung / 1000,
412			scenario=scenario,
413			district_heating=row.district_heating,
414			electrical_bus_id=row.bus_id,
415			voltage_level=row.voltage_level,
416			geom=f"SRID=4326;POINT({row.Laengengrad} {row.Breitengrad})",
417			)
418			session.add(entry)
419			session.commit()
420
421
422			def insert_chp_egon2035():
423			"""Insert CHP plants for eGon2035 considering NEP and MaStR data
424
425			Returns
426			-------
427			None.
428
429			"""
430
431			sources = config.datasets()["chp_location"]["sources"]
432
433			targets = config.datasets()["chp_location"]["targets"]
434
435			insert_biomass_chp("eGon2035")
436
437			# Insert large CHPs based on NEP's list of conventional power plants
438			MaStR_konv = insert_large_chp(sources, targets["chp_table"], EgonChp)
439
440			# Insert smaller CHPs (< 10MW) based on existing locations from MaStR
441			existing_chp_smaller_10mw(sources, MaStR_konv, EgonChp)
442
443			gpd.GeoDataFrame(
444			MaStR_konv[
445			[
446			"EinheitMastrNummer",
447			"el_capacity",
448			"geometry",
449			"carrier",
450			"plz",
451			"city",
452			"federal_state",
453			]
454			]
455			).to_postgis(
456			targets["mastr_conventional_without_chp"]["table"],
457			schema=targets["mastr_conventional_without_chp"]["schema"],
458			con=db.engine(),
459			if_exists="replace",
460			)
461
462
463			def extension_BW():
464			extension_per_federal_state("BadenWuerttemberg", EgonChp)
465
466
467			def extension_BY():
468			extension_per_federal_state("Bayern", EgonChp)
469
470
471			def extension_HB():
472			extension_per_federal_state("Bremen", EgonChp)
473
474
475			def extension_BB():
476			extension_per_federal_state("Brandenburg", EgonChp)
477
478
479			def extension_HH():
480			extension_per_federal_state("Hamburg", EgonChp)
481
482
483			def extension_HE():
484			extension_per_federal_state("Hessen", EgonChp)
485
486
487			def extension_MV():
488			extension_per_federal_state("MecklenburgVorpommern", EgonChp)
489
490
491			def extension_NS():
492			extension_per_federal_state("Niedersachsen", EgonChp)
493
494
495			def extension_NW():
496			extension_per_federal_state("NordrheinWestfalen", EgonChp)
497
498
499			def extension_SN():
500			extension_per_federal_state("Sachsen", EgonChp)
501
502
503			def extension_TH():
504			extension_per_federal_state("Thueringen", EgonChp)
505
506
507			def extension_SL():
508			extension_per_federal_state("Saarland", EgonChp)
509
510
511			def extension_ST():
512			extension_per_federal_state("SachsenAnhalt", EgonChp)
513
514
515			def extension_RP():
516			extension_per_federal_state("RheinlandPfalz", EgonChp)
517
518
519			def extension_BE():
520			extension_per_federal_state("Berlin", EgonChp)
521
522
523			def extension_SH():
524			extension_per_federal_state("SchleswigHolstein", EgonChp)
525
526
527			def insert_chp_egon100re():
528			"""Insert CHP plants for eGon100RE considering results from pypsa-eur-sec
529
530			Returns
531			-------
532			None.
533
534			"""
535
536			sources = config.datasets()["chp_location"]["sources"]
537
538			db.execute_sql(
539			f"""
540			DELETE FROM {EgonChp.__table__.schema}.{EgonChp.__table__.name}
541			WHERE scenario = 'eGon100RE'
542			"""
543			)
544
545			# select target values from pypsa-eur-sec
546			additional_capacity = db.select_dataframe(
547			"""
548			SELECT capacity
549			FROM supply.egon_scenario_capacities
550			WHERE scenario_name = 'eGon100RE'
551			AND carrier = 'urban_central_gas_CHP'
552			"""
553			).capacity[0]
554
555			if config.settings()["egon-data"]["--dataset-boundary"] != "Everything":
556			additional_capacity /= 16
557			target_file = (
558			Path(".")
559			/ "data_bundle_egon_data"
560			/ "pypsa_eur_sec"
561			/ "2022-07-26-egondata-integration"
562			/ "postnetworks"
563			/ "elec_s_37_lv2.0__Co2L0-1H-T-H-B-I-dist1_2050.nc"
564			)
565
566			network = pypsa.Network(str(target_file))
567			chp_index = "DE0 0 urban central gas CHP"
568
569			standard_chp_th = 10
570			standard_chp_el = (
571			standard_chp_th
572			* network.links.loc[chp_index, "efficiency"]
573			/ network.links.loc[chp_index, "efficiency2"]
574			)
575
576			areas = db.select_geodataframe(
577			f"""
578			SELECT
579			residential_and_service_demand as demand, area_id,
580			ST_Transform(ST_PointOnSurface(geom_polygon), 4326) as geom
581			FROM
582			{sources['district_heating_areas']['schema']}.
583			{sources['district_heating_areas']['table']}
584			WHERE scenario = 'eGon100RE'
585			"""
586			)
587
588			existing_chp = pd.DataFrame(
589			data={
590			"el_capacity": standard_chp_el,
591			"th_capacity": standard_chp_th,
592			"voltage_level": 5,
593			},
594			index=range(1),
595			)
596
597			flh = (
598			network.links_t.p0[chp_index].sum()
599			/ network.links.p_nom_opt[chp_index]
600			)
601
602			extension_to_areas(
603			areas,
604			additional_capacity,
605			existing_chp,
606			flh,
607			EgonChp,
608			district_heating=True,
609			scenario="eGon100RE",
610			)
611
612
613			# Add one task per federal state for small CHP extension
614			if (
615			config.settings()["egon-data"]["--dataset-boundary"]
616			== "Schleswig-Holstein"
617			):
618			extension = extension_SH
619			else:
620			extension = {
621			extension_BW,
622			extension_BY,
623			extension_HB,
624			extension_BB,
625			extension_HE,
626			extension_MV,
627			extension_NS,
628			extension_NW,
629			extension_SH,
630			extension_HH,
631			extension_RP,
632			extension_SL,
633			extension_SN,
634			extension_ST,
635			extension_TH,
636			extension_BE,
637			}
638
639
640			class Chp(Dataset):
641			"""
642			Extract combined heat and power plants for each scenario
643
644			This dataset creates combined heat and power (CHP) plants for each scenario and defines their use case.
645			The method bases on existing CHP plants from Marktstammdatenregister. For the eGon2035 scenario,
646			a list of CHP plans from the grid operator is used for new largescale CHP plants. CHP < 10MW are
647			randomly distributed.
648			Depending on the distance to a district heating grid, it is decided if the CHP is used to
649			supply a district heating grid or used by an industrial site.
650
651
652			Dependencies
653			* :py:class:`GasAreaseGon100RE <egon.data.datasets.gas_areas.GasAreaseGon100RE>`
654			* :py:class:`GasAreaseGon2035 <egon.data.datasets.gas_areas.GasAreaseGon2035>`
655			* :py:class:`DistrictHeatingAreas <egon.data.datasets.district_heating_areas.DistrictHeatingAreas>`
656			* :py:class:`IndustrialDemandCurves <egon.data.datasets.industry.IndustrialDemandCurves>`
657			* :py:class:`OsmLanduse <egon.data.datasets.loadarea.OsmLanduse>`
658			* :py:func:`download_mastr_data <egon.data.datasets.mastr.download_mastr_data>`
659			* :py:func:`define_mv_grid_districts <egon.data.datasets.mv_grid_districts.define_mv_grid_districts>`
660			* :py:class:`ScenarioCapacities <egon.data.datasets.scenario_capacities.ScenarioCapacities>`
661
662
663			Resulting tables
664			* :py:class:`supply.egon_chp_plants <egon.data.datasets.chp.EgonChp>` is created and filled
665			* :py:class:`supply.egon_mastr_conventional_without_chp <egon.data.datasets.chp.EgonMaStRConventinalWithoutChp>` is created and filled
666
667			"""
668
669			#:
670			name: str = "Chp"
671			#:
672			version: str = "0.0.7"
673
674			def __init__(self, dependencies):
675			super().__init__(
676			name=self.name,
677			version=self.version,
678			dependencies=dependencies,
679			tasks=(
680			create_tables,
681			{insert_chp_egon2035, insert_chp_egon100re},
682			assign_heat_bus,
683			extension,
684			metadata,
685			),
686			)
687

openego / eGon-data

Pull Request — dev (#568)

data.datasets.chp.insert_biomass_chp() B

Complexity

Size

Duplication

Importance

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like