Source code for dave_core.topology.medium_voltage

# Copyright (c) 2022-2024 by Fraunhofer Institute for Energy Economics and Energy System Technology (IEE)
# Kassel and individual contributors (see AUTHORS file for details). All rights reserved.
# Use of this source code is governed by a BSD-style license that can be found in the LICENSE file.

from geopandas import GeoDataFrame
from geopandas import GeoSeries
from pandas import Series
from pandas import concat
from shapely.geometry import LineString
from shapely.geometry import MultiLineString
from shapely.geometry import Point
from shapely.ops import linemerge
from shapely.wkb import loads

from dave_core.datapool.oep_request import oep_request
from dave_core.progressbar import create_tqdm
from dave_core.settings import dave_settings
from dave_core.toolbox import intersection_with_area


[docs] def create_hv_mv_substations(grid_data): """ This function requests data for the hv/mv substations if there not already included in grid data """ if grid_data.components_power.substations.hv_mv.empty: hvmv_substations, meta_data = oep_request( table="ego_dp_hvmv_substation" ) # take polygon for full area # add meta data if ( bool(meta_data) and f"{meta_data['Main'].Titel.loc[0]}" not in grid_data.meta_data.keys() ): grid_data.meta_data[f"{meta_data['Main'].Titel.loc[0]}"] = ( meta_data ) hvmv_substations.rename( columns={ "version": "ego_version", "subst_id": "ego_subst_id", "voltage": "voltage_kv", "ags_0": "Gemeindeschluessel", }, inplace=True, ) # filter substations which are within the grid area hvmv_substations = intersection_with_area( hvmv_substations, grid_data.area ) if not hvmv_substations.empty: hvmv_substations["voltage_level"] = 4 # add dave name hvmv_substations.reset_index(drop=True, inplace=True) hvmv_substations.insert( 0, "dave_name", Series([f"substation_4_{x}" for x in hvmv_substations.index]), ) # set crs hvmv_substations.set_crs(dave_settings["crs_main"], inplace=True) # add ehv substations to grid data grid_data.components_power.substations.hv_mv = concat( [ grid_data.components_power.substations.hv_mv, hvmv_substations, ] ) else: hvmv_substations = grid_data.components_power.substations.hv_mv.copy() return hvmv_substations
[docs] def create_mv_lv_substations(grid_data): """ This function requests data for the mv/lv substations if there not already included in grid data """ mvlv_substations, meta_data = oep_request(table="ego_dp_mvlv_substation") # add meta data if ( bool(meta_data) and f"{meta_data['Main'].Titel.loc[0]}" not in grid_data.meta_data.keys() ): grid_data.meta_data[f"{meta_data['Main'].Titel.loc[0]}"] = meta_data # change wrong crs from oep mvlv_substations.crs = dave_settings["crs_main"] mvlv_substations.rename( columns={"version": "ego_version", "mvlv_subst_id": "ego_subst_id"}, inplace=True, ) # filter trafos which are within the grid area mvlv_substations = intersection_with_area(mvlv_substations, grid_data.area) if ( grid_data.components_power.substations.mv_lv.empty and not mvlv_substations.empty ): mvlv_substations["voltage_level"] = 6 # add dave name mvlv_substations.reset_index(drop=True, inplace=True) mvlv_substations.insert( 0, "dave_name", Series([f"substation_6_{x}" for x in mvlv_substations.index]), ) # add ehv substations to grid data grid_data.components_power.substations.mv_lv = concat( [grid_data.components_power.substations.mv_lv, mvlv_substations], ignore_index=True, ) else: mvlv_substations = grid_data.components_power.substations.mv_lv.copy() return mvlv_substations
[docs] def search_connection_line(bus, mv_buses): """ Search connection line """ nearest_bus_idx = ( mv_buses.drop([bus.name]) .geometry.apply(lambda x: bus.geometry.distance(x)) .idxmin() ) return LineString([bus.geometry, mv_buses.loc[nearest_bus_idx].geometry])
[docs] def create_mv_topology(grid_data): """ This function creates a dictonary with all relevant parameters for the medium voltage level INPUT: **grid_data** (dict) - all Informations about the target area OUTPUT: Writes data in the DaVe dataset """ # set progress bar pbar = create_tqdm(desc="create medium voltage topology") # --- create substations # create hv/mv substations hvmv_substations = create_hv_mv_substations(grid_data) # update progress pbar.update(5) # create mv/lv substations mvlv_substations = create_mv_lv_substations(grid_data) # update progress pbar.update(10) # --- create mv nodes # copy data for mv node creation mvlv_buses = mvlv_substations.copy() # nodes for mv/lv traofs hv side mvlv_buses.drop( columns=( [ "dave_name", "la_id", "subst_id", "geom", "is_dummy", "subst_cnt", "voltage_level", ] ), inplace=True, ) mvlv_buses["node_type"] = "mvlv_substation" # update progress pbar.update(5) # nodes for hv/mv trafos us side hvmv_buses = hvmv_substations.copy() if not hvmv_buses.empty: hvmv_buses.drop( columns=( [ "dave_name", "lon", "lat", "polygon", "voltage_kv", "power_type", "substation", "osm_id", "osm_www", "frequency", "subst_name", "ref", "operator", "dbahn", "status", "otg_id", "Gemeindeschluessel", "geom", "geometry", "voltage_level", ] ), inplace=True, ) hvmv_buses["node_type"] = "hvmv_substation" # change geometry to point hvmv_buses["geometry"] = hvmv_buses.point.apply( lambda x: loads(x, hex=True) ) # filter trafos which are within the grid area hvmv_buses = intersection_with_area(hvmv_buses, grid_data.area) hvmv_buses.drop( columns=( [ "point", ] ), inplace=True, ) hvmv_buses["node_type"] = "hvmv_substation" # update progress pbar.update(10) # consider data only if there are more than one node in the target area mv_buses = concat([mvlv_buses, hvmv_buses]) if len(mv_buses) > 1: # search for the substations dave name substations_rel = concat([hvmv_substations, mvlv_substations]) mv_buses["subs_dave_name"] = mv_buses.ego_subst_id.apply( lambda x: substations_rel[substations_rel.ego_subst_id == x] .iloc[0] .dave_name ) mv_buses["voltage_level"] = 5 mv_buses["voltage_kv"] = dave_settings["mv_voltage"] # add oep as source mv_buses["source"] = "OEP" # add dave name mv_buses.reset_index(drop=True, inplace=True) mv_buses.insert( 0, "dave_name", Series([f"node_5_{x}" for x in mv_buses.index]), ) # set crs mv_buses.set_crs(dave_settings["crs_main"], inplace=True) # add mv nodes to grid data grid_data.mv_data.mv_nodes = concat( [grid_data.mv_data.mv_nodes, mv_buses], ignore_index=True ) # --- create mv lines # lines to connect node with the nearest node # mv_line = mv_buses.apply(lambda x: search_connection_line(x, mv_buses), axis=1) # mv_line.drop_duplicates(inplace=True) # list(map(lambda x: search_connection_line(x, mv_buses), mv_buses)) mv_lines = GeoSeries([]) for i, bus in mv_buses.iterrows(): mv_line = search_connection_line(bus, mv_buses) # check if line already exists if not mv_lines.geom_equals(mv_line).any(): mv_lines[i] = mv_line # update progress pbar.update(10) mv_lines.set_crs(dave_settings["crs_main"], inplace=True) mv_lines.reset_index(drop=True, inplace=True) # connect line segments with each other while 1: # search for related lines and merge them mv_lines_rel = mv_lines.copy() for _, bus in mv_buses.iterrows(): # check if bus is conected to more than one line lines_intersect = mv_lines_rel[ mv_lines_rel.intersects(bus.geometry) ] if len(lines_intersect) > 1: # get list with line objects lines_list = lines_intersect.tolist() # search for multilines and split them new_line = [ list(x.geoms) if isinstance(x, MultiLineString) else [x] for x in lines_list ] new_line = [ line for sublist in new_line for line in sublist ] # merge found lines and add new line to line quantity mv_lines_rel[len(mv_lines)] = linemerge(new_line) # delete found lines from line quantity mv_lines_rel.drop( lines_intersect.index.tolist(), inplace=True ) mv_lines_rel.reset_index(drop=True, inplace=True) # break loop if all lines connected if len(mv_lines_rel) == 1: break # create lines for connecting line segments for i, line in enumerate(mv_lines_rel.to_list()): # find nearest line to considered one nearest_line_idx = ( mv_lines_rel.drop([i]).geometry.distance(line).idxmin() ) # get line coordinates if isinstance(line, MultiLineString): line_points = GeoSeries( [ Point(coords) for segment in line.geoms for coords in segment.coords[:] ] ) else: line_points = GeoSeries( [Point(coords) for coords in line.coords[:]] ) # set crs line_points = line_points.set_crs(dave_settings["crs_main"]) # get nearest line coordinates nearest_line = mv_lines_rel.loc[nearest_line_idx] if isinstance(nearest_line, MultiLineString): nearest_line_points = GeoSeries( [ Point(coords) for segment in nearest_line.geoms for coords in segment.coords[:] ] ) else: nearest_line_points = GeoSeries( [ Point( mv_lines_rel.loc[nearest_line_idx].coords[:][j] ) for j in range(len(nearest_line.coords[:])) ] ) # set crs nearest_line_points.set_crs( dave_settings["crs_main"], inplace=True ) # define minimal distance for initialize distance_min = 1000 # any big number # find pair of nearest nodes for point in line_points: distance = nearest_line_points.geometry.apply( lambda x, point=point: point.distance(x) ) if distance_min > distance.min(): distance_min = distance.min() nearest_point = nearest_line_points[distance.idxmin()] line_point = point # add created connection line into mv lines mv_line = LineString([line_point, nearest_point]) if not mv_lines.geom_equals(mv_line).any(): mv_lines[len(mv_lines)] = mv_line # update progress pbar.update(40) # prepare dataframe for mv lines mv_lines = GeoDataFrame(geometry=mv_lines) # project lines to crs with unit in meter for length calculation mv_lines.set_crs(dave_settings["crs_main"], inplace=True) mv_lines_3035 = mv_lines.to_crs(dave_settings["crs_meter"]) # add parameters to lines for _, line in mv_lines.iterrows(): # get from bus name mv_lines.at[line.name, "from_bus"] = mv_buses.loc[ mv_buses.geometry.apply( lambda x, line=line: Point( line.geometry.coords[:][0] ).distance(x) ).idxmin() ].dave_name # get to bus name mv_lines.at[line.name, "to_bus"] = mv_buses.loc[ mv_buses.geometry.apply( lambda x, line=line: Point( line.geometry.coords[:][1] ).distance(x) ).idxmin() ].dave_name # calculate length in km mv_lines["length_km"] = mv_lines_3035.geometry.length / 100 # line dave name mv_lines.insert( 0, "dave_name", Series([f"line_5_{x}" for x in mv_lines.index]), ) # additional informations mv_lines["voltage_kv"] = dave_settings["mv_voltage"] mv_lines["voltage_level"] = 5 mv_lines["source"] = "dave internal" # set crs mv_lines.set_crs(dave_settings["crs_main"], inplace=True) # add mv lines to grid data grid_data.mv_data.mv_lines = concat( [grid_data.mv_data.mv_lines, mv_lines], ignore_index=True ) # update progress pbar.update(20) else: # update progress pbar.update(80) # close progress bar pbar.close()