Connecting a new consumer
This demo shows how to use power-grid-model-ds to simulate a new consumer to be connected to the grid
First we create an extension of the Grid objects with properties we want to use in this context
We create a random grid structure for the purpose of the demo
We define functions that add a new consumer to the grid and simulate its impact
from dataclasses import dataclass
import numpy as np
from numpy.typing import NDArray
from power_grid_model_ds import Grid
from power_grid_model_ds.arrays import LineArray, NodeArray
class ExtendedNodeArray(NodeArray):
"""Extends the node array with the simulated voltage and coordinates"""
_defaults = {"u": 0}
u: NDArray[np.float64]
x_coor: NDArray[np.float64]
y_coor: NDArray[np.float64]
@property
def is_overloaded(self):
return np.logical_or(self.u > 1.1 * self.u_rated, self.u < 0.9 * self.u_rated)
class ExtendedLineArray(LineArray):
"""Extends the line array with current output"""
_defaults = {"i_from": 0}
i_from: NDArray[np.float64]
@property
def is_overloaded(self):
return self.i_from > self.i_n
@dataclass
class ExtendedGrid(Grid):
node: ExtendedNodeArray
line: ExtendedLineArray
from power_grid_model_ds.generators import RadialGridGenerator
grid_generator = RadialGridGenerator(grid_class=ExtendedGrid, nr_nodes=20, nr_sources=1, nr_nops=10)
grid = grid_generator.run(seed=0)
grid.set_feeder_ids()
grid.node.x_coor = np.random.uniform(100, 500, len(grid.node))
grid.node.y_coor = np.random.uniform(100, 500, len(grid.node))
First we create a new consumer, with a location and a load demand
from power_grid_model import LoadGenType
from power_grid_model_ds.arrays import SymLoadArray
from power_grid_model_ds.enums import NodeType
def create_new_consumer_arrays(
u_rated: float, x_coor: float, y_coor: float, p_specified: float, q_specified: float
) -> tuple[ExtendedNodeArray, SymLoadArray]:
new_consumer = ExtendedNodeArray(
u_rated=[u_rated],
node_type=[NodeType.UNSPECIFIED],
x_coor=[x_coor],
y_coor=[y_coor],
)
new_consumer_load = SymLoadArray(
node=[new_consumer.get_empty_value("id")],
status=[1],
type=[LoadGenType.const_power],
p_specified=[p_specified],
q_specified=[q_specified],
)
return new_consumer, new_consumer_load
new_consumer, new_consumer_load = create_new_consumer_arrays(10_500, 300, 300, 1_000_000, 200_000)
Now lets define some functions that add the new consumer by connecting it to the closest node
from power_grid_model_ds import PowerGridModelInterface
R_PER_KM = 0.1
X_PER_KM = 0.1
def find_closest_node(grid: ExtendedGrid, x: float, y: float) -> int:
dist = np.sqrt((grid.node.x_coor - x) ** 2 + (grid.node.y_coor - y) ** 2)
return np.argmin(dist)
def connect_new_consumer(
grid: ExtendedGrid,
new_consumer: ExtendedNodeArray,
new_consumer_load: SymLoadArray,
):
closest_node_idx = find_closest_node(
grid=grid,
x=new_consumer.x_coor[0],
y=new_consumer.y_coor[0],
)
closest_node = grid.node[closest_node_idx]
grid.append(new_consumer)
new_consumer_load.node = new_consumer.id
grid.append(new_consumer_load)
dist = np.sqrt((closest_node.x_coor - new_consumer.x_coor) ** 2 + (closest_node.y_coor - new_consumer.y_coor) ** 2)
new_line = ExtendedLineArray(
from_node=[closest_node.id],
to_node=[new_consumer.id],
from_status=[1],
to_status=[1],
r1=[R_PER_KM * dist / 1_000],
x1=[X_PER_KM * dist / 1_000],
c1=[0],
tan1=[0],
i_n=[200],
)
grid.append(new_line)
def update_grid(grid: ExtendedGrid):
# Set the new feeder ids
grid.set_feeder_ids()
# Update the power flow
core_interface = PowerGridModelInterface(grid=grid)
core_interface.create_input_from_grid()
core_interface.calculate_power_flow()
core_interface.update_grid()
connect_new_consumer(grid, new_consumer, new_consumer_load)
update_grid(grid)
---------------------------------------------------------------------------
NotImplementedError Traceback (most recent call last)
Cell In[5], line 1
----> 1 connect_new_consumer(grid, new_consumer, new_consumer_load)
2 update_grid(grid)
Cell In[4], line 22, in connect_new_consumer(grid, new_consumer, new_consumer_load)
12 def connect_new_consumer(
13 grid: ExtendedGrid,
14 new_consumer: ExtendedNodeArray,
15 new_consumer_load: SymLoadArray,
16 ):
17 closest_node_idx = find_closest_node(
18 grid=grid,
19 x=new_consumer.x_coor[0],
20 y=new_consumer.y_coor[0],
21 )
---> 22 closest_node = grid.node[closest_node_idx]
24 grid.append(new_consumer)
25 new_consumer_load.node = new_consumer.id
File ~/checkouts/readthedocs.org/user_builds/power-grid-model-ds/checkouts/v1.7.0/src/power_grid_model_ds/_core/model/arrays/base/array.py:179, in FancyArray.__getitem__(self, item)
177 if np.issubdtype(item_array.dtype, np.str_):
178 return self._data[item_array.tolist()]
--> 179 raise NotImplementedError(
180 f"FancyArray[{type(item).__name__}] is not supported. Try FancyArray.data[{type(item).__name__}] instead."
181 )
NotImplementedError: FancyArray[int64] is not supported. Try FancyArray.data[int64] instead.
We can inspect the results
The grid has been extended (graph and arrays)
Load values have been updated on node and line arrays
The feeder ids have been updated for the new consumer
print(grid.node)
print(grid.line)
print(f"Overloaded nodes: {grid.node[grid.node.is_overloaded].id}")
print(f"Overloaded lines: {grid.line[grid.line.is_overloaded].id}")
Now simulate more consumers being added, as to see how this will lead to overloads
for _ in range(10):
new_consumer, new_consumer_load = create_new_consumer_arrays(
10_500, np.random.uniform(0, 500), np.random.uniform(0, 500), 1_000_000, 200_000
)
connect_new_consumer(grid, new_consumer, new_consumer_load)
update_grid(grid)
print(f"Overloaded nodes: {grid.node[grid.node.is_overloaded].id}")
print(f"Overloaded lines: {grid.line[grid.line.is_overloaded].id}")