from __future__ import annotations
from abc import ABC, abstractmethod
import matplotlib.pyplot as plt
import networkx as nx
import pandas as pd
from networkx.drawing.nx_pydot import graphviz_layout
from edisgo.network.components import Generator, Load, Switch
from edisgo.tools.geopandas_helper import to_geopandas
from edisgo.tools.networkx_helper import translate_df_to_graph
[docs]
class Grid(ABC):
"""
Defines a basic grid in eDisGo.
Parameters
-----------
edisgo_obj : :class:`~.EDisGo`
id : str or int, optional
Identifier
"""
def __init__(self, **kwargs):
self._id = kwargs.get("id", None)
if isinstance(self._id, float):
self._id = int(self._id)
self._edisgo_obj = kwargs.get("edisgo_obj", None)
self._nominal_voltage = None
@property
def id(self):
"""
ID of the grid.
"""
return self._id
@property
def edisgo_obj(self):
"""
EDisGo object the grid is stored in.
"""
return self._edisgo_obj
@property
def nominal_voltage(self):
"""
Nominal voltage of network in kV.
Parameters
----------
nominal_voltage : float
Returns
-------
float
Nominal voltage of network in kV.
"""
if self._nominal_voltage is None:
self._nominal_voltage = self.buses_df.v_nom.max()
return self._nominal_voltage
@nominal_voltage.setter
def nominal_voltage(self, nominal_voltage):
self._nominal_voltage = nominal_voltage
@property
def graph(self):
"""
Graph representation of the grid.
Returns
-------
:networkx:`networkx.Graph<>`
Graph representation of the grid as networkx Ordered Graph,
where lines are represented by edges in the graph, and buses and
transformers are represented by nodes.
"""
return translate_df_to_graph(self.buses_df, self.lines_df)
@property
def geopandas(self):
"""
Returns components as :geopandas:`GeoDataFrame`\\ s
Returns container with :geopandas:`GeoDataFrame`\\ s containing all
georeferenced components within the grid.
Returns
-------
:class:`~.tools.geopandas_helper.GeoPandasGridContainer` or \
list(:class:`~.tools.geopandas_helper.GeoPandasGridContainer`)
Data container with GeoDataFrames containing all georeferenced components
within the grid(s).
"""
return to_geopandas(self)
@property
def station(self):
"""
DataFrame with form of buses_df with only grid's station's secondary
side bus information.
"""
return self.buses_df.loc[self.transformers_df.iloc[0].bus1].to_frame().T
@property
def station_name(self):
"""
Name of station to the overlying voltage level.
Name of station is composed of grid name with the extension '_station'.
"""
return f"{self}_station"
@property
def generators_df(self):
"""
Connected generators within the network.
Returns
-------
:pandas:`pandas.DataFrame<DataFrame>`
Dataframe with all generators in topology. For more information on
the dataframe see
:attr:`~.network.topology.Topology.generators_df`.
"""
return self.edisgo_obj.topology.generators_df[
self.edisgo_obj.topology.generators_df.bus.isin(self.buses_df.index)
]
@property
def generators(self):
"""
Connected generators within the network.
Returns
-------
list(:class:`~.network.components.Generator`)
List of generators within the network.
"""
for gen in self.generators_df.index:
yield Generator(id=gen, edisgo_obj=self.edisgo_obj)
@property
def loads_df(self):
"""
Connected loads within the network.
Returns
-------
:pandas:`pandas.DataFrame<DataFrame>`
Dataframe with all loads in topology. For more information on the
dataframe see :attr:`~.network.topology.Topology.loads_df`.
"""
return self.edisgo_obj.topology.loads_df[
self.edisgo_obj.topology.loads_df.bus.isin(self.buses_df.index)
]
@property
def loads(self):
"""
Connected loads within the network.
Returns
-------
list(:class:`~.network.components.Load`)
List of loads within the network.
"""
for load in self.loads_df.index:
yield Load(id=load, edisgo_obj=self.edisgo_obj)
@property
def storage_units_df(self):
"""
Connected storage units within the network.
Returns
-------
:pandas:`pandas.DataFrame<DataFrame>`
Dataframe with all storage units in topology. For more information
on the dataframe see
:attr:`~.network.topology.Topology.storage_units_df`.
"""
return self.edisgo_obj.topology.storage_units_df[
self.edisgo_obj.topology.storage_units_df.bus.isin(self.buses_df.index)
]
@property
def charging_points_df(self):
"""
Connected charging points within the network.
Returns
-------
:pandas:`pandas.DataFrame<DataFrame>`
Dataframe with all charging points in topology. For more
information on the dataframe see
:attr:`~.network.topology.Topology.loads_df`.
"""
return self.loads_df[self.loads_df.type == "charging_point"]
@property
def switch_disconnectors_df(self):
"""
Switch disconnectors in network.
Switch disconnectors are points where rings are split under normal
operating conditions.
Returns
-------
:pandas:`pandas.DataFrame<DataFrame>`
Dataframe with all switch disconnectors in network. For more
information on the dataframe see
:attr:`~.network.topology.Topology.switches_df`.
"""
return self.edisgo_obj.topology.switches_df[
self.edisgo_obj.topology.switches_df.bus_closed.isin(self.buses_df.index)
][self.edisgo_obj.topology.switches_df.type_info == "Switch Disconnector"]
@property
def switch_disconnectors(self):
"""
Switch disconnectors within the network.
Returns
-------
list(:class:`~.network.components.Switch`)
List of switch disconnectory within the network.
"""
for s in self.switch_disconnectors_df.index:
yield Switch(id=s, edisgo_obj=self.edisgo_obj)
@property
def lines_df(self):
"""
Lines within the network.
Returns
-------
:pandas:`pandas.DataFrame<DataFrame>`
Dataframe with all buses in topology. For more information on the
dataframe see :attr:`~.network.topology.Topology.lines_df`.
"""
return self.edisgo_obj.topology.lines_df[
self.edisgo_obj.topology.lines_df.bus0.isin(self.buses_df.index)
& self.edisgo_obj.topology.lines_df.bus1.isin(self.buses_df.index)
]
@property
@abstractmethod
def buses_df(self):
"""
Buses within the network.
Returns
-------
:pandas:`pandas.DataFrame<DataFrame>`
Dataframe with all buses in topology. For more information on the
dataframe see :attr:`~.network.topology.Topology.buses_df`.
"""
@property
def weather_cells(self):
"""
Weather cells in network.
Returns
-------
list(int)
List of weather cell IDs in network.
"""
return self.generators_df.weather_cell_id.dropna().unique()
@property
def peak_generation_capacity(self):
"""
Cumulative peak generation capacity of generators in the network in MW.
Returns
-------
float
Cumulative peak generation capacity of generators in the network
in MW.
"""
return self.generators_df.p_nom.sum()
@property
def peak_generation_capacity_per_technology(self):
"""
Cumulative peak generation capacity of generators in the network per
technology type in MW.
Returns
-------
:pandas:`pandas.DataFrame<DataFrame>`
Cumulative peak generation capacity of generators in the network
per technology type in MW.
"""
return self.generators_df.groupby(["type"]).sum()["p_nom"]
@property
def p_set(self):
"""
Cumulative peak load of loads in the network in MW.
Returns
-------
float
Cumulative peak load of loads in the network in MW.
"""
return self.loads_df.p_set.sum()
@property
def p_set_per_sector(self):
"""
Cumulative peak load of loads in the network per sector in MW.
Returns
-------
:pandas:`pandas.DataFrame<DataFrame>`
Cumulative peak load of loads in the network per sector in MW.
"""
return self.loads_df.groupby(["sector"]).sum()["p_set"]
[docs]
def assign_length_to_grid_station(self):
"""
Assign length in km from each bus in the grid to the grid's station.
The length is written to column 'length_to_grid_station' in
:attr:`~.network.topology.Topology.buses_df`.
"""
buses_df = self._edisgo_obj.topology.buses_df
graph = self.graph
station = self.station.index[0]
for bus in self.buses_df.index:
buses_df.at[bus, "length_to_grid_station"] = nx.shortest_path_length(
graph, source=station, target=bus, weight="length"
)
[docs]
def assign_grid_feeder(self, mode: str = "grid_feeder"):
"""
Assigns MV or LV feeder to each bus and line, depending on the `mode`.
See :attr:`~.network.topology.Topology.assign_feeders` for more information.
Parameters
----------
mode : str
Specifies whether to assign MV or grid feeder.
If mode is "mv_feeder" the MV feeder the buses and lines are in are
determined. If mode is "grid_feeder" LV buses and lines are assigned the
LV feeder they are in and MV buses and lines are assigned the MV feeder
they are in. Default: "grid_feeder".
"""
buses_df = self._edisgo_obj.topology.buses_df
lines_df = self._edisgo_obj.topology.lines_df
if mode == "grid_feeder":
graph = self.graph
column_name = "grid_feeder"
elif mode == "mv_feeder":
graph = self._edisgo_obj.topology.to_graph()
column_name = "mv_feeder"
else:
raise ValueError("Choose an existing mode.")
station = self.station.index[0]
# get all buses in network and remove station to get separate sub-graphs
graph_nodes = list(graph.nodes())
graph_nodes.remove(station)
subgraph = graph.subgraph(graph_nodes)
buses_df.at[station, column_name] = "station_node"
for neighbor in graph.neighbors(station):
# get all nodes in that feeder by doing a DFS in the disconnected
# subgraph starting from the node adjacent to the station `neighbor`
feeder_graph = nx.dfs_tree(subgraph, source=neighbor)
feeder_lines = set()
for node in feeder_graph.nodes():
buses_df.at[node, column_name] = neighbor
feeder_lines.update(
{edge[2]["branch_name"] for edge in graph.edges(node, data=True)}
)
lines_df.loc[lines_df.index.isin(feeder_lines), column_name] = neighbor
[docs]
def get_feeder_stats(self) -> pd.DataFrame:
"""
Generate statistics of the grid's feeders.
So far, only the feeder length is determined.
Returns
-------
:pandas:`pandas.DataFrame<DataFrame>`
Dataframe with feeder name in index and column 'length' containing the
respective feeder length in km.
"""
self.assign_grid_feeder()
self.assign_length_to_grid_station()
buses_df = self.buses_df
feeders = (
buses_df.loc[
buses_df["grid_feeder"] != "station_node",
["grid_feeder", "length_to_grid_station"],
]
.groupby("grid_feeder")
.max()
.rename(columns={"length_to_grid_station": "length"})
)
return feeders
[docs]
def __repr__(self):
return "_".join([self.__class__.__name__, str(self.id)])
[docs]
class MVGrid(Grid):
"""
Defines a medium voltage network in eDisGo.
"""
def __init__(self, **kwargs):
super().__init__(**kwargs)
@property
def lv_grids(self):
"""
Yields generator object with all underlying low voltage grids.
Returns
--------
:class:`~.network.grids.LVGrid`
Yields generator object with :class:`~.network.grids.LVGrid` object.
"""
return self.edisgo_obj.topology.lv_grids
@property
def buses_df(self):
"""
Buses within the network.
Returns
-------
:pandas:`pandas.DataFrame<DataFrame>`
Dataframe with all buses in topology. For more information on the
dataframe see :attr:`~.network.topology.Topology.buses_df`.
"""
return self.edisgo_obj.topology.buses_df.drop(
self.edisgo_obj.topology.buses_df.lv_grid_id.dropna().index
)
@property
def transformers_df(self):
"""
Transformers to overlaying network.
Returns
-------
:pandas:`pandas.DataFrame<DataFrame>`
Dataframe with all transformers to overlaying network. For more
information on the dataframe see
:attr:`~.network.topology.Topology.transformers_df`.
"""
return self.edisgo_obj.topology.transformers_hvmv_df
[docs]
def draw(self):
"""
Draw MV network.
"""
raise NotImplementedError
[docs]
class LVGrid(Grid):
"""
Defines a low voltage network in eDisGo.
"""
def __init__(self, **kwargs):
super().__init__(**kwargs)
@property
def buses_df(self):
"""
Buses within the network.
Returns
-------
:pandas:`pandas.DataFrame<DataFrame>`
Dataframe with all buses in topology. For more information on the
dataframe see :attr:`~.network.topology.Topology.buses_df`.
"""
return self.edisgo_obj.topology.buses_df.loc[
self.edisgo_obj.topology.buses_df.lv_grid_id == self.id
]
@property
def transformers_df(self):
"""
Transformers to overlaying network.
Returns
-------
:pandas:`pandas.DataFrame<DataFrame>`
Dataframe with all transformers to overlaying network. For more
information on the dataframe see
:attr:`~.network.topology.Topology.transformers_df`.
"""
return self.edisgo_obj.topology.transformers_df[
self.edisgo_obj.topology.transformers_df.bus1.isin(self.buses_df.index)
]
[docs]
def draw(
self,
node_color="black",
edge_color="black",
colorbar=False,
labels=False,
filename=None,
):
"""
Draw LV network.
Currently, edge width is proportional to nominal apparent power of
the line and node size is proportional to peak load of connected loads.
Parameters
-----------
node_color : str or :pandas:`pandas.Series<Series>`
Color of the nodes (buses) of the grid. If provided as string
all nodes will have that color. If provided as series, the
index of the series must contain all buses in the LV grid and the
corresponding values must be float values, that will be translated
to the node color using a colormap, currently set to "Blues".
Default: "black".
edge_color : str or :pandas:`pandas.Series<Series>`
Color of the edges (lines) of the grid. If provided as string
all edges will have that color. If provided as series, the
index of the series must contain all lines in the LV grid and the
corresponding values must be float values, that will be translated
to the edge color using a colormap, currently set to "inferno_r".
Default: "black".
colorbar : bool
If True, a colorbar is added to the plot for node and edge colors,
in case these are sequences. Default: False.
labels : bool
If True, displays bus names. As bus names are quite long, this
is currently not very pretty. Default: False.
filename : str or None
If a filename is provided, the plot is saved under that name but
not displayed. If no filename is provided, the plot is only
displayed. Default: None.
"""
G = self.graph
pos = graphviz_layout(G, prog="dot")
# assign edge width + color and node size + color
top = self.edisgo_obj.topology
edge_width = [
top.get_line_connecting_buses(u, v).s_nom.sum() * 10 for u, v in G.edges()
]
if isinstance(edge_color, pd.Series):
edge_color = [
edge_color.loc[top.get_line_connecting_buses(u, v).index[0]]
for u, v in G.edges()
]
edge_color_is_sequence = True
else:
edge_color_is_sequence = False
node_size = [
top.get_connected_components_from_bus(v)["loads"].p_set.sum() * 50000 + 10
for v in G
]
if isinstance(node_color, pd.Series):
node_color = [node_color.loc[v] for v in G]
node_color_is_sequence = True
else:
node_color_is_sequence = False
# draw edges and nodes of the graph
fig, ax = plt.subplots(figsize=(12, 12))
cm_edges = nx.draw_networkx_edges(
G,
pos,
width=edge_width,
edge_color=edge_color,
edge_cmap=plt.cm.get_cmap("inferno_r"),
)
cm_nodes = nx.draw_networkx_nodes(
G, pos, node_size=node_size, node_color=node_color, cmap="Blues"
)
if colorbar:
if edge_color_is_sequence:
fig.colorbar(cm_edges, ax=ax)
if node_color_is_sequence:
fig.colorbar(cm_nodes, ax=ax)
if labels:
# ToDo find nicer way to display bus names
label_options = {"ec": "k", "fc": "white", "alpha": 0.7}
nx.draw_networkx_labels(
G,
pos,
font_size=8,
bbox=label_options,
horizontalalignment="right",
)
if filename is None:
plt.show()
else:
plt.savefig(filename, dpi=150, bbox_inches="tight", pad_inches=0.1)
plt.close()
@property
def geopandas(self):
"""
TODO: Remove this as soon as LVGrids are georeferenced
"""
raise NotImplementedError("LV Grids are not georeferenced yet.")