data.datasets.power_plants.assign_weather_data.assign_bus_id() - Code Metrics - Inspection of "Merge pull request #1273 from openego/fixes/#1256-..." - openego/eGon-data - Measure and Improve Code Quality continuously with Scrutinizer

Completed

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

unknown

created 2025-06-12 13:09 UTC

assign_bus_id() A

↳ Parent: data.datasets.power_plants.assign_weather_data

Complexity

Conditions

Size

Total Lines	63
Code Lines	29

Duplication

Lines	63
Ratio	100 %

Importance

Changes

Metric	Value
eloc	29
dl	63
loc	63
rs	9.184
c	0
b	0
f	0
cc	4
nop	2

How to fix Long Method

import geopandas as gpd

from egon.data import db
from egon.data.datasets.power_plants.pv_rooftop_buildings import timer_func
import egon.data.config

def assign_bus_id(power_plants, cfg):

    """Assigns bus_ids to power plants according to location and voltage level

    Parameters
    ----------
    power_plants : pandas.DataFrame
        Power plants including voltage level

    Returns
    -------
    power_plants : pandas.DataFrame
        Power plants including voltage level and bus_id

    """

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

    # Assign power plants in hv and below to hvmv bus
    power_plants_hv = power_plants[power_plants.voltage_level >= 3].index
    if len(power_plants_hv) > 0:
        power_plants.loc[power_plants_hv, "bus_id"] = gpd.sjoin(
            power_plants[power_plants.index.isin(power_plants_hv)],
            mv_grid_districts,
        ).bus_id

    # Assign power plants in ehv to ehv bus
    power_plants_ehv = power_plants[power_plants.voltage_level < 3].index

    if len(power_plants_ehv) > 0:
        ehv_join = gpd.sjoin(
            power_plants[power_plants.index.isin(power_plants_ehv)],
            ehv_grid_districts,
        )

        if "bus_id_right" in ehv_join.columns:
            power_plants.loc[power_plants_ehv, "bus_id"] = gpd.sjoin(
                power_plants[power_plants.index.isin(power_plants_ehv)],
                ehv_grid_districts,
            ).bus_id_right

        else:
            power_plants.loc[power_plants_ehv, "bus_id"] = gpd.sjoin(
                power_plants[power_plants.index.isin(power_plants_ehv)],
                ehv_grid_districts,
            ).bus_id

    # Assert that all power plants have a bus_id
    assert power_plants.bus_id.notnull().all(), f"""Some power plants are
    not attached to a bus: {power_plants[power_plants.bus_id.isnull()]}"""

    return power_plants


@timer_func
def add_missing_bus_ids(scn_name):
    """Assign busses by spatal intersection of mvgrid districts or ehv voronois."""

    sql = f"""
                -- Assign missing buses to mv grid district buses for HV and below
                UPDATE supply.egon_power_plants AS epp
                SET bus_id = (
                    SELECT emgd.bus_id
                    FROM grid.egon_mv_grid_district AS emgd
                    WHERE ST_Intersects(ST_Transform(epp.geom, 4326), ST_Transform(emgd.geom, 4326))
                    ORDER BY ST_Transform(emgd.geom, 4326) <-> ST_Transform(epp.geom, 4326)
                    LIMIT 1
                )
                WHERE (epp.carrier = 'solar'
                    OR epp.carrier = 'wind_onshore'
                    OR epp.carrier = 'solar_rooftop'
                    OR epp.carrier = 'wind_offshore')
                AND epp.scenario = '{scn_name}'
                AND epp.bus_id is null
                AND epp.voltage_level >= 3; -- HV and below


                -- Assign missing buses to EHV buses for EHV
                UPDATE supply.egon_power_plants AS epp
                SET bus_id = (
                    SELECT eesv.bus_id
                    FROM grid.egon_ehv_substation_voronoi AS eesv
                    WHERE ST_Intersects(ST_Transform(epp.geom, 4326), ST_Transform(eesv.geom, 4326))
                    ORDER BY ST_Transform(eesv.geom, 4326) <-> ST_Transform(epp.geom, 4326)
                    LIMIT 1
                )
                WHERE (epp.carrier = 'solar'
                    OR epp.carrier = 'wind_onshore'
                    OR epp.carrier = 'solar_rooftop'
                    OR epp.carrier = 'wind_offshore')
                AND epp.scenario = '{scn_name}'
                AND epp.bus_id is null
                AND epp.voltage_level < 3; --EHV


        """

    db.execute_sql(sql)


@timer_func
def find_weather_id(scn_name):

    sql = f"""UPDATE supply.egon_power_plants AS epp
                SET weather_cell_id = (
                    SELECT eewc.w_id
                    FROM supply.egon_era5_weather_cells AS eewc
                    WHERE ST_Intersects(epp.geom, eewc.geom)
                    ORDER BY eewc.geom <-> epp.geom
                    LIMIT 1
                )
                WHERE (epp.carrier = 'solar'
                    OR epp.carrier = 'solar_rooftop'
                    OR epp.carrier = 'wind_onshore'
                    OR epp.carrier = 'wind_offshore')
                AND epp.scenario = '{scn_name}';
        """

    db.execute_sql(sql)


def weatherId_and_busId(scn_name):
    find_weather_id(scn_name)
    add_missing_bus_ids(scn_name)


1		import geopandas as gpd
2
3		from egon.data import db
4		from egon.data.datasets.power_plants.pv_rooftop_buildings import timer_func
5		import egon.data.config
6
7	View Code Duplication	def assign_bus_id(power_plants, cfg):
		0 ignored issues – show Duplication introduced 2024-11-08 07:56 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
8		"""Assigns bus_ids to power plants according to location and voltage level
9
10		Parameters
11		----------
12		power_plants : pandas.DataFrame
13		Power plants including voltage level
14
15		Returns
16		-------
17		power_plants : pandas.DataFrame
18		Power plants including voltage level and bus_id
19
20		"""
21
22		mv_grid_districts = db.select_geodataframe(
23		f"""
24		SELECT * FROM {cfg['sources']['egon_mv_grid_district']}
25		""",
26		epsg=4326,
27		)
28
29		ehv_grid_districts = db.select_geodataframe(
30		f"""
31		SELECT * FROM {cfg['sources']['ehv_voronoi']}
32		""",
33		epsg=4326,
34		)
35
36		# Assign power plants in hv and below to hvmv bus
37		power_plants_hv = power_plants[power_plants.voltage_level >= 3].index
38		if len(power_plants_hv) > 0:
39		power_plants.loc[power_plants_hv, "bus_id"] = gpd.sjoin(
40		power_plants[power_plants.index.isin(power_plants_hv)],
41		mv_grid_districts,
42		).bus_id
43
44		# Assign power plants in ehv to ehv bus
45		power_plants_ehv = power_plants[power_plants.voltage_level < 3].index
46
47		if len(power_plants_ehv) > 0:
48		ehv_join = gpd.sjoin(
49		power_plants[power_plants.index.isin(power_plants_ehv)],
50		ehv_grid_districts,
51		)
52
53		if "bus_id_right" in ehv_join.columns:
54		power_plants.loc[power_plants_ehv, "bus_id"] = gpd.sjoin(
55		power_plants[power_plants.index.isin(power_plants_ehv)],
56		ehv_grid_districts,
57		).bus_id_right
58
59		else:
60		power_plants.loc[power_plants_ehv, "bus_id"] = gpd.sjoin(
61		power_plants[power_plants.index.isin(power_plants_ehv)],
62		ehv_grid_districts,
63		).bus_id
64
65		# Assert that all power plants have a bus_id
66		assert power_plants.bus_id.notnull().all(), f"""Some power plants are
67		not attached to a bus: {power_plants[power_plants.bus_id.isnull()]}"""
68
69		return power_plants
70
71
72		@timer_func
73		def add_missing_bus_ids(scn_name):
74		"""Assign busses by spatal intersection of mvgrid districts or ehv voronois."""
75
76		sql = f"""
77		-- Assign missing buses to mv grid district buses for HV and below
78		UPDATE supply.egon_power_plants AS epp
79		SET bus_id = (
80		SELECT emgd.bus_id
81		FROM grid.egon_mv_grid_district AS emgd
82		WHERE ST_Intersects(ST_Transform(epp.geom, 4326), ST_Transform(emgd.geom, 4326))
83		ORDER BY ST_Transform(emgd.geom, 4326) <-> ST_Transform(epp.geom, 4326)
84		LIMIT 1
85		)
86		WHERE (epp.carrier = 'solar'
87		OR epp.carrier = 'wind_onshore'
88		OR epp.carrier = 'solar_rooftop'
89		OR epp.carrier = 'wind_offshore')
90		AND epp.scenario = '{scn_name}'
91		AND epp.bus_id is null
92		AND epp.voltage_level >= 3; -- HV and below
93
94
95		-- Assign missing buses to EHV buses for EHV
96		UPDATE supply.egon_power_plants AS epp
97		SET bus_id = (
98		SELECT eesv.bus_id
99		FROM grid.egon_ehv_substation_voronoi AS eesv
100		WHERE ST_Intersects(ST_Transform(epp.geom, 4326), ST_Transform(eesv.geom, 4326))
101		ORDER BY ST_Transform(eesv.geom, 4326) <-> ST_Transform(epp.geom, 4326)
102		LIMIT 1
103		)
104		WHERE (epp.carrier = 'solar'
105		OR epp.carrier = 'wind_onshore'
106		OR epp.carrier = 'solar_rooftop'
107		OR epp.carrier = 'wind_offshore')
108		AND epp.scenario = '{scn_name}'
109		AND epp.bus_id is null
110		AND epp.voltage_level < 3; --EHV
111
112
113		"""
114
115		db.execute_sql(sql)
116
117
118		@timer_func
119		def find_weather_id(scn_name):
120
121		sql = f"""UPDATE supply.egon_power_plants AS epp
122		SET weather_cell_id = (
123		SELECT eewc.w_id
124		FROM supply.egon_era5_weather_cells AS eewc
125		WHERE ST_Intersects(epp.geom, eewc.geom)
126		ORDER BY eewc.geom <-> epp.geom
127		LIMIT 1
128		)
129		WHERE (epp.carrier = 'solar'
130		OR epp.carrier = 'solar_rooftop'
131		OR epp.carrier = 'wind_onshore'
132		OR epp.carrier = 'wind_offshore')
133		AND epp.scenario = '{scn_name}';
134		"""
135
136		db.execute_sql(sql)
137
138
139		def weatherId_and_busId(scn_name):
140		find_weather_id(scn_name)
141		add_missing_bus_ids(scn_name)
142

openego / eGon-data

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

assign_bus_id() A

Complexity

Size

Duplication

Importance

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like