Interfaces

toop_engine_interfaces

toop_engine_interfaces.folder_structure

Defines constants for the folder structure.

File: folder_structure.py Author: Benjamin Petrick Created: 2024-09-12

PREPROCESSING_PATHS module-attribute

PREPROCESSING_PATHS = {
    "grid_file_path_powsybl": "grid.xiidm",
    "grid_file_path_pandapower": "grid.json",
    "masks_path": "masks",
    "static_information_file_path": "static_information.hdf5",
    "importer_auxiliary_file_path": "importer_auxiliary_data.json",
    "initial_topology_path": "initial_topology",
    "LF_CA_path": "initial_topology/LF_CA",
    "single_line_diagram_path": "initial_topology/single_line_diagram",
    "asset_topology_file_path": "initial_topology/asset_topology.json",
    "asset_topology_runtime_file_path": "initial_topology/asset_topology_runtime.json",
    "asset_topology_master_file_path": "initial_topology/asset_topology_master_data.json",
    "asset_topology_master_data_file_path": "initial_topology/asset_topology_master_data.json",
    "original_gridfile_path": "initial_topology/original_gridfile",
    "logs_path": "logs",
    "start_datetime_info_file_path": "logs/start_datetime.info",
    "chronics_path": "chronics",
    "action_set_file_path": "action_set.json",
    "action_set_diff_path": "action_set_diffs.hdf5",
    "nminus1_definition_file_path": "nminus1_definition.json",
    "ignore_file_path": "ignore_elements.csv",
    "contingency_list_file_path": "contingency_list.csv",
    "static_information_stats_file_path": "static_information_stats.json",
    "loadflow_parameters_file_path": "loadflow_parameters.json",
}

POSTPROCESSING_PATHS module-attribute

POSTPROCESSING_PATHS = {
    "optimizer_snapshots_path": "optimizer_snapshots",
    "dc_optimizer_snapshots_path": "optimizer_snapshots/dc",
    "ac_optimizer_snapshots_path": "optimizer_snapshots/ac",
    "dc_plus_optimizer_snapshots_path": "optimizer_snapshots/dc_plus",
    "LF_CA_ac_path": "optimizer_snapshots/ac/LF_CA",
    "single_line_diagram_ac_path": "optimizer_snapshots/ac/single_line_diagram",
    "logs_path": "logs",
    "orao_summary": "orao_summary",
}

NETWORK_MASK_NAMES module-attribute

NETWORK_MASK_NAMES = {
    "relevant_subs": "relevant_subs.npy",
    "line_for_nminus1": "line_for_nminus1.npy",
    "line_for_reward": "line_for_reward.npy",
    "line_overload_weight": "line_overload_weight.npy",
    "line_disconnectable": "line_disconnectable.npy",
    "line_tso_border": "line_tso_border.npy",
    "line_blacklisted": "line_blacklisted.npy",
    "trafo_for_nminus1": "trafo_for_nminus1.npy",
    "trafo_for_reward": "trafo_for_reward.npy",
    "trafo_overload_weight": "trafo_overload_weight.npy",
    "trafo_disconnectable": "trafo_disconnectable.npy",
    "trafo_controllable": "trafo_controllable.npy",
    "trafo_dso_border": "trafo_dso_border.npy",
    "trafo_n0_n1_max_diff_factor": "trafo_n0_n1_max_diff_factor.npy",
    "trafo_blacklisted": "trafo_blacklisted.npy",
    "trafo3w_for_nminus1": "trafo3w_for_nminus1.npy",
    "trafo3w_for_reward": "trafo3w_for_reward.npy",
    "trafo3w_overload_weight": "trafo3w_overload_weight.npy",
    "trafo3w_disconnectable": "trafo3w_disconnectable.npy",
    "trafo3w_n0_n1_max_diff_factor": "trafo3w_n0_n1_max_diff_factor.npy",
    "tie_line_for_reward": "tie_line_for_reward.npy",
    "tie_line_for_nminus1": "tie_line_for_nminus1.npy",
    "tie_line_overload_weight": "tie_line_overload_weight.npy",
    "tie_line_disconnectable": "tie_line_disconnectable.npy",
    "tie_line_tso_border": "tie_line_tso_border.npy",
    "boundary_line_for_nminus1": "boundary_line_for_nminus1.npy",
    "generator_for_nminus1": "generator_for_nminus1.npy",
    "load_for_nminus1": "load_for_nminus1.npy",
    "switch_for_nminus1": "switch_for_nminus1.npy",
    "switch_for_reward": "switch_for_reward.npy",
    "cross_coupler_limits": "cross_coupler_limits.npy",
    "sgen_for_nminus1": "sgen_for_nminus1.npy",
    "busbar_for_nminus1": "busbar_for_nminus1.npy",
}

OUTPUT_FILE_NAMES module-attribute

OUTPUT_FILE_NAMES = {
    "multiple_topologies": "repertoire.json",
    "realized_asset_topology": "asset_topology.json",
    "postprocessed_topology": "topology.json",
    "loadflows_ac": "loadflows_ac.hdf5",
    "loadflows_dc": "loadflows_dc.hdf5",
    "loadflows_ac_cross_coupler": "loadflows_ac_cross_coupler.hdf5",
    "loadflows_dc_cross_coupler": "loadflows_dc_cross_coupler.hdf5",
}

CHRONICS_FILE_NAMES module-attribute

CHRONICS_FILE_NAMES = {
    "load_p": "load_p.npy",
    "gen_p": "gen_p.npy",
    "sgen_p": "sgen_p.npy",
    "dcline_p": "dcline_p.npy",
}

toop_engine_interfaces.backend

The abstract interface definition for accessing data from pandapower/powerfactory/...

BackendInterface

Bases: ABC

Interface for the backend.

The task of this interface is to provide routines for accessing data from the grid modelling software (pandapower/powerfactory/...)

Specifically not task of this interface is to perform any validations or processing of the data

This assume a node-branch model, hence busbars would be nodes and lines, trafos, etc would be branches. Injections inject onto a node and represent both generators, loads, sgens, ...

get_ptdf

get_ptdf()

Get the PTDF matrix, if it was computed already

For the relevant substations it is important that only node A is given as a column in the reference topology. This is to ensure node A and B are treated properly by the algorithm.

If None is returned, the PTDF matrix will be computed by the solver based on from_node, to_node and susceptance.

RETURNS	DESCRIPTION
Float[ndarray, ' n_branch n_node']	The unextended PTDF matrix, not including second nodes for the relevant substations, and not including the PSDF.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/backend.py

def get_ptdf(self) -> Optional[Float[np.ndarray, " n_branch n_node"]]:
    """Get the PTDF matrix, if it was computed already

    For the relevant substations it is important that only node A is given
    as a column in the reference topology. This is to ensure node A and B
    are treated properly by the algorithm.

    If None is returned, the PTDF matrix will be computed by the solver based on
    from_node, to_node and susceptance.

    Returns
    -------
    Float[np.ndarray, " n_branch n_node"]
        The unextended PTDF matrix, not including second nodes for the relevant
        substations, and not including the PSDF.
    """
    return None

get_psdf

get_psdf()

Get the PSDF matrix, if it was computed already

If None is returned, the PSDF matrix will be computed by the solver based on shift_angle and susceptance.

This refers to the already reduced PSDF matrix, i.e. without elements that will never have a shift angle. See get_phase_shifters for more information.

RETURNS	DESCRIPTION
Float[ndarray, ' n_branch n_phaseshifters']	The PSDF matrix, not including the PTDF.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/backend.py

def get_psdf(self) -> Optional[Float[np.ndarray, " n_branch n_phaseshifters"]]:
    """Get the PSDF matrix, if it was computed already

    If None is returned, the PSDF matrix will be computed by the solver based on
    shift_angle and susceptance.

    This refers to the already reduced PSDF matrix, i.e. without elements that will
    never have a shift angle. See get_phase_shifters for more information.

    Returns
    -------
    Float[np.ndarray, " n_branch n_phaseshifters"]
        The PSDF matrix, not including the PTDF.
    """
    return None

get_slack abstractmethod

get_slack()

Get the index of the slack node

Note that the solver does not support distributed slack nodes, if you have a distributed slack, replace all but one slack node by their injections or create a virtual slack node that is connected with same-impendance lines to the other slack nodes.

RETURNS	DESCRIPTION
int	The index of the slack node

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/backend.py

@abstractmethod
def get_slack(self) -> int:
    """Get the index of the slack node

    Note that the solver does not support distributed slack nodes, if you have a
    distributed slack, replace all but one slack node by their injections or create
    a virtual slack node that is connected with same-impendance lines to the other
    slack nodes.

    Returns
    -------
    int
        The index of the slack node
    """

get_ac_dc_mismatch

get_ac_dc_mismatch()

Get the AC-DC mismatch for each branch

This is the difference between the AC and DC flow on each branch, i.e. the difference between the AC and DC loadflow results.

This is used in the solver to adjust the DC flow to match the AC flow in the N-0 case. If all zeros are returned, the solver will return pure DC flows.

Positive values mean the AC flow is higher than the DC flow, negative values mean the AC flow is lower than the DC flow.

RETURNS	DESCRIPTION
Float[ndarray, ' n_timestep n_branch']	The AC-DC mismatch for each branch and per timestep

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/backend.py

def get_ac_dc_mismatch(self) -> Float[np.ndarray, " n_timestep n_branch"]:
    """Get the AC-DC mismatch for each branch

    This is the difference between the AC and DC flow on each branch, i.e. the
    difference between the AC and DC loadflow results.

    This is used in the solver to adjust the DC flow to match the AC flow in the N-0 case.
    If all zeros are returned, the solver will return pure DC flows.

    Positive values mean the AC flow is higher than the DC flow, negative values mean the
    AC flow is lower than the DC flow.

    Returns
    -------
    Float[np.ndarray, " n_timestep n_branch"]
        The AC-DC mismatch for each branch and per timestep
    """
    return np.zeros_like(self.get_max_mw_flows(), dtype=float)

get_basecase_dc_branch_flows abstractmethod

get_basecase_dc_branch_flows()

Return base-case DC branch flows in the solver's branch orientation.

Positive values flow from get_from_nodes() to get_to_nodes(). The returned values must be ordered consistently with get_branch_ids() and contain one row per timestep.

RETURNS	DESCRIPTION
Float[ndarray, ' n_timestep n_branch']	Base-case DC active branch flows in MW.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/backend.py

@abstractmethod
def get_basecase_dc_branch_flows(self) -> Float[np.ndarray, " n_timestep n_branch"]:
    """Return base-case DC branch flows in the solver's branch orientation.

    Positive values flow from ``get_from_nodes()`` to ``get_to_nodes()``. The returned
    values must be ordered consistently with ``get_branch_ids()`` and contain one row per
    timestep.

    Returns
    -------
    Float[np.ndarray, " n_timestep n_branch"]
        Base-case DC active branch flows in MW.
    """

get_max_mw_flows abstractmethod

get_max_mw_flows()

Get the maximum flow per branch

The timestep dimension is added to represent temperature-dependent capacity limits. If the capacity limits are not temperature-dependent, the same value should be returned for all timesteps.

RETURNS	DESCRIPTION
Float[ndarray, ' n_timestep n_branch']	The maximum flow per branch and per timestep

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/backend.py

@abstractmethod
def get_max_mw_flows(self) -> Float[np.ndarray, " n_timestep n_branch"]:
    """Get the maximum flow per branch

    The timestep dimension is added to represent temperature-dependent capacity
    limits. If the capacity limits are not temperature-dependent, the same value
    should be returned for all timesteps.

    Returns
    -------
    Float[np.ndarray, " n_timestep n_branch"]
        The maximum flow per branch and per timestep
    """

get_max_mw_flows_n_1

get_max_mw_flows_n_1()

Get a varying max flow for N-1 if there is a difference or NaN.

In some circumstances, a higher N-1 load is allowed than N-0 as N-1 leaves some time to address an overload in practice - a line won't melt right away if it's overloaded for a few seconds until the operators can react.

If not overloaded, returns all NaNs and hence the values from get_max_mw_flows will be used.

RETURNS	DESCRIPTION
Float[ndarray, ' n_timestep n_branch']	The maximum flow per branch and per timestep if overridden, else NaN

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/backend.py

def get_max_mw_flows_n_1(self) -> Float[np.ndarray, " n_timestep n_branch"]:
    """Get a varying max flow for N-1 if there is a difference or NaN.

    In some circumstances, a higher N-1 load is allowed than N-0 as N-1 leaves some time to
    address an overload in practice - a line won't melt right away if it's overloaded for a
    few seconds until the operators can react.

    If not overloaded, returns all NaNs and hence the values from get_max_mw_flows will be
    used.

    Returns
    -------
    Float[np.ndarray, " n_timestep n_branch"]
        The maximum flow per branch and per timestep if overridden, else NaN
    """
    return np.full_like(self.get_max_mw_flows(), np.nan)

get_overload_weights

get_overload_weights()

Get a factor that the overloads are multiplied with for each branch

This can be used to penalize overloads on certain branches more than on others.

If this funcion is not overloaded, returns all ones.

RETURNS	DESCRIPTION
Float[ndarray, ' n_branch']	The overload weights for each branch

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/backend.py

def get_overload_weights(self) -> Float[np.ndarray, " n_branch"]:
    """Get a factor that the overloads are multiplied with for each branch

    This can be used to penalize overloads on certain branches more than on others.

    If this funcion is not overloaded, returns all ones.

    Returns
    -------
    Float[np.ndarray, " n_branch"]
        The overload weights for each branch
    """
    return np.ones(self.get_max_mw_flows().shape[-1])

get_n0_n1_max_diff_factors

get_n0_n1_max_diff_factors()

Get limits for the relative difference between N-0 and N-1 flows.

This is an array of factors to the base case flows. Negative factors or NaN values mean the branch will be ignored and always have a penalty of 0. For example if a branch has a 20 MW diff between N-0 and N-1 in the base case (in the unsplit configuration) and the factor is 2, then the maximum allowed diff for the n0_n1_delta penalty would be 40 MW. If a negative factor is used, this branch has no N-0 to N-1 maximum delta and will always incur a penalty of 0. See dc_solver.jax.aggregate_results.compute_n0_n1_max_diff for how these factors are used

If this function is not overloaded, returns all minus ones (i.e. no branch has a limit).

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/backend.py

def get_n0_n1_max_diff_factors(self) -> Float[np.ndarray, " n_branch"]:
    """Get limits for the relative difference between N-0 and N-1 flows.

    This is an array of factors to the base case flows. Negative factors or NaN values mean the
    branch will be ignored and always have a penalty of 0.
    For example if a branch has a 20 MW diff between N-0 and N-1 in the base case (in the
    unsplit configuration) and the factor is 2, then the maximum allowed diff for the
    n0_n1_delta penalty would be 40 MW. If a negative factor is used, this branch has no
    N-0 to N-1 maximum delta and will always incur a penalty of 0. See
    dc_solver.jax.aggregate_results.compute_n0_n1_max_diff for how these factors are used

    If this function is not overloaded, returns all minus ones (i.e. no branch has a limit).
    """
    return -np.ones(self.get_max_mw_flows().shape[-1])

get_cross_coupler_limits

get_cross_coupler_limits()

Get the cross-coupler limits for each relevant substation.

Returns over all buses to match conventions and if relevant substations are modified independently of the cross-coupler limits.

The limits are a P[MW] Value for each coupler.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/backend.py

def get_cross_coupler_limits(self) -> Float[np.ndarray, " n_bus"]:
    """Get the cross-coupler limits for each relevant substation.

    Returns over all buses to match conventions and if relevant substations are modified
    independently of the cross-coupler limits.

    The limits are a P[MW] Value for each coupler.
    """
    return np.zeros(self.get_relevant_node_mask().shape, dtype=float)

get_susceptances abstractmethod

get_susceptances()

Get the susceptances of the branches

RETURNS	DESCRIPTION
Float[ndarray, ' n_branch']	The susceptances of the branches

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/backend.py

@abstractmethod
def get_susceptances(self) -> Float[np.ndarray, " n_branch"]:
    """Get the susceptances of the branches

    Returns
    -------
    Float[np.ndarray, " n_branch"]
        The susceptances of the branches
    """

get_from_nodes abstractmethod

get_from_nodes()

Get the from nodes of the branches

RETURNS	DESCRIPTION
Int[ndarray, ' n_branch']	The from nodes of the branches

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/backend.py

@abstractmethod
def get_from_nodes(self) -> Int[np.ndarray, " n_branch"]:
    """Get the from nodes of the branches

    Returns
    -------
    Int[np.ndarray, " n_branch"]
        The from nodes of the branches
    """

get_to_nodes abstractmethod

get_to_nodes()

Get the to nodes of the branches

RETURNS	DESCRIPTION
Int[ndarray, ' n_branch']	The to nodes of the branches

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/backend.py

@abstractmethod
def get_to_nodes(self) -> Int[np.ndarray, " n_branch"]:
    """Get the to nodes of the branches

    Returns
    -------
    Int[np.ndarray, " n_branch"]
        The to nodes of the branches
    """

get_controllable_pst_node_mask

get_controllable_pst_node_mask()

Get the mask of controllable phase shifters over nodes

True means a node is (bogus node and) a controllable phase shifter, i.e. is connected to a branch that is a controllable phase shifter. False means it normal node.

RETURNS	DESCRIPTION
Bool[ndarray, ' n_node']	The mask of controllable phase shifters over nodes

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/backend.py

def get_controllable_pst_node_mask(self) -> Bool[np.ndarray, " n_node"]:
    """Get the mask of controllable phase shifters over nodes

    True means a node is (bogus node and) a controllable phase shifter, i.e. is connected to a branch
    that is a controllable phase shifter. False means it normal node.

    Returns
    -------
    Bool[np.ndarray, " n_node"]
        The mask of controllable phase shifters over nodes
    """
    return np.zeros(self.get_relevant_node_mask().shape, dtype=bool)

get_shift_angles abstractmethod

get_shift_angles()

Get the shift angles of the branches in degree

The timestep dimension is added to represent time-varying phase shift angles. If the phase shift angles are not time-varying, the same value should be returned for all timesteps.

RETURNS	DESCRIPTION
Float[ndarray, ' n_timestep n_branch']	The shift angles of the branches

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/backend.py

@abstractmethod
def get_shift_angles(self) -> Float[np.ndarray, " n_timestep n_branch"]:
    """Get the shift angles of the branches in degree

    The timestep dimension is added to represent time-varying phase shift angles.
    If the phase shift angles are not time-varying, the same value should be returned
    for all timesteps.

    Returns
    -------
    Float[np.ndarray, " n_timestep n_branch"]
        The shift angles of the branches
    """

get_phase_shift_mask abstractmethod

get_phase_shift_mask()

Get the mask of phase shifters

True means a branch is a phase shifter, i.e. can have shift_degree != 0 False means it is not a phase shifter. Note that the controllable phase shifters are a subset of this, i.e. not every phase shifter is controllable.

RETURNS	DESCRIPTION
Bool[ndarray, ' n_branch']	The mask of phase shifters

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/backend.py

@abstractmethod
def get_phase_shift_mask(self) -> Bool[np.ndarray, " n_branch"]:
    """Get the mask of phase shifters

    True means a branch is a phase shifter, i.e. can have shift_degree != 0
    False means it is not a phase shifter. Note that the controllable phase shifters are a subset of this, i.e. not every
    phase shifter is controllable.

    Returns
    -------
    Bool[np.ndarray, " n_branch"]
        The mask of phase shifters
    """

get_controllable_phase_shift_mask

get_controllable_phase_shift_mask()

Which of the phase shifters are controllable

This must be a subset of get_phase_shift_mask()

RETURNS	DESCRIPTION
Bool[ndarray, ' n_branch']	The mask of controllable phase shifters

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/backend.py

def get_controllable_phase_shift_mask(self) -> Bool[np.ndarray, " n_branch"]:
    """Which of the phase shifters are controllable

    This must be a subset of get_phase_shift_mask()

    Returns
    -------
    Bool[np.ndarray, " n_branch"]
        The mask of controllable phase shifters
    """
    return np.zeros_like(self.get_phase_shift_mask())

get_phase_shift_linearity

get_phase_shift_linearity()

Which of the phase shifters have linear shift behavior

RETURNS	DESCRIPTION
Bool[ndarray, ' n_controllable_psts']	The mask of phase shifters with linear shift behavior

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/backend.py

def get_phase_shift_linearity(self) -> Bool[np.ndarray, " n_controllable_psts"]:
    """Which of the phase shifters have linear shift behavior

    Returns
    -------
    Bool[np.ndarray, " n_controllable_psts"]
        The mask of phase shifters with linear shift behavior
    """
    return np.zeros(sum(self.get_controllable_phase_shift_mask()), dtype=bool)

get_phase_shift_taps

get_phase_shift_taps()

Return the possible tap positions of each controllable PST.

The outer list has as many entries as there are controllable PSTs (see controllable_phase_shift_mask). The inner np array has as many entries as there are taps for the given PST with each value representing the angle shift for the given tap position. The taps are ordered smallest to largest angle shift. Each controllable PST must have at least one tap position.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/backend.py

def get_phase_shift_taps(self) -> list[Float[np.ndarray, " n_tap_positions"]]:
    """Return the possible tap positions of each controllable PST.

    The outer list has as many entries as there are controllable PSTs (see
    controllable_phase_shift_mask). The inner np array has as many entries as there are taps for the given PST with each
    value representing the angle shift for the given tap position. The taps are ordered smallest to largest angle shift.
    Each controllable PST must have at least one tap position.
    """
    # Get the viable shift from the zeroth timestep as a viable default value if the user hasn't overloaded the function
    viable_shifts = self.get_shift_angles()[0, self.get_controllable_phase_shift_mask()]
    return [np.array([shift]) for shift in viable_shifts]

get_phase_shift_susceptance_taps

get_phase_shift_susceptance_taps()

Return the effective branch susceptance at every controllable PST tap.

The returned lists must align with get_phase_shift_taps() and get_controllable_phase_shift_mask(). By default, repeat the current branch susceptance for every tap, which is correct for PSTs whose effective branch parameters do not vary with the tap.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/backend.py

def get_phase_shift_susceptance_taps(self) -> list[Float[np.ndarray, " n_tap_positions"]]:
    """Return the effective branch susceptance at every controllable PST tap.

    The returned lists must align with get_phase_shift_taps() and get_controllable_phase_shift_mask().
    By default, repeat the current branch susceptance for every tap, which is correct for PSTs whose
    effective branch parameters do not vary with the tap.
    """
    controllable_pst_indices = np.flatnonzero(self.get_controllable_phase_shift_mask())
    susceptances = self.get_susceptances()
    tap_values = self.get_phase_shift_taps()
    return [
        np.full_like(taps, fill_value=float(susceptances[branch_idx]), dtype=float)
        for taps, branch_idx in zip(tap_values, controllable_pst_indices, strict=True)
    ]

get_phase_shift_starting_taps

get_phase_shift_starting_taps()

Get the starting tap position for each controllable PST, given as an integer index into pst_tap_values.

Note that taps in the original grid model might not start at zero, while in our optimization we assume taps to always be zero terminated. To translate back into original grid model, add get_phase_shift_low_taps.

The outer list has as many entries as there are controllable PSTs (see controllable_phase_shift_mask). The inner np array has as many entries as there are taps for the given PST with each value representing the angle shift for the given tap position. The taps are ordered smallest to largest angle shift.

If this function is not overloaded, it is assumed that all controllable PSTs start at their lowest tap position (i.e. index 0).

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/backend.py

def get_phase_shift_starting_taps(self) -> Int[np.ndarray, " n_controllable_pst"]:
    """Get the starting tap position for each controllable PST, given as an integer index into pst_tap_values.

    Note that taps in the original grid model might not start at zero, while in our optimization we assume taps to always
    be zero terminated. To translate back into original grid model, add get_phase_shift_low_taps.

    The outer list has as many entries as there are controllable PSTs (see
    controllable_phase_shift_mask). The inner np array has as many entries as there are taps for the given PST with each
    value representing the angle shift for the given tap position. The taps are ordered smallest to largest angle shift.

    If this function is not overloaded, it is assumed that all controllable PSTs
    start at their lowest tap position (i.e. index 0).
    """
    return np.zeros(sum(self.get_controllable_phase_shift_mask()), dtype=int)

get_phase_shift_low_taps

get_phase_shift_low_taps()

Get the lowest tap position in the original grid model

Original taps are needed so taps as integer indices into tap values can be converted back to the original tap positions by tap + low_tap

If this function is not overloaded, it is assumed that all controllable PSTs have a low tap of 0.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/backend.py

def get_phase_shift_low_taps(self) -> Int[np.ndarray, " n_controllable_psts"]:
    """Get the lowest tap position in the original grid model

    Original taps are needed so taps as integer indices into tap values
    can be converted back to the original tap positions by tap + low_tap

    If this function is not overloaded, it is assumed that all controllable PSTs have a low tap of 0.
    """
    return np.zeros(sum(self.get_controllable_phase_shift_mask()), dtype=int)

get_controllable_phase_shift_ids

get_controllable_phase_shift_ids()

Get branch ids of controllable PSTs aligned with controllable PST arrays.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/backend.py

def get_controllable_phase_shift_ids(self) -> list[str]:
    """Get branch ids of controllable PSTs aligned with controllable PST arrays."""
    branch_ids = self.get_branch_ids()
    controllable_pst_mask = self.get_controllable_phase_shift_mask()
    return [
        str(branch_id)
        for branch_id, is_controllable in zip(branch_ids, controllable_pst_mask, strict=True)
        if is_controllable
    ]

get_parallel_pst_group_mask abstractmethod

get_parallel_pst_group_mask()

Get a PST group mask aligned with the controllable PST arrays.

RETURNS	DESCRIPTION
Optional[Bool[np.ndarray, " n_parallel_pst_groups n_controllable_pst"]	The mask for parallel PST groups, or None if no explicit grouping metadata is available.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/backend.py

@abstractmethod
def get_parallel_pst_group_mask(self) -> Optional[Bool[np.ndarray, " n_parallel_pst_groups n_controllable_pst"]]:
    """Get a PST group mask aligned with the controllable PST arrays.

    Returns
    -------
    Optional[Bool[np.ndarray, " n_parallel_pst_groups n_controllable_pst"]
        The mask for parallel PST groups, or None if no explicit grouping metadata is available.
    """

get_parallel_pst_group_ids abstractmethod

get_parallel_pst_group_ids()

Get PST group identifiers aligned with rows of get_parallel_pst_group_mask().

RETURNS	DESCRIPTION
Optional[list[str]]	The identifiers for parallel PST groups, or None if no explicit grouping metadata is available.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/backend.py

@abstractmethod
def get_parallel_pst_group_ids(self) -> Optional[list[str]]:
    """Get PST group identifiers aligned with rows of get_parallel_pst_group_mask().

    Returns
    -------
    Optional[list[str]]
        The identifiers for parallel PST groups, or None if no explicit grouping metadata is available.
    """

get_relevant_node_mask abstractmethod

get_relevant_node_mask()

Get true if a node is part of the relevant nodes

This refers to the node A (the node that is present in the un-extended PTDF) of the relevant substations. The relevant nodes are those that can be split later on in the solver.

RETURNS	DESCRIPTION
Bool[ndarray, ' n_node']	The mask over nodes, indicating if they are relevant (splittable)

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/backend.py

@abstractmethod
def get_relevant_node_mask(self) -> Bool[np.ndarray, " n_node"]:
    """Get true if a node is part of the relevant nodes

    This refers to the node A (the node that is present in the un-extended PTDF) of the
    relevant substations. The relevant nodes are those that can be split later on
    in the solver.

    Returns
    -------
    Bool[np.ndarray, " n_node"]
        The mask over nodes, indicating if they are relevant (splittable)
    """

get_monitored_branch_mask abstractmethod

get_monitored_branch_mask()

Get the mask of monitored branches for the reward calculation

True means a branch is monitored, False means it is not monitored

RETURNS	DESCRIPTION
Bool[ndarray, ' n_branch']	The mask of monitored branches

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/backend.py

@abstractmethod
def get_monitored_branch_mask(self) -> Bool[np.ndarray, " n_branch"]:
    """Get the mask of monitored branches for the reward calculation

    True means a branch is monitored, False means it is not monitored

    Returns
    -------
    Bool[np.ndarray, " n_branch"]
        The mask of monitored branches
    """

get_branches_in_maintenance abstractmethod

get_branches_in_maintenance()

Get the mask of branches in maintenance

True means a branch is in maintenance, False means it is not in maintenance

The timestep dimension is added to represent time-varying maintenance schedules. If the maintenance schedules are not time-varying, the same value should be returned for all timesteps.

RETURNS	DESCRIPTION
Bool[ndarray, ' n_timestep n_branch']	The mask of branches in maintenance

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/backend.py

@abstractmethod
def get_branches_in_maintenance(
    self,
) -> Bool[np.ndarray, " n_timestep n_branch"]:
    """Get the mask of branches in maintenance

    True means a branch is in maintenance, False means it is not in maintenance

    The timestep dimension is added to represent time-varying maintenance schedules.
    If the maintenance schedules are not time-varying, the same value should be returned
    for all timesteps.

    Returns
    -------
    Bool[np.ndarray, " n_timestep n_branch"]
        The mask of branches in maintenance
    """

get_disconnectable_branch_mask abstractmethod

get_disconnectable_branch_mask()

Get the mask of disconnectable branches

True means a branch is disconnectable as a remedial action, False means it must stay online

RETURNS	DESCRIPTION
Bool[ndarray, ' n_branch']	The mask of disconnectable branches

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/backend.py

@abstractmethod
def get_disconnectable_branch_mask(self) -> Bool[np.ndarray, " n_branch"]:
    """Get the mask of disconnectable branches

    True means a branch is disconnectable as a remedial action, False means it must stay online

    Returns
    -------
    Bool[np.ndarray, " n_branch"]
        The mask of disconnectable branches
    """

get_outaged_branch_mask abstractmethod

get_outaged_branch_mask()

Get the mask of outaged branches for the N-1 computation

True means a branch is outaged, False means it is not outaged

RETURNS	DESCRIPTION
Bool[ndarray, ' n_branch']	The mask of outaged branches

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/backend.py

@abstractmethod
def get_outaged_branch_mask(self) -> Bool[np.ndarray, " n_branch"]:
    """Get the mask of outaged branches for the N-1 computation

    True means a branch is outaged, False means it is not outaged

    Returns
    -------
    Bool[np.ndarray, " n_branch"]
        The mask of outaged branches
    """

get_outaged_injection_mask abstractmethod

get_outaged_injection_mask()

Get the mask of outaged injections for the N-1 computation

True means an injection is outaged, False means it is not outaged

RETURNS	DESCRIPTION
Bool[ndarray, ' n_injection']	The mask of outaged injections

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/backend.py

@abstractmethod
def get_outaged_injection_mask(self) -> Bool[np.ndarray, " n_injection"]:
    """Get the mask of outaged injections for the N-1 computation

    True means an injection is outaged, False means it is not outaged

    Returns
    -------
    Bool[np.ndarray, " n_injection"]
        The mask of outaged injections
    """

get_multi_outage_branches abstractmethod

get_multi_outage_branches()

Get the mask of outaged branches for potential multi-outages

True means a branch is outaged, False means it is not outaged.

get_multi_outage_branches, get_multi_outage_nodes and get_multi_outage_names have to return the same first dimension, i.e. the same number of multi-outages.

RETURNS	DESCRIPTION
Bool[ndarray, ' n_multi_outages n_branch']	The mask of outaged branches for every multi-outage

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/backend.py

@abstractmethod
def get_multi_outage_branches(
    self,
) -> Bool[np.ndarray, " n_multi_outages n_branch"]:
    """Get the mask of outaged branches for potential multi-outages

    True means a branch is outaged, False means it is not outaged.

    get_multi_outage_branches, get_multi_outage_nodes and get_multi_outage_names have to return
    the same first dimension, i.e. the same number of multi-outages.

    Returns
    -------
    Bool[np.ndarray, " n_multi_outages n_branch"]
        The mask of outaged branches for every multi-outage
    """

get_multi_outage_nodes abstractmethod

get_multi_outage_nodes()

Get the mask of outaged nodes for potential multi-outages

True means a node is outaged, False means it is not outaged.

get_multi_outage_branches, get_multi_outage_nodes and get_multi_outage_names have to return the same first dimension, i.e. the same number of multi-outages.

RETURNS	DESCRIPTION
Bool[ndarray, ' n_multi_outages n_node']	The mask of outaged nodes for every multi-outage

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/backend.py

@abstractmethod
def get_multi_outage_nodes(
    self,
) -> Bool[np.ndarray, " n_multi_outages n_node"]:
    """Get the mask of outaged nodes for potential multi-outages

    True means a node is outaged, False means it is not outaged.

    get_multi_outage_branches, get_multi_outage_nodes and get_multi_outage_names have to return
    the same first dimension, i.e. the same number of multi-outages.

    Returns
    -------
    Bool[np.ndarray, " n_multi_outages n_node"]
        The mask of outaged nodes for every multi-outage
    """

get_injection_nodes abstractmethod

get_injection_nodes()

Get the node index of the injections

RETURNS	DESCRIPTION
Int[ndarray, ' n_injection']	The node index that the injection injects onto

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/backend.py

@abstractmethod
def get_injection_nodes(self) -> Int[np.ndarray, " n_injection"]:
    """Get the node index of the injections

    Returns
    -------
    Int[np.ndarray, " n_injection"]
        The node index that the injection injects onto
    """

get_mw_injections abstractmethod

get_mw_injections()

Get the MW injections of the injections

The timestep dimension is added to represent time-varying injections. If the injections are not time-varying, the same value should be returned for all timesteps.

RETURNS	DESCRIPTION
Float[ndarray, ' n_timestep n_injection']	The MW injections of the injections

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/backend.py

@abstractmethod
def get_mw_injections(self) -> Float[np.ndarray, " n_timestep n_injection"]:
    """Get the MW injections of the injections

    The timestep dimension is added to represent time-varying injections.
    If the injections are not time-varying, the same value should be returned
    for all timesteps.

    Returns
    -------
    Float[np.ndarray, " n_timestep n_injection"]
        The MW injections of the injections
    """

get_base_mva abstractmethod

get_base_mva()

Get the baseMVA of the grid

RETURNS	DESCRIPTION
float	The base MVA of the grid

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/backend.py

@abstractmethod
def get_base_mva(self) -> float:
    """Get the baseMVA of the grid

    Returns
    -------
    float
        The base MVA of the grid
    """

get_master_asset_topology

get_master_asset_topology()

Get canonical asset-topology master data for the grid, if available.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/backend.py

def get_master_asset_topology(self) -> Optional[MasterAssetTopology]:
    """Get canonical asset-topology master data for the grid, if available."""
    return None

get_runtime_asset_topology

get_runtime_asset_topology()

Get live runtime-enriched topology payloads for the current grid, if available.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/backend.py

def get_runtime_asset_topology(self) -> Optional[RuntimeAssetTopology]:
    """Get live runtime-enriched topology payloads for the current grid, if available."""
    return None

get_node_ids abstractmethod

get_node_ids()

Get the ids of the nodes as a Sequence of length N_node

RETURNS	DESCRIPTION
Union[Sequence[str], Sequence[int]]	The ids of the nodes

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/backend.py

@abstractmethod
def get_node_ids(
    self,
) -> Union[Sequence[str], Sequence[int]]:
    """Get the ids of the nodes as a Sequence of length N_node

    Returns
    -------
    Union[Sequence[str], Sequence[int]]
        The ids of the nodes
    """

get_branch_ids abstractmethod

get_branch_ids()

Get the ids of the branches as a Sequence of length N_branch

RETURNS	DESCRIPTION
Union[Sequence[str], Sequence[int]]	The ids of the branches

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/backend.py

@abstractmethod
def get_branch_ids(
    self,
) -> Union[Sequence[str], Sequence[int]]:
    """Get the ids of the branches as a Sequence of length N_branch

    Returns
    -------
    Union[Sequence[str], Sequence[int]]
        The ids of the branches
    """

get_injection_ids abstractmethod

get_injection_ids()

Get the ids of the injections as a Sequence of length N_injection

RETURNS	DESCRIPTION
Union[Sequence[str], Sequence[int]]	The ids of the injections

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/backend.py

@abstractmethod
def get_injection_ids(
    self,
) -> Union[Sequence[str], Sequence[int]]:
    """Get the ids of the injections as a Sequence of length N_injection

    Returns
    -------
    Union[Sequence[str], Sequence[int]]
        The ids of the injections
    """

get_multi_outage_ids abstractmethod

get_multi_outage_ids()

Get the ids of the multi-outages as a Sequence of length N_multi_outages

RETURNS	DESCRIPTION
Union[Sequence[str], Sequence[int]]	The ids of the multi-outages

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/backend.py

@abstractmethod
def get_multi_outage_ids(self) -> Union[Sequence[str], Sequence[int]]:
    """Get the ids of the multi-outages as a Sequence of length N_multi_outages

    Returns
    -------
    Union[Sequence[str], Sequence[int]]
        The ids of the multi-outages
    """

get_node_names abstractmethod

get_node_names()

Get the names of the nodes as a Sequence of length N_node

RETURNS	DESCRIPTION
Sequence[str]	The names of the nodes

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/backend.py

@abstractmethod
def get_node_names(self) -> Sequence[str]:
    """Get the names of the nodes as a Sequence of length N_node

    Returns
    -------
    Sequence[str]
        The names of the nodes
    """

get_branch_names abstractmethod

get_branch_names()

Get the names of the branches as a Sequence of length N_branch

RETURNS	DESCRIPTION
Sequence[str]	The names of the branches

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/backend.py

@abstractmethod
def get_branch_names(self) -> Sequence[str]:
    """Get the names of the branches as a Sequence of length N_branch

    Returns
    -------
    Sequence[str]
        The names of the branches
    """

get_injection_names abstractmethod

get_injection_names()

Get the names of the injections

RETURNS	DESCRIPTION
Sequence[str]	The names of the injections

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/backend.py

@abstractmethod
def get_injection_names(self) -> Sequence[str]:
    """Get the names of the injections

    Returns
    -------
    Sequence[str]
        The names of the injections
    """

get_multi_outage_names abstractmethod

get_multi_outage_names()

Get the names of the multi-outages as a Sequence of length N_multi_outages

If more than one element are involved in a multi-outage you can return a concatenated name

RETURNS	DESCRIPTION
Sequence[str]	The names of the multi-outages

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/backend.py

@abstractmethod
def get_multi_outage_names(self) -> Sequence[str]:
    """Get the names of the multi-outages as a Sequence of length N_multi_outages

    If more than one element are involved in a multi-outage you can return a concatenated name

    Returns
    -------
    Sequence[str]
        The names of the multi-outages
    """

get_branch_types abstractmethod

get_branch_types()

Get the type of the branches

RETURNS	DESCRIPTION
Sequence[str]	The type of the branches

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/backend.py

@abstractmethod
def get_branch_types(self) -> Sequence[str]:
    """Get the type of the branches

    Returns
    -------
    Sequence[str]
        The type of the branches
    """

get_node_types abstractmethod

get_node_types()

Get the type of the nodes

RETURNS	DESCRIPTION
Sequence[str]	The type of the nodes

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/backend.py

@abstractmethod
def get_node_types(self) -> Sequence[str]:
    """Get the type of the nodes

    Returns
    -------
    Sequence[str]
        The type of the nodes
    """

get_injection_types abstractmethod

get_injection_types()

Get the type of the injections

RETURNS	DESCRIPTION
Sequence[str]	The type of the injections

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/backend.py

@abstractmethod
def get_injection_types(self) -> Sequence[str]:
    """Get the type of the injections

    Returns
    -------
    Sequence[str]
        The type of the injections
    """

get_multi_outage_types abstractmethod

get_multi_outage_types()

Get the type of the multi-outages as a Sequence of length N_multi_outages

RETURNS	DESCRIPTION
Sequence[str]	The type of the multi-outages

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/backend.py

@abstractmethod
def get_multi_outage_types(self) -> Sequence[str]:
    """Get the type of the multi-outages as a Sequence of length N_multi_outages

    Returns
    -------
    Sequence[str]
        The type of the multi-outages
    """

get_metadata abstractmethod

get_metadata()

Can be used to return metadata or additional information about the grid.

This is not used by the solver but rather to easy postprocessing and validation. You can return an empty dict if you don't want to use this field.

RETURNS	DESCRIPTION
dict	The metadata of the grid

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/backend.py

@abstractmethod
def get_metadata(self) -> dict:
    """Can be used to return metadata or additional information about the grid.

    This is not used by the solver but rather to easy postprocessing and validation. You can
    return an empty dict if you don't want to use this field.

    Returns
    -------
    dict
        The metadata of the grid
    """

get_busbar_outage_map

get_busbar_outage_map()

Get the mapping of stations to busbars for the busbar-outages

The key of the dict is the station's bus_group_id and the value is a list of grid_model_ids of the busbars that have to be outaged. If this method is not overloaded, all the physical busbars of the relevant stations will be outaged.

RETURNS	DESCRIPTION
Optional[dict[str, Sequence[str]]]	The mapping of busbar-outages to the relevant nodes

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/backend.py

def get_busbar_outage_map(
    self,
) -> Optional[dict[str, Sequence[str]]]:
    """Get the mapping of stations to busbars for the busbar-outages

    The key of the dict is the station's bus_group_id and the value is a list of grid_model_ids
    of the busbars that have to be outaged. If this method is not overloaded, all the physical
    busbars of the relevant stations will be outaged.

    Returns
    -------
    Optional[dict[str, Sequence[str]]]
        The mapping of busbar-outages to the relevant nodes
    """
    return None

toop_engine_interfaces.types

Define global type aliases.

Currently this holds only the MetricType, a Literal of all possible metrics.

The meaning of the metrics is documented in metrics.md in the docs folder.

MatrixMetric module-attribute

MatrixMetric = Literal[
    "max_flow_n_0",
    "median_flow_n_0",
    "overload_energy_n_0",
    "underload_energy_n_0",
    "overload_energy_limited_n_0",
    "exponential_overload_energy_n_0",
    "exponential_overload_energy_limited_n_0",
    "critical_branch_count_n_0",
    "critical_branch_count_limited_n_0",
    "max_flow_n_1",
    "median_flow_n_1",
    "overload_energy_n_1",
    "underload_energy_n_1",
    "overload_energy_limited_n_1",
    "exponential_overload_energy_n_1",
    "exponential_overload_energy_limited_n_1",
    "critical_branch_count_n_1",
    "critical_branch_count_limited_n_1",
    "top_k_overloads_n_1",
    "cumulative_overload_n_0",
    "cumulative_overload_n_1",
    "transport_n_0",
    "transport_n_1",
]

OperationMetric module-attribute

OperationMetric = Literal[
    "switching_distance",
    "split_subs",
    "disconnected_branches",
    "fitness",
    "pst_switching_distance",
    "pst_switching_distance_squared",
    "pst_activated",
]

OtherMetric module-attribute

OtherMetric = Literal[
    "n0_n1_delta",
    "cross_coupler_flow",
    "bb_outage_penalty",
    "bb_outage_overload",
    "bb_outage_grid_splits",
    "max_va_diff_n_0",
    "max_va_diff_n_1",
    "critical_va_diff_count_n_0",
    "critical_va_diff_count_n_1",
    "overload_current_n_0",
    "overload_current_n_1",
    "voltage_jump_count_n_1",
    "non_converging_loadflows",
    "fitness_dc",
]

MetricType module-attribute

MetricType = Literal[
    MatrixMetric, OperationMetric, OtherMetric
]

toop_engine_interfaces.status_update

The in-process progress reporting of the preprocessing routines.

This holds the callback that preprocessing routines invoke to report their progress, along with the statistics they report next to a stage. Both are an internal interface between the routines and whoever drives them, and the statistics only ever end up in the logs. They live here rather than in a package of their own so that the importer and the DC solver can depend on them without depending on each other.

The stages themselves are user-facing and are therefore defined alongside the heartbeat message in toop_engine_interfaces.messages.preprocess.preprocess_heartbeat.

logger module-attribute

logger = structlog.get_logger(__name__)

NetworkDataStats module-attribute

NetworkDataStats = dict[str, int]

StatusUpdateFn

Bases: Protocol

The callback used to report progress through the preprocessing pipeline.

__call__

__call__(stage, message, *, stats=None)

Report that a preprocessing stage was entered.

PARAMETER	DESCRIPTION
stage	The stage that is being entered TYPE: PreprocessStage
message	An optional message with more detail on the stage TYPE: Optional[str]
stats	Size statistics of the network data as it looks when entering the stage, if available. Only stages that operate on network data report these. TYPE: Optional[NetworkDataStats] DEFAULT: None

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/status_update.py

def __call__(
    self,
    stage: PreprocessStage,
    message: Optional[str],
    *,
    stats: Optional[NetworkDataStats] = None,
) -> None:
    """Report that a preprocessing stage was entered.

    Parameters
    ----------
    stage : PreprocessStage
        The stage that is being entered
    message : Optional[str]
        An optional message with more detail on the stage
    stats : Optional[NetworkDataStats]
        Size statistics of the network data as it looks when entering the stage, if available.
        Only stages that operate on network data report these.
    """
    ...

empty_status_update_fn

empty_status_update_fn(stage, message, *, stats=None)

Log an empty status update to logging.

Use this function when no status_update_fn is provided.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/status_update.py

def empty_status_update_fn(
    stage: PreprocessStage, message: Optional[str], *, stats: Optional[NetworkDataStats] = None
) -> None:
    """Log an empty status update to logging.

    Use this function when no status_update_fn is provided.
    """
    stats_kwargs = {} if stats is None else {"network_stats": stats}
    if message is None:
        logger.info(f"Preprocessing stage {stage}", preprocess_stage=stage, **stats_kwargs)
    else:
        logger.info(f"Preprocessing stage {stage}, {message}", preprocess_stage=stage, message=message, **stats_kwargs)

toop_engine_interfaces.stored_action_set

Holds a format for storing the action set for later use in postprocessing.

This is different from the jax-internal action set as defined in jax/types.py where only jax-relevant information is stored, but is instead aimed at use in postprocessing and visualization. Instead of just storing the electrical switching state, this bases on the asset topology to store physical switchings to make a translation to .dgs or other formats easier.

One of the decisions to take was was whether to use a single action set for all timesteps or a different one for each timestep. As the jax part currently also only supports one action set for all timesteps, we decide to mirror this for the time being, i.e. we do not store strategies but topologies in the action set.

Furthermore, it should also be possible to use a global action set if necessary. Meaning, by default an action is substation-local, but it it can span multiple substations as well. Using a format that is suitable for both options is desirable for easier collaboration.

Another question was whether to store the switching distance and busbar information in the action set, but the switching distance can be trivially recomputed by using the bus-group diff between the starting topology and the bus group in the action set. BB outage information can also be retrieved from the asset topology.

There is a slim hope of storing the action set independent of the grid state but based on the master grid, however right now there is a fundamental way that 'binds' an action set to the specific grid it has been computed on: During the enumerations, all electrical actions are enumerated and then physical realizations are found for it based on heuristics. These heuristics take the grid state into account, so it could be that an electrical action can not be realized the same way if maintenances are active. Hence, for the moment, it is no problem to tie the initial topology into the action set.

STATION_DIFF_ORDER_ATTR module-attribute

STATION_DIFF_ORDER_ATTR = 'station_order'

PSTRange

Bases: GridElement

Phase shifting transformers can be set within the scope of non-costly optimization.

A PST has a list of taps, each with an angle shift.

starting_tap instance-attribute

starting_tap

The tap the PST was set to before optimization. To filter out actions that do not change anything in the UI, this is required.

low_tap instance-attribute

low_tap

The lowest tap the PST supports

high_tap instance-attribute

high_tap

The highest tap the PST supports

pst_group class-attribute instance-attribute

pst_group = None

The optimization group of the PST.

When omitted in serialized action sets, this defaults to the PST id for backward compatibility.

id instance-attribute

id

The id of the element. For powsybl grids this is the global string id, for pandapower this is the integer index into the dataframe

name class-attribute instance-attribute

name = ''

The name of the element. This is optional, but can be used to provide a more human-readable name for the element.

type instance-attribute

type

For pandapower, we need to further specify a type which corresponds to the table pandapower stores the information in. Valid tables are 'line', 'trafo', 'ext_grid', 'gen', 'load', 'shunt', ... For powsybl, this is not strictly needed to identify the element however it makes it easier. In that case, type will be something like TIE_LINE, LINE, TWO_WINDING_TRANSFORMER, GENERATOR, etc.

kind instance-attribute

kind

The kind of the element. Usually these are handled differently in the grid modelling software, so it can make assembling an N-1 analysis easier if it is known if the element is a branch, bus or injection. This could be inferred from the type, however for conveniece it is stored separately.

For the bus type there is some potential confusion in powsybl. In pandapower, this always refers to the net.bus df. In powsybl in a bus/branch model, there are no busbar sections in powsybl, i.e. net.get_node_breaker_topology does not deliver busbar sections. Meaning, the "bus" type refers to the net.get_bus_breaker_topology buses if it's a bus/breaker topology bus. If it's a node/breaker topology, then "bus" refers to the busbar section.

HVDCRange

Bases: GridElement

High voltage direct current lines can be set within the scope of non-costly optimization.

An HVDC has a minimum and maximum power setpoint

min_power instance-attribute

min_power

The lowest power setpoint the HVDC supports

max_power instance-attribute

max_power

The highest power setpoint the HVDC supports

id instance-attribute

id

The id of the element. For powsybl grids this is the global string id, for pandapower this is the integer index into the dataframe

name class-attribute instance-attribute

name = ''

The name of the element. This is optional, but can be used to provide a more human-readable name for the element.

type instance-attribute

type

For pandapower, we need to further specify a type which corresponds to the table pandapower stores the information in. Valid tables are 'line', 'trafo', 'ext_grid', 'gen', 'load', 'shunt', ... For powsybl, this is not strictly needed to identify the element however it makes it easier. In that case, type will be something like TIE_LINE, LINE, TWO_WINDING_TRANSFORMER, GENERATOR, etc.

kind instance-attribute

kind

The kind of the element. Usually these are handled differently in the grid modelling software, so it can make assembling an N-1 analysis easier if it is known if the element is a branch, bus or injection. This could be inferred from the type, however for conveniece it is stored separately.

For the bus type there is some potential confusion in powsybl. In pandapower, this always refers to the net.bus df. In powsybl in a bus/branch model, there are no busbar sections in powsybl, i.e. net.get_node_breaker_topology does not deliver busbar sections. Meaning, the "bus" type refers to the net.get_bus_breaker_topology buses if it's a bus/breaker topology bus. If it's a node/breaker topology, then "bus" refers to the busbar section.

ActionSet

Bases: BaseModel

A collection of actions available to the optimizer in readable form.

All actions are also stored directly in jax, but without IDs, names or other useful information to introspect them.

model_config class-attribute instance-attribute

model_config = ConfigDict(extra='forbid')

starting_bus_groups instance-attribute

starting_bus_groups

Runtime-aware bus-group snapshots for the starting grid state. These are used to get to the switches of disconnectable branches. In the future this will be handled differently via the circuit groups.

simplified_starting_bus_groups instance-attribute

simplified_starting_bus_groups

Runtime-aware bus-group snapshots for the simplified starting grid state.

These snapshots define the bus-group and asset ordering contract for local_actions. They are still runtime snapshots, but projected to the reduced DC-solver asset view rather than the full physical bus-group view.

These are used to apply the split topology.

connectable_branches instance-attribute

connectable_branches

A list of assets that can be connected as a remedial action.

disconnectable_branches instance-attribute

disconnectable_branches

A list of assets that can be disconnected as a remedial action. Currently the DC solver supports only branches.

pst_ranges instance-attribute

pst_ranges

A list of phase shifting transformers that can be set as a remedial action.

hvdc_ranges instance-attribute

hvdc_ranges

A list of high voltage direct current lines that can be set as a remedial action. This is currently not implemented yet in the solver.

local_actions instance-attribute

local_actions

A list of split/reconfiguration actions that affect exactly one electrical bus group.

Actions affecting the same bus group are adjacent. The bus_group_id identifies the affected bus group, and asset ordering matches the corresponding entry in simplified_starting_bus_groups.

get_starting_bus_groups

get_starting_bus_groups()

Return normalized runtime-aware bus-group snapshots for the starting topology.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/stored_action_set.py

def get_starting_bus_groups(self) -> list[RuntimeBusGroup]:
    """Return normalized runtime-aware bus-group snapshots for the starting topology."""
    return self.starting_bus_groups

get_simplified_starting_bus_groups

get_simplified_starting_bus_groups()

Return normalized runtime-aware bus-group snapshots for the simplified starting topology.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/stored_action_set.py

def get_simplified_starting_bus_groups(self) -> list[SimplifiedBusGroup]:
    """Return normalized runtime-aware bus-group snapshots for the simplified starting topology."""
    return self.simplified_starting_bus_groups

BusGroupDiffArray

Bases: BaseModel

A difference between copies of a bus group in the local action set and the starting topology.

So that the action set does not have to store copies of the full bus group with all associated information, we only store the changes in the bus group that are typical for the actions in the action set, i.e. the switching table and coupler states. Furthermore, we store them in array form for the entire action set, so that we can potentially store them in parquet format.

A full action set consists of bus-group diffs for every switchable bus group in the grid.

model_config class-attribute instance-attribute

model_config = ConfigDict(arbitrary_types_allowed=True)

grid_model_id instance-attribute

grid_model_id

The grid model id of the bus group.

coupler_open instance-attribute

coupler_open

The state of the "open" field for every coupler in the bus group. The array dimension n_couplers is equivalent to bus_group.couplers in length and order and the entries correspond to open (True) and closed (False). The n_actions dimension provides an entry per action in the action set.

branch_switching_table instance-attribute

branch_switching_table

Branch switching tables for the bus-group actions.

The busbar and branch-asset dimensions match bus_group.branch_switching_table.

injection_switching_table instance-attribute

injection_switching_table

Injection switching tables for the bus-group actions.

The busbar and injection-asset dimensions match bus_group.injection_switching_table.

validate_actions_grouped

validate_actions_grouped(actions)

Validate that actions are grouped by bus-group grid model id.

PARAMETER	DESCRIPTION
actions	Action bus groups to validate. TYPE: list[SimplifiedBusGroup]

RAISES	DESCRIPTION
ValueError	If a bus-group grid model id appears in multiple non-contiguous groups.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/stored_action_set.py

def validate_actions_grouped(actions: list[SimplifiedBusGroup]) -> None:
    """Validate that actions are grouped by bus-group grid model id.

    Parameters
    ----------
    actions : list[SimplifiedBusGroup]
        Action bus groups to validate.

    Raises
    ------
    ValueError
        If a bus-group grid model id appears in multiple non-contiguous groups.
    """
    seen_grid_model_ids: set[str] = set()
    last_grid_model_id: str | None = None
    for action in actions:
        grid_model_id = action.bus_group_id
        if grid_model_id != last_grid_model_id:
            if grid_model_id in seen_grid_model_ids:
                raise ValueError(
                    f"Actions are not grouped by station. Grid model id {grid_model_id} appears in multiple groups."
                )
            seen_grid_model_ids.add(grid_model_id)
            last_grid_model_id = grid_model_id

store_bus_group_diff_fs

store_bus_group_diff_fs(
    filesystem, bus_group_diffs, diff_file_path
)

Store bus-group diffs to an HDF5 file, using a different group for every bus group.

Use load_bus_group_diff_fs to load it again

PARAMETER	DESCRIPTION
filesystem	A filesystem to store the bus-group diffs in. TYPE: AbstractFileSystem
bus_group_diffs	A list of bus-group diffs to store. TYPE: list[BusGroupDiffArray]
diff_file_path	The file path to store the bus-group diffs in. TYPE: str | Path

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/stored_action_set.py

def store_bus_group_diff_fs(
    filesystem: AbstractFileSystem, bus_group_diffs: list[BusGroupDiffArray], diff_file_path: str | Path
) -> None:
    """Store bus-group diffs to an HDF5 file, using a different group for every bus group.

    Use load_bus_group_diff_fs to load it again

    Parameters
    ----------
    filesystem : AbstractFileSystem
        A filesystem to store the bus-group diffs in.
    bus_group_diffs : list[BusGroupDiffArray]
        A list of bus-group diffs to store.
    diff_file_path : str | Path
        The file path to store the bus-group diffs in.
    """
    filesystem.makedirs(Path(diff_file_path).parent.as_posix(), exist_ok=True)

    bytes_io = io.BytesIO()
    with h5py.File(bytes_io, mode="w") as file:
        file.attrs[STATION_DIFF_ORDER_ATTR] = np.array(
            [bus_group_diff.grid_model_id for bus_group_diff in bus_group_diffs],
            dtype=h5py.string_dtype(encoding="utf-8"),
        )
        for bus_group_diff in bus_group_diffs:
            group = file.create_group(bus_group_diff.grid_model_id)
            group.create_dataset("coupler_open", data=bus_group_diff.coupler_open)
            group.create_dataset("branch_switching_table", data=bus_group_diff.branch_switching_table)
            group.create_dataset("injection_switching_table", data=bus_group_diff.injection_switching_table)
    bytes_io.seek(0)
    with filesystem.open(str(diff_file_path), "wb") as file:
        file.write(bytes_io.getbuffer())

load_bus_group_diff_fs

load_bus_group_diff_fs(filesystem, diff_file_path)

Load bus-group diffs from an HDF5 file, using a different group for every bus group.

Use store_station_diff_io to store it.

PARAMETER	DESCRIPTION
filesystem	A filesystem to load the bus-group diffs from. TYPE: AbstractFileSystem
diff_file_path	The file path to load the bus-group diffs from. TYPE: str | Path

RETURNS	DESCRIPTION
list[BusGroupDiffArray]	A list of bus-group diffs loaded from the file.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/stored_action_set.py

def load_bus_group_diff_fs(filesystem: AbstractFileSystem, diff_file_path: str | Path) -> list[BusGroupDiffArray]:
    """Load bus-group diffs from an HDF5 file, using a different group for every bus group.

    Use store_station_diff_io to store it.

    Parameters
    ----------
    filesystem : AbstractFileSystem
        A filesystem to load the bus-group diffs from.
    diff_file_path : str | Path
        The file path to load the bus-group diffs from.

    Returns
    -------
    list[BusGroupDiffArray]
        A list of bus-group diffs loaded from the file.
    """
    with filesystem.open(str(diff_file_path), "rb") as file:
        file_bytes = file.read()
    buffer = io.BytesIO(file_bytes)
    return _load_bus_group_diff_io(buffer)

expand_single_bus_group_diff_to_actions

expand_single_bus_group_diff_to_actions(
    starting_busgroup, bus_group_diff
)

Expand one densely stored bus-group diff to action bus groups.

PARAMETER	DESCRIPTION
starting_busgroup	The starting bus group. Its static fields are copied while coupler states and switching tables are replaced. TYPE: SimplifiedBusGroup
bus_group_diff	The bus-group diff to expand. TYPE: BusGroupDiffArray

RETURNS	DESCRIPTION
list[SimplifiedBusGroup]	Action bus groups represented by the diff action dimension.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/stored_action_set.py

def expand_single_bus_group_diff_to_actions(
    starting_busgroup: SimplifiedBusGroup, bus_group_diff: BusGroupDiffArray
) -> list[SimplifiedBusGroup]:
    """Expand one densely stored bus-group diff to action bus groups.

    Parameters
    ----------
    starting_busgroup : SimplifiedBusGroup
        The starting bus group. Its static fields are copied while coupler states and switching tables are replaced.
    bus_group_diff : BusGroupDiffArray
        The bus-group diff to expand.

    Returns
    -------
    list[SimplifiedBusGroup]
        Action bus groups represented by the diff action dimension.
    """
    actions = []
    coupler_state_cache: dict[tuple[bool, ...], list] = {}
    for i in range(bus_group_diff.coupler_open.shape[0]):
        coupler_state_key = tuple(bool(coupler_open) for coupler_open in bus_group_diff.coupler_open[i])
        couplers = coupler_state_cache.get(coupler_state_key)
        if couplers is None:
            couplers = [
                coupler.model_copy(update={"open": coupler_open}, deep=False)
                for coupler, coupler_open in zip(starting_busgroup.couplers, coupler_state_key, strict=True)
            ]
            coupler_state_cache[coupler_state_key] = couplers

        branch_switching_table = bus_group_diff.branch_switching_table[i]
        injection_switching_table = bus_group_diff.injection_switching_table[i]

        action = _construct_action_from_bus_group_diff(
            starting_busgroup=starting_busgroup,
            couplers=couplers,
            branch_switching_table=branch_switching_table,
            injection_switching_table=injection_switching_table,
        )
        actions.append(action)
    return actions

expand_bus_group_diffs_from_starting_bus_groups

expand_bus_group_diffs_from_starting_bus_groups(
    starting_bus_groups, bus_group_diffs
)

Expand densely stored bus-group diffs from reference runtime bus groups.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/stored_action_set.py

def expand_bus_group_diffs_from_starting_bus_groups(
    starting_bus_groups: list[SimplifiedBusGroup],
    bus_group_diffs: list[BusGroupDiffArray],
) -> list[SimplifiedBusGroup]:
    """Expand densely stored bus-group diffs from reference runtime bus groups."""
    grid_model_id_to_station = {station.bus_group_id: station for station in starting_bus_groups}
    actions = []
    for bus_group_diff in bus_group_diffs:
        starting_station = grid_model_id_to_station[bus_group_diff.grid_model_id]
        actions.extend(expand_single_bus_group_diff_to_actions(starting_station, bus_group_diff))
    return actions

compress_actions_to_bus_group_diffs_from_starting_bus_groups

compress_actions_to_bus_group_diffs_from_starting_bus_groups(
    starting_bus_groups,
    actions,
    validate_diff_hypothesis=False,
)

Compress action bus groups to bus-group diffs using reference runtime bus groups.

This is the inverse of expand_bus_group_diffs_from_starting_bus_groups and keeps only the state that actually varies across local actions: coupler openness and the two switching tables.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/stored_action_set.py

def compress_actions_to_bus_group_diffs_from_starting_bus_groups(
    starting_bus_groups: list[SimplifiedBusGroup],
    actions: list[SimplifiedBusGroup],
    validate_diff_hypothesis: bool = False,
) -> list[BusGroupDiffArray]:
    """Compress action bus groups to bus-group diffs using reference runtime bus groups.

    This is the inverse of ``expand_bus_group_diffs_from_starting_bus_groups`` and keeps only the
    state that actually varies across local actions: coupler openness and the two switching tables.
    """
    grid_model_id_to_station = {station.bus_group_id: station for station in starting_bus_groups}
    bus_group_diffs = {}
    for grid_model_id, group in itertools.groupby(actions, key=lambda action: action.bus_group_id):
        if grid_model_id not in grid_model_id_to_station:
            raise ValueError(f"Action station id {grid_model_id} not found in starting topology.")
        starting_station = grid_model_id_to_station[grid_model_id]

        coupler_open = []
        branch_switching_tables = []
        injection_switching_tables = []
        for action in group:
            assert len(action.couplers) == len(starting_station.couplers), (
                "Number of couplers in action station does not match starting station."
            )
            assert action.branch_switching_table.shape == starting_station.branch_switching_table.shape, (
                "Branch switching table shape in action station does not match starting station."
            )
            assert action.injection_switching_table.shape == starting_station.injection_switching_table.shape, (
                "Injection switching table shape in action station does not match starting station."
            )
            if validate_diff_hypothesis:
                _validate_bus_group_diff_hypothesis(starting_busgroups=starting_station, action=action)
            coupler_open.append([coupler.open for coupler in action.couplers])
            branch_switching_tables.append(action.branch_switching_table)
            injection_switching_tables.append(action.injection_switching_table)
        coupler_open_array = np.array(coupler_open).astype(bool)
        branch_switching_table_array = np.array(branch_switching_tables).astype(bool)
        injection_switching_table_array = np.array(injection_switching_tables).astype(bool)
        bus_group_diff = BusGroupDiffArray(
            grid_model_id=grid_model_id,
            coupler_open=coupler_open_array,
            branch_switching_table=branch_switching_table_array,
            injection_switching_table=injection_switching_table_array,
        )
        if bus_group_diff.grid_model_id in bus_group_diffs:
            raise ValueError(f"Duplicate station diff for grid_model_id {grid_model_id}, actions were not in order.")
        bus_group_diffs[grid_model_id] = bus_group_diff
    return list(bus_group_diffs.values())

load_action_set_fs

load_action_set_fs(
    filesystem, json_file_path, diff_file_path
)

Load an action set from a file system.

PARAMETER	DESCRIPTION
filesystem	The file system to use to load the action set. TYPE: AbstractFileSystem
json_file_path	The path to the JSON file containing the action set without local actions. TYPE: Union[str, Path]
diff_file_path	The path to the HDF5 file containing the station diffs to expand to local actions. If this is none, the local_actions field will not be filled and be the empty list. TYPE: Union[str, Path] | None

RETURNS	DESCRIPTION
ActionSet	The action set loaded from the file. When diff_file_path is given, local_actions are reconstructed from the stored bus-group diffs and simplified_starting_bus_groups.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/stored_action_set.py

def load_action_set_fs(
    filesystem: AbstractFileSystem, json_file_path: Union[str, Path], diff_file_path: Union[str, Path] | None
) -> ActionSet:
    """Load an action set from a file system.

    Parameters
    ----------
    filesystem : AbstractFileSystem
        The file system to use to load the action set.
    json_file_path : Union[str, Path]
        The path to the JSON file containing the action set without local actions.
    diff_file_path : Union[str, Path] | None
        The path to the HDF5 file containing the station diffs to expand to local actions. If this is none, the
        local_actions field will not be filled and be the empty list.

    Returns
    -------
    ActionSet
        The action set loaded from the file. When ``diff_file_path`` is given, ``local_actions``
        are reconstructed from the stored bus-group diffs and ``simplified_starting_bus_groups``.
    """
    with filesystem.open(str(json_file_path), "r") as f:
        payload = json.loads(f.read())
    action_set = ActionSet.model_validate(payload)
    if diff_file_path is not None:
        bus_group_diffs = load_bus_group_diff_fs(filesystem, diff_file_path)
        local_actions = expand_bus_group_diffs_from_starting_bus_groups(
            starting_bus_groups=action_set.get_simplified_starting_bus_groups(),
            bus_group_diffs=bus_group_diffs,
        )
        action_set = action_set.model_copy(update={"local_actions": local_actions})
    return action_set

load_action_set

load_action_set(json_file_path, diff_file_path)

Load an action set from a file.

PARAMETER	DESCRIPTION
json_file_path	The path to the JSON file containing the action set without local actions. TYPE: Union[str, Path]
diff_file_path	The path to the HDF5 file containing the station diffs to expand to local actions. If this is none, the local_actions field will not be filled and be the empty list. TYPE: Union[str, Path] | None

RETURNS	DESCRIPTION
ActionSet	The action set loaded from the file.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/stored_action_set.py

def load_action_set(json_file_path: Union[str, Path], diff_file_path: Union[str, Path] | None) -> ActionSet:
    """Load an action set from a file.

    Parameters
    ----------
    json_file_path : Union[str, Path]
        The path to the JSON file containing the action set without local actions.
    diff_file_path : Union[str, Path] | None
        The path to the HDF5 file containing the station diffs to expand to local actions. If this is none, the
        local_actions field will not be filled and be the empty list.

    Returns
    -------
    ActionSet
        The action set loaded from the file.
    """
    return load_action_set_fs(LocalFileSystem(), json_file_path=json_file_path, diff_file_path=diff_file_path)

save_action_set_fs

save_action_set_fs(
    filesystem,
    json_file_path,
    diff_file_path,
    action_set,
    validate_diff_hypothesis=False,
    revalidate_action_set=True,
)

Save an action set to a file system.

PARAMETER	DESCRIPTION
filesystem	The file system to use to save the action set. TYPE: AbstractFileSystem
json_file_path	The path to the JSON file to save the pydantic payload. TYPE: Union[str, Path]
diff_file_path	The path to the HDF5 file to save the station diffs. TYPE: Union[str, Path]
action_set	The action set to save. TYPE: ActionSet
validate_diff_hypothesis	Whether to validate that local action changes only affect coupler open states and switching tables. This is intended for debugging and can make saving slower. TYPE: bool DEFAULT: False
revalidate_action_set	Whether to round-trip the action set through Pydantic validation before saving. Disable this in hot paths when the caller already constructed a validated ActionSet. TYPE: bool DEFAULT: True

Notes

The JSON payload stores only the reference stations and scalar metadata. local_actions are serialized separately as dense station diffs in HDF5 to avoid repeating unchanged station payloads.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/stored_action_set.py

def save_action_set_fs(
    filesystem: AbstractFileSystem,
    json_file_path: Union[str, Path],
    diff_file_path: Union[str, Path],
    action_set: ActionSet,
    validate_diff_hypothesis: bool = False,
    revalidate_action_set: bool = True,
) -> None:
    """Save an action set to a file system.

    Parameters
    ----------
    filesystem : AbstractFileSystem
        The file system to use to save the action set.
    json_file_path : Union[str, Path]
        The path to the JSON file to save the pydantic payload.
    diff_file_path : Union[str, Path]
        The path to the HDF5 file to save the station diffs.
    action_set : ActionSet
        The action set to save.
    validate_diff_hypothesis : bool
        Whether to validate that local action changes only affect coupler open states and switching tables.
        This is intended for debugging and can make saving slower.
    revalidate_action_set : bool
        Whether to round-trip the action set through Pydantic validation before saving.
        Disable this in hot paths when the caller already constructed a validated ``ActionSet``.

    Notes
    -----
    The JSON payload stores only the reference stations and scalar metadata. ``local_actions`` are
    serialized separately as dense station diffs in HDF5 to avoid repeating unchanged station payloads.
    """
    if revalidate_action_set:
        action_set = ActionSet.model_validate(action_set.model_dump(mode="python", round_trip=True))
    bus_group_diffs = compress_actions_to_bus_group_diffs_from_starting_bus_groups(
        starting_bus_groups=action_set.get_simplified_starting_bus_groups(),
        actions=action_set.local_actions,
        validate_diff_hypothesis=validate_diff_hypothesis,
    )

    # local_actions are persisted in the HDF5 file as compressed station diffs.
    action_set_without_local_actions = action_set.model_copy(update={"local_actions": []})
    with filesystem.open(str(json_file_path), "w") as f:
        f.write(action_set_without_local_actions.model_dump_json(indent=2, exclude_none=True))
    store_bus_group_diff_fs(filesystem, bus_group_diffs, diff_file_path)

save_action_set

save_action_set(
    json_file_path,
    diff_file_path,
    action_set,
    validate_diff_hypothesis=False,
    revalidate_action_set=True,
)

Save an action set to a file.

PARAMETER	DESCRIPTION
json_file_path	The path to the JSON file to save the pydantic payload. TYPE: Union[str, Path]
diff_file_path	The path to the HDF5 file to save the station diffs. TYPE: Union[str, Path]
action_set	The action set to save. TYPE: ActionSet
validate_diff_hypothesis	Whether to validate that local action changes only affect coupler open states and switching tables. This is intended for debugging and can make saving slower. TYPE: bool DEFAULT: False
revalidate_action_set	Whether to round-trip the action set through Pydantic validation before saving. TYPE: bool DEFAULT: True

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/stored_action_set.py

def save_action_set(
    json_file_path: Union[str, Path],
    diff_file_path: Union[str, Path],
    action_set: ActionSet,
    validate_diff_hypothesis: bool = False,
    revalidate_action_set: bool = True,
) -> None:
    """Save an action set to a file.

    Parameters
    ----------
    json_file_path : Union[str, Path]
        The path to the JSON file to save the pydantic payload.
    diff_file_path : Union[str, Path]
        The path to the HDF5 file to save the station diffs.
    action_set : ActionSet
        The action set to save.
    validate_diff_hypothesis : bool
        Whether to validate that local action changes only affect coupler open states and switching tables.
        This is intended for debugging and can make saving slower.
    revalidate_action_set : bool
        Whether to round-trip the action set through Pydantic validation before saving.

    """
    save_action_set_fs(
        filesystem=LocalFileSystem(),
        json_file_path=json_file_path,
        diff_file_path=diff_file_path,
        action_set=action_set,
        validate_diff_hypothesis=validate_diff_hypothesis,
        revalidate_action_set=revalidate_action_set,
    )

random_actions

random_actions(action_set, rng, n_split_subs)

Sample a random topology from the action set.

Makes sure to sample each substation at most once.

PARAMETER	DESCRIPTION
action_set	The action set to sample the random topology from. TYPE: ActionSet
rng	The random number generator to use. TYPE: Generator
n_split_subs	The number of substations to split. If this is more than total number of substations, all substations are split. (i.e. will be clipped to the number of substations) TYPE: int

RETURNS	DESCRIPTION
list[int]	A list of indices of the action set with substations to split.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/stored_action_set.py

def random_actions(action_set: ActionSet, rng: np.random.Generator, n_split_subs: int) -> list[int]:
    """Sample a random topology from the action set.

    Makes sure to sample each substation at most once.

    Parameters
    ----------
    action_set : ActionSet
        The action set to sample the random topology from.
    rng : np.random.Generator
        The random number generator to use.
    n_split_subs : int
        The number of substations to split. If this is more than total number of substations, all substations are split.
        (i.e. will be clipped to the number of substations)

    Returns
    -------
    list[int]
        A list of indices of the action set with substations to split.
    """
    # First sample the substations to split
    substations = list(set(station.bus_group_id for station in action_set.local_actions))
    substations.sort()  # Sort to make sure the order is deterministic for the same random seed
    sub_choice = rng.choice(substations, size=min(n_split_subs, len(substations)), replace=False).tolist()

    # Then sample an action for each substation
    actions = []
    for grid_model_id in sub_choice:
        applicable_indices = [
            i for i, station in enumerate(action_set.local_actions) if station.bus_group_id == grid_model_id
        ]
        actions.append(rng.choice(applicable_indices).item())
    return actions

Messages

toop_engine_interfaces.messages.preprocess

toop_engine_interfaces.messages.lf_service

toop_engine_interfaces.messages.lf_service.loadflow_commands

Describes the interfaces for a loadflow service providing N-1 computations in AC or DC to customers.

The communication follows a 2 step pattern:

1. Grid load - Load a grid file into the engine and potentially perform some preprocessing. The engine should return a grid reference upon this call, which is used in the job to reference the grid. Multiple grids can be loaded at the same time, it is the responsibility of the engine to perform memory management (i.e. swap out a grid to disk if memory is full).
2. Execute jobs - Run a loadflow job on the engine. The job references the grid that was loaded in step 2. A call to this can contain multiple jobs, allowing the engine to parallelize over jobs in addition to parallelizing over timesteps/outages.

The LoadflowEngine protocol describes this two step process in detail.

BaseFilter

Bases: BaseModel, ABC

A base class for filters.

The idea behind filters is to implement logics that can not be implemented with the monitored elements directly but that are dependent on the loadflow results

This is not to be used directly, but to be subclassed.

WorstContingencyBranchFilter

Bases: BaseFilter

If this filter is applied, it will reduce the branch results to only the worst N-1 case/side per branch and timestep.

Worst is determined with respect to the loading value. Branches that don't have a rating will never be returned.

If there are multiple worst N-1 cases/sides that produce a tie, one will be chosen at random. This filtering happens on a per-timestep basis, meaning for every timestep there should be as many results as monitored branches, but they can refer to different N-1 cases.

return_basecase class-attribute instance-attribute

return_basecase = False

Whether to return the basecase still. If this is set to True, the basecase will always be returned even if it is not the worst case.

filter_type class-attribute instance-attribute

filter_type = 'worst_contingency'

An identifier for the discriminated union

VoltageBandFilter

Bases: BaseFilter

If this filter is applied, it will reduce the node results to only the results that are outside of a specified band.

The band is defined by a minimum and maximum p.u. value for all nodes

return_basecase class-attribute instance-attribute

return_basecase = False

Whether to return the basecase at all times. If this is set to True, the basecase will always be returned even if it is inside the band and hence should be filtered out.

v_min instance-attribute

v_min

The minimum voltage in p.u. - values below this will be returned.

v_max instance-attribute

v_max

The maximum voltage in p.u. - values above this will be returned

filter_type class-attribute instance-attribute

filter_type = 'voltage_band'

An identifier for the discriminated union

PercentCutoffBranchFilter

Bases: BaseFilter

Filter, if applied returns only branch results that are above a loading threshold.

Elements for which no loading could be computed (e.g. due to missing ratings) are never returned. This filtering happens on a per-timestep basis, i.e. if a branch/contingency is above the threshold in one timestep, it will be returned in exactly that timestep.

loading_threshold instance-attribute

loading_threshold

The loading threshold in percent. Only branches with a loading above this threshold are returned.

filter_type class-attribute instance-attribute

filter_type = 'percent_cutoff'

An identifier for the discriminated union

Job

Bases: BaseModel

A job constitutes a single workload and will produce a LoadflowResults object.

There are different types of jobs based on the workload, the simple being a base job with no changes to the base grid.

id instance-attribute

id

A unique identifier for the job. This is used to reference the job in the results.

branch_filter class-attribute instance-attribute

branch_filter = Field(
    default=None, discriminator="filter_type"
)

Filters for the branch results table. Exactly one filter can be active per table and job

node_filter class-attribute instance-attribute

node_filter = Field(
    default=None, discriminator="filter_type"
)

Filters for the node results table. Exactly one filter can be active per table and job

job_type class-attribute instance-attribute

job_type = 'bare'

An identifier for the discriminated union

timestep_subselection class-attribute instance-attribute

timestep_subselection = None

If this is set, only the timesteps in this list are computed. If this is not set, all timesteps are computed. Timesteps are referenced by their index in the grid file, starting at 0.

JobWithCGMESChanges

Bases: Job

A job that includes changes in CGMES format. This is only applicable if the grid is a CGMES grid

tp_files instance-attribute

tp_files

The file including the topology changes that shall be applied.

There must be as many entries as timesteps in the grid, but the same file can be referenced multiple times.

ssh_files instance-attribute

ssh_files

The file including the state/injection changes that shall be applied.

There must be as many entries as timesteps in the grid, but the same file can be referenced multiple times.

job_type class-attribute instance-attribute

job_type = 'cgmes_changes'

An identifier for the discriminated union

id instance-attribute

id

A unique identifier for the job. This is used to reference the job in the results.

branch_filter class-attribute instance-attribute

branch_filter = Field(
    default=None, discriminator="filter_type"
)

Filters for the branch results table. Exactly one filter can be active per table and job

node_filter class-attribute instance-attribute

node_filter = Field(
    default=None, discriminator="filter_type"
)

Filters for the node results table. Exactly one filter can be active per table and job

timestep_subselection class-attribute instance-attribute

timestep_subselection = None

If this is set, only the timesteps in this list are computed. If this is not set, all timesteps are computed. Timesteps are referenced by their index in the grid file, starting at 0.

InjectionAddition

Bases: BaseModel

A single addition of an injection at a node.

This feature only support PQ nodes, if attempted to apply to a branch, pv node or slack node, the engine should ignore this addition and log a warning.

Positive values shall have the same effect as sgens, i.e. power is produced, while negative values will have the same effect as loads, i.e. power is consumed.

node instance-attribute

node

The node to which the injection is added

p_mw instance-attribute

p_mw

The active power in MW that is added to the node

q_mw instance-attribute

q_mw

The reactive power that is added to the node

timestep_subselection class-attribute instance-attribute

timestep_subselection = None

If this is given, the addition only happens in the timesteps that are in this list. If this is not given, the addition happens in all timesteps.

JobWithInjectionAdditions

Bases: Job

Adds a constant injection to a node in the grid.

This feature assumes all injections are added to PQ nodes - otherwise they will be ignored.

Positive values shall have the same effect as sgens, i.e. power is produced, while negative values will have the same effect as loads, i.e. power is consumed.

additions instance-attribute

additions

The injections that are added to the grid

job_type class-attribute instance-attribute

job_type = 'injection_additions'

An identifier for the discriminated union

id instance-attribute

id

A unique identifier for the job. This is used to reference the job in the results.

branch_filter class-attribute instance-attribute

branch_filter = Field(
    default=None, discriminator="filter_type"
)

Filters for the branch results table. Exactly one filter can be active per table and job

node_filter class-attribute instance-attribute

node_filter = Field(
    default=None, discriminator="filter_type"
)

Filters for the node results table. Exactly one filter can be active per table and job

timestep_subselection class-attribute instance-attribute

timestep_subselection = None

If this is set, only the timesteps in this list are computed. If this is not set, all timesteps are computed. Timesteps are referenced by their index in the grid file, starting at 0.

BaseGrid

Bases: BaseModel, ABC

A base class for grid files. This is not to be used directly, but to be subclassed

n_1_definition class-attribute instance-attribute

n_1_definition = None

The N-1 cases that are to be computed. If this is provided, this shall overwrite the N-1 cases that are defined in the grid files if the format supports such definition. If this is not provided, the N-1 cases that are defined in the grid files shall be used. If neither is provided, the engine should throw an error.

grid_type instance-attribute

grid_type

An identifier for the discriminated union, to be set by the subclasses

CGMESGrid

Bases: BaseGrid

A CGMES grid file does not need to store much additional information

grid_files instance-attribute

grid_files

A list of paths to grid files. This can include multiple .tp and .ssh files which are to be interpreted as multiple timesteps. If a .tp and a .ssh file have the same filename or the same timestep metadata inside the file, they correspond to the same timestep. Timesteps should be sorted by the timestep information inside the CGMES files.

grid_type class-attribute instance-attribute

grid_type = 'cgmes'

An identifier for the discriminated union

n_1_definition class-attribute instance-attribute

n_1_definition = None

The N-1 cases that are to be computed. If this is provided, this shall overwrite the N-1 cases that are defined in the grid files if the format supports such definition. If this is not provided, the N-1 cases that are defined in the grid files shall be used. If neither is provided, the engine should throw an error.

UCTEGrid

Bases: BaseGrid

A list of UCTE files that are to be loaded into the engine

grid_files instance-attribute

grid_files

A list of paths to grid files. This can include multiple .ucte files which are to be interpreted as multiple timesteps. Timesteps should be interpreted in the order of this list

grid_type class-attribute instance-attribute

grid_type = 'ucte'

An identifier for the discriminated union

n_1_definition class-attribute instance-attribute

n_1_definition = None

The N-1 cases that are to be computed. If this is provided, this shall overwrite the N-1 cases that are defined in the grid files if the format supports such definition. If this is not provided, the N-1 cases that are defined in the grid files shall be used. If neither is provided, the engine should throw an error.

PowsyblGrid

Bases: BaseGrid

A list of powsybl xiidm files that are to be loaded into the engine

grid_files instance-attribute

grid_files

A list of xiidm files that represent the timesteps. Timesteps should be interpreted in the order of this list

grid_type class-attribute instance-attribute

grid_type = 'powsybl'

n_1_definition class-attribute instance-attribute

n_1_definition = None

The N-1 cases that are to be computed. If this is provided, this shall overwrite the N-1 cases that are defined in the grid files if the format supports such definition. If this is not provided, the N-1 cases that are defined in the grid files shall be used. If neither is provided, the engine should throw an error.

PandapowerGrid

Bases: BaseGrid

A list of pandapower json files that are to be loaded into the engine

grid_files instance-attribute

grid_files

A list of pandapower files that represent the timesteps. Timesteps should be interpreted in the order of this list

grid_type class-attribute instance-attribute

grid_type = 'pandapower'

n_1_definition class-attribute instance-attribute

n_1_definition = None

The N-1 cases that are to be computed. If this is provided, this shall overwrite the N-1 cases that are defined in the grid files if the format supports such definition. If this is not provided, the N-1 cases that are defined in the grid files shall be used. If neither is provided, the engine should throw an error.

StartCalculationCommand

Bases: BaseModel

A command to run a list of jobs on the engine.

This can involve multiple N-1 computations with different changes to the base grid, but a job must share the same grid file.

loadflow_id instance-attribute

loadflow_id

A unique identifier for the loadflow run. This is used to identify the result

grid_data class-attribute instance-attribute

grid_data = Field(discriminator='grid_type')

The string that was returned by load_grid, identifying the grid file that this job collection shall run on

method instance-attribute

method

The method that is to be used for the loadflow computations. This can be either AC or DC. This must be the same for all jobs in the list

jobs instance-attribute

jobs

The jobs to be executed

LoadflowEngine

Bases: Protocol

A protocol for a loadflow engine.

Roughly, an engine shall be able to load grids and execute jobs on them. There is some memory management that the engine needs to perform internally, i.e. it could happen that two users want to use the same engine in parallel. In that case, the engine should swap out grids to disk if memory is full.

run_job

run_job(job)

Run a job on the engine.

This can involve multiple N-1 computations with different changes to the base grid, identified through multiple jobs in the BatchJob. The engine should return the results of the jobs as in-memory dataframes.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/messages/lf_service/loadflow_commands.py

def run_job(self, job: StartCalculationCommand) -> list[LoadflowResults]:
    """Run a job on the engine.

    This can involve multiple N-1 computations with different changes to the base grid,
    identified through multiple jobs in the BatchJob. The engine should return the results of the jobs as in-memory
    dataframes.
    """

ShutdownCommand

Bases: BaseModel

A command to shut down the preprocessing worker

exit_code class-attribute instance-attribute

exit_code = 0

The exit code to return

LoadflowServiceCommand

Bases: BaseModel

A wrapper to aid deserialization

command instance-attribute

command

The actual command posted

timestamp class-attribute instance-attribute

timestamp = Field(
    default_factory=lambda: str(datetime.now())
)

When the command was sent

uuid class-attribute instance-attribute

uuid = Field(default_factory=lambda: str(uuid.uuid4()))

A unique identifier for this command message, used to avoid duplicates during processing

toop_engine_interfaces.messages.lf_service.loadflow_heartbeat

Loadflow Heartbeat Commands for the kafka worker.

LoadflowStatusInfo

Bases: BaseModel

A status info to inform about an ongoint Loadflow solving action.

loadflow_id instance-attribute

loadflow_id

The id of the loadflow solving job.

runtime instance-attribute

runtime

The amount of time since the start of the optimization.

message class-attribute instance-attribute

message = ''

An optional message

LoadflowHeartbeat

Bases: BaseModel

A message class for heartbeats from the loadflow worker.

When idle, this just sends a hello, and when solving it also conveys the current status of the Loadflow Analysis

idle instance-attribute

idle

Whether the worker is idle

status_info instance-attribute

status_info

If not idle, a status update

timestamp class-attribute instance-attribute

timestamp = Field(
    default_factory=lambda: str(datetime.now())
)

When the heartbeat was sent

uuid class-attribute instance-attribute

uuid = Field(default_factory=lambda: str(uuid.uuid4()))

A unique identifier for this heartbeat message, used to avoid duplicates during processing

toop_engine_interfaces.messages.lf_service.stored_loadflow_reference

A reference to a stored loadflow, in a separate file because many locations started to reference it.

Also, we might add helper functions in the future like load/save to azure bucket.

StoredLoadflowReference

Bases: BaseModel

A reference to a stored loadflow result on disk or in an object store.

Loadflow results are too large to be sent directly over kafka, so they need to be stored somewhere and referenced. They are stored and written using the functions in loadflow_result_helpers_new.py, which use the fsspec library to abstract away the filesystem. Hence, these can write to local disk, Azure bucket, ...

The reference contains the filename relative to the base path or bucket defined in the filesystem, i.e. if a DirFileSystem is used with base_path="/path/to/base" and the filename is "loadflows" then the full path is "/path/to/base/loadflows/node_results.parquet", "/path/to/base/loadflows/branch_results.parquet", ...

relative_path instance-attribute

relative_path

The folder of the loadflow result relative to the base path or bucket in the filesystem. This points to a directory under which the files "node_results.parquet", "branch_results.parquet", "metadata.json", ... are stored.

toop_engine_interfaces.messages.protobuf_message_factory

Module: protobuf_message_factory

This module provides utility functions for serializing and deserializing messages using the MessageWrapper protobuf schema. It enables conversion between string messages and their protobuf-encoded byte representations for efficient message exchange.

FUNCTION	DESCRIPTION
- serialize_message	Serializes a string message into bytes using the MessageWrapper protobuf schema.
- deserialize_message	Deserializes bytes into a string message using the MessageWrapper protobuf schema.

serialize_message

serialize_message(message)

Serialize a message string into bytes using MessageWrapper.

PARAMETER	DESCRIPTION
message	The message to be serialized. TYPE: str

RETURNS	DESCRIPTION
bytes	The serialized message as a bytes object.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/messages/protobuf_message_factory.py

def serialize_message(message: str) -> bytes:
    """
    Serialize a message string into bytes using MessageWrapper.

    Parameters
    ----------
    message : str
        The message to be serialized.

    Returns
    -------
    bytes
        The serialized message as a bytes object.
    """
    mw = MessageWrapper(message=message)
    return mw.SerializeToString()

deserialize_message

deserialize_message(msg_bytes)

Deserializes a protobuf message from bytes and returns its string representation.

PARAMETER	DESCRIPTION
msg_bytes	The serialized protobuf message as a byte string. TYPE: bytes

RETURNS	DESCRIPTION
str	The deserialized message as a string.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/messages/protobuf_message_factory.py

def deserialize_message(msg_bytes: bytes) -> str:
    """
    Deserializes a protobuf message from bytes and returns its string representation.

    Parameters
    ----------
    msg_bytes : bytes
        The serialized protobuf message as a byte string.

    Returns
    -------
    str
        The deserialized message as a string.
    """
    mw = MessageWrapper()
    mw.ParseFromString(msg_bytes)
    return mw.message

toop_engine_interfaces.messages.preprocess.preprocess_results

Contains the result classes for the preprocessing process.

ImportResult

Bases: BaseModel

Statistics and results from an import process of UCTE data

data_folder instance-attribute

data_folder

The path where the entry point where the timestep data folder structure starts. The folder structure is defined in dc_solver.interfaces.folder_structure. Can be on a temp dir

n_relevant_subs class-attribute instance-attribute

n_relevant_subs = 0

The number of relevant substations

n_low_impedance_lines class-attribute instance-attribute

n_low_impedance_lines = 0

The number of low impedance lines that have been converted to a switch

n_branch_across_switch class-attribute instance-attribute

n_branch_across_switch = 0

The number of branches across a switch that have been removed

n_line_for_nminus1 class-attribute instance-attribute

n_line_for_nminus1 = 0

The number of lines in the N-1 definition

n_line_for_reward class-attribute instance-attribute

n_line_for_reward = 0

The number of lines that are observed

n_line_disconnectable class-attribute instance-attribute

n_line_disconnectable = 0

The number of lines that are disconnectable

n_trafo_for_nminus1 class-attribute instance-attribute

n_trafo_for_nminus1 = 0

The number of trafos in the N-1 definition

n_trafo_for_reward class-attribute instance-attribute

n_trafo_for_reward = 0

The number of trafos that are observed

n_trafo_disconnectable class-attribute instance-attribute

n_trafo_disconnectable = 0

The number of trafos in the N-1 definition

n_tie_line_for_reward class-attribute instance-attribute

n_tie_line_for_reward = 0

The number of tie lines that are observed

n_tie_line_for_nminus1 class-attribute instance-attribute

n_tie_line_for_nminus1 = 0

The number of tie lines in the N-1 definition

n_tie_line_disconnectable class-attribute instance-attribute

n_tie_line_disconnectable = 0

The number of tie lines that are disconnectable

n_boundary_line_for_nminus1 class-attribute instance-attribute

n_boundary_line_for_nminus1 = 0

The number of boundary lines in the N-1 definition

n_generator_for_nminus1 class-attribute instance-attribute

n_generator_for_nminus1 = 0

The number of generators in the N-1 definition

n_load_for_nminus1 class-attribute instance-attribute

n_load_for_nminus1 = 0

The number of loads in the N-1 definition

n_switch_for_nminus1 class-attribute instance-attribute

n_switch_for_nminus1 = 0

The number of switches in the N-1 definition

n_switch_for_reward class-attribute instance-attribute

n_switch_for_reward = 0

The number of switches that are observed

n_white_list class-attribute instance-attribute

n_white_list = 0

The number of elements in the whitelist in total

n_white_list_applied class-attribute instance-attribute

n_white_list_applied = 0

The number of elements in the whitelist that were successfully matched and applied

n_black_list class-attribute instance-attribute

n_black_list = 0

The number of elements in the blacklist in total

n_black_list_applied class-attribute instance-attribute

n_black_list_applied = 0

The number of elements in the blacklist that were successfully matched and applied

grid_type class-attribute instance-attribute

grid_type = 'cgmes'

The type of grid that was imported, e.g. ucte or cgmes

DynamicInformationStats

Bases: BaseModel

Stats about the static information class

time class-attribute instance-attribute

time = None

The timestep that was optimized, if given

fp_dtype class-attribute instance-attribute

fp_dtype = ''

A string representation of the floating point type used in the static informations, e.g. 'float32' or 'float64'.

device class-attribute instance-attribute

device = ''

A string representation of the device(s) the static information arrays are placed on, e.g. 'TFRT_CPU_0' or 'cuda:0'. If the arrays are replicated or sharded across multiple devices, all of them are listed comma-separated.

total_size_bytes class-attribute instance-attribute

total_size_bytes = 0

The total storage space of all arrays in the dynamic information, in bytes. This is the logical size of the arrays, so a replicated array is counted once and not once per device.

ptdf_size_bytes class-attribute instance-attribute

ptdf_size_bytes = 0

The storage space of the PTDF matrix alone, in bytes. Part of total_size_bytes.

action_set_size_bytes class-attribute instance-attribute

action_set_size_bytes = 0

The storage space of the action set, in bytes, excluding the busbar outage data that is stored within it. Part of total_size_bytes and disjoint from bb_outage_size_bytes.

bb_outage_size_bytes class-attribute instance-attribute

bb_outage_size_bytes = 0

The storage space of all busbar outage data, in bytes. This covers the rel_bb_outage_data inside the action set, the non_rel_bb_outage_data and the bb_outage_baseline_analysis. It is 0 if the static information was preprocessed without busbar outages. Part of total_size_bytes.

has_double_limits class-attribute instance-attribute

has_double_limits = False

Whether the static information has max_mw_flow_limited set or not

n_branches class-attribute instance-attribute

n_branches = 0

The number of branches in the PTDF matrix

n_nodes class-attribute instance-attribute

n_nodes = 0

The number of nodes in the PTDF matrix

n_branch_outages class-attribute instance-attribute

n_branch_outages = 0

How many branch outages are part of the N-1 computation

n_multi_outages class-attribute instance-attribute

n_multi_outages = 0

How many multi-outages are part of the N-1 computation

n_injection_outages class-attribute instance-attribute

n_injection_outages = 0

How many injection outages are part of the N-1 computation

n_busbar_outages class-attribute instance-attribute

n_busbar_outages = 0

How many busbar outages are part of the N-1 computation

n_nminus1_cases class-attribute instance-attribute

n_nminus1_cases = 0

How many N-1 cases are there in total

n_controllable_psts class-attribute instance-attribute

n_controllable_psts = 0

How many controllable phase shifting transformers are in the grid

n_monitored_branches class-attribute instance-attribute

n_monitored_branches = 0

How many branches are monitored

n_timesteps class-attribute instance-attribute

n_timesteps = 0

How many timesteps are optimized at the same time

n_relevant_subs class-attribute instance-attribute

n_relevant_subs = 0

How many relevant substations are in the grid

n_disc_branches class-attribute instance-attribute

n_disc_branches = 0

How many disconnectable branches are in the definition

overload_energy_n0 class-attribute instance-attribute

overload_energy_n0 = 0.0

What is the N-0 overload energy of the unsplit configuration

overload_energy_n1 class-attribute instance-attribute

overload_energy_n1 = 0.0

What is the N-1 overload energy of the unsplit configuration

n_actions class-attribute instance-attribute

n_actions = 0

How many actions have been precomputed in the action set. This is the size of the branch action set, note that combinations of actions within that set are possible (product set wise) if multiple substations are split

max_station_branch_degree class-attribute instance-attribute

max_station_branch_degree = 0

The maximum number of branches connected to any station in the grid

max_station_injection_degree class-attribute instance-attribute

max_station_injection_degree = 0

The maximum number of injections connected to any station in the grid

max_reassignment_distance class-attribute instance-attribute

max_reassignment_distance = 0

The maximum reassignment distance associated with any action

PreprocessingSuccessResult

Bases: BaseModel

Results of a preprocessing run, mainly including the static_information and network_data files.

data_folder instance-attribute

data_folder

The path where the entry point where the timestep data folder structure starts. The folder structure is defined in dc_solver.interfaces.folder_structure. Can be on a temp dir

initial_loadflow instance-attribute

initial_loadflow

The initial AC loadflow results, i.e. the N-1 analysis without any actions applied to the grid.

initial_metrics instance-attribute

initial_metrics

The initial metrics computed for the loadflow results

static_information_stats instance-attribute

static_information_stats

Statistics about the static information file that was produced

importer_results instance-attribute

importer_results

The results of the importer process

result_type class-attribute instance-attribute

result_type = 'preprocessing_success'

The discriminator for the Result Union

PreprocessingStartedResult

Bases: BaseModel

A message that is sent when the preprocessing process has started

result_type class-attribute instance-attribute

result_type = 'preprocessing_started'

The discriminator for the Result Union

ErrorResult

Bases: BaseModel

A message that is sent if an error occurred

error instance-attribute

error

The error message

result_type class-attribute instance-attribute

result_type = 'error'

The discriminator for the Result Union

Result

Bases: BaseModel

A generic class for result, holding either a successful or an unsuccessful result

preprocess_id instance-attribute

preprocess_id

The preprocess_id that was sent in the preprocess_command, used to identify the result

instance_id class-attribute instance-attribute

instance_id = ''

The instance id of the importer worker that created this result

runtime instance-attribute

runtime

The runtime in seconds that the preprocessing took until the result

result class-attribute instance-attribute

result = Field(discriminator='result_type')

The actual result data in a discriminated union

uuid class-attribute instance-attribute

uuid = Field(default_factory=lambda: str(uuid.uuid4()))

A unique identifier for this result message, used to avoid duplicates during processing

timestamp class-attribute instance-attribute

timestamp = Field(
    default_factory=lambda: str(datetime.now())
)

When the result was sent

toop_engine_interfaces.messages.preprocess.preprocess_commands

Defines the commands that can be sent to a preprocessing worker.

UCTERegionType module-attribute

UCTERegionType = Literal[
    "A",
    "B",
    "C",
    "D",
    "D1",
    "D2",
    "D4",
    "D6",
    "D7",
    "D8",
    "E",
    "F",
    "G",
    "H",
    "I",
    "J",
    "K",
    "L",
    "M",
    "N",
    "O",
    "P",
    "Q",
    "R",
    "S",
    "T",
    "U",
    "V",
    "W",
    "X",
    "Y",
    "Z",
    "0",
    "2",
    "_",
]

CGMESRegionType module-attribute

CGMESRegionType = Literal[
    "AL",
    "AD",
    "AM",
    "AT",
    "AZ",
    "BY",
    "BE",
    "BA",
    "BG",
    "HR",
    "CY",
    "CZ",
    "DK",
    "EE",
    "FI",
    "FR",
    "GE",
    "DE",
    "GR",
    "HU",
    "IS",
    "IE",
    "IT",
    "LV",
    "LI",
    "LT",
    "LU",
    "MT",
    "MD",
    "MC",
    "ME",
    "NL",
    "MK",
    "NO",
    "PL",
    "PT",
    "RO",
    "RU",
    "SM",
    "RS",
    "SK",
    "SI",
    "ES",
    "SE",
    "CH",
    "TR",
    "UA",
    "GB",
    "VA",
    "HK",
]

AllCountriesRegionType module-attribute

AllCountriesRegionType = Literal['']

CustomRegionType module-attribute

CustomRegionType = Literal['S3']

RegionType module-attribute

RegionType = Union[
    UCTERegionType,
    CGMESRegionType,
    AllCountriesRegionType,
    CustomRegionType,
]

GridModelType module-attribute

GridModelType = Literal['ucte', 'cgmes']

LimitAdjustmentParameters

Bases: BaseModel

Parameters for the adjustment of limits of special branches.

The new operational limits will be created like this: 1) Compute AC Loadflow 2) new_limit = current_flow * n_0_factor (or n_1_factor) 3) If new_limit > min_increase (=old_limit * n_0_min_increase) new_limit = min_increase 4) If new_limit > old_limit new_limit = old_limit

n_0_factor class-attribute instance-attribute

n_0_factor = 1.2

The factor for the N-0 current limit. Default is 1.2

n_1_factor class-attribute instance-attribute

n_1_factor = 1.4

The factor for the N-1 current limit. Default is 1.4

n_0_min_increase class-attribute instance-attribute

n_0_min_increase = 0.05

The minimal allowed load increase in percent for the n-0 case. This value multiplied with the current limit gives a lower border for the new limit. This makes sure, that lines that currently are barely loaded can be used at all Default is 5%.

n_1_min_increase class-attribute instance-attribute

n_1_min_increase = 0.05

The minimal allowed load increase in percent for the n-1 case. This value multiplied with the current limit gives a lower border for the new limit. This makes sure, that lines that currently are barely loaded can be used at all Default is 5%.

get_parameters_for_case

get_parameters_for_case(case)

Get the factors for the specific case

PARAMETER	DESCRIPTION
case	Which case should be returned TYPE: Literal[n0, n1]

RETURNS	DESCRIPTION
tuple[PositiveFloat, PositiveFloat]	The factor and the minimal increase

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/messages/preprocess/preprocess_commands.py

def get_parameters_for_case(self, case: Literal["n0", "n1"]) -> tuple[PositiveFloat, PositiveFloat]:
    """Get the factors for the specific case

    Parameters
    ----------
    case: Literal["n0", "n1"]
        Which case should be returned

    Returns
    -------
    tuple[PositiveFloat, PositiveFloat]
        The factor and the minimal increase
    """
    if case == "n0":
        return self.n_0_factor, self.n_0_min_increase
    if case == "n1":
        return self.n_1_factor, self.n_1_min_increase
    raise ValueError(f"Case {case} not defined")

AreaSettings

Bases: BaseModel

Setting related to the areas that are imported

control_area instance-attribute

control_area

The area in which switching can take place. Substations from this area will automatically become relevant substations (switchable) except they are below the cutoff voltage. Also lines in this area will become disconnectable.

view_area instance-attribute

view_area

The areas in which branches shall be part of the overload computation, i.e. for which regions shall line flows be computed.

nminus1_area instance-attribute

nminus1_area

The areas where elements shall be part of the N-1 computation, i.e. which elements to fail.

cutoff_voltage class-attribute instance-attribute

cutoff_voltage = 220

The cutoff voltage under which to ignore equipment. Equipment that doesn't have at least one end equal or above this nominal voltage will not be part of the reward/nminus1 computation

dso_trafo_factors class-attribute instance-attribute

dso_trafo_factors = None

If given, the N-0 and N-1 flows across the dso trafos in the specied region will be limited to the current N-0 flows in the unsplit configuration. For each case (n0 or n1) a new operational limit with the name "border_limit_n0"/"border_limit_n1" is added.

dso_trafo_weight class-attribute instance-attribute

dso_trafo_weight = 1.0

A weight that is used for trafos that leave the n-1 area, to underlying DSOs

border_line_factors class-attribute instance-attribute

border_line_factors = None

If given, the N-0 and N-1 flows across the border lines leaving or entering the specied region will be limited to the current N-0 flows in the unsplit configuration. For each case (n0 or n1) a new operational limit with the name "border_limit_n0"/"border_limit_n1" is added.

border_line_weight class-attribute instance-attribute

border_line_weight = 1.0

A weight that is used for lines that leave the n-1 area, to neighbouring TSOs

RelevantStationRules

Bases: BaseModel

Rules to determine whether a substation is relevant or not.

min_busbars class-attribute instance-attribute

min_busbars = 2

The minimum number of busbars a substation must have to be considered relevant.

min_connected_branches class-attribute instance-attribute

min_connected_branches = 4

The minimum number of connected branches a substation must have to be considered relevant. This only counts branches (lines, transformers, tie-lines), not injections (generators, loads, shunts, etc.).

min_connected_elements class-attribute instance-attribute

min_connected_elements = 4

The minimum number of connected elements a substation must have to be considered relevant. This includes branches and injections (generators, loads, shunts, etc.).

BaseImporterParameters

Bases: BaseModel

Parameters that are required to import any data format.

area_settings instance-attribute

area_settings

Which areas of the grid are to be imported and how to handle boundaries

fail_on_non_convergence class-attribute instance-attribute

fail_on_non_convergence = True

Whether to raise an error if the loadflow does not converge in the basecase. If set to False, the preprocessing will continue and dc-optimization can still happen. Mostly for debugging purposes

data_folder instance-attribute

data_folder

The path where the entry point where the timestep data folder structure starts.

The folder structure is defined in interfaces.folder_structure. This folder is relative to the processed_grid_folder that is configured in the backend/importer. A typical default would be grid_model_file.stem

grid_model_file instance-attribute

grid_model_file

The path to the input grid model file.

This file should contain the grid model in the format defined by the data_type. For instance a .uct for UCTE data or a .zip for CGMES data.

data_type instance-attribute

data_type

The type of data that is being imported.

This will determine the importer that is used to load the data.

white_list_file class-attribute instance-attribute

white_list_file = None

The path to the white lists if present

black_list_file class-attribute instance-attribute

black_list_file = None

The path to the balck lists if present

ignore_list_file class-attribute instance-attribute

ignore_list_file = None

The path to the ignore lists if present

A csv file with the following columns: grid_model_id, reason

The implementation is expected to ignore all elements that are in the ignore list.

select_by_voltage_level_id_list class-attribute instance-attribute

select_by_voltage_level_id_list = None

If given, only the voltage levels in this list will be imported. Note: not all voltage levels in this list might be considered relevant after preprocessing. This can happen if the requirements for relevant substations are not met. E.g. minimum number of busbars, connected branches or missing busbar couplers.

ingress_id class-attribute instance-attribute

ingress_id = None

An optional id that is used to identify the source of the data. This can be used to track where the data came from, e.g. if it was imported from a specific database or a specific user.

contingency_list_file class-attribute instance-attribute

contingency_list_file = None

The path to the contingency lists if present expected format see: importer/contingency_from_power_factory/PF_data_class.py

schema_format class-attribute instance-attribute

schema_format = None

The schema format of the contingency list file if present. This can be either "ContingencyImportSchemaPowerFactory" or "ContingencyImportSchema". found in: - importer/contingency_from_power_factory/PF_data_class.py - importer/pypowsybl_import/contingency_from_file/contingency_file_models.py

relevant_station_rules class-attribute instance-attribute

relevant_station_rules = RelevantStationRules()

Rules to determine whether a substation is relevant or not.

loadflow_parameters_file class-attribute instance-attribute

loadflow_parameters_file = None

The path to the loadflow parameters file if present. This file should contain the loadflow parameters in the format defined by the data_type.

network_reduction_voltage_level_range class-attribute instance-attribute

network_reduction_voltage_level_range = -1

The range of the network reduction to be applied, based on AreaSettings.view_area. If set to -1, no reduction is applied. The range defines how many voltage levels away from the view_area should be included in the reduction.

UcteImporterParameters

Bases: BaseImporterParameters

Parameters that are required to import the data from a UCTE file.

This will utilize powsybl and the powsybl backend to the loadflow solver

area_settings class-attribute instance-attribute

area_settings = AreaSettings(
    control_area=["D8"],
    view_area=["D2", "D4", "D7", "D8"],
    nminus1_area=["D2", "D4", "D7", "D8"],
)

By default the D8 is controllable and the german grid is viewable

grid_model_file instance-attribute

grid_model_file

The path to the UCTE file to load. Note that only a single timestep, i.e. only a single UCTE file will be loaded in one import/preprocessing run. For multiple timesteps, the preprocessing is triggered multiple times.

data_type class-attribute instance-attribute

data_type = 'ucte'

A constant field to indicate that this is a UCTE importer

fail_on_non_convergence class-attribute instance-attribute

fail_on_non_convergence = True

Whether to raise an error if the loadflow does not converge in the basecase. If set to False, the preprocessing will continue and dc-optimization can still happen. Mostly for debugging purposes

data_folder instance-attribute

data_folder

The path where the entry point where the timestep data folder structure starts.

The folder structure is defined in interfaces.folder_structure. This folder is relative to the processed_grid_folder that is configured in the backend/importer. A typical default would be grid_model_file.stem

white_list_file class-attribute instance-attribute

white_list_file = None

The path to the white lists if present

black_list_file class-attribute instance-attribute

black_list_file = None

The path to the balck lists if present

ignore_list_file class-attribute instance-attribute

ignore_list_file = None

The path to the ignore lists if present

A csv file with the following columns: grid_model_id, reason

The implementation is expected to ignore all elements that are in the ignore list.

select_by_voltage_level_id_list class-attribute instance-attribute

select_by_voltage_level_id_list = None

If given, only the voltage levels in this list will be imported. Note: not all voltage levels in this list might be considered relevant after preprocessing. This can happen if the requirements for relevant substations are not met. E.g. minimum number of busbars, connected branches or missing busbar couplers.

ingress_id class-attribute instance-attribute

ingress_id = None

An optional id that is used to identify the source of the data. This can be used to track where the data came from, e.g. if it was imported from a specific database or a specific user.

contingency_list_file class-attribute instance-attribute

contingency_list_file = None

The path to the contingency lists if present expected format see: importer/contingency_from_power_factory/PF_data_class.py

schema_format class-attribute instance-attribute

schema_format = None

The schema format of the contingency list file if present. This can be either "ContingencyImportSchemaPowerFactory" or "ContingencyImportSchema". found in: - importer/contingency_from_power_factory/PF_data_class.py - importer/pypowsybl_import/contingency_from_file/contingency_file_models.py

relevant_station_rules class-attribute instance-attribute

relevant_station_rules = RelevantStationRules()

Rules to determine whether a substation is relevant or not.

loadflow_parameters_file class-attribute instance-attribute

loadflow_parameters_file = None

The path to the loadflow parameters file if present. This file should contain the loadflow parameters in the format defined by the data_type.

network_reduction_voltage_level_range class-attribute instance-attribute

network_reduction_voltage_level_range = -1

The range of the network reduction to be applied, based on AreaSettings.view_area. If set to -1, no reduction is applied. The range defines how many voltage levels away from the view_area should be included in the reduction.

CgmesImporterParameters

Bases: BaseImporterParameters

Parameters to start an import data from a CGMES file.

This will utilize powsybl and the powsybl backend to the loadflow solver.

area_settings class-attribute instance-attribute

area_settings = AreaSettings(
    control_area=["BE"],
    view_area=["BE", "LU", "D4", "D2", "NL", "FR"],
    nminus1_area=["BE"],
    cutoff_voltage=220,
)

The area settings for the CGMES importer

grid_model_file instance-attribute

grid_model_file

The path to the CGMES .zip file to load.

Note that only a single timestep, i.e. only a single CGMES .zip file will be loaded in one import/preprocessing run. For multiple timesteps, the preprocessing is triggered multiple times. Note: the .zip file must contain all xml files in the same root folder, i.e. the following files: - EQ.xml - SSH.xml - SV.xml - TP.xml - EQBD.xml - TPBD.xml

data_type class-attribute instance-attribute

data_type = 'cgmes'

A constant field to indicate that this is a CGMES importer

fail_on_non_convergence class-attribute instance-attribute

fail_on_non_convergence = True

Whether to raise an error if the loadflow does not converge in the basecase. If set to False, the preprocessing will continue and dc-optimization can still happen. Mostly for debugging purposes

data_folder instance-attribute

data_folder

The path where the entry point where the timestep data folder structure starts.

The folder structure is defined in interfaces.folder_structure. This folder is relative to the processed_grid_folder that is configured in the backend/importer. A typical default would be grid_model_file.stem

white_list_file class-attribute instance-attribute

white_list_file = None

The path to the white lists if present

black_list_file class-attribute instance-attribute

black_list_file = None

The path to the balck lists if present

ignore_list_file class-attribute instance-attribute

ignore_list_file = None

The path to the ignore lists if present

A csv file with the following columns: grid_model_id, reason

The implementation is expected to ignore all elements that are in the ignore list.

select_by_voltage_level_id_list class-attribute instance-attribute

select_by_voltage_level_id_list = None

If given, only the voltage levels in this list will be imported. Note: not all voltage levels in this list might be considered relevant after preprocessing. This can happen if the requirements for relevant substations are not met. E.g. minimum number of busbars, connected branches or missing busbar couplers.

ingress_id class-attribute instance-attribute

ingress_id = None

An optional id that is used to identify the source of the data. This can be used to track where the data came from, e.g. if it was imported from a specific database or a specific user.

contingency_list_file class-attribute instance-attribute

contingency_list_file = None

The path to the contingency lists if present expected format see: importer/contingency_from_power_factory/PF_data_class.py

schema_format class-attribute instance-attribute

schema_format = None

The schema format of the contingency list file if present. This can be either "ContingencyImportSchemaPowerFactory" or "ContingencyImportSchema". found in: - importer/contingency_from_power_factory/PF_data_class.py - importer/pypowsybl_import/contingency_from_file/contingency_file_models.py

relevant_station_rules class-attribute instance-attribute

relevant_station_rules = RelevantStationRules()

Rules to determine whether a substation is relevant or not.

loadflow_parameters_file class-attribute instance-attribute

loadflow_parameters_file = None

The path to the loadflow parameters file if present. This file should contain the loadflow parameters in the format defined by the data_type.

network_reduction_voltage_level_range class-attribute instance-attribute

network_reduction_voltage_level_range = -1

The range of the network reduction to be applied, based on AreaSettings.view_area. If set to -1, no reduction is applied. The range defines how many voltage levels away from the view_area should be included in the reduction.

ReassignmentLimits

Bases: BaseModel

Reassignment limits for electrical reconfiguration at substations.

max_reassignments_per_sub class-attribute instance-attribute

max_reassignments_per_sub = 1000

The maximum number of reassignments to perform during the electrical reconfiguration. Gets overriden by station_specific_limits if a voltage level id is given.

station_specific_limits class-attribute instance-attribute

station_specific_limits = Field(default_factory=dict)

Specific reassignment limits per station to override the global reassignment limit. Expects a voltage level id as key and the maximum number of reassignments as value. Note: the key must match the runtime station voltage_level_id used during preprocessing.

PreprocessParameters

Bases: BaseModel

Parameters for the preprocessing procedure which is independent of the data source

filter_disconnectable_branches_processes class-attribute instance-attribute

filter_disconnectable_branches_processes = 1

When checking for disconnectable branches, multiple worker processes can be used as it is a costly operation.

action_set_filter_bridge_lookup class-attribute instance-attribute

action_set_filter_bridge_lookup = True

Whether to filter the action set using bridge lookups. This will remove all assignments that have less than two non-bridges on every side

action_set_filter_bsdf_lodf class-attribute instance-attribute

action_set_filter_bsdf_lodf = True

Whether to filter the action set using a consecutive BSDF/LODF application. This will filter out all actions that are also filtered by bridge lookup and additionally all actions that split the grid under N-1 branch outages, i.e. all assignments that created a new bridge in the graph. This is a relatively costly process to run, only set to true if you can afford the extra preprocessing time.

action_set_filter_bsdf_lodf_batch_size class-attribute instance-attribute

action_set_filter_bsdf_lodf_batch_size = 8

If filtering with bsdf/lodf - which batch size to use. Larger will use more memory but be faster.

action_set_clip class-attribute instance-attribute

action_set_clip = 2 ** 23

After which size to randomly subselect actions at a substation. If a substations has a lot of branches, the action space will explode exponentially and a safe-guard is to clip after a certain number of actions.

asset_topo_close_couplers class-attribute instance-attribute

asset_topo_close_couplers = False

Whether to close open couplers in all stations in the asset topology. This might accidentally cancel a maintenance

separation_set_clip_hamming_distance class-attribute instance-attribute

separation_set_clip_hamming_distance = 0

If a large configuration table comes out of a substation, the table size can be reduced by removing configurations that are close to each other. This parameter sets the definition of close in terms of hamming distance, by default 0 (no reduction).

separation_set_clip_at_size class-attribute instance-attribute

separation_set_clip_at_size = 100

By what size a table is considered large. If the table is larger than this size, the clip_hamming_distance will be used to reduce the table size, by default 100. If a table is smaller, no reduction will be applied.

realise_station_busbar_choice_heuristic class-attribute instance-attribute

realise_station_busbar_choice_heuristic = (
    "least_connected_busbar"
)

The heuristic to use when there are multiple physical busbars available for an asset. The options are: - "first": Use the first busbar in the list of busbars (fastest preprocessing) - "least_connected_busbar": Use the busbar with the least number of connections to other assets (best results)

The "least_connected_busbar" heuristic is the default and is recommended for most cases, trying to spread the assets evenly across the busbars in a station.

electrical_reassignment_limits class-attribute instance-attribute

electrical_reassignment_limits = None

If given, limits for the electrical reassignment at substations.

physical_reassignment_limits class-attribute instance-attribute

physical_reassignment_limits = None

If given, limits for the physical reassignment at substations.

ac_dc_interpolation class-attribute instance-attribute

ac_dc_interpolation = 0.0

Whether to use the DC loadflow as the base loadflow (0) or the AC loadflow (1). Can also be anything in between.

preprocess_bb_outages class-attribute instance-attribute

preprocess_bb_outages = False

Whether to preprocess and persist busbar outage data into the grid file. This does not decide whether the optimizer will eventually include busbar outage effects.

double_limit_n0 class-attribute instance-attribute

double_limit_n0 = 0.9

If passed, then double limits will be computed for the N-0 flows. Lines that are below double_limit_n0 relative load in the unsplit configuration will have their capacity multiplied by double_limit_n0 to prevent loading them up to their maximum capacity.

double_limit_n1 class-attribute instance-attribute

double_limit_n1 = 0.9

If passed, then double limits will be computed for the N-1 flows. Lines that are below double_limit_n1 relative load in the unsplit configuration will have their capacities multiplied by double_limit_n1 to prevent loading them up to their maximum capacity.

initial_loadflow_processes class-attribute instance-attribute

initial_loadflow_processes = 8

How many processes to use to compute the initial AC loadflow

fail_on_non_convergence class-attribute instance-attribute

fail_on_non_convergence = True

Whether to raise an error if the initial loadflow does not converge. If False, a warning is logged instead and the backend is initialized with the dc loadflow results.

StartPreprocessingCommand

Bases: BaseModel

A command to launch a preprocessing run of a timestep upon reception.

importer_parameters class-attribute instance-attribute

importer_parameters = Field(discriminator='data_type')

The parameters to the importer, depending which input source was chosen

preprocess_parameters class-attribute instance-attribute

preprocess_parameters = PreprocessParameters()

Parameters required for preprocessing independent of the data source

preprocess_id instance-attribute

preprocess_id

The id of the preprocessing run, should be included in all responses to identify where the data came from

command_type class-attribute instance-attribute

command_type = 'start_preprocessing'

ShutdownCommand

Bases: BaseModel

A command to shut down the preprocessing worker

exit_code class-attribute instance-attribute

exit_code = 0

The exit code to return

command_type class-attribute instance-attribute

command_type = 'shutdown'

Command

Bases: BaseModel

A wrapper to aid deserialization

command class-attribute instance-attribute

command = Field(discriminator='command_type')

The actual command posted

timestamp class-attribute instance-attribute

timestamp = Field(
    default_factory=lambda: str(datetime.now())
)

When the command was sent

uuid class-attribute instance-attribute

uuid = Field(default_factory=lambda: str(uuid.uuid4()))

A unique identifier for this command message, used to avoid duplicates during processing

toop_engine_interfaces.messages.preprocess.preprocess_heartbeat

Contains the message classes for a preprocessing worker's heartbeat messages

ConvertToJaxStage module-attribute

ConvertToJaxStage = Literal[
    "convert_to_jax_started",
    "convert_tot_stat",
    "convert_relevant_inj",
    "convert_masks",
    "switching_distance_info",
    "pad_out_branch_actions",
    "convert_rel_bb_outage_data",
    "create_static_information",
    "filter_branch_actions",
    "bb_outage_baseline_analysis",
    "convert_to_jax_done",
]

NumpyPreprocessStage module-attribute

NumpyPreprocessStage = Literal[
    "preprocess_started",
    "extract_network_data_from_interface",
    "exclude_nonlinear_psts_from_controllable",
    "filter_relevant_nodes",
    "assert_network_data",
    "compute_ptdf_if_not_given",
    "compute_psdf_if_not_given",
    "add_nodal_injections_to_network_data",
    "combine_phaseshift_and_injection",
    "compute_bridging_branches",
    "exclude_bridges_from_outage_masks",
    "reduce_branch_dimension",
    "reduce_node_dimension",
    "filter_disconnectable_branches_nminus2",
    "compute_branch_topology_info",
    "compute_electrical_actions",
    "enumerate_station_realizations",
    "remove_relevant_subs_without_actions",
    "simplify_asset_topology",
    "compute_separation_set",
    "convert_multi_outages",
    "filter_inactive_injections",
    "compute_injection_topology_info",
    "process_injection_outages",
    "add_missing_asset_topo_info",
    "add_bus_b_columns_to_ptdf",
    "enumerate_injection_actions",
    "preprocess_bb_outage",
    "preprocess_done",
]

LoadGridStage module-attribute

LoadGridStage = Literal[
    "load_grid_into_loadflow_solver_backend",
    "compute_base_loadflows",
    "save_artifacts",
]

InitialLoadflowStage module-attribute

InitialLoadflowStage = Literal[
    "prepare_contingency_analysis",
    "run_contingency_analysis",
]

ImporterStage module-attribute

ImporterStage = Literal[
    "start",
    "load_from_fs",
    "reduce_network_to_view_area",
    "get_topology_model",
    "modify_low_impedance_lines",
    "modify_branches_over_switches",
    "apply_cb_list",
    "cross_border_current",
    "get_masks",
    "end",
]

PreprocessStage module-attribute

PreprocessStage = Literal[
    ImporterStage,
    NumpyPreprocessStage,
    ConvertToJaxStage,
    LoadGridStage,
    InitialLoadflowStage,
]

PreprocessStatusInfo

Bases: BaseModel

A status info to inform about an ongoing preprocess action.

preprocess_id instance-attribute

preprocess_id

The id of the preprocess job.

runtime instance-attribute

runtime

The amount of time since the start of the optimization.

stage instance-attribute

stage

The stage in which the preprocessing job currently is.

message instance-attribute

message

An optional message

PreprocessHeartbeat

Bases: BaseModel

A message class for heartbeats from the preprocessing worker.

When idle, this just sends a hello, and when preprocessing it also conveys a status update at which stage the preprocessing is so it can be tracked in the frontend.

idle instance-attribute

idle

Whether the worker is idle

status_info class-attribute instance-attribute

status_info = None

If not idle, a status update

instance_id class-attribute instance-attribute

instance_id = ''

The ID of the worker instance that sent this heartbeat.

timestamp class-attribute instance-attribute

timestamp = Field(
    default_factory=lambda: str(datetime.now())
)

When the heartbeat was sent

uuid class-attribute instance-attribute

uuid = Field(default_factory=lambda: str(uuid.uuid4()))

A unique identifier for this heartbeat message, used to avoid duplicates during processing

Loadflow Service

toop_engine_interfaces.loadflow_results

Defines interfaces for loadflow results.

The overall process is that a job is called on a loadflow engine for a grid. The grid holds some information that is referenced in the results: - an N-1 definition, which can include multi-contingencys. An N-1 case is uniquely identified by a string descriptor but can include multiple failing elements. The string identifier of the N-1 case should be delivered upon loading. - a number of timesteps, which are uniquely identified by an integer index. - branches which are uniquely identified by a string descriptor and have either two or three sides. - nodes which are uniquely identified by a string descriptor and have a type (PV, PQ, REF) - regulating elements which are uniquely identified by a string descriptor and have a type (generator, regulating transformer, SVC, ...)

LoadflowResultTable module-attribute

LoadflowResultTable = Union[
    pat.DataFrame[NodeResultSchema],
    pat.DataFrame[BranchResultSchema],
    pat.DataFrame[VADiffResultSchema],
    pat.DataFrame[ConnectivityResultSchema],
    pat.DataFrame[SwitchResultsSchema],
    pat.DataFrame[RegulatingElementResultSchema],
    pat.DataFrame[ConvergedSchema],
    pat.DataFrame[SppsResultsSchema],
    pat.DataFrame[CascadeResultSchema],
]

BranchSide

Bases: Enum

The side of a branch.

ONE class-attribute instance-attribute

ONE = 1

The following side for the types of branches: - line: from side - 2 winding trafo: high voltage side - 3 winding trafo: high voltage side - other: from side

TWO class-attribute instance-attribute

TWO = 2

The following side for the types of branches: - line: to side - 2 winding trafo: low voltage side - 3 winding trafo: medium voltage side - other: to side

THREE class-attribute instance-attribute

THREE = 3

Only valid for 3 winding transformers, representing the low voltage side.

NONE class-attribute instance-attribute

NONE = 4

No side specified.

RegulatingElementType

Bases: Enum

A list of known regulating elements, TODO expand

GENERATOR_Q class-attribute instance-attribute

GENERATOR_Q = 'GENERATOR_Q'

A generator that is used to control the reactive power output.

SLACK_P class-attribute instance-attribute

SLACK_P = 'SLACK_P'

The active power output of the slack node.

SLACK_Q class-attribute instance-attribute

SLACK_Q = 'SLACK_Q'

The reactive power output of the slack node.

REGULATING_TRANSFORMER_TAP class-attribute instance-attribute

REGULATING_TRANSFORMER_TAP = 'REGULATING_TRANSFORMER_TAP'

A regulating transformer that is used to control the tap position.

SVC_Q class-attribute instance-attribute

SVC_Q = 'SVC_Q'

A static var compensator that is used to control the reactive power output.

HVDC_CONVERTER_Q class-attribute instance-attribute

HVDC_CONVERTER_Q = 'HVDC_CONVERTER_Q'

An HVDC converter station.

OTHER class-attribute instance-attribute

OTHER = 'OTHER'

A placeholder for not yet known regulating elements.

ConvergenceStatus

Bases: Enum

The convergence status of the loadflow in a single timestep/contingency/component

CONVERGED class-attribute instance-attribute

CONVERGED = 'CONVERGED'

The loadflow converged

FAILED class-attribute instance-attribute

FAILED = 'FAILED'

The loadflow failed to start, e.g. because no slack bus was available

MAX_ITERATION_REACHED class-attribute instance-attribute

MAX_ITERATION_REACHED = 'MAX_ITERATION_REACHED'

The maximum number of iterations was reached, i.e. the loadflow did not converge.

NO_CALCULATION class-attribute instance-attribute

NO_CALCULATION = 'NO_CALCULATION'

The component was ignored due to other reasons (engine did not support it)

BranchResultSchema

Bases: DataFrameModel

A schema for the branch results table.

This holds i, p and q values for all monitored branches with a multi-index of timestep, contingency (CO), branch (CB) and side.

If no branches are monitored, this is the empty DataFrame.

TODO Decide if this should be used for injections aswell

timestep instance-attribute

timestep

The timestep of this result. This indexes into the timesteps that were loaded

contingency instance-attribute

contingency

The contingency that caused this loadflow. For N-0 results, the special CO 'BASECASE' without GridElements is used, if its added.

element instance-attribute

element

The branch that these loadflow results correspond to

side class-attribute instance-attribute

side = pa.Field(isin=[(side.value) for side in BranchSide])

The side of the branch that these results correspond to

i class-attribute instance-attribute

i = pa.Field(nullable=True)

The current in the branch in A

This should only be NaN if the branch has no connection to the slack bus.

p class-attribute instance-attribute

p = pa.Field(nullable=True)

The active power in the branch in MW

This should only be NaN if the branch has no connection to the slack bus.

q class-attribute instance-attribute

q = pa.Field(nullable=True)

The reactive power in the branch in MVar

This should only be NaN if the branch has no connection to the slack bus.

loading class-attribute instance-attribute

loading = pa.Field(nullable=True)

The loading of the branch in % of rated current. This always refers to the permanent/default rating of the branch if there are multiple ratings available. If no rating is available for the branch, this should be set to NaN. If the engine does not support the computation of this value, the column can be omitted.

element_name instance-attribute

element_name

The name of the Branch, if available. This is not used for the loadflow computation, but can be used for display purposes. If no name is available, this should be set to an empty string.

contingency_name instance-attribute

contingency_name

The name of the contingency, if available. This is not used for the loadflow computation, but can be used for display purposes. If no name is available, this should be set to an empty string.

NodeResultSchema

Bases: DataFrameModel

A schema for the node results table.

This holds p and q values for all monitored nodes with a multi-index of timestep and contingency. If no nodes are monitored, this is the empty DataFrame.

timestep instance-attribute

timestep

The timestep of this result. This indexes into the timesteps that were loaded

contingency instance-attribute

contingency

The contingency that caused this loadflow. For N-0 results, the special CO 'BASECASE' is used.

element instance-attribute

element

The node that these loadflow results correspond to

vm class-attribute instance-attribute

vm = pa.Field(nullable=True)

The voltage magnitude at the node in kV.

In DC, this should be the nominal voltage of the node. This should only be NaN if the node does not have a connection to the slack bus.

vm_loading class-attribute instance-attribute

vm_loading = pa.Field(nullable=True)

How close the voltage magnitude is to the max/min voltage limits in percent. This is computed as: (vm - v_nominal) / (v_max - v_nominal) for vm > v_nominal and (vm - v_nominal) / (v_nominal - v_min) for vm < v_nominal.

va class-attribute instance-attribute

va = pa.Field(nullable=True)

The voltage angle at the node in degrees

This should only be NaN if the node does not have a connection to the slack bus.

p class-attribute instance-attribute

p = pa.Field(nullable=True)

The accumulated absolute active power at the node in MW, obtained by summing the absolute active power of all branches and injections connected to the node.

If the engine does not support the computation of this value, the column can be omitted.

q class-attribute instance-attribute

q = pa.Field(nullable=True)

The accumulated absolute reactive power at the node in MVar, obtained by summing the absolute reactive power of all branches and injections connected to the node

If the engine does not support the computation of this value, the column can be omitted.

vm_basecase_deviation class-attribute instance-attribute

vm_basecase_deviation = pa.Field(nullable=True)

Voltage magnitude deviation from the basecase (N-0) in percent. Computed as: abs(vm_contingency - vm_basecase) / vm_basecase * 100 For basecase contingency, the deviation will be 0.0 (basecase vs basecase). NaN if no valid basecase voltage exists(basecase not converged).

element_name instance-attribute

element_name

The name of the node, if available. This is not used for the loadflow computation, but can be used for display purposes. If no name is available, this should be set to an empty string.

contingency_name instance-attribute

contingency_name

The name of the contingency, if available. This is not used for the loadflow computation, but can be used for display purposes. If no name is available, this should be set to an empty string.

ConnectivityResultSchema

Bases: DataFrameModel

Schema defining the contingency-to-element connectivity mapping.

Each row represents a relationship between a contingency and an affected grid element, based on outage group logic.

contingency instance-attribute

contingency

Global unique identifier of the contingency event.

Represents the triggering outage (e.g., line, transformer, generator). This is the first level of the MultiIndex.

element instance-attribute

element

Global unique identifier of a grid element affected by the contingency.

Each element listed here belongs to the outage group associated with the contingency and is therefore considered disconnected when the contingency occurs. This is the second level of the MultiIndex.

outage_group_id instance-attribute

outage_group_id

Identifier of the outage group shared by the contingency and element.

Outage groups represent sets of elements that become de-energized together when separated from the rest of the network by circuit breakers. Multiple contingencies may map to the same outage group.

VADiffResultSchema

Bases: DataFrameModel

A schema for the voltage angle results.

Holds information about the voltage angle difference between busses that could be (re)connected by power switches.

timestep instance-attribute

timestep

The timestep of this result. This indexes into the timesteps that were loaded

contingency instance-attribute

contingency

The critical contingency that caused this loadflow. For N-0 results, the special CO 'BASECASE' is used.

element instance-attribute

element

The element over which the voltage angle difference is computed. Can be either an open switch or any switch or branch under N-1. If under N-1, then element and contingency are the same.

va_diff class-attribute instance-attribute

va_diff = pa.Field(nullable=True)

The voltage angle difference in degrees between the two ends of the element. nan if at least one of the ends has no voltage angle (island, out of service)

element_name instance-attribute

element_name

The name of the Branch or Switch, if available. This is not used for the loadflow computation, but can be used for display purposes. If no name is available, this should be set to an empty string.

contingency_name instance-attribute

contingency_name

The name of the contingency, if available. This is not used for the loadflow computation, but can be used for display purposes. If no name is available, this should be set to an empty string.

SwitchResultsSchema

Bases: DataFrameModel

A schema for the voltage angle results.

Holds information about the voltage angle difference between busses that could be (re)connected by power switches.

timestep instance-attribute

timestep

The timestep of this result. This indexes into the timesteps that were loaded

contingency instance-attribute

contingency

The critical contingency that caused this loadflow. For N-0 results, the special CO 'BASECASE' is used.

element instance-attribute

element

The element over which the voltage angle difference is computed. Can be either an open switch or any switch or branch under N-1. If under N-1, then element and contingency are the same.

p class-attribute instance-attribute

p = pa.Field(nullable=True)

The accumulated absolute active power at the node in MW, obtained by summing the absolute active power of all branches and injections connected to the node.

If the engine does not support the computation of this value, the column can be omitted.

q class-attribute instance-attribute

q = pa.Field(nullable=True)

The accumulated absolute reactive power at the node in MVar, obtained by summing the absolute reactive power of all branches and injections connected to the node

If the engine does not support the computation of this value, the column can be omitted.

vm class-attribute instance-attribute

vm = pa.Field(nullable=True)

The voltage magnitude at the node in kV.

In DC, this should be the nominal voltage of the node. This should only be NaN if the node does not have a connection to the slack bus.

i class-attribute instance-attribute

i = pa.Field(nullable=True)

The current in the branch in A

This should only be NaN if the branch has no connection to the slack bus.

element_name instance-attribute

element_name

The name of the Branch or Switch, if available. This is not used for the loadflow computation, but can be used for display purposes. If no name is available, this should be set to an empty string.

contingency_name instance-attribute

contingency_name

The name of the contingency, if available. This is not used for the loadflow computation, but can be used for display purposes. If no name is available, this should be set to an empty string.

side class-attribute instance-attribute

side = pa.Field(nullable=True)

The measurement side of the switch result.

• "from": values measured at the from-bus terminal (taken from net.res_switch).
• "to": values measured at the to-bus terminal (taken from net.res_switch).
• null: result was computed by aggregating branch flows and node injections. Switches modelled without impedance have identical electrical conditions on both terminals (no voltage drop, no power loss across the switch), so a single aggregated value is sufficient and no side distinction is needed.

SwitchElementMappingSchema

Bases: DataFrameModel

Schema for mapping switches to connected elements.

This table defines which elements are electrically connected to each switch. It is used to aggregate branch flows and node injections when computing switch-level results.

The mapping includes both: - branch-like elements (lines, trafos, impedances, etc.) - buses

If no switches are mapped, this is an empty DataFrame.

switch_id class-attribute instance-attribute

switch_id = pa.Field(nullable=False)

The pandapower index of the switch.

This identifies the switch for which connected elements are collected and used in result aggregation.

element class-attribute instance-attribute

element = pa.Field(nullable=False)

The globally unique identifier of the connected element.

This can represent either: - a branch-like element (e.g. "12__line", "5__trafo") - a bus (e.g. "3__bus")

side class-attribute instance-attribute

side = pa.Field(nullable=True)

The side of the branch element.

• For branch-like elements: Indicates the terminal of the element connected to the bus (e.g. BranchSide.ONE, BranchSide.TWO, BranchSide.THREE).
• For bus entries: This value is NaN, since buses do not have sides.

RegulatingElementResultSchema

Bases: DataFrameModel

A schema for the regulating elements.

A regulating element can either be a branch (a trafo with regulating tap) or a node (a generator, SVC, ...). If no regulating elements are monitored, this is the empty DataFrame.

timestep instance-attribute

timestep

The timestep of this result. This indexes into the timesteps that were loaded

contingency instance-attribute

contingency

The critical contingency that caused this loadflow. For N-0 results, the special CO 'BASECASE' is used.

element instance-attribute

element

The regulating element that these loadflow results correspond to

value instance-attribute

value

The value of the regulating element. Depending on the type of the regulating element, this can mean different things.

regulating_element_type class-attribute instance-attribute

regulating_element_type = pa.Field(
    isin=[(side.value) for side in RegulatingElementType]
)

The type of the regulating element (generator, regulating transformer, SVC, ...).

element_name instance-attribute

element_name

The name of the Regulating Element, if available. This is not used for the loadflow computation, but can be used for display purposes. If no name is available, this should be set to an empty string.

contingency_name instance-attribute

contingency_name

The name of the contingency, if available. This is not used for the loadflow computation, but can be used for display purposes. If no name is available, this should be set to an empty string.

ConvergedSchema

Bases: DataFrameModel

A schema for the converged table. This holds the convergence information for each timestep.

Potentially, multiple islands can exist in the same grid. In this case, the synchronous component needs to be distinguished. A synchronous component is a grid area consisting of all nodes and branches that are connected to the same slack through AC lines (no HVDC). The largest component must always be called 'MAIN' while the names of the other components are arbitrary. Usually only one component is present. If no convergence information is available, this is the empty DataFrame.

timestep instance-attribute

timestep

The timestep of this result. This indexes into the timesteps that were loaded

contingency instance-attribute

contingency

The critical contingency that caused this loadflow. For N-0 results, the special CO 'BASECASE' is used.

status class-attribute instance-attribute

status = pa.Field(
    isin=[(side.value) for side in ConvergenceStatus]
)

Whether the loadflow converged at this timestep/contingency.

iteration_count class-attribute instance-attribute

iteration_count = pa.Field(nullable=True)

The number of iterations required for the loadflow to converge.

warnings class-attribute instance-attribute

warnings = pa.Field(nullable=True)

An additional string field that carries warnings or error logs for specific timesteps/contingencys/components.

contingency_name instance-attribute

contingency_name

The name of the contingency, if available. This is not used for the loadflow computation, but can be used for display purposes. If no name is available, this should be set to an empty string.

SppsResultsSchema

Bases: DataFrameModel

SpPS run summaries, one row per (timestep, contingency).

activated_schemes_per_iter holds a JSON string encoding of list[list[str]] (outer list = SpPS outer iterations, inner = scheme names that fired in that iteration), so the table remains parquet-friendly. Use json.loads to recover the nested structure.

timestep instance-attribute

timestep

Loadflow timestep index for the outage that produced this row.

contingency instance-attribute

contingency

Globally unique id of the contingency (same value as in other loadflow result tables).

iterations instance-attribute

iterations

Number of SpPS iterations that executed (1-based count).

activated_schemes_per_iter class-attribute instance-attribute

activated_schemes_per_iter = pa.Field()

JSON string for list[list[str]] of scheme names that activated per iteration.

max_iterations_reached instance-attribute

max_iterations_reached

Whether the run stopped because the iteration cap was hit while schemes still fired.

power_flow_failed instance-attribute

power_flow_failed

Whether a post-action power flow failed in keep_previous mode.

CascadeResultSchema

Bases: DataFrameModel

A schema for cascade simulation event results.

This table contains one row per cascade event created after a converged contingency load flow. It records why the cascade advanced and which element/outage group was affected.

timestep instance-attribute

timestep

The timestep of this cascade event.

contingency instance-attribute

contingency

Globally unique id of the contingency that started the cascade.

cascade_number instance-attribute

cascade_number

Cascade step number where the event happened.

element_mrid class-attribute instance-attribute

element_mrid = pa.Field(nullable=True)

External identifier of the affected element, if known.

element_id class-attribute instance-attribute

element_id = pa.Field(nullable=True)

Globally unique id of the affected element, if known.

contingency_outage_id class-attribute instance-attribute

contingency_outage_id = pa.Field(nullable=True)

Identifier of the outage group for the contingency that started the cascade.

contingency_name class-attribute instance-attribute

contingency_name = pa.Field(nullable=True)

Human-readable name of the contingency that started the cascade, if known.

element_outage_group_id class-attribute instance-attribute

element_outage_group_id = pa.Field(nullable=True)

Stable identifier of the affected element's outage group.

element_name class-attribute instance-attribute

element_name = pa.Field(nullable=True)

Human-readable name of the affected element, if known.

cascade_reason instance-attribute

cascade_reason

Reason for the cascade event, such as current overload or distance protection.

loading class-attribute instance-attribute

loading = pa.Field(nullable=True)

Branch loading value that triggered the cascade event, if available.

r_ohm class-attribute instance-attribute

r_ohm = pa.Field(nullable=True)

Relay resistance value for distance-protection events, if available.

x_ohm class-attribute instance-attribute

x_ohm = pa.Field(nullable=True)

Relay reactance value for distance-protection events, if available.

distance_protection_severity class-attribute instance-attribute

distance_protection_severity = pa.Field(nullable=True)

Distance-protection severity for relay events. Empty for other event types.

activated_schemes_per_iter class-attribute instance-attribute

activated_schemes_per_iter = pa.Field(nullable=True)

JSON string of SpPS scheme names that activated per inner cascade load-flow iteration.

LoadflowResults

Bases: BaseModel

A container for the loadflow results for a computation job.

job_id instance-attribute

job_id

The id of the computation job that created these loadflows

branch_results class-attribute instance-attribute

branch_results = None

The results for the branches. If no branches are monitored, this is the empty DataFrame. Non converging contingencys/timesteps are to be omitted

node_results class-attribute instance-attribute

node_results = None

The results for the nodes. If no nodes are monitored, this is the empty DataFrame.

regulating_element_results class-attribute instance-attribute

regulating_element_results = None

The results for the regulating elements. If no regulating elements are monitored, this is the empty DataFrame.

converged class-attribute instance-attribute

converged = None

The convergence information for each timestep and contingency. If there were non-converging loadflows for some timesteps/contingencys, these results should be omitted from the other tables.

va_diff_results class-attribute instance-attribute

va_diff_results = None

The voltage angle difference results for each timestep and contingency. Considers the ends of the outaged branch, aswell as all open switches in monitored elements.

switch_results class-attribute instance-attribute

switch_results = None

The results for the switches.

Contains aggregated power flow and injection results per switch for each timestep and contingency.

Switch results are computed by aggregating contributions from all elements (branches and buses) electrically connected to one side of the switch. This represents the power flowing through the switch.

If no switches are monitored, this is the empty DataFrame. For non-converging contingencies/timesteps, result values are present but set to NaN.

connectivity_result class-attribute instance-attribute

connectivity_result = None

Connectivity mapping between contingencies and affected grid elements. This DataFrame defines which elements become unavailable for each contingency, based on outage group logic. Each row represents a (contingency, element) pair, indicating that the element is part of the outage group triggered by the contingency.

warnings class-attribute instance-attribute

warnings = Field(default_factory=list)

Global warnings that occured during the computation (e.g. monitored elements/contingencies that were not found)

spps_results class-attribute instance-attribute

spps_results = None

SpPS run summaries, concatenated in single-outage order. When SpPS did not run for a case, that chunk contributes no rows. If no job recorded SpPS, this is the empty DataFrame (default).

cascade_results class-attribute instance-attribute

cascade_results = None

Cascade simulation events, one row per event. Empty when cascade simulation is disabled or has no events.

__eq__

__eq__(lf_result)

Compare two LoadflowResults objects for equality.

Rounds floats to 6 decimal places for comparison. This is necessary because floating point arithmetic can lead to small differences in the results.

Ignores the order of the DataFrames, but checks that the indices are equal.

PARAMETER	DESCRIPTION
lf_result	The LoadflowResults object to compare with. TYPE: LoadflowResults

RETURNS	DESCRIPTION
bool	True if the two LoadflowResults objects are equal, False otherwise.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/loadflow_results.py

def __eq__(self, lf_result: object) -> bool:
    """Compare two LoadflowResults objects for equality.

    Rounds floats to 6 decimal places for comparison.
    This is necessary because floating point arithmetic can lead to small differences in the results.

    Ignores the order of the DataFrames, but checks that the indices are equal.

    Parameters
    ----------
    lf_result : LoadflowResults
        The LoadflowResults object to compare with.

    Returns
    -------
    bool
        True if the two LoadflowResults objects are equal, False otherwise.
    """
    rounding_accuracy = 6

    if not isinstance(lf_result, LoadflowResults):
        return False

    def required_frame_matches(left: pd.DataFrame, right: pd.DataFrame) -> bool:
        """Compare required result frames while ignoring row and column order."""
        if left.shape != right.shape:
            return False
        if not all(left.index.isin(right.index)):
            return False
        if not all(left.columns.isin(right.columns)):
            return False

        ordered_left = left.loc[right.index.drop_duplicates(), right.columns].round(rounding_accuracy)
        return ordered_left.equals(right.round(rounding_accuracy))

    def optional_frame_matches(left: pd.DataFrame | None, right: pd.DataFrame | None) -> bool:
        """Compare optional result frames while treating None and empty frames as equal."""
        if left is None or right is None:
            other = right if left is None else left
            return other is None or other.empty
        return required_frame_matches(left, right)

    return (
        self.job_id == lf_result.job_id
        and self.warnings == lf_result.warnings
        and required_frame_matches(self.branch_results, lf_result.branch_results)
        and required_frame_matches(self.node_results, lf_result.node_results)
        and required_frame_matches(self.regulating_element_results, lf_result.regulating_element_results)
        and required_frame_matches(self.va_diff_results, lf_result.va_diff_results)
        and required_frame_matches(self.converged, lf_result.converged)
        and optional_frame_matches(self.spps_results, lf_result.spps_results)
        and optional_frame_matches(self.cascade_results, lf_result.cascade_results)
    )

toop_engine_interfaces.loadflow_results_polars

Defines performance-improved polars versions of the loadflow results.

The loadflow results here mirror what is defined in loadflow_results.py, but use polars dataframes which are faster.

LoadflowResultTablePolars module-attribute

LoadflowResultTablePolars = Union[
    patpl.LazyFrame[NodeResultSchemaPolars],
    patpl.LazyFrame[BranchResultSchemaPolars],
    patpl.LazyFrame[VADiffResultSchemaPolars],
    patpl.LazyFrame[ConnectivityResultSchemaPolars],
    patpl.LazyFrame[SwitchResultsSchemaPolars],
    patpl.LazyFrame[RegulatingElementResultSchemaPolars],
    patpl.LazyFrame[ConvergedSchemaPolars],
    patpl.LazyFrame[SppsResultsSchemaPolars],
    patpl.LazyFrame[CascadeResultSchemaPolars],
]

BranchResultSchemaPolars

Bases: DataFrameModel, BranchResultSchema

Polars variant of BranchResultSchema.

timestep instance-attribute

timestep

The timestep of this result. This indexes into the timesteps that were loaded

contingency instance-attribute

contingency

The contingency that caused this loadflow. For N-0 results, the special CO 'BASECASE' without GridElements is used, if its added.

element instance-attribute

element

The branch that these loadflow results correspond to

side class-attribute instance-attribute

side = pa.Field(isin=[(side.value) for side in BranchSide])

The side of the branch that these results correspond to

i class-attribute instance-attribute

i = pa.Field(nullable=True)

The current in the branch in A

This should only be NaN if the branch has no connection to the slack bus.

p class-attribute instance-attribute

p = pa.Field(nullable=True)

The active power in the branch in MW

This should only be NaN if the branch has no connection to the slack bus.

q class-attribute instance-attribute

q = pa.Field(nullable=True)

The reactive power in the branch in MVar

This should only be NaN if the branch has no connection to the slack bus.

loading class-attribute instance-attribute

loading = pa.Field(nullable=True)

The loading of the branch in % of rated current. This always refers to the permanent/default rating of the branch if there are multiple ratings available. If no rating is available for the branch, this should be set to NaN. If the engine does not support the computation of this value, the column can be omitted.

element_name instance-attribute

element_name

The name of the Branch, if available. This is not used for the loadflow computation, but can be used for display purposes. If no name is available, this should be set to an empty string.

contingency_name instance-attribute

contingency_name

The name of the contingency, if available. This is not used for the loadflow computation, but can be used for display purposes. If no name is available, this should be set to an empty string.

NodeResultSchemaPolars

Bases: DataFrameModel, NodeResultSchema

Polars variant of NodeResultSchema.

timestep instance-attribute

timestep

The timestep of this result. This indexes into the timesteps that were loaded

contingency instance-attribute

contingency

The contingency that caused this loadflow. For N-0 results, the special CO 'BASECASE' is used.

element instance-attribute

element

The node that these loadflow results correspond to

vm class-attribute instance-attribute

vm = pa.Field(nullable=True)

The voltage magnitude at the node in kV.

In DC, this should be the nominal voltage of the node. This should only be NaN if the node does not have a connection to the slack bus.

vm_loading class-attribute instance-attribute

vm_loading = pa.Field(nullable=True)

How close the voltage magnitude is to the max/min voltage limits in percent. This is computed as: (vm - v_nominal) / (v_max - v_nominal) for vm > v_nominal and (vm - v_nominal) / (v_nominal - v_min) for vm < v_nominal.

va class-attribute instance-attribute

va = pa.Field(nullable=True)

The voltage angle at the node in degrees

This should only be NaN if the node does not have a connection to the slack bus.

p class-attribute instance-attribute

p = pa.Field(nullable=True)

The accumulated absolute active power at the node in MW, obtained by summing the absolute active power of all branches and injections connected to the node.

If the engine does not support the computation of this value, the column can be omitted.

q class-attribute instance-attribute

q = pa.Field(nullable=True)

The accumulated absolute reactive power at the node in MVar, obtained by summing the absolute reactive power of all branches and injections connected to the node

If the engine does not support the computation of this value, the column can be omitted.

vm_basecase_deviation class-attribute instance-attribute

vm_basecase_deviation = pa.Field(nullable=True)

Voltage magnitude deviation from the basecase (N-0) in percent. Computed as: abs(vm_contingency - vm_basecase) / vm_basecase * 100 For basecase contingency, the deviation will be 0.0 (basecase vs basecase). NaN if no valid basecase voltage exists(basecase not converged).

element_name instance-attribute

element_name

The name of the node, if available. This is not used for the loadflow computation, but can be used for display purposes. If no name is available, this should be set to an empty string.

contingency_name instance-attribute

contingency_name

The name of the contingency, if available. This is not used for the loadflow computation, but can be used for display purposes. If no name is available, this should be set to an empty string.

ConnectivityResultSchemaPolars

Bases: DataFrameModel, ConnectivityResultSchema

Polars variant of ConnectivityResultSchema.

contingency instance-attribute

contingency

Global unique identifier of the contingency event.

Represents the triggering outage (e.g., line, transformer, generator). This is the first level of the MultiIndex.

element instance-attribute

element

Global unique identifier of a grid element affected by the contingency.

Each element listed here belongs to the outage group associated with the contingency and is therefore considered disconnected when the contingency occurs. This is the second level of the MultiIndex.

outage_group_id instance-attribute

outage_group_id

Identifier of the outage group shared by the contingency and element.

Outage groups represent sets of elements that become de-energized together when separated from the rest of the network by circuit breakers. Multiple contingencies may map to the same outage group.

VADiffResultSchemaPolars

Bases: DataFrameModel, VADiffResultSchema

Polars variant of VADiffResultSchema.

timestep instance-attribute

timestep

The timestep of this result. This indexes into the timesteps that were loaded

contingency instance-attribute

contingency

The critical contingency that caused this loadflow. For N-0 results, the special CO 'BASECASE' is used.

element instance-attribute

element

The element over which the voltage angle difference is computed. Can be either an open switch or any switch or branch under N-1. If under N-1, then element and contingency are the same.

va_diff class-attribute instance-attribute

va_diff = pa.Field(nullable=True)

The voltage angle difference in degrees between the two ends of the element. nan if at least one of the ends has no voltage angle (island, out of service)

element_name instance-attribute

element_name

The name of the Branch or Switch, if available. This is not used for the loadflow computation, but can be used for display purposes. If no name is available, this should be set to an empty string.

contingency_name instance-attribute

contingency_name

The name of the contingency, if available. This is not used for the loadflow computation, but can be used for display purposes. If no name is available, this should be set to an empty string.

SwitchResultsSchemaPolars

Bases: DataFrameModel, SwitchResultsSchema

Polars variant of SwitchResultsSchema.

timestep instance-attribute

timestep

The timestep of this result. This indexes into the timesteps that were loaded

contingency instance-attribute

contingency

The critical contingency that caused this loadflow. For N-0 results, the special CO 'BASECASE' is used.

element instance-attribute

element

The element over which the voltage angle difference is computed. Can be either an open switch or any switch or branch under N-1. If under N-1, then element and contingency are the same.

p class-attribute instance-attribute

p = pa.Field(nullable=True)

The accumulated absolute active power at the node in MW, obtained by summing the absolute active power of all branches and injections connected to the node.

If the engine does not support the computation of this value, the column can be omitted.

q class-attribute instance-attribute

q = pa.Field(nullable=True)

The accumulated absolute reactive power at the node in MVar, obtained by summing the absolute reactive power of all branches and injections connected to the node

If the engine does not support the computation of this value, the column can be omitted.

vm class-attribute instance-attribute

vm = pa.Field(nullable=True)

The voltage magnitude at the node in kV.

In DC, this should be the nominal voltage of the node. This should only be NaN if the node does not have a connection to the slack bus.

i class-attribute instance-attribute

i = pa.Field(nullable=True)

The current in the branch in A

This should only be NaN if the branch has no connection to the slack bus.

element_name instance-attribute

element_name

The name of the Branch or Switch, if available. This is not used for the loadflow computation, but can be used for display purposes. If no name is available, this should be set to an empty string.

contingency_name instance-attribute

contingency_name

The name of the contingency, if available. This is not used for the loadflow computation, but can be used for display purposes. If no name is available, this should be set to an empty string.

side class-attribute instance-attribute

side = pa.Field(nullable=True)

The measurement side of the switch result.

• "from": values measured at the from-bus terminal (taken from net.res_switch).
• "to": values measured at the to-bus terminal (taken from net.res_switch).
• null: result was computed by aggregating branch flows and node injections. Switches modelled without impedance have identical electrical conditions on both terminals (no voltage drop, no power loss across the switch), so a single aggregated value is sufficient and no side distinction is needed.

RegulatingElementResultSchemaPolars

Bases: DataFrameModel, RegulatingElementResultSchema

Polars variant of RegulatingElementResultSchema.

timestep instance-attribute

timestep

The timestep of this result. This indexes into the timesteps that were loaded

contingency instance-attribute

contingency

The critical contingency that caused this loadflow. For N-0 results, the special CO 'BASECASE' is used.

element instance-attribute

element

The regulating element that these loadflow results correspond to

value instance-attribute

value

The value of the regulating element. Depending on the type of the regulating element, this can mean different things.

regulating_element_type class-attribute instance-attribute

regulating_element_type = pa.Field(
    isin=[(side.value) for side in RegulatingElementType]
)

The type of the regulating element (generator, regulating transformer, SVC, ...).

element_name instance-attribute

element_name

The name of the Regulating Element, if available. This is not used for the loadflow computation, but can be used for display purposes. If no name is available, this should be set to an empty string.

contingency_name instance-attribute

contingency_name

The name of the contingency, if available. This is not used for the loadflow computation, but can be used for display purposes. If no name is available, this should be set to an empty string.

ConvergedSchemaPolars

Bases: DataFrameModel, ConvergedSchema

Polars variant of ConvergedSchema.

timestep instance-attribute

timestep

The timestep of this result. This indexes into the timesteps that were loaded

contingency instance-attribute

contingency

The critical contingency that caused this loadflow. For N-0 results, the special CO 'BASECASE' is used.

status class-attribute instance-attribute

status = pa.Field(
    isin=[(side.value) for side in ConvergenceStatus]
)

Whether the loadflow converged at this timestep/contingency.

iteration_count class-attribute instance-attribute

iteration_count = pa.Field(nullable=True)

The number of iterations required for the loadflow to converge.

warnings class-attribute instance-attribute

warnings = pa.Field(nullable=True)

An additional string field that carries warnings or error logs for specific timesteps/contingencys/components.

contingency_name instance-attribute

contingency_name

The name of the contingency, if available. This is not used for the loadflow computation, but can be used for display purposes. If no name is available, this should be set to an empty string.

SppsResultsSchemaPolars

Bases: DataFrameModel, SppsResultsSchema

Polars variant of SppsResultsSchema.

timestep instance-attribute

timestep

Loadflow timestep index for the outage that produced this row.

contingency instance-attribute

contingency

Globally unique id of the contingency (same value as in other loadflow result tables).

iterations instance-attribute

iterations

Number of SpPS iterations that executed (1-based count).

activated_schemes_per_iter class-attribute instance-attribute

activated_schemes_per_iter = pa.Field()

JSON string for list[list[str]] of scheme names that activated per iteration.

max_iterations_reached instance-attribute

max_iterations_reached

Whether the run stopped because the iteration cap was hit while schemes still fired.

power_flow_failed instance-attribute

power_flow_failed

Whether a post-action power flow failed in keep_previous mode.

CascadeResultSchemaPolars

Bases: DataFrameModel, CascadeResultSchema

Polars variant of CascadeResultSchema.

timestep instance-attribute

timestep

The timestep of this cascade event.

contingency instance-attribute

contingency

Globally unique id of the contingency that started the cascade.

cascade_number instance-attribute

cascade_number

Cascade step number where the event happened.

element_mrid class-attribute instance-attribute

element_mrid = pa.Field(nullable=True)

External identifier of the affected element, if known.

element_id class-attribute instance-attribute

element_id = pa.Field(nullable=True)

Globally unique id of the affected element, if known.

contingency_outage_id class-attribute instance-attribute

contingency_outage_id = pa.Field(nullable=True)

Identifier of the outage group for the contingency that started the cascade.

contingency_name class-attribute instance-attribute

contingency_name = pa.Field(nullable=True)

Human-readable name of the contingency that started the cascade, if known.

element_outage_group_id class-attribute instance-attribute

element_outage_group_id = pa.Field(nullable=True)

Stable identifier of the affected element's outage group.

element_name class-attribute instance-attribute

element_name = pa.Field(nullable=True)

Human-readable name of the affected element, if known.

cascade_reason instance-attribute

cascade_reason

Reason for the cascade event, such as current overload or distance protection.

loading class-attribute instance-attribute

loading = pa.Field(nullable=True)

Branch loading value that triggered the cascade event, if available.

r_ohm class-attribute instance-attribute

r_ohm = pa.Field(nullable=True)

Relay resistance value for distance-protection events, if available.

x_ohm class-attribute instance-attribute

x_ohm = pa.Field(nullable=True)

Relay reactance value for distance-protection events, if available.

distance_protection_severity class-attribute instance-attribute

distance_protection_severity = pa.Field(nullable=True)

Distance-protection severity for relay events. Empty for other event types.

activated_schemes_per_iter class-attribute instance-attribute

activated_schemes_per_iter = pa.Field(nullable=True)

JSON string of SpPS scheme names that activated per inner cascade load-flow iteration.

LoadflowResultsPolars

Bases: BaseModel

A container for the loadflow results for a computation job.

job_id instance-attribute

job_id

The id of the computation job that created these loadflows

branch_results class-attribute instance-attribute

branch_results = None

The results for the branches. If no branches are monitored, this is the empty DataFrame. Non converging contingencys/timesteps are to be omitted

node_results class-attribute instance-attribute

node_results = None

The results for the nodes. If no nodes are monitored, this is the empty DataFrame.

regulating_element_results class-attribute instance-attribute

regulating_element_results = None

The results for the regulating elements. If no regulating elements are monitored, this is the empty DataFrame.

converged class-attribute instance-attribute

converged = None

The convergence information for each timestep and contingency. If there were non-converging loadflows for some timesteps/contingencys, these results should be omitted from the other tables.

va_diff_results class-attribute instance-attribute

va_diff_results = None

The voltage angle difference results for each timestep and contingency. Considers the ends of the outaged branch, aswell as all open switches in monitored elements.

switch_results class-attribute instance-attribute

switch_results = None

The results for the switches.

Contains aggregated power flow and injection results per switch for each timestep and contingency.

Switch results are computed by aggregating contributions from all elements (branches and buses) electrically connected to one side of the switch. This represents the power flowing through the switch.

If no switches are monitored, this is the empty DataFrame. For non-converging contingencies/timesteps, result values are present but set to NaN.

connectivity_result class-attribute instance-attribute

connectivity_result = None

Connectivity mapping between contingencies and affected grid elements. This DataFrame defines which elements become unavailable for each contingency, based on outage group logic. Each row represents a (contingency, element) pair, indicating that the element is part of the outage group triggered by the contingency.

warnings class-attribute instance-attribute

warnings = Field(default_factory=list)

Global warnings that occured during the computation (e.g. monitored elements/contingencies that were not found)

spps_results class-attribute instance-attribute

spps_results = None

SpPS run summaries, concatenated in single-outage order. Empty when no SpPS was recorded (default).

cascade_results class-attribute instance-attribute

cascade_results = None

Cascade simulation events. Empty when cascade simulation is disabled or has no events.

Config

Pydantic configuration for the LoadflowResultsPolars model.

arbitrary_types_allowed class-attribute instance-attribute

arbitrary_types_allowed = True

Allow arbitrary types in the model.

__eq__

__eq__(lf_result)

Compare two LoadflowResults objects for equality.

Rounds floats to 6 decimal places for comparison. This is necessary because floating point arithmetic can lead to small differences in the results.

Ignores the order of the DataFrames, but checks that the indices are equal.

Note: This functions is not very efficient and can take up to half a minute for >10Mio rows.

PARAMETER	DESCRIPTION
lf_result	The LoadflowResults object to compare with. TYPE: LoadflowResults

RETURNS	DESCRIPTION
bool	True if the two LoadflowResults objects are equal, False otherwise.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/loadflow_results_polars.py

def __eq__(self, lf_result: Self) -> bool:
    """Compare two LoadflowResults objects for equality.

    Rounds floats to 6 decimal places for comparison.
    This is necessary because floating point arithmetic can lead to small differences in the results.

    Ignores the order of the DataFrames, but checks that the indices are equal.

    Note: This functions is not very efficient and can take up to half a minute for >10Mio rows.

    Parameters
    ----------
    lf_result : LoadflowResults
        The LoadflowResults object to compare with.

    Returns
    -------
    bool
        True if the two LoadflowResults objects are equal, False otherwise.
    """
    rounding_accuracy = 1e-6

    if not isinstance(lf_result, LoadflowResultsPolars):
        return False

    job_match = self.job_id == lf_result.job_id
    warnings_match = self.warnings == lf_result.warnings
    simple_checks = job_match and warnings_match
    if not simple_checks:
        return False

    kw_args_testing = {
        "check_row_order": True,
        "check_column_order": False,
        "check_dtypes": True,
        "check_exact": False,
        "abs_tol": rounding_accuracy,
    }

    def assert_optional_frame_equal(left: pl.LazyFrame | None, right: pl.LazyFrame | None) -> None:
        """Assert that optional polars frames are equal.

        Parameters
        ----------
        left : pl.LazyFrame | None
            First optional frame.
        right : pl.LazyFrame | None
            Second optional frame.

        Raises
        ------
        AssertionError
            If one frame is None and the other is not, or if frame contents differ.
        """
        if left is None or right is None:
            if left is not right:
                raise AssertionError("One frame is None and the other is not.")
            return
        row_identifier_columns = [
            column
            for column in ("timestep", "contingency", "element", "side", "cascade_number")
            if column in left.collect_schema().names()
        ]
        assert_frame_equal(
            left.fill_nan(None).sort(row_identifier_columns).collect(),
            right.fill_nan(None).sort(row_identifier_columns).collect(),
            **kw_args_testing,
        )

    try:
        assert_optional_frame_equal(self.branch_results, lf_result.branch_results)
        assert_optional_frame_equal(self.node_results, lf_result.node_results)
        assert_optional_frame_equal(self.regulating_element_results, lf_result.regulating_element_results)
        assert_optional_frame_equal(self.va_diff_results, lf_result.va_diff_results)
        assert_optional_frame_equal(self.converged, lf_result.converged)
        assert_optional_frame_equal(self.spps_results, lf_result.spps_results)
        assert_optional_frame_equal(self.cascade_results, lf_result.cascade_results)
    except AssertionError:
        return False

    return True

toop_engine_interfaces.loadflow_result_helpers_polars

Loadflow result helpers for polars LazyFrames or DataFrames.

Holds functions to work with the loadflow results interfaces.

save_loadflow_results_polars

save_loadflow_results_polars(fs, file_path, loadflows)

Save loadflow results to a file in hdf5 format.

PARAMETER	DESCRIPTION
fs	The filesystem to use to save the results. This can be a local filesystem or an object store like S3 or Azure, using the fsspec library. For writing to local disk, you should use the DirFilesystem to inject a base path like this: from fsspec.implementations.local import DirFileSystem fs = DirFileSystem(base_path="/path/to/base") Similarly, buckets can be used with the appropriate fsspec filesystem like adbs TYPE: AbstractFileSystem
file_path	The file path where to save the results to. This is relative to the base name or bucket defined in the storage TYPE: str | Path
loadflows	The loadflow results to save. TYPE: LoadflowResultsPolars

RETURNS	DESCRIPTION
StoredLoadflowReference	A reference to the stored loadflow results.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/loadflow_result_helpers_polars.py

def save_loadflow_results_polars(
    fs: AbstractFileSystem, file_path: str | Path, loadflows: LoadflowResultsPolars
) -> StoredLoadflowReference:
    """Save loadflow results to a file in hdf5 format.

    Parameters
    ----------
    fs : AbstractFileSystem
        The filesystem to use to save the results. This can be a local filesystem or an object store like S3 or Azure, using
        the fsspec library. For writing to local disk, you should use the DirFilesystem to inject a base path like this:
        ```python
        from fsspec.implementations.local import DirFileSystem
        fs = DirFileSystem(base_path="/path/to/base")
        ```
        Similarly, buckets can be used with the appropriate fsspec filesystem like adbs
    file_path: str | Path
        The file path where to save the results to. This is relative to the base name or bucket defined in the storage
    loadflows : LoadflowResultsPolars
        The loadflow results to save.

    Returns
    -------
    StoredLoadflowReference
        A reference to the stored loadflow results.
    """
    file_path = str(file_path)
    fs.makedirs(file_path, exist_ok=True)
    metadata = {
        "job_id": loadflows.job_id,
        "warnings": loadflows.warnings,
    }
    with fs.open(file_path + "/metadata.json", "w") as f:
        json.dump(metadata, f)

    with fs.open(file_path + "/branch_results.parquet", "wb") as f:
        loadflows.branch_results.sink_parquet(f)
    with fs.open(file_path + "/node_results.parquet", "wb") as f:
        loadflows.node_results.sink_parquet(f)
    with fs.open(file_path + "/regulating_element_results.parquet", "wb") as f:
        loadflows.regulating_element_results.sink_parquet(f)
    with fs.open(file_path + "/converged.parquet", "wb") as f:
        loadflows.converged.sink_parquet(f)
    with fs.open(file_path + "/va_diff_results.parquet", "wb") as f:
        loadflows.va_diff_results.sink_parquet(f)
    if loadflows.cascade_results is not None:
        with fs.open(file_path + "/cascade_results.parquet", "wb") as f:
            loadflows.cascade_results.sink_parquet(f)
    if loadflows.switch_results is not None:
        with fs.open(file_path + "/switch_results.parquet", "wb") as f:
            loadflows.switch_results.sink_parquet(f)
    if loadflows.spps_results is not None:
        with fs.open(file_path + "/spps_results.parquet", "wb") as f:
            loadflows.spps_results.sink_parquet(f)

    return StoredLoadflowReference(
        relative_path=str(file_path),
    )

load_loadflow_results_polars

load_loadflow_results_polars(fs, reference, validate=True)

Load loadflow results from a StoredLoadflowReference.

PARAMETER	DESCRIPTION
fs	The filesystem to use to load the results. This can be a local filesystem or an object store like S3 or Azure, using the fsspec library. TYPE: AbstractFileSystem
reference	The reference to the stored loadflow results. TYPE: StoredLoadflowReference
validate	Whether to validate the loaded results against the schemas defined in the interfaces. For dataframes with a lot of data this can take a few seconds. TYPE: bool DEFAULT: True

RETURNS	DESCRIPTION
LoadflowResults	The loaded loadflow results.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/loadflow_result_helpers_polars.py

def load_loadflow_results_polars(
    fs: AbstractFileSystem, reference: StoredLoadflowReference, validate: bool = True
) -> LoadflowResultsPolars:
    """Load loadflow results from a StoredLoadflowReference.

    Parameters
    ----------
    fs: AbstractFileSystem
        The filesystem to use to load the results. This can be a local filesystem or an object store like S3 or Azure, using
        the fsspec library.
    reference: StoredLoadflowReference
        The reference to the stored loadflow results.
    validate: bool
        Whether to validate the loaded results against the schemas defined in the interfaces.
        For dataframes with a lot of data this can take a few seconds.

    Returns
    -------
    LoadflowResults
        The loaded loadflow results.
    """
    file_path = str(reference.relative_path)
    with fs.open(file_path + "/metadata.json", "r") as f:
        metadata = json.load(f)
    job_id = metadata["job_id"]
    warnings = metadata["warnings"]

    with fs.open(file_path + "/branch_results.parquet", "rb") as f:
        branch_results = pl.scan_parquet(f)
    with fs.open(file_path + "/node_results.parquet", "rb") as f:
        node_results = pl.scan_parquet(f)
    with fs.open(file_path + "/regulating_element_results.parquet", "rb") as f:
        regulating_element_results = pl.scan_parquet(f)
    with fs.open(file_path + "/converged.parquet", "rb") as f:
        converged = pl.scan_parquet(f)
    with fs.open(file_path + "/va_diff_results.parquet", "rb") as f:
        va_diff_results = pl.scan_parquet(f)
    if fs.exists(file_path + "/cascade_results.parquet"):
        with fs.open(file_path + "/cascade_results.parquet", "rb") as f:
            cascade_results = pl.scan_parquet(f)
    else:
        cascade_results = None
    if fs.exists(file_path + "/switch_results.parquet"):
        with fs.open(file_path + "/switch_results.parquet", "rb") as f:
            switch_results = pl.scan_parquet(f)
    else:
        switch_results = None
    if fs.exists(file_path + "/spps_results.parquet"):
        with fs.open(file_path + "/spps_results.parquet", "rb") as f:
            spps_results = pl.scan_parquet(f)
    else:
        spps_results = None

    if validate:
        return LoadflowResultsPolars(
            job_id=job_id,
            branch_results=BranchResultSchemaPolars.validate(branch_results),
            node_results=NodeResultSchemaPolars.validate(node_results),
            regulating_element_results=RegulatingElementResultSchemaPolars.validate(regulating_element_results),
            converged=ConvergedSchemaPolars.validate(converged),
            va_diff_results=VADiffResultSchemaPolars.validate(va_diff_results),
            cascade_results=(CascadeResultSchemaPolars.validate(cascade_results) if cascade_results is not None else None),
            switch_results=(SwitchResultsSchemaPolars.validate(switch_results) if switch_results is not None else None),
            spps_results=(SppsResultsSchemaPolars.validate(spps_results) if spps_results is not None else None),
            warnings=warnings,
        )

    return LoadflowResultsPolars.model_construct(
        job_id=job_id,
        branch_results=branch_results,
        node_results=node_results,
        regulating_element_results=regulating_element_results,
        converged=converged,
        va_diff_results=va_diff_results,
        cascade_results=cascade_results,
        switch_results=switch_results,
        spps_results=spps_results,
        warnings=warnings,
    )

concatenate_loadflow_results_polars

concatenate_loadflow_results_polars(loadflow_results_list)

Concatenate the results of the loadflow results.

PARAMETER	DESCRIPTION
loadflow_results_list	The list of loadflow results to concatenate TYPE: list

RETURNS	DESCRIPTION
LoadflowResultsPolars	The concatenated loadflow results

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/loadflow_result_helpers_polars.py

def concatenate_loadflow_results_polars(
    loadflow_results_list: list[LoadflowResultsPolars],
) -> LoadflowResultsPolars:
    """Concatenate the results of the loadflow results.

    Parameters
    ----------
    loadflow_results_list : list
        The list of loadflow results to concatenate

    Returns
    -------
    LoadflowResultsPolars
        The concatenated loadflow results
    """
    assert len(loadflow_results_list) > 0, "The list of loadflow results must not be empty"
    assert all(loadflow_results_list[0].job_id == res.job_id for res in loadflow_results_list), (
        "All loadflow results must have the same job_id"
    )
    # make sure None values are not included in the concatenation
    branch_results_list = [res.branch_results for res in loadflow_results_list if res.branch_results is not None]
    node_results_list = [res.node_results for res in loadflow_results_list if res.node_results is not None]
    regulating_element_results_list = [
        res.regulating_element_results for res in loadflow_results_list if res.regulating_element_results is not None
    ]
    converged_list = [res.converged for res in loadflow_results_list if res.converged is not None]
    va_diff_results_list = [res.va_diff_results for res in loadflow_results_list if res.va_diff_results is not None]
    switch_results_list = [res.switch_results for res in loadflow_results_list if res.switch_results is not None]
    spps_results_list = [res.spps_results for res in loadflow_results_list if res.spps_results is not None]
    cascade_results_list = [res.cascade_results for res in loadflow_results_list if res.cascade_results is not None]

    # how="diagonal" aligns by column name, tolerating per-outage column-order differences
    # (e.g. an empty schema-derived frame vs a built one).
    branch_results = pl.concat(branch_results_list, how="diagonal")
    node_results = pl.concat(node_results_list, how="diagonal")
    regulating_element_results = pl.concat(regulating_element_results_list, how="diagonal")
    converged = pl.concat(converged_list, how="diagonal")
    va_diff_results = pl.concat(va_diff_results_list, how="diagonal")
    switch_results = pl.concat(switch_results_list, how="diagonal") if switch_results_list else None
    spps_results = pl.concat(spps_results_list, how="diagonal") if spps_results_list else None
    cascade_results = pl.concat(cascade_results_list, how="diagonal") if cascade_results_list else None
    warnings = [warning for lf_results in loadflow_results_list for warning in lf_results.warnings]
    # model_construct: the per-outage frames are eager DataFrames (the field type is LazyFrame);
    # skip validation here and rely on the final schema conversion.
    return LoadflowResultsPolars.model_construct(
        job_id=loadflow_results_list[0].job_id,
        branch_results=branch_results,
        node_results=node_results,
        regulating_element_results=regulating_element_results,
        converged=converged,
        va_diff_results=va_diff_results,
        switch_results=switch_results,
        spps_results=spps_results,
        cascade_results=cascade_results,
        warnings=warnings,
    )

select_timestep_polars

select_timestep_polars(loadflow_results, timestep)

Select a single timestep from the loadflow results.

PARAMETER	DESCRIPTION
loadflow_results	The loadflow results to select the timestep from. TYPE: LoadflowResultsPolars
timestep	The timestep to select. TYPE: int

RETURNS	DESCRIPTION
LoadflowResultsPolars	The loadflow results for the selected timestep.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/loadflow_result_helpers_polars.py

def select_timestep_polars(loadflow_results: LoadflowResultsPolars, timestep: int) -> LoadflowResultsPolars:
    """Select a single timestep from the loadflow results.

    Parameters
    ----------
    loadflow_results : LoadflowResultsPolars
        The loadflow results to select the timestep from.
    timestep : int
        The timestep to select.

    Returns
    -------
    LoadflowResultsPolars
        The loadflow results for the selected timestep.
    """
    return LoadflowResultsPolars(
        job_id=loadflow_results.job_id,
        branch_results=loadflow_results.branch_results.filter(pl.col("timestep") == timestep),
        node_results=loadflow_results.node_results.filter(pl.col("timestep") == timestep),
        regulating_element_results=loadflow_results.regulating_element_results.filter(pl.col("timestep") == timestep),
        converged=loadflow_results.converged.filter(pl.col("timestep") == timestep),
        va_diff_results=loadflow_results.va_diff_results.filter(pl.col("timestep") == timestep),
        cascade_results=(
            loadflow_results.cascade_results.filter(pl.col("timestep") == timestep)
            if loadflow_results.cascade_results is not None
            else None
        ),
        warnings=loadflow_results.warnings,
    )

subset_contingencies_polars

subset_contingencies_polars(
    loadflow_results, contingencies
)

Subset the loadflow results to a list of contingencies.

PARAMETER	DESCRIPTION
loadflow_results	The loadflow results to subset. TYPE: LoadflowResultsPolars
contingencies	The list of contingencies to subset to. TYPE: list[str]

RETURNS	DESCRIPTION
LoadflowResultsPolars	The loadflow results for the selected contingencies.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/loadflow_result_helpers_polars.py

def subset_contingencies_polars(loadflow_results: LoadflowResultsPolars, contingencies: list[str]) -> LoadflowResultsPolars:
    """Subset the loadflow results to a list of contingencies.

    Parameters
    ----------
    loadflow_results : LoadflowResultsPolars
        The loadflow results to subset.
    contingencies : list[str]
        The list of contingencies to subset to.

    Returns
    -------
    LoadflowResultsPolars
        The loadflow results for the selected contingencies.
    """
    return LoadflowResultsPolars(
        job_id=loadflow_results.job_id,
        branch_results=loadflow_results.branch_results.filter(pl.col("contingency").is_in(contingencies)),
        node_results=loadflow_results.node_results.filter(pl.col("contingency").is_in(contingencies)),
        regulating_element_results=loadflow_results.regulating_element_results.filter(
            pl.col("contingency").is_in(contingencies)
        ),
        converged=loadflow_results.converged.filter(pl.col("contingency").is_in(contingencies)),
        va_diff_results=loadflow_results.va_diff_results.filter(pl.col("contingency").is_in(contingencies)),
        cascade_results=(
            loadflow_results.cascade_results.filter(pl.col("contingency").is_in(contingencies))
            if loadflow_results.cascade_results is not None
            else None
        ),
        warnings=loadflow_results.warnings,
    )

extract_branch_results_polars

extract_branch_results_polars(
    branch_results,
    timestep,
    contingencies,
    monitored_branches,
    basecase,
)

Extract the branch results for a specific timestep.

PARAMETER	DESCRIPTION
branch_results	The branch results dataframe to extract the branch results from. TYPE: LazyFrame[BranchResultSchemaPolars]
timestep	The selected timestep to pull from the loadflow results. TYPE: int
basecase	The basecase contingency id to use for the N-0 results. TYPE: str
contingencies	The list of contingencies to extract the results for. TYPE: list[str]
monitored_branches	The list of monitored branches to extract the results for. buses switches etc should not be included here, only branches. TYPE: list[GridElement]

RETURNS	DESCRIPTION
Float[ndarray, ' n_contingencies n_branches_monitored']	The branch results with the following: - shape (n_contingencies, n_branches_monitored) - only the p values of the monitored branches at the from-end - For three winding transformers, the p values are split into three rows for each side (hv, mv, lv).

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/loadflow_result_helpers_polars.py

def extract_branch_results_polars(
    branch_results: patpl.LazyFrame[BranchResultSchemaPolars],
    timestep: int,
    contingencies: list[str],
    monitored_branches: list[GridElement],
    basecase: str,
) -> tuple[Float[np.ndarray, " n_branches_monitored"], Float[np.ndarray, " n_contingencies n_branches_monitored"]]:
    """Extract the branch results for a specific timestep.

    Parameters
    ----------
    branch_results: patpl.LazyFrame[BranchResultSchemaPolars],
        The branch results dataframe to extract the branch results from.
    timestep : int
        The selected timestep to pull from the loadflow results.
    basecase : str
        The basecase contingency id to use for the N-0 results.
    contingencies : list[str]
        The list of contingencies to extract the results for.
    monitored_branches : list[GridElement]
        The list of monitored branches to extract the results for.
        buses switches etc should not be included here, only branches.

    Returns
    -------
    Float[np.ndarray, " n_contingencies n_branches_monitored"]
        The branch results with the following:
        - shape (n_contingencies, n_branches_monitored)
        - only the p values of the monitored branches at the from-end
        - For three winding transformers, the p values are split into three rows for each side (hv, mv, lv).
    """
    assert basecase not in contingencies, "Basecase contingency should not be in the list of N-k contingencies"
    n_monitored_branches = len(monitored_branches)
    n_contingencies = len(contingencies)
    if (n_monitored_branches == 0) or (n_contingencies == 0 and basecase is None):
        # If there are no monitored branches, return empty arrays
        return (
            np.full((n_monitored_branches,), np.nan, dtype=float),
            np.full((n_contingencies, n_monitored_branches), np.nan, dtype=float),
        )
    # Get the branch results for the given job_id and timestep
    three_winding_side_dict = {
        "trafo3w_hv": BranchSide.ONE.value,
        "trafo3w_mv": BranchSide.TWO.value,
        "trafo3w_lv": BranchSide.THREE.value,
    }
    all_cases = [basecase, *contingencies]

    normal_branches_ids = [elem.id for elem in monitored_branches if elem.type not in three_winding_side_dict]

    timestep_df = branch_results.select(pl.col("timestep").unique())
    contingency_df = timestep_df.join(pl.LazyFrame({"contingency": all_cases}), how="cross")

    normal_branch_df = contingency_df.join(pl.LazyFrame({"element": normal_branches_ids}), how="cross").with_columns(
        side=BranchSide.ONE.value
    )

    trafo3w_dfs = []
    for trafo_type, side in three_winding_side_dict.items():
        three_winding_branches = [element.id for element in monitored_branches if element.type == trafo_type]
        trafo_side_df = contingency_df.join(pl.LazyFrame({"element": three_winding_branches}), how="cross").with_columns(
            side=side
        )
        trafo3w_dfs.append(trafo_side_df)
    all_branches_df = pl.concat([normal_branch_df, *trafo3w_dfs], how="vertical")

    merge_columns = ["timestep", "contingency", "element", "side"]
    all_p_results = (
        all_branches_df.join(
            branch_results.filter(pl.col("timestep") == timestep).select([*merge_columns, "p"]), on=merge_columns, how="left"
        )
        .fill_null(0.0)
        .fill_nan(0.0)
    )

    n_0_results = all_p_results.filter(pl.col("contingency") == basecase).collect()
    n_0_vector = n_0_results["p"].to_numpy()

    sort_by = [
        pl.col("contingency").cast(pl.Enum(contingencies)),
        pl.col("element").cast(pl.Enum([elem.id for elem in monitored_branches])),
    ]
    is_not_basecase = pl.col("contingency") != basecase
    n_1_results = all_p_results.filter(is_not_basecase).sort(sort_by).collect()
    n_1_array = n_1_results["p"].to_numpy().reshape(n_contingencies, n_monitored_branches)
    return n_0_vector, n_1_array

extract_node_matrices_polars

extract_node_matrices_polars(
    node_results,
    timestep,
    contingencies,
    monitored_nodes,
    basecase="BASECASE",
)

Extract the node results for a specific timestep.

PARAMETER	DESCRIPTION
node_results	The node results polars dataframe to extract the node results from. TYPE: LazyFrame[NodeResultSchemaPolars]
timestep	The selected timestep to pull from the loadflow results. TYPE: int
basecase	The basecase contingency id to use for the N-0 results. TYPE: str DEFAULT: 'BASECASE'
contingencies	The list of contingencies to extract the results for. TYPE: list[str]
monitored_nodes	The list of monitored nodes to extract the results for. buses switches etc should not be included here, only nodes. TYPE: list[GridElement]

RETURNS	DESCRIPTION
vm_n0	The voltage magnitude results for the basecase contingency at the monitored nodes. TYPE: Float[ndarray, ' n_nodes_monitored']
va_n0	The voltage angle results for the basecase contingency at the monitored nodes. TYPE: Float[ndarray, ' n_nodes_monitored']
vm_n1	The voltage magnitude results for the contingencies at the monitored nodes. TYPE: Float[ndarray, ' n_contingencies n_nodes_monitored']
va_n1	The voltage angle results for the contingencies at the monitored nodes. TYPE: Float[ndarray, ' n_contingencies n_nodes_monitored']

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/loadflow_result_helpers_polars.py

def extract_node_matrices_polars(
    node_results: patpl.LazyFrame[NodeResultSchemaPolars],
    timestep: int,
    contingencies: list[str],
    monitored_nodes: list[GridElement],
    basecase: str = "BASECASE",
) -> tuple[
    Float[np.ndarray, " n_nodes_monitored"],
    Float[np.ndarray, "  n_nodes_monitored"],
    Float[np.ndarray, " n_contingencies n_nodes_monitored"],
    Float[np.ndarray, " n_contingencies n_nodes_monitored"],
]:
    """Extract the node results for a specific timestep.

    Parameters
    ----------
    node_results: patpl.LazyFrame[NodeResultSchemaPolars],
        The node results polars dataframe to extract the node results from.
    timestep : int
        The selected timestep to pull from the loadflow results.
    basecase : str
        The basecase contingency id to use for the N-0 results.
    contingencies : list[str]
        The list of contingencies to extract the results for.
    monitored_nodes : list[GridElement]
        The list of monitored nodes to extract the results for.
        buses switches etc should not be included here, only nodes.

    Returns
    -------
    vm_n0 : Float[np.ndarray, " n_nodes_monitored"]
        The voltage magnitude results for the basecase contingency at the monitored nodes.
    va_n0 : Float[np.ndarray, " n_nodes_monitored"]
        The voltage angle results for the basecase contingency at the monitored nodes.
    vm_n1 : Float[np.ndarray, " n_contingencies n_nodes_monitored"]
        The voltage magnitude results for the contingencies at the monitored nodes.
    va_n1 : Float[np.ndarray, " n_contingencies n_nodes_monitored"]
        The voltage angle results for the contingencies at the monitored nodes.
    """
    assert basecase not in contingencies, "Basecase contingency should not be in the list of N-k contingencies"
    n_contingencies = len(contingencies)
    n_monitored_nodes = len(monitored_nodes)
    if (n_monitored_nodes == 0) or (n_contingencies == 0 and basecase is None):
        # If there are no monitored nodes, return empty arrays
        return (
            np.full(n_monitored_nodes, dtype=float),
            np.full(n_monitored_nodes, dtype=float),
            np.full((n_contingencies, n_monitored_nodes), dtype=float),
            np.full((n_contingencies, n_monitored_nodes), dtype=float),
        )

    # Get the node results for the given job_id and timestep
    contingency_df = pl.LazyFrame({"contingency": [basecase, *contingencies]})
    all_cases_df = contingency_df.join(pl.LazyFrame({"element": [elem.id for elem in monitored_nodes]}), how="cross")
    node_results = all_cases_df.join(
        node_results.filter(pl.col("timestep") == timestep).select(["contingency", "element", "vm", "va"]),
        on=["contingency", "element"],
        how="left",
    )
    v_n0 = node_results.filter(pl.col("contingency") == basecase).select(["vm", "va"]).collect()
    vm_n0 = v_n0["vm"].to_numpy()
    va_n0 = v_n0["va"].to_numpy()

    v_n1 = node_results.filter(pl.col("contingency") != basecase).select(["vm", "va"]).collect()
    vm_n1 = v_n1["vm"].to_numpy()
    va_n1 = v_n1["va"].to_numpy()
    # reshape the results to have the contingencies as first dimension
    vm_n1_reshaped = vm_n1.reshape(len(contingencies), len(monitored_nodes))
    va_n1_reshaped = va_n1.reshape(len(contingencies), len(monitored_nodes))
    return vm_n0, va_n0, vm_n1_reshaped, va_n1_reshaped

extract_solver_matrices_polars

extract_solver_matrices_polars(
    loadflow_results, nminus1_definition, timestep
)

Extract the N-0 and N-1 matrices in a similar format to the DC solver.

PARAMETER	DESCRIPTION
loadflow_results	The loadflow results to extract the matrices from. TYPE: LoadflowResults
nminus1_definition	The N-1 definition to use for the contingencies and monitored elements. TYPE: Nminus1Definition
timestep	The selected timestep to pull from the loadflow results. TYPE: int

RETURNS	DESCRIPTION
Float[ndarray, ' n_branches_monitored']	The N-0 matrix
Float[ndarray, ' n_solver_contingencies n_branches_monitored']	The N-1 matrix
Bool[ndarray, ' n_solver_contingencies']	The convergence status of the contingencies in the N-1 matrix True if converged or not calculated, False if not converged.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/loadflow_result_helpers_polars.py

def extract_solver_matrices_polars(
    loadflow_results: LoadflowResultsPolars,
    nminus1_definition: Nminus1Definition,
    timestep: int,
) -> tuple[
    Float[np.ndarray, " n_branches_monitored"],
    Float[np.ndarray, " n_solver_contingencies n_branches_monitored"],
    Bool[np.ndarray, " n_solver_contingencies"],
]:
    """Extract the N-0 and N-1 matrices in a similar format to the DC solver.

    Parameters
    ----------
    loadflow_results : LoadflowResults
        The loadflow results to extract the matrices from.
    nminus1_definition : Nminus1Definition
        The N-1 definition to use for the contingencies and monitored elements.
    timestep : int
        The selected timestep to pull from the loadflow results.

    Returns
    -------
    Float[np.ndarray, " n_branches_monitored"]
        The N-0 matrix
    Float[np.ndarray, " n_solver_contingencies n_branches_monitored"]
        The N-1 matrix
    Bool[np.ndarray, " n_solver_contingencies"]
        The convergence status of the contingencies in the N-1 matrix
        True if converged or not calculated, False if not converged.
    """
    basecase = next((cont for cont in nminus1_definition.contingencies if cont.is_basecase()), None)
    assert basecase is not None, "No basecase contingency found in the N-1 definition."
    contingency_order = [cont.id for cont in nminus1_definition.contingencies if not cont.is_basecase()]

    # Only consider the selected timestep
    timestep_filter = pl.col("timestep") == timestep
    # For n-1 results, only consider non-basecase contingencies
    not_basecase_filter = pl.col("contingency") != basecase.id
    # A contingency is considered successful if it converged or if no calculation was performed
    # (e.g. for disconnected elements)
    filtered_converged = (
        loadflow_results.converged.filter(timestep_filter & not_basecase_filter).select(["contingency", "status"]).collect()
    )
    success_by_contingency = {
        row["contingency"]: row["status"] in {ConvergenceStatus.CONVERGED.value, ConvergenceStatus.NO_CALCULATION.value}
        for row in filtered_converged.iter_rows(named=True)
    }
    success = np.array(
        [success_by_contingency.get(contingency_id, False) for contingency_id in contingency_order], dtype=bool
    )

    branch_elements = [elem for elem in nminus1_definition.monitored_elements if elem.kind == "branch"]
    n_0_vector, n1_matrix = extract_branch_results_polars(
        branch_results=loadflow_results.branch_results,
        timestep=timestep,
        contingencies=contingency_order,
        monitored_branches=branch_elements,
        basecase=basecase.id,
    )

    return n_0_vector, n1_matrix, success

toop_engine_interfaces.loadflow_result_helpers

Loadflow result helpers. Holds functions to work with the loadflow results interfaces.

save_loadflow_results

save_loadflow_results(fs, file_path, loadflows)

Save loadflow results to a file in hdf5 format.

PARAMETER	DESCRIPTION
fs	The filesystem to use to save the results. This can be a local filesystem or an object store like S3 or Azure, using the fsspec library. For writing to local disk, you should use the DirFilesystem to inject a base path like this: from fsspec.implementations.local import DirFileSystem fs = DirFileSystem(base_path="/path/to/base") Similarly, buckets can be used with the appropriate fsspec filesystem like adbs TYPE: AbstractFileSystem
file_path	The file path where to save the results to. This is relative to the base name or bucket defined in the storage TYPE: str | Path
loadflows	The loadflow results to save. TYPE: LoadflowResults

RETURNS	DESCRIPTION
StoredLoadflowReference	A reference to the stored loadflow results.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/loadflow_result_helpers.py

def save_loadflow_results(
    fs: AbstractFileSystem, file_path: str | Path, loadflows: LoadflowResults
) -> StoredLoadflowReference:
    """Save loadflow results to a file in hdf5 format.

    Parameters
    ----------
    fs : AbstractFileSystem
        The filesystem to use to save the results. This can be a local filesystem or an object store like S3 or Azure, using
        the fsspec library. For writing to local disk, you should use the DirFilesystem to inject a base path like this:
        ```python
        from fsspec.implementations.local import DirFileSystem
        fs = DirFileSystem(base_path="/path/to/base")
        ```
        Similarly, buckets can be used with the appropriate fsspec filesystem like adbs
    file_path: str | Path
        The file path where to save the results to. This is relative to the base name or bucket defined in the storage
    loadflows : LoadflowResults
        The loadflow results to save.

    Returns
    -------
    StoredLoadflowReference
        A reference to the stored loadflow results.
    """
    file_path = str(file_path)
    fs.makedirs(file_path, exist_ok=True)
    metadata = {
        "job_id": loadflows.job_id,
        "warnings": loadflows.warnings,
    }
    with fs.open(file_path + "/metadata.json", "w") as f:
        json.dump(metadata, f)

    with fs.open(file_path + "/branch_results.parquet", "wb") as f:
        loadflows.branch_results.to_parquet(f)
    with fs.open(file_path + "/node_results.parquet", "wb") as f:
        loadflows.node_results.to_parquet(f)
    with fs.open(file_path + "/regulating_element_results.parquet", "wb") as f:
        loadflows.regulating_element_results.to_parquet(f)
    with fs.open(file_path + "/converged.parquet", "wb") as f:
        loadflows.converged.to_parquet(f)
    with fs.open(file_path + "/va_diff_results.parquet", "wb") as f:
        loadflows.va_diff_results.to_parquet(f)
    if loadflows.cascade_results is not None:
        with fs.open(file_path + "/cascade_results.parquet", "wb") as f:
            loadflows.cascade_results.to_parquet(f)

    return StoredLoadflowReference(
        relative_path=str(file_path),
    )

load_loadflow_results

load_loadflow_results(fs, reference, validate=True)

Load loadflow results from a StoredLoadflowReference.

PARAMETER	DESCRIPTION
fs	The filesystem to use to load the results. This can be a local filesystem or an object store like S3 or Azure, using the fsspec library. TYPE: AbstractFileSystem
reference	The reference to the stored loadflow results. TYPE: StoredLoadflowReference
validate	Whether to validate the loaded results against the schemas defined in the interfaces. For dataframes with a lot of data this can take a few seconds. TYPE: bool DEFAULT: True

RETURNS	DESCRIPTION
LoadflowResults	The loaded loadflow results.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/loadflow_result_helpers.py

def load_loadflow_results(
    fs: AbstractFileSystem, reference: StoredLoadflowReference, validate: bool = True
) -> LoadflowResults:
    """Load loadflow results from a StoredLoadflowReference.

    Parameters
    ----------
    fs: AbstractFileSystem
        The filesystem to use to load the results. This can be a local filesystem or an object store like S3 or Azure, using
        the fsspec library.
    reference: StoredLoadflowReference
        The reference to the stored loadflow results.
    validate: bool
        Whether to validate the loaded results against the schemas defined in the interfaces.
        For dataframes with a lot of data this can take a few seconds.

    Returns
    -------
    LoadflowResults
        The loaded loadflow results.
    """
    file_path = str(reference.relative_path)
    with fs.open(file_path + "/metadata.json", "r") as f:
        metadata = json.load(f)
    job_id = metadata["job_id"]
    warnings = metadata["warnings"]

    with fs.open(file_path + "/branch_results.parquet", "rb") as f:
        branch_results = pd.read_parquet(f)
    with fs.open(file_path + "/node_results.parquet", "rb") as f:
        node_results = pd.read_parquet(f)
    with fs.open(file_path + "/regulating_element_results.parquet", "rb") as f:
        regulating_element_results = pd.read_parquet(f)
    with fs.open(file_path + "/converged.parquet", "rb") as f:
        converged = pd.read_parquet(f)
    with fs.open(file_path + "/va_diff_results.parquet", "rb") as f:
        va_diff_results = pd.read_parquet(f)
    if fs.exists(file_path + "/cascade_results.parquet"):
        with fs.open(file_path + "/cascade_results.parquet", "rb") as f:
            cascade_results = pd.read_parquet(f)
    else:
        cascade_results = get_empty_dataframe_from_model(CascadeResultSchema)

    if validate:
        BranchResultSchema.validate(branch_results)
        NodeResultSchema.validate(node_results)
        RegulatingElementResultSchema.validate(regulating_element_results)
        ConvergedSchema.validate(converged)
        VADiffResultSchema.validate(va_diff_results)
        CascadeResultSchema.validate(cascade_results)
        return LoadflowResults(
            job_id=job_id,
            branch_results=branch_results,
            node_results=node_results,
            regulating_element_results=regulating_element_results,
            converged=converged,
            va_diff_results=va_diff_results,
            cascade_results=cascade_results,
            warnings=warnings,
        )
    return LoadflowResults.model_construct(
        job_id=job_id,
        branch_results=branch_results,
        node_results=node_results,
        regulating_element_results=regulating_element_results,
        converged=converged,
        va_diff_results=va_diff_results,
        cascade_results=cascade_results,
        warnings=warnings,
    )

concatenate_loadflow_results

concatenate_loadflow_results(loadflow_results_list)

Concatenate the results of the loadflow results.

PARAMETER	DESCRIPTION
loadflow_results_list	The list of loadflow results to concatenate TYPE: list

RETURNS	DESCRIPTION
LoadflowResults	The concatenated loadflow results

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/loadflow_result_helpers.py

def concatenate_loadflow_results(
    loadflow_results_list: list[LoadflowResults],
) -> LoadflowResults:
    """Concatenate the results of the loadflow results.

    Parameters
    ----------
    loadflow_results_list : list
        The list of loadflow results to concatenate

    Returns
    -------
    LoadflowResults
        The concatenated loadflow results
    """
    assert len(loadflow_results_list) > 0, "The list of loadflow results must not be empty"
    assert all(loadflow_results_list[0].job_id == res.job_id for res in loadflow_results_list), (
        "All loadflow results must have the same job_id"
    )
    branch_results = pd.concat([res.branch_results for res in loadflow_results_list], axis=0)
    node_results = pd.concat([res.node_results for res in loadflow_results_list], axis=0)
    regulating_element_results = pd.concat([res.regulating_element_results for res in loadflow_results_list], axis=0)
    converged = pd.concat([res.converged for res in loadflow_results_list], axis=0)
    va_diff_results = pd.concat([res.va_diff_results for res in loadflow_results_list], axis=0)
    switch_results = pd.concat([res.switch_results for res in loadflow_results_list], axis=0)
    cascade_results = pd.concat(
        [
            lf.cascade_results if lf.cascade_results is not None else get_empty_dataframe_from_model(CascadeResultSchema)
            for lf in loadflow_results_list
        ],
        axis=0,
    )
    warnings = [warning for lf_results in loadflow_results_list for warning in lf_results.warnings]
    spps_results = pd.concat(
        [
            lf.spps_results if lf.spps_results is not None else get_empty_dataframe_from_model(SppsResultsSchema)
            for lf in loadflow_results_list
        ],
        axis=0,
    )
    return LoadflowResults(
        job_id=loadflow_results_list[0].job_id,
        branch_results=branch_results,
        node_results=node_results,
        regulating_element_results=regulating_element_results,
        converged=converged,
        va_diff_results=va_diff_results,
        switch_results=switch_results,
        warnings=warnings,
        spps_results=spps_results,
        cascade_results=cascade_results,
    )

get_failed_branch_results

get_failed_branch_results(
    timestep,
    failed_outages,
    monitored_2_end_branches,
    monitored_3_end_branches,
)

Get the failed branch results.

PARAMETER	DESCRIPTION
timestep	The timestep of the results TYPE: int
failed_outages	The list of failed outages TYPE: list
monitored_2_end_branches	The list of monitored 2 end branches. i.e. most branches TYPE: list
monitored_3_end_branches	The list of monitored 3 end branches. i.e. 3 winding transformers TYPE: list

RETURNS	DESCRIPTION
DataFrame[BranchResultSchema]	The failed branch results

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/loadflow_result_helpers.py

@pa.check_types
def get_failed_branch_results(
    timestep: int, failed_outages: list[str], monitored_2_end_branches: list[str], monitored_3_end_branches: list[str]
) -> pat.DataFrame[BranchResultSchema]:
    """Get the failed branch results.

    Parameters
    ----------
    timestep : int
        The timestep of the results
    failed_outages : list
        The list of failed outages
    monitored_2_end_branches : list
        The list of monitored 2 end branches. i.e. most branches
    monitored_3_end_branches : list
        The list of monitored 3 end branches. i.e. 3 winding transformers

    Returns
    -------
    pat.DataFrame[BranchResultSchema]
        The failed branch results
    """
    # With two Sides
    failed_branch_results = pd.DataFrame(
        index=pd.MultiIndex.from_product(
            [[timestep], failed_outages, monitored_2_end_branches, [BranchSide.ONE.value, BranchSide.TWO.value]],
            names=["timestep", "contingency", "element", "side"],
        )
    ).assign(p=np.nan, q=np.nan, i=np.nan, loading=np.nan)
    # Add results for non convergent contingencies
    failed_trafo3w_results = pd.DataFrame(
        index=pd.MultiIndex.from_product(
            [
                [timestep],
                failed_outages,
                monitored_3_end_branches,
                [BranchSide.ONE.value, BranchSide.TWO.value, BranchSide.THREE.value],
            ],
            names=["timestep", "contingency", "element", "side"],
        )
    ).assign(p=np.nan, q=np.nan, i=np.nan, loading=np.nan)
    converted_branch_results = pd.concat([failed_branch_results, failed_trafo3w_results], axis=0)
    # add empty element_name and contingency_name columns to match the schema
    converted_branch_results["element_name"] = ""
    converted_branch_results["contingency_name"] = ""
    return converted_branch_results

get_failed_node_results

get_failed_node_results(
    timestep, failed_outages, monitored_nodes
)

Get the failed node results.

PARAMETER	DESCRIPTION
timestep	The timestep of the results TYPE: int
failed_outages	The list of failed outages TYPE: list
monitored_nodes	The list of monitored nodes TYPE: list

RETURNS	DESCRIPTION
DataFrame[NodeResultSchema]	The failed node results

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/loadflow_result_helpers.py

@pa.check_types
def get_failed_node_results(
    timestep: int, failed_outages: list[str], monitored_nodes: list[str]
) -> pat.DataFrame[NodeResultSchema]:
    """Get the failed node results.

    Parameters
    ----------
    timestep : int
        The timestep of the results
    failed_outages : list
        The list of failed outages
    monitored_nodes : list
        The list of monitored nodes

    Returns
    -------
    pat.DataFrame[NodeResultSchema]
        The failed node results
    """
    failed_node_results = pd.DataFrame(
        index=pd.MultiIndex.from_product(
            [[timestep], failed_outages, monitored_nodes],
            names=["timestep", "contingency", "element"],
        )
    ).assign(
        vm=np.nan,
        va=np.nan,
        vm_loading=np.nan,
        p=np.nan,
        q=np.nan,
        vm_basecase_deviation=np.nan,
        element_name="",
        contingency_name="",
    )
    # fill in empty columns to match the schema
    failed_node_results["p"] = np.nan
    failed_node_results["q"] = np.nan
    failed_node_results["vm_basecase_deviation"] = np.nan
    failed_node_results["element_name"] = ""
    failed_node_results["contingency_name"] = ""
    return failed_node_results

extract_branch_results

extract_branch_results(
    branch_results,
    timestep,
    contingencies,
    monitored_branches,
    basecase,
)

Extract the branch results for a specific timestep.

PARAMETER	DESCRIPTION
branch_results	The branch results dataframe to extract the branch results from. TYPE: DataFrame[BranchResultSchema]
timestep	The selected timestep to pull from the loadflow results. TYPE: int
basecase	The basecase contingency id to use for the N-0 results. TYPE: str
contingencies	The list of contingencies to extract the results for. TYPE: list[str]
monitored_branches	The list of monitored branches to extract the results for. buses switches etc should not be included here, only branches. TYPE: list[GridElement]

RETURNS	DESCRIPTION
Float[ndarray, ' n_contingencies n_branches_monitored']	The branch results with the following: - shape (n_contingencies, n_branches_monitored) - only the p values of the monitored branches at the from-end - For three winding transformers, the p values are split into three rows for each side (hv, mv, lv).

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/loadflow_result_helpers.py

def extract_branch_results(
    branch_results: pat.DataFrame[BranchResultSchema],
    timestep: int,
    contingencies: list[str],
    monitored_branches: list[GridElement],
    basecase: str,
) -> tuple[Float[np.ndarray, " n_branches_monitored"], Float[np.ndarray, " n_contingencies n_branches_monitored"]]:
    """Extract the branch results for a specific timestep.

    Parameters
    ----------
    branch_results: BranchResultSchema,
        The branch results dataframe to extract the branch results from.
    timestep : int
        The selected timestep to pull from the loadflow results.
    basecase : str
        The basecase contingency id to use for the N-0 results.
    contingencies : list[str]
        The list of contingencies to extract the results for.
    monitored_branches : list[GridElement]
        The list of monitored branches to extract the results for.
        buses switches etc should not be included here, only branches.

    Returns
    -------
    Float[np.ndarray, " n_contingencies n_branches_monitored"]
        The branch results with the following:
        - shape (n_contingencies, n_branches_monitored)
        - only the p values of the monitored branches at the from-end
        - For three winding transformers, the p values are split into three rows for each side (hv, mv, lv).
    """
    assert basecase not in contingencies, "Basecase contingency should not be in the list of N-k contingencies"
    n_monitored_branches = len(monitored_branches)
    n_contingencies = len(contingencies)
    if (n_monitored_branches == 0) or (n_contingencies == 0 and basecase is None):
        # If there are no monitored branches, return empty arrays
        return np.full(n_monitored_branches, dtype=float), np.full((n_contingencies, n_monitored_branches), dtype=float)
    # Get the branch results for the given job_id and timestep
    three_winding_side_dict = {
        "trafo3w_hv": [BranchSide.ONE.value],
        "trafo3w_mv": [BranchSide.TWO.value],
        "trafo3w_lv": [BranchSide.THREE.value],
    }
    all_cases = [basecase, *contingencies]
    normal_branches_ids = [elem.id for elem in monitored_branches if elem.type not in three_winding_side_dict]
    normal_branch_multi_idx = [*product([timestep], all_cases, normal_branches_ids, [BranchSide.ONE.value])]
    trafo3w_idx = []
    for trafo_type, sides in three_winding_side_dict.items():
        three_winding_branches = [element.id for element in monitored_branches if element.type == trafo_type]
        multi_idx = product([timestep], all_cases, three_winding_branches, sides)
        trafo3w_idx.extend(multi_idx)

    multi_index = pd.MultiIndex.from_tuples(
        [*normal_branch_multi_idx, *trafo3w_idx],
        names=["timestep", "contingency", "element", "side"],
    )

    # Drop timestep and side, since we do not need them anymore
    p_results = branch_results.reindex(multi_index, fill_value=0.0).droplevel(["side", "timestep"])["p"]

    # bring into correct order
    monitored_branches_order = [elem.id for elem in monitored_branches]

    n_0_vector = p_results.fillna(0.0).loc[basecase, monitored_branches_order].values

    n_1_index = pd.MultiIndex.from_product([contingencies, monitored_branches_order], names=["contingency", "element"])
    n_1_results = p_results.reindex(n_1_index, fill_value=0.0)
    n_1_array = n_1_results.fillna(0.0).values.reshape(len(contingencies), len(monitored_branches_order))
    return n_0_vector, n_1_array

extract_node_matrices

extract_node_matrices(
    node_results,
    timestep,
    contingencies,
    monitored_nodes,
    basecase="BASECASE",
)

Extract the node results for a specific timestep.

PARAMETER	DESCRIPTION
node_results	The node results dataframe to extract the node results from. TYPE: DataFrame[NodeResultSchema]
timestep	The selected timestep to pull from the loadflow results. TYPE: int
basecase	The basecase contingency id to use for the N-0 results. TYPE: str DEFAULT: 'BASECASE'
contingencies	The list of contingencies to extract the results for. TYPE: list[str]
monitored_nodes	The list of monitored nodes to extract the results for. buses switches etc should not be included here, only nodes. TYPE: list[GridElement]

RETURNS	DESCRIPTION
vm_n0	The voltage magnitude results for the basecase contingency at the monitored nodes. TYPE: Float[ndarray, ' n_nodes_monitored']
va_n0	The voltage angle results for the basecase contingency at the monitored nodes. TYPE: Float[ndarray, ' n_nodes_monitored']
vm_n1	The voltage magnitude results for the contingencies at the monitored nodes. TYPE: Float[ndarray, ' n_contingencies n_nodes_monitored']
va_n1	The voltage angle results for the contingencies at the monitored nodes. TYPE: Float[ndarray, ' n_contingencies n_nodes_monitored']

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/loadflow_result_helpers.py

def extract_node_matrices(
    node_results: pat.DataFrame[NodeResultSchema],
    timestep: int,
    contingencies: list[str],
    monitored_nodes: list[GridElement],
    basecase: str = "BASECASE",
) -> tuple[
    Float[np.ndarray, " n_nodes_monitored"],
    Float[np.ndarray, "  n_nodes_monitored"],
    Float[np.ndarray, " n_contingencies n_nodes_monitored"],
    Float[np.ndarray, " n_contingencies n_nodes_monitored"],
]:
    """Extract the node results for a specific timestep.

    Parameters
    ----------
    node_results: NodeResultSchema,
        The node results dataframe to extract the node results from.
    timestep : int
        The selected timestep to pull from the loadflow results.
    basecase : str
        The basecase contingency id to use for the N-0 results.
    contingencies : list[str]
        The list of contingencies to extract the results for.
    monitored_nodes : list[GridElement]
        The list of monitored nodes to extract the results for.
        buses switches etc should not be included here, only nodes.

    Returns
    -------
    vm_n0 : Float[np.ndarray, " n_nodes_monitored"]
        The voltage magnitude results for the basecase contingency at the monitored nodes.
    va_n0 : Float[np.ndarray, " n_nodes_monitored"]
        The voltage angle results for the basecase contingency at the monitored nodes.
    vm_n1 : Float[np.ndarray, " n_contingencies n_nodes_monitored"]
        The voltage magnitude results for the contingencies at the monitored nodes.
    va_n1 : Float[np.ndarray, " n_contingencies n_nodes_monitored"]
        The voltage angle results for the contingencies at the monitored nodes.
    """
    assert basecase not in contingencies, "Basecase contingency should not be in the list of N-k contingencies"
    n_contingencies = len(contingencies)
    n_monitored_nodes = len(monitored_nodes)
    if (n_monitored_nodes == 0) or (n_contingencies == 0 and basecase is None):
        # If there are no monitored nodes, return empty arrays
        return (
            np.full(n_monitored_nodes, dtype=float),
            np.full(n_monitored_nodes, dtype=float),
            np.full((n_contingencies, n_monitored_nodes), dtype=float),
            np.full((n_contingencies, n_monitored_nodes), dtype=float),
        )

    # Get the node results for the given job_id and timestep
    node_results = node_results.xs(timestep, level="timestep")
    # Reindex to ensure all contingencies and monitored nodes are present, fill missing with 0
    product_index = pd.MultiIndex.from_product(
        [[basecase, *contingencies], [elem.id for elem in monitored_nodes]],
        names=["contingency", "element"],
    )
    node_results = node_results.reindex(product_index, fill_value=np.nan)
    vm_n0 = node_results.loc[basecase, "vm"].values
    va_n0 = node_results.loc[basecase, "va"].values

    vm_n1 = node_results.loc[contingencies, :]["vm"].values
    va_n1 = node_results.loc[contingencies, :]["va"].values
    # reshape the results to have the contingencies as first dimension
    vm_n1 = vm_n1.reshape(len(contingencies), len(monitored_nodes))
    va_n1 = va_n1.reshape(len(contingencies), len(monitored_nodes))
    return vm_n0, va_n0, vm_n1, va_n1

extract_solver_matrices

extract_solver_matrices(
    loadflow_results, nminus1_definition, timestep
)

Extract the N-0 and N-1 matrices in a similar format to the DC solver.

PARAMETER	DESCRIPTION
loadflow_results	The loadflow results to extract the matrices from. TYPE: LoadflowResults
nminus1_definition	The N-1 definition to use for the contingencies and monitored elements. TYPE: Nminus1Definition
timestep	The selected timestep to pull from the loadflow results. TYPE: int

RETURNS	DESCRIPTION
Float[ndarray, ' n_branches_monitored']	The N-0 matrix
Float[ndarray, ' n_solver_contingencies n_branches_monitored']	The N-1 matrix
Bool[ndarray, ' n_solver_contingencies']	The convergence status of the contingencies in the N-1 matrix True if converged or not calculated, False if not converged.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/loadflow_result_helpers.py

def extract_solver_matrices(
    loadflow_results: LoadflowResults,
    nminus1_definition: Nminus1Definition,
    timestep: int,
) -> tuple[
    Float[np.ndarray, " n_branches_monitored"],
    Float[np.ndarray, " n_solver_contingencies n_branches_monitored"],
    Bool[np.ndarray, " n_solver_contingencies"],
]:
    """Extract the N-0 and N-1 matrices in a similar format to the DC solver.

    Parameters
    ----------
    loadflow_results : LoadflowResults
        The loadflow results to extract the matrices from.
    nminus1_definition : Nminus1Definition
        The N-1 definition to use for the contingencies and monitored elements.
    timestep : int
        The selected timestep to pull from the loadflow results.

    Returns
    -------
    Float[np.ndarray, " n_branches_monitored"]
        The N-0 matrix
    Float[np.ndarray, " n_solver_contingencies n_branches_monitored"]
        The N-1 matrix
    Bool[np.ndarray, " n_solver_contingencies"]
        The convergence status of the contingencies in the N-1 matrix
        True if converged or not calculated, False if not converged.
    """
    basecase = next((cont for cont in nminus1_definition.contingencies if cont.is_basecase()), None)
    assert basecase is not None, "No basecase contingency found in the N-1 definition."
    contingency_order = [cont.id for cont in nminus1_definition.contingencies if not cont.is_basecase()]

    success = (
        (
            loadflow_results.converged["status"]
            .loc[timestep]
            .isin([ConvergenceStatus.CONVERGED.value, ConvergenceStatus.NO_CALCULATION.value])
        )
        .reindex(contingency_order, fill_value=False)
        .values
    )
    branch_elements = [elem for elem in nminus1_definition.monitored_elements if elem.kind == "branch"]
    n_0_vector, n1_matrix = extract_branch_results(
        branch_results=loadflow_results.branch_results,
        timestep=timestep,
        contingencies=contingency_order,
        monitored_branches=branch_elements,
        basecase=basecase.id,
    )

    return n_0_vector, n1_matrix, success

select_timestep

select_timestep(loadflow_results, timestep)

Select a specific timestep from the loadflow results.

PARAMETER	DESCRIPTION
loadflow_results	The loadflow results to select the timestep from. TYPE: LoadflowResults
timestep	The timestep to select. TYPE: Integral

RETURNS	DESCRIPTION
LoadflowResults	The loadflow results for the selected timestep.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/loadflow_result_helpers.py

def select_timestep(loadflow_results: LoadflowResults, timestep: Integral) -> LoadflowResults:
    """Select a specific timestep from the loadflow results.

    Parameters
    ----------
    loadflow_results : LoadflowResults
        The loadflow results to select the timestep from.
    timestep : Integral
        The timestep to select.

    Returns
    -------
    LoadflowResults
        The loadflow results for the selected timestep.
    """

    def safe_xs(df: pd.DataFrame | None) -> pd.DataFrame | None:
        """Safely select a timestep from a DataFrame."""
        if df is None:
            return None
        try:
            return df.xs(timestep, level="timestep", drop_level=False)
        except KeyError:
            return df.iloc[0:0]

    return LoadflowResults(
        job_id=loadflow_results.job_id,
        warnings=loadflow_results.warnings,
        branch_results=safe_xs(loadflow_results.branch_results),
        node_results=safe_xs(loadflow_results.node_results),
        regulating_element_results=safe_xs(loadflow_results.regulating_element_results),
        converged=safe_xs(loadflow_results.converged),
        va_diff_results=safe_xs(loadflow_results.va_diff_results),
        cascade_results=safe_xs(loadflow_results.cascade_results),
    )

convert_polars_loadflow_results_to_pandas

convert_polars_loadflow_results_to_pandas(
    loadflow_results_polars,
)

Convert the LoadflowResultsPolars class to LoadflowResults class.

PARAMETER	DESCRIPTION
loadflow_results_polars	The loadflow results in polars format. TYPE: LoadflowResultsPolars

RETURNS	DESCRIPTION
LoadflowResults	The loadflow results in pandas format.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/loadflow_result_helpers.py

def convert_polars_loadflow_results_to_pandas(
    loadflow_results_polars: LoadflowResultsPolars,
) -> LoadflowResults:
    """Convert the LoadflowResultsPolars class to LoadflowResults class.

    Parameters
    ----------
    loadflow_results_polars : LoadflowResultsPolars
        The loadflow results in polars format.

    Returns
    -------
    LoadflowResults
        The loadflow results in pandas format.
    """

    def polars_to_pandas(df: Optional[Union[pl.DataFrame, pl.LazyFrame]], index_cols: list[str]) -> Optional[pd.DataFrame]:
        """Convert a polars DataFrame or LazyFrame to an indexed pandas DataFrame.

        *index_cols* is the schema's index for this result type (only those present in the
        frame are used). Note ``side`` is an index level for branch results but a plain
        column for switch results, so the index differs per result type.
        """
        if df is None:
            return None
        if hasattr(df, "collect"):
            df = df.collect()
        pdf = df.to_pandas()
        present = [col for col in index_cols if col in pdf.columns]
        if present:
            pdf = pdf.set_index(present)
        return pdf

    element_index = ["timestep", "contingency", "element"]
    # These are optional on both models, and not every producer fills them (powsybl leaves
    # the switch/SpPS/connectivity frames unset). Pass them only when present: the pandas
    # schema field validates the value it is given, and validating an explicit None fails.
    optional = {
        "connectivity_result": polars_to_pandas(loadflow_results_polars.connectivity_result, ["contingency", "element"]),
        "switch_results": polars_to_pandas(loadflow_results_polars.switch_results, element_index),
        "spps_results": polars_to_pandas(loadflow_results_polars.spps_results, ["timestep", "contingency"]),
    }
    optional = {field: value for field, value in optional.items() if value is not None}

    return LoadflowResults(
        job_id=loadflow_results_polars.job_id,
        branch_results=polars_to_pandas(loadflow_results_polars.branch_results, [*element_index, "side"]),
        node_results=polars_to_pandas(loadflow_results_polars.node_results, element_index),
        regulating_element_results=polars_to_pandas(loadflow_results_polars.regulating_element_results, element_index),
        converged=polars_to_pandas(loadflow_results_polars.converged, ["timestep", "contingency"]),
        va_diff_results=polars_to_pandas(loadflow_results_polars.va_diff_results, element_index),
        cascade_results=polars_to_pandas(
            loadflow_results_polars.cascade_results,
            ["timestep", "contingency", "cascade_number", "element_mrid"],
        ),
        warnings=loadflow_results_polars.warnings,
        **optional,
    )

pandas_to_polars

pandas_to_polars(df, lazy, *, nan_to_null=False)

Convert a pandas DataFrame to a polars DataFrame.

PARAMETER	DESCRIPTION
df	The pandas DataFrame to convert. TYPE: Optional[DataFrame]
lazy	Whether to return a LazyFrame or a DataFrame. TYPE: bool
nan_to_null	Whether NaN values should be converted to polars null values. TYPE: bool DEFAULT: False

RETURNS	DESCRIPTION
Optional[DataFrame]	The polars DataFrame or None if the input was None.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/loadflow_result_helpers.py

def pandas_to_polars(
    df: Optional[pd.DataFrame],
    lazy: bool,
    *,
    nan_to_null: bool = False,
) -> Optional[pl.DataFrame | pl.LazyFrame]:
    """Convert a pandas DataFrame to a polars DataFrame.

    Parameters
    ----------
    df : Optional[pd.DataFrame]
        The pandas DataFrame to convert.
    lazy : bool
        Whether to return a LazyFrame or a DataFrame.
    nan_to_null : bool
        Whether NaN values should be converted to polars null values.

    Returns
    -------
    Optional[pl.DataFrame]
        The polars DataFrame or None if the input was None.
    """
    if df is None:
        return None
    if isinstance(df, pd.DataFrame):
        df = pl.from_pandas(df, include_index=True, nan_to_null=nan_to_null)
        for column in ["element_name", "contingency_name", "warnings"]:
            if column in df.columns:
                df = df.with_columns(pl.col(column).cast(pl.String))
    if lazy:
        df = df.lazy()  # Assume it's a pandas DataFrame
    return df  # Assume it's already a polars DataFrame

cascade_pandas_to_polars

cascade_pandas_to_polars(df, lazy)

Convert cascade results to polars with stable nullable dtypes.

PARAMETER	DESCRIPTION
df	Cascade results in pandas format. TYPE: Optional[DataFrame]
lazy	Whether to return a LazyFrame or a DataFrame. TYPE: bool

RETURNS	DESCRIPTION
Optional[DataFrame | LazyFrame]	Cascade results converted to polars, or None if the input was None.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/loadflow_result_helpers.py

def cascade_pandas_to_polars(df: Optional[pd.DataFrame], lazy: bool) -> Optional[pl.DataFrame | pl.LazyFrame]:
    """Convert cascade results to polars with stable nullable dtypes.

    Parameters
    ----------
    df : Optional[pd.DataFrame]
        Cascade results in pandas format.
    lazy : bool
        Whether to return a LazyFrame or a DataFrame.

    Returns
    -------
    Optional[pl.DataFrame | pl.LazyFrame]
        Cascade results converted to polars, or None if the input was None.
    """
    if df is None:
        return None
    if isinstance(df, pd.DataFrame):
        df = df.copy()
        for column in ["loading", "r_ohm", "x_ohm"]:
            if column in df.columns:
                df[column] = pd.to_numeric(df[column], errors="coerce")
    return pandas_to_polars(df, lazy=lazy, nan_to_null=True)

convert_pandas_loadflow_results_to_polars

convert_pandas_loadflow_results_to_polars(loadflow_results)

Convert the LoadflowResults class to LoadflowResultsPolars class.

PARAMETER	DESCRIPTION
loadflow_results	The loadflow results in pandas format. TYPE: LoadflowResults

RETURNS	DESCRIPTION
LoadflowResultsPolars	The loadflow results in polars format.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/loadflow_result_helpers.py

def convert_pandas_loadflow_results_to_polars(loadflow_results: LoadflowResults) -> LoadflowResultsPolars:
    """Convert the LoadflowResults class to LoadflowResultsPolars class.

    Parameters
    ----------
    loadflow_results : LoadflowResults
        The loadflow results in pandas format.

    Returns
    -------
    LoadflowResultsPolars
        The loadflow results in polars format.
    """
    return LoadflowResultsPolars(
        job_id=loadflow_results.job_id,
        branch_results=pandas_to_polars(loadflow_results.branch_results, lazy=True),
        node_results=pandas_to_polars(loadflow_results.node_results, lazy=True),
        regulating_element_results=pandas_to_polars(loadflow_results.regulating_element_results, lazy=True),
        converged=pandas_to_polars(loadflow_results.converged, lazy=True),
        va_diff_results=pandas_to_polars(loadflow_results.va_diff_results, lazy=True),
        switch_results=pandas_to_polars(loadflow_results.switch_results, lazy=True),
        connectivity_result=pandas_to_polars(loadflow_results.connectivity_result, lazy=True),
        cascade_results=cascade_pandas_to_polars(loadflow_results.cascade_results, lazy=True),
        spps_results=pandas_to_polars(loadflow_results.spps_results, lazy=True),
        warnings=loadflow_results.warnings,
        lazy=True,
    )

toop_engine_interfaces.nminus1_definition

The N-1 definition holds monitored and outaged elements for a grid.

This information is not present in the grid models and hence needs to be stored separately to run an N-1 computation. The order of the outages should be the same as in the jax code, where it's hardcoded to the following: - branch outages - multi outage - non-relevant injection outages - relevant injection outages

POWSYBL_SUPPORTED_ID_TYPES module-attribute

POWSYBL_SUPPORTED_ID_TYPES = Literal[
    "powsybl", "cgmes", "ucte"
]

PANDAPOWER_SUPPORTED_ID_TYPES module-attribute

PANDAPOWER_SUPPORTED_ID_TYPES = Literal[
    "unique_pandapower", "cgmes"
]

ELEMENT_ID_TYPES module-attribute

ELEMENT_ID_TYPES = Literal[
    PANDAPOWER_SUPPORTED_ID_TYPES,
    POWSYBL_SUPPORTED_ID_TYPES,
]

GridElement

Bases: BaseModel

A grid element is identified by its id(powsybl) or its id and type (pandapower)

id instance-attribute

id

The id of the element. For powsybl grids this is the global string id, for pandapower this is the integer index into the dataframe

name class-attribute instance-attribute

name = ''

The name of the element. This is optional, but can be used to provide a more human-readable name for the element.

type instance-attribute

type

For pandapower, we need to further specify a type which corresponds to the table pandapower stores the information in. Valid tables are 'line', 'trafo', 'ext_grid', 'gen', 'load', 'shunt', ... For powsybl, this is not strictly needed to identify the element however it makes it easier. In that case, type will be something like TIE_LINE, LINE, TWO_WINDING_TRANSFORMER, GENERATOR, etc.

kind instance-attribute

kind

The kind of the element. Usually these are handled differently in the grid modelling software, so it can make assembling an N-1 analysis easier if it is known if the element is a branch, bus or injection. This could be inferred from the type, however for conveniece it is stored separately.

For the bus type there is some potential confusion in powsybl. In pandapower, this always refers to the net.bus df. In powsybl in a bus/branch model, there are no busbar sections in powsybl, i.e. net.get_node_breaker_topology does not deliver busbar sections. Meaning, the "bus" type refers to the net.get_bus_breaker_topology buses if it's a bus/breaker topology bus. If it's a node/breaker topology, then "bus" refers to the busbar section.

SwitchMonitoringScope

Bases: str, Enum

A single monitoring aspect for a switch element.

Set monitoring_scope on a :class:MonitoredElement with kind="switch" to restrict monitoring to one aspect. None (the default) activates all three.

• FLOW: active power (p), reactive power (q), and current (i) through the switch.
• ANGLE: voltage-angle difference (va) and voltage magnitude (vm) across the open switch; used to assess reclosing conditions.
• PROTECTION: the switch carries a protection relay and participates in cascade-tripping.

FLOW class-attribute instance-attribute

FLOW = 'flow'

ANGLE class-attribute instance-attribute

ANGLE = 'angle'

PROTECTION class-attribute instance-attribute

PROTECTION = 'protection'

MonitoredElement

Bases: GridElement

A grid element that is observed during N-1 computation.

Extends :class:GridElement with an optional monitoring_scope that restricts which aspects are computed for switch elements.

For kind="switch", None activates all three aspects (flow, angle, protection). Passing a frozenset restricts monitoring to those aspects only. For other kinds this field must remain None.

monitoring_scope class-attribute instance-attribute

monitoring_scope = None

Which aspects of this switch to monitor. None enables all defaults.

For kind="switch": - None → monitor flow, angle, and protection (the extractor stores frozenset(SwitchMonitoringScope) in the DataFrame column). - A frozenset of :class:SwitchMonitoringScope values → monitor only those aspects. For other kinds this field must be None.

id instance-attribute

id

The id of the element. For powsybl grids this is the global string id, for pandapower this is the integer index into the dataframe

name class-attribute instance-attribute

name = ''

The name of the element. This is optional, but can be used to provide a more human-readable name for the element.

type instance-attribute

type

For pandapower, we need to further specify a type which corresponds to the table pandapower stores the information in. Valid tables are 'line', 'trafo', 'ext_grid', 'gen', 'load', 'shunt', ... For powsybl, this is not strictly needed to identify the element however it makes it easier. In that case, type will be something like TIE_LINE, LINE, TWO_WINDING_TRANSFORMER, GENERATOR, etc.

kind instance-attribute

kind

The kind of the element. Usually these are handled differently in the grid modelling software, so it can make assembling an N-1 analysis easier if it is known if the element is a branch, bus or injection. This could be inferred from the type, however for conveniece it is stored separately.

For the bus type there is some potential confusion in powsybl. In pandapower, this always refers to the net.bus df. In powsybl in a bus/branch model, there are no busbar sections in powsybl, i.e. net.get_node_breaker_topology does not deliver busbar sections. Meaning, the "bus" type refers to the net.get_bus_breaker_topology buses if it's a bus/breaker topology bus. If it's a node/breaker topology, then "bus" refers to the busbar section.

Contingency

Bases: BaseModel

A single N-1 case

elements instance-attribute

elements

The grid elements that are to be outaged under this contingency. Usually, this will be exactly one element however exceptional contingencies and multi-outages might include more than one element.

id instance-attribute

id

The id of the contingency. This is used to identify the contingency in the results. It should be unique across all contingencies in the N-1 definition.

name class-attribute instance-attribute

name = ''

The name of the contingency. This is optional, but can be used to provide a more human-readable name. This will show up in the Loadflowresult-tables as column contingency_name.

is_multi_outage

is_multi_outage()

Check if the contingency is a multi-outage.

A multi-outage is defined as a contingency that has more than one element in it.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/nminus1_definition.py

def is_multi_outage(self) -> bool:
    """Check if the contingency is a multi-outage.

    A multi-outage is defined as a contingency that has more than one element in it.
    """
    return len(self.elements) > 1

is_basecase

is_basecase()

Check if the contingency is the N-0 base case.

A base case is defined as a contingency that has no elements in it.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/nminus1_definition.py

def is_basecase(self) -> bool:
    """Check if the contingency is the N-0 base case.

    A base case is defined as a contingency that has no elements in it.
    """
    return len(self.elements) == 0

is_single_outage

is_single_outage()

Check if the contingency is a normal single-element outage.

A single outage is defined as a contingency that has exactly one element in it.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/nminus1_definition.py

def is_single_outage(self) -> bool:
    """Check if the contingency is a normal single-element outage.

    A single outage is defined as a contingency that has exactly one element in it.
    """
    return len(self.elements) == 1

Condition

Bases: BaseModel

Represents a single condition in a rule.

condition_type instance-attribute

condition_type

Type of condition to evaluate.

condition_check_type class-attribute instance-attribute

condition_check_type = None

Comparison operator or special check.

condition_side class-attribute instance-attribute

condition_side = None

Element side or aggregation mode.

condition_limit_value class-attribute instance-attribute

condition_limit_value = None

Threshold value for numeric checks.

condition_element_unique_id instance-attribute

condition_element_unique_id

Globally unique identifier of the condition element.

condition_mode class-attribute instance-attribute

condition_mode = SppsConditionMode.CON

Network state against which the condition is evaluated.

• BC — base-case results (before the contingency is applied).
• CON — post-contingency results (default).

Action

Bases: BaseModel

Represents a single action in a rule.

measure_element_unique_id instance-attribute

measure_element_unique_id

Globally unique identifier of the element to apply the action to.

measure_type instance-attribute

measure_type

Type of action.

measure_value instance-attribute

measure_value

Target value (numeric or 'Open'/'Closed').

SppsRule

Bases: BaseModel

Represents a full scheme with multiple conditions and actions.

scheme_name instance-attribute

scheme_name

Unique scheme identifier.

condition_logic class-attribute instance-attribute

condition_logic = SppsConditionLogic.ALL

Whether every condition must pass SppsConditionLogic.ALL or at least one SppsConditionLogic.ANY for the scheme to activate.

conditions instance-attribute

conditions

Condition rows for this scheme; combined according to condition_logic.

actions instance-attribute

actions

Actions applied when the rule is activated.

Nminus1Definition

Bases: BaseModel

An N-1 definition holds monitored and outaged elements for a grid.

For powsybl, ids are unique across types (i.e. a branch and an injection can not have the same id), however in pandapower, ids are not unique and we have to store the type alongside with them.

monitored_elements instance-attribute

monitored_elements

A list of monitored elements that should be observed during the N-1 computation.

contingencies instance-attribute

contingencies

A list of contingencies that should be computed during the N-1 computation.

spps_rules class-attribute instance-attribute

spps_rules = None

The spps ruleset to use for the N-1 computation. This is optional. If not provided, spps ruleset will not be used.

id_type class-attribute instance-attribute

id_type = None

The type of the ids used in the N-1 definition. This is used to determine how to interpret the ids in the monitored elements and contingencies. See ELEMENT_ID_TYPES for more information. If none, pandapower will try to use the globally unique ids, and powsybl will use the global string ids.

base_case property

base_case

Get the base case contingency, which is the contingency with no elements in it.

__getitem__

__getitem__(key)

Get a subset of the nminus1definition based on the contingencies.

If a string is given, the contingency id must be in the contingencies list. If an integer or slice is given, the case id will be indexed by the integer or slice.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/nminus1_definition.py

def __getitem__(self, key: str | int | slice) -> "Nminus1Definition":
    """Get a subset of the nminus1definition based on the contingencies.

    If a string is given, the contingency id must be in the contingencies list.
    If an integer or slice is given, the case id will be indexed by the integer or slice.
    """
    if isinstance(key, str):
        contingency_ids = [contingency.id for contingency in self.contingencies]
        if key not in contingency_ids:
            raise KeyError(f"Contingency id {key} not in contingencies.")
        index = contingency_ids.index(key)
        index = slice(index, index + 1)
    elif isinstance(key, int):
        index = slice(key, key + 1)
    elif isinstance(key, slice):
        index = key
    else:
        raise TypeError("Key must be a string, int or slice.")

    # pylint: disable=unsubscriptable-object
    return Nminus1Definition(
        monitored_elements=self.monitored_elements,
        contingencies=self.contingencies[index],
    )

load_nminus1_definition_fs

load_nminus1_definition_fs(filesystem, file_path)

Load an N-1 definition from a file system.

PARAMETER	DESCRIPTION
filesystem	The file system to use to load the N-1 definition. TYPE: AbstractFileSystem
file_path	The path to the file containing the N-1 definition in json format. TYPE: Union[str, Path]

RETURNS	DESCRIPTION
Nminus1Definition	The loaded N-1 definition.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/nminus1_definition.py

def load_nminus1_definition_fs(
    filesystem: AbstractFileSystem,
    file_path: Union[str, Path],
) -> Nminus1Definition:
    """Load an N-1 definition from a file system.

    Parameters
    ----------
    filesystem : AbstractFileSystem
        The file system to use to load the N-1 definition.
    file_path : Union[str, Path]
        The path to the file containing the N-1 definition in json format.

    Returns
    -------
    Nminus1Definition
        The loaded N-1 definition.
    """
    return load_pydantic_model_fs(
        filesystem=filesystem,
        file_path=file_path,
        model_class=Nminus1Definition,
    )

load_nminus1_definition

load_nminus1_definition(filename)

Load an N-1 definition from a json file

PARAMETER	DESCRIPTION
filename	The path to the json file containing the N-1 definition. TYPE: Path

RETURNS	DESCRIPTION
Nminus1Definition	The loaded N-1 definition.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/nminus1_definition.py

def load_nminus1_definition(filename: Path) -> Nminus1Definition:
    """Load an N-1 definition from a json file

    Parameters
    ----------
    filename : Path
        The path to the json file containing the N-1 definition.

    Returns
    -------
    Nminus1Definition
        The loaded N-1 definition.
    """
    return load_nminus1_definition_fs(
        filesystem=LocalFileSystem(),
        file_path=filename,
    )

save_nminus1_definition

save_nminus1_definition(filename, nminus1_definition)

Save an N-1 definition to a json file

PARAMETER	DESCRIPTION
filename	The path to the json file to save the N-1 definition to. TYPE: Path
nminus1_definition	The N-1 definition to save. TYPE: Nminus1Definition

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/nminus1_definition.py

def save_nminus1_definition(filename: Path, nminus1_definition: Nminus1Definition) -> None:
    """Save an N-1 definition to a json file

    Parameters
    ----------
    filename : Path
        The path to the json file to save the N-1 definition to.
    nminus1_definition : Nminus1Definition
        The N-1 definition to save.
    """
    save_pydantic_model_fs(filesystem=LocalFileSystem(), file_path=filename, pydantic_model=nminus1_definition)

Asset Topology

toop_engine_interfaces.asset_topology

Public asset topology API.

This package re-exports the main topology data models so downstream imports and documentation can refer to a stable package-level namespace.

RealizedBusGroup module-attribute

RealizedBusGroup = RealizedStation

__all__ module-attribute

__all__ = [
    "AppliedStation",
    "AssetBay",
    "AssetSetpoint",
    "BusGroupAssetConnection",
    "Busbar",
    "BusbarCoupler",
    "CircuitGroup",
    "MasterAssetTopology",
    "MasterBusGroup",
    "RealizedBusGroup",
    "RealizedStation",
    "RealizedTopology",
    "RuntimeAssetConnection",
    "RuntimeAssetTopology",
    "RuntimeBranchAsset",
    "RuntimeBusGroup",
    "RuntimeBusbar",
    "RuntimeBusbarCoupler",
    "RuntimeInjectionAsset",
    "RuntimeSwitchableAsset",
    "SimplifiedAssetTopology",
    "SimplifiedBusGroup",
    "SwitchableAsset",
    "get_asset_bay_ids_for_bus_group_asset",
    "get_asset_bays_for_bus_group_asset",
    "validate_runtime_bus_group_asset_references",
]

AppliedStation

Bases: RealizedStation

Backward-compatible alias for RealizedStation.

bus_group instance-attribute

bus_group

Runtime bus-group snapshot after applying the action.

coupler_diff instance-attribute

coupler_diff

A list of couplers that have been switched.

branch_reassignment_diff instance-attribute

branch_reassignment_diff

Branch reassignments as (branch_index, busbar_index, connected) tuples.

injection_reassignment_diff instance-attribute

injection_reassignment_diff

Injection reassignments as (injection_index, busbar_index, connected) tuples.

branch_disconnection_diff instance-attribute

branch_disconnection_diff

Branch indices that were disconnected.

injection_disconnection_diff instance-attribute

injection_disconnection_diff

Injection indices that were disconnected.

RealizedStation

Bases: BaseModel

Legacy station-level diff wrapper around one runtime bus group.

The wrapped bus_group contains the full runtime snapshot after applying an action, while the diff fields isolate the switching changes relative to the reference state.

bus_group instance-attribute

bus_group

Runtime bus-group snapshot after applying the action.

coupler_diff instance-attribute

coupler_diff

A list of couplers that have been switched.

branch_reassignment_diff instance-attribute

branch_reassignment_diff

Branch reassignments as (branch_index, busbar_index, connected) tuples.

injection_reassignment_diff instance-attribute

injection_reassignment_diff

Injection reassignments as (injection_index, busbar_index, connected) tuples.

branch_disconnection_diff instance-attribute

branch_disconnection_diff

Branch indices that were disconnected.

injection_disconnection_diff instance-attribute

injection_disconnection_diff

Injection indices that were disconnected.

RealizedTopology

Bases: BaseModel

A realized topology, including the new topology and the changes made to the original topology.

DeprecationWarning: This model is deprecated and should be replaced by RuntimeAssetTopology plus explicit switch updates in new code.

This is similar to RealizedStation but holds information for all stations in the topology. The diffs are include a station identifier that shows which station in the topology was affected by the diff.

master_data class-attribute instance-attribute

master_data = None

Canonical master data associated with the realized runtime stations when available.

bus_groups class-attribute instance-attribute

bus_groups = Field(default_factory=list)

The realized asset stations that were directly applied or compared.

coupler_diff instance-attribute

coupler_diff

A list of couplers that have been switched. Each tuple contains the station id and the coupler that was switched.

branch_reassignment_diff instance-attribute

branch_reassignment_diff

Branch reassignments as (station_id, branch_index, busbar_index, connected) tuples.

injection_reassignment_diff instance-attribute

injection_reassignment_diff

Injection reassignments as (station_id, injection_index, busbar_index, connected) tuples.

branch_disconnection_diff instance-attribute

branch_disconnection_diff

Branch disconnections as (station_id, branch_index) tuples.

injection_disconnection_diff instance-attribute

injection_disconnection_diff

Injection disconnections as (station_id, injection_index) tuples.

BusGroupAssetConnection

Bases: BaseModel

Bus-group-local association between a switching-table column and a topology asset.

asset_id instance-attribute

asset_id

Grid model id of the topology-owned asset referenced by this bus-group-local column.

branch_end class-attribute instance-attribute

branch_end = None

Optional branch-end metadata for this bus-group-local asset occurrence.

asset_bay_id class-attribute instance-attribute

asset_bay_id = None

Optional topology-scoped asset bay identifier for this bus-group-local asset occurrence.

CircuitGroup

Bases: BaseModel

A circuit group represents assets connected without power switches.

All assets inside the same circuit group are treated as jointly outaged. In master data this is a structural grouping only: it records which assets share a no-breaker path in the station design. Runtime open or closed disconnector states do not mutate this list; callers must combine the structural group with runtime switch states when reasoning about the current electrical reachability.

asset_ids instance-attribute

asset_ids

Grid-model ids of the assets contained in the circuit group.

asset_bay_ids instance-attribute

asset_bay_ids

Asset-bay ids whose switches implement the circuit-group outage effect.

The ids identify the relevant bays structurally. Their live switch state is stored only on runtime station snapshots.

MasterAssetTopology

Bases: BaseModel

Canonical grid master data without runtime switching or outage state.

topology_id instance-attribute

topology_id

The unique identifier of the topology master data.

grid_model_file class-attribute instance-attribute

grid_model_file = None

The source grid model file the master data was derived from.

name class-attribute instance-attribute

name = None

The name of the topology master data.

bus_groups instance-attribute

bus_groups

Canonical bus groups with asset references and physical connectivity only.

circuit_groups class-attribute instance-attribute

circuit_groups = None

Topology-owned structural circuit groups.

These groups are derived from the master bus-group layout and are not updated when runtime disconnectors open or close. Runtime callers must therefore interpret them together with the current bus-group switch state.

branch_assets class-attribute instance-attribute

branch_assets = Field(default_factory=list)

The canonical branch master data payloads.

injection_assets class-attribute instance-attribute

injection_assets = Field(default_factory=list)

The canonical injection master data payloads.

asset_bays class-attribute instance-attribute

asset_bays = Field(default_factory=list)

The canonical asset-bay payloads.

asset_setpoints class-attribute instance-attribute

asset_setpoints = None

Optional topology-owned setpoint payloads.

check_bus_group_ids_unique classmethod

check_bus_group_ids_unique(v)

Validate uniqueness of canonical bus-group identifiers.

PARAMETER	DESCRIPTION
v	Canonical bus groups assigned to the topology master data. TYPE: list[MasterBusGroup]

RETURNS	DESCRIPTION
list[MasterBusGroup]	Validated canonical stations.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/asset_topology/asset_topology.py

@field_validator("bus_groups")
@classmethod
def check_bus_group_ids_unique(cls, v: list[MasterBusGroup]) -> list[MasterBusGroup]:
    """Validate uniqueness of canonical bus-group identifiers.

    Parameters
    ----------
    v : list[MasterBusGroup]
        Canonical bus groups assigned to the topology master data.

    Returns
    -------
    list[MasterBusGroup]
        Validated canonical stations.
    """
    station_ids = [station.bus_group_id for station in v]
    if len(station_ids) != len(set(station_ids)):
        raise ValueError("bus_group_id must be unique for topology master data stations")
    return v

check_branch_asset_ids_unique classmethod

check_branch_asset_ids_unique(v)

Validate uniqueness of canonical branch asset ids.

PARAMETER	DESCRIPTION
v	Canonical branch assets. TYPE: list[BranchAsset]

RETURNS	DESCRIPTION
list[BranchAsset]	Validated branch assets.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/asset_topology/asset_topology.py

@field_validator("branch_assets")
@classmethod
def check_branch_asset_ids_unique(cls, v: list[BranchAsset]) -> list[BranchAsset]:
    """Validate uniqueness of canonical branch asset ids.

    Parameters
    ----------
    v : list[BranchAsset]
        Canonical branch assets.

    Returns
    -------
    list[BranchAsset]
        Validated branch assets.
    """
    asset_ids = [asset.grid_model_id for asset in v]
    if len(asset_ids) != len(set(asset_ids)):
        raise ValueError("grid_model_id must be unique for topology branch assets")
    return v

check_injection_asset_ids_unique classmethod

check_injection_asset_ids_unique(v)

Validate uniqueness of canonical injection asset ids.

PARAMETER	DESCRIPTION
v	Canonical injection assets. TYPE: list[InjectionAsset]

RETURNS	DESCRIPTION
list[InjectionAsset]	Validated injection assets.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/asset_topology/asset_topology.py

@field_validator("injection_assets")
@classmethod
def check_injection_asset_ids_unique(cls, v: list[InjectionAsset]) -> list[InjectionAsset]:
    """Validate uniqueness of canonical injection asset ids.

    Parameters
    ----------
    v : list[InjectionAsset]
        Canonical injection assets.

    Returns
    -------
    list[InjectionAsset]
        Validated injection assets.
    """
    asset_ids = [asset.grid_model_id for asset in v]
    if len(asset_ids) != len(set(asset_ids)):
        raise ValueError("grid_model_id must be unique for topology injection assets")
    return v

check_asset_bay_ids_unique classmethod

check_asset_bay_ids_unique(v)

Validate uniqueness and presence of canonical asset-bay ids.

PARAMETER	DESCRIPTION
v	Canonical asset-bay payloads. TYPE: list[AssetBay]

RETURNS	DESCRIPTION
list[AssetBay]	Validated asset-bay payloads.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/asset_topology/asset_topology.py

@field_validator("asset_bays")
@classmethod
def check_asset_bay_ids_unique(cls, v: list[AssetBay]) -> list[AssetBay]:
    """Validate uniqueness and presence of canonical asset-bay ids.

    Parameters
    ----------
    v : list[AssetBay]
        Canonical asset-bay payloads.

    Returns
    -------
    list[AssetBay]
        Validated asset-bay payloads.
    """
    asset_bay_ids = [asset_bay.asset_bay_id for asset_bay in v]
    if any(asset_bay_id is None for asset_bay_id in asset_bay_ids):
        raise ValueError("All topology asset bays must define asset_bay_id")
    if len(asset_bay_ids) != len(set(asset_bay_ids)):
        raise ValueError("asset_bay_id must be unique for topology asset bays")
    return v

check_station_asset_references

check_station_asset_references()

Validate station asset references against canonical topology collections.

RETURNS	DESCRIPTION
MasterAssetTopology	Validated topology master data.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/asset_topology/asset_topology.py

@model_validator(mode="after")
def check_station_asset_references(self: "MasterAssetTopology") -> "MasterAssetTopology":
    """Validate station asset references against canonical topology collections.

    Returns
    -------
    MasterAssetTopology
        Validated topology master data.
    """
    _validate_station_asset_references(
        (
            (station.bus_group_id, "branch", asset_connection.asset_id, asset_connection.asset_bay_id)
            for station in self.bus_groups
            for asset_connection in station.branch_connections
        ),
        self.branch_assets,
        self.injection_assets,
        self.asset_bays,
    )
    _validate_station_asset_references(
        (
            (station.bus_group_id, "injection", asset_connection.asset_id, asset_connection.asset_bay_id)
            for station in self.bus_groups
            for asset_connection in station.injection_connections
        ),
        self.branch_assets,
        self.injection_assets,
        self.asset_bays,
    )

    return self

MasterBusGroup

Bases: BaseModel

Canonical bus-group master data without runtime switching state.

model_config class-attribute instance-attribute

model_config = ConfigDict(arbitrary_types_allowed=True)

bus_group_id instance-attribute

bus_group_id

The unique identifier of the canonical bus-group view.

This is the stable identifier used to align master data, runtime snapshots, and action sets. Deterministic suffixes such as _a or _b distinguish multiple structural bus groups that originate from the same physical substation.

voltage_level_id class-attribute instance-attribute

voltage_level_id = None

The voltage level identifier backing this canonical bus-group view.

name class-attribute instance-attribute

name = None

The name of the station.

station_type class-attribute instance-attribute

station_type = None

The type of the station.

region class-attribute instance-attribute

region = None

The region of the station.

voltage_level class-attribute instance-attribute

voltage_level = None

The voltage level of the station in kV.

busbars instance-attribute

busbars

Canonical busbars owned by the bus group.

Runtime outage state is stripped; all busbars are assumed in service in this model.

couplers instance-attribute

couplers

Canonical couplers owned by the bus group.

Runtime switch state is stripped; all couplers are assumed closed and in service.

branch_connections class-attribute instance-attribute

branch_connections = Field(default_factory=list)

Bus-group-local canonical branch references aligned with branch_connectivity.

injection_connections class-attribute instance-attribute

injection_connections = Field(default_factory=list)

Bus-group-local canonical injection references aligned with injection_connectivity.

branch_connectivity class-attribute instance-attribute

branch_connectivity = None

Physically possible branch-to-busbar assignments for the bus group.

injection_connectivity class-attribute instance-attribute

injection_connectivity = None

Physically possible injection-to-busbar assignments for the bus group.

model_copy

model_copy(*, update=None, deep=False)

Copy and revalidate the bus group.

PARAMETER	DESCRIPTION
update	Field updates to merge into the copied bus group. TYPE: Optional[dict[str, Any]] DEFAULT: None
deep	Whether to deep-copy nested structures before validation. TYPE: bool DEFAULT: False

RETURNS	DESCRIPTION
MasterBusGroup	Copied and revalidated bus-group instance.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/asset_topology/asset_topology.py

def model_copy(self, *, update: Optional[dict[str, Any]] = None, deep: bool = False) -> "MasterBusGroup":
    """Copy and revalidate the bus group.

    Parameters
    ----------
    update : Optional[dict[str, Any]], optional
        Field updates to merge into the copied bus group.
    deep : bool, default=False
        Whether to deep-copy nested structures before validation.

    Returns
    -------
    MasterBusGroup
        Copied and revalidated bus-group instance.
    """
    payload = merged_round_trip_payload(self, update, deep=deep)
    return type(self).model_validate(payload)

normalize_connectivity_tables classmethod

normalize_connectivity_tables(v)

Normalize connectivity table inputs to boolean arrays.

PARAMETER	DESCRIPTION
v	Raw connectivity table input. TYPE: object | None

RETURNS	DESCRIPTION
Optional[ndarray]	Boolean connectivity table or None.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/asset_topology/asset_topology.py

@field_validator("branch_connectivity", "injection_connectivity", mode="before")
@classmethod
def normalize_connectivity_tables(cls, v: object | None) -> Optional[np.ndarray]:
    """Normalize connectivity table inputs to boolean arrays.

    Parameters
    ----------
    v : object | None
        Raw connectivity table input.

    Returns
    -------
    Optional[np.ndarray]
        Boolean connectivity table or ``None``.
    """
    if v is None:
        return None
    return np.asarray(v, dtype=bool)

check_busbar_int_ids_unique classmethod

check_busbar_int_ids_unique(v)

Validate that bus-group busbar integer ids are unique.

PARAMETER	DESCRIPTION
v	Busbars assigned to the bus group. TYPE: list[Busbar]

RETURNS	DESCRIPTION
list[Busbar]	Validated busbars.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/asset_topology/asset_topology.py

@field_validator("busbars")
@classmethod
def check_busbar_int_ids_unique(cls, v: list[Busbar]) -> list[Busbar]:
    """Validate that bus-group busbar integer ids are unique.

    Parameters
    ----------
    v : list[Busbar]
        Busbars assigned to the bus group.

    Returns
    -------
    list[Busbar]
        Validated busbars.
    """
    int_ids = [busbar.int_id for busbar in v]
    if len(int_ids) != len(set(int_ids)):
        raise ValueError("busbar int_ids must be unique per station")
    return v

check_asset_reference_alignment

check_asset_reference_alignment()

Validate connectivity matrices against canonical asset references.

RETURNS	DESCRIPTION
MasterBusGroup	Validated bus-group instance.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/asset_topology/asset_topology.py

@model_validator(mode="after")
def check_asset_reference_alignment(self: "MasterBusGroup") -> "MasterBusGroup":
    """Validate connectivity matrices against canonical asset references.

    Returns
    -------
    MasterBusGroup
        Validated bus-group instance.
    """
    _validate_master_bus_group_connectivity(
        station_grid_model_id=self.bus_group_id,
        station_name=self.name,
        busbar_count=len(self.busbars),
        asset_count=len(self.branch_connections),
        asset_connectivity=self.branch_connectivity,
        asset_kind="branch",
    )
    _validate_master_bus_group_connectivity(
        station_grid_model_id=self.bus_group_id,
        station_name=self.name,
        busbar_count=len(self.busbars),
        asset_count=len(self.injection_connections),
        asset_connectivity=self.injection_connectivity,
        asset_kind="injection",
    )
    return self

AssetBay

Bases: BaseModel

Saves the physical connection from the asset to the substation busbars - a bay (Schaltfeld).

A line usually has three switches, before it is connected to the busbar. Two disconnector switches and one circuit breaker switch. A transformer usually has two switches, before it is connected to the busbar. One disconnector switch and one circuit breaker switch.

type: n - node type: b - busbar (Sammelschiene) type: CB - Breaker / Power Switch (Leistungsschalter) type: DS - Disconnector Switch (Trennschalter)

------------------ busbar 1 - type: b | / type: DS - Busbar disconnector busbar 1 -> used for reassigning the asset to another busbar | ------|----------- busbar 2 - type: b | | / | type: DS - Busbar disconnector busbar 2 -> used for reassigning the asset to another busbar | | --------- bus_3 - type: n - busbar section bus | / type: CB - Breaker / Power Switch -> used for disconnecting the asset from the busbar | --------- bus_2 - type: n - circuit breaker bus | / type: DS - Asset disconnector (optional) -> not used by the asset | --------- bus_1 - type: n - asset bus ^ | Line/Transformer

asset_bay_id instance-attribute

asset_bay_id

Topology-scoped identifier for the asset bay.

asset_disconnector_grid_model_id class-attribute instance-attribute

asset_disconnector_grid_model_id = None

The id of the switch, which connects the asset to the circuit breaker node. This switch is a disconnector switch. Do not use for anything, leave state as found. Default should be closed.

breaker_grid_model_id instance-attribute

breaker_grid_model_id

This switch is a circuit breaker / power switch. Use for disconnecting / reconnecting the asset from the busbar.

busbar_disconnector_grid_model_id instance-attribute

busbar_disconnector_grid_model_id

The ids of the switches, which assign the asset to the busbars. key: busbar_grid_model_id e.g. 4%%bus value: busbar_disconnector_grid_model_id This switch is a disconnector switch. Use for reassigning the asset to another busbar. Only one switch should be closed at a time.

Master data stores only the switch ids, not their live open or closed state. Runtime materialization interprets the current switch states to reconstruct the active station assignment.

check_is_empty classmethod

check_is_empty(v)

Check if the dict is empty.

PARAMETER	DESCRIPTION
v	The dictionary of busbar_disconnector_grid_model_id to check. TYPE: dict[str, str]

RETURNS	DESCRIPTION
dict[str, str]	The dictionary itself.

RAISES	DESCRIPTION
ValueError	If the dictionary is empty.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/asset_topology/assets.py

@field_validator("busbar_disconnector_grid_model_id")
@classmethod
def check_is_empty(cls, v: dict[str, str]) -> dict[str, str]:
    """Check if the dict is empty.

    Parameters
    ----------
    v : dict[str, str]
        The dictionary of busbar_disconnector_grid_model_id to check.

    Returns
    -------
    dict[str, str]
        The dictionary itself.

    Raises
    ------
    ValueError
        If the dictionary is empty.
    """
    if len(v) == 0:
        raise ValueError("busbar_disconnector_grid_model_id must not be empty")
    return v

AssetSetpoint

Bases: BaseModel

Asset data describing a single asset with a setpoint.

This could for example be a PST or HVDC setpoint. Note: The same asset can both be switchable and have a setpoint. In this case, the asset will be represented twice.

grid_model_id instance-attribute

grid_model_id

The unique identifier of the asset. Corresponds to the asset's id in the grid model.

asset_type class-attribute instance-attribute

asset_type = None

The type of the asset, might be useful for finding the asset later on

name class-attribute instance-attribute

name = None

The name of the asset, might be useful for finding the asset later on

setpoint instance-attribute

setpoint

The setpoint of the asset.

Busbar

Bases: BaseModel

Canonical busbar data describing a physical busbar in a bus group.

model_config class-attribute instance-attribute

model_config = ConfigDict(extra='forbid')

grid_model_id instance-attribute

grid_model_id

The unique identifier of the busbar. Corresponds to the busbar's id in the grid model.

busbar_type class-attribute instance-attribute

busbar_type = None

The type of the busbar, might be useful for finding the busbar later on

name class-attribute instance-attribute

name = None

The name of the busbar, might be useful for finding the busbar later on

int_id instance-attribute

int_id

Is used to reference busbars in the couplers. Needs to be unique per bus group

bus_breaker_bus_id class-attribute instance-attribute

bus_breaker_bus_id = None

Physical bus-breaker bus id backing this busbar section.

This field is primarily populated by node-breaker backends such as Powsybl. It identifies the bus in the source bus-breaker topology that the physical busbar section belongs to. Unlike bus_branch_bus_id this is not meant to reflect runtime regrouping after switching operations.

BusbarCoupler

Bases: BaseModel

Canonical coupler data describing a physical busbar coupler at a station.

This references only busbar couplers, i.e. couplers connecting two busbars. Switches connecting assets to a busbar are represented in the asset_switching_table in the station model.

Note: A busbar couple is a physical connection between two busbars, this can be also a cross coupler. To further specify the connection of an asset to a busbar, the asset connection

model_config class-attribute instance-attribute

model_config = ConfigDict(extra='forbid')

grid_model_id instance-attribute

grid_model_id

The unique identifier of the coupler. Corresponds to the coupler's id in the grid model.

coupler_type class-attribute instance-attribute

coupler_type = None

The type of the coupler, might be useful for finding the coupler later on

name class-attribute instance-attribute

name = None

The name of the coupler, might be useful for finding the coupler later on

asset_bay class-attribute instance-attribute

asset_bay = None

The asset bay (Schaltfeld) of the coupler. Note: A coupler can have multiple from and to busbars. The asset bay busbar_disconnector_grid_model_id is used save the selector switches of the coupler. Note: A coupler has never an asset_disconnector_grid_model_id. Central coupler-path switches are stored on the coupler bay via coupler_breaker_ids and coupler_disconnector_ids.

Multiple selector-switch endpoints may exist in runtime data. Current action generation still assumes one active selector per side and does not synthesize switching plans for multiple simultaneously closed selector paths on one coupler side.

coupler_bay class-attribute instance-attribute

coupler_bay = None

Side-aware coupler bay metadata used to reconstruct runtime endpoints.

SwitchableAsset

Bases: BaseModel

Canonical asset data describing a single switchable asset.

An asset can be for instance a transformer, line, generator, load, shunt.

Note: An asset that is out-of-service can be represented, but its switching entries will be ignored.

model_config class-attribute instance-attribute

model_config = ConfigDict(extra='forbid')

grid_model_id instance-attribute

grid_model_id

The unique identifier of the asset. Corresponds to the asset's id in the grid model.

asset_type class-attribute instance-attribute

asset_type = None

The type of the asset. These refer loosely to the types in the pandapower/powsybl grid models. If set, this can be used to disambiguate branches from injections

name class-attribute instance-attribute

name = None

The name of the asset, might be useful for finding the asset later on

RuntimeBranchAsset

Bases: BranchAsset, RuntimeSwitchableAsset

Runtime switchable asset representing a branch-type element.

model_config class-attribute instance-attribute

model_config = ConfigDict(extra='forbid')

grid_model_id instance-attribute

grid_model_id

The unique identifier of the asset. Corresponds to the asset's id in the grid model.

asset_type class-attribute instance-attribute

asset_type = None

name class-attribute instance-attribute

name = None

The name of the asset, might be useful for finding the asset later on

in_service class-attribute instance-attribute

in_service = True

Whether the asset is in service in the current runtime state.

RuntimeBusbar

Bases: Busbar

Runtime busbar with live service and electrical-bus state.

model_config class-attribute instance-attribute

model_config = ConfigDict(extra='forbid')

grid_model_id instance-attribute

grid_model_id

The unique identifier of the busbar. Corresponds to the busbar's id in the grid model.

busbar_type class-attribute instance-attribute

busbar_type = None

The type of the busbar, might be useful for finding the busbar later on

name class-attribute instance-attribute

name = None

The name of the busbar, might be useful for finding the busbar later on

int_id instance-attribute

int_id

Is used to reference busbars in the couplers. Needs to be unique per bus group

bus_breaker_bus_id class-attribute instance-attribute

bus_breaker_bus_id = None

Physical bus-breaker bus id backing this busbar section.

This field is primarily populated by node-breaker backends such as Powsybl. It identifies the bus in the source bus-breaker topology that the physical busbar section belongs to. Unlike bus_branch_bus_id this is not meant to reflect runtime regrouping after switching operations.

in_service class-attribute instance-attribute

in_service = True

Whether the busbar is in service in the current runtime state.

bus_branch_bus_id class-attribute instance-attribute

bus_branch_bus_id = None

Runtime-only electrical bus id for this physical busbar.

RuntimeBusbarCoupler

Bases: BusbarCoupler

Runtime coupler with live open and service state.

model_config class-attribute instance-attribute

model_config = ConfigDict(extra='forbid')

grid_model_id instance-attribute

grid_model_id

The unique identifier of the coupler. Corresponds to the coupler's id in the grid model.

coupler_type class-attribute instance-attribute

coupler_type = None

The type of the coupler, might be useful for finding the coupler later on

name class-attribute instance-attribute

name = None

The name of the coupler, might be useful for finding the coupler later on

asset_bay class-attribute instance-attribute

asset_bay = None

The asset bay (Schaltfeld) of the coupler. Note: A coupler can have multiple from and to busbars. The asset bay busbar_disconnector_grid_model_id is used save the selector switches of the coupler. Note: A coupler has never an asset_disconnector_grid_model_id. Central coupler-path switches are stored on the coupler bay via coupler_breaker_ids and coupler_disconnector_ids.

Multiple selector-switch endpoints may exist in runtime data. Current action generation still assumes one active selector per side and does not synthesize switching plans for multiple simultaneously closed selector paths on one coupler side.

coupler_bay class-attribute instance-attribute

coupler_bay = None

Side-aware coupler bay metadata used to reconstruct runtime endpoints.

busbar_from_id instance-attribute

busbar_from_id

Resolved runtime busbar int id on the from side.

busbar_to_id instance-attribute

busbar_to_id

Resolved runtime busbar int id on the to side.

open class-attribute instance-attribute

open = False

Whether the coupler is open in the current runtime state.

in_service class-attribute instance-attribute

in_service = True

Whether the coupler is in service in the current runtime state.

RuntimeInjectionAsset

Bases: InjectionAsset, RuntimeSwitchableAsset

Runtime switchable asset representing an injection-type element.

model_config class-attribute instance-attribute

model_config = ConfigDict(extra='forbid')

grid_model_id instance-attribute

grid_model_id

The unique identifier of the asset. Corresponds to the asset's id in the grid model.

asset_type class-attribute instance-attribute

asset_type = None

name class-attribute instance-attribute

name = None

The name of the asset, might be useful for finding the asset later on

in_service class-attribute instance-attribute

in_service = True

Whether the asset is in service in the current runtime state.

RuntimeSwitchableAsset

Bases: SwitchableAsset

Runtime switchable asset with live service state.

model_config class-attribute instance-attribute

model_config = ConfigDict(extra='forbid')

grid_model_id instance-attribute

grid_model_id

The unique identifier of the asset. Corresponds to the asset's id in the grid model.

asset_type class-attribute instance-attribute

asset_type = None

The type of the asset. These refer loosely to the types in the pandapower/powsybl grid models. If set, this can be used to disambiguate branches from injections

name class-attribute instance-attribute

name = None

The name of the asset, might be useful for finding the asset later on

in_service class-attribute instance-attribute

in_service = True

Whether the asset is in service in the current runtime state.

RuntimeAssetConnection

Bases: BaseModel

Busgroup-local association between a switching-table column and a materialized asset payload.

ATTRIBUTE	DESCRIPTION
asset	Bus-group-local runtime asset payload aligned with one switching-table column. TYPE: RuntimeSwitchableAsset
branch_end	Optional canonical branch-end metadata for the bus-group-local occurrence. TYPE: Optional[BranchEnd]
asset_bay	Optional bus-group-local asset-bay payload describing the physical switch path. TYPE: Optional[AssetBay]

asset instance-attribute

asset

Bus-group-local asset payload aligned with one switching-table column.

branch_end class-attribute instance-attribute

branch_end = None

Optional branch-end metadata for this bus-group-local asset occurrence.

asset_bay class-attribute instance-attribute

asset_bay = None

Optional bus-group-local asset bay payload for this bus-group-local asset occurrence.

get_busbar_disconnector

get_busbar_disconnector()

Return the selector-switch mapping of the asset bay, if available.

RETURNS	DESCRIPTION
Optional[dict[str, str]]	Mapping from busbar id to selector-switch id, or None when the bus-group-local asset has no asset-bay payload.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/asset_topology/runtime_topology.py

def get_busbar_disconnector(self) -> Optional[dict[str, str]]:
    """Return the selector-switch mapping of the asset bay, if available.

    Returns
    -------
    Optional[dict[str, str]]
        Mapping from busbar id to selector-switch id, or ``None`` when the
        bus-group-local asset has no asset-bay payload.
    """
    if self.asset_bay is not None:
        return self.asset_bay.busbar_disconnector_grid_model_id
    return None

RuntimeAssetTopology

Bases: BaseModel

Runtime topology payload grouped independently from canonical master data.

The wrapper carries runtime station snapshots and optional runtime-visible circuit-group metadata aligned with the same topology view. It is the runtime companion of MasterAssetTopology and intentionally carries no topology-owned canonical asset collections itself.

bus_groups instance-attribute

bus_groups

Runtime station snapshots for the topology view.

circuit_groups class-attribute instance-attribute

circuit_groups = None

Optional circuit-group metadata carried alongside the runtime stations.

check_bus_group_ids_unique classmethod

check_bus_group_ids_unique(v)

Validate uniqueness of runtime bus-group identifiers.

PARAMETER	DESCRIPTION
v	Runtime bus groups assigned to the wrapper. TYPE: list[RuntimeBusGroup]

RETURNS	DESCRIPTION
list[RuntimeBusGroup]	Validated runtime stations.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/asset_topology/runtime_topology.py

@field_validator("bus_groups")
@classmethod
def check_bus_group_ids_unique(cls, v: list[RuntimeBusGroup]) -> list[RuntimeBusGroup]:
    """Validate uniqueness of runtime bus-group identifiers.

    Parameters
    ----------
    v : list[RuntimeBusGroup]
        Runtime bus groups assigned to the wrapper.

    Returns
    -------
    list[RuntimeBusGroup]
        Validated runtime stations.
    """
    station_ids = [station.bus_group_id for station in v]
    if len(station_ids) != len(set(station_ids)):
        raise ValueError("bus_group_id must be unique for runtime topology stations")
    return v

RuntimeBusGroup

Bases: BaseModel

Bus-group data describing a single materialized bus group.

The bus-group identity refers to a bus-group or station-view identifier. A physical substation or voltage level may contain multiple bus-branch model bus ids. The bus-group assets are aligned with the switching tables and describe the assets visible in that bus-group view; they are not intended to define a topology-owned canonical asset list.

model_config class-attribute instance-attribute

model_config = ConfigDict(arbitrary_types_allowed=True)

bus_group_id instance-attribute

bus_group_id

The unique identifier of the bus-group or station view.

This is a bus-group-view identifier and may be synthetic. Runtime electrical bus ids are tracked separately on the busbars.

Included are all assets, busbars and couplers that are connectable via switches. Buses in the same bus group that are connected via branches are excluded in this specific bus.

This means that two bus groups can have the same elements if the underlying station is currently split.

voltage_level_id class-attribute instance-attribute

voltage_level_id = None

Voltage level identifier backing this bus-group view in the source grid.

name class-attribute instance-attribute

name = None

The name of the station.

station_type class-attribute instance-attribute

station_type = None

The type of the station.

region class-attribute instance-attribute

region = None

The region of the station.

voltage_level class-attribute instance-attribute

voltage_level = None

The voltage level of the station.

busbars instance-attribute

busbars

The list of busbars in the bus group.

bus_branch_bus_ids class-attribute instance-attribute

bus_branch_bus_ids = Field(default_factory=list)

Unique non-empty bus-branch bus ids currently represented by this bus-group view.

couplers instance-attribute

couplers

The list of couplers in the bus group.

branch_switching_table instance-attribute

branch_switching_table

Holds the switching of each branch asset to each busbar, shape (n_bus, n_branch_asset).

An entry is true if the asset is connected to the busbar. Note: An asset can be connected to multiple busbars, in which case a closed coupler is assumed to be present between these busbars. Note: An asset can be connected to none of the busbars. In this case, the asset is intentionally disconnected as part of a transmission line switching action. In practice, this usually involves a separate switch from the asset-to-busbar couplers, as each asset usually has a switch that completely disconnects it from the bus group. These switches are not modelled here, a postprocessing routine needs to do the translation to this physical layout. Do not use in_service for intentional disconnections.

injection_switching_table instance-attribute

injection_switching_table

Holds the switching of each injection asset to each busbar, shape (n_bus, n_injection_asset).

branch_connectivity class-attribute instance-attribute

branch_connectivity = None

Holds all physically possible branch layouts, shape (n_bus, n_branch_asset).

injection_connectivity class-attribute instance-attribute

injection_connectivity = None

Holds all physically possible injection layouts, shape (n_bus, n_injection_asset).

model_log class-attribute instance-attribute

model_log = None

Holds log messages from the model creation process.

This can be used to store information about the model creation process, e.g. warnings or errors. A potential use case is to inform the user about data quality issues e.g. missing the Asset Bay switches.

branch_connections class-attribute instance-attribute

branch_connections = Field(default_factory=list)

Bus-group-local branch payloads aligned with branch_switching_table.

injection_connections class-attribute instance-attribute

injection_connections = Field(default_factory=list)

Bus-group-local injection payloads aligned with injection_switching_table.

grid_model_id property

grid_model_id

Backward-compatible alias for the station identifier.

assets property

assets

Return combined station-local assets in legacy column order.

asset_switching_table property

asset_switching_table

Return the combined switching table in legacy branch-then-injection order.

asset_connectivity property

asset_connectivity

Return the combined connectivity matrix in legacy branch-then-injection order.

normalize_station_tables classmethod

normalize_station_tables(v)

Normalize switching and connectivity table inputs to boolean arrays.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/asset_topology/runtime_topology.py

@field_validator(
    "branch_switching_table",
    "injection_switching_table",
    "branch_connectivity",
    "injection_connectivity",
    mode="before",
)
@classmethod
def normalize_station_tables(
    cls,
    v: object | None,
) -> Optional[np.ndarray]:
    """Normalize switching and connectivity table inputs to boolean arrays."""
    if v is None:
        return None
    return np.asarray(v, dtype=bool)

check_int_id_unique classmethod

check_int_id_unique(v)

Check if the int_ids of the busbars are unique.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/asset_topology/runtime_topology.py

@field_validator("busbars")
@classmethod
def check_int_id_unique(cls, v: list[RuntimeBusbar]) -> list[RuntimeBusbar]:
    """Check if the int_ids of the busbars are unique."""
    int_ids = [busbar.int_id for busbar in v]
    if len(int_ids) != len(set(int_ids)):
        raise ValueError("busbar int_ids must be unique per station")
    return v

normalize_runtime_busbars classmethod

normalize_runtime_busbars(v)

Validate bus-group busbars as runtime busbars.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/asset_topology/runtime_topology.py

@field_validator("busbars", mode="before")
@classmethod
def normalize_runtime_busbars(cls, v: object) -> list[RuntimeBusbar]:
    """Validate bus-group busbars as runtime busbars."""
    if not isinstance(v, list):
        return v
    runtime_busbars: list[RuntimeBusbar] = []
    for busbar in v:
        if isinstance(busbar, RuntimeBusbar):
            runtime_busbars.append(busbar)
        elif isinstance(busbar, dict):
            runtime_busbars.append(RuntimeBusbar.model_validate(busbar))
        else:
            runtime_busbars.append(RuntimeBusbar(**busbar.model_dump()))
    return runtime_busbars

normalize_bus_branch_bus_ids classmethod

normalize_bus_branch_bus_ids(v)

Normalize explicit bus-branch bus ids to a unique sorted list.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/asset_topology/runtime_topology.py

@field_validator("bus_branch_bus_ids", mode="before")
@classmethod
def normalize_bus_branch_bus_ids(cls, v: Optional[list[str]]) -> list[str]:
    """Normalize explicit bus-branch bus ids to a unique sorted list."""
    if v is None:
        return []
    return sorted({bus_id for bus_id in v if bus_id not in {None, ""}})

check_coupler_busbars_different classmethod

check_coupler_busbars_different(v)

Check if the couplers do not connect the same busbar on both ends.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/asset_topology/runtime_topology.py

@field_validator("couplers")
@classmethod
def check_coupler_busbars_different(cls, v: list[RuntimeBusbarCoupler]) -> list[RuntimeBusbarCoupler]:
    """Check if the couplers do not connect the same busbar on both ends."""
    for coupler in v:
        if coupler.busbar_from_id == coupler.busbar_to_id:
            raise ValueError(f"Coupler {coupler.grid_model_id} connects the same busbar on both ends")
    return v

normalize_runtime_couplers classmethod

normalize_runtime_couplers(v)

Validate bus-group couplers as runtime couplers.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/asset_topology/runtime_topology.py

@field_validator("couplers", mode="before")
@classmethod
def normalize_runtime_couplers(cls, v: object) -> list[RuntimeBusbarCoupler]:
    """Validate bus-group couplers as runtime couplers."""
    if not isinstance(v, list):
        return v
    runtime_couplers: list[RuntimeBusbarCoupler] = []
    for coupler in v:
        if isinstance(coupler, RuntimeBusbarCoupler):
            runtime_couplers.append(coupler)
        elif isinstance(coupler, dict):
            runtime_couplers.append(RuntimeBusbarCoupler.model_validate(coupler))
        else:
            runtime_couplers.append(RuntimeBusbarCoupler(**coupler.model_dump()))
    return runtime_couplers

check_coupler_busbars_exist

check_coupler_busbars_exist()

Check if all coupler busbar ids exist in busbars.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/asset_topology/runtime_topology.py

@model_validator(mode="after")
def check_coupler_busbars_exist(self: "RuntimeBusGroup") -> "RuntimeBusGroup":
    """Check if all coupler busbar ids exist in busbars."""
    busbar_ids = [busbar.int_id for busbar in self.busbars]
    for coupler in self.couplers:
        if coupler.busbar_from_id not in busbar_ids or coupler.busbar_to_id not in busbar_ids:
            raise ValueError(
                f"Coupler {coupler.grid_model_id} references non-existing busbars"
                f" Station_id: {self.bus_group_id}, Name: {self.name}"
            )
    return self

check_coupler_references

check_coupler_references()

Check if all closed couplers reference in-service busbars.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/asset_topology/runtime_topology.py

@model_validator(mode="after")
def check_coupler_references(self: "RuntimeBusGroup") -> "RuntimeBusGroup":
    """Check if all closed couplers reference in-service busbars."""
    busbar_state_map = {busbar.int_id: busbar.in_service for busbar in self.busbars}
    for coupler in self.couplers:
        if coupler.open or not coupler.in_service:
            continue
        if busbar_state_map[coupler.busbar_from_id] != busbar_state_map[coupler.busbar_to_id]:
            raise ValueError(
                f"Closed coupler {coupler.grid_model_id} connects out-of-service busbar with in-service busbar."
                f" Station_id: {self.bus_group_id}, Name: {self.name}"
            )
    return self

sync_bus_branch_bus_ids

sync_bus_branch_bus_ids()

Store the unique bus-branch bus ids contained in the station busbars.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/asset_topology/runtime_topology.py

@model_validator(mode="after")
def sync_bus_branch_bus_ids(self: "RuntimeBusGroup") -> "RuntimeBusGroup":
    """Store the unique bus-branch bus ids contained in the station busbars."""
    bus_branch_bus_ids = sorted(
        {busbar.bus_branch_bus_id for busbar in self.busbars if busbar.bus_branch_bus_id not in {None, ""}}
    )
    self.bus_branch_bus_ids = bus_branch_bus_ids if bus_branch_bus_ids else self.bus_branch_bus_ids
    return self

check_asset_shapes

check_asset_shapes()

Check if switching-table-aligned station-local assets match the matrix shapes.

RETURNS	DESCRIPTION
RuntimeBusGroup	The validated station instance.

RAISES	DESCRIPTION
ValueError	If switching tables, connectivity tables, or asset-bay busbar references do not align with the station-local structure.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/asset_topology/runtime_topology.py

@model_validator(mode="after")
def check_asset_shapes(self: "RuntimeBusGroup") -> "RuntimeBusGroup":
    """Check if switching-table-aligned station-local assets match the matrix shapes.

    Returns
    -------
    RuntimeBusGroup
        The validated station instance.

    Raises
    ------
    ValueError
        If switching tables, connectivity tables, or asset-bay busbar references
        do not align with the station-local structure.
    """
    _validate_busgroup_switching_tables(
        station_grid_model_id=self.bus_group_id,
        station_name=self.name,
        busbar_count=len(self.busbars),
        asset_count=len(self.branch_connections),
        asset_switching_table=self.branch_switching_table,
        asset_connectivity=self.branch_connectivity,
        asset_kind="branch",
    )
    _validate_busgroup_physical_assignments(
        station_grid_model_id=self.bus_group_id,
        station_name=self.name,
        asset_switching_table=self.branch_switching_table,
        asset_connectivity=self.branch_connectivity,
        asset_kind="branch",
    )
    _validate_busgroup_switching_tables(
        station_grid_model_id=self.bus_group_id,
        station_name=self.name,
        busbar_count=len(self.busbars),
        asset_count=len(self.injection_connections),
        asset_switching_table=self.injection_switching_table,
        asset_connectivity=self.injection_connectivity,
        asset_kind="injection",
    )
    _validate_busgroup_physical_assignments(
        station_grid_model_id=self.bus_group_id,
        station_name=self.name,
        asset_switching_table=self.injection_switching_table,
        asset_connectivity=self.injection_connectivity,
        asset_kind="injection",
    )
    busbar_ids = {busbar.grid_model_id for busbar in self.busbars}
    for asset_connection in [*self.branch_connections, *self.injection_connections]:
        if asset_connection.asset_bay is None:
            continue
        for busbar_id in asset_connection.asset_bay.busbar_disconnector_grid_model_id:
            if busbar_id not in busbar_ids:
                raise ValueError(
                    f"busbar_id {busbar_id} in asset {asset_connection.asset.grid_model_id} does not exist in busbars"
                )
    return self

__eq__

__eq__(other)

Check if two stations are equal.

PARAMETER	DESCRIPTION
other	The other station to compare to. TYPE: object

RETURNS	DESCRIPTION
bool	True if the stations are equal, False otherwise.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/asset_topology/runtime_topology.py

def __eq__(self, other: object) -> bool:
    """Check if two stations are equal.

    Parameters
    ----------
    other : object
        The other station to compare to.

    Returns
    -------
    bool
        True if the stations are equal, False otherwise.
    """
    if not isinstance(other, RuntimeBusGroup):
        return False
    return (
        self.bus_group_id == other.bus_group_id
        and self.voltage_level_id == other.voltage_level_id
        and self.region == other.region
        and self.busbars == other.busbars
        and self.bus_branch_bus_ids == other.bus_branch_bus_ids
        and self.couplers == other.couplers
        and self.branch_connections == other.branch_connections
        and self.injection_connections == other.injection_connections
        and np.array_equal(self.branch_switching_table, other.branch_switching_table)
        and np.array_equal(self.injection_switching_table, other.injection_switching_table)
        and (
            np.array_equal(self.branch_connectivity, other.branch_connectivity)
            if (self.branch_connectivity is not None and other.branch_connectivity is not None)
            else self.branch_connectivity == other.branch_connectivity
        )
        and (
            np.array_equal(self.injection_connectivity, other.injection_connectivity)
            if (self.injection_connectivity is not None and other.injection_connectivity is not None)
            else self.injection_connectivity == other.injection_connectivity
        )
    )

model_copy

model_copy(*, update=None, deep=False)

Copy and revalidate the station.

PARAMETER	DESCRIPTION
update	Field updates to merge into the copied station. TYPE: Optional[dict[str, Any]] DEFAULT: None
deep	Whether to deep-copy nested structures before validation. TYPE: bool DEFAULT: False

RETURNS	DESCRIPTION
RuntimeBusGroup	Copied and revalidated station instance.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/asset_topology/runtime_topology.py

def model_copy(self, *, update: Optional[dict[str, Any]] = None, deep: bool = False) -> "RuntimeBusGroup":
    """Copy and revalidate the station.

    Parameters
    ----------
    update : Optional[dict[str, Any]], optional
        Field updates to merge into the copied station.
    deep : bool, default=False
        Whether to deep-copy nested structures before validation.

    Returns
    -------
    RuntimeBusGroup
        Copied and revalidated station instance.
    """
    if update and ("asset_switching_table" in update or "asset_connectivity" in update):
        normalized_update = dict(update)

        if "asset_switching_table" in normalized_update:
            asset_switching_table = normalized_update.pop("asset_switching_table")
            branch_count = len(self.branch_connections)
            normalized_update["branch_switching_table"] = asset_switching_table[:, :branch_count]
            normalized_update["injection_switching_table"] = asset_switching_table[:, branch_count:]

        if "asset_connectivity" in normalized_update:
            asset_connectivity = normalized_update.pop("asset_connectivity")
            if asset_connectivity is None:
                normalized_update["branch_connectivity"] = None
                normalized_update["injection_connectivity"] = None
            else:
                branch_count = len(self.branch_connections)
                normalized_update["branch_connectivity"] = asset_connectivity[:, :branch_count]
                normalized_update["injection_connectivity"] = asset_connectivity[:, branch_count:]

        update = normalized_update

    payload = merged_round_trip_payload(self, update, deep=deep)
    return type(self).model_validate(payload)

is_split

is_split()

Return whether the station view spans more than one non-empty bus-branch bus id.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/asset_topology/runtime_topology.py

def is_split(self) -> bool:
    """Return whether the station view spans more than one non-empty bus-branch bus id."""
    return len(self.bus_branch_bus_ids) > 1

get_connected_assets

get_connected_assets(
    busbar_index, topology_assets=None, asset_scope="all"
)

Return in-service assets connected to one busbar.

PARAMETER	DESCRIPTION
busbar_index	Row index into the station switching tables. TYPE: int
topology_assets	Ignored for materialized stations because payloads are embedded locally. TYPE: Optional[list[RuntimeSwitchableAsset]] DEFAULT: None
asset_scope	Restrict the lookup to branch or injection connections. TYPE: Literal[all, branch, injection] DEFAULT: 'all'

RETURNS	DESCRIPTION
list[RuntimeSwitchableAsset]	Connected in-service assets for the requested busbar and scope.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/asset_topology/runtime_topology.py

def get_connected_assets(
    self,
    busbar_index: int,
    topology_assets: Optional[list[RuntimeSwitchableAsset]] = None,
    asset_scope: Literal["all", "branch", "injection"] = "all",
) -> list[RuntimeSwitchableAsset]:
    """Return in-service assets connected to one busbar.

    Parameters
    ----------
    busbar_index : int
        Row index into the station switching tables.
    topology_assets : Optional[list[RuntimeSwitchableAsset]]
        Ignored for materialized stations because payloads are embedded locally.
    asset_scope : Literal["all", "branch", "injection"]
        Restrict the lookup to branch or injection connections.

    Returns
    -------
    list[RuntimeSwitchableAsset]
        Connected in-service assets for the requested busbar and scope.
    """
    del topology_assets
    if asset_scope == "branch":
        return [
            asset_connection.asset
            for asset_connection, is_connected in zip(
                self.branch_connections,
                self.branch_switching_table[busbar_index],
                strict=True,
            )
            if is_connected and asset_connection.asset.in_service
        ]
    if asset_scope == "injection":
        return [
            asset_connection.asset
            for asset_connection, is_connected in zip(
                self.injection_connections,
                self.injection_switching_table[busbar_index],
                strict=True,
            )
            if is_connected and asset_connection.asset.in_service
        ]
    return [
        *self.get_connected_assets(busbar_index, asset_scope="branch"),
        *self.get_connected_assets(busbar_index, asset_scope="injection"),
    ]

SimplifiedAssetTopology

Bases: RuntimeAssetTopology

Runtime topology subclass carrying already simplified bus-group views.

circuit_groups class-attribute instance-attribute

circuit_groups = None

Optional circuit-group metadata carried alongside the runtime stations.

bus_groups instance-attribute

bus_groups

Simplified runtime bus-group snapshots for the topology view.

check_bus_group_ids_unique classmethod

check_bus_group_ids_unique(v)

Validate uniqueness of runtime bus-group identifiers.

PARAMETER	DESCRIPTION
v	Runtime bus groups assigned to the wrapper. TYPE: list[RuntimeBusGroup]

RETURNS	DESCRIPTION
list[RuntimeBusGroup]	Validated runtime stations.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/asset_topology/runtime_topology.py

@field_validator("bus_groups")
@classmethod
def check_bus_group_ids_unique(cls, v: list[RuntimeBusGroup]) -> list[RuntimeBusGroup]:
    """Validate uniqueness of runtime bus-group identifiers.

    Parameters
    ----------
    v : list[RuntimeBusGroup]
        Runtime bus groups assigned to the wrapper.

    Returns
    -------
    list[RuntimeBusGroup]
        Validated runtime stations.
    """
    station_ids = [station.bus_group_id for station in v]
    if len(station_ids) != len(set(station_ids)):
        raise ValueError("bus_group_id must be unique for runtime topology stations")
    return v

SimplifiedBusGroup

Bases: RuntimeBusGroup

Runtime bus-group subclass marking groups that passed preprocessing simplification.

model_config class-attribute instance-attribute

model_config = ConfigDict(arbitrary_types_allowed=True)

bus_group_id instance-attribute

bus_group_id

The unique identifier of the bus-group or station view.

This is a bus-group-view identifier and may be synthetic. Runtime electrical bus ids are tracked separately on the busbars.

Included are all assets, busbars and couplers that are connectable via switches. Buses in the same bus group that are connected via branches are excluded in this specific bus.

This means that two bus groups can have the same elements if the underlying station is currently split.

voltage_level_id class-attribute instance-attribute

voltage_level_id = None

Voltage level identifier backing this bus-group view in the source grid.

name class-attribute instance-attribute

name = None

The name of the station.

station_type class-attribute instance-attribute

station_type = None

The type of the station.

region class-attribute instance-attribute

region = None

The region of the station.

voltage_level class-attribute instance-attribute

voltage_level = None

The voltage level of the station.

busbars instance-attribute

busbars

The list of busbars in the bus group.

bus_branch_bus_ids class-attribute instance-attribute

bus_branch_bus_ids = Field(default_factory=list)

Unique non-empty bus-branch bus ids currently represented by this bus-group view.

couplers instance-attribute

couplers

The list of couplers in the bus group.

branch_switching_table instance-attribute

branch_switching_table

Holds the switching of each branch asset to each busbar, shape (n_bus, n_branch_asset).

An entry is true if the asset is connected to the busbar. Note: An asset can be connected to multiple busbars, in which case a closed coupler is assumed to be present between these busbars. Note: An asset can be connected to none of the busbars. In this case, the asset is intentionally disconnected as part of a transmission line switching action. In practice, this usually involves a separate switch from the asset-to-busbar couplers, as each asset usually has a switch that completely disconnects it from the bus group. These switches are not modelled here, a postprocessing routine needs to do the translation to this physical layout. Do not use in_service for intentional disconnections.

injection_switching_table instance-attribute

injection_switching_table

Holds the switching of each injection asset to each busbar, shape (n_bus, n_injection_asset).

branch_connectivity class-attribute instance-attribute

branch_connectivity = None

Holds all physically possible branch layouts, shape (n_bus, n_branch_asset).

injection_connectivity class-attribute instance-attribute

injection_connectivity = None

Holds all physically possible injection layouts, shape (n_bus, n_injection_asset).

model_log class-attribute instance-attribute

model_log = None

Holds log messages from the model creation process.

This can be used to store information about the model creation process, e.g. warnings or errors. A potential use case is to inform the user about data quality issues e.g. missing the Asset Bay switches.

branch_connections class-attribute instance-attribute

branch_connections = Field(default_factory=list)

Bus-group-local branch payloads aligned with branch_switching_table.

injection_connections class-attribute instance-attribute

injection_connections = Field(default_factory=list)

Bus-group-local injection payloads aligned with injection_switching_table.

grid_model_id property

grid_model_id

Backward-compatible alias for the station identifier.

assets property

assets

Return combined station-local assets in legacy column order.

asset_switching_table property

asset_switching_table

Return the combined switching table in legacy branch-then-injection order.

asset_connectivity property

asset_connectivity

Return the combined connectivity matrix in legacy branch-then-injection order.

normalize_station_tables classmethod

normalize_station_tables(v)

Normalize switching and connectivity table inputs to boolean arrays.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/asset_topology/runtime_topology.py

@field_validator(
    "branch_switching_table",
    "injection_switching_table",
    "branch_connectivity",
    "injection_connectivity",
    mode="before",
)
@classmethod
def normalize_station_tables(
    cls,
    v: object | None,
) -> Optional[np.ndarray]:
    """Normalize switching and connectivity table inputs to boolean arrays."""
    if v is None:
        return None
    return np.asarray(v, dtype=bool)

check_int_id_unique classmethod

check_int_id_unique(v)

Check if the int_ids of the busbars are unique.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/asset_topology/runtime_topology.py

@field_validator("busbars")
@classmethod
def check_int_id_unique(cls, v: list[RuntimeBusbar]) -> list[RuntimeBusbar]:
    """Check if the int_ids of the busbars are unique."""
    int_ids = [busbar.int_id for busbar in v]
    if len(int_ids) != len(set(int_ids)):
        raise ValueError("busbar int_ids must be unique per station")
    return v

normalize_runtime_busbars classmethod

normalize_runtime_busbars(v)

Validate bus-group busbars as runtime busbars.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/asset_topology/runtime_topology.py

@field_validator("busbars", mode="before")
@classmethod
def normalize_runtime_busbars(cls, v: object) -> list[RuntimeBusbar]:
    """Validate bus-group busbars as runtime busbars."""
    if not isinstance(v, list):
        return v
    runtime_busbars: list[RuntimeBusbar] = []
    for busbar in v:
        if isinstance(busbar, RuntimeBusbar):
            runtime_busbars.append(busbar)
        elif isinstance(busbar, dict):
            runtime_busbars.append(RuntimeBusbar.model_validate(busbar))
        else:
            runtime_busbars.append(RuntimeBusbar(**busbar.model_dump()))
    return runtime_busbars

normalize_bus_branch_bus_ids classmethod

normalize_bus_branch_bus_ids(v)

Normalize explicit bus-branch bus ids to a unique sorted list.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/asset_topology/runtime_topology.py

@field_validator("bus_branch_bus_ids", mode="before")
@classmethod
def normalize_bus_branch_bus_ids(cls, v: Optional[list[str]]) -> list[str]:
    """Normalize explicit bus-branch bus ids to a unique sorted list."""
    if v is None:
        return []
    return sorted({bus_id for bus_id in v if bus_id not in {None, ""}})

check_coupler_busbars_different classmethod

check_coupler_busbars_different(v)

Check if the couplers do not connect the same busbar on both ends.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/asset_topology/runtime_topology.py

@field_validator("couplers")
@classmethod
def check_coupler_busbars_different(cls, v: list[RuntimeBusbarCoupler]) -> list[RuntimeBusbarCoupler]:
    """Check if the couplers do not connect the same busbar on both ends."""
    for coupler in v:
        if coupler.busbar_from_id == coupler.busbar_to_id:
            raise ValueError(f"Coupler {coupler.grid_model_id} connects the same busbar on both ends")
    return v

normalize_runtime_couplers classmethod

normalize_runtime_couplers(v)

Validate bus-group couplers as runtime couplers.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/asset_topology/runtime_topology.py

@field_validator("couplers", mode="before")
@classmethod
def normalize_runtime_couplers(cls, v: object) -> list[RuntimeBusbarCoupler]:
    """Validate bus-group couplers as runtime couplers."""
    if not isinstance(v, list):
        return v
    runtime_couplers: list[RuntimeBusbarCoupler] = []
    for coupler in v:
        if isinstance(coupler, RuntimeBusbarCoupler):
            runtime_couplers.append(coupler)
        elif isinstance(coupler, dict):
            runtime_couplers.append(RuntimeBusbarCoupler.model_validate(coupler))
        else:
            runtime_couplers.append(RuntimeBusbarCoupler(**coupler.model_dump()))
    return runtime_couplers

check_coupler_busbars_exist

check_coupler_busbars_exist()

Check if all coupler busbar ids exist in busbars.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/asset_topology/runtime_topology.py

@model_validator(mode="after")
def check_coupler_busbars_exist(self: "RuntimeBusGroup") -> "RuntimeBusGroup":
    """Check if all coupler busbar ids exist in busbars."""
    busbar_ids = [busbar.int_id for busbar in self.busbars]
    for coupler in self.couplers:
        if coupler.busbar_from_id not in busbar_ids or coupler.busbar_to_id not in busbar_ids:
            raise ValueError(
                f"Coupler {coupler.grid_model_id} references non-existing busbars"
                f" Station_id: {self.bus_group_id}, Name: {self.name}"
            )
    return self

check_coupler_references

check_coupler_references()

Check if all closed couplers reference in-service busbars.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/asset_topology/runtime_topology.py

@model_validator(mode="after")
def check_coupler_references(self: "RuntimeBusGroup") -> "RuntimeBusGroup":
    """Check if all closed couplers reference in-service busbars."""
    busbar_state_map = {busbar.int_id: busbar.in_service for busbar in self.busbars}
    for coupler in self.couplers:
        if coupler.open or not coupler.in_service:
            continue
        if busbar_state_map[coupler.busbar_from_id] != busbar_state_map[coupler.busbar_to_id]:
            raise ValueError(
                f"Closed coupler {coupler.grid_model_id} connects out-of-service busbar with in-service busbar."
                f" Station_id: {self.bus_group_id}, Name: {self.name}"
            )
    return self

sync_bus_branch_bus_ids

sync_bus_branch_bus_ids()

Store the unique bus-branch bus ids contained in the station busbars.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/asset_topology/runtime_topology.py

@model_validator(mode="after")
def sync_bus_branch_bus_ids(self: "RuntimeBusGroup") -> "RuntimeBusGroup":
    """Store the unique bus-branch bus ids contained in the station busbars."""
    bus_branch_bus_ids = sorted(
        {busbar.bus_branch_bus_id for busbar in self.busbars if busbar.bus_branch_bus_id not in {None, ""}}
    )
    self.bus_branch_bus_ids = bus_branch_bus_ids if bus_branch_bus_ids else self.bus_branch_bus_ids
    return self

check_asset_shapes

check_asset_shapes()

Check if switching-table-aligned station-local assets match the matrix shapes.

RETURNS	DESCRIPTION
RuntimeBusGroup	The validated station instance.

RAISES	DESCRIPTION
ValueError	If switching tables, connectivity tables, or asset-bay busbar references do not align with the station-local structure.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/asset_topology/runtime_topology.py

@model_validator(mode="after")
def check_asset_shapes(self: "RuntimeBusGroup") -> "RuntimeBusGroup":
    """Check if switching-table-aligned station-local assets match the matrix shapes.

    Returns
    -------
    RuntimeBusGroup
        The validated station instance.

    Raises
    ------
    ValueError
        If switching tables, connectivity tables, or asset-bay busbar references
        do not align with the station-local structure.
    """
    _validate_busgroup_switching_tables(
        station_grid_model_id=self.bus_group_id,
        station_name=self.name,
        busbar_count=len(self.busbars),
        asset_count=len(self.branch_connections),
        asset_switching_table=self.branch_switching_table,
        asset_connectivity=self.branch_connectivity,
        asset_kind="branch",
    )
    _validate_busgroup_physical_assignments(
        station_grid_model_id=self.bus_group_id,
        station_name=self.name,
        asset_switching_table=self.branch_switching_table,
        asset_connectivity=self.branch_connectivity,
        asset_kind="branch",
    )
    _validate_busgroup_switching_tables(
        station_grid_model_id=self.bus_group_id,
        station_name=self.name,
        busbar_count=len(self.busbars),
        asset_count=len(self.injection_connections),
        asset_switching_table=self.injection_switching_table,
        asset_connectivity=self.injection_connectivity,
        asset_kind="injection",
    )
    _validate_busgroup_physical_assignments(
        station_grid_model_id=self.bus_group_id,
        station_name=self.name,
        asset_switching_table=self.injection_switching_table,
        asset_connectivity=self.injection_connectivity,
        asset_kind="injection",
    )
    busbar_ids = {busbar.grid_model_id for busbar in self.busbars}
    for asset_connection in [*self.branch_connections, *self.injection_connections]:
        if asset_connection.asset_bay is None:
            continue
        for busbar_id in asset_connection.asset_bay.busbar_disconnector_grid_model_id:
            if busbar_id not in busbar_ids:
                raise ValueError(
                    f"busbar_id {busbar_id} in asset {asset_connection.asset.grid_model_id} does not exist in busbars"
                )
    return self

__eq__

__eq__(other)

Check if two stations are equal.

PARAMETER	DESCRIPTION
other	The other station to compare to. TYPE: object

RETURNS	DESCRIPTION
bool	True if the stations are equal, False otherwise.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/asset_topology/runtime_topology.py

def __eq__(self, other: object) -> bool:
    """Check if two stations are equal.

    Parameters
    ----------
    other : object
        The other station to compare to.

    Returns
    -------
    bool
        True if the stations are equal, False otherwise.
    """
    if not isinstance(other, RuntimeBusGroup):
        return False
    return (
        self.bus_group_id == other.bus_group_id
        and self.voltage_level_id == other.voltage_level_id
        and self.region == other.region
        and self.busbars == other.busbars
        and self.bus_branch_bus_ids == other.bus_branch_bus_ids
        and self.couplers == other.couplers
        and self.branch_connections == other.branch_connections
        and self.injection_connections == other.injection_connections
        and np.array_equal(self.branch_switching_table, other.branch_switching_table)
        and np.array_equal(self.injection_switching_table, other.injection_switching_table)
        and (
            np.array_equal(self.branch_connectivity, other.branch_connectivity)
            if (self.branch_connectivity is not None and other.branch_connectivity is not None)
            else self.branch_connectivity == other.branch_connectivity
        )
        and (
            np.array_equal(self.injection_connectivity, other.injection_connectivity)
            if (self.injection_connectivity is not None and other.injection_connectivity is not None)
            else self.injection_connectivity == other.injection_connectivity
        )
    )

model_copy

model_copy(*, update=None, deep=False)

Copy and revalidate the station.

PARAMETER	DESCRIPTION
update	Field updates to merge into the copied station. TYPE: Optional[dict[str, Any]] DEFAULT: None
deep	Whether to deep-copy nested structures before validation. TYPE: bool DEFAULT: False

RETURNS	DESCRIPTION
RuntimeBusGroup	Copied and revalidated station instance.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/asset_topology/runtime_topology.py

def model_copy(self, *, update: Optional[dict[str, Any]] = None, deep: bool = False) -> "RuntimeBusGroup":
    """Copy and revalidate the station.

    Parameters
    ----------
    update : Optional[dict[str, Any]], optional
        Field updates to merge into the copied station.
    deep : bool, default=False
        Whether to deep-copy nested structures before validation.

    Returns
    -------
    RuntimeBusGroup
        Copied and revalidated station instance.
    """
    if update and ("asset_switching_table" in update or "asset_connectivity" in update):
        normalized_update = dict(update)

        if "asset_switching_table" in normalized_update:
            asset_switching_table = normalized_update.pop("asset_switching_table")
            branch_count = len(self.branch_connections)
            normalized_update["branch_switching_table"] = asset_switching_table[:, :branch_count]
            normalized_update["injection_switching_table"] = asset_switching_table[:, branch_count:]

        if "asset_connectivity" in normalized_update:
            asset_connectivity = normalized_update.pop("asset_connectivity")
            if asset_connectivity is None:
                normalized_update["branch_connectivity"] = None
                normalized_update["injection_connectivity"] = None
            else:
                branch_count = len(self.branch_connections)
                normalized_update["branch_connectivity"] = asset_connectivity[:, :branch_count]
                normalized_update["injection_connectivity"] = asset_connectivity[:, branch_count:]

        update = normalized_update

    payload = merged_round_trip_payload(self, update, deep=deep)
    return type(self).model_validate(payload)

is_split

is_split()

Return whether the station view spans more than one non-empty bus-branch bus id.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/asset_topology/runtime_topology.py

def is_split(self) -> bool:
    """Return whether the station view spans more than one non-empty bus-branch bus id."""
    return len(self.bus_branch_bus_ids) > 1

get_connected_assets

get_connected_assets(
    busbar_index, topology_assets=None, asset_scope="all"
)

Return in-service assets connected to one busbar.

PARAMETER	DESCRIPTION
busbar_index	Row index into the station switching tables. TYPE: int
topology_assets	Ignored for materialized stations because payloads are embedded locally. TYPE: Optional[list[RuntimeSwitchableAsset]] DEFAULT: None
asset_scope	Restrict the lookup to branch or injection connections. TYPE: Literal[all, branch, injection] DEFAULT: 'all'

RETURNS	DESCRIPTION
list[RuntimeSwitchableAsset]	Connected in-service assets for the requested busbar and scope.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/asset_topology/runtime_topology.py

def get_connected_assets(
    self,
    busbar_index: int,
    topology_assets: Optional[list[RuntimeSwitchableAsset]] = None,
    asset_scope: Literal["all", "branch", "injection"] = "all",
) -> list[RuntimeSwitchableAsset]:
    """Return in-service assets connected to one busbar.

    Parameters
    ----------
    busbar_index : int
        Row index into the station switching tables.
    topology_assets : Optional[list[RuntimeSwitchableAsset]]
        Ignored for materialized stations because payloads are embedded locally.
    asset_scope : Literal["all", "branch", "injection"]
        Restrict the lookup to branch or injection connections.

    Returns
    -------
    list[RuntimeSwitchableAsset]
        Connected in-service assets for the requested busbar and scope.
    """
    del topology_assets
    if asset_scope == "branch":
        return [
            asset_connection.asset
            for asset_connection, is_connected in zip(
                self.branch_connections,
                self.branch_switching_table[busbar_index],
                strict=True,
            )
            if is_connected and asset_connection.asset.in_service
        ]
    if asset_scope == "injection":
        return [
            asset_connection.asset
            for asset_connection, is_connected in zip(
                self.injection_connections,
                self.injection_switching_table[busbar_index],
                strict=True,
            )
            if is_connected and asset_connection.asset.in_service
        ]
    return [
        *self.get_connected_assets(busbar_index, asset_scope="branch"),
        *self.get_connected_assets(busbar_index, asset_scope="injection"),
    ]

get_asset_bay_ids_for_bus_group_asset

get_asset_bay_ids_for_bus_group_asset(
    stations, asset_grid_model_id
)

Return ordered unique asset-bay ids for one asset.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/asset_topology/runtime_topology.py

def get_asset_bay_ids_for_bus_group_asset(stations: list[RuntimeBusGroup], asset_grid_model_id: str) -> list[str]:
    """Return ordered unique asset-bay ids for one asset."""
    asset_bay_ids: list[str] = []
    seen_ids: set[str] = set()
    for _, _, asset_id, asset_bay_id in iter_bus_group_asset_references(stations):
        if asset_id != asset_grid_model_id or asset_bay_id is None or asset_bay_id in seen_ids:
            continue
        seen_ids.add(asset_bay_id)
        asset_bay_ids.append(asset_bay_id)
    return asset_bay_ids

get_asset_bays_for_bus_group_asset

get_asset_bays_for_bus_group_asset(
    stations, asset_bays, asset_grid_model_id
)

Return ordered unique asset-bay payloads for one asset.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/asset_topology/runtime_topology.py

def get_asset_bays_for_bus_group_asset(
    stations: list[RuntimeBusGroup],
    asset_bays: list[AssetBay],
    asset_grid_model_id: str,
) -> list[AssetBay]:
    """Return ordered unique asset-bay payloads for one asset."""
    asset_bay_map = {asset_bay.asset_bay_id: asset_bay for asset_bay in asset_bays}
    return [
        asset_bay_map[asset_bay_id] for asset_bay_id in get_asset_bay_ids_for_bus_group_asset(stations, asset_grid_model_id)
    ]

validate_runtime_bus_group_asset_references

validate_runtime_bus_group_asset_references(
    stations, branch_assets, injection_assets, asset_bays
)

Validate runtime station references against canonical topology payloads.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/asset_topology/runtime_topology.py

def validate_runtime_bus_group_asset_references(
    stations: list[RuntimeBusGroup],
    branch_assets: list[BranchAsset],
    injection_assets: list[InjectionAsset],
    asset_bays: list[AssetBay],
) -> None:
    """Validate runtime station references against canonical topology payloads."""
    branch_asset_ids = {asset.grid_model_id for asset in branch_assets}
    injection_asset_ids = {asset.grid_model_id for asset in injection_assets}
    asset_bay_ids = {asset_bay.asset_bay_id for asset_bay in asset_bays}

    allowed_asset_ids = {
        "branch": branch_asset_ids,
        "injection": injection_asset_ids,
    }
    error_prefix = {
        "branch": "Branch",
        "injection": "Injection",
    }

    for station_id, asset_kind, asset_id, asset_bay_id in iter_bus_group_asset_references(stations):
        if asset_id not in allowed_asset_ids[asset_kind]:
            raise ValueError(
                f"{error_prefix[asset_kind]} asset grid_model_id {asset_id} referenced by station "
                f"{station_id} does not exist in topology assets"
            )
        if asset_bay_id is not None and asset_bay_id not in asset_bay_ids:
            raise ValueError(
                f"asset_bay_id {asset_bay_id} referenced by station {station_id} does not exist in topology asset bays"
            )

toop_engine_interfaces.asset_topology.topology_conversion

Conversions between materialized bus groups and canonical master data.

logger module-attribute

logger = structlog.get_logger(__name__)

RuntimeSwitchingState dataclass

RuntimeSwitchingState(
    busbar_bus_branch_bus_ids=None,
    branch_current_bus_ids=None,
    injection_current_bus_ids=None,
    busbar_out_of_service_ids=frozenset(),
    open_coupler_ids=frozenset(),
    out_of_service_coupler_ids=frozenset(),
    open_switch_ids=frozenset(),
)

Compact runtime overlay describing the current live switching state of a bus group.

busbar_bus_branch_bus_ids class-attribute instance-attribute

busbar_bus_branch_bus_ids = None

Mapping from canonical busbar ids to current runtime bus-branch bus ids.

branch_current_bus_ids class-attribute instance-attribute

branch_current_bus_ids = None

Current runtime bus ids for branch connections, aligned with the bus group's canonical branch connections.

injection_current_bus_ids class-attribute instance-attribute

injection_current_bus_ids = None

Current runtime bus ids for injection connections, aligned with the bus group's canonical injection connections.

busbar_out_of_service_ids class-attribute instance-attribute

busbar_out_of_service_ids = frozenset()

Canonical busbar ids that are currently out of service.

open_coupler_ids class-attribute instance-attribute

open_coupler_ids = frozenset()

Canonical coupler ids that are currently open.

out_of_service_coupler_ids class-attribute instance-attribute

out_of_service_coupler_ids = frozenset()

Canonical coupler ids that are currently out of service.

open_switch_ids class-attribute instance-attribute

open_switch_ids = frozenset()

Asset-bay switch ids that are currently open.

validate_complete_master_asset_topology

validate_complete_master_asset_topology(master_data)

Validate that canonical master data already contains productive asset metadata.

PARAMETER	DESCRIPTION
master_data	Canonical master data to validate. TYPE: MasterAssetTopology

RAISES	DESCRIPTION
ValueError	If canonical assets or station-local branch references are missing metadata that productive preprocessing requires.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/asset_topology/topology_conversion.py

def validate_complete_master_asset_topology(master_data: MasterAssetTopology) -> None:
    """Validate that canonical master data already contains productive asset metadata.

    Parameters
    ----------
    master_data : MasterAssetTopology
        Canonical master data to validate.

    Raises
    ------
    ValueError
        If canonical assets or station-local branch references are missing metadata that productive
        preprocessing requires.
    """
    for asset in master_data.branch_assets:
        if asset.name is None:
            raise ValueError(f"Branch asset {asset.grid_model_id} is missing canonical name in master data")

    for station in master_data.bus_groups:
        for asset_connection in station.branch_connections:
            if asset_connection.branch_end is None:
                raise ValueError(
                    f"Branch asset {asset_connection.asset_id} in station {station.bus_group_id} "
                    "is missing canonical branch_end in master data"
                )

materialize_runtime_bus_group_from_runtime_state

materialize_runtime_bus_group_from_runtime_state(
    canonical_bus_group,
    branch_asset_map,
    injection_asset_map,
    asset_bay_map,
    runtime_switching_state,
    *,
    model_log=None,
)

Materialize one master bus group from compact runtime switching state.

PARAMETER	DESCRIPTION
canonical_bus_group	Canonical bus-group definition. TYPE: MasterBusGroup
branch_asset_map	Canonical branch assets keyed by grid-model id. TYPE: dict[str, BranchAsset]
injection_asset_map	Canonical injection assets keyed by grid-model id. TYPE: dict[str, InjectionAsset]
asset_bay_map	Canonical asset bays keyed by asset-bay id. TYPE: dict[str, AssetBay]
runtime_switching_state	Compact runtime overlay describing the current live state. TYPE: RuntimeSwitchingState
model_log	Optional log messages to attach to the materialized bus group. TYPE: Optional[list[str]] DEFAULT: None

RETURNS	DESCRIPTION
RuntimeBusGroup	Runtime bus-group snapshot combining canonical structure and live switching state.

Source code in packages/interfaces_pkg/src/toop_engine_interfaces/asset_topology/topology_conversion.py

def materialize_runtime_bus_group_from_runtime_state(
    canonical_bus_group: MasterBusGroup,
    branch_asset_map: dict[str, BranchAsset],
    injection_asset_map: dict[str, InjectionAsset],
    asset_bay_map: dict[str, AssetBay],
    runtime_switching_state: RuntimeSwitchingState,
    *,
    model_log: Optional[list[str]] = None,
) -> RuntimeBusGroup:
    """Materialize one master bus group from compact runtime switching state.

    Parameters
    ----------
    canonical_bus_group : MasterBusGroup
        Canonical bus-group definition.
    branch_asset_map : dict[str, BranchAsset]
        Canonical branch assets keyed by grid-model id.
    injection_asset_map : dict[str, InjectionAsset]
        Canonical injection assets keyed by grid-model id.
    asset_bay_map : dict[str, AssetBay]
        Canonical asset bays keyed by asset-bay id.
    runtime_switching_state : RuntimeSwitchingState
        Compact runtime overlay describing the current live state.
    model_log : Optional[list[str]], optional
        Optional log messages to attach to the materialized bus group.

    Returns
    -------
    RuntimeBusGroup
        Runtime bus-group snapshot combining canonical structure and live switching state.
    """
    materialized_busbars = [
        RuntimeBusbar(
            **busbar.model_dump(),
            in_service=busbar.grid_model_id not in runtime_switching_state.busbar_out_of_service_ids,
            bus_branch_bus_id=(
                runtime_switching_state.busbar_bus_branch_bus_ids.get(busbar.grid_model_id)
                if runtime_switching_state.busbar_bus_branch_bus_ids is not None
                else None
            ),
        )
        for busbar in canonical_bus_group.busbars
    ]
    busbar_int_id_by_grid_model_id = {busbar.grid_model_id: busbar.int_id for busbar in canonical_bus_group.busbars}

    return RuntimeBusGroup(
        bus_group_id=canonical_bus_group.bus_group_id,
        voltage_level_id=canonical_bus_group.voltage_level_id,
        name=canonical_bus_group.name,
        station_type=canonical_bus_group.station_type,
        region=canonical_bus_group.region,
        voltage_level=canonical_bus_group.voltage_level,
        busbars=materialized_busbars,
        couplers=[
            _materialize_runtime_coupler(
                coupler=coupler,
                busbar_int_id_by_grid_model_id=busbar_int_id_by_grid_model_id,
                runtime_switching_state=runtime_switching_state,
            )
            for coupler in canonical_bus_group.couplers
        ],
        branch_connections=[
            RuntimeAssetConnection(
                asset=RuntimeBranchAsset.model_validate(
                    _copy_runtime_or_raise(branch_asset_map, asset_connection.asset_id, "branch asset").model_dump()
                ),
                branch_end=asset_connection.branch_end,
                asset_bay=(
                    _copy_runtime_or_raise(asset_bay_map, asset_connection.asset_bay_id, "asset bay").model_copy(deep=True)
                    if asset_connection.asset_bay_id is not None
                    else None
                ),
            )
            for asset_connection in canonical_bus_group.branch_connections
        ],
        injection_connections=[
            RuntimeAssetConnection(
                asset=RuntimeInjectionAsset.model_validate(
                    _copy_runtime_or_raise(
                        injection_asset_map,
                        asset_connection.asset_id,
                        "injection asset",
                    ).model_dump()
                ),
                branch_end=asset_connection.branch_end,
                asset_bay=(
                    _copy_runtime_or_raise(asset_bay_map, asset_connection.asset_bay_id, "asset bay").model_copy(deep=True)
                    if asset_connection.asset_bay_id is not None
                    else None
                ),
            )
            for asset_connection in canonical_bus_group.injection_connections
        ],
        branch_switching_table=_build_switching_table_from_compact_runtime(
            master_bus_group=canonical_bus_group,
            asset_connections=canonical_bus_group.branch_connections,
            asset_connectivity=(
                np.asarray(canonical_bus_group.branch_connectivity, dtype=bool)
                if canonical_bus_group.branch_connectivity is not None
                else None
            ),
            asset_bay_map=asset_bay_map,
            runtime_switching_state=runtime_switching_state,
            asset_kind="branch",
        ),
        injection_switching_table=_build_switching_table_from_compact_runtime(
            master_bus_group=canonical_bus_group,
            asset_connections=canonical_bus_group.injection_connections,
            asset_connectivity=(
                np.asarray(canonical_bus_group.injection_connectivity, dtype=bool)
                if canonical_bus_group.injection_connectivity is not None
                else None
            ),
            asset_bay_map=asset_bay_map,
            runtime_switching_state=runtime_switching_state,
            asset_kind="injection",
        ),
        branch_connectivity=(
            np.asarray(canonical_bus_group.branch_connectivity, dtype=bool)
            if canonical_bus_group.branch_connectivity is not None
            else None
        ),
        injection_connectivity=(
            np.asarray(canonical_bus_group.injection_connectivity, dtype=bool)
            if canonical_bus_group.injection_connectivity is not None
            else None
        ),
        model_log=list(model_log) if model_log is not None else None,
    )

Switch Updates

toop_engine_interfaces.switch_update_schema

Shared schema for switch updates induced by a topology.

A switch update is a collection of switch ids in the grid model and their new open/closed state. The internal representation of actions does not follow this structure but instead reference the action set. To enable export of the actions to other tools, this needs to be translated into multiple formats, e.g. .dgs for powerfactory or .json for OpenRAO.

SwitchUpdateSchema

Bases: DataFrameModel

Schema for switch update dataframes.

grid_model_id class-attribute instance-attribute

grid_model_id = pa.Field(nullable=True)

The grid_model_id to be updated.

open class-attribute instance-attribute

open = pa.Field(nullable=True)

The value to be set for the switch, True for open, False for closed.