data.datasets.storages.allocate_storage_units_sq() - Code Metrics - Inspection of "Fix/#1180 missing cts buildings" - openego/eGon-data - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — dev (#1181)

unknown

created 2025-02-03 09:46 UTC

data.datasets.storages.allocate_storage_units_sq() D

↳ Parent: data.datasets.storages

Complexity

Conditions

Size

Total Lines	265
Code Lines	152

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	152
dl	0
loc	265
rs	4.6666
c	0
b	0
f	0
cc	9
nop	2

How to fix Long Method

"""The central module containing all code dealing with power plant data.
"""

from pathlib import Path

from geoalchemy2 import Geometry
from sqlalchemy import BigInteger, 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.scenario_parameters import get_sector_parameters
from egon.data.datasets.electrical_neighbours import entsoe_to_bus_etrago
from egon.data.datasets.mastr import (
    WORKING_DIR_MASTR_NEW,
    WORKING_DIR_MASTR_OLD,
)
from egon.data.datasets.mv_grid_districts import Vg250GemClean
from egon.data.datasets.power_plants import assign_bus_id, assign_voltage_level
from egon.data.datasets.storages.home_batteries import (
    allocate_home_batteries_to_buildings,
)
from egon.data.datasets.storages.pumped_hydro import (
    apply_voltage_level_thresholds,
    get_location,
    match_storage_units,
    select_mastr_pumped_hydro,
    select_nep_pumped_hydro,
)
from egon.data.db import session_scope

Base = declarative_base()


class EgonStorages(Base):

    __tablename__ = "egon_storages"
    __table_args__ = {"schema": "supply"}
    id = Column(BigInteger, Sequence("storage_seq"), primary_key=True)
    sources = Column(JSONB)
    source_id = Column(JSONB)
    carrier = Column(String)
    el_capacity = Column(Float)
    bus_id = Column(Integer)
    voltage_level = Column(Integer)
    scenario = Column(String)
    geom = Column(Geometry("POINT", 4326))


class Storages(Dataset):
    def __init__(self, dependencies):
        super().__init__(
            name="Storages",
            version="0.0.8",
            dependencies=dependencies,
            tasks=(
                create_tables,
                allocate_pumped_hydro_scn,
                allocate_other_storage_units,
                # allocate_pv_home_batteries_to_grids,
                # allocate_home_batteries_to_buildings,
            ),
        )


def create_tables():
    """Create tables for power plant data
    Returns
    -------
    None.
    """

    cfg = config.datasets()["storages"]
    db.execute_sql(f"CREATE SCHEMA IF NOT EXISTS {cfg['target']['schema']};")
    engine = db.engine()
    db.execute_sql(
        f"""DROP TABLE IF EXISTS
        {cfg['target']['schema']}.{cfg['target']['table']}"""
    )

    db.execute_sql("""DROP SEQUENCE IF EXISTS pp_seq""")
    EgonStorages.__table__.create(bind=engine, checkfirst=True)


def allocate_pumped_hydro(scn, export=True):
    """Allocates pumped_hydro plants for eGon2035 and scenario2019 scenarios
    and either exports results to data base or returns as a dataframe

    Parameters
    ----------
    export : bool
        Choose if allocated pumped hydro plants should be exported to the data
        base. The default is True.
        If export=False a data frame will be returned

    Returns
    -------
    power_plants : pandas.DataFrame
        List of pumped hydro plants in 'eGon2035' and 'scenario2019' scenarios
    """

    carrier = "pumped_hydro"

    cfg = config.datasets()["power_plants"]

    nep = select_nep_pumped_hydro(scn=scn)
    mastr = select_mastr_pumped_hydro()

    # Assign voltage level to MaStR
    mastr["voltage_level"] = assign_voltage_level(
        mastr.rename({"el_capacity": "Nettonennleistung"}, axis=1),
        cfg,
        WORKING_DIR_MASTR_OLD,
    )

    # Initalize DataFrame for matching power plants
    matched = gpd.GeoDataFrame(
        columns=[
            "carrier",
            "el_capacity",
            "scenario",
            "geometry",
            "MaStRNummer",
            "source",
            "voltage_level",
        ]
    )

    # Match pumped_hydro units from NEP list
    # using PLZ and capacity
    matched, mastr, nep = match_storage_units(
        nep,
        mastr,
        matched,
        buffer_capacity=0.1,
        consider_carrier=False,
        scn=scn,
    )

    # Match plants from NEP list using plz,
    # neglecting the capacity
    matched, mastr, nep = match_storage_units(
        nep,
        mastr,
        matched,
        consider_location="plz",
        consider_carrier=False,
        consider_capacity=False,
        scn=scn,
    )

    # Match plants from NEP list using city,
    # neglecting the capacity
    matched, mastr, nep = match_storage_units(
        nep,
        mastr,
        matched,
        consider_location="city",
        consider_carrier=False,
        consider_capacity=False,
        scn=scn,
    )

    # Match remaining plants from NEP using the federal state
    matched, mastr, nep = match_storage_units(
        nep,
        mastr,
        matched,
        buffer_capacity=0.1,
        consider_location="federal_state",
        consider_carrier=False,
        scn=scn,
    )

    # Match remaining plants from NEP using the federal state
    matched, mastr, nep = match_storage_units(
        nep,
        mastr,
        matched,
        buffer_capacity=0.7,
        consider_location="federal_state",
        consider_carrier=False,
        scn=scn,
    )

    print(f"{matched.el_capacity.sum()} MW of {carrier} matched")
    print(f"{nep.elec_capacity.sum()} MW of {carrier} not matched")

    if nep.elec_capacity.sum() > 0:
        # Get location using geolocator and city information
        located, unmatched = get_location(nep)

        # Bring both dataframes together
        matched = pd.concat(
            [
                matched,
                located[
                    [
                        "carrier",
                        "el_capacity",
                        "scenario",
                        "geometry",
                        "source",
                        "MaStRNummer",
                    ]
                ],
            ],
            ignore_index=True,
        )

    # Set CRS
    matched.crs = "EPSG:4326"

    # Assign voltage level
    matched = apply_voltage_level_thresholds(matched)

    # Assign bus_id
    # Load grid district polygons
    mv_grid_districts = db.select_geodataframe(
        f"""
    SELECT * FROM {cfg['sources']['egon_mv_grid_district']}
    """,
        epsg=4326,
    )

    ehv_grid_districts = db.select_geodataframe(
        f"""
    SELECT * FROM {cfg['sources']['ehv_voronoi']}
    """,
        epsg=4326,
    )

    # Perform spatial joins for plants in ehv and hv level seperately
    power_plants_hv = gpd.sjoin(
        matched[matched.voltage_level >= 3],
        mv_grid_districts[["bus_id", "geom"]],
        how="left",
    ).drop(columns=["index_right"])
    power_plants_ehv = gpd.sjoin(
        matched[matched.voltage_level < 3],
        ehv_grid_districts[["bus_id", "geom"]],
        how="left",
    ).drop(columns=["index_right"])

    # Combine both dataframes
    power_plants = pd.concat([power_plants_hv, power_plants_ehv])

    # Delete existing units in the target table
    db.execute_sql(
        f""" DELETE FROM {cfg ['sources']['storages']}
        WHERE carrier IN ('pumped_hydro')
        AND scenario='{scn}';"""
    )

    # If export = True export pumped_hydro plants to data base

    if export:
        # Insert into target table
        with session_scope() as session:
            for i, row in power_plants.iterrows():
                entry = EgonStorages(
                    sources={"el_capacity": row.source},
                    source_id={"MastrNummer": row.MaStRNummer},
                    carrier=row.carrier,
                    el_capacity=row.el_capacity,
                    voltage_level=row.voltage_level,
                    bus_id=row.bus_id,
                    scenario=row.scenario,
                    geom=f"SRID=4326;POINT({row.geometry.x} {row.geometry.y})",
                )
                session.add(entry)
            session.commit()

    else:
        return power_plants


def allocate_storage_units_sq(scn_name, storage_types):
    """
    Allocate storage units by mastr data only. Capacities outside
    germany are assigned to foreign buses.

    Parameters
    ----------
    scn_name: str
        Scenario name
    storage_types: list
        contains all the required storage units carriers to be imported

    Returns
    -------

    """
    sources = config.datasets()["power_plants"]["sources"]
    scn_parameters = get_sector_parameters("global", scn_name)
    scenario_date_max = str(scn_parameters["weather_year"]) + "-12-31 23:59:00"

    map_storage = {
        "battery": "Batterie",
        "pumped_hydro": "Pumpspeicher",
        "compressed_air": "Druckluft",
        "flywheel": "Schwungrad",
        "other": "Sonstige",
    }

    for storage_type in storage_types:
        # Read-in data from MaStR
        mastr_ph = pd.read_csv(
            WORKING_DIR_MASTR_NEW / sources["mastr_storage"],
            delimiter=",",
            usecols=[
                "Nettonennleistung",
                "EinheitMastrNummer",
                "Kraftwerksnummer",
                "Technologie",
                "Postleitzahl",
                "Laengengrad",
                "Breitengrad",
                "EinheitBetriebsstatus",
                "LokationMastrNummer",
                "Ort",
                "Bundesland",
                "DatumEndgueltigeStilllegung",
                "Inbetriebnahmedatum",
            ],
            dtype={"Postleitzahl": str},
        )

        # Rename columns
        mastr_ph = mastr_ph.rename(
            columns={
                "Kraftwerksnummer": "bnetza_id",
                "Technologie": "carrier",
                "Postleitzahl": "plz",
                "Ort": "city",
                "Bundesland": "federal_state",
                "Nettonennleistung": "el_capacity",
                "DatumEndgueltigeStilllegung": "decommissioning_date",
            }
        )

        # Select only the required type of storage
        mastr_ph = mastr_ph.loc[mastr_ph.carrier == map_storage[storage_type]]

        # Select only storage units in operation
        mastr_ph.loc[
            mastr_ph["decommissioning_date"] > scenario_date_max,
            "EinheitBetriebsstatus",
        ] = "InBetrieb"
        mastr_ph = mastr_ph.loc[
            mastr_ph.EinheitBetriebsstatus.isin(
                ["InBetrieb", "VoruebergehendStillgelegt"]
            )
        ]

        # Select only storage units installed before scenario_date_max
        mastr_ph = mastr_ph[
            pd.to_datetime(mastr_ph["Inbetriebnahmedatum"]) < scenario_date_max
        ]

        # Calculate power in MW
        mastr_ph.loc[:, "el_capacity"] *= 1e-3

        # Create geodataframe from long, lat
        mastr_ph = gpd.GeoDataFrame(
            mastr_ph,
            geometry=gpd.points_from_xy(
                mastr_ph["Laengengrad"], mastr_ph["Breitengrad"]
            ),
            crs="4326",
        )

        # Identify pp without geocord
        mastr_ph_nogeo = mastr_ph.loc[mastr_ph["Laengengrad"].isna()]

        # Remove all PP without geocord
        mastr_ph = mastr_ph.dropna(subset="Laengengrad")

        # Get geometry of villages/cities with same name of pp with missing geocord
        with session_scope() as session:
            query = session.query(Vg250GemClean.gen, Vg250GemClean.geometry)
            df_cities = gpd.read_postgis(
                query.statement,
                query.session.bind,
                geom_col="geometry",
                crs="3035",
            )

        # Keep only useful cities
        df_cities = df_cities[df_cities["gen"].isin(mastr_ph_nogeo["city"])]

        # Just take the first entry, inaccuracy is negligible as centroid is taken afterwards
        df_cities = df_cities.drop_duplicates("gen", keep="first")

        # Use the centroid instead of polygon of region
        df_cities.loc[:, "geometry"] = df_cities["geometry"].centroid

        # Change coordinate system
        df_cities.to_crs("4326", inplace=True)

        # Add centroid geometry to pp without geometry
        mastr_ph_nogeo = pd.merge(
            left=df_cities,
            right=mastr_ph_nogeo,
            right_on="city",
            left_on="gen",
            suffixes=("", "_no-geo"),
            how="inner",
        ).drop("gen", axis=1)

        mastr_ph = pd.concat([mastr_ph, mastr_ph_nogeo], axis=0)

        # aggregate capacity per location
        agg_cap = mastr_ph.groupby("geometry")["el_capacity"].sum()

        # list mastr number by location
        agg_mastr = mastr_ph.groupby("geometry")["EinheitMastrNummer"].apply(
            list
        )

        # remove duplicates by location
        mastr_ph = mastr_ph.drop_duplicates(
            subset="geometry", keep="first"
        ).drop(["el_capacity", "EinheitMastrNummer"], axis=1)

        # Adjust capacity
        mastr_ph = pd.merge(
            left=mastr_ph,
            right=agg_cap,
            left_on="geometry",
            right_on="geometry",
        )

        # Adjust capacity
        mastr_ph = pd.merge(
            left=mastr_ph,
            right=agg_mastr,
            left_on="geometry",
            right_on="geometry",
        )

        # Drop small pp <= 30 kW
        mastr_ph = mastr_ph.loc[mastr_ph["el_capacity"] > 0.03]

        # Apply voltage level by capacity
        mastr_ph = apply_voltage_level_thresholds(mastr_ph)
        mastr_ph["voltage_level"] = mastr_ph["voltage_level"].astype(int)

        # Capacity located outside germany -> will be assigned to foreign buses
        mastr_ph_foreign = mastr_ph.loc[mastr_ph["federal_state"].isna()]

        # Keep only capacities within germany
        mastr_ph = mastr_ph.dropna(subset="federal_state")

        # Asign buses within germany
        mastr_ph = assign_bus_id(
            mastr_ph, cfg=config.datasets()["power_plants"], drop_missing=True
        )
        mastr_ph["bus_id"] = mastr_ph["bus_id"].astype(int)

        # Get foreign central buses
        sql = f"""
        SELECT * FROM grid.egon_etrago_bus
        WHERE scn_name = '{scn_name}'
        and country != 'DE'
        """
        df_foreign_buses = db.select_geodataframe(
            sql, geom_col="geom", epsg="4326"
        )
        central_bus = entsoe_to_bus_etrago(scn_name).to_frame()
        central_bus["geom"] = (
            df_foreign_buses.set_index("bus_id")
            .loc[central_bus[0], "geom"]
            .values
        )
        df_foreign_buses = df_foreign_buses[
            df_foreign_buses["geom"].isin(central_bus["geom"])
        ]

        if len(mastr_ph_foreign) > 0:
            # Assign closest bus at voltage level to foreign pp
            nearest_neighbors = []
            for vl, v_nom in {1: 380, 2: 220, 3: 110}.items():
                ph = mastr_ph_foreign.loc[
                    mastr_ph_foreign["voltage_level"] == vl
                ]
                if ph.empty:
                    continue
                bus = df_foreign_buses.loc[
                    df_foreign_buses["v_nom"] == v_nom,
                    ["v_nom", "country", "bus_id", "geom"],
                ]
                results = gpd.sjoin_nearest(
                    left_df=ph,
                    right_df=bus,
                    how="left",
                    distance_col="distance",
                )
                nearest_neighbors.append(results)
            mastr_ph_foreign = pd.concat(nearest_neighbors)

            # Merge foreign pp
            mastr_ph = pd.concat([mastr_ph, mastr_ph_foreign])

        # Reduce to necessary columns
        mastr_ph = mastr_ph[
            [
                "el_capacity",
                "voltage_level",
                "bus_id",
                "geometry",
                "EinheitMastrNummer",
            ]
        ]

        # Rename and format columns
        mastr_ph["carrier"] = storage_type
        mastr_ph = mastr_ph.rename(
            columns={"EinheitMastrNummer": "source_id", "geometry": "geom"}
        )
        mastr_ph["source_id"] = mastr_ph["source_id"].apply(
            lambda x: {"MastrNummer": ", ".join(x)}
        )
        mastr_ph = mastr_ph.set_geometry("geom")
        mastr_ph["geom"] = mastr_ph["geom"].apply(lambda x: x.wkb_hex)
        mastr_ph["scenario"] = scn_name
        mastr_ph["sources"] = [
            {"el_capacity": "MaStR aggregated by location"}
        ] * mastr_ph.shape[0]

        # Delete existing units in the target table
        db.execute_sql(
            f""" DELETE FROM supply.egon_storages
            WHERE carrier = '{storage_type}'
            AND scenario = '{scn_name}'
            AND sources ->> 'el_capacity' = 'MaStR aggregated by location';"""
        )

        with db.session_scope() as session:
            session.bulk_insert_mappings(
                EgonStorages,
                mastr_ph.to_dict(orient="records"),
            )


def allocate_pumped_hydro_eGon100RE():
    """Allocates pumped_hydro plants for eGon100RE scenario based on a
    prox-to-now method applied on allocated pumped-hydro plants in the eGon2035
    scenario.

    Parameters
    ----------
    None

    Returns
    -------
    None
    """

    carrier = "pumped_hydro"
    cfg = config.datasets()["power_plants"]
    boundary = config.settings()["egon-data"]["--dataset-boundary"]

    # Select installed capacity for pumped_hydro in eGon100RE scenario from
    # scenario capacities table
    capacity = db.select_dataframe(
        f"""
        SELECT capacity
        FROM {cfg['sources']['capacities']}
        WHERE carrier = '{carrier}'
        AND scenario_name = 'eGon100RE';
        """
    )

    if boundary == "Schleswig-Holstein":
        # Break capacity of pumped hydron plants down SH share in eGon2035
        capacity_phes = capacity.iat[0, 0] * 0.0176

    elif boundary == "Everything":
        # Select national capacity for pumped hydro
        capacity_phes = capacity.iat[0, 0]

    else:
        raise ValueError(f"'{boundary}' is not a valid dataset boundary.")

    # Get allocation of pumped_hydro plants in eGon2035 scenario as the
    # reference for the distribution in eGon100RE scenario
    allocation = allocate_pumped_hydro(scn="status2019", export=False)

    scaling_factor = capacity_phes / allocation.el_capacity.sum()

    power_plants = allocation.copy()
    power_plants["scenario"] = "eGon100RE"
    power_plants["el_capacity"] = allocation.el_capacity * scaling_factor

    # Insert into target table
    session = sessionmaker(bind=db.engine())()
    for i, row in power_plants.iterrows():
        entry = EgonStorages(
            sources={"el_capacity": row.source},
            source_id={"MastrNummer": row.MaStRNummer},
            carrier=row.carrier,
            el_capacity=row.el_capacity,
            voltage_level=row.voltage_level,
            bus_id=row.bus_id,
            scenario=row.scenario,
            geom=f"SRID=4326;POINT({row.geometry.x} {row.geometry.y})",
        )
        session.add(entry)
    session.commit()


def home_batteries_per_scenario(scenario):
    """Allocates home batteries which define a lower boundary for extendable
    battery storage units. The overall installed capacity is taken from NEP
    for eGon2035 scenario. The spatial distribution of installed battery
    capacities is based on the installed pv rooftop capacity.

    Parameters
    ----------
    None

    Returns
    -------
    None
    """

    cfg = config.datasets()["storages"]
    dataset = config.settings()["egon-data"]["--dataset-boundary"]

    if scenario == "eGon2035":
        target_file = (
            Path(".")
            / "data_bundle_egon_data"
            / "nep2035_version2021"
            / cfg["sources"]["nep_capacities"]
        )

        capacities_nep = pd.read_excel(
            target_file,
            sheet_name="1.Entwurf_NEP2035_V2021",
            index_col="Unnamed: 0",
        )

        # Select target value in MW
        target = capacities_nep.Summe["PV-Batteriespeicher"] * 1000

    else:
        target = db.select_dataframe(
            f"""
            SELECT capacity
            FROM {cfg['sources']['capacities']}
            WHERE scenario_name = '{scenario}'
            AND carrier = 'battery';
            """
        ).capacity[0]

    pv_rooftop = db.select_dataframe(
        f"""
        SELECT bus, p_nom, generator_id
        FROM {cfg['sources']['generators']}
        WHERE scn_name = '{scenario}'
        AND carrier = 'solar_rooftop'
        AND bus IN
            (SELECT bus_id FROM {cfg['sources']['bus']}
               WHERE scn_name = '{scenario}' AND country = 'DE' );
        """
    )

    if dataset == "Schleswig-Holstein":
        target = target / 16

    battery = pv_rooftop
    battery["p_nom_min"] = target * battery["p_nom"] / battery["p_nom"].sum()
    battery = battery.drop(columns=["p_nom"])

    battery["carrier"] = "home_battery"
    battery["scenario"] = scenario

    if (scenario == "eGon2035") | ("status" in scenario):
        source = "NEP"

    else:
        source = "p-e-s"

    battery["source"] = (
        f"{source} capacity allocated based in installed PV rooftop capacity"
    )

    # Insert into target table
    session = sessionmaker(bind=db.engine())()
    for i, row in battery.iterrows():
        entry = EgonStorages(
            sources={"el_capacity": row.source},
            source_id={"generator_id": row.generator_id},
            carrier=row.carrier,
            el_capacity=row.p_nom_min,
            bus_id=row.bus,
            scenario=row.scenario,
        )
        session.add(entry)
    session.commit()


def allocate_pv_home_batteries_to_grids():
    for scn in config.settings()["egon-data"]["--scenarios"]:
        home_batteries_per_scenario(scn)


def allocate_pumped_hydro_scn():
    for scn in config.settings()["egon-data"]["--scenarios"]:
        if scn == "eGon2035":
            allocate_pumped_hydro(scn="eGon2035")
        elif scn == "eGon100RE":
            allocate_pumped_hydro_eGon100RE()
        elif "status" in scn:
            allocate_storage_units_sq(scn_name=scn, storage_types=["pumped_hydro"])


def allocate_other_storage_units():
    for scn in config.settings()["egon-data"]["--scenarios"]:
        if "status" in scn:
            allocate_storage_units_sq(
                scn_name=scn, storage_types=["battery"]
            )


1		"""The central module containing all code dealing with power plant data.
2		"""
3
4		from pathlib import Path
5
6		from geoalchemy2 import Geometry
7		from sqlalchemy import BigInteger, Column, Float, Integer, Sequence, String
8		from sqlalchemy.dialects.postgresql import JSONB
9		from sqlalchemy.ext.declarative import declarative_base
10		from sqlalchemy.orm import sessionmaker
11		import geopandas as gpd
12		import pandas as pd
13
14		from egon.data import config, db
15		from egon.data.datasets import Dataset
16		from egon.data.datasets.scenario_parameters import get_sector_parameters
17		from egon.data.datasets.electrical_neighbours import entsoe_to_bus_etrago
18		from egon.data.datasets.mastr import (
19		WORKING_DIR_MASTR_NEW,
20		WORKING_DIR_MASTR_OLD,
21		)
22		from egon.data.datasets.mv_grid_districts import Vg250GemClean
23		from egon.data.datasets.power_plants import assign_bus_id, assign_voltage_level
24		from egon.data.datasets.storages.home_batteries import (
25		allocate_home_batteries_to_buildings,
26		)
27		from egon.data.datasets.storages.pumped_hydro import (
28		apply_voltage_level_thresholds,
29		get_location,
30		match_storage_units,
31		select_mastr_pumped_hydro,
32		select_nep_pumped_hydro,
33		)
34		from egon.data.db import session_scope
35
36		Base = declarative_base()
37
38
39	View Code Duplication	class EgonStorages(Base):
		0 ignored issues – show Duplication introduced 2021-11-25 08:07 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
40		__tablename__ = "egon_storages"
41		__table_args__ = {"schema": "supply"}
42		id = Column(BigInteger, Sequence("storage_seq"), primary_key=True)
43		sources = Column(JSONB)
44		source_id = Column(JSONB)
45		carrier = Column(String)
46		el_capacity = Column(Float)
47		bus_id = Column(Integer)
48		voltage_level = Column(Integer)
49		scenario = Column(String)
50		geom = Column(Geometry("POINT", 4326))
51
52
53		class Storages(Dataset):
54		def __init__(self, dependencies):
55		super().__init__(
56		name="Storages",
57		version="0.0.8",
58		dependencies=dependencies,
59		tasks=(
60		create_tables,
61		allocate_pumped_hydro_scn,
62		allocate_other_storage_units,
63		# allocate_pv_home_batteries_to_grids,
64		# allocate_home_batteries_to_buildings,
65		),
66		)
67
68
69		def create_tables():
70		"""Create tables for power plant data
71		Returns
72		-------
73		None.
74		"""
75
76		cfg = config.datasets()["storages"]
77		db.execute_sql(f"CREATE SCHEMA IF NOT EXISTS {cfg['target']['schema']};")
78		engine = db.engine()
79		db.execute_sql(
80		f"""DROP TABLE IF EXISTS
81		{cfg['target']['schema']}.{cfg['target']['table']}"""
82		)
83
84		db.execute_sql("""DROP SEQUENCE IF EXISTS pp_seq""")
85		EgonStorages.__table__.create(bind=engine, checkfirst=True)
86
87
88		def allocate_pumped_hydro(scn, export=True):
89		"""Allocates pumped_hydro plants for eGon2035 and scenario2019 scenarios
90		and either exports results to data base or returns as a dataframe
91
92		Parameters
93		----------
94		export : bool
95		Choose if allocated pumped hydro plants should be exported to the data
96		base. The default is True.
97		If export=False a data frame will be returned
98
99		Returns
100		-------
101		power_plants : pandas.DataFrame
102		List of pumped hydro plants in 'eGon2035' and 'scenario2019' scenarios
103		"""
104
105		carrier = "pumped_hydro"
106
107		cfg = config.datasets()["power_plants"]
108
109		nep = select_nep_pumped_hydro(scn=scn)
110		mastr = select_mastr_pumped_hydro()
111
112		# Assign voltage level to MaStR
113		mastr["voltage_level"] = assign_voltage_level(
114		mastr.rename({"el_capacity": "Nettonennleistung"}, axis=1),
115		cfg,
116		WORKING_DIR_MASTR_OLD,
117		)
118
119		# Initalize DataFrame for matching power plants
120		matched = gpd.GeoDataFrame(
121		columns=[
122		"carrier",
123		"el_capacity",
124		"scenario",
125		"geometry",
126		"MaStRNummer",
127		"source",
128		"voltage_level",
129		]
130		)
131
132		# Match pumped_hydro units from NEP list
133		# using PLZ and capacity
134		matched, mastr, nep = match_storage_units(
135		nep,
136		mastr,
137		matched,
138		buffer_capacity=0.1,
139		consider_carrier=False,
140		scn=scn,
141		)
142
143		# Match plants from NEP list using plz,
144		# neglecting the capacity
145		matched, mastr, nep = match_storage_units(
146		nep,
147		mastr,
148		matched,
149		consider_location="plz",
150		consider_carrier=False,
151		consider_capacity=False,
152		scn=scn,
153		)
154
155		# Match plants from NEP list using city,
156		# neglecting the capacity
157		matched, mastr, nep = match_storage_units(
158		nep,
159		mastr,
160		matched,
161		consider_location="city",
162		consider_carrier=False,
163		consider_capacity=False,
164		scn=scn,
165		)
166
167		# Match remaining plants from NEP using the federal state
168		matched, mastr, nep = match_storage_units(
169		nep,
170		mastr,
171		matched,
172		buffer_capacity=0.1,
173		consider_location="federal_state",
174		consider_carrier=False,
175		scn=scn,
176		)
177
178		# Match remaining plants from NEP using the federal state
179		matched, mastr, nep = match_storage_units(
180		nep,
181		mastr,
182		matched,
183		buffer_capacity=0.7,
184		consider_location="federal_state",
185		consider_carrier=False,
186		scn=scn,
187		)
188
189		print(f"{matched.el_capacity.sum()} MW of {carrier} matched")
190		print(f"{nep.elec_capacity.sum()} MW of {carrier} not matched")
191
192		if nep.elec_capacity.sum() > 0:
193		# Get location using geolocator and city information
194		located, unmatched = get_location(nep)
195
196		# Bring both dataframes together
197		matched = pd.concat(
198		[
199		matched,
200		located[
201		[
202		"carrier",
203		"el_capacity",
204		"scenario",
205		"geometry",
206		"source",
207		"MaStRNummer",
208		]
209		],
210		],
211		ignore_index=True,
212		)
213
214		# Set CRS
215		matched.crs = "EPSG:4326"
216
217		# Assign voltage level
218		matched = apply_voltage_level_thresholds(matched)
219
220		# Assign bus_id
221		# Load grid district polygons
222		mv_grid_districts = db.select_geodataframe(
223		f"""
224		SELECT * FROM {cfg['sources']['egon_mv_grid_district']}
225		""",
226		epsg=4326,
227		)
228
229		ehv_grid_districts = db.select_geodataframe(
230		f"""
231		SELECT * FROM {cfg['sources']['ehv_voronoi']}
232		""",
233		epsg=4326,
234		)
235
236		# Perform spatial joins for plants in ehv and hv level seperately
237		power_plants_hv = gpd.sjoin(
238		matched[matched.voltage_level >= 3],
239		mv_grid_districts[["bus_id", "geom"]],
240		how="left",
241		).drop(columns=["index_right"])
242		power_plants_ehv = gpd.sjoin(
243		matched[matched.voltage_level < 3],
244		ehv_grid_districts[["bus_id", "geom"]],
245		how="left",
246		).drop(columns=["index_right"])
247
248		# Combine both dataframes
249		power_plants = pd.concat([power_plants_hv, power_plants_ehv])
250
251		# Delete existing units in the target table
252		db.execute_sql(
253		f""" DELETE FROM {cfg ['sources']['storages']}
254		WHERE carrier IN ('pumped_hydro')
255		AND scenario='{scn}';"""
256		)
257
258		# If export = True export pumped_hydro plants to data base
259
260		if export:
261		# Insert into target table
262		with session_scope() as session:
263		for i, row in power_plants.iterrows():
264		entry = EgonStorages(
265		sources={"el_capacity": row.source},
266		source_id={"MastrNummer": row.MaStRNummer},
267		carrier=row.carrier,
268		el_capacity=row.el_capacity,
269		voltage_level=row.voltage_level,
270		bus_id=row.bus_id,
271		scenario=row.scenario,
272		geom=f"SRID=4326;POINT({row.geometry.x} {row.geometry.y})",
273		)
274		session.add(entry)
275		session.commit()
276
277		else:
278		return power_plants
279
280
281		def allocate_storage_units_sq(scn_name, storage_types):
282		"""
283		Allocate storage units by mastr data only. Capacities outside
284		germany are assigned to foreign buses.
285
286		Parameters
287		----------
288		scn_name: str
289		Scenario name
290		storage_types: list
291		contains all the required storage units carriers to be imported
292
293		Returns
294		-------
295
296		"""
297		sources = config.datasets()["power_plants"]["sources"]
298		scn_parameters = get_sector_parameters("global", scn_name)
299		scenario_date_max = str(scn_parameters["weather_year"]) + "-12-31 23:59:00"
300
301		map_storage = {
302		"battery": "Batterie",
303		"pumped_hydro": "Pumpspeicher",
304		"compressed_air": "Druckluft",
305		"flywheel": "Schwungrad",
306		"other": "Sonstige",
307		}
308
309		for storage_type in storage_types:
310		# Read-in data from MaStR
311		mastr_ph = pd.read_csv(
312		WORKING_DIR_MASTR_NEW / sources["mastr_storage"],
313		delimiter=",",
314		usecols=[
315		"Nettonennleistung",
316		"EinheitMastrNummer",
317		"Kraftwerksnummer",
318		"Technologie",
319		"Postleitzahl",
320		"Laengengrad",
321		"Breitengrad",
322		"EinheitBetriebsstatus",
323		"LokationMastrNummer",
324		"Ort",
325		"Bundesland",
326		"DatumEndgueltigeStilllegung",
327		"Inbetriebnahmedatum",
328		],
329		dtype={"Postleitzahl": str},
330		)
331
332		# Rename columns
333		mastr_ph = mastr_ph.rename(
334		columns={
335		"Kraftwerksnummer": "bnetza_id",
336		"Technologie": "carrier",
337		"Postleitzahl": "plz",
338		"Ort": "city",
339		"Bundesland": "federal_state",
340		"Nettonennleistung": "el_capacity",
341		"DatumEndgueltigeStilllegung": "decommissioning_date",
342		}
343		)
344
345		# Select only the required type of storage
346		mastr_ph = mastr_ph.loc[mastr_ph.carrier == map_storage[storage_type]]
347
348		# Select only storage units in operation
349		mastr_ph.loc[
350		mastr_ph["decommissioning_date"] > scenario_date_max,
351		"EinheitBetriebsstatus",
352		] = "InBetrieb"
353		mastr_ph = mastr_ph.loc[
354		mastr_ph.EinheitBetriebsstatus.isin(
355		["InBetrieb", "VoruebergehendStillgelegt"]
356		)
357		]
358
359		# Select only storage units installed before scenario_date_max
360		mastr_ph = mastr_ph[
361		pd.to_datetime(mastr_ph["Inbetriebnahmedatum"]) < scenario_date_max
362		]
363
364		# Calculate power in MW
365		mastr_ph.loc[:, "el_capacity"] *= 1e-3
366
367		# Create geodataframe from long, lat
368		mastr_ph = gpd.GeoDataFrame(
369		mastr_ph,
370		geometry=gpd.points_from_xy(
371		mastr_ph["Laengengrad"], mastr_ph["Breitengrad"]
372		),
373		crs="4326",
374		)
375
376		# Identify pp without geocord
377		mastr_ph_nogeo = mastr_ph.loc[mastr_ph["Laengengrad"].isna()]
378
379		# Remove all PP without geocord
380		mastr_ph = mastr_ph.dropna(subset="Laengengrad")
381
382		# Get geometry of villages/cities with same name of pp with missing geocord
383		with session_scope() as session:
384		query = session.query(Vg250GemClean.gen, Vg250GemClean.geometry)
385		df_cities = gpd.read_postgis(
386		query.statement,
387		query.session.bind,
388		geom_col="geometry",
389		crs="3035",
390		)
391
392		# Keep only useful cities
393		df_cities = df_cities[df_cities["gen"].isin(mastr_ph_nogeo["city"])]
394
395		# Just take the first entry, inaccuracy is negligible as centroid is taken afterwards
396		df_cities = df_cities.drop_duplicates("gen", keep="first")
397
398		# Use the centroid instead of polygon of region
399		df_cities.loc[:, "geometry"] = df_cities["geometry"].centroid
400
401		# Change coordinate system
402		df_cities.to_crs("4326", inplace=True)
403
404		# Add centroid geometry to pp without geometry
405		mastr_ph_nogeo = pd.merge(
406		left=df_cities,
407		right=mastr_ph_nogeo,
408		right_on="city",
409		left_on="gen",
410		suffixes=("", "_no-geo"),
411		how="inner",
412		).drop("gen", axis=1)
413
414		mastr_ph = pd.concat([mastr_ph, mastr_ph_nogeo], axis=0)
415
416		# aggregate capacity per location
417		agg_cap = mastr_ph.groupby("geometry")["el_capacity"].sum()
418
419		# list mastr number by location
420		agg_mastr = mastr_ph.groupby("geometry")["EinheitMastrNummer"].apply(
421		list
422		)
423
424		# remove duplicates by location
425		mastr_ph = mastr_ph.drop_duplicates(
426		subset="geometry", keep="first"
427		).drop(["el_capacity", "EinheitMastrNummer"], axis=1)
428
429		# Adjust capacity
430		mastr_ph = pd.merge(
431		left=mastr_ph,
432		right=agg_cap,
433		left_on="geometry",
434		right_on="geometry",
435		)
436
437		# Adjust capacity
438		mastr_ph = pd.merge(
439		left=mastr_ph,
440		right=agg_mastr,
441		left_on="geometry",
442		right_on="geometry",
443		)
444
445		# Drop small pp <= 30 kW
446		mastr_ph = mastr_ph.loc[mastr_ph["el_capacity"] > 0.03]
447
448		# Apply voltage level by capacity
449		mastr_ph = apply_voltage_level_thresholds(mastr_ph)
450		mastr_ph["voltage_level"] = mastr_ph["voltage_level"].astype(int)
451
452		# Capacity located outside germany -> will be assigned to foreign buses
453		mastr_ph_foreign = mastr_ph.loc[mastr_ph["federal_state"].isna()]
454
455		# Keep only capacities within germany
456		mastr_ph = mastr_ph.dropna(subset="federal_state")
457
458		# Asign buses within germany
459		mastr_ph = assign_bus_id(
460		mastr_ph, cfg=config.datasets()["power_plants"], drop_missing=True
461		)
462		mastr_ph["bus_id"] = mastr_ph["bus_id"].astype(int)
463
464		# Get foreign central buses
465		sql = f"""
466		SELECT * FROM grid.egon_etrago_bus
467		WHERE scn_name = '{scn_name}'
468		and country != 'DE'
469		"""
470		df_foreign_buses = db.select_geodataframe(
471		sql, geom_col="geom", epsg="4326"
472		)
473		central_bus = entsoe_to_bus_etrago(scn_name).to_frame()
474		central_bus["geom"] = (
475		df_foreign_buses.set_index("bus_id")
476		.loc[central_bus[0], "geom"]
477		.values
478		)
479		df_foreign_buses = df_foreign_buses[
480		df_foreign_buses["geom"].isin(central_bus["geom"])
481		]
482
483		if len(mastr_ph_foreign) > 0:
484		# Assign closest bus at voltage level to foreign pp
485		nearest_neighbors = []
486		for vl, v_nom in {1: 380, 2: 220, 3: 110}.items():
487		ph = mastr_ph_foreign.loc[
488		mastr_ph_foreign["voltage_level"] == vl
489		]
490		if ph.empty:
491		continue
492		bus = df_foreign_buses.loc[
493		df_foreign_buses["v_nom"] == v_nom,
494		["v_nom", "country", "bus_id", "geom"],
495		]
496		results = gpd.sjoin_nearest(
497		left_df=ph,
498		right_df=bus,
499		how="left",
500		distance_col="distance",
501		)
502		nearest_neighbors.append(results)
503		mastr_ph_foreign = pd.concat(nearest_neighbors)
504
505		# Merge foreign pp
506		mastr_ph = pd.concat([mastr_ph, mastr_ph_foreign])
507
508		# Reduce to necessary columns
509		mastr_ph = mastr_ph[
510		[
511		"el_capacity",
512		"voltage_level",
513		"bus_id",
514		"geometry",
515		"EinheitMastrNummer",
516		]
517		]
518
519		# Rename and format columns
520		mastr_ph["carrier"] = storage_type
521		mastr_ph = mastr_ph.rename(
522		columns={"EinheitMastrNummer": "source_id", "geometry": "geom"}
523		)
524		mastr_ph["source_id"] = mastr_ph["source_id"].apply(
525		lambda x: {"MastrNummer": ", ".join(x)}
526		)
527		mastr_ph = mastr_ph.set_geometry("geom")
528		mastr_ph["geom"] = mastr_ph["geom"].apply(lambda x: x.wkb_hex)
529		mastr_ph["scenario"] = scn_name
530		mastr_ph["sources"] = [
531		{"el_capacity": "MaStR aggregated by location"}
532		] * mastr_ph.shape[0]
533
534		# Delete existing units in the target table
535		db.execute_sql(
536		f""" DELETE FROM supply.egon_storages
537		WHERE carrier = '{storage_type}'
538		AND scenario = '{scn_name}'
539		AND sources ->> 'el_capacity' = 'MaStR aggregated by location';"""
540		)
541
542		with db.session_scope() as session:
543		session.bulk_insert_mappings(
544		EgonStorages,
545		mastr_ph.to_dict(orient="records"),
546		)
547
548
549		def allocate_pumped_hydro_eGon100RE():
550		"""Allocates pumped_hydro plants for eGon100RE scenario based on a
551		prox-to-now method applied on allocated pumped-hydro plants in the eGon2035
552		scenario.
553
554		Parameters
555		----------
556		None
557
558		Returns
559		-------
560		None
561		"""
562
563		carrier = "pumped_hydro"
564		cfg = config.datasets()["power_plants"]
565		boundary = config.settings()["egon-data"]["--dataset-boundary"]
566
567		# Select installed capacity for pumped_hydro in eGon100RE scenario from
568		# scenario capacities table
569		capacity = db.select_dataframe(
570		f"""
571		SELECT capacity
572		FROM {cfg['sources']['capacities']}
573		WHERE carrier = '{carrier}'
574		AND scenario_name = 'eGon100RE';
575		"""
576		)
577
578		if boundary == "Schleswig-Holstein":
579		# Break capacity of pumped hydron plants down SH share in eGon2035
580		capacity_phes = capacity.iat[0, 0] * 0.0176
581
582		elif boundary == "Everything":
583		# Select national capacity for pumped hydro
584		capacity_phes = capacity.iat[0, 0]
585
586		else:
587		raise ValueError(f"'{boundary}' is not a valid dataset boundary.")
588
589		# Get allocation of pumped_hydro plants in eGon2035 scenario as the
590		# reference for the distribution in eGon100RE scenario
591		allocation = allocate_pumped_hydro(scn="status2019", export=False)
592
593		scaling_factor = capacity_phes / allocation.el_capacity.sum()
594
595		power_plants = allocation.copy()
596		power_plants["scenario"] = "eGon100RE"
597		power_plants["el_capacity"] = allocation.el_capacity * scaling_factor
598
599		# Insert into target table
600		session = sessionmaker(bind=db.engine())()
601		for i, row in power_plants.iterrows():
602		entry = EgonStorages(
603		sources={"el_capacity": row.source},
604		source_id={"MastrNummer": row.MaStRNummer},
605		carrier=row.carrier,
606		el_capacity=row.el_capacity,
607		voltage_level=row.voltage_level,
608		bus_id=row.bus_id,
609		scenario=row.scenario,
610		geom=f"SRID=4326;POINT({row.geometry.x} {row.geometry.y})",
611		)
612		session.add(entry)
613		session.commit()
614
615
616		def home_batteries_per_scenario(scenario):
617		"""Allocates home batteries which define a lower boundary for extendable
618		battery storage units. The overall installed capacity is taken from NEP
619		for eGon2035 scenario. The spatial distribution of installed battery
620		capacities is based on the installed pv rooftop capacity.
621
622		Parameters
623		----------
624		None
625
626		Returns
627		-------
628		None
629		"""
630
631		cfg = config.datasets()["storages"]
632		dataset = config.settings()["egon-data"]["--dataset-boundary"]
633
634		if scenario == "eGon2035":
635		target_file = (
636		Path(".")
637		/ "data_bundle_egon_data"
638		/ "nep2035_version2021"
639		/ cfg["sources"]["nep_capacities"]
640		)
641
642		capacities_nep = pd.read_excel(
643		target_file,
644		sheet_name="1.Entwurf_NEP2035_V2021",
645		index_col="Unnamed: 0",
646		)
647
648		# Select target value in MW
649		target = capacities_nep.Summe["PV-Batteriespeicher"] * 1000
650
651		else:
652		target = db.select_dataframe(
653		f"""
654		SELECT capacity
655		FROM {cfg['sources']['capacities']}
656		WHERE scenario_name = '{scenario}'
657		AND carrier = 'battery';
658		"""
659		).capacity[0]
660
661		pv_rooftop = db.select_dataframe(
662		f"""
663		SELECT bus, p_nom, generator_id
664		FROM {cfg['sources']['generators']}
665		WHERE scn_name = '{scenario}'
666		AND carrier = 'solar_rooftop'
667		AND bus IN
668		(SELECT bus_id FROM {cfg['sources']['bus']}
669		WHERE scn_name = '{scenario}' AND country = 'DE' );
670		"""
671		)
672
673		if dataset == "Schleswig-Holstein":
674		target = target / 16
675
676		battery = pv_rooftop
677		battery["p_nom_min"] = target * battery["p_nom"] / battery["p_nom"].sum()
678		battery = battery.drop(columns=["p_nom"])
679
680		battery["carrier"] = "home_battery"
681		battery["scenario"] = scenario
682
683		if (scenario == "eGon2035") \| ("status" in scenario):
684		source = "NEP"
685
686		else:
687		source = "p-e-s"
688
689		battery["source"] = (
690		f"{source} capacity allocated based in installed PV rooftop capacity"
691		)
692
693		# Insert into target table
694		session = sessionmaker(bind=db.engine())()
695		for i, row in battery.iterrows():
696		entry = EgonStorages(
697		sources={"el_capacity": row.source},
698		source_id={"generator_id": row.generator_id},
699		carrier=row.carrier,
700		el_capacity=row.p_nom_min,
701		bus_id=row.bus,
702		scenario=row.scenario,
703		)
704		session.add(entry)
705		session.commit()
706
707
708		def allocate_pv_home_batteries_to_grids():
709		for scn in config.settings()["egon-data"]["--scenarios"]:
710		home_batteries_per_scenario(scn)
711
712
713		def allocate_pumped_hydro_scn():
714		for scn in config.settings()["egon-data"]["--scenarios"]:
715		if scn == "eGon2035":
716		allocate_pumped_hydro(scn="eGon2035")
717		elif scn == "eGon100RE":
718		allocate_pumped_hydro_eGon100RE()
719		elif "status" in scn:
720		allocate_storage_units_sq(scn_name=scn, storage_types=["pumped_hydro"])
721
722
723		def allocate_other_storage_units():
724		for scn in config.settings()["egon-data"]["--scenarios"]:
725		if "status" in scn:
726		allocate_storage_units_sq(
727		scn_name=scn, storage_types=["battery"]
728		)
729

openego / eGon-data

Pull Request — dev (#1181)

data.datasets.storages.allocate_storage_units_sq() D

Complexity

Size

Duplication

Importance

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like