power_grid_model_ds

This is the API reference for the power-grid-model-ds library.

class power_grid_model_ds.Grid(_id_counter: int, graphs: GraphContainer, node: NodeArray, transformer: TransformerArray, three_winding_transformer: ThreeWindingTransformerArray, line: LineArray, link: LinkArray, source: SourceArray, sym_load: SymLoadArray, sym_gen: SymGenArray, transformer_tap_regulator: TransformerTapRegulatorArray, sym_power_sensor: SymPowerSensorArray, sym_voltage_sensor: SymVoltageSensorArray, asym_voltage_sensor: AsymVoltageSensorArray)

Bases: FancyArrayContainer

Grid object containing the entire network and interface to interact with it.

Examples

>>> from power_grid_model_ds import Grid
>>> grid = Grid.empty()
>>> grid

graphs: GraphContainer: The graph representations of the grid.

node: NodeArray

transformer: TransformerArray

three_winding_transformer: ThreeWindingTransformerArray

line: LineArray

link: LinkArray

source: SourceArray

sym_load: SymLoadArray

sym_gen: SymGenArray

transformer_tap_regulator: TransformerTapRegulatorArray

sym_power_sensor: SymPowerSensorArray

sym_voltage_sensor: SymVoltageSensorArray

asym_voltage_sensor: AsymVoltageSensorArray

property branches: BranchArray: Converts all branch arrays into a single BranchArray.

property branch_arrays: list[BranchArray]: Returns all branch arrays

get_typed_branches(branch_ids: list[int] | ndarray[tuple[int, ...], dtype[int32]]) → BranchArray

Find a matching LineArray, LinkArray or TransformerArray for the given branch_ids

Raises:

ValueError –

If no branch_ids are provided. - If not all branch_ids are of the same type.

reverse_branches(branches: BranchArray): Reverse the direction of the branches.

classmethod empty(graph_model: type[~power_grid_model_ds._core.model.graphs.models.base.BaseGraphModel] = <class 'power_grid_model_ds._core.model.graphs.models.rustworkx.RustworkxGraphModel'>) → Self

Create an empty grid

Parameters:: graph_model (type[BaseGraphModel], optional) – The graph model to use. Defaults to RustworkxGraphModel.
Returns:: An empty grid
Return type:: Grid

append(array: FancyArray, check_max_id: bool = True)

Append an array to the grid. Both ‘grid arrays’ and ‘grid.graphs’ will be updated.

Parameters:

array (FancyArray) – The array to append.
check_max_id (bool, optional) – Whether to check if the array id is the maximum id. Defaults to True.

add_branch(branch: BranchArray) → None

Add a branch to the grid

Parameters:: branch (BranchArray) – The branch to add

delete_branch(branch: BranchArray) → None

Remove a branch from the grid

Parameters:: branch (BranchArray) – The branch to remove

delete_branch3(branch: Branch3Array) → None

Remove a branch3 from the grid

Parameters:: branch (Branch3Array) – The branch3 to remove

add_node(node: NodeArray) → None

Add a new node to the grid

Parameters:: node (NodeArray) – The node to add

delete_node(node: NodeArray) → None

Remove a node from the grid

Parameters:: node (NodeArray) – The node to remove

make_active(branch: BranchArray) → None

Make a branch active

Parameters:: branch (BranchArray) – The branch to make active

make_inactive(branch: BranchArray, at_to_side: bool = True) → None

Make a branch inactive. This is done by setting from or to status to 0.

Parameters:

branch (BranchArray) – The branch to make inactive
at_to_side (bool, optional) – Whether to deactivate the to_status instead of the from_status.
True. (Defaults to)

get_branches_in_path(nodes_in_path: list[int]) → BranchArray

Returns all branches within a path of nodes

Parameters:: nodes_in_path (list[int]) – The nodes in the path
Returns:: The branches in the path
Return type:: BranchArray

get_nearest_substation_node(node_id: int)

Find the nearest substation node.

Parameters:: node_id (int) – The id of the node to find the nearest substation node for.
Returns:: The nearest substation node.
Return type:: NodeArray
Raises:: RecordDoesNotExist – If no substation node is connected to the input node.

get_downstream_nodes(node_id: int, inclusive: bool = False)

Get the downstream nodes from a node.

Example

given this graph: [1] - [2] - [3] - [4], with 1 being a substation node

>>> graph.get_downstream_nodes(2) == [3, 4]
>>> graph.get_downstream_nodes(3) == [4]
>>> graph.get_downstream_nodes(3, inclusive=True) == [3, 4]

Parameters:

node_id (int) – The id of the node to get the downstream nodes from.
inclusive (bool) – Whether to include the input node in the result.

Raises:

NotImplementedError – If the input node is a substation node.

Returns:

The downstream nodes.

Return type:

list[int]

cache(cache_dir: Path, cache_name: str, compress: bool = True)

Cache Grid to a folder

Parameters:

cache_dir (Path) – The directory to save the cache to.
cache_name (str) – The name of the cache.
compress (bool, optional) – Whether to compress the cache. Defaults to True.

classmethod from_cache(cache_path: Path, load_graphs: bool = True) → Self

Read from cache and build .graphs from arrays

Parameters:

cache_path (Path) – The path to the cache
load_graphs (bool, optional) – Whether to load the graphs. Defaults to True.

Returns:

The grid loaded from cache

Return type:

Self

classmethod from_txt(*args: str)

Build a grid from a list of strings

See the documentation for the expected format of the txt_lines
Args:
*args (str): The lines of the grid

Examples:
>>> Grid.from_txt("1 2", "2 3", "3 4 transformer", "4 5", "S1 6")
alternative: Grid.from_txt("1 2

2 3 3 4 transformer 4 5 S1 6”)

classmethod from_txt_file(txt_file_path: Path)

Load grid from txt file

Parameters:: txt_file_path (Path) – The path to the txt file

set_feeder_ids(): Sets feeder and substation id properties in the grids arrays

__init__(_id_counter: int, graphs: GraphContainer, node: NodeArray, transformer: TransformerArray, three_winding_transformer: ThreeWindingTransformerArray, line: LineArray, link: LinkArray, source: SourceArray, sym_load: SymLoadArray, sym_gen: SymGenArray, transformer_tap_regulator: TransformerTapRegulatorArray, sym_power_sensor: SymPowerSensorArray, sym_voltage_sensor: SymVoltageSensorArray, asym_voltage_sensor: AsymVoltageSensorArray) → None

class power_grid_model_ds.GraphContainer(active_graph: BaseGraphModel, complete_graph: BaseGraphModel)

Bases: object

Contains graphs

active_graph: BaseGraphModel: The graph containing only active branches.

complete_graph: BaseGraphModel: The graph containing all branches.

property graph_attributes: Generator

Get all graph attributes of the container.

Yields:: BaseGraphModel – The next graph attribute.

classmethod empty(graph_model: type[~power_grid_model_ds._core.model.graphs.models.base.BaseGraphModel] = <class 'power_grid_model_ds._core.model.graphs.models.rustworkx.RustworkxGraphModel'>) → GraphContainer

Get empty instance of GraphContainer.

Parameters:: graph_model (type[BaseGraphModel]) – The graph model to use. Defaults to RustworkxGraphModel. An alternative graph model can be passed as an argument.
Returns:: The empty graph container.
Return type:: GraphContainer

add_branch(branch: BranchArray) → None: Add a branch to all graphs

add_branch3(branch: Branch3Array) → None: Add a branch to all graphs

delete_branch(branch: BranchArray) → None: Remove a branch from all graphs

delete_branch3(branch: Branch3Array) → None: Remove a branch from all graphs

add_node(node: NodeArray) → None: Add a node to all graphs

delete_node(node: NodeArray) → None: Remove a node from all graphs

make_active(branch: BranchArray) → None: Add branch to all active_only graphs

make_inactive(branch: BranchArray) → None: Remove a branch from all active_only graphs

classmethod from_arrays(arrays: MinimalGridArrays) → GraphContainer: Build from arrays

__init__(active_graph: BaseGraphModel, complete_graph: BaseGraphModel) → None

class power_grid_model_ds.PowerGridModelInterface(grid: Grid, input_data: Dict | None = None, system_frequency: float = 50.0)

Bases: object

Interface between the Grid and the PowerGridModel (pgm).

Can convert grid data to pgm input
Can calculate power flow
Can do batch calculations using pgm
Can update grid with output from power flow

__init__(grid: Grid, input_data: Dict | None = None, system_frequency: float = 50.0)

create_input_from_grid(): Create input for the PowerGridModel

calculate_power_flow(calculation_method: CalculationMethod = CalculationMethod.newton_raphson, update_data: Dict | None = None, **kwargs)

Initialize the PowerGridModel and calculate power flow over input data.

If input data is not available, self.create_input_from_grid() will be called to create it.

Returns output of the power flow calculation (also stored in self.output_data)

update_model(update_data: Dict)

Updates the power-grid-model using update_data, this allows for batch calculations

Example

Example of update_data creation:

>>> update_sym_load = initialize_array('update', 'sym_load', 2)
>>> update_sym_load['id'] = [4, 7]  # same ID
>>> update_sym_load['p_specified'] = [30e6, 15e6]  # change active power
>>> # leave reactive power the same, no need to specify
>>>
>>> update_line = initialize_array('update', 'line', 1)
>>> update_line['id'] = [3]  # change line ID 3
>>> update_line['from_status'] = [0]  # switch off at from side
>>> # leave to-side swichint status the same, no need to specify
>>>
>>> update_data = {
>>>    'sym_load': update_sym_load,
>>>    'line': update_line
>>> }

update_grid() → None: Fills the output values in the grid for the values that are present

arrays

class power_grid_model_ds.arrays.IdArray(*args, data: ndarray[tuple[int, ...], dtype[_ScalarType_co]] | None = None, **kwargs)

Bases: Id, FancyArray

id: NDArray[np.int32]

class power_grid_model_ds.arrays.NodeArray(*args, data: ndarray[tuple[int, ...], dtype[_ScalarType_co]] | None = None, **kwargs)

Bases: IdArray, Node

id: NDArray[np.int32]

u_rated: NDArray[np.float64]

node_type: NDArray[np.int8]

feeder_branch_id: NDArray[np.int32]

feeder_node_id: NDArray[np.int32]

class power_grid_model_ds.arrays.BranchArray(*args, data: ndarray[tuple[int, ...], dtype[_ScalarType_co]] | None = None, **kwargs)

Bases: IdArray, Branch

property node_ids: Return both from_node and to_node in one array

property is_active: ndarray[tuple[int, ...], dtype[bool]]: Returns boolean whether branch is closed at both ends

filter_parallel(n_parallel: int, mode: Literal['eq', 'neq']) → BranchArray

Return branches that have n_parallel connections.

Parameters:

branches – BranchArray.
n_parallel – the number of connections between the same nodes
mode – mode of comparison. “eq” (equal) or “neq” (non-equal).

Returns:

when n_parallel is 1 and mode is ‘eq’, the function returns branches that are not parallel.
when n_parallel is 1 and mode is ‘neq’, the function returns branches that are parallel.

id: NDArray[np.int32]

from_node: NDArray[np.int32]

to_node: NDArray[np.int32]

from_status: NDArray[np.int8]

to_status: NDArray[np.int8]

feeder_branch_id: NDArray[np.int32]

feeder_node_id: NDArray[np.int32]

is_feeder: NDArray[np.bool_]

class power_grid_model_ds.arrays.LinkArray(*args, data: ndarray[tuple[int, ...], dtype[_ScalarType_co]] | None = None, **kwargs): Bases: Link, BranchArray

class power_grid_model_ds.arrays.LineArray(*args, data: ndarray[tuple[int, ...], dtype[_ScalarType_co]] | None = None, **kwargs)

Bases: Line, BranchArray

r1: NDArray[np.float64]

x1: NDArray[np.float64]

c1: NDArray[np.float64]

tan1: NDArray[np.float64]

i_n: NDArray[np.float64]

class power_grid_model_ds.arrays.TransformerArray(*args, data: ndarray[tuple[int, ...], dtype[_ScalarType_co]] | None = None, **kwargs)

Bases: Transformer, BranchArray

u1: NDArray[np.float64]

u2: NDArray[np.float64]

sn: NDArray[np.float64]

tap_size: NDArray[np.float64]

uk: NDArray[np.float64]

pk: NDArray[np.float64]

i0: NDArray[np.float64]

p0: NDArray[np.float64]

winding_from: NDArray[np.int8]

winding_to: NDArray[np.int8]

clock: NDArray[np.int8]

tap_side: NDArray[np.int8]

tap_pos: NDArray[np.int8]

tap_min: NDArray[np.int8]

tap_max: NDArray[np.int8]

tap_nom: NDArray[np.int8]

class power_grid_model_ds.arrays.Branch3Array(*args, data: ndarray[tuple[int, ...], dtype[_ScalarType_co]] | None = None, **kwargs)

Bases: IdArray, Branch3

id: NDArray[np.int32]

node_1: NDArray[np.int32]

node_2: NDArray[np.int32]

node_3: NDArray[np.int32]

status_1: NDArray[np.int8]

status_2: NDArray[np.int8]

status_3: NDArray[np.int8]

class power_grid_model_ds.arrays.ThreeWindingTransformerArray(*args, data: ndarray[tuple[int, ...], dtype[_ScalarType_co]] | None = None, **kwargs)

Bases: Branch3Array, ThreeWindingTransformer

node_1: NDArray[np.int32]

node_2: NDArray[np.int32]

node_3: NDArray[np.int32]

status_1: NDArray[np.int8]

status_2: NDArray[np.int8]

status_3: NDArray[np.int8]

id: NDArray[np.int32]

u1: NDArray[np.float64]

u2: NDArray[np.float64]

u3: NDArray[np.float64]

sn_1: NDArray[np.float64]

sn_2: NDArray[np.float64]

sn_3: NDArray[np.float64]

uk_12: NDArray[np.float64]

uk_13: NDArray[np.float64]

uk_23: NDArray[np.float64]

pk_12: NDArray[np.float64]

pk_13: NDArray[np.float64]

pk_23: NDArray[np.float64]

i0: NDArray[np.float64]

p0: NDArray[np.float64]

winding_1: NDArray[np.int8]

winding_2: NDArray[np.int8]

winding_3: NDArray[np.int8]

clock_12: NDArray[np.int8]

clock_13: NDArray[np.int8]

tap_side: NDArray[np.int8]

tap_pos: NDArray[np.int8]

tap_min: NDArray[np.int8]

tap_max: NDArray[np.int8]

tap_nom: NDArray[np.int8]

tap_size: NDArray[np.float64]

uk_12_min: NDArray[np.float64]

uk_13_min: NDArray[np.float64]

uk_23_min: NDArray[np.float64]

pk_12_min: NDArray[np.float64]

pk_13_min: NDArray[np.float64]

pk_23_min: NDArray[np.float64]

uk_12_max: NDArray[np.float64]

uk_13_max: NDArray[np.float64]

uk_23_max: NDArray[np.float64]

pk_12_max: NDArray[np.float64]

pk_13_max: NDArray[np.float64]

pk_23_max: NDArray[np.float64]

class power_grid_model_ds.arrays.SourceArray(*args, data: ndarray[tuple[int, ...], dtype[_ScalarType_co]] | None = None, **kwargs)

Bases: IdArray, Source

id: NDArray[np.int32]

u_ref: NDArray[np.float64]

node: NDArray[np.int32]

status: NDArray[np.int8]

class power_grid_model_ds.arrays.SymGenArray(*args, data: ndarray[tuple[int, ...], dtype[_ScalarType_co]] | None = None, **kwargs)

Bases: IdArray, SymGen

id: NDArray[np.int32]

type: NDArray[np.int_]

p_specified: NDArray[np.float64]

q_specified: NDArray[np.float64]

node: NDArray[np.int32]

status: NDArray[np.int8]

class power_grid_model_ds.arrays.SymLoadArray(*args, data: ndarray[tuple[int, ...], dtype[_ScalarType_co]] | None = None, **kwargs)

Bases: IdArray, SymLoad

id: NDArray[np.int32]

type: NDArray[np.int8]

p_specified: NDArray[np.float64]

q_specified: NDArray[np.float64]

node: NDArray[np.int32]

status: NDArray[np.int8]

class power_grid_model_ds.arrays.TransformerTapRegulatorArray(*args, data: ndarray[tuple[int, ...], dtype[_ScalarType_co]] | None = None, **kwargs)

Bases: IdArray, TransformerTapRegulator

id: NDArray[np.int32]

control_side: NDArray[np.int8]

u_set: NDArray[np.float64]

u_band: NDArray[np.float64]

line_drop_compensation_r: NDArray[np.float64]

line_drop_compensation_x: NDArray[np.float64]

regulated_object: NDArray[np.int32]

status: NDArray[np.int8]

class power_grid_model_ds.arrays.AsymVoltageSensorArray(*args, data: ndarray[tuple[int, ...], dtype[_ScalarType_co]] | None = None, **kwargs)

Bases: IdArray, AsymVoltageSensor

id: NDArray[np.int32]

u_sigma: NDArray3[np.float64]

u_measured: NDArray3[np.float64]

u_angle_measured: NDArray3[np.float64]

measured_object: NDArray[np.int32]

class power_grid_model_ds.arrays.SymPowerSensorArray(*args, data: ndarray[tuple[int, ...], dtype[_ScalarType_co]] | None = None, **kwargs)

Bases: IdArray, SymPowerSensor

id: NDArray[np.int32]

p_measured: NDArray[np.float64]

q_measured: NDArray[np.float64]

p_sigma: NDArray[np.float64]

q_sigma: NDArray[np.float64]

measured_terminal_type: NDArray[np.int32]

power_sigma: NDArray[np.float64]

measured_object: NDArray[np.int32]

class power_grid_model_ds.arrays.SymVoltageSensorArray(*args, data: ndarray[tuple[int, ...], dtype[_ScalarType_co]] | None = None, **kwargs)

Bases: IdArray, SymVoltageSensor

id: NDArray[np.int32]

u_sigma: NDArray[np.float64]

u_measured: NDArray[np.float64]

u_angle_measured: NDArray[np.float64]

measured_object: NDArray[np.int32]

constants

power_grid_model_ds.constants.empty(dtype: type): Returns the empty value for the given dtype.

enums

class power_grid_model_ds.enums.NodeType(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: IntEnum

Node Types Nodes located within a substation, are marked as Substation nodes.

UNSPECIFIED = 0

SUBSTATION_NODE = 1

errors

exception power_grid_model_ds.errors.PGMCoreException

Bases: Exception

Raised when there is an error in running the power grid model

exception power_grid_model_ds.errors.GraphError

Bases: Exception

Raised when there is an error in the graph

exception power_grid_model_ds.errors.ArrayDefinitionError

Bases: Exception

Raised when an array is defined incorrectly

exception power_grid_model_ds.errors.RecordDoesNotExist

Bases: IndexError

Raised when a record is accessed that does not exist

exception power_grid_model_ds.errors.MultipleRecordsReturned

Bases: IndexError

Raised when unexpectedly, multiple records are found.

exception power_grid_model_ds.errors.MissingNodeError

Bases: GraphError

Raised when a node is missing

exception power_grid_model_ds.errors.MissingBranchError

Bases: GraphError

Raised when a branch is missing

exception power_grid_model_ds.errors.NoPathBetweenNodes

Bases: GraphError

Raised when there is no path between two nodes

fancypy

class power_grid_model_ds.fancypy.FancyArray(*args, data: ndarray[tuple[int, ...], dtype[_ScalarType_co]] | None = None, **kwargs)

Bases: ABC

Base class for all arrays.

You can create your own array by subclassing FancyArray. Array-columns can be defined by adding class attributes with the column name and the numpy dtype.

Example

>>> class MyArray(FancyArray):
>>>     id: NDArray[np.int64]
>>>     name: NDArray[np.str_]
>>>     value: NDArray[np.float64]

Note on string-columns:: The default length for string columns is stored in _DEFAULT_STR_LENGTH. To change this, you can set the _str_lengths class attribute.

Example

>>> class MyArray(FancyArray):
>>>     name: NDArray[np.str_]
>>>     _str_lengths = {"name": 100}

Extra note on string-columns:: Where possible, it is recommended use IntEnum’s instead of string-columns to reduce memory usage.

__init__(*args, data: ndarray[tuple[int, ...], dtype[_ScalarType_co]] | None = None, **kwargs)

property data

classmethod get_defaults() → dict[str, Any]

classmethod get_dtype()

copy(): Return a copy of this array including its data

classmethod zeros(num: int, empty_id: bool = True) → Self

Construct an array filled with zeros.

If ‘empty_id’ is True, the ‘id’ column will be filled with EMPTY_ID values.

classmethod empty(num: int, use_defaults: bool = True) → Self

Construct an array filled with ‘empty’ values.

‘empty’ values differs per dtype:: string: “” (empty string) float: np.nan integer: minimum value

if ‘use_defaults’ is True, the default values will be used instead where possible.

is_empty(column: str) → ndarray[tuple[int, ...], dtype[bool]]: Check if a column is filled with ‘empty’ values.

get_empty_value(column: str) → float | int | str | bool

set_empty(column: str): Set a column to its ‘empty’ value.

property columns: list[str]

property record: Return a named tuple of the first record in the array.

filter(*args: int | Iterable[int] | ndarray, mode_: Literal['AND', 'OR'] = 'AND', **kwargs: Any | list[Any] | ndarray) → Self

exclude(*args: int | Iterable[int] | ndarray, mode_: Literal['AND', 'OR'] = 'AND', **kwargs: Any | list[Any] | ndarray) → Self

get(*args: int | Iterable[int] | ndarray, mode_: Literal['AND', 'OR'] = 'AND', **kwargs: Any | list[Any] | ndarray) → Self

filter_mask(*args: int | Iterable[int] | ndarray, mode_: Literal['AND', 'OR'] = 'AND', **kwargs: Any | list[Any] | ndarray) → ndarray

exclude_mask(*args: int | Iterable[int] | ndarray, mode_: Literal['AND', 'OR'] = 'AND', **kwargs: Any | list[Any] | ndarray) → ndarray

re_order(new_order: _Buffer | _SupportsArray[dtype[Any]] | _NestedSequence[_SupportsArray[dtype[Any]]] | bool | int | float | complex | str | bytes | _NestedSequence[bool | int | float | complex | str | bytes], column: str = 'id') → Self

update_by_id(ids: _Buffer | _SupportsArray[dtype[Any]] | _NestedSequence[_SupportsArray[dtype[Any]]] | bool | int | float | complex | str | bytes | _NestedSequence[bool | int | float | complex | str | bytes], allow_missing: bool = False, **kwargs) → None

get_updated_by_id(ids: _Buffer | _SupportsArray[dtype[Any]] | _NestedSequence[_SupportsArray[dtype[Any]]] | bool | int | float | complex | str | bytes | _NestedSequence[bool | int | float | complex | str | bytes], allow_missing: bool = False, **kwargs) → Self

check_ids(return_duplicates: bool = False) → ndarray[tuple[int, ...], dtype[_ScalarType_co]] | None

as_table(column_width: int | str = 'auto', rows: int = 10) → str

as_df(): Convert to pandas DataFrame

class power_grid_model_ds.fancypy.FancyArrayContainer(_id_counter: int)

Bases: object

Base class for ArrayContainers. Contains general functionality that is nonspecific to the type of array being stored.

property id_counter: Returns the private _id_counter field (as read-only)

classmethod empty() → Self: Create an empty grid

all_arrays(): Returns all arrays in the container.

classmethod find_array_field(array_type: Type[FancyArray]) → Field

Find the Field that holds an array of type array_type.

Parameters:: array_type (type[FancyArray]) – FancyArray subclass.
Raises:: TypeError – if no field with the given array_type is found or if multiple fields are found.
Returns:: a Field instance.

property max_id: int: Returns the max id across all arrays within the container.

check_ids(check_between_arrays: bool = True, check_within_arrays: bool = True) → None

Checks for duplicate id values across all arrays in the container.

Parameters:

check_between_arrays (bool) – whether to check for duplicate ids across arrays
check_within_arrays (bool) – whether to check for duplicate ids within each array

Raises:

ValueError – if duplicates are found.

append(array: FancyArray, check_max_id: bool = True) → None

Append the given asset_array to the corresponding field of ArrayContainer and generate ids.

Parameters:

array (FancyArray) – the asset_array to be appended (e.g. a NodeArray instance).
check_max_id (bool) – whether to check max(array.id) with the id counter

Returns:

None

attach_ids(array: FancyArray) → FancyArray

Generate and attach ids to the given FancyArray. Also updates _id_counter.

Parameters:: array (FancyArray) – the array of which the id column is set.
Returns:: initial array with updated id column.
Return type:: FancyArray

search_for_id(record_id: int) → list[FancyArray]

Attempts to find a record across all id-arrays within the container.

This method is only intended for debugging purposes since it is very inefficient. In normal circumstances you should use get or filter to find records within a specific array.

Parameters:: record_id (int) – the id of the record to be found.
Returns:: a list of arrays that contain the given record_id. Each array within the list contains all records with the given array.
Return type:: list[FancyArray]

__init__(_id_counter: int) → None

power_grid_model_ds.fancypy.concatenate(fancy_array: FancyArray, *other_arrays: FancyArray | ndarray) → FancyArray: Concatenate arrays.

power_grid_model_ds.fancypy.unique(array: FancyArray, **kwargs): Return the unique elements of the array.

power_grid_model_ds.fancypy.sort(array: FancyArray, axis=-1, kind=None, order=None) → FancyArray: Sort the array in-place and return sorted array.

power_grid_model_ds.fancypy.array_equal(array1: FancyArray, array2: FancyArray, equal_nan: bool = True) → bool: Return True if two arrays are equal.

generators

class power_grid_model_ds.generators.RadialGridGenerator(grid_class: ~typing.Type[~power_grid_model_ds._core.model.grids.base.Grid], nr_nodes: int = 100, nr_sources: int = 2, nr_nops: int = 10, graph_model: type[~power_grid_model_ds._core.model.graphs.models.base.BaseGraphModel] = <class 'power_grid_model_ds._core.model.graphs.models.rustworkx.RustworkxGraphModel'>)

Bases: object

Generates a random but structurally correct radial grid with the given specifications

__init__(grid_class: ~typing.Type[~power_grid_model_ds._core.model.grids.base.Grid], nr_nodes: int = 100, nr_sources: int = 2, nr_nops: int = 10, graph_model: type[~power_grid_model_ds._core.model.graphs.models.base.BaseGraphModel] = <class 'power_grid_model_ds._core.model.graphs.models.rustworkx.RustworkxGraphModel'>)

run(seed=None, create_10_3_kv_net: bool = False)

Run the generator to create a random radial grid.

if a seed is provided, this will be used to set rng.

class power_grid_model_ds.generators.NodeGenerator(grid: Grid, seed: int)

Bases: BaseGenerator

Generator for node elements in the grid

run(amount: int, voltage_level: int = 10500): Generate nodes in a grid with two possible load scenarios

class power_grid_model_ds.generators.LineGenerator(grid: Grid, seed: int)

Bases: BaseGenerator

Generator for line elements in the grid

__init__(grid: Grid, seed: int) → None: Initializes generator with grid and amount

run(amount: int, number_of_routes: int | None = None) → LineArray: Generate routes, lines and normally open points (NOPs)

create_routes(number_of_routes: int): Create a number of lines from the substation to unconnected nodes

determine_number_of_routes() → int: Decide on a number of routes based on expected route-size

connect_nodes(): Add a new line between an active and inactive line

create_nop_lines(number_of_nops: int): Create the inactive lines between different routes (Normally Open Points)

set_unconnected_nodes() → None: From a line array and total set of nodes determine which are not yet connected

class power_grid_model_ds.generators.TransformerGenerator(grid: Grid, seed: int)

Bases: BaseGenerator

Generator for tranformer elements in the grid

run(amount: int) → TransformerArray: Generate transformers

class power_grid_model_ds.generators.SourceGenerator(grid: Grid, seed: int)

Bases: BaseGenerator

Generator for source elements in the grid (substations)

run(amount: int) → tuple[NodeArray, SourceArray]: Generate nodes in a grid which are sources (substations)

graph_models

class power_grid_model_ds.graph_models.BaseGraphModel(active_only=False)

Bases: ABC

Base class for graph models

__init__(active_only=False) → None

abstract property nr_nodes: Returns the number of nodes in the graph

abstract property nr_branches: Returns the number of branches in the graph

property all_branches: Generator[tuple[int, int], None, None]: Returns all branches in the graph.

abstract external_to_internal(ext_node_id: int) → int

Convert external node id to internal node id (internal)

Raises:: MissingNodeError – if the external node id does not exist in the graph

abstract internal_to_external(int_node_id: int) → int: Convert internal id (internal) to external node id

abstract property external_ids: list[int]

Return all external node ids

Warning: Depending on graph engine, performance could be slow for large graphs

has_node(node_id: int) → bool: Check if a node exists.

in_branches(node_id: int) → Generator[tuple[int, int], None, None]: Return all branches that have the node as an endpoint.

add_node(ext_node_id: int, raise_on_fail: bool = True) → None: Add a node to the graph.

delete_node(ext_node_id: int, raise_on_fail: bool = True) → None

Remove a node from the graph.

Parameters:

ext_node_id (int) – id of the node to remove
raise_on_fail (bool) – whether to raise an error if the node does not exist. Defaults to True

Raises:

MissingNodeError – if the node does not exist in the graph and raise_on_fail=True

add_node_array(node_array: NodeArray, raise_on_fail: bool = True) → None: Add all nodes in the node array to the graph.

delete_node_array(node_array: NodeArray, raise_on_fail: bool = True) → None: Delete all nodes in node_array from the graph

has_branch(from_ext_node_id: int, to_ext_node_id: int) → bool: Check if a branch exists between two nodes.

add_branch(from_ext_node_id: int, to_ext_node_id: int) → None: Add a new branch to the graph.

delete_branch(from_ext_node_id: int, to_ext_node_id: int, raise_on_fail: bool = True) → None

Remove an existing branch from the graph.

Parameters:

from_ext_node_id – id of the from node
to_ext_node_id – id of the to node
raise_on_fail – whether to raise an error if the branch does not exist

Raises:

MissingBranchError – if branch does not exist in the graph and raise_on_fail=True

add_branch_array(branch_array: BranchArray) → None: Add all branches in the branch array to the graph.

add_branch3_array(branch3_array: Branch3Array) → None: Add all branch3s in the branch3 array to the graph.

delete_branch_array(branch_array: BranchArray, raise_on_fail: bool = True) → None: Delete all branches in branch_array from the graph.

delete_branch3_array(branch3_array: Branch3Array, raise_on_fail: bool = True) → None: Delete all branch3s in the branch3 array from the graph.

tmp_remove_nodes(nodes: list[int]) → Generator

Context manager that temporarily removes nodes and their branches from the graph. .. rubric:: Example

>>> with graph.tmp_remove_nodes([1, 2, 3]):
>>>    assert not graph.has_node(1)
>>> assert graph.has_node(1)

In practice, this is useful when you want to e.g. calculate the shortest path between two nodes without considering certain nodes.

get_shortest_path(ext_start_node_id: int, ext_end_node_id: int) → tuple[list[int], int]

Calculate the shortest path between two nodes

Example

given this graph: [1] - [2] - [3] - [4]

>>> graph.get_shortest_path(1, 4) == [1, 2, 3, 4], 3
>>> graph.get_shortest_path(1, 1) == [1], 0

Returns:: a tuple where the first element is a list of external nodes from start to end. The second element is the distance of the path in number of edges.
Return type:: tuple[list[int], int]
Raises:: NoPathBetweenNodes – if no path exists between the given nodes

get_all_paths(ext_start_node_id: int, ext_end_node_id: int) → list[list[int]]: Retrieves all paths between two (external) nodes. Returns a list of paths, each path containing a list of external nodes.

get_components(substation_nodes: ndarray[tuple[int, ...], dtype[int32]]) → list[list[int]]: Returns all separate components when the substation_nodes are removed of the graph as lists

get_connected(node_id: int, nodes_to_ignore: list[int] | None = None, inclusive: bool = False) → list[int]

Find all nodes connected to the node_id

Parameters:

node_id – node id to start the search from
inclusive – whether to include the given node id in the result
nodes_to_ignore – list of node ids to ignore while traversing the graph. Any nodes connected to node_id (solely) through these nodes will not be included in the result

Returns:

list of node ids sorted by distance, connected to the node id

Return type:

nodes

find_fundamental_cycles() → list[list[int]]: Find all fundamental cycles in the graph. :returns: list of cycles, each cycle is a list of (external) node ids :rtype: list[list[int]]

classmethod from_arrays(arrays: MinimalGridArrays, active_only=False) → BaseGraphModel: Build from arrays

class power_grid_model_ds.graph_models.RustworkxGraphModel(active_only=False)

Bases: BaseGraphModel

A wrapper around the graph from the ‘rustworkx’ package

__init__(active_only=False) → None

property nr_nodes: Returns the number of nodes in the graph

property nr_branches: Returns the number of branches in the graph

property external_ids: list[int]

Return all external node ids

Warning: Depending on graph engine, performance could be slow for large graphs

external_to_internal(ext_node_id: int)

Convert external node id to internal node id (internal)

Raises:: MissingNodeError – if the external node id does not exist in the graph

internal_to_external(int_node_id: int): Convert internal id (internal) to external node id