power_grid_model_ds

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

class power_grid_model_ds.FancyArray(*args, data: ndarray[tuple[Any, ...], dtype[_ScalarT]] | 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 default values:: Default values for columns can be defined by adding a class attribute _defaults, which is a dictionary mapping column names to their default values. The _defaults attribute is inherited by child classes.
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. The _str_lengths attribute is inherited by child classes.
Note on id columns:: Columns that contain ids or reference ids (e.g., ‘id’, ‘from_node’, etc.) should be defined in the _id_columns class attribute to ensure correct functionality of certain grid helper methods (e.g. grid.merge). The _id_columns attribute is inherited and combined with those of the parent classes.

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[Any, ...], dtype[_ScalarT]] | None = None, **kwargs)

property data: ndarray[tuple[Any, ...], dtype[_ScalarT]]

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

classmethod get_id_columns() → set[str]

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[Any, ...], 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.

property size: int

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: ArrayLike, column: str = 'id') → Self

update_by_id(ids: ArrayLike, allow_missing: bool = False, **kwargs) → None

get_updated_by_id(ids: ArrayLike, allow_missing: bool = False, **kwargs) → Self

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

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

as_df(): Convert to pandas DataFrame

classmethod from_extended(extended: Self) → Self: Create an instance from an extended array.

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[Field, None, None]

Get all graph attributes of the container.

Yields:: BaseGraphModel – The next graph attribute.

classmethod empty(graph_model: type[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_node_array(node_array: NodeArray) → None: Add a node to all graphs

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

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

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

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

delete_branch3(branch: Branch3Array) → None: Remove a branch 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: Grid) → GraphContainer: Build from arrays. DEPRECATED: Use .from_grid instead.

classmethod from_grid(grid: Grid) → GraphContainer: Build from grid

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

class power_grid_model_ds.Grid(_ids: set[int], graphs: GraphContainer, node: NodeArray, transformer: TransformerArray, three_winding_transformer: ThreeWindingTransformerArray, line: LineArray, link: LinkArray, generic_branch: GenericBranchArray, asym_line: AsymLineArray, source: SourceArray, sym_load: SymLoadArray, sym_gen: SymGenArray, asym_load: AsymLoadArray, asym_gen: AsymGenArray, shunt: ShuntArray, transformer_tap_regulator: TransformerTapRegulatorArray, voltage_regulator: VoltageRegulatorArray, sym_power_sensor: SymPowerSensorArray, sym_voltage_sensor: SymVoltageSensorArray, sym_current_sensor: SymCurrentSensorArray, asym_power_sensor: AsymPowerSensorArray, asym_voltage_sensor: AsymVoltageSensorArray, asym_current_sensor: AsymCurrentSensorArray, fault: FaultArray)

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

generic_branch: GenericBranchArray

asym_line: AsymLineArray

source: SourceArray

sym_load: SymLoadArray

sym_gen: SymGenArray

asym_load: AsymLoadArray

asym_gen: AsymGenArray

shunt: ShuntArray

transformer_tap_regulator: TransformerTapRegulatorArray

voltage_regulator: VoltageRegulatorArray

sym_power_sensor: SymPowerSensorArray

sym_voltage_sensor: SymVoltageSensorArray

sym_current_sensor: SymCurrentSensorArray

asym_power_sensor: AsymPowerSensorArray

asym_voltage_sensor: AsymVoltageSensorArray

asym_current_sensor: AsymCurrentSensorArray

fault: FaultArray

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

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

classmethod from_txt(*args: str) → G

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) → G

Load grid from txt file

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

classmethod from_extended(extended: G) → G: Create a grid from an extended Grid object.

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

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

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.

delete_branch(branch: BranchArray) → None

Remove a branch array from the grid

Supports removing multiple branches at once. Also removes assets connected to the branch (e.g. sensors) and updates the graphs accordingly.

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

delete_branch3(branch: Branch3Array) → None

Remove a branch3 array from the grid

Supports removing multiple branch3 records at once Also removes assets connected to the branch3 (e.g. sensors) and updates the graphs accordingly.

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

delete_node(node: NodeArray) → None

Remove a node array from the grid

Supports removing multiple branches at once. Also removes assets connected to the node (e.g. branches, sensors, loads) and updates the graphs accordingly.

Parameters:: node (NodeArray) – node array 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)

set_feeder_ids()

Determines all feeder groups in the network and gives them a feeder identifier

feeder_branch_id := All assets being connected through the same feeding branch(es) on the substation feeder_node_id := All assets connected to the same feeder node (substation)

Example: Nodes Topology (SUBSTATION) 101 — 102 — 103 -|- 104 — 105 — 101 (SUBSTATION)

{-} 106

Branches Topology: (SUBSTATION) * 201 * 202 * 203 * 601 * 204 * (SUBSTATION)

Substation ID | Feeder ID

101 | -1 | -1 102 | 101 | 201 103 | 101 | 201 104 | 101 | 204 105 | 101 | 204 106 | 101 | 201 201 | 101 | 201 202 | 101 | 201 203 | 101 | -1 204 | 101 | 204 301 | 101 | 201 601 | 101 | 204

get_typed_branches(branch_ids: list[int] | ndarray[tuple[Any, ...], 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.

get_reversed_branches() → BranchArray

Get the branch array of branches that are oriented towards the source(s).

Orientation is determined by the distance of the branch’s from_node and to_node to the source node. The node that is closer to the source is considered the “from_node”. Note that this might not reflect the actual power flow direction in the grid.

If a source is connected to another source. The orientation will be set from the perspective of the source with the lowest node id.
Parallel edges (multiple edges between the same two nodes) and cycles are supported.
Open branches will be corrected so that the from_status is 1 and the to_status is 0.

Returns:: All branches that are oriented towards the source(s)
Return type:: BranchArray

set_branch_orientations() → BranchArray

Set branch orientations in the grid so that all branches are oriented away from the sources. See also get_reversed_branches() for how the branches to reverse are determined.

Returns:: All branches that were reversed.
Return type:: BranchArray

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. Assuming each node has a single feeding substation and the grid is radial

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]

merge(other_grid: G, mode: Literal['recalculate_ids']) → int

merge(other_grid: G, mode: Literal['keep_ids']) → None

Merge another grid into this grid.

Parameters:

other_grid (Grid) – The grid to merge into this grid.
mode (Literal["keep_ids", "recalculate_ids"]) – The merge mode: - “recalculate_ids”: ids in the arrays of other_grid are offset to avoid conflicts. IMPORTANT: we currently only update any id column and all id references in the default PGM-DS grid. - “keep_ids”: Keep ids of other_grid. Raises an error if grids contain overlapping indices.

Returns:

The offset of the IDs in the merged grid.: If mode is “keep_ids”, the offset will be 0. If mode is “recalculate_ids”, the offset will be the value that was added to the other_grid ids to avoid conflicts.

Return type:

int

__init__(_ids: set[int], graphs: GraphContainer, node: NodeArray, transformer: TransformerArray, three_winding_transformer: ThreeWindingTransformerArray, line: LineArray, link: LinkArray, generic_branch: GenericBranchArray, asym_line: AsymLineArray, source: SourceArray, sym_load: SymLoadArray, sym_gen: SymGenArray, asym_load: AsymLoadArray, asym_gen: AsymGenArray, shunt: ShuntArray, transformer_tap_regulator: TransformerTapRegulatorArray, voltage_regulator: VoltageRegulatorArray, sym_power_sensor: SymPowerSensorArray, sym_voltage_sensor: SymVoltageSensorArray, sym_current_sensor: SymCurrentSensorArray, asym_power_sensor: AsymPowerSensorArray, asym_voltage_sensor: AsymVoltageSensorArray, asym_current_sensor: AsymCurrentSensorArray, fault: FaultArray) → None

serialize(path: Path, mode: Literal['json'] = 'json', **kwargs) → Path

serialize(path: None = None, *, mode: Literal['json_string'], **kwargs) → str

Serialize the grid.

Parameters:

path – Destination file path. Required when mode is "json", ignored otherwise.
mode – Serialization target. Use "json" (default) to write a JSON file, or "json_string" to return a JSON string.
**kwargs – Additional keyword arguments forwarded to json.dump / json.dumps.

Returns:

Path when mode is "json", str when mode is "json_string".

classmethod from_json_string(json_string: str) → Self

Deserialize the grid from a JSON string.

Parameters:: json_string – A JSON string as produced by serialize(mode="json_string").
Returns:: The deserialized grid instance.
Return type:: Self

classmethod deserialize(path: Path) → Self

Deserialize the grid from a JSON file.

Parameters:: path – Path to the JSON file.
Returns:: The deserialized grid instance.
Return type:: Self

rebuild_graphs() → None: (Re)build the graphs in the grid.

diff(other_grid: Self) → None

Print the differences between two grids

Intended for debugging.

Note: Only the content of the arrays is compared. Differences in the ordering within arrays are ignored.

Parameters:: other_grid (Grid) – The grid to compare with.

class power_grid_model_ds.PowerGridModelInterface(grid: Grid | None = None, 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 | None = None, input_data: dict | None = None, system_frequency: float = 50.0)

property input_data: dict[str, ndarray[tuple[Any, ...], dtype[_ScalarT]]]: Get the input data for the PowerGridModel.

create_input_from_grid(): Create input for the PowerGridModel

create_grid_from_input_data(check_ids: bool = True) → Grid

Create Grid object from PowerGridModel input. Note that for some arrays, not all fields are available in the PowerGridModel input. In this case, the default values are used.

Parameters:: check_ids – if True, check if the ids are unique

Returns a Grid object with the arrays filled with the PowerGridModel input.

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

setup_model(): Setup the PowerGridModel with the input data.

arrays

Arrays

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

Bases: Id, FancyArray

id: ndarray[tuple[Any, ...], dtype[int32]]

class power_grid_model_ds.arrays.ApplianceArray(*args, data: ndarray[tuple[Any, ...], dtype[_ScalarT]] | None = None, **kwargs)

Bases: IdArray, Appliance

node: ndarray[tuple[Any, ...], dtype[int32]]

status: ndarray[tuple[Any, ...], dtype[int8]]

id: ndarray[tuple[Any, ...], dtype[int32]]

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

Bases: ApplianceArray, SymLoad

type: ndarray[tuple[Any, ...], dtype[int8]]

p_specified: ndarray[tuple[Any, ...], dtype[float64]]

q_specified: ndarray[tuple[Any, ...], dtype[float64]]

node: ndarray[tuple[Any, ...], dtype[int32]]

status: ndarray[tuple[Any, ...], dtype[int8]]

id: ndarray[tuple[Any, ...], dtype[int32]]

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

Bases: ApplianceArray, SymGen

type: ndarray[tuple[Any, ...], dtype[int8]]

p_specified: ndarray[tuple[Any, ...], dtype[float64]]

q_specified: ndarray[tuple[Any, ...], dtype[float64]]

node: ndarray[tuple[Any, ...], dtype[int32]]

status: ndarray[tuple[Any, ...], dtype[int8]]

id: ndarray[tuple[Any, ...], dtype[int32]]

class power_grid_model_ds.arrays.AsymLoadArray(*args, data: ndarray[tuple[Any, ...], dtype[_ScalarT]] | None = None, **kwargs)

Bases: ApplianceArray, AsymLoad

type: ndarray[tuple[Any, ...], dtype[int8]]

p_specified: Annotated[ndarray[tuple[Any, ...], dtype[float64]], Literal[3]]

q_specified: Annotated[ndarray[tuple[Any, ...], dtype[float64]], Literal[3]]

node: ndarray[tuple[Any, ...], dtype[int32]]

status: ndarray[tuple[Any, ...], dtype[int8]]

id: ndarray[tuple[Any, ...], dtype[int32]]

class power_grid_model_ds.arrays.AsymGenArray(*args, data: ndarray[tuple[Any, ...], dtype[_ScalarT]] | None = None, **kwargs)

Bases: ApplianceArray, AsymGen

type: ndarray[tuple[Any, ...], dtype[int8]]

p_specified: Annotated[ndarray[tuple[Any, ...], dtype[float64]], Literal[3]]

q_specified: Annotated[ndarray[tuple[Any, ...], dtype[float64]], Literal[3]]

node: ndarray[tuple[Any, ...], dtype[int32]]

status: ndarray[tuple[Any, ...], dtype[int8]]

id: ndarray[tuple[Any, ...], dtype[int32]]

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

Bases: ApplianceArray, Source

u_ref: ndarray[tuple[Any, ...], dtype[float64]]

node: ndarray[tuple[Any, ...], dtype[int32]]

status: ndarray[tuple[Any, ...], dtype[int8]]

id: ndarray[tuple[Any, ...], dtype[int32]]

class power_grid_model_ds.arrays.ShuntArray(*args, data: ndarray[tuple[Any, ...], dtype[_ScalarT]] | None = None, **kwargs)

Bases: ApplianceArray, Shunt

g1: ndarray[tuple[Any, ...], dtype[float64]]

b1: ndarray[tuple[Any, ...], dtype[float64]]

node: ndarray[tuple[Any, ...], dtype[int32]]

status: ndarray[tuple[Any, ...], dtype[int8]]

id: ndarray[tuple[Any, ...], dtype[int32]]

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

Bases: IdArray, Node

u_rated: ndarray[tuple[Any, ...], dtype[float64]]

node_type: ndarray[tuple[Any, ...], dtype[int8]]

feeder_branch_id: ndarray[tuple[Any, ...], dtype[int32]]

feeder_node_id: ndarray[tuple[Any, ...], dtype[int32]]

id: ndarray[tuple[Any, ...], dtype[int32]]

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

Bases: IdArray, Branch

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

property is_active: ndarray[tuple[Any, ...], 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.

from_node: ndarray[tuple[Any, ...], dtype[int32]]

to_node: ndarray[tuple[Any, ...], dtype[int32]]

from_status: ndarray[tuple[Any, ...], dtype[int8]]

to_status: ndarray[tuple[Any, ...], dtype[int8]]

feeder_branch_id: ndarray[tuple[Any, ...], dtype[int32]]

feeder_node_id: ndarray[tuple[Any, ...], dtype[int32]]

is_feeder: ndarray[tuple[Any, ...], dtype[bool]]

id: ndarray[tuple[Any, ...], dtype[int32]]

class power_grid_model_ds.arrays.LinkArray(*args, data: ndarray[tuple[Any, ...], dtype[_ScalarT]] | None = None, **kwargs)

Bases: Link, BranchArray

from_node: ndarray[tuple[Any, ...], dtype[int32]]

to_node: ndarray[tuple[Any, ...], dtype[int32]]

from_status: ndarray[tuple[Any, ...], dtype[int8]]

to_status: ndarray[tuple[Any, ...], dtype[int8]]

feeder_branch_id: ndarray[tuple[Any, ...], dtype[int32]]

feeder_node_id: ndarray[tuple[Any, ...], dtype[int32]]

is_feeder: ndarray[tuple[Any, ...], dtype[bool]]

id: ndarray[tuple[Any, ...], dtype[int32]]

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

Bases: Line, BranchArray

r1: ndarray[tuple[Any, ...], dtype[float64]]

x1: ndarray[tuple[Any, ...], dtype[float64]]

c1: ndarray[tuple[Any, ...], dtype[float64]]

tan1: ndarray[tuple[Any, ...], dtype[float64]]

i_n: ndarray[tuple[Any, ...], dtype[float64]]

from_node: ndarray[tuple[Any, ...], dtype[int32]]

to_node: ndarray[tuple[Any, ...], dtype[int32]]

from_status: ndarray[tuple[Any, ...], dtype[int8]]

to_status: ndarray[tuple[Any, ...], dtype[int8]]

feeder_branch_id: ndarray[tuple[Any, ...], dtype[int32]]

feeder_node_id: ndarray[tuple[Any, ...], dtype[int32]]

is_feeder: ndarray[tuple[Any, ...], dtype[bool]]

id: ndarray[tuple[Any, ...], dtype[int32]]

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

Bases: Transformer, BranchArray

u1: ndarray[tuple[Any, ...], dtype[float64]]

u2: ndarray[tuple[Any, ...], dtype[float64]]

sn: ndarray[tuple[Any, ...], dtype[float64]]

tap_size: ndarray[tuple[Any, ...], dtype[float64]]

uk: ndarray[tuple[Any, ...], dtype[float64]]

pk: ndarray[tuple[Any, ...], dtype[float64]]

i0: ndarray[tuple[Any, ...], dtype[float64]]

p0: ndarray[tuple[Any, ...], dtype[float64]]

winding_from: ndarray[tuple[Any, ...], dtype[int8]]

winding_to: ndarray[tuple[Any, ...], dtype[int8]]

clock: ndarray[tuple[Any, ...], dtype[int8]]

tap_side: ndarray[tuple[Any, ...], dtype[int8]]

tap_pos: ndarray[tuple[Any, ...], dtype[int8]]

tap_min: ndarray[tuple[Any, ...], dtype[int8]]

tap_max: ndarray[tuple[Any, ...], dtype[int8]]

tap_nom: ndarray[tuple[Any, ...], dtype[int8]]

from_node: ndarray[tuple[Any, ...], dtype[int32]]

to_node: ndarray[tuple[Any, ...], dtype[int32]]

from_status: ndarray[tuple[Any, ...], dtype[int8]]

to_status: ndarray[tuple[Any, ...], dtype[int8]]

feeder_branch_id: ndarray[tuple[Any, ...], dtype[int32]]

feeder_node_id: ndarray[tuple[Any, ...], dtype[int32]]

is_feeder: ndarray[tuple[Any, ...], dtype[bool]]

id: ndarray[tuple[Any, ...], dtype[int32]]

class power_grid_model_ds.arrays.GenericBranchArray(*args, data: ndarray[tuple[Any, ...], dtype[_ScalarT]] | None = None, **kwargs)

Bases: GenericBranch, BranchArray

r1: ndarray[tuple[Any, ...], dtype[float64]]

x1: ndarray[tuple[Any, ...], dtype[float64]]

g1: ndarray[tuple[Any, ...], dtype[float64]]

b1: ndarray[tuple[Any, ...], dtype[float64]]

from_node: ndarray[tuple[Any, ...], dtype[int32]]

to_node: ndarray[tuple[Any, ...], dtype[int32]]

from_status: ndarray[tuple[Any, ...], dtype[int8]]

to_status: ndarray[tuple[Any, ...], dtype[int8]]

feeder_branch_id: ndarray[tuple[Any, ...], dtype[int32]]

feeder_node_id: ndarray[tuple[Any, ...], dtype[int32]]

is_feeder: ndarray[tuple[Any, ...], dtype[bool]]

id: ndarray[tuple[Any, ...], dtype[int32]]

class power_grid_model_ds.arrays.AsymLineArray(*args, data: ndarray[tuple[Any, ...], dtype[_ScalarT]] | None = None, **kwargs)

Bases: AsymLine, BranchArray

r_aa: ndarray[tuple[Any, ...], dtype[float64]]

r_ba: ndarray[tuple[Any, ...], dtype[float64]]

r_bb: ndarray[tuple[Any, ...], dtype[float64]]

r_ca: ndarray[tuple[Any, ...], dtype[float64]]

r_cb: ndarray[tuple[Any, ...], dtype[float64]]

r_cc: ndarray[tuple[Any, ...], dtype[float64]]

x_aa: ndarray[tuple[Any, ...], dtype[float64]]

x_ba: ndarray[tuple[Any, ...], dtype[float64]]

x_bb: ndarray[tuple[Any, ...], dtype[float64]]

x_ca: ndarray[tuple[Any, ...], dtype[float64]]

x_cb: ndarray[tuple[Any, ...], dtype[float64]]

x_cc: ndarray[tuple[Any, ...], dtype[float64]]

c0: ndarray[tuple[Any, ...], dtype[float64]]

c1: ndarray[tuple[Any, ...], dtype[float64]]

i_n: ndarray[tuple[Any, ...], dtype[float64]]

from_node: ndarray[tuple[Any, ...], dtype[int32]]

to_node: ndarray[tuple[Any, ...], dtype[int32]]

from_status: ndarray[tuple[Any, ...], dtype[int8]]

to_status: ndarray[tuple[Any, ...], dtype[int8]]

feeder_branch_id: ndarray[tuple[Any, ...], dtype[int32]]

feeder_node_id: ndarray[tuple[Any, ...], dtype[int32]]

is_feeder: ndarray[tuple[Any, ...], dtype[bool]]

id: ndarray[tuple[Any, ...], dtype[int32]]

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

Bases: IdArray, Branch3

as_branches() → BranchArray: Convert Branch3Array to BranchArray.

node_1: ndarray[tuple[Any, ...], dtype[int32]]

node_2: ndarray[tuple[Any, ...], dtype[int32]]

node_3: ndarray[tuple[Any, ...], dtype[int32]]

status_1: ndarray[tuple[Any, ...], dtype[int8]]

status_2: ndarray[tuple[Any, ...], dtype[int8]]

status_3: ndarray[tuple[Any, ...], dtype[int8]]

id: ndarray[tuple[Any, ...], dtype[int32]]

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

Bases: Branch3Array, ThreeWindingTransformer

as_branches() → BranchArray: Convert Branch3Array to BranchArray.

u1: ndarray[tuple[Any, ...], dtype[float64]]

u2: ndarray[tuple[Any, ...], dtype[float64]]

u3: ndarray[tuple[Any, ...], dtype[float64]]

sn_1: ndarray[tuple[Any, ...], dtype[float64]]

sn_2: ndarray[tuple[Any, ...], dtype[float64]]

sn_3: ndarray[tuple[Any, ...], dtype[float64]]

uk_12: ndarray[tuple[Any, ...], dtype[float64]]

uk_13: ndarray[tuple[Any, ...], dtype[float64]]

uk_23: ndarray[tuple[Any, ...], dtype[float64]]

pk_12: ndarray[tuple[Any, ...], dtype[float64]]

pk_13: ndarray[tuple[Any, ...], dtype[float64]]

pk_23: ndarray[tuple[Any, ...], dtype[float64]]

i0: ndarray[tuple[Any, ...], dtype[float64]]

p0: ndarray[tuple[Any, ...], dtype[float64]]

winding_1: ndarray[tuple[Any, ...], dtype[int8]]

winding_2: ndarray[tuple[Any, ...], dtype[int8]]

winding_3: ndarray[tuple[Any, ...], dtype[int8]]

clock_12: ndarray[tuple[Any, ...], dtype[int8]]

clock_13: ndarray[tuple[Any, ...], dtype[int8]]

tap_side: ndarray[tuple[Any, ...], dtype[int8]]

tap_pos: ndarray[tuple[Any, ...], dtype[int8]]

tap_min: ndarray[tuple[Any, ...], dtype[int8]]

tap_max: ndarray[tuple[Any, ...], dtype[int8]]

tap_nom: ndarray[tuple[Any, ...], dtype[int8]]

tap_size: ndarray[tuple[Any, ...], dtype[float64]]

uk_12_min: ndarray[tuple[Any, ...], dtype[float64]]

uk_13_min: ndarray[tuple[Any, ...], dtype[float64]]

uk_23_min: ndarray[tuple[Any, ...], dtype[float64]]

pk_12_min: ndarray[tuple[Any, ...], dtype[float64]]

pk_13_min: ndarray[tuple[Any, ...], dtype[float64]]

pk_23_min: ndarray[tuple[Any, ...], dtype[float64]]

uk_12_max: ndarray[tuple[Any, ...], dtype[float64]]

uk_13_max: ndarray[tuple[Any, ...], dtype[float64]]

uk_23_max: ndarray[tuple[Any, ...], dtype[float64]]

pk_12_max: ndarray[tuple[Any, ...], dtype[float64]]

pk_13_max: ndarray[tuple[Any, ...], dtype[float64]]

pk_23_max: ndarray[tuple[Any, ...], dtype[float64]]

node_1: ndarray[tuple[Any, ...], dtype[int32]]

node_2: ndarray[tuple[Any, ...], dtype[int32]]

node_3: ndarray[tuple[Any, ...], dtype[int32]]

status_1: ndarray[tuple[Any, ...], dtype[int8]]

status_2: ndarray[tuple[Any, ...], dtype[int8]]

status_3: ndarray[tuple[Any, ...], dtype[int8]]

id: ndarray[tuple[Any, ...], dtype[int32]]

class power_grid_model_ds.arrays.VoltageRegulatorArray(*args, data: ndarray[tuple[Any, ...], dtype[_ScalarT]] | None = None, **kwargs)

Bases: IdArray, VoltageRegulator

u_ref: ndarray[tuple[Any, ...], dtype[float64]]

regulated_object: ndarray[tuple[Any, ...], dtype[int32]]

status: ndarray[tuple[Any, ...], dtype[int8]]

id: ndarray[tuple[Any, ...], dtype[int32]]

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

Bases: IdArray, TransformerTapRegulator

control_side: ndarray[tuple[Any, ...], dtype[int8]]

u_set: ndarray[tuple[Any, ...], dtype[float64]]

u_band: ndarray[tuple[Any, ...], dtype[float64]]

line_drop_compensation_r: ndarray[tuple[Any, ...], dtype[float64]]

line_drop_compensation_x: ndarray[tuple[Any, ...], dtype[float64]]

regulated_object: ndarray[tuple[Any, ...], dtype[int32]]

status: ndarray[tuple[Any, ...], dtype[int8]]

id: ndarray[tuple[Any, ...], dtype[int32]]

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

Bases: IdArray, SymPowerSensor

p_measured: ndarray[tuple[Any, ...], dtype[float64]]

q_measured: ndarray[tuple[Any, ...], dtype[float64]]

p_sigma: ndarray[tuple[Any, ...], dtype[float64]]

q_sigma: ndarray[tuple[Any, ...], dtype[float64]]

measured_terminal_type: ndarray[tuple[Any, ...], dtype[int8]]

power_sigma: ndarray[tuple[Any, ...], dtype[float64]]

measured_object: ndarray[tuple[Any, ...], dtype[int32]]

id: ndarray[tuple[Any, ...], dtype[int32]]

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

Bases: IdArray, SymVoltageSensor

u_sigma: ndarray[tuple[Any, ...], dtype[float64]]

u_measured: ndarray[tuple[Any, ...], dtype[float64]]

u_angle_measured: ndarray[tuple[Any, ...], dtype[float64]]

measured_object: ndarray[tuple[Any, ...], dtype[int32]]

id: ndarray[tuple[Any, ...], dtype[int32]]

class power_grid_model_ds.arrays.SymCurrentSensorArray(*args, data: ndarray[tuple[Any, ...], dtype[_ScalarT]] | None = None, **kwargs)

Bases: IdArray, SymCurrentSensor

i_measured: ndarray[tuple[Any, ...], dtype[float64]]

i_angle_measured: ndarray[tuple[Any, ...], dtype[float64]]

measured_terminal_type: ndarray[tuple[Any, ...], dtype[int8]]

angle_measurement_type: ndarray[tuple[Any, ...], dtype[int8]]

i_sigma: ndarray[tuple[Any, ...], dtype[float64]]

i_angle_sigma: ndarray[tuple[Any, ...], dtype[float64]]

measured_object: ndarray[tuple[Any, ...], dtype[int32]]

id: ndarray[tuple[Any, ...], dtype[int32]]

class power_grid_model_ds.arrays.AsymPowerSensorArray(*args, data: ndarray[tuple[Any, ...], dtype[_ScalarT]] | None = None, **kwargs)

Bases: IdArray, AsymPowerSensor

p_measured: Annotated[ndarray[tuple[Any, ...], dtype[float64]], Literal[3]]

q_measured: Annotated[ndarray[tuple[Any, ...], dtype[float64]], Literal[3]]

p_sigma: Annotated[ndarray[tuple[Any, ...], dtype[float64]], Literal[3]]

q_sigma: Annotated[ndarray[tuple[Any, ...], dtype[float64]], Literal[3]]

measured_terminal_type: ndarray[tuple[Any, ...], dtype[int8]]

power_sigma: ndarray[tuple[Any, ...], dtype[float64]]

measured_object: ndarray[tuple[Any, ...], dtype[int32]]

id: ndarray[tuple[Any, ...], dtype[int32]]

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

Bases: IdArray, AsymVoltageSensor

u_sigma: ndarray[tuple[Any, ...], dtype[float64]]

u_measured: Annotated[ndarray[tuple[Any, ...], dtype[float64]], Literal[3]]

u_angle_measured: Annotated[ndarray[tuple[Any, ...], dtype[float64]], Literal[3]]

measured_object: ndarray[tuple[Any, ...], dtype[int32]]

id: ndarray[tuple[Any, ...], dtype[int32]]

class power_grid_model_ds.arrays.AsymCurrentSensorArray(*args, data: ndarray[tuple[Any, ...], dtype[_ScalarT]] | None = None, **kwargs)

Bases: IdArray, AsymCurrentSensor

i_measured: Annotated[ndarray[tuple[Any, ...], dtype[float64]], Literal[3]]

i_angle_measured: Annotated[ndarray[tuple[Any, ...], dtype[float64]], Literal[3]]

measured_terminal_type: ndarray[tuple[Any, ...], dtype[int8]]

angle_measurement_type: ndarray[tuple[Any, ...], dtype[int8]]

i_sigma: ndarray[tuple[Any, ...], dtype[float64]]

i_angle_sigma: ndarray[tuple[Any, ...], dtype[float64]]

measured_object: ndarray[tuple[Any, ...], dtype[int32]]

id: ndarray[tuple[Any, ...], dtype[int32]]

class power_grid_model_ds.arrays.FaultArray(*args, data: ndarray[tuple[Any, ...], dtype[_ScalarT]] | None = None, **kwargs)

Bases: IdArray, Fault

fault_object: ndarray[tuple[Any, ...], dtype[int32]]

status: ndarray[tuple[Any, ...], dtype[int8]]

fault_type: ndarray[tuple[Any, ...], dtype[int8]]

fault_phase: ndarray[tuple[Any, ...], dtype[int8]]

r_f: ndarray[tuple[Any, ...], dtype[float64]]

x_f: ndarray[tuple[Any, ...], dtype[float64]]

id: ndarray[tuple[Any, ...], dtype[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(*values)

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.ArrayDefinitionError

Bases: Exception

Raised when an array is defined incorrectly

exception power_grid_model_ds.errors.GraphError

Bases: Exception

Raised when there is an error in the graph

exception power_grid_model_ds.errors.MissingBranchError

Bases: GraphError

Raised when a branch is missing

exception power_grid_model_ds.errors.MissingNodeError

Bases: GraphError

Raised when a node is missing

exception power_grid_model_ds.errors.MultipleRecordsReturned

Bases: IndexError

Raised when unexpectedly, multiple records are found.

exception power_grid_model_ds.errors.NoPathBetweenNodes

Bases: GraphError

Raised when there is no path between two nodes

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.RecordDoesNotExist

Bases: IndexError

Raised when a record is accessed that does not exist

fancypy

class power_grid_model_ds.fancypy.FancyArray(*args, data: ndarray[tuple[Any, ...], dtype[_ScalarT]] | 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 default values:: Default values for columns can be defined by adding a class attribute _defaults, which is a dictionary mapping column names to their default values. The _defaults attribute is inherited by child classes.
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. The _str_lengths attribute is inherited by child classes.
Note on id columns:: Columns that contain ids or reference ids (e.g., ‘id’, ‘from_node’, etc.) should be defined in the _id_columns class attribute to ensure correct functionality of certain grid helper methods (e.g. grid.merge). The _id_columns attribute is inherited and combined with those of the parent classes.

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[Any, ...], dtype[_ScalarT]] | None = None, **kwargs)

property data: ndarray[tuple[Any, ...], dtype[_ScalarT]]

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

classmethod get_id_columns() → set[str]

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[Any, ...], 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.

property size: int

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: ArrayLike, column: str = 'id') → Self

update_by_id(ids: ArrayLike, allow_missing: bool = False, **kwargs) → None

get_updated_by_id(ids: ArrayLike, allow_missing: bool = False, **kwargs) → Self

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

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

as_df(): Convert to pandas DataFrame

classmethod from_extended(extended: Self) → Self: Create an instance from an extended array.

class power_grid_model_ds.fancypy.FancyArrayContainer(_ids: set[int])

Bases: object

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

property ids: Returns the ids across all arrays

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

rebuild_ids() → None

Rebuild self._ids based on the arrays in the container.

Raises:: ValueError – if duplicate ids are found between or within arrays.

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.

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

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

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 _ids.

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]

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.

__init__(_ids: set[int]) → None

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

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

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

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

generators

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.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.RadialGridGenerator(grid_class: type[T], 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: Generic

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

__init__(grid_class: type[T], 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) → T

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

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

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: int: Returns the number of nodes in the graph

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

abstractmethod has_parallel_edges() → bool: Check if the graph has parallel edges (multiple edges between the same two nodes)

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

abstractmethod 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

abstractmethod 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.

tmp_remove_branches(branches: list[tuple[int, int]]) → Generator

Context manager that temporarily removes branches from the graph.

Example

>>> with graph.tmp_remove_branches([(1, 2), (2, 3)]):
>>>    assert not graph.has_branch(1, 2)
>>>    assert not graph.has_branch(2, 3)
>>> assert graph.has_branch(1, 2)
>>> assert graph.has_branch(2, 3)

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() → list[list[int]]

Returns all separate components of the graph as lists

If you want to get the components of the graph without certain nodes, use the tmp_remove_nodes context manager.

Example: >>> with graph.tmp_remove_nodes(substation_nodes): >>> components = graph.get_components()

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_first_connected(node_id: int, candidate_node_ids: list[int]) → int

Find the first connected node to the node_id from the candidate_node_ids

Note

If multiple candidate nodes are connected to the node, the first one found is returned. There is no guarantee that the same candidate node will be returned each time.

Raises:

MissingNodeError – if no connected node is found
ValueError – if the node_id is in candidate_node_ids

get_downstream_nodes(node_id: int, start_node_ids: list[int], inclusive: bool = False) → list[int]

Find all nodes downstream of the node_id with respect to the start_node_ids

Example

given this graph: [1] - [2] - [3] - [4] >>> graph.get_downstream_nodes(2, [1]) == [3, 4] >>> graph.get_downstream_nodes(2, [1], inclusive=True) == [2, 3, 4]

Parameters:

node_id – node id to start the search from
start_node_ids – list of node ids considered ‘above’ the node_id
inclusive – whether to include the given node id in the result

Returns:

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

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: Grid, active_only=False) → BaseGraphModel: Build from arrays. DEPRECATED: Use .from_grid instead.

classmethod from_grid(grid: Grid, active_only=False) → BaseGraphModel: Build from grid.

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: int: Returns the number of nodes in the graph

property nr_branches: int: 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

has_parallel_edges() → bool: Check if the graph has parallel edges (multiple edges between the same two nodes)

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