Importer

Importer Worker

toop_engine_importer.worker.worker

Module contains functions for the kafka communication in the importer repo.

File: worker.py Author: Nico Westerbeck Created: 2024

logger module-attribute

logger = structlog.get_logger(__name__)

Args

Bases: BaseModel

Holds arguments which must be provided at the launch of the worker.

Contains arguments that static for each preprocessing run.

kafka_broker class-attribute instance-attribute

kafka_broker = 'localhost:9092'

The Kafka broker to connect to.

importer_command_topic class-attribute instance-attribute

importer_command_topic = 'importer_commands'

The Kafka topic to listen for commands on.

importer_results_topic class-attribute instance-attribute

importer_results_topic = 'importer_results'

The topic to push results to.

importer_heartbeat_topic class-attribute instance-attribute

importer_heartbeat_topic = 'importer_heartbeat'

The topic to push heartbeats to.

heartbeat_interval_ms class-attribute instance-attribute

heartbeat_interval_ms = 1000

The interval in milliseconds to send heartbeats.

idle_loop

idle_loop(
    consumer, send_heartbeat_fn, heartbeat_interval_ms
)

Start the idle loop of the worker.

This will be running when the worker is currently not preprocessing This will wait until a StartPreprocessingCommand is received and return it. In case a ShutdownCommand is received, the worker will exit with the exit code provided in the command.

PARAMETER	DESCRIPTION
consumer	The initialized Kafka consumer to listen for commands on. TYPE: LongRunningKafkaConsumer
send_heartbeat_fn	A function to call when there were no messages received for a while. TYPE: Callable
heartbeat_interval_ms	The time to wait for a new command in milliseconds. If no command has been received, a heartbeat will be sent and then the receiver will wait for commands again. TYPE: int

RETURNS	DESCRIPTION
StartPreprocessingCommand	The start preprocessing command to start the preprocessing run with

Source code in packages/importer_pkg/src/toop_engine_importer/worker/worker.py

def idle_loop(
    consumer: LongRunningKafkaConsumer,
    send_heartbeat_fn: Callable[[], None],
    heartbeat_interval_ms: int,
) -> StartPreprocessingCommand:
    """Start the idle loop of the worker.

    This will be running when the worker is currently not preprocessing
    This will wait until a StartPreprocessingCommand is received and return it. In case a
    ShutdownCommand is received, the worker will exit with the exit code provided in the command.

    Parameters
    ----------
    consumer : LongRunningKafkaConsumer
        The initialized Kafka consumer to listen for commands on.
    send_heartbeat_fn : Callable
        A function to call when there were no messages received for a while.
    heartbeat_interval_ms : int
        The time to wait for a new command in milliseconds. If no command has been received, a
        heartbeat will be sent and then the receiver will wait for commands again.

    Returns
    -------
    StartPreprocessingCommand
        The start preprocessing command to start the preprocessing run with
    """
    send_heartbeat_fn()
    logger.info("Entering idle loop")
    while True:
        message = consumer.poll(timeout=heartbeat_interval_ms / 1000)

        # Wait timeout exceeded
        if not message:
            send_heartbeat_fn()
            continue

        command = Command.model_validate_json(deserialize_message(message.value()))

        if isinstance(command.command, StartPreprocessingCommand):
            return command.command

        consumer.commit()
        if isinstance(command.command, ShutdownCommand):
            consumer.consumer.close()
            raise SystemExit(command.command.exit_code)

        # If we are here, we received a command that we do not know
        logger.warning(f"Received unknown command, dropping: {command}")

main

main(
    args,
    producer,
    consumer,
    unprocessed_gridfile_fs,
    processed_gridfile_fs,
    loadflow_result_fs,
)

Start main function of the worker.

PARAMETER	DESCRIPTION
args	The arguments to start the worker with. TYPE: Args
unprocessed_gridfile_fs	A filesystem where the unprocessed gridfiles are stored. The concrete folder to use is determined by the start command, which contains an import location relative to the root of the unprocessed_gridfile_fs. TYPE: AbstractFileSystem
processed_gridfile_fs	The target filesystem for the preprocessing worker. This contains all processed grid files. During the import job, a new folder import_results.data_folder was created which will be completed with the preprocess call to this function. Internally, only the data folder is passed around as a dirfs. Note that the unprocessed_gridfile_fs is not needed here anymore, as all preprocessing steps that need the unprocessed gridfiles were already done. TYPE: AbstractFileSystem
loadflow_result_fs	A filesystem where the loadflow results are stored. Loadflows will be stored here using the uuid generation process and passed as a StoredLoadflowReference which contains the subfolder in this filesystem. TYPE: AbstractFileSystem
producer	The Kafka producer to send results and heartbeats with. TYPE: Producer
consumer	The Kafka consumer to receive commands with. TYPE: LongRunningKafkaConsumer

Source code in packages/importer_pkg/src/toop_engine_importer/worker/worker.py

def main(
    args: Args,
    producer: Producer,
    consumer: LongRunningKafkaConsumer,
    unprocessed_gridfile_fs: AbstractFileSystem,
    processed_gridfile_fs: AbstractFileSystem,
    loadflow_result_fs: AbstractFileSystem,
) -> None:
    """Start main function of the worker.

    Parameters
    ----------
    args: Args
        The arguments to start the worker with.
    unprocessed_gridfile_fs: AbstractFileSystem
        A filesystem where the unprocessed gridfiles are stored. The concrete folder to use is determined by the start
        command, which contains an import location relative to the root of the unprocessed_gridfile_fs.
    processed_gridfile_fs: AbstractFileSystem
        The target filesystem for the preprocessing worker. This contains all processed grid files.
        During the import job,  a new folder import_results.data_folder was created
        which will be completed with the preprocess call to this function.
        Internally, only the data folder is passed around as a dirfs.
        Note that the unprocessed_gridfile_fs is not needed here anymore, as all preprocessing steps that need the
        unprocessed gridfiles were already done.
    loadflow_result_fs: AbstractFileSystem
        A filesystem where the loadflow results are stored. Loadflows will be stored here using the uuid generation process
        and passed as a StoredLoadflowReference which contains the subfolder in this filesystem.
    producer: Producer
        The Kafka producer to send results and heartbeats with.
    consumer: LongRunningKafkaConsumer
        The Kafka consumer to receive commands with.
    """
    instance_id = str(uuid4())
    logger.info(f"Starting importer instance {instance_id} with arguments {args}")
    jax.config.update("jax_enable_x64", True)
    jax.config.update("jax_logging_level", "INFO")

    def heartbeat_idle() -> None:
        producer.produce(
            args.importer_heartbeat_topic,
            value=serialize_message(
                PreprocessHeartbeat(
                    idle=True,
                    status_info=None,
                    instance_id=instance_id,
                ).model_dump_json()
            ),
            key=instance_id.encode("utf-8"),
        )
        producer.flush()

    def heartbeat_working(
        stage: PreprocessStage,
        message: Optional[str],
        preprocess_id: str,
        start_time: float,
        stats: Optional[NetworkDataStats] = None,
    ) -> None:
        logger.info(
            f"Preprocessing stage {stage} for job {preprocess_id} after {time.time() - start_time}s: {message}",
            preprocess_stage=stage,
            preprocess_id=preprocess_id,
            **({} if stats is None else {"network_stats": stats}),
        )
        producer.produce(
            args.importer_heartbeat_topic,
            value=serialize_message(
                PreprocessHeartbeat(
                    idle=False,
                    status_info=PreprocessStatusInfo(
                        preprocess_id=preprocess_id,
                        runtime=time.time() - start_time,
                        stage=stage,
                        message=message,
                    ),
                    instance_id=instance_id,
                ).model_dump_json()
            ),
            key=preprocess_id.encode("utf-8"),
        )
        producer.flush()
        # Ping the command consumer to show we are still alive
        consumer.heartbeat()

    while True:
        command = idle_loop(
            consumer=consumer,
            send_heartbeat_fn=heartbeat_idle,
            heartbeat_interval_ms=args.heartbeat_interval_ms,
        )

        with structlog.contextvars.bound_contextvars(
            preprocess_id=command.preprocess_id,
        ):
            consumer.start_processing()

            start_time = time.time()
            heartbeat_fn = partial(
                heartbeat_working,
                preprocess_id=command.preprocess_id,
                start_time=start_time,
            )
            producer.produce(
                args.importer_results_topic,
                value=serialize_message(
                    Result(
                        preprocess_id=command.preprocess_id,
                        runtime=0,
                        result=PreprocessingStartedResult(),
                    ).model_dump_json()
                ),
                key=command.preprocess_id.encode(),
            )
            producer.flush()
            heartbeat_fn("start", "Preprocessing run started")

            try:
                result = run_preprocessing(
                    start_command=command,
                    unprocessed_gridfile_fs=unprocessed_gridfile_fs,
                    processed_gridfile_fs=processed_gridfile_fs,
                    loadflow_result_fs=loadflow_result_fs,
                    status_update_fn=heartbeat_fn,
                )

                heartbeat_fn("end", "Preprocessing run done")

                producer.produce(
                    topic=args.importer_results_topic,
                    value=serialize_message(
                        Result(
                            preprocess_id=command.preprocess_id,
                            runtime=time.time() - start_time,
                            result=result,
                        ).model_dump_json()
                    ),
                    key=command.preprocess_id.encode(),
                )
            except Exception as e:
                logger.error(f"Error while processing {command.preprocess_id}", exc_info=e)
                producer.produce(
                    topic=args.importer_results_topic,
                    value=serialize_message(
                        Result(
                            preprocess_id=command.preprocess_id,
                            runtime=time.time() - start_time,
                            result=ErrorResult(error=str(e)),
                        ).model_dump_json()
                    ),
                    key=command.preprocess_id.encode(),
                )
            producer.flush()
            consumer.stop_processing()

toop_engine_importer.worker.preprocessor

Module contains functions holds preprocessor commands for kafka communication in the importer repo.

File: preprocessor.py Author: Nico Westerbeck Created: 2024

logger module-attribute

logger = structlog.get_logger(__name__)

run_preprocessing

run_preprocessing(
    start_command,
    unprocessed_gridfile_fs,
    processed_gridfile_fs,
    loadflow_result_fs,
    status_update_fn,
)

Run the whole preprocessing of a start command: import, initial loadflow and DC preprocessing.

The AC N-1 security analysis runs once, during the import. It sets the artificial limits and is stored as the initial loadflow, before the DC preprocessing reads the grid with these limits.

PARAMETER	DESCRIPTION
start_command	The command to start the preprocessing run with TYPE: StartPreprocessingCommand
unprocessed_gridfile_fs	A filesystem where the unprocessed gridfiles are stored. TYPE: AbstractFileSystem
processed_gridfile_fs	The target filesystem for the preprocessing worker. This contains all processed grid files. TYPE: AbstractFileSystem
loadflow_result_fs	A filesystem where the loadflow results are stored - this should be a NFS share together with the backend and optimizer. The initial loadflow is stored here. TYPE: AbstractFileSystem
status_update_fn	A function to call to signal progress in the preprocessing pipeline. Takes a stage, an optional message as parameters and network size stats. TYPE: StatusUpdateFn

RETURNS	DESCRIPTION
PreprocessingSuccessResult | NoOptimizationRequiredResult	A result dataclass for the entire preprocessing, including paths to the ready static_information and network_data dataclasses.

RAISES	DESCRIPTION
BaseCaseNotConvergedError	If the base case loadflow of the import did not converge.
Exception	Any exception raised will be caught by the worker and sent back

Source code in packages/importer_pkg/src/toop_engine_importer/worker/preprocessor.py

def run_preprocessing(
    start_command: StartPreprocessingCommand,
    unprocessed_gridfile_fs: AbstractFileSystem,
    processed_gridfile_fs: AbstractFileSystem,
    loadflow_result_fs: AbstractFileSystem,
    status_update_fn: StatusUpdateFn,
) -> PreprocessingSuccessResult | NoOptimizationRequiredResult:
    """Run the whole preprocessing of a start command: import, initial loadflow and DC preprocessing.

    The AC N-1 security analysis runs once, during the import. It sets the artificial limits and is stored as the
    initial loadflow, before the DC preprocessing reads the grid with these limits.

    Parameters
    ----------
    start_command: StartPreprocessingCommand
        The command to start the preprocessing run with
    unprocessed_gridfile_fs: AbstractFileSystem
        A filesystem where the unprocessed gridfiles are stored.
    processed_gridfile_fs: AbstractFileSystem
        The target filesystem for the preprocessing worker. This contains all processed grid files.
    loadflow_result_fs: AbstractFileSystem
        A filesystem where the loadflow results are stored - this should be a NFS share together with the backend and
        optimizer. The initial loadflow is stored here.
    status_update_fn: StatusUpdateFn
        A function to call to signal progress in the preprocessing pipeline. Takes a stage, an
        optional message as parameters and network size stats.

    Returns
    -------
    PreprocessingSuccessResult | NoOptimizationRequiredResult
        A result dataclass for the entire preprocessing, including paths to the ready
        static_information and network_data dataclasses.

    Raises
    ------
    BaseCaseNotConvergedError
        If the base case loadflow of the import did not converge.
    Exception
        Any exception raised will be caught by the worker and sent back
    """
    logger.info("Starting preprocessing", preprocess_id=start_command.preprocess_id)
    imported_grid_model = import_grid_model(
        start_command=start_command,
        unprocessed_gridfile_fs=unprocessed_gridfile_fs,
        processed_gridfile_fs=processed_gridfile_fs,
        status_update_fn=status_update_fn,
    )
    import_result = imported_grid_model.import_result.model_copy(
        update={
            "optimization_required": _determine_optimization_required(
                imported_grid_model.import_result,
                start_command.preprocess_parameters.minimum_initial_ac_overload_energy,
            ),
        }
    )
    initial_metrics = imported_grid_model.initial_metrics
    initial_loadflow = store_initial_ca_results(
        start_command=start_command,
        initial_loadflow=imported_grid_model.initial_loadflow,
        loadflow_result_fs=loadflow_result_fs,
    )
    # Free the AC results before the memory peak of the DC preprocessing
    del imported_grid_model

    preprocessing_results: PreprocessingSuccessResult | NoOptimizationRequiredResult
    if import_result.optimization_required:
        preprocessing_results = preprocess(
            start_command=start_command,
            import_results=import_result,
            status_update_fn=status_update_fn,
            processed_gridfile_fs=processed_gridfile_fs,
            initial_loadflow=initial_loadflow,
            initial_metrics=initial_metrics,
        )
    else:
        preprocessing_results = NoOptimizationRequiredResult(
            data_folder=import_result.data_folder,
            importer_results=import_result,
        )
        logger.warning("No optimization required for this preprocessing run", preprocess_id=start_command.preprocess_id)

    logger.info("Finished preprocessing", preprocess_id=start_command.preprocess_id)
    return preprocessing_results

import_grid_model

import_grid_model(
    start_command,
    unprocessed_gridfile_fs,
    processed_gridfile_fs,
    status_update_fn,
)

Run the import procedure, including the AC security analysis that sets the artificial limits.

This only performs the import until there's a grid model, the preprocessing in the loadflow solver is run by preprocess

PARAMETER	DESCRIPTION
start_command	The command to start the preprocessing run with TYPE: StartPreprocessingCommand
unprocessed_gridfile_fs	A filesystem where the unprocessed gridfiles are stored. The concrete folder to use is determined by the start command, which contains an import location relative to the root of the unprocessed_gridfile_fs. TYPE: AbstractFileSystem
processed_gridfile_fs	The target filesystem for the preprocessing worker. This contains all processed grid files. During the import job, a new folder import_results.data_folder is created. TYPE: AbstractFileSystem
status_update_fn	A function to call to signal progress in the preprocessing pipeline. Takes a stage, an optional message as parameters and network size stats. TYPE: StatusUpdateFn

RETURNS	DESCRIPTION
ImportedGridModel	The import statistics, mainly including the grid folder, and the initial loadflow with its metrics.

RAISES	DESCRIPTION
Exception	Any exception raised will be caught by the worker and sent back

Source code in packages/importer_pkg/src/toop_engine_importer/worker/preprocessor.py

def import_grid_model(
    start_command: StartPreprocessingCommand,
    unprocessed_gridfile_fs: AbstractFileSystem,
    processed_gridfile_fs: AbstractFileSystem,
    status_update_fn: StatusUpdateFn,
) -> ImportedGridModel:
    """Run the import procedure, including the AC security analysis that sets the artificial limits.

    This only performs the import until there's a grid model, the preprocessing in the loadflow
    solver is run by preprocess

    Parameters
    ----------
    start_command: StartPreprocessingCommand
        The command to start the preprocessing run with
    unprocessed_gridfile_fs: AbstractFileSystem
        A filesystem where the unprocessed gridfiles are stored. The concrete folder to use is determined by the start
        command, which contains an import location relative to the root of the unprocessed_gridfile_fs.
    processed_gridfile_fs: AbstractFileSystem
        The target filesystem for the preprocessing worker. This contains all processed grid files.
        During the import job,  a new folder import_results.data_folder is created.
    status_update_fn: StatusUpdateFn
        A function to call to signal progress in the preprocessing pipeline. Takes a stage, an
        optional message as parameters and network size stats.

    Returns
    -------
    ImportedGridModel
        The import statistics, mainly including the grid folder, and the initial loadflow with its metrics.

    Raises
    ------
    Exception
        Any exception raised will be caught by the worker and sent back
    """
    imported_grid_model = preprocessing.import_grid_with_security_analysis(
        importer_parameters=start_command.importer_parameters,
        status_update_fn=status_update_fn,
        unprocessed_gridfile_fs=unprocessed_gridfile_fs,
        processed_gridfile_fs=processed_gridfile_fs,
        job_id=start_command.preprocess_id,
    )
    return imported_grid_model

store_initial_ca_results

store_initial_ca_results(
    start_command, initial_loadflow, loadflow_result_fs
)

Store the AC N-1 results of the import as the initial loadflow.

PARAMETER	DESCRIPTION
start_command	The command that was sent to the worker TYPE: StartPreprocessingCommand
initial_loadflow	The AC N-1 results of the import, with the permanent and the artificial branch limits. TYPE: LoadflowResultsPolars
loadflow_result_fs	A filesystem where the loadflow results are stored - this should be a NFS share together with the backend and optimizer. TYPE: AbstractFileSystem

RETURNS	DESCRIPTION
StoredLoadflowReference	A reference to the stored loadflow results

Source code in packages/importer_pkg/src/toop_engine_importer/worker/preprocessor.py

def store_initial_ca_results(
    start_command: StartPreprocessingCommand,
    initial_loadflow: LoadflowResultsPolars,
    loadflow_result_fs: AbstractFileSystem,
) -> StoredLoadflowReference:
    """Store the AC N-1 results of the import as the initial loadflow.

    Parameters
    ----------
    start_command: StartPreprocessingCommand
        The command that was sent to the worker
    initial_loadflow: LoadflowResultsPolars
        The AC N-1 results of the import, with the permanent and the artificial branch limits.
    loadflow_result_fs: AbstractFileSystem
        A filesystem where the loadflow results are stored - this should be a NFS share together with the backend and
        optimizer.

    Returns
    -------
    StoredLoadflowReference
        A reference to the stored loadflow results
    """
    reference = save_loadflow_results_polars(
        loadflow_result_fs, f"initial_loadflow_{start_command.preprocess_id}", initial_loadflow
    )
    logger.info(
        "Stored the initial loadflow",
        preprocess_id=start_command.preprocess_id,
        relative_path=reference.relative_path,
    )
    return reference

preprocess

preprocess(
    start_command,
    import_results,
    status_update_fn,
    processed_gridfile_fs,
    initial_loadflow,
    initial_metrics,
)

Run the preprocessing pipeline that is independent of the data source.

This only performs the preprocessing in the loadflow solver

PARAMETER	DESCRIPTION
start_command	The command to start the preprocessing run with TYPE: StartPreprocessingCommand
import_results	Results from the import procedure TYPE: ImportResult
status_update_fn	A function to call to signal progress in the preprocessing pipeline. Takes a stage and an optional message as parameters TYPE: StatusUpdateFn
processed_gridfile_fs	The target filesystem for the preprocessing worker. This contains all processed grid files. During the import job, a new folder import_results.data_folder was created which will be completed with the preprocess call to this function. Internally, only the data folder is passed around as a dirfs. Note that the unprocessed_gridfile_fs is not needed here anymore, as all preprocessing steps that need the unprocessed gridfiles were already done. TYPE: AbstractFileSystem
initial_loadflow	The reference to the stored initial loadflow of the import. TYPE: StoredLoadflowReference
initial_metrics	The metrics of the initial loadflow. TYPE: dict[MetricType, float]

RETURNS	DESCRIPTION
PreprocessingSuccessResult	The result of completed DC preprocessing.

RAISES	DESCRIPTION
Exception	Any exception raised will be caught by the worker and sent back

Source code in packages/importer_pkg/src/toop_engine_importer/worker/preprocessor.py

def preprocess(
    start_command: StartPreprocessingCommand,
    import_results: ImportResult,
    status_update_fn: StatusUpdateFn,
    processed_gridfile_fs: AbstractFileSystem,
    initial_loadflow: StoredLoadflowReference,
    initial_metrics: dict[MetricType, float],
) -> PreprocessingSuccessResult:
    """Run the preprocessing pipeline that is independent of the data source.

    This only performs the preprocessing in the loadflow solver

    Parameters
    ----------
    start_command: StartPreprocessingCommand
        The command to start the preprocessing run with
    import_results: ImportResult
        Results from the import procedure
    status_update_fn: StatusUpdateFn
        A function to call to signal progress in the preprocessing pipeline. Takes a stage and an
        optional message as parameters
    processed_gridfile_fs: AbstractFileSystem
        The target filesystem for the preprocessing worker. This contains all processed grid files.
        During the import job,  a new folder import_results.data_folder was created
        which will be completed with the preprocess call to this function.
        Internally, only the data folder is passed around as a dirfs.
        Note that the unprocessed_gridfile_fs is not needed here anymore, as all preprocessing steps that need the
        unprocessed gridfiles were already done.
    initial_loadflow: StoredLoadflowReference
        The reference to the stored initial loadflow of the import.
    initial_metrics: dict[MetricType, float]
        The metrics of the initial loadflow.

    Returns
    -------
    PreprocessingSuccessResult
        The result of completed DC preprocessing.

    Raises
    ------
    Exception
        Any exception raised will be caught by the worker and sent back
    """
    preprocess_parameters = start_command.preprocess_parameters
    pandapower = False
    if import_results.grid_type == "power_factory":
        pandapower = True

    # Create a dirfs that points to the data folder, so we can pass around the dirfs instead of the path + fs
    output_dirfs = DirFileSystem(path=str(import_results.data_folder), fs=processed_gridfile_fs)

    lf_params = load_lf_params_from_fs(output_dirfs, Path(PREPROCESSING_PATHS["loadflow_parameters_file_path"]))
    info, _, _ = load_grid(
        data_folder_dirfs=output_dirfs,
        pandapower=pandapower,
        parameters=preprocess_parameters,
        status_update_fn=status_update_fn,
        lf_params=lf_params,
    )

    return PreprocessingSuccessResult(
        data_folder=import_results.data_folder,
        static_information_stats=info,
        importer_results=import_results,
        initial_loadflow=initial_loadflow,
        initial_metrics=initial_metrics,
    )

Contingency from PowerFactory

toop_engine_importer.contingency_from_power_factory

Import contingency from PowerFactory.

__all__ module-attribute

__all__ = [
    "AllGridElementsSchema",
    "ContingencyImportSchemaPowerFactory",
    "ContingencyMatchSchema",
    "get_contingencies_from_file",
    "match_contingencies",
    "power_factory_data_class",
]

AllGridElementsSchema

Bases: DataFrameModel

A AllGridElementsSchema is a DataFrameModel for all grid elements in the grid model.

The grid model is loaded from the CGMES file in either PyPowsybl or Pandapower.

element_type class-attribute instance-attribute

element_type = pa.Field(
    nullable=True, isin=list(GridElementType.__args__)
)

The grid model type of the contingency. e.g. LINE, SWITCH, BUS, etc.

grid_model_id class-attribute instance-attribute

grid_model_id = pa.Field(nullable=True)

The grid model id of the contingency. e.g. a CGMES id (cryptic number)

grid_model_name class-attribute instance-attribute

grid_model_name = pa.Field(nullable=True)

The grid model name of the contingency. e.g. a CGMES name (human readable name)

ContingencyImportSchemaPowerFactory

Bases: DataFrameModel

A ContingencyImportSchemaPowerFactory is a DataFrameModel defining the expected data of the contingency import.

From PowerFactory: You may find the list of contingencies in the PowerFactory GUI under "Calculation > Contingency Analysis > Show Contingencies...".

index class-attribute instance-attribute

index = pa.Field(nullable=False)

The unique index of the DataFrame. This index is used as a unique id for the dataframe.

contingency_name class-attribute instance-attribute

contingency_name = pa.Field(nullable=False)

The id of contingency found in the contingency table. Attribute: "loc_name" of contingency table May be a multi index to group the contingencies together.

contingency_id class-attribute instance-attribute

contingency_id = pa.Field(nullable=False)

A id for the contingency. This id is used to group the contingencies together. Attribute: "number" of contingency table.

power_factory_grid_model_name class-attribute instance-attribute

power_factory_grid_model_name = pa.Field(nullable=False)

The name of the grid model element Attribute: "loc_name" of grid model element

power_factory_grid_model_fid class-attribute instance-attribute

power_factory_grid_model_fid = pa.Field(nullable=True)

The foreign Key of the grid model element Attribute: "for_name" of grid model element Note: True spacing of FID must be kept in the string.

power_factory_grid_model_rdf_id class-attribute instance-attribute

power_factory_grid_model_rdf_id = pa.Field(nullable=True)

The rdf id (CIM) of the grid model element Attribute: "cimRdfId" of grid model element

comment class-attribute instance-attribute

comment = pa.Field(nullable=True)

May contain information about the contingency. Leave empty if not needed. Fill if comments or descriptions exist in the contingency table.

power_factory_element_type class-attribute instance-attribute

power_factory_element_type = pa.Field(
    nullable=True, isin=list(GridElementType.__args__)
)

The type of the contingency based on the PowerFactory type. Gives a hint where to look for the contingency.

ContingencyMatchSchema

Bases: ContingencyImportSchemaPowerFactory, AllGridElementsSchema

A ContingencyMatchSchema is a DataFrameModel for matching the ContingencyImportSchema with the grid model.

ContingencyMatchSchema is a merge of: ContingencyImportSchema.merge( AllGridElementsSchema, how="left", left_on="power_factory_grid_model_rdf_id", right_on="grid_model_id" ) Note: the power_factory_grid_model_rdf_id has a leading underscore and may need modification.

element_type class-attribute instance-attribute

element_type = pa.Field(
    nullable=True, isin=list(GridElementType.__args__)
)

The grid model type of the contingency. e.g. LINE, SWITCH, BUS, etc.

grid_model_id class-attribute instance-attribute

grid_model_id = pa.Field(nullable=True)

The grid model id of the contingency. e.g. a CGMES id (cryptic number)

grid_model_name class-attribute instance-attribute

grid_model_name = pa.Field(nullable=True)

The grid model name of the contingency. e.g. a CGMES name (human readable name)

index class-attribute instance-attribute

index = pa.Field(nullable=False)

The unique index of the DataFrame. This index is used as a unique id for the dataframe.

contingency_name class-attribute instance-attribute

contingency_name = pa.Field(nullable=False)

The id of contingency found in the contingency table. Attribute: "loc_name" of contingency table May be a multi index to group the contingencies together.

contingency_id class-attribute instance-attribute

contingency_id = pa.Field(nullable=False)

A id for the contingency. This id is used to group the contingencies together. Attribute: "number" of contingency table.

power_factory_grid_model_name class-attribute instance-attribute

power_factory_grid_model_name = pa.Field(nullable=False)

The name of the grid model element Attribute: "loc_name" of grid model element

power_factory_grid_model_fid class-attribute instance-attribute

power_factory_grid_model_fid = pa.Field(nullable=True)

The foreign Key of the grid model element Attribute: "for_name" of grid model element Note: True spacing of FID must be kept in the string.

power_factory_grid_model_rdf_id class-attribute instance-attribute

power_factory_grid_model_rdf_id = pa.Field(nullable=True)

The rdf id (CIM) of the grid model element Attribute: "cimRdfId" of grid model element

comment class-attribute instance-attribute

comment = pa.Field(nullable=True)

May contain information about the contingency. Leave empty if not needed. Fill if comments or descriptions exist in the contingency table.

power_factory_element_type class-attribute instance-attribute

power_factory_element_type = pa.Field(
    nullable=True, isin=list(GridElementType.__args__)
)

The type of the contingency based on the PowerFactory type. Gives a hint where to look for the contingency.

get_contingencies_from_file

get_contingencies_from_file(
    n1_file, delimiter=";", filesystem=None
)

Get the contingencies from the file.

This function reads the contingencies from the file and returns a DataFrame in the ContingencyImportSchema format.

PARAMETER	DESCRIPTION
n1_file	The path to the file. TYPE: Path
delimiter	The delimiter of the file. Default is ";". TYPE: str DEFAULT: ';'
filesystem	The filesystem to use to read the file. If None, the local filesystem is used. TYPE: AbstractFileSystem | None DEFAULT: None

RETURNS	DESCRIPTION
ContingencyImportSchema	A DataFrame containing the contingencies.

Source code in packages/importer_pkg/src/toop_engine_importer/contingency_from_power_factory/contingency_from_file.py

def get_contingencies_from_file(
    n1_file: Path, delimiter: str = ";", filesystem: AbstractFileSystem | None = None
) -> pat.DataFrame[ContingencyImportSchemaPowerFactory]:
    """Get the contingencies from the file.

    This function reads the contingencies from the file and returns a DataFrame in the
    ContingencyImportSchema format.

    Parameters
    ----------
    n1_file : Path
        The path to the file.
    delimiter : str
        The delimiter of the file. Default is ";".
    filesystem : AbstractFileSystem | None
        The filesystem to use to read the file. If None, the local filesystem is used.

    Returns
    -------
    ContingencyImportSchema
        A DataFrame containing the contingencies.
    """
    if filesystem is None:
        filesystem = LocalFileSystem()
    with filesystem.open(str(n1_file), "r") as f:
        n1_definition = pd.read_csv(f, delimiter=delimiter)
    cond = n1_definition["power_factory_grid_model_name"].isna()
    n1_definition.loc[cond, "power_factory_grid_model_name"] = n1_definition.loc[cond, "contingency_name"]
    n1_definition["contingency_id"] = n1_definition["contingency_id"].astype(int)
    ContingencyImportSchemaPowerFactory.validate(n1_definition)
    return n1_definition

match_contingencies

match_contingencies(
    n1_definition, all_element_names, match_by_name=True
)

Match the contingencies from the file with the elements in the grid model.

This function matches the contingencies from the file with the elements in the grid model. It first tries to match by index, then by name.

PARAMETER	DESCRIPTION
n1_definition	The contingencies from the file. TYPE: DataFrame[ContingencyImportSchemaPowerFactory]
all_element_names	The elements in the grid model. TYPE: DataFrame[AllGridElementsSchema]
match_by_name	If True, match by name. Default is True. If False, only match by index. TYPE: bool DEFAULT: True

RETURNS	DESCRIPTION
DataFrame[ContingencyMatchSchema]	A DataFrame containing the matched contingencies.

Source code in packages/importer_pkg/src/toop_engine_importer/contingency_from_power_factory/contingency_from_file.py

def match_contingencies(
    n1_definition: pat.DataFrame[ContingencyImportSchemaPowerFactory],
    all_element_names: pat.DataFrame[AllGridElementsSchema],
    match_by_name: bool = True,
) -> pat.DataFrame[ContingencyMatchSchema]:
    """Match the contingencies from the file with the elements in the grid model.

    This function matches the contingencies from the file with the elements in the grid model.
    It first tries to match by index, then by name.

    Parameters
    ----------
    n1_definition : pat.DataFrame[ContingencyImportSchemaPowerFactory]
        The contingencies from the file.
    all_element_names : pat.DataFrame[AllGridElementsSchema]
        The elements in the grid model.
    match_by_name : bool
        If True, match by name. Default is True.
        If False, only match by index.

    Returns
    -------
    pat.DataFrame[ContingencyMatchSchema]
        A DataFrame containing the matched contingencies.
    """
    processed_n1_definition = match_contingencies_by_index(n1_definition, all_element_names)
    if match_by_name:
        processed_n1_definition = match_contingencies_by_name(processed_n1_definition, all_element_names)
    return processed_n1_definition

toop_engine_importer.contingency_from_power_factory.power_factory_data_class

Classes for the contingency list import from PowerFactory.

This importing has the focus on CIM bases grid models. UCTE has not been tested.

Author: Benjamin Petrick Created: 2025-05-13

GridModelTypePowerFactory module-attribute

GridModelTypePowerFactory = Literal[
    "ElmTr2",
    "ElmLne",
    "ElmGenstat",
    "ElmLod",
    "ElmSym",
    "ElmNec",
    "ElmZpu",
    "ElmTr3",
    "ElmSind",
    "ElmTerm",
    "ElmShnt",
    "ElmVac",
]

GridElementType module-attribute

GridElementType = Literal[
    "BUS",
    "BUSBAR_SECTION",
    "LINE",
    "HVDC_LINE",
    "SWITCH",
    "TWO_WINDINGS_TRANSFORMER",
    "THREE_WINDINGS_TRANSFORMER",
    "GENERATOR",
    "LOAD",
    "SHUNT_COMPENSATOR",
    "BOUNDARY_LINE",
    "TIE_LINE",
]

ContingencyImportSchemaPowerFactory

Bases: DataFrameModel

A ContingencyImportSchemaPowerFactory is a DataFrameModel defining the expected data of the contingency import.

From PowerFactory: You may find the list of contingencies in the PowerFactory GUI under "Calculation > Contingency Analysis > Show Contingencies...".

index class-attribute instance-attribute

index = pa.Field(nullable=False)

The unique index of the DataFrame. This index is used as a unique id for the dataframe.

contingency_name class-attribute instance-attribute

contingency_name = pa.Field(nullable=False)

The id of contingency found in the contingency table. Attribute: "loc_name" of contingency table May be a multi index to group the contingencies together.

contingency_id class-attribute instance-attribute

contingency_id = pa.Field(nullable=False)

A id for the contingency. This id is used to group the contingencies together. Attribute: "number" of contingency table.

power_factory_grid_model_name class-attribute instance-attribute

power_factory_grid_model_name = pa.Field(nullable=False)

The name of the grid model element Attribute: "loc_name" of grid model element

power_factory_grid_model_fid class-attribute instance-attribute

power_factory_grid_model_fid = pa.Field(nullable=True)

The foreign Key of the grid model element Attribute: "for_name" of grid model element Note: True spacing of FID must be kept in the string.

power_factory_grid_model_rdf_id class-attribute instance-attribute

power_factory_grid_model_rdf_id = pa.Field(nullable=True)

The rdf id (CIM) of the grid model element Attribute: "cimRdfId" of grid model element

comment class-attribute instance-attribute

comment = pa.Field(nullable=True)

May contain information about the contingency. Leave empty if not needed. Fill if comments or descriptions exist in the contingency table.

power_factory_element_type class-attribute instance-attribute

power_factory_element_type = pa.Field(
    nullable=True, isin=list(GridElementType.__args__)
)

The type of the contingency based on the PowerFactory type. Gives a hint where to look for the contingency.

AllGridElementsSchema

Bases: DataFrameModel

A AllGridElementsSchema is a DataFrameModel for all grid elements in the grid model.

The grid model is loaded from the CGMES file in either PyPowsybl or Pandapower.

element_type class-attribute instance-attribute

element_type = pa.Field(
    nullable=True, isin=list(GridElementType.__args__)
)

The grid model type of the contingency. e.g. LINE, SWITCH, BUS, etc.

grid_model_id class-attribute instance-attribute

grid_model_id = pa.Field(nullable=True)

The grid model id of the contingency. e.g. a CGMES id (cryptic number)

grid_model_name class-attribute instance-attribute

grid_model_name = pa.Field(nullable=True)

The grid model name of the contingency. e.g. a CGMES name (human readable name)

ContingencyMatchSchema

Bases: ContingencyImportSchemaPowerFactory, AllGridElementsSchema

A ContingencyMatchSchema is a DataFrameModel for matching the ContingencyImportSchema with the grid model.

ContingencyMatchSchema is a merge of: ContingencyImportSchema.merge( AllGridElementsSchema, how="left", left_on="power_factory_grid_model_rdf_id", right_on="grid_model_id" ) Note: the power_factory_grid_model_rdf_id has a leading underscore and may need modification.

element_type class-attribute instance-attribute

element_type = pa.Field(
    nullable=True, isin=list(GridElementType.__args__)
)

The grid model type of the contingency. e.g. LINE, SWITCH, BUS, etc.

grid_model_id class-attribute instance-attribute

grid_model_id = pa.Field(nullable=True)

The grid model id of the contingency. e.g. a CGMES id (cryptic number)

grid_model_name class-attribute instance-attribute

grid_model_name = pa.Field(nullable=True)

The grid model name of the contingency. e.g. a CGMES name (human readable name)

index class-attribute instance-attribute

index = pa.Field(nullable=False)

The unique index of the DataFrame. This index is used as a unique id for the dataframe.

contingency_name class-attribute instance-attribute

contingency_name = pa.Field(nullable=False)

The id of contingency found in the contingency table. Attribute: "loc_name" of contingency table May be a multi index to group the contingencies together.

contingency_id class-attribute instance-attribute

contingency_id = pa.Field(nullable=False)

A id for the contingency. This id is used to group the contingencies together. Attribute: "number" of contingency table.

power_factory_grid_model_name class-attribute instance-attribute

power_factory_grid_model_name = pa.Field(nullable=False)

The name of the grid model element Attribute: "loc_name" of grid model element

power_factory_grid_model_fid class-attribute instance-attribute

power_factory_grid_model_fid = pa.Field(nullable=True)

The foreign Key of the grid model element Attribute: "for_name" of grid model element Note: True spacing of FID must be kept in the string.

power_factory_grid_model_rdf_id class-attribute instance-attribute

power_factory_grid_model_rdf_id = pa.Field(nullable=True)

The rdf id (CIM) of the grid model element Attribute: "cimRdfId" of grid model element

comment class-attribute instance-attribute

comment = pa.Field(nullable=True)

May contain information about the contingency. Leave empty if not needed. Fill if comments or descriptions exist in the contingency table.

power_factory_element_type class-attribute instance-attribute

power_factory_element_type = pa.Field(
    nullable=True, isin=list(GridElementType.__args__)
)

The type of the contingency based on the PowerFactory type. Gives a hint where to look for the contingency.

toop_engine_importer.contingency_from_power_factory.contingency_from_file

Import contingencies from a file.

This module contains functions to import contingencies from a file and match them with the grid model.

Author: Benjamin Petrick Created: 2025-05-13

logger module-attribute

logger = structlog.get_logger(__name__)

get_contingencies_from_file

get_contingencies_from_file(
    n1_file, delimiter=";", filesystem=None
)

Get the contingencies from the file.

This function reads the contingencies from the file and returns a DataFrame in the ContingencyImportSchema format.

PARAMETER	DESCRIPTION
n1_file	The path to the file. TYPE: Path
delimiter	The delimiter of the file. Default is ";". TYPE: str DEFAULT: ';'
filesystem	The filesystem to use to read the file. If None, the local filesystem is used. TYPE: AbstractFileSystem | None DEFAULT: None

RETURNS	DESCRIPTION
ContingencyImportSchema	A DataFrame containing the contingencies.

Source code in packages/importer_pkg/src/toop_engine_importer/contingency_from_power_factory/contingency_from_file.py

def get_contingencies_from_file(
    n1_file: Path, delimiter: str = ";", filesystem: AbstractFileSystem | None = None
) -> pat.DataFrame[ContingencyImportSchemaPowerFactory]:
    """Get the contingencies from the file.

    This function reads the contingencies from the file and returns a DataFrame in the
    ContingencyImportSchema format.

    Parameters
    ----------
    n1_file : Path
        The path to the file.
    delimiter : str
        The delimiter of the file. Default is ";".
    filesystem : AbstractFileSystem | None
        The filesystem to use to read the file. If None, the local filesystem is used.

    Returns
    -------
    ContingencyImportSchema
        A DataFrame containing the contingencies.
    """
    if filesystem is None:
        filesystem = LocalFileSystem()
    with filesystem.open(str(n1_file), "r") as f:
        n1_definition = pd.read_csv(f, delimiter=delimiter)
    cond = n1_definition["power_factory_grid_model_name"].isna()
    n1_definition.loc[cond, "power_factory_grid_model_name"] = n1_definition.loc[cond, "contingency_name"]
    n1_definition["contingency_id"] = n1_definition["contingency_id"].astype(int)
    ContingencyImportSchemaPowerFactory.validate(n1_definition)
    return n1_definition

match_contingencies

match_contingencies(
    n1_definition, all_element_names, match_by_name=True
)

Match the contingencies from the file with the elements in the grid model.

This function matches the contingencies from the file with the elements in the grid model. It first tries to match by index, then by name.

PARAMETER	DESCRIPTION
n1_definition	The contingencies from the file. TYPE: DataFrame[ContingencyImportSchemaPowerFactory]
all_element_names	The elements in the grid model. TYPE: DataFrame[AllGridElementsSchema]
match_by_name	If True, match by name. Default is True. If False, only match by index. TYPE: bool DEFAULT: True

RETURNS	DESCRIPTION
DataFrame[ContingencyMatchSchema]	A DataFrame containing the matched contingencies.

Source code in packages/importer_pkg/src/toop_engine_importer/contingency_from_power_factory/contingency_from_file.py

def match_contingencies(
    n1_definition: pat.DataFrame[ContingencyImportSchemaPowerFactory],
    all_element_names: pat.DataFrame[AllGridElementsSchema],
    match_by_name: bool = True,
) -> pat.DataFrame[ContingencyMatchSchema]:
    """Match the contingencies from the file with the elements in the grid model.

    This function matches the contingencies from the file with the elements in the grid model.
    It first tries to match by index, then by name.

    Parameters
    ----------
    n1_definition : pat.DataFrame[ContingencyImportSchemaPowerFactory]
        The contingencies from the file.
    all_element_names : pat.DataFrame[AllGridElementsSchema]
        The elements in the grid model.
    match_by_name : bool
        If True, match by name. Default is True.
        If False, only match by index.

    Returns
    -------
    pat.DataFrame[ContingencyMatchSchema]
        A DataFrame containing the matched contingencies.
    """
    processed_n1_definition = match_contingencies_by_index(n1_definition, all_element_names)
    if match_by_name:
        processed_n1_definition = match_contingencies_by_name(processed_n1_definition, all_element_names)
    return processed_n1_definition

match_contingencies_by_index

match_contingencies_by_index(
    n1_definition, all_element_names
)

Match the contingencies from the file with the elements in the grid model.

PARAMETER	DESCRIPTION
n1_definition	The contingencies from the file. TYPE: DataFrame[ContingencyImportSchemaPowerFactory]
all_element_names	The elements in the grid model. TYPE: DataFrame[AllGridElementsSchema]

RETURNS	DESCRIPTION
DataFrame[ContingencyMatchSchema]	A DataFrame containing the matched contingencies.

Source code in packages/importer_pkg/src/toop_engine_importer/contingency_from_power_factory/contingency_from_file.py

def match_contingencies_by_index(
    n1_definition: pat.DataFrame[ContingencyImportSchemaPowerFactory],
    all_element_names: pat.DataFrame[AllGridElementsSchema],
) -> pat.DataFrame[ContingencyMatchSchema]:
    """Match the contingencies from the file with the elements in the grid model.

    Parameters
    ----------
    n1_definition : pat.DataFrame[ContingencyImportSchemaPowerFactory]
        The contingencies from the file.
    all_element_names : pat.DataFrame[AllGridElementsSchema]
        The elements in the grid model.

    Returns
    -------
    pat.DataFrame[ContingencyMatchSchema]
        A DataFrame containing the matched contingencies.
    """
    # match grid_model_ids directly
    processed_n1_definition = n1_definition.merge(
        all_element_names, how="left", left_on="power_factory_grid_model_rdf_id", right_on="grid_model_id"
    )
    if (~processed_n1_definition["grid_model_name"].isna()).sum() == 0:
        # no elements found, try to remove underscore from rdf_id
        n1_definition["power_factory_grid_model_rdf_id"] = n1_definition["power_factory_grid_model_rdf_id"].str[1:]
        processed_n1_definition = n1_definition.merge(
            all_element_names, how="left", left_on="power_factory_grid_model_rdf_id", right_on="grid_model_id"
        )
    if (~processed_n1_definition["grid_model_name"].isna()).sum() == 0:
        logger.warning("No elements found in the grid model via CIM id. Check the grid model and the contingency file.")

    ContingencyMatchSchema.validate(processed_n1_definition)
    return processed_n1_definition

match_contingencies_by_name

match_contingencies_by_name(
    processed_n1_definition, all_element_names
)

Match the contingencies from the file with the elements in the grid model by name.

Matches by name and replaces the grid_model_name, element_type and grid_model_id. First tries to match 100% of the name. Second tries to match by removing spaces and replacing "+" with "##_##".

PARAMETER	DESCRIPTION
processed_n1_definition	The contingencies from the file. TYPE: DataFrame[ContingencyMatchSchema]
all_element_names	The elements in the grid model. TYPE: DataFrame[AllGridElementsSchema]

RETURNS	DESCRIPTION
DataFrame[ContingencyMatchSchema]	A DataFrame containing the matched contingencies.

Source code in packages/importer_pkg/src/toop_engine_importer/contingency_from_power_factory/contingency_from_file.py

def match_contingencies_by_name(
    processed_n1_definition: pat.DataFrame[ContingencyMatchSchema],
    all_element_names: pat.DataFrame[AllGridElementsSchema],
) -> pat.DataFrame[ContingencyMatchSchema]:
    """Match the contingencies from the file with the elements in the grid model by name.

    Matches by name and replaces the grid_model_name, element_type and grid_model_id.
    First tries to match 100% of the name.
    Second tries to match by removing spaces and replacing "+" with "##_##".

    Parameters
    ----------
    processed_n1_definition : pat.DataFrame[ContingencyMatchSchema]
        The contingencies from the file.
    all_element_names : pat.DataFrame[AllGridElementsSchema]
        The elements in the grid model.

    Returns
    -------
    pat.DataFrame[ContingencyMatchSchema]
        A DataFrame containing the matched contingencies.
    """
    processed_n1_definition = match_contingencies_column(
        processed_n1_definition=processed_n1_definition,
        all_element_names=all_element_names,
        n1_column="power_factory_grid_model_name",
        element_column="grid_model_name",
    )

    # a "+" sometimes makes problems in the name matching
    all_element_names["grid_model_name_no_space"] = (
        all_element_names["grid_model_name"].str.replace(" ", "").str.replace("+", "##_##")
    )
    processed_n1_definition["power_factory_grid_model_name_no_space"] = (
        processed_n1_definition["power_factory_grid_model_name"].str.replace(" ", "").str.replace("+", "##_##")
    )
    processed_n1_definition = match_contingencies_column(
        processed_n1_definition=processed_n1_definition,
        all_element_names=all_element_names,
        n1_column="power_factory_grid_model_name_no_space",
        element_column="grid_model_name_no_space",
    )

    processed_n1_definition.drop(
        columns=[
            "power_factory_grid_model_name_no_space",
        ],
        inplace=True,
    )
    all_element_names.drop(columns=["grid_model_name_no_space"], inplace=True)

    return processed_n1_definition

match_contingencies_with_suffix

match_contingencies_with_suffix(
    processed_n1_definition,
    all_element_names,
    grid_model_suffix,
)

Match the contingencies from the file with the elements in the grid model by name.

Matches by name and replaces the grid_model_name with power_factory_grid_model_name. Removes suffix from the grid_model_name.

PARAMETER	DESCRIPTION
processed_n1_definition	The contingencies from the file. TYPE: DataFrame[ContingencyMatchSchema]
all_element_names	The elements in the grid model. TYPE: DataFrame[AllGridElementsSchema]
grid_model_suffix	The suffixes to match the grid model names. TYPE: list[str]

RETURNS	DESCRIPTION
DataFrame[ContingencyMatchSchema]	A DataFrame containing the matched contingencies.

Source code in packages/importer_pkg/src/toop_engine_importer/contingency_from_power_factory/contingency_from_file.py

def match_contingencies_with_suffix(
    processed_n1_definition: pat.DataFrame[ContingencyMatchSchema],
    all_element_names: pat.DataFrame[AllGridElementsSchema],
    grid_model_suffix: list[str],
) -> pat.DataFrame[ContingencyMatchSchema]:
    """Match the contingencies from the file with the elements in the grid model by name.

    Matches by name and replaces the grid_model_name with power_factory_grid_model_name.
    Removes suffix from the grid_model_name.

    Parameters
    ----------
    processed_n1_definition : pat.DataFrame[ContingencyMatchSchema]
        The contingencies from the file.
    all_element_names : pat.DataFrame[AllGridElementsSchema]
        The elements in the grid model.
    grid_model_suffix : list[str]
        The suffixes to match the grid model names.

    Returns
    -------
    pat.DataFrame[ContingencyMatchSchema]
        A DataFrame containing the matched contingencies.
    """
    all_element_names["grid_model_name_suffix"] = all_element_names["grid_model_name"]
    for suffix in grid_model_suffix:
        cond_suffix = all_element_names["grid_model_name_suffix"].str.endswith(suffix)
        all_element_names.loc[cond_suffix, "grid_model_name_suffix"] = all_element_names.loc[
            cond_suffix, "grid_model_name_suffix"
        ].str[: -len(suffix)]

    processed_n1_definition = match_contingencies_column(
        processed_n1_definition=processed_n1_definition,
        all_element_names=all_element_names,
        n1_column="power_factory_grid_model_name",
        element_column="grid_model_name_suffix",
    )

    all_element_names.drop(columns=["grid_model_name_suffix"], inplace=True)

    return processed_n1_definition

match_contingencies_column

match_contingencies_column(
    processed_n1_definition,
    all_element_names,
    n1_column,
    element_column,
)

Match a column processed_n1_definition with a column from all_element_names.

This functions matches based on 100% name match and replaces the grid_model_name, element_type and grid_model_id

PARAMETER	DESCRIPTION
processed_n1_definition	The contingencies from the file. TYPE: DataFrame[ContingencyMatchSchema]
all_element_names	The elements in the grid model. TYPE: DataFrame[AllGridElementsSchema]
n1_column	The column name in processed_n1_definition. TYPE: str
element_column	The column name in all_element_names. TYPE: str

RETURNS	DESCRIPTION
DataFrame[ContingencyMatchSchema]	A DataFrame containing the matched contingencies.

Source code in packages/importer_pkg/src/toop_engine_importer/contingency_from_power_factory/contingency_from_file.py

def match_contingencies_column(
    processed_n1_definition: pat.DataFrame[ContingencyMatchSchema],
    all_element_names: pat.DataFrame[AllGridElementsSchema],
    n1_column: str,
    element_column: str,
) -> pat.DataFrame[ContingencyMatchSchema]:
    """Match a column processed_n1_definition with a column from all_element_names.

    This functions matches based on 100% name match and replaces the grid_model_name, element_type and grid_model_id

    Parameters
    ----------
    processed_n1_definition : pat.DataFrame[ContingencyMatchSchema]
        The contingencies from the file.
    all_element_names : pat.DataFrame[AllGridElementsSchema]
        The elements in the grid model.
    n1_column : str
        The column name in processed_n1_definition.
    element_column : str
        The column name in all_element_names.

    Returns
    -------
    pat.DataFrame[ContingencyMatchSchema]
        A DataFrame containing the matched contingencies.
    """
    # merge the n1_definition with all_element_names
    processed_n1_definition = processed_n1_definition.merge(
        all_element_names,
        how="left",
        left_on=n1_column,
        right_on=element_column,
        suffixes=("", "_2"),
    )
    # get new matched elements
    cond_not_matched_elements = processed_n1_definition["grid_model_name"].isna()
    cond_name_found = ~processed_n1_definition["grid_model_name_2"].isna()
    cond_replace = cond_not_matched_elements & cond_name_found
    # replace new matched elements
    processed_n1_definition.loc[cond_replace, "grid_model_name"] = processed_n1_definition.loc[
        cond_replace, "grid_model_name_2"
    ]
    processed_n1_definition.loc[cond_replace, "element_type"] = processed_n1_definition.loc[cond_replace, "element_type_2"]
    processed_n1_definition.loc[cond_replace, "grid_model_id"] = processed_n1_definition.loc[cond_replace, "grid_model_id_2"]
    processed_n1_definition.drop(columns=["grid_model_name_2", "element_type_2", "grid_model_id_2"], inplace=True)
    ContingencyMatchSchema.validate(processed_n1_definition)
    return processed_n1_definition

UCTE Toolset

toop_engine_importer.ucte_toolset

A collection of tools to work with UCTE data.

• ucte_toolset.py: Functions load, manipulate, and save UCTE data using pd.DataFrame.

Importer Pandapower

toop_engine_importer.pandapower_import

Contains functions to import data from pandapower networks to the Topology Optimizer.

__all__ module-attribute

__all__ = [
    "add_substation_column_to_bus",
    "create_virtual_slack",
    "drop_elements_connected_to_one_bus",
    "drop_unsupplied_buses",
    "fuse_closed_switches_by_bus_ids",
    "fuse_closed_switches_fast",
    "get_all_switches_from_bus_ids",
    "get_closed_switch",
    "get_coupler_types_of_substation",
    "get_indirect_connected_switch",
    "get_master_asset_topology_from_network",
    "get_station_id_list",
    "get_substation_buses_from_bus_id",
    "get_type_b_nodes",
    "move_elements_based_on_labels",
    "preprocess_net_step1",
    "preprocess_net_step2_master_asset_topology",
    "remove_out_of_service",
    "replace_zero_branches",
    "select_connected_subnet",
    "validate_asset_topology_stations",
]

get_master_asset_topology_from_network

get_master_asset_topology_from_network(
    network,
    topology_id,
    grid_model_file,
    station_id_list,
    foreign_key="equipment",
)

Return canonical asset-topology master data derived from a pandapower network.

PARAMETER	DESCRIPTION
network	Source pandapower network. TYPE: pandapowerNet
topology_id	Identifier to store on the resulting master data. TYPE: str
grid_model_file	Source grid-model file name stored in the master data. TYPE: str
station_id_list	Station definitions as lists of pandapower bus indices. TYPE: list[list[int]]
foreign_key	Column name used as the preferred human-readable identifier. TYPE: str DEFAULT: "equipment"

RETURNS	DESCRIPTION
MasterAssetTopology	Canonical master data split into structural station groups.

Source code in packages/grid_helpers_pkg/src/toop_engine_grid_helpers/pandapower/asset_topology.py

def get_master_asset_topology_from_network(
    network: pp.pandapowerNet,
    topology_id: str,
    grid_model_file: str,
    station_id_list: List[List[int]],
    foreign_key: str = "equipment",
) -> MasterAssetTopology:
    """Return canonical asset-topology master data derived from a pandapower network.

    Parameters
    ----------
    network : pp.pandapowerNet
        Source pandapower network.
    topology_id : str
        Identifier to store on the resulting master data.
    grid_model_file : str
        Source grid-model file name stored in the master data.
    station_id_list : list[list[int]]
        Station definitions as lists of pandapower bus indices.
    foreign_key : str, default="equipment"
        Column name used as the preferred human-readable identifier.

    Returns
    -------
    MasterAssetTopology
        Canonical master data split into structural station groups.
    """
    master_stations: list[MasterBusGroup] = []
    branch_assets_by_id: dict[str, BranchAsset] = {}
    injection_assets_by_id: dict[str, InjectionAsset] = {}
    asset_bays_by_id: dict[str, AssetBay] = {}

    for station_ids in station_id_list:
        structural_groups = _get_structural_station_bus_groups(list(station_ids), network)
        for group_index, structural_group in enumerate(structural_groups):
            master_station, branch_assets, injection_assets, asset_bays = _build_master_bus_group_from_station_id(
                network=network,
                station_id_list=structural_group,
                group_index=group_index,
                foreign_key=foreign_key,
            )
            master_stations.append(master_station)
            for asset in branch_assets:
                _register_unique_payload(branch_assets_by_id, asset.grid_model_id, asset, "branch asset")
            for asset in injection_assets:
                _register_unique_payload(injection_assets_by_id, asset.grid_model_id, asset, "injection asset")
            for asset_bay in asset_bays:
                if asset_bay.asset_bay_id is None:
                    continue
                _register_unique_payload(asset_bays_by_id, asset_bay.asset_bay_id, asset_bay, "asset bay")

    master_data = MasterAssetTopology(
        topology_id=topology_id,
        grid_model_file=grid_model_file,
        bus_groups=master_stations,
        branch_assets=list(branch_assets_by_id.values()),
        injection_assets=list(injection_assets_by_id.values()),
        asset_bays=list(asset_bays_by_id.values()),
    )
    validate_complete_master_asset_topology(master_data)
    return master_data

create_virtual_slack

create_virtual_slack(net)

Create a virtual slack bus for all ext_grids in the network.

PARAMETER	DESCRIPTION
net	The pandapower network to create a virtual slack for, will be modified in-place. Note: network is modified in-place. TYPE: pandapowerNet

RETURNS	DESCRIPTION
pandapowerNet	The network with a virtual slack.

Source code in packages/grid_helpers_pkg/src/toop_engine_grid_helpers/pandapower/pandapower_import_helpers.py

def create_virtual_slack(net: pp.pandapowerNet) -> None:
    """Create a virtual slack bus for all ext_grids in the network.

    Parameters
    ----------
    net: pp.pandapowerNet
        The pandapower network to create a virtual slack for, will be modified in-place.
        Note: network is modified in-place.

    Returns
    -------
    pp.pandapowerNet
        The network with a virtual slack.
    """
    if net.gen.slack.sum() <= 1:
        return
    # Create a virtual slack where all ext_grids are connected to
    virtual_slack_bus = pp.create_bus(net, vn_kv=380, in_service=True, name="virtual_slack")

    for generator in net.gen[net.gen.slack].index:
        cur_bus = net.gen.loc[generator].bus
        # Connect each gen through a trafo to the virtual slack
        pp.create_transformer_from_parameters(
            net,
            hv_bus=virtual_slack_bus,
            lv_bus=cur_bus,
            name="con_" + str(net.gen.loc[generator].name),
            sn_mva=9999,
            vn_hv_kv=net.bus.vn_kv[cur_bus],
            vn_lv_kv=net.bus.vn_kv[cur_bus],
            # shift_degree=net.ext_grid.loc[generator].va_degree,
            shift_degree=0,
            pfe_kw=1,
            i0_percent=0.1,
            vk_percent=1,
            vkr_percent=0.1,
            xn_ohm=10,
        )

    net.gen.drop(net.gen[net.gen.slack].index, inplace=True)

    pp.create_ext_grid(
        net,
        virtual_slack_bus,
        vm_pu=1,
        va_degree=0,
        in_service=True,
        name="virtual_slack",
    )

drop_elements_connected_to_one_bus

drop_elements_connected_to_one_bus(net, branch_types=None)

Drop elements connected to one bus.

• impedance -> Capacitor will end up on the same bus
• trafo3w -> edgecase: trafo3w that goes from one hv to the same level but two different busbars will end up on the same bus

PARAMETER	DESCRIPTION
net	pandapower network Note: the network is modified in place TYPE: pandapowerNet
branch_types	list of branch types to drop elements connected to one bus TYPE: list[str] DEFAULT: None

RETURNS	DESCRIPTION
None

Source code in packages/grid_helpers_pkg/src/toop_engine_grid_helpers/pandapower/pandapower_import_helpers.py

def drop_elements_connected_to_one_bus(net: pp.pandapowerNet, branch_types: Optional[list[str]] = None) -> None:
    """Drop elements connected to one bus.

    - impedance -> Capacitor will end up on the same bus
    - trafo3w -> edgecase: trafo3w that goes from one hv to the same level but two
                 different busbars will end up on the same bus

    Parameters
    ----------
    net : pp.pandapowerNet
        pandapower network
        Note: the network is modified in place
    branch_types : list[str]
        list of branch types to drop elements connected to one bus

    Returns
    -------
    None

    """
    if branch_types is None:
        branch_types = ["line", "trafo", "trafo3w", "impedance", "switch"]

    for branch_type in branch_types:
        handle_elements_connected_to_one_bus(net, branch_type)

drop_unsupplied_buses

drop_unsupplied_buses(net)

Drop all unsupplied buses from the network.

PARAMETER	DESCRIPTION
net	The pandapower network to drop unsupplied buses from, will be modified in-place. TYPE: pandapowerNet

Source code in packages/grid_helpers_pkg/src/toop_engine_grid_helpers/pandapower/pandapower_import_helpers.py

def drop_unsupplied_buses(net: pp.pandapowerNet) -> None:
    """Drop all unsupplied buses from the network.

    Parameters
    ----------
    net: pp.pandapowerNet
        The pandapower network to drop unsupplied buses from, will be modified in-place.
    """
    pp.toolbox.drop_buses(net, pp.topology.unsupplied_buses(net))
    assert len(pp.topology.unsupplied_buses(net)) == 0

fuse_closed_switches_fast

fuse_closed_switches_fast(net, switch_ids=None)

Fuse closed switches in the network by merging busbars.

This routine uses an algorithm to number each busbar and then find the lowest connected busbar iteratively. If a busbar is connected to a lower-numbered busbar, it will be re-labeled to the lower-numbered busbar. This algorithm needs as many iterations as the maximum number of hops between the lowest and highest busbar in any of the substations.

PARAMETER	DESCRIPTION
net	The pandapower network to fuse closed switches in, will be modified in-place. TYPE: pandapowerNet
switch_ids	The switch ids to fuse. If None, all closed switches are fused. TYPE: Optional[list[int]] DEFAULT: None

RETURNS	DESCRIPTION
DataFrame	The closed switches that were fused.
DataFrame	The buses that were dropped because they were relabeled to a lower-numbered busbar.

Source code in packages/grid_helpers_pkg/src/toop_engine_grid_helpers/pandapower/pandapower_import_helpers.py

def fuse_closed_switches_fast(
    net: pp.pandapowerNet,
    switch_ids: Optional[list[int]] = None,
) -> tuple[pd.DataFrame, pd.DataFrame]:
    """Fuse closed switches in the network by merging busbars.

    This routine uses an algorithm to number each busbar and then find the lowest connected busbar
    iteratively. If a busbar is connected to a lower-numbered busbar, it will be re-labeled to the
    lower-numbered busbar. This algorithm needs as many iterations as the maximum number of hops
    between the lowest and highest busbar in any of the substations.

    Parameters
    ----------
    net: pp.pandapowerNet
        The pandapower network to fuse closed switches in, will be modified in-place.
    switch_ids: list[int]
        The switch ids to fuse. If None, all closed switches are fused.

    Returns
    -------
    pd.DataFrame
        The closed switches that were fused.
    pd.DataFrame
        The buses that were dropped because they were relabeled to a lower-numbered busbar.
    """
    # Label the busbars, find the lowest index that every busbar is coupled to
    labels = np.arange(np.max(net.bus.index) + 1)
    closed_switches = net.switch[net.switch.closed & (net.switch.et == "b") & (net.switch.bus != net.switch.element)]
    if switch_ids is not None:
        closed_switches = closed_switches[closed_switches.index.isin(switch_ids)]
    while not np.array_equal(labels[closed_switches.bus.values], labels[closed_switches.element.values]):
        bus_smaller = labels[closed_switches.bus.values] < labels[closed_switches.element.values]
        element_smaller = labels[closed_switches.bus.values] > labels[closed_switches.element.values]

        # Where the element is smaller, set the bus labels to the element labels
        _, change_idx = np.unique(closed_switches.bus.values[element_smaller], return_index=True)
        labels[closed_switches.bus.values[element_smaller][change_idx]] = labels[
            closed_switches.element.values[element_smaller][change_idx]
        ]

        # Where the bus is smaller (and where the element was not already touched), set the element labels to the bus labels
        was_touched = np.isin(
            closed_switches.element.values,
            closed_switches.bus.values[element_smaller][change_idx],
        )
        cond = bus_smaller & ~was_touched
        _, change_idx = np.unique(closed_switches.element.values[cond], return_index=True)
        labels[closed_switches.element.values[cond][change_idx]] = labels[closed_switches.bus.values[cond][change_idx]]

    # Move all elements over to the lowest index busbar
    move_elements_based_on_labels(net, labels)
    # Drop all busbars that were re-labeled because they were connected to a lower-labeled bus
    buses_to_drop = net.bus[~np.isin(net.bus.index, labels)]
    switch_cond = (net.switch.et == "b") & (net.switch.bus == net.switch.element)
    switch_to_drop = net.switch[switch_cond]
    pp.toolbox.drop_elements(net, "switch", switch_to_drop.index)
    pp.toolbox.drop_buses(net, buses_to_drop.index)
    return closed_switches, buses_to_drop

move_elements_based_on_labels

move_elements_based_on_labels(net, labels)

Move all elements in the network to the lowest labeled busbar.

PARAMETER	DESCRIPTION
net	The pandapower network to move elements in, will be modified in-place. TYPE: pandapowerNet
labels	The labels of the busbars to move the elements to. TYPE: ndarray

Source code in packages/grid_helpers_pkg/src/toop_engine_grid_helpers/pandapower/pandapower_import_helpers.py

def move_elements_based_on_labels(
    net: pp.pandapowerNet,
    labels: np.ndarray,
) -> None:
    """Move all elements in the network to the lowest labeled busbar.

    Parameters
    ----------
    net: pp.pandapowerNet
        The pandapower network to move elements in, will be modified in-place.
    labels: np.ndarray
        The labels of the busbars to move the elements to.
    """
    for element, column in pp.element_bus_tuples():
        if element == "switch":
            net[element][column] = labels[net[element][column]]
            switch_cond = net[element].et == "b"
            net[element].loc[switch_cond, "element"] = labels[net[element].loc[switch_cond, "element"]]
            net[element].loc[net[element].index, "bus"] = labels[net[element].loc[net[element].index, "bus"]]
        else:
            net[element][column] = labels[net[element][column]]

remove_out_of_service

remove_out_of_service(net)

Remove all out-of-service elements from the network.

PARAMETER	DESCRIPTION
net	The pandapower network to remove out-of-service elements from, will be modified in-place. TYPE: pandapowerNet

Source code in packages/grid_helpers_pkg/src/toop_engine_grid_helpers/pandapower/pandapower_import_helpers.py

def remove_out_of_service(net: pp.pandapowerNet) -> None:
    """Remove all out-of-service elements from the network.

    Parameters
    ----------
    net: pp.pandapowerNet
        The pandapower network to remove out-of-service elements from, will be modified in-place.
    """
    for element in pp.pp_elements():
        if "bus" == element and "in_service" in net[element]:
            pp.toolbox.drop_buses(net, net[element][~net[element]["in_service"]].index)
        elif "in_service" in net[element]:
            net[element] = net[element][net[element]["in_service"]]

replace_zero_branches

replace_zero_branches(net)

Replace zero-impedance branches with switches in the network.

Some leftover lines and xwards will be bumped to a higher impedance to avoid numerical issues.

PARAMETER	DESCRIPTION
net	The pandapower network to replace zero branches in, will be modified in-place. TYPE: pandapowerNet

RETURNS	DESCRIPTION
pandapowerNet	The network with zero branches replaced.

Source code in packages/grid_helpers_pkg/src/toop_engine_grid_helpers/pandapower/pandapower_import_helpers.py

def replace_zero_branches(net: pp.pandapowerNet) -> None:
    """Replace zero-impedance branches with switches in the network.

    Some leftover lines and xwards will be bumped to a higher impedance to avoid numerical issues.

    Parameters
    ----------
    net: pp.pandapowerNet
        The pandapower network to replace zero branches in, will be modified in-place.

    Returns
    -------
    pp.pandapowerNet
        The network with zero branches replaced.
    """
    pp.toolbox.replace_zero_branches_with_switches(
        net,
        min_length_km=0.0,
        min_r_ohm_per_km=0.002,
        min_x_ohm_per_km=0.002,
        min_c_nf_per_km=0,
        min_rft_pu=0,
        min_xft_pu=0,
    )
    threshold_x_ohm = 0.001
    # net.xward.x_ohm[net.xward.x_ohm == 1e-6] = 1e-2
    net.xward.loc[net.xward.x_ohm < threshold_x_ohm, "x_ohm"] = 0.01
    zero_lines = (net.line.x_ohm_per_km * net.line.length_km) < threshold_x_ohm
    net.line.loc[zero_lines, "x_ohm_per_km"] = 0.01
    net.line.loc[zero_lines, "length_km"] = 1.0

select_connected_subnet

select_connected_subnet(net)

Select the connected subnet of the grid that has a slack and return it.

PARAMETER	DESCRIPTION
net	The pandapower network to select the connected subnet from. TYPE: pandapowerNet

RETURNS	DESCRIPTION
pandapowerNet	The connected subnet of the grid that has a slack.

Source code in packages/grid_helpers_pkg/src/toop_engine_grid_helpers/pandapower/pandapower_import_helpers.py

def select_connected_subnet(net: pp.pandapowerNet) -> pp.pandapowerNet:
    """Select the connected subnet of the grid that has a slack and return it.

    Parameters
    ----------
    net: pp.pandapowerNet
        The pandapower network to select the connected subnet from.

    Returns
    -------
    pp.pandapowerNet
        The connected subnet of the grid that has a slack.
    """
    name = net.name
    mg = pp.topology.create_nxgraph(net, respect_switches=True)

    slack_bus = net.ext_grid[net.ext_grid.in_service].bus
    if len(slack_bus) == 0:
        slack_bus = net.gen[net.gen.slack & net.gen.in_service].bus
        if len(slack_bus) == 0:
            raise ValueError("No slack bus found in the network.")
    slack_bus = slack_bus.iloc[0]

    cc = pp.topology.connected_component(mg, slack_bus)

    next_grid_buses = set(cc)
    net_new = pp.toolbox.select_subnet(
        net,
        next_grid_buses,
        include_switch_buses=True,
        include_results=False,
        keep_everything_else=True,
    )
    net_new.name = name
    return net_new

add_substation_column_to_bus

add_substation_column_to_bus(
    network,
    substation_col="substat",
    get_name_col="name",
    only_closed_switches=False,
)

Add a substation column to the bus DataFrame.

This function will go through all busbars of type 'b' and add the substation name to all buses connected to the busbar.

PARAMETER	DESCRIPTION
network	The pandapower network to add the substation column to. Note: the network will be modified in-place. TYPE: pandapowerNet
substation_col	The name of the new substation column where the value from the get_name_col is added. TYPE: Optional[str] DEFAULT: 'substat'
get_name_col	The name of the column to get the substation name from. TYPE: Optional[str] DEFAULT: 'name'
only_closed_switches	If True, only closed switches are considered. The result will lead substation naming after the the electrical voltage level. TYPE: bool DEFAULT: False

Source code in packages/importer_pkg/src/toop_engine_importer/pandapower_import/pandapower_toolset_node_breaker.py

def add_substation_column_to_bus(
    network: pp.pandapowerNet,
    substation_col: Optional[str] = "substat",
    get_name_col: Optional[str] = "name",
    only_closed_switches: bool = False,
) -> None:
    """Add a substation column to the bus DataFrame.

    This function will go through all busbars of type 'b' and add the substation name to all buses connected to the busbar.

    Parameters
    ----------
    network: pp.pandapowerNet
        The pandapower network to add the substation column to.
        Note: the network will be modified in-place.
    substation_col: Optional[str]
        The name of the new substation column where the value from the get_name_col is added.
    get_name_col: Optional[str]
        The name of the column to get the substation name from.
    only_closed_switches: bool
        If True, only closed switches are considered.
        The result will lead substation naming after the the electrical voltage level.
    """
    bus_type_b = get_type_b_nodes(network).index
    network.bus[substation_col] = ""
    found_list = []
    name_list = []
    for bus_id in bus_type_b:
        if bus_id in found_list:
            continue
        station_buses = list(get_substation_buses_from_bus_id(network, bus_id, only_closed_switches=only_closed_switches))
        station_name = str(network.bus.loc[bus_id, get_name_col])
        counter = 0
        while station_name in name_list:
            station_name = str(network.bus.loc[bus_id, get_name_col]) + f"_{counter}"
            counter += 1
        network.bus.loc[station_buses, substation_col] = station_name
        found_list.extend(station_buses)
        name_list.append(station_name)

fuse_closed_switches_by_bus_ids

fuse_closed_switches_by_bus_ids(network, switch_bus_ids)

Fuse a series of closed switches in the network by merging busbars (type b).

Note: this function expects that there are only switches between the busbars.
Warning: this function will break the model if gaps are between the buses or other elements in between.
This function will not work if you try to fuse multiple busbars,
that are not directly connected by the the switch_bus_ids.
e.g.
----busbar1----switch1----switch2---switch3----busbar2----
will result in:
----busbar1---

This will not work:
(no connection between busbar1 and busbar3 / busbar2 and 4)
----busbar1----switch1----switch2---switch3----busbar2----
----busbar3----switch4----switch5---switch6----busbar4----
call this function twice to fuse busbar2 into busbar1 and busbar4 into busbar3

PARAMETER	DESCRIPTION
network	The pandapower network to fuse closed switches in, will be modified in-place. TYPE: pandapowerNet
switch_bus_ids	The bus ids of the switches to fuse. Note: this must include the bus_id that is expected to be the final busbar. TYPE: list[int]

RETURNS	DESCRIPTION
bus_labels	An with the length of the highest bus id in the network representing the busbar index. At the index of the array(old busbar index), the new busbar index is stored. TYPE: array

Source code in packages/importer_pkg/src/toop_engine_importer/pandapower_import/pandapower_toolset_node_breaker.py

def fuse_closed_switches_by_bus_ids(network: pp.pandapowerNet, switch_bus_ids: list[int]) -> np.ndarray:
    """Fuse a series of closed switches in the network by merging busbars (type b).

    ```
    Note: this function expects that there are only switches between the busbars.
    Warning: this function will break the model if gaps are between the buses or other elements in between.
    This function will not work if you try to fuse multiple busbars,
    that are not directly connected by the the switch_bus_ids.
    e.g.
    ----busbar1----switch1----switch2---switch3----busbar2----
    will result in:
    ----busbar1---

    This will not work:
    (no connection between busbar1 and busbar3 / busbar2 and 4)
    ----busbar1----switch1----switch2---switch3----busbar2----
    ----busbar3----switch4----switch5---switch6----busbar4----
    call this function twice to fuse busbar2 into busbar1 and busbar4 into busbar3
    ```

    Parameters
    ----------
    network: pp.pandapowerNet
        The pandapower network to fuse closed switches in, will be modified in-place.
    switch_bus_ids: list[int]
        The bus ids of the switches to fuse.
        Note: this must include the bus_id that is expected to be the final busbar.

    Returns
    -------
    bus_labels: np.array
        An with the length of the highest bus id in the network representing the busbar index.
        At the index of the array(old busbar index), the new busbar index is stored.

    """
    # get a label dict for the buses
    # make sure that missing/deleted buses do not break the algorithm
    bus_labels = np.arange(np.max(network.bus.index) + 1)
    # remove duplicate bus ids -> can happen if cross couplers have only one breaker
    switch_bus_ids_pruned = list(set(switch_bus_ids))
    switch_buses = network.bus.loc[switch_bus_ids_pruned]
    switch_buses_type_b = switch_buses[switch_buses["type"] == "b"]
    if len(switch_buses_type_b) == 0:
        raise ValueError(f"No busbars found in the switch_bus_ids list {switch_bus_ids}")
    # select the first busbar of type 'b' to be the reference busbar
    for bus_id in switch_bus_ids_pruned:
        # set all busbars to the first busbar
        bus_labels[bus_id] = switch_buses_type_b.index[0]
        # Move all elements over to the lowest index busbar

    move_elements_based_on_labels(network, bus_labels)
    # Drop all busbars that were re-labeled because they were connected to a lower-labeled bus
    buses_to_drop = network.bus[~np.isin(network.bus.index, bus_labels)]
    # drop switches that are connected to one bus -> have been fused
    network["switch"] = network["switch"][network["switch"]["bus"] != network["switch"]["element"]]
    pp.toolbox.drop_buses(network, buses_to_drop.index)

    return bus_labels

get_all_switches_from_bus_ids

get_all_switches_from_bus_ids(
    network, bus_ids, only_closed_switches=True
)

Get all switches connected to a list of buses.

PARAMETER	DESCRIPTION
network	The pandapower network to get the switches from. TYPE: pandapowerNet
bus_ids	The buses to get the switches from. TYPE: list[int] | Index
only_closed_switches	If True, only closed switches are considered. TYPE: bool DEFAULT: True

RETURNS	DESCRIPTION
DataFrame	A DataFrame with all switches connected to the buses in bus_ids.

Source code in packages/importer_pkg/src/toop_engine_importer/pandapower_import/pandapower_toolset_node_breaker.py

def get_all_switches_from_bus_ids(
    network: pp.pandapowerNet, bus_ids: list[int] | pd.Index, only_closed_switches: bool = True
) -> pd.DataFrame:
    """Get all switches connected to a list of buses.

    Parameters
    ----------
    network: pp.pandapowerNet
        The pandapower network to get the switches from.
    bus_ids: list[int]
        The buses to get the switches from.
    only_closed_switches: bool
        If True, only closed switches are considered.

    Returns
    -------
    pd.DataFrame
        A DataFrame with all switches connected to the buses in bus_ids.
    """
    connected = pp.toolbox.get_connected_elements_dict(
        network,
        bus_ids,
        respect_switches=only_closed_switches,
        include_empty_lists=True,
    )
    station_switches = network.switch[network.switch.index.isin(connected["switch"])]
    return station_switches

get_closed_switch

get_closed_switch(switches, column, column_ids)

Get the closed switch based on the column and column_ids.

PARAMETER	DESCRIPTION
switches	The switches df to filter the closed switch from. TYPE: DataFrame
column	The column to filter the column_ids. e.g. foreign_id TYPE: str
column_ids	The column ids to filter the closed switch from. TYPE: Iterable[Union[str, int, float, None]]

RETURNS	DESCRIPTION
DataFrame	The closed switch filtered by the column_ids.

Source code in packages/importer_pkg/src/toop_engine_importer/pandapower_import/pandapower_toolset_node_breaker.py

def get_closed_switch(
    switches: pd.DataFrame, column: str, column_ids: Iterable[Union[str, int, float, None]]
) -> pd.DataFrame:
    """Get the closed switch based on the column and column_ids.

    Parameters
    ----------
    switches: pd.DataFrame
        The switches df to filter the closed switch from.
    column: str
        The column to filter the column_ids. e.g. foreign_id
    column_ids: list[Union[str, int, float]]
        The column ids to filter the closed switch from.

    Returns
    -------
    pd.DataFrame
        The closed switch filtered by the column_ids.
    """
    closed_switch = switches[(switches[column].isin(column_ids)) & (switches.closed)]
    return closed_switch

get_coupler_types_of_substation

get_coupler_types_of_substation(
    network, substation_bus_list, only_closed_switches=True
)

Get the cross coupler (German: Querkuppler), busbar coupler and a cross connector of a substation.

A busbar coupler is a connection between two busbars, where assets can be connected to both busbars. A cross coupler is a connection between two busbars B1 and B2, where assets A1 can not be connected to both busbars directly. Asset A1 can only be connected directly to B1 and is connected indirectly to B2 by the cross coupler. A coupler is always a disconnector (DS), a power switch (CB) and a DS in series. In unique cases, there can be two CB switches in series. A cross connector is a single disconnector between two busbars.

PARAMETER	DESCRIPTION
network	The pandapower network to get the Cross coupler/quercoupler from. TYPE: pandapowerNet
substation_bus_list	The bus list of the substation. All buses in the list represent a substation. TYPE: list[int] | Index
only_closed_switches	If True, only closed switches are considered. TYPE: bool DEFAULT: True

RETURNS

DESCRIPTION

coupler

a dictionary with 4 keys: - 1. key: "busbar_coupler_bus_ids" - 2. key: "cross_coupler_bus_ids" - 3. key: "busbar_coupler_switch_ids" - 4. key: "cross_coupler_switch_ids" bus_ids: list of bus ids representing the busbar coupler and cross coupler switch_ids: list of switch ids representing the busbar coupler and cross coupler switch_ids = [CB, DS1, DS2] Note: the switches are not filtered by open/closed. Note: if there is only one switch or two switches: switch_ids_1sw = [CB, CB, CB] switch_ids_2sw = [CB, CB, DS2] TYPE: dict[str, list[list[int]]]

Source code in packages/importer_pkg/src/toop_engine_importer/pandapower_import/pandapower_toolset_node_breaker.py

def get_coupler_types_of_substation(
    network: pp.pandapowerNet,
    substation_bus_list: list[int] | pd.Index,
    only_closed_switches: bool = True,
) -> dict[str, list[list[int]]]:
    """Get the cross coupler (German: Querkuppler),  busbar coupler and a cross connector of a substation.

    A busbar coupler is a connection between two busbars, where assets can be connected to both busbars.
    A cross coupler is a connection between two busbars B1 and B2,
    where assets A1 can not be connected to both busbars directly.
    Asset A1 can only be connected directly to B1 and is connected indirectly to B2 by the cross coupler.
    A coupler is always a disconnector (DS), a power switch (CB) and a DS in series. In unique cases, there can be
    two CB switches in series.
    A cross connector is a single disconnector between two busbars.


    Parameters
    ----------
    network: pp.pandapowerNet
        The pandapower network to get the Cross coupler/quercoupler from.
    substation_bus_list: list[int]
        The bus list of the substation.
        All buses in the list represent a substation.
    only_closed_switches: bool
        If True, only closed switches are considered.

    Returns
    -------
    coupler : dict[str, list[list[int]]]
        a dictionary with 4 keys:
        - 1. key: "busbar_coupler_bus_ids"
        - 2. key: "cross_coupler_bus_ids"
        - 3. key: "busbar_coupler_switch_ids"
        - 4. key: "cross_coupler_switch_ids"
        bus_ids: list of bus ids representing the busbar coupler and cross coupler
        switch_ids: list of switch ids representing the busbar coupler and cross coupler
            switch_ids = [CB, DS1, DS2]
        Note: the switches are not filtered by open/closed.
        Note: if there is only one switch or two switches:
            switch_ids_1sw = [CB, CB, CB]
            switch_ids_2sw = [CB, CB, DS2]
    """
    coupler = {
        "busbar_coupler_bus_ids": [],
        "cross_coupler_bus_ids": [],
        "busbar_coupler_switch_ids": [],
        "cross_coupler_switch_ids": [],
    }  # type: dict[str, list[list[int]]]
    bus_type_b = get_type_b_nodes(network, substation_bus_list)
    if len(bus_type_b) == 0 or len(bus_type_b) == 1:
        # no coupled busbars
        return coupler
    vertical_busbars = get_vertical_connected_busbars(network, substation_bus_list)
    busbar_combinations = [
        (int(bus_1), int(bus_2)) for i, bus_1 in enumerate(bus_type_b.index) for bus_2 in bus_type_b.index[i + 1 :]
    ]
    # sort by busbar coupler and cross coupler
    for bus_1, bus_2 in busbar_combinations:
        # get connection between busbars
        switches, _connection = get_connection_between_busbars(
            network=network,
            bus_1=bus_1,
            bus_2=bus_2,
            exlcude_ids=bus_type_b.index,
            only_closed_switches=only_closed_switches,
        )
        if len(switches) != 0:
            # check for parallel switches
            for cb_switch_id in switches:
                # if not consider_three_buses:
                #     cb_switch_id_list = [cb_switch_id]
                # else:
                #     cb_switch_id_list = cb_switch_id
                cb_switch_id_list = [cb_switch_id]
                power_switch = network.switch.loc[cb_switch_id_list]
                switch_buses = np.append(power_switch.element.values, power_switch.bus.values)
                ds_switch_1 = pp.toolbox.get_connecting_branches(
                    network,
                    [bus_1],
                    switch_buses,
                )
                ds_switch_2 = pp.toolbox.get_connecting_branches(
                    network,
                    [bus_2],
                    switch_buses,
                )
                if bus_1 in vertical_busbars and bus_2 in vertical_busbars[bus_1]:
                    bus_key = "busbar_coupler_bus_ids"
                    switch_key = "busbar_coupler_switch_ids"
                else:
                    bus_key = "cross_coupler_bus_ids"
                    switch_key = "cross_coupler_switch_ids"

                # handle cases with two buses
                bus_res = [
                    bus_1,
                    bus_2,
                    int(power_switch.element.values[0]),
                    int(power_switch.bus.values[0]),
                ]
                switch_res = [
                    int(cb_switch_id_list[0]),
                    int(list(ds_switch_1["switch"])[0]),  # noqa: RUF015
                    int(list(ds_switch_2["switch"])[0]),  # noqa: RUF015
                ]
                # # handle cases with three buses
                # # if consider_three_buses:
                # if len(cb_switch_id_list) > 1:
                #     switch_res.append(cb_switch_id_list[1])
                #     node_list = list(
                #         set(
                #             np.append(
                #                 power_switch.element.values, power_switch.bus.values
                #             )
                #         )
                #     )
                #     bus_res = [bus_1, bus_2] + node_list
                coupler[bus_key].append(bus_res)
                coupler[switch_key].append(switch_res)

    return coupler

get_indirect_connected_switch

get_indirect_connected_switch(
    net,
    bus_1,
    bus_2,
    only_closed_switches=True,
    consider_three_buses=False,
    exclude_buses=None,
)

Get a switch, that is indirectly connected by two buses and only by two buses.

This function will only return the indirect connection between two buses. e.g. switchB or any switch that is parallel to switchB.

busA---switchA---busB---switchB---busC---switchC---busD

Note: this function will also return an empty dict for bus1 and bus3. Note: this function will return an empty dict if e.g. switch1 & bus2 are missing.

PARAMETER	DESCRIPTION
net	The pandapower network to get the indirect connections from. TYPE: pandapowerNet
bus_1	The bus to get the indirect connections from. TYPE: int
bus_2	The bus to get the indirect connections to. TYPE: int
only_closed_switches	If True, only closed switches are considered. TYPE: bool DEFAULT: True
consider_three_buses	If True, the function will also consider three buses in between. Bus1---switch1---bus2---switch2---bus3---switch3---bus4---switch4---bus5 TYPE: bool DEFAULT: False
exclude_buses	The buses to exclude from the indirect connection. e.g. give all other busbars (type b) in the substation. TYPE: Optional[list[int] | Index] DEFAULT: None

RETURNS	DESCRIPTION
dict[str, list[int]]	A dictionary with the indirect connections from bus_1 to bus_2

RAISES	DESCRIPTION
ValueError	If the indirect connection contains more than one switch. e.g. a parallel line to the switch.

Source code in packages/importer_pkg/src/toop_engine_importer/pandapower_import/pandapower_toolset_node_breaker.py

def get_indirect_connected_switch(
    net: pp.pandapowerNet,
    bus_1: int,
    bus_2: int,
    only_closed_switches: bool = True,
    consider_three_buses: bool = False,
    exclude_buses: Optional[list[int] | pd.Index] = None,
) -> dict[str, list[int]]:
    """Get a switch, that is indirectly connected by two buses and only by two buses.

    This function will only return the indirect connection between two buses.
    e.g. switchB or any switch that is parallel to switchB.
    ```
    busA---switchA---busB---switchB---busC---switchC---busD
    ```
    Note: this function will also return an empty dict for bus1 and bus3.
    Note: this function will return an empty dict if e.g. switch1 & bus2 are missing.

    Parameters
    ----------
    net: pp.pandapowerNet
        The pandapower network to get the indirect connections from.
    bus_1: int
        The bus to get the indirect connections from.
    bus_2: int
        The bus to get the indirect connections to.
    only_closed_switches: bool
        If True, only closed switches are considered.
    consider_three_buses: bool
        If True, the function will also consider three buses in between.
        Bus1---switch1---bus2---switch2---bus3---switch3---bus4---switch4---bus5
    exclude_buses: Optional[list[int]]
        The buses to exclude from the indirect connection.
        e.g. give all other busbars (type b) in the substation.

    Returns
    -------
    dict[str, list[int]]
        A dictionary with the indirect connections from bus_1 to bus_2

    Raises
    ------
    ValueError
        If the indirect connection contains more than one switch.
        e.g. a parallel line to the switch.
    """
    if exclude_buses is None:
        exclude_buses = [bus_1, bus_2]
    bus_1_connected = list(pp.toolbox.get_connected_buses(net, [bus_1], respect_switches=only_closed_switches, consider="s"))
    bus_1_connected = [el for el in bus_1_connected if el not in exclude_buses]
    bus_2_connected = list(pp.toolbox.get_connected_buses(net, [bus_2], respect_switches=only_closed_switches, consider="s"))
    bus_2_connected = [el for el in bus_2_connected if el not in exclude_buses]

    indirect_connection = pp.toolbox.get_connecting_branches(net, bus_1_connected, bus_2_connected)
    if consider_three_buses:
        indirect_connection_3 = get_indirect_connected_switches_three_buses(
            net,
            bus_1,
            bus_2,
            bus_1_connected,
            bus_2_connected,
            only_closed_switches,
            exclude_buses,
        )
        if "switch" in indirect_connection:
            indirect_connection["switch"] = indirect_connection["switch"] | set(indirect_connection_3["switch"])
        else:
            indirect_connection["switch"] = set(indirect_connection_3["switch"])

    indirect_connection = {
        key: list(indirect_connection[key])
        for key in indirect_connection
        if len(indirect_connection[key]) > 0 or key == "switch"
    }
    # filter only closed switches in the indirect connection
    closed_switches = []
    if "switch" in indirect_connection and only_closed_switches:
        for switch_id in indirect_connection["switch"]:
            if net.switch.loc[switch_id].closed:
                closed_switches.append(switch_id)
        indirect_connection["switch"] = closed_switches
        if len(indirect_connection["switch"]) == 0:
            del indirect_connection["switch"]
    if ("switch" in indirect_connection and len(indirect_connection) != 1) or (
        "switch" not in indirect_connection and len(indirect_connection) > 0
    ):
        error_value = [f"{key!s}:{value!s}" for key, values in indirect_connection.items() for value in values]
        raise ValueError(
            f"Indirect connection between bus {bus_1} and {bus_2} must contain only switches {' '.join(error_value)}"
        )
    return indirect_connection

get_station_id_list

get_station_id_list(bus_df, substation_col='substat')

Get all station ids from the network.

This function will return all unique station ids from the network.

PARAMETER	DESCRIPTION
bus_df	The bus DataFrame to get the station ids from. e.g. pre filtered bus DataFrame with only busbars of type 'b'. TYPE: DataFrame
substation_col	The column name of the substation TYPE: str DEFAULT: 'substat'

RETURNS	DESCRIPTION
list[int]	A list of station ids in the order of the stations in the substation_col.

Source code in packages/importer_pkg/src/toop_engine_importer/pandapower_import/pandapower_toolset_node_breaker.py

def get_station_id_list(bus_df: pd.DataFrame, substation_col: str = "substat") -> list[pd.Index]:
    """Get all station ids from the network.

    This function will return all unique station ids from the network.

    Parameters
    ----------
    bus_df: pd.DataFrame
        The bus DataFrame to get the station ids from.
        e.g. pre filtered bus DataFrame with only busbars of type 'b'.
    substation_col: str
        The column name of the substation

    Returns
    -------
    list[int]
        A list of station ids in the order of the stations in the substation_col.
    """
    substation_names = bus_df[substation_col].unique()
    return [bus_df[bus_df[substation_col] == substation_name].index for substation_name in substation_names]

get_substation_buses_from_bus_id

get_substation_buses_from_bus_id(
    network, start_bus_id, only_closed_switches=False
)

Get all buses of a substation from a start bus id.

This function will return all buses that are connected to the start bus id via switches. Note: The input expects a bus ids only containing the busbars you want to get the connection for. See diagram for references. e.g:: input [BB1, BB2] -> get BC½ input [BB1, BB2, BB3, BB4] -> get BC½, BC¾, CC⅓, CC2/4

PARAMETER	DESCRIPTION
network	The pandapower network to get the substation buses from. TYPE: pandapowerNet
start_bus_id	The bus id to start the search from. TYPE: int
only_closed_switches	If True, only closed switches are considered. TYPE: bool DEFAULT: False

RETURNS	DESCRIPTION
set[int]	A set of bus ids that are connected to the start bus id.

RAISES	DESCRIPTION
RuntimeError	If the function detects an infinite loop.

Source code in packages/importer_pkg/src/toop_engine_importer/pandapower_import/pandapower_toolset_node_breaker.py

def get_substation_buses_from_bus_id(
    network: pp.pandapowerNet, start_bus_id: int, only_closed_switches: bool = False
) -> set[int]:
    """Get all buses of a substation from a start bus id.

    This function will return all buses that are connected to the start bus id via switches.
    Note: The input expects a bus ids only containing the busbars you want to get the connection for.
    See diagram for references. e.g::
    input [BB1, BB2] -> get BC1/2
    input [BB1, BB2, BB3, BB4] -> get BC1/2, BC3/4, CC1/3, CC2/4

    Parameters
    ----------
    network: pp.pandapowerNet
        The pandapower network to get the substation buses from.
    start_bus_id: int
        The bus id to start the search from.
    only_closed_switches: bool
        If True, only closed switches are considered.

    Returns
    -------
    set[int]
        A set of bus ids that are connected to the start bus id.

    Raises
    ------
    RuntimeError
        If the function detects an infinite loop.
    """
    station_buses = {start_bus_id}
    len_station = len(station_buses)
    len_update = 0
    break_counter = 0
    max_loop_count = 25
    while len_station != len_update:
        len_station = len(station_buses)
        update_bus = pp.toolbox.get_connected_buses(
            network, station_buses, consider="s", respect_switches=only_closed_switches
        )
        station_buses.update(update_bus)
        len_update = len(station_buses)
        break_counter += 1
        # maximum hops is 7 for a standard substation as drawn the module header if you start at a branch
        if break_counter > max_loop_count:
            raise RuntimeError(
                "Infinite loop detected, please check the network model. "
                + f"Substation: {network.bus.loc[start_bus_id, 'name']}, with bus_id: {start_bus_id}"
            )
    return station_buses

get_type_b_nodes

get_type_b_nodes(
    network,
    substation_bus_list=None,
    substation_column="substat",
)

Get all nodes of type 'b' (busbar) in a network or substation.

PARAMETER	DESCRIPTION
network	The pandapower network to get the busbars from. TYPE: pandapowerNet
substation_bus_list	The bus ids of the substation. TYPE: Optional[list[int] | Index] DEFAULT: None
substation_column	The column containing the substation. TYPE: str DEFAULT: 'substat'

RETURNS	DESCRIPTION
DataFrame	A DataFrame with all busbars of type 'b' in the substation.

Source code in packages/importer_pkg/src/toop_engine_importer/pandapower_import/pandapower_toolset_node_breaker.py

def get_type_b_nodes(
    network: pp.pandapowerNet, substation_bus_list: Optional[list[int] | pd.Index] = None, substation_column: str = "substat"
) -> pd.DataFrame:
    """Get all nodes of type 'b' (busbar) in a network or substation.

    Parameters
    ----------
    network: pp.pandapowerNet
        The pandapower network to get the busbars from.
    substation_bus_list: Optional[list[int] | pd.Index]
        The bus ids of the substation.
    substation_column: str
        The column containing the substation.

    Returns
    -------
    pd.DataFrame
        A DataFrame with all busbars of type 'b' in the substation.
    """
    if substation_bus_list is None:
        substation_bus_list = network.bus.index
    substation_buses = network.bus.loc[substation_bus_list]
    bus_type_b = substation_buses[substation_buses.type == "b"]
    if substation_column not in bus_type_b.columns:
        bus_type_b[substation_column] = np.nan
    no_substations_name = bus_type_b[substation_column].isna() | (bus_type_b[substation_column] == "")
    bus_type_b.loc[no_substations_name, substation_column] = bus_type_b.loc[no_substations_name].index.astype(str)
    return bus_type_b

preprocess_net_step1

preprocess_net_step1(net)

General preprocessing - e.g. a PowerFactory network may converge in AC -> change elements.

Step 1: General preprocessing - select connected subnet - Remove zero branches - remove out of service elements - handle_constant_z_load - drop elements connected to one bus - replace xward by internal elements - replace ward by internal elements - drop controler

PARAMETER	DESCRIPTION
net	pandapower network Note: the network is modified in place TYPE: pandapowerNet

RETURNS	DESCRIPTION
net	modified pandapower network TYPE: pandapowerNet

Source code in packages/importer_pkg/src/toop_engine_importer/pandapower_import/preprocessing.py

def preprocess_net_step1(net: pp.pandapowerNet) -> pp.pandapowerNet:
    """General preprocessing - e.g. a PowerFactory network may converge in AC -> change elements.

    Step 1: General preprocessing
        - select connected subnet
        - Remove zero branches
        - remove out of service elements
        - handle_constant_z_load
        - drop elements connected to one bus
        - replace xward by internal elements
        - replace ward by internal elements
        - drop controler

    Parameters
    ----------
    net : pp.pandapowerNet
        pandapower network
        Note: the network is modified in place

    Returns
    -------
    net : pp.pandapowerNet
        modified pandapower network


    """
    # preprocessing: remove zero branches, fuse closed switches, remove out of service elements
    net = select_connected_subnet(net)
    replace_zero_branches(net)
    remove_out_of_service(net)
    # sometimes if the load is 100% constan z it will not converge -> investigate, cosinder setting to 99%
    modify_constan_z_load(net)
    drop_elements_connected_to_one_bus(net)
    pp.toolbox.replace_xward_by_internal_elements(net)
    pp.toolbox.replace_ward_by_internal_elements(net)
    validate_trafo_model(net)
    if "controler" in net:
        del net["controler"]

    return net

preprocess_net_step2_master_asset_topology

preprocess_net_step2_master_asset_topology(
    network, master_data
)

Run pandapower preprocessing step 2 on canonical master data.

Source code in packages/importer_pkg/src/toop_engine_importer/pandapower_import/preprocessing.py

def preprocess_net_step2_master_asset_topology(
    network: pp.pandapowerNet,
    master_data: MasterAssetTopology,
) -> MasterAssetTopology:
    """Run pandapower preprocessing step 2 on canonical master data."""
    handle_switches(network)
    drop_elements_connected_to_one_bus(network)
    if "bus_geodata" in network:
        del network["bus_geodata"]
    old_index = pp.toolbox.create_continuous_bus_index(network, start=0, store_old_index=True)
    updated_stations: list[MasterBusGroup] = []
    for station in master_data.bus_groups:
        station_id = _get_station_index(station)
        new_id = old_index[station_id]
        current_suffix = station.bus_group_id.rsplit("_", 1)[1] if "_" in station.bus_group_id else "a"
        updated_stations.append(station.model_copy(update={"bus_group_id": f"{new_id}{SEPARATOR}bus_{current_suffix}"}))
    return master_data.model_copy(update={"bus_groups": updated_stations})

validate_asset_topology_stations

validate_asset_topology_stations(net, master_data)

Validate canonical station connection counts directly against the pandapower network.

Source code in packages/importer_pkg/src/toop_engine_importer/pandapower_import/preprocessing.py

def validate_asset_topology_stations(net: pp.pandapowerNet, master_data: MasterAssetTopology) -> None:
    """Validate canonical station connection counts directly against the pandapower network."""
    for station in master_data.bus_groups:
        s_id = _get_station_index(station)
        station_connections = [*station.branch_connections, *station.injection_connections]
        connection_dict = pp.toolbox.get_connected_elements_dict(net, [s_id])
        del connection_dict["bus"]
        len_connection = len([element for key in connection_dict for element in connection_dict[key]])
        if len_connection != len(station_connections):
            logger.warning(
                f"Station {s_id} has {len(station_connections)} assets but only "
                + f"{len_connection} connections in the network",
                **connection_dict,
            )
            for asset_connection in station_connections:
                logger.warning(
                    f"Station {s_id} with assets: {asset_connection.asset_id}",
                    asset_id=asset_connection.asset_id,
                )
            raise ValueError(
                f"Station {s_id} has {len(station_connections)} assets but only "
                + f"{len_connection} connections in the network"
            )

Importer Pypowsybl

toop_engine_importer.pypowsybl_import

Import data from PyPowSyBl networks to the Topology Optimizer.

__all__ module-attribute

__all__ = [
    "NetworkMasks",
    "PowsyblSecurityAnalysisParam",
    "PreProcessingStatistics",
    "apply_cb_lists",
    "apply_preprocessing_changes_to_network",
    "apply_white_list_to_operational_limits",
    "assign_element_id_to_cb_df",
    "convert_low_impedance_lines",
    "create_default_network_masks",
    "get_branches_df_with_element_name",
    "get_bus_breaker_master_asset_topology",
    "import_grid_with_security_analysis",
    "load_preprocessing_statistics_filesystem",
    "make_masks",
    "remove_branches_across_switch",
    "save_masks_to_files",
    "save_preprocessing_statistics_filesystem",
    "validate_network_masks",
]

PowsyblSecurityAnalysisParam

Bases: BaseModel

Contains all the parameter for a Security Analysis with pypowsybl.

single_element_contingencies_ids instance-attribute

single_element_contingencies_ids

The ids of the single element contingencies for the different element types.

The keys are the element types and the values are the ids of the elements. keys example: "dangling", "generator", "line", "switch", "tie", "transformer", "load", "custom"

current_limit_factor instance-attribute

current_limit_factor

The factor to reduce the current limit on the lines.

This factor needs to be applied before the security analysis in for current limit and after in the violation dataframe.

monitored_branches instance-attribute

monitored_branches

The branches that are monitored during the security analysis.

monitored_buses instance-attribute

monitored_buses

The buses that are monitored during the security analysis.

ac_run class-attribute instance-attribute

ac_run = True

Define load flow type.

True: run AC N-1 Analysis. False: run DC N-1 Analysis.

PreProcessingStatistics

Bases: BaseModel

Contains all the statistics of the postprocessing.

id_lists class-attribute instance-attribute

id_lists = Field(default_factory=dict)

Contains the ids of the N-1 analysis, border line currents and CB lists. keys: relevant_subs, line_for_nminus1, trafo_for_nminus1, tie_line_for_nminus1, boundary_line_for_nminus1, generator_for_nminus1, load_for_nminus1, switches_for_nminus1 white_list, black_list

import_result instance-attribute

import_result

Statistics and results from an import process.

border_current class-attribute instance-attribute

border_current = Field(default_factory=dict)

Contains the statistics of the current limit for the lines that leave the tso area.

network_changes class-attribute instance-attribute

network_changes = Field(default_factory=dict)

Contains the statistics of the changes made to the network. keys: black_list, white_list, low_impedance_lines, branches_across_switch

import_parameter class-attribute instance-attribute

import_parameter = None

Contains the statistics of the post processing.

get_bus_breaker_master_asset_topology

get_bus_breaker_master_asset_topology(
    network,
    relevant_stations,
    topology_id,
    grid_model_file=None,
)

Return canonical topology master data derived from the current bus-breaker structure.

PARAMETER	DESCRIPTION
network	Source powsybl network. TYPE: Network
relevant_stations	Relevant stations as bus ids or as a boolean mask over network.get_buses(). TYPE: Union[list[str], Bool[ndarray, ' n_buses']]
topology_id	Identifier to store on the resulting master data. TYPE: str
grid_model_file	Source grid-model file name stored in the master data. TYPE: Optional[str] DEFAULT: None

RETURNS	DESCRIPTION
MasterAssetTopology	Canonical master data grouped by structural bus-breaker station views.

Source code in packages/grid_helpers_pkg/src/toop_engine_grid_helpers/powsybl/powsybl_asset_topo.py

def get_bus_breaker_master_asset_topology(
    network: Network,
    relevant_stations: Union[list[str], Bool[np.ndarray, " n_buses"]],
    topology_id: str,
    grid_model_file: Optional[str] = None,
) -> MasterAssetTopology:
    """Return canonical topology master data derived from the current bus-breaker structure.

    Parameters
    ----------
    network : Network
        Source powsybl network.
    relevant_stations : Union[list[str], Bool[np.ndarray, " n_buses"]]
        Relevant stations as bus ids or as a boolean mask over ``network.get_buses()``.
    topology_id : str
        Identifier to store on the resulting master data.
    grid_model_file : Optional[str], optional
        Source grid-model file name stored in the master data.

    Returns
    -------
    MasterAssetTopology
        Canonical master data grouped by structural bus-breaker station views.
    """
    buses_with_substation_and_voltage, switches, dangling_lines, element_names = get_relevant_network_data(
        network=network,
        relevant_stations=relevant_stations,
    )
    master_stations: list[MasterBusGroup] = []
    topology_branch_assets: list[BranchAsset] = []
    topology_injection_assets: list[InjectionAsset] = []
    branches = network.get_branches(attributes=["voltage_level1_id", "voltage_level2_id", "bus1_id", "bus2_id"])
    for voltage_level_id, voltage_level_rows in buses_with_substation_and_voltage.groupby("voltage_level_id", sort=False):
        station_topology = network.get_bus_breaker_topology(voltage_level_id)
        structural_groups = _get_bus_breaker_structural_bus_groups(
            station_topology_buses=station_topology.buses,
            station_topology_switches=station_topology.switches,
        )
        relevant_bus_ids = {str(bus_id) for bus_id in voltage_level_rows.index}
        representative_row = voltage_level_rows.iloc[0]

        for group_index, structural_group in enumerate(structural_groups):
            station_buses = _get_bus_breaker_station_bus_info_from_group(
                station_buses=station_topology.buses,
                selected_busbar_ids=structural_group,
            )
            local_bus_ids = set(station_buses["bus_branch_bus_id"])
            if relevant_bus_ids.isdisjoint(local_bus_ids):
                continue

            coupler_elements = get_coupler_info_from_topology(station_topology.switches, switches["name"], station_buses)
            _station_elements, normalized_assets, asset_terminals, switching_matrix, asset_connectivity = (
                _get_station_asset_inputs_from_topology(
                    station_topology.elements,
                    station_buses,
                    dangling_lines,
                    element_names,
                )
            )
            (
                station_branch_assets,
                branch_terminals,
                _branch_switching_table,
                branch_connectivity,
            ) = _get_branch_station_assets_from_df(
                normalized_assets,
                asset_terminals,
                switching_matrix,
                asset_connectivity,
            )
            branch_terminals = [
                branch_terminal
                if branch_terminal is not None
                else _infer_branch_end_from_branch_table(
                    asset_grid_model_id=asset.grid_model_id,
                    station_voltage_level_id=voltage_level_id,
                    local_bus_ids=local_bus_ids,
                    branches=branches,
                )
                for asset, branch_terminal in zip(station_branch_assets, branch_terminals, strict=True)
            ]
            (
                station_injection_assets,
                injection_terminals,
                _injection_switching_table,
                injection_connectivity,
            ) = _get_injection_station_assets_from_df(
                normalized_assets,
                asset_terminals,
                switching_matrix,
                asset_connectivity,
            )

            station_busbars = get_list_of_busbars_from_df(station_buses)
            station_couplers = get_list_of_coupler_from_df(coupler_elements)

            topology_branch_assets.extend(station_branch_assets)
            topology_injection_assets.extend(station_injection_assets)
            master_stations.append(
                MasterBusGroup(
                    bus_group_id=_build_structural_station_id(voltage_level_id, group_index),
                    voltage_level_id=voltage_level_id,
                    name=representative_row.substation_id,
                    region=str(voltage_level_id)[0:2],
                    voltage_level=representative_row.nominal_v,
                    busbars=station_busbars,
                    couplers=station_couplers,
                    branch_connections=[
                        BusGroupAssetConnection(asset_id=asset.grid_model_id, branch_end=asset_terminal, asset_bay_id=None)
                        for asset, asset_terminal in zip(station_branch_assets, branch_terminals, strict=True)
                    ],
                    injection_connections=[
                        BusGroupAssetConnection(asset_id=asset.grid_model_id, branch_end=asset_terminal, asset_bay_id=None)
                        for asset, asset_terminal in zip(station_injection_assets, injection_terminals, strict=True)
                    ],
                    branch_connectivity=branch_connectivity,
                    injection_connectivity=injection_connectivity,
                )
            )

    master_data = MasterAssetTopology(
        topology_id=topology_id,
        grid_model_file=grid_model_file,
        bus_groups=master_stations,
        branch_assets=_dedupe_assets_by_id(topology_branch_assets),
        injection_assets=_dedupe_assets_by_id(topology_injection_assets),
    )
    return master_data

apply_white_list_to_operational_limits

apply_white_list_to_operational_limits(
    network, white_list_df
)

Apply the white list to the operational limits of the network.

PARAMETER	DESCRIPTION
network	The network to modify. Note: The network is modified in place. TYPE: Network
white_list_df	DataFrame with the columns "element_id", "Anfangsknoten", "Endknoten", "Auslastungsgrenze_n_0", "Auslastungsgrenze_n_1" TYPE: DataFrame

Source code in packages/importer_pkg/src/toop_engine_importer/pypowsybl_import/dacf_whitelists.py

def apply_white_list_to_operational_limits(network: Network, white_list_df: pd.DataFrame) -> None:
    """Apply the white list to the operational limits of the network.

    Parameters
    ----------
    network : Network
        The network to modify. Note: The network is modified in place.
    white_list_df : pd.DataFrame
        DataFrame with the columns "element_id", "Anfangsknoten", "Endknoten",
        "Auslastungsgrenze_n_0", "Auslastungsgrenze_n_1"

    """
    white_list_df["Auslastungsgrenze_n_0"] = white_list_df["Auslastungsgrenze_n_0"] / 100
    white_list_df["Auslastungsgrenze_n_1"] = white_list_df["Auslastungsgrenze_n_1"] / 100
    # get the current operational limits
    op_lim = network.get_operational_limits().reset_index()
    # filter the operational limits to the elements in the white list
    op_lim = op_lim[op_lim["element_id"].isin(white_list_df["element_id"].to_list())]
    # merge the white list with the operational limits -> add the "Auslastungsgrenze_n_0" and "Auslastungsgrenze_n_1" columns
    op_lim = op_lim.merge(white_list_df, how="left", left_on="element_id", right_on="element_id")
    op_lim.set_index("element_id", inplace=True)
    # remove tie lines, as they can't be set
    op_lim = op_lim[op_lim["element_type"] != "TIE_LINE"]
    # copy the operational limits for N-1 limits
    n1_limits = op_lim[op_lim["Auslastungsgrenze_n_0"] != op_lim["Auslastungsgrenze_n_1"]].copy()
    n1_limits["acceptable_duration"] = 3600
    n1_limits["name"] = OperationalLimit.DACF_PERMANENT_N_1.value
    # apply the limits to the operational limits
    n1_limits["value"] = n1_limits["value"] * n1_limits["Auslastungsgrenze_n_1"]
    op_lim["value"] = op_lim["value"] * op_lim["Auslastungsgrenze_n_0"]
    # merge the operational limits with the N-1 limits
    op_lim = pd.concat([op_lim, n1_limits]).sort_index()
    # drop white list columns, to be able to create new operational limits
    op_lim = op_lim.set_index(["side", "type", "group_name"], append=True)[["acceptable_duration", "name", "value"]]
    # drop element_type column -> deprecated
    # create the new operational limits
    network.create_operational_limits(op_lim)

assign_element_id_to_cb_df

assign_element_id_to_cb_df(
    branches_with_elementname, cb_df
)

Get the element_id for the elements in the cb_df based on the power network model.

PARAMETER	DESCRIPTION
branches_with_elementname	powsybl branches DataFrame with the columns "elementName", "bus_breaker_bus1_id", "bus_breaker_bus2_id", "voltage_level1_id", "voltage_level2_id", "pairing_key" TYPE: DataFrame
cb_df	DataFrame with the columns "Elementname", "Anfangsknoten", "Endknoten" Note: The element_id column is added to the DataFrame in place TYPE: DataFrame

RETURNS	DESCRIPTION
None

Source code in packages/importer_pkg/src/toop_engine_importer/pypowsybl_import/dacf_whitelists.py

def assign_element_id_to_cb_df(branches_with_elementname: pd.DataFrame, cb_df: pd.DataFrame) -> None:
    """Get the element_id for the elements in the cb_df based on the power network model.

    Parameters
    ----------
    branches_with_elementname : pd.DataFrame
        powsybl branches DataFrame with the columns "elementName", "bus_breaker_bus1_id",
        "bus_breaker_bus2_id", "voltage_level1_id", "voltage_level2_id", "pairing_key"
    cb_df : pd.DataFrame
        DataFrame with the columns "Elementname", "Anfangsknoten", "Endknoten"
        Note: The element_id column is added to the DataFrame in place

    Returns
    -------
    None

    """
    eight_letter_nodes = 8
    seven_letter_nodes = 7
    cb_df["element_id"] = None
    for index, row in cb_df.iterrows():
        # determine the column names for the bus ids, based on the length of the bus id given in the cb_df
        if len(row["Anfangsknoten"]) == eight_letter_nodes:
            column_start_node = "bus_breaker_bus"
        elif len(row["Anfangsknoten"]) == seven_letter_nodes:
            column_start_node = "voltage_level"
        else:
            # should this trigger an error? -> would be an error in the black/white list
            pass

        if len(row["Endknoten"]) == eight_letter_nodes:
            column_end_node = "bus_breaker_bus"
        elif len(row["Endknoten"]) == seven_letter_nodes:
            column_end_node = "voltage_level"
        else:
            # should this trigger an error? -> would be an error in the black/white list
            pass

        # search for the element in the power network model
        # search for the element name in the branches_with_elementname
        # search for "Anfangsknoten"/start node and "Endknoten"/end node in the bus ids left and right + pairing key
        condition_name = branches_with_elementname["elementName"].str.contains(row["Elementname"])
        condition_start_node_left = (branches_with_elementname[f"{column_start_node}1_id"] == row["Anfangsknoten"]) | (
            branches_with_elementname["pairing_key"] == row["Anfangsknoten"]
        )
        condition_start_node_right = (branches_with_elementname[f"{column_start_node}2_id"] == row["Anfangsknoten"]) | (
            branches_with_elementname["pairing_key"] == row["Anfangsknoten"]
        )
        condition_end_node_left = (branches_with_elementname[f"{column_end_node}1_id"] == row["Endknoten"]) | (
            branches_with_elementname["pairing_key"] == row["Endknoten"]
        )
        condition_end_node_right = (branches_with_elementname[f"{column_end_node}2_id"] == row["Endknoten"]) | (
            branches_with_elementname["pairing_key"] == row["Endknoten"]
        )
        condition_key_order1 = condition_start_node_left & condition_end_node_right
        condition_key_order2 = condition_start_node_right & condition_end_node_left

        # apply the conditions to the branches_with_elementname DataFrame
        found_list = branches_with_elementname[
            condition_name & (condition_key_order1 | condition_key_order2)
        ].index.to_list()
        # check if only one element was found -> add to the cb_df
        if len(found_list) == 1:
            cb_df.at[index, "element_id"] = found_list[0]
        # if more than one/no element was found -> check if the id is in the "bus_breaker_bus" column
        # this often happens for TWO_WINDINGS_TRANSFORMER elements, where powsybl creates it's own id for the voltage level
        # this could be done immediately, but it would be a lot slower
        else:
            column_start_node = "bus_breaker_bus"
            column_end_node = "bus_breaker_bus"
            condition_name = branches_with_elementname["elementName"].str.contains(row["Elementname"])
            condition_start_node_left = (
                branches_with_elementname[f"{column_start_node}1_id"].str.contains(row["Anfangsknoten"])
            ) | (branches_with_elementname["pairing_key"] == row["Anfangsknoten"])
            condition_start_node_right = (
                branches_with_elementname[f"{column_start_node}2_id"].str.contains(row["Anfangsknoten"])
            ) | (branches_with_elementname["pairing_key"] == row["Anfangsknoten"])
            condition_end_node_left = (
                branches_with_elementname[f"{column_end_node}1_id"].str.contains(row["Endknoten"])
            ) | (branches_with_elementname["pairing_key"] == row["Endknoten"])
            condition_end_node_right = (
                branches_with_elementname[f"{column_end_node}2_id"].str.contains(row["Endknoten"])
            ) | (branches_with_elementname["pairing_key"] == row["Endknoten"])
            condition_key_order1 = condition_start_node_left & condition_end_node_right
            condition_key_order2 = condition_start_node_right & condition_end_node_left
            found_list = branches_with_elementname[
                condition_name & (condition_key_order1 | condition_key_order2)
            ].index.to_list()
            if len(found_list) == 1:
                cb_df.at[index, "element_id"] = found_list[0]

apply_cb_lists

apply_cb_lists(
    network,
    statistics,
    white_list_file,
    black_list_file,
    fs,
)

Run the black or white list to the powsybl network.

PARAMETER	DESCRIPTION
network	The network to modify. Note: The network is modified in place. TYPE: Network
statistics	The statistics to fill with the id lists of the black and white list Note: The statistics are modified in place. TYPE: PreProcessingStatistics
white_list_file	The path to the white list file, if None, no white list is applied. TYPE: str | Path | None
black_list_file	The path to the black list file, if None, no black list is applied. TYPE: str | Path | None
fs	The filesystem to use to read the files. TYPE: AbstractFileSystem

RETURNS	DESCRIPTION
statistics	The statistics with the id lists of the black and white list TYPE: PreProcessingStatistics

Source code in packages/importer_pkg/src/toop_engine_importer/pypowsybl_import/network_analysis.py

def apply_cb_lists(
    network: Network,
    statistics: PreProcessingStatistics,
    white_list_file: str | Path | None,
    black_list_file: str | Path | None,
    fs: AbstractFileSystem,
) -> PreProcessingStatistics:
    """Run the black or white list to the powsybl network.

    Parameters
    ----------
    network : Network
        The network to modify. Note: The network is modified in place.
    statistics : PreProcessingStatistics
        The statistics to fill with the id lists of the black and white list
        Note: The statistics are modified in place.
    white_list_file : str | Path | None
        The path to the white list file, if None, no white list is applied.
    black_list_file : str | Path | None
        The path to the black list file, if None, no black list is applied.
    fs : AbstractFileSystem
        The filesystem to use to read the files.

    Returns
    -------
    statistics: PreProcessingStatistics
        The statistics with the id lists of the black and white list

    """
    branches_with_elementname = get_branches_df_with_element_name(network)
    branches_with_elementname["pairing_key"] = branches_with_elementname["pairing_key"].str[0:7]
    # get only the rows needed
    branches_with_elementname = branches_with_elementname[
        powsybl_masks.get_mask_for_area_codes(branches_with_elementname, ["D"], "voltage_level1_id", "voltage_level2_id")
    ]
    op_lim = network.get_operational_limits(attributes=[]).index.get_level_values("element_id").to_list()
    branches_with_elementname = branches_with_elementname[branches_with_elementname.index.isin(op_lim)]

    if white_list_file is not None:
        with fs.open(str(white_list_file), "r") as f:
            white_list_df = pd.read_csv(f, delimiter=";").fillna("")
        dacf_whitelists.assign_element_id_to_cb_df(cb_df=white_list_df, branches_with_elementname=branches_with_elementname)
        statistics.import_result.n_white_list = len(white_list_df)
        white_list_df = white_list_df[white_list_df["element_id"].notnull()]
        apply_white_list_to_operational_limits(network, white_list_df)
        statistics.id_lists["white_list"] = white_list_df["element_id"].to_list()
        statistics.import_result.n_white_list_applied = len(white_list_df["element_id"])
    else:
        statistics.id_lists["white_list"] = []
    if black_list_file is not None:
        with fs.open(str(black_list_file), "r") as f:
            black_list_df = pd.read_csv(f, delimiter=";").fillna("")
        dacf_whitelists.assign_element_id_to_cb_df(cb_df=black_list_df, branches_with_elementname=branches_with_elementname)
        statistics.import_result.n_black_list = len(black_list_df)
        black_list_df = black_list_df[black_list_df["element_id"].notnull()]
        statistics.id_lists["black_list"] = black_list_df["element_id"].to_list()
        statistics.import_result.n_black_list_applied = len(black_list_df["element_id"])
    else:
        statistics.id_lists["black_list"] = []
    return statistics

convert_low_impedance_lines

convert_low_impedance_lines(
    net, voltage_level_prefix, x_threshold_line=0.05
)

Convert all lines in the same voltage level with very low impedance to breakers.

PARAMETER	DESCRIPTION
net	The network to modify. Note: This function modifies the network in place. TYPE: Network
voltage_level_prefix	The prefix of the voltage level to consider. TYPE: str
x_threshold_line	The threshold for x, everything below will be converted. TYPE: float DEFAULT: 0.05

RETURNS	DESCRIPTION
low_impedance_lines	The lines that were converted to breakers. TYPE: DataFrame

Source code in packages/importer_pkg/src/toop_engine_importer/pypowsybl_import/network_analysis.py

def convert_low_impedance_lines(net: Network, voltage_level_prefix: str, x_threshold_line: float = 0.05) -> pd.DataFrame:
    """Convert all lines in the same voltage level with very low impedance to breakers.

    Parameters
    ----------
    net: Network
        The network to modify. Note: This function modifies the network in place.
    voltage_level_prefix: str
        The prefix of the voltage level to consider.
    x_threshold_line: float
        The threshold for x, everything below will be converted.

    Returns
    -------
    low_impedance_lines: pd.DataFrame
        The lines that were converted to breakers.

    """
    lines = net.get_lines(all_attributes=True)
    low_impedance_lines = lines[
        (lines["voltage_level1_id"] == lines["voltage_level2_id"])
        & (lines["x"] <= x_threshold_line)
        & (lines["voltage_level1_id"].str.startswith(voltage_level_prefix))
        & (lines["connected1"] & lines["connected2"])
    ]
    net.remove_elements(low_impedance_lines.index)
    low_impedance_lines = low_impedance_lines[
        [
            "bus_breaker_bus1_id",
            "bus_breaker_bus2_id",
            "elementName",
            "voltage_level1_id",
        ]
    ].rename(
        columns={
            "bus_breaker_bus1_id": "bus1_id",
            "bus_breaker_bus2_id": "bus2_id",
            "elementName": "name",
            "voltage_level1_id": "voltage_level_id",
        }
    )
    low_impedance_lines["kind"] = "BREAKER"
    low_impedance_lines["open"] = False
    low_impedance_lines["retained"] = True
    net.create_switches(low_impedance_lines)
    return low_impedance_lines

get_branches_df_with_element_name

get_branches_df_with_element_name(network)

Get the branches with the element name.

PARAMETER	DESCRIPTION
network	The network object TYPE: Network

RETURNS	DESCRIPTION
DataFrame	The branches with the element name

Source code in packages/importer_pkg/src/toop_engine_importer/pypowsybl_import/network_analysis.py

def get_branches_df_with_element_name(network: Network) -> pd.DataFrame:
    """Get the branches with the element name.

    Parameters
    ----------
    network : Network
        The network object

    Returns
    -------
    pd.DataFrame
        The branches with the element name

    """
    branches = network.get_branches(all_attributes=True)
    lines = network.get_lines(all_attributes=True)["elementName"]
    trafos = network.get_2_windings_transformers(all_attributes=True)["elementName"]
    tie_lines = network.get_tie_lines(all_attributes=True)[["elementName_1", "elementName_2", "pairing_key"]]
    tie_lines["elementName"] = tie_lines["elementName_1"] + " + " + tie_lines["elementName_2"]
    tie_lines = tie_lines[["elementName", "pairing_key"]]

    branches = branches.merge(lines, how="left", on="id", suffixes=("", "_1"))
    branches = branches.merge(trafos, how="left", on="id", suffixes=("", "_2"))
    branches = branches.merge(tie_lines, how="left", on="id", suffixes=("", "_3"))
    branches["elementName"] = branches["elementName"].combine_first(branches["elementName_2"])
    branches["elementName"] = branches["elementName"].combine_first(branches["elementName_3"])
    branches = branches.drop(columns=["elementName_2", "elementName_3"])
    return branches

remove_branches_across_switch

remove_branches_across_switch(net)

Remove all branches that span across a closed switch, i.e. have the same from+to bus.

PARAMETER	DESCRIPTION
net	The network to modify. Note: This function modifies the network in place. TYPE: Network

RETURNS	DESCRIPTION
to_remove	The branches that were removed. TYPE: DataFrame

Source code in packages/importer_pkg/src/toop_engine_importer/pypowsybl_import/network_analysis.py

def remove_branches_across_switch(net: Network) -> pd.DataFrame:
    """Remove all branches that span across a closed switch, i.e. have the same from+to bus.

    Parameters
    ----------
    net: Network
        The network to modify. Note: This function modifies the network in place.

    Returns
    -------
    to_remove: pd.DataFrame
        The branches that were removed.

    """
    # remove branches that span across a closed switch
    # the bus1_id == bus2_id, in case the branch is a closed switch
    # in case no switch is between the buses, bus1_id != bus2_id
    # -> a line between two buses is not removed, if there is no switch between them
    to_remove = net.get_branches()[
        (net.get_branches()["bus1_id"] == net.get_branches()["bus2_id"])
        & (net.get_branches()["connected1"] & net.get_branches()["connected2"])
    ]
    net.remove_elements(to_remove.index)
    return to_remove

make_masks

make_masks(
    network,
    slack_id,
    importer_parameters,
    filesystem=None,
    blacklisted_ids=None,
)

Create all masks for the network, depending on the import parameters.

PARAMETER	DESCRIPTION
network	The network to get the masks for. TYPE: Network
slack_id	The id of the slack bus in the network. This is needed to exclude the slack bus from the relevant_subs mask. TYPE: str
importer_parameters	The import parameters including control_area, nminus1_area, cutoff_voltage Optional: border_line_factors, border_line_weight, dso_trafo_factors, dso_trafo_weight TYPE: Union[UcteImporterParameters, CgmesImporterParameters]
filesystem	The filesystem to use for loading the contingency lists from. If not provided, the local filesystem is used. TYPE: AbstractFileSystem DEFAULT: None
blacklisted_ids	The ids of the branche that are blacklisted. TYPE: list[str] | None DEFAULT: None

RETURNS	DESCRIPTION
network_masks	The masks for the network. TYPE: NetworkMasks

Source code in packages/importer_pkg/src/toop_engine_importer/pypowsybl_import/powsybl_masks.py

def make_masks(
    network: Network,
    slack_id: str,
    importer_parameters: Union[UcteImporterParameters, CgmesImporterParameters],
    filesystem: AbstractFileSystem = None,
    blacklisted_ids: list[str] | None = None,
) -> NetworkMasks:
    """Create all masks for the network, depending on the import parameters.

    Parameters
    ----------
    network: Network
        The network to get the masks for.
    slack_id: str
        The id of the slack bus in the network. This is needed to exclude the slack bus from the relevant_subs mask.
    importer_parameters: Union[UcteImporterParameters, CgmesImporterParameters]
        The import parameters including control_area, nminus1_area, cutoff_voltage
        Optional: border_line_factors, border_line_weight, dso_trafo_factors, dso_trafo_weight
    filesystem: AbstractFileSystem
        The filesystem to use for loading the contingency lists from. If not provided, the local filesystem is used.
    blacklisted_ids: list[str] | None
        The ids of the branche that are blacklisted.

    Returns
    -------
    network_masks: NetworkMasks
        The masks for the network.
    """
    if filesystem is None:
        filesystem = LocalFileSystem()
    if blacklisted_ids is None:
        blacklisted_ids = []
    default_masks = create_default_network_masks(network)

    network_masks = update_line_masks(
        default_masks,
        network,
        importer_parameters,
        blacklisted_ids,
    )
    network_masks = update_trafo_masks(
        network_masks,
        network,
        importer_parameters,
        blacklisted_ids,
    )
    network_masks = update_tie_and_dangling_line_masks(network_masks, network, importer_parameters, blacklisted_ids)
    network_masks = update_load_and_generation_masks(network_masks, network, importer_parameters, blacklisted_ids)
    network_masks = update_switch_masks(network_masks, network, importer_parameters, blacklisted_ids)
    network_masks = update_bus_masks(
        network_masks,
        network,
        importer_parameters,
        blacklisted_ids,
    )
    network_masks = remove_slack_from_relevant_subs(network_masks, network, slack_id=slack_id)

    if importer_parameters.contingency_list_file is not None:
        if importer_parameters.schema_format == "ContingencyImportSchemaPowerFactory":
            network_masks = update_masks_from_power_factory_contingency_list_file(
                network_masks, network, importer_parameters, filesystem=filesystem
            )
        elif importer_parameters.schema_format == "ContingencyImportSchema":
            network_masks = update_masks_from_contingency_list_file(
                network_masks, network, importer_parameters, filesystem=filesystem
            )
        else:
            logger.warning(f"Contingency list processing for {importer_parameters.ingress_id} is not implemented yet.")
    if not validate_network_masks(network_masks, default_masks):
        raise RuntimeError("Network masks are not created correctly.")

    network_masks = remove_slack_busbar_sections(network_masks, network, slack_id=slack_id)

    return network_masks

save_masks_to_files

save_masks_to_files(network_masks, data_folder)

Save the network masks to files.

PARAMETER	DESCRIPTION
network_masks	The network masks to save. TYPE: NetworkMasks
data_folder	The folder to save the masks to. TYPE: Path

Source code in packages/importer_pkg/src/toop_engine_importer/pypowsybl_import/powsybl_masks.py

def save_masks_to_files(network_masks: NetworkMasks, data_folder: Path) -> None:
    """Save the network masks to files.

    Parameters
    ----------
    network_masks: NetworkMasks
        The network masks to save.
    data_folder: Path
        The folder to save the masks to.
    """
    save_masks_to_filesystem(network_masks, data_folder, filesystem=LocalFileSystem())

validate_network_masks

validate_network_masks(network_masks, default_mask)

Validate if the network masks are created correctly.

PARAMETER	DESCRIPTION
network_masks	The network masks to validate. TYPE: NetworkMasks
default_mask	The default network masks to validate against. TYPE: NetworkMasks

RETURNS	DESCRIPTION
bool	True if the network masks are created correctly, False otherwise.

Source code in packages/importer_pkg/src/toop_engine_importer/pypowsybl_import/powsybl_masks.py

def validate_network_masks(network_masks: NetworkMasks, default_mask: NetworkMasks) -> bool:
    """Validate if the network masks are created correctly.

    Parameters
    ----------
    network_masks: NetworkMasks
        The network masks to validate.
    default_mask: NetworkMasks
        The default network masks to validate against.

    Returns
    -------
    bool
        True if the network masks are created correctly, False otherwise.

    """
    if not isinstance(network_masks, NetworkMasks):
        logger.warning("network_masks are not of type NetworkMasks.")
        return False
    for mask_key, mask in asdict(network_masks).items():
        if not isinstance(mask, np.ndarray):
            logger.warning(f"Mask {mask_key} is not a numpy array.")
            return False
        if not mask.shape == asdict(default_mask)[mask_key].shape:
            logger.warning(
                f"Shape of mask {mask_key} is not correct. got: "
                + f"{mask.shape}, expected: {asdict(default_mask)[mask_key].shape}"
            )
            return False
        if mask.dtype != asdict(default_mask)[mask_key].dtype:
            logger.warning(
                f"Dtype of mask {mask_key} is not correct. got: "
                + f"{mask.dtype}, expected: {asdict(default_mask)[mask_key].dtype}"
            )
            return False
    for element_type in ["line", "trafo", "tie_line", "switch"]:
        overlap = getattr(network_masks, f"{element_type}_for_optimized") & getattr(
            network_masks, f"{element_type}_for_non_worsening"
        )
        if overlap.any():
            logger.warning(
                f"{element_type} elements must be either optimized or non-worsening, not both. "
                f"{int(overlap.sum())} elements are in both masks."
            )
            return False
    return True

apply_preprocessing_changes_to_network

apply_preprocessing_changes_to_network(
    network, statistics, status_update_fn=None
)

Apply the default changes to the network.

These changes include: - removing low impedance lines - removing branches across switches

PARAMETER	DESCRIPTION
network	The network to apply the changes to. Note: This function modifies the network in place. TYPE: Network
statistics	The statistics of the preprocessing. Note: This function modifies the statistics in place. TYPE: PreProcessingStatistics
status_update_fn	A function to call to signal progress in the preprocessing pipeline. Takes a stage and an optional message as parameters TYPE: Optional[StatusUpdateFn] DEFAULT: None

Source code in packages/importer_pkg/src/toop_engine_importer/pypowsybl_import/preprocessing.py

def apply_preprocessing_changes_to_network(
    network: Network,
    statistics: PreProcessingStatistics,
    status_update_fn: Optional[StatusUpdateFn] = None,
) -> None:
    """Apply the default changes to the network.

    These changes include:
    - removing low impedance lines
    - removing branches across switches

    Parameters
    ----------
    network: Network
        The network to apply the changes to.
        Note: This function modifies the network in place.
    statistics: PreprocessingStatistics
        The statistics of the preprocessing.
        Note: This function modifies the statistics in place.
    status_update_fn: Optional[StatusUpdateFn]
        A function to call to signal progress in the preprocessing pipeline. Takes a stage and an
        optional message as parameters

    """
    if status_update_fn is None:
        status_update_fn = empty_status_update_fn
    status_update_fn("modify_low_impedance_lines", "Converting low impedance lines to breakers")
    low_impedance_lines = network_analysis.convert_low_impedance_lines(network, "D8")
    statistics.import_result.n_low_impedance_lines = len(low_impedance_lines)
    statistics.network_changes["low_impedance_lines"] = low_impedance_lines.index.to_list()

    status_update_fn("modify_branches_over_switches", "Removing branches across switches")
    branches_across_switch = network_analysis.remove_branches_across_switch(network)
    statistics.import_result.n_branch_across_switch = len(branches_across_switch)
    statistics.network_changes["branches_across_switch"] = branches_across_switch.index.to_list()

import_grid_with_security_analysis

import_grid_with_security_analysis(
    importer_parameters,
    status_update_fn=empty_status_update_fn,
    processed_gridfile_fs=None,
    unprocessed_gridfile_fs=None,
    *,
    job_id="",
)

Import the grid file and run its AC security analysis, which sets the artificial limits.

Saves data and network to the output folder. The security analysis runs once. Its results, with the artificial limits attached, are returned as the initial loadflow, so no second analysis is needed.

PARAMETER	DESCRIPTION
importer_parameters	Parameters that are required to import the data from a UCTE or CGMES file. This will utilize powsybl and the powsybl backend to the loadflow solver TYPE: BaseImporterParameters
status_update_fn	A function to call to signal progress in the preprocessing pipeline. Takes a stage, an optional message and network stats as parameters TYPE: StatusUpdateFn DEFAULT: empty_status_update_fn
processed_gridfile_fs	A filesystem where the processed gridfiles are stored. If None, the local filesystem is used TYPE: Optional[AbstractFileSystem] DEFAULT: None
unprocessed_gridfile_fs	A filesystem where the unprocessed gridfiles are stored. If None, the local filesystem is used. TYPE: Optional[AbstractFileSystem] DEFAULT: None
job_id	The job id stored with the initial loadflow. TYPE: str DEFAULT: ''

RETURNS	DESCRIPTION
ImportedGridModel	The statistics of the import, the initial loadflow and its metrics.

RAISES	DESCRIPTION
BaseCaseNotConvergedError	If the base case loadflow does not converge, before or after the preprocessing changes to the network.

Source code in packages/importer_pkg/src/toop_engine_importer/pypowsybl_import/preprocessing.py

def import_grid_with_security_analysis(  # noqa: PLR0915
    importer_parameters: BaseImporterParameters,
    status_update_fn: StatusUpdateFn = empty_status_update_fn,
    processed_gridfile_fs: Optional[AbstractFileSystem] = None,
    unprocessed_gridfile_fs: Optional[AbstractFileSystem] = None,
    *,
    job_id: str = "",
) -> ImportedGridModel:
    """Import the grid file and run its AC security analysis, which sets the artificial limits.

    Saves data and network to the output folder. The security analysis runs once. Its results, with the artificial
    limits attached, are returned as the initial loadflow, so no second analysis is needed.

    Parameters
    ----------
    importer_parameters: BaseImporterParameters
        Parameters that are required to import the data from a UCTE or CGMES file. This will utilize
        powsybl and the powsybl backend to the loadflow solver
    status_update_fn: StatusUpdateFn
        A function to call to signal progress in the preprocessing pipeline. Takes a stage, an
        optional message and network stats as parameters
    processed_gridfile_fs: Optional[AbstractFileSystem]
        A filesystem where the processed gridfiles are stored. If None, the local filesystem is used
    unprocessed_gridfile_fs: Optional[AbstractFileSystem]
        A filesystem where the unprocessed gridfiles are stored. If None, the local filesystem is used.
    job_id: str
        The job id stored with the initial loadflow.

    Returns
    -------
    ImportedGridModel
        The statistics of the import, the initial loadflow and its metrics.

    Raises
    ------
    BaseCaseNotConvergedError
        If the base case loadflow does not converge, before or after the preprocessing changes to the network.
    """
    if unprocessed_gridfile_fs is None:
        unprocessed_gridfile_fs = LocalFileSystem()
    if processed_gridfile_fs is None:
        processed_gridfile_fs = LocalFileSystem()
    network, input_nminus1_definition = load_and_prepare_network(
        importer_parameters=importer_parameters,
        processed_gridfile_fs=processed_gridfile_fs,
        unprocessed_gridfile_fs=unprocessed_gridfile_fs,
        status_update_fn=status_update_fn,
    )

    # Iterate over Loadflow parameters and voltage initialization methods to find a converging loadflow.
    # This is necessary because some grid files do not converge with the
    # default loadflow parameters and voltage initialization method.

    statistics = PreProcessingStatistics(
        import_result=ImportResult(data_folder=importer_parameters.data_folder, grid_type=importer_parameters.data_type),
        import_parameter=importer_parameters,
    )

    # a loadflow is needed for the network set_tie_line_boundary_equivalents and later for the reduction
    lf_params, main_result = _run_initial_loadflow(importer_parameters, network, unprocessed_gridfile_fs)
    if main_result.status != pypowsybl.loadflow.ComponentStatus.CONVERGED:
        raise BaseCaseNotConvergedError("The base case loadflow did not converge, preprocessing stopped.")

    # set_tie_line_boundary_equivalents
    # sets the p0 and q0 of the boundary lines to match the actual flow over the tie line
    # This is needed if the grid is reduced and a tie line is removed -> wrong loadflow if the p0 and q0 is not set
    network_analysis.set_tie_line_boundary_equivalents(net=network)

    if importer_parameters.network_reduction_voltage_level_range >= 0:
        status_update_fn("reduce_network_to_view_area", "Reducing network to view area")
        reduce_network_based_on_area_settings(
            net=network,
            importer_parameters=importer_parameters,
            extra_voltage_level_ids=get_input_nminus1_definition_voltage_level_ids(network, input_nminus1_definition),
        )

    status_update_fn("apply_cb_list", "Applying Whitelists")
    if importer_parameters.data_type == "ucte":
        # TODO: move to UCTE Toolset after all PRs are merged
        apply_preprocessing_changes_to_network(
            network=network,
            statistics=statistics,
            status_update_fn=status_update_fn,
        )

        # apply black and white list
        statistics = network_analysis.apply_cb_lists(
            network=network,
            statistics=statistics,
            white_list_file=importer_parameters.white_list_file,
            black_list_file=importer_parameters.black_list_file,
            fs=unprocessed_gridfile_fs,
        )
    elif importer_parameters.data_type == "cgmes":
        statistics = network_analysis.apply_cb_lists_cgmes(
            statistics=statistics,
            white_list_file=importer_parameters.white_list_file,
            ignore_list_file=importer_parameters.ignore_list_file,
            filesystem=unprocessed_gridfile_fs,
        )

    # Save and reload Network due to powsybl changing order during save
    grid_file_path = importer_parameters.data_folder / PREPROCESSING_PATHS["grid_file_path_powsybl"]
    save_powsybl_to_fs(
        network,
        filesystem=processed_gridfile_fs,
        file_path=grid_file_path,
    )

    # Reload Network because powsybl likes to change order during save
    network = load_powsybl_from_fs(
        filesystem=processed_gridfile_fs,
        file_path=grid_file_path,
    )
    lf_params, main_result = _run_initial_loadflow(importer_parameters, network, unprocessed_gridfile_fs)
    if main_result.status != pypowsybl.loadflow.ComponentStatus.CONVERGED:
        raise BaseCaseNotConvergedError(
            "The base case loadflow did not converge after the preprocessing changes, preprocessing stopped."
        )
    save_lf_params_to_fs(
        lf_params=lf_params,
        filesystem=processed_gridfile_fs,
        file_path=importer_parameters.data_folder / PREPROCESSING_PATHS["loadflow_parameters_file_path"],
    )

    # get N-1 masks
    status_update_fn("get_masks", "Creating Network Masks")
    network_masks = compute_network_masks(
        importer_parameters,
        processed_gridfile_fs,
        unprocessed_gridfile_fs,
        network,
        statistics,
    )

    if (
        importer_parameters.area_settings.dso_trafo_factors is not None
        or importer_parameters.area_settings.border_line_factors is not None
    ):
        status_update_fn("cross_border_current", "Setting cross border current limit")
        create_new_border_limits(network, network_masks, importer_parameters)
        # save new border limits

    status_update_fn("get_topology_model", "Creating canonical asset-topology master data")
    topology_master_data = get_master_asset_topology_artifact(
        network,
        network_masks,
        importer_parameters,
    )
    fill_statistics_for_network_masks(network=network, statistics=statistics, network_masks=network_masks)

    save_masks_to_filesystem(
        data_folder=importer_parameters.data_folder, network_masks=network_masks, filesystem=processed_gridfile_fs
    )

    # get nminus1 definition
    nminus1_definition = create_nminus1_definition(network, network_masks, topology_master_data, input_nminus1_definition)
    fill_statistics_for_nminus1_definition(statistics=statistics, nminus1_definition=nminus1_definition)
    save_pydantic_model_fs(
        filesystem=processed_gridfile_fs,
        file_path=importer_parameters.data_folder / PREPROCESSING_PATHS["nminus1_definition_file_path"],
        pydantic_model=nminus1_definition,
    )

    if importer_parameters.create_artificial_limit_when_missing_permanent:
        # Before the analysis, as the loading of a branch is measured against its permanent limit
        add_default_permanent_limits(
            network,
            [element.id for element in nminus1_definition.monitored_elements if element.kind == "branch"],
            importer_parameters.default_permanent,
        )

    status_update_fn("security_analysis", "Running the AC security analysis, stored as the initial loadflow")
    security_analysis_results = run_contingency_analysis_powsybl(
        net=network,
        n_minus_1_definition=nminus1_definition,
        job_id=job_id,
        timestep=0,
        method="ac",
        polars=True,
        lf_params=lf_params,
        n_processes=importer_parameters.n_contingency_worker,
        batch_size=importer_parameters.contingency_batch_size,
    )
    assert isinstance(security_analysis_results, LoadflowResultsPolars), (
        "Security analysis did not return LoadflowResultsPolars"
    )
    status_update_fn("set_artificial_limits", "Setting the optimization limits of the monitored branches")
    initial_loadflow = attach_artificial_limits_to_results(
        network=network,
        nminus1_definition=nminus1_definition,
        security_analysis_results=security_analysis_results,
        double_limits=importer_parameters.double_limits,
    )
    initial_metrics = compute_initial_metrics(initial_loadflow, nminus1_definition)

    status_update_fn("set_artificial_dc_limits", "Setting the DC optimization limits of the monitored branches")
    dc_metrics = compute_dc_artificial_limit_metrics(
        network=network,
        nminus1_definition=nminus1_definition,
        double_limits=importer_parameters.double_limits,
        lf_params=lf_params,
        job_id=job_id,
        n_processes=importer_parameters.n_contingency_worker,
        batch_size=importer_parameters.contingency_batch_size,
    )
    fill_statistics_for_overload_energy(statistics=statistics, metrics=initial_metrics, dc_metrics=dc_metrics)

    save_powsybl_to_fs(
        network,
        filesystem=processed_gridfile_fs,
        file_path=grid_file_path,
    )

    save_preprocessing_statistics_filesystem(
        statistics=statistics,
        file_path=importer_parameters.data_folder / PREPROCESSING_PATHS["importer_auxiliary_file_path"],
        filesystem=processed_gridfile_fs,
    )

    save_pydantic_model_fs(
        filesystem=processed_gridfile_fs,
        file_path=importer_parameters.data_folder / PREPROCESSING_PATHS["asset_topology_master_data_file_path"],
        pydantic_model=topology_master_data,
        indent=4,
    )
    return ImportedGridModel(
        import_result=statistics.import_result,
        initial_loadflow=initial_loadflow,
        initial_metrics=initial_metrics,
    )

load_preprocessing_statistics_filesystem

load_preprocessing_statistics_filesystem(
    file_path, filesystem
)

Load the preprocessing statistics from the file.

PARAMETER	DESCRIPTION
file_path	The file to load the preprocessing statistics from. TYPE: Path
filesystem	The filesystem to load the file from. TYPE: AbstractFileSystem

RETURNS	DESCRIPTION
statistics	The loaded statistics. TYPE: PreProcessingStatistics

Source code in packages/importer_pkg/src/toop_engine_importer/pypowsybl_import/preprocessing.py

def load_preprocessing_statistics_filesystem(file_path: Path, filesystem: AbstractFileSystem) -> PreProcessingStatistics:
    """Load the preprocessing statistics from the file.

    Parameters
    ----------
    file_path: Path
        The file to load the preprocessing statistics from.
    filesystem: AbstractFileSystem
        The filesystem to load the file from.

    Returns
    -------
    statistics: PreProcessingStatistics
        The loaded statistics.

    """
    with filesystem.open(str(file_path), "r") as f:
        statistics = json.load(f)
    import_result = PreProcessingStatistics(**statistics)
    return import_result

save_preprocessing_statistics_filesystem

save_preprocessing_statistics_filesystem(
    statistics, filesystem, file_path
)

Save the preprocessing statistics to the filesystem.

PARAMETER	DESCRIPTION
statistics	The statistics to save. TYPE: PreProcessingStatistics
file_path	The file to save the preprocessing statistics to. TYPE: Union[str, Path]
filesystem	The filesystem to save the file to. TYPE: AbstractFileSystem

Source code in packages/importer_pkg/src/toop_engine_importer/pypowsybl_import/preprocessing.py

def save_preprocessing_statistics_filesystem(
    statistics: PreProcessingStatistics, filesystem: AbstractFileSystem, file_path: Union[str, Path]
) -> None:
    """Save the preprocessing statistics to the filesystem.

    Parameters
    ----------
    statistics: PreProcessingStatistics
        The statistics to save.
    file_path: Path
        The file to save the preprocessing statistics to.
    filesystem: AbstractFileSystem
        The filesystem to save the file to.
    """
    with filesystem.open(str(file_path), "w") as f:
        f.write(statistics.model_dump_json(indent=4))

toop_engine_importer.pypowsybl_import.preprocessing

Module contains functions for the pypowsybl preprocessing for the grid export into the loadflow solver.

File: preprocessing.py Author: Benjamin Petrick Created: 2024-09-04

logger module-attribute

logger = structlog.get_logger(__name__)

BaseCaseNotConvergedError

Bases: RuntimeError

The base case loadflow of the import did not converge, so the import cannot continue.

ImportedGridModel dataclass

ImportedGridModel(
    import_result, initial_loadflow, initial_metrics
)

The outcome of an import, including its AC security analysis.

import_result instance-attribute

import_result

The statistics of the import.

initial_loadflow instance-attribute

initial_loadflow

The AC N-1 results of the imported grid, with the permanent and the artificial branch limits.

initial_metrics instance-attribute

initial_metrics

The metrics of the initial loadflow. Metrics of limits that no branch has are left out.

save_preprocessing_statistics_filesystem

save_preprocessing_statistics_filesystem(
    statistics, filesystem, file_path
)

Save the preprocessing statistics to the filesystem.

PARAMETER	DESCRIPTION
statistics	The statistics to save. TYPE: PreProcessingStatistics
file_path	The file to save the preprocessing statistics to. TYPE: Union[str, Path]
filesystem	The filesystem to save the file to. TYPE: AbstractFileSystem

Source code in packages/importer_pkg/src/toop_engine_importer/pypowsybl_import/preprocessing.py

def save_preprocessing_statistics_filesystem(
    statistics: PreProcessingStatistics, filesystem: AbstractFileSystem, file_path: Union[str, Path]
) -> None:
    """Save the preprocessing statistics to the filesystem.

    Parameters
    ----------
    statistics: PreProcessingStatistics
        The statistics to save.
    file_path: Path
        The file to save the preprocessing statistics to.
    filesystem: AbstractFileSystem
        The filesystem to save the file to.
    """
    with filesystem.open(str(file_path), "w") as f:
        f.write(statistics.model_dump_json(indent=4))

load_preprocessing_statistics_filesystem

load_preprocessing_statistics_filesystem(
    file_path, filesystem
)

Load the preprocessing statistics from the file.

PARAMETER	DESCRIPTION
file_path	The file to load the preprocessing statistics from. TYPE: Path
filesystem	The filesystem to load the file from. TYPE: AbstractFileSystem

RETURNS	DESCRIPTION
statistics	The loaded statistics. TYPE: PreProcessingStatistics

Source code in packages/importer_pkg/src/toop_engine_importer/pypowsybl_import/preprocessing.py

def load_preprocessing_statistics_filesystem(file_path: Path, filesystem: AbstractFileSystem) -> PreProcessingStatistics:
    """Load the preprocessing statistics from the file.

    Parameters
    ----------
    file_path: Path
        The file to load the preprocessing statistics from.
    filesystem: AbstractFileSystem
        The filesystem to load the file from.

    Returns
    -------
    statistics: PreProcessingStatistics
        The loaded statistics.

    """
    with filesystem.open(str(file_path), "r") as f:
        statistics = json.load(f)
    import_result = PreProcessingStatistics(**statistics)
    return import_result

create_nminus1_definition_from_masks

create_nminus1_definition_from_masks(
    network, network_masks, master_topology
)

Create the N-1 definition from the network masks.

PARAMETER	DESCRIPTION
network	The network to create the N-1 definition for. TYPE: Network
network_masks	The network masks to create the N-1 definition from. TYPE: NetworkMasks
master_topology	The asset-topology master data providing the busbars and couplers of the relevant stations. TYPE: MasterAssetTopology

RETURNS	DESCRIPTION
Nminus1Definition	The created N-1 definition.

Source code in packages/importer_pkg/src/toop_engine_importer/pypowsybl_import/preprocessing.py

def create_nminus1_definition_from_masks(
    network: Network, network_masks: NetworkMasks, master_topology: MasterAssetTopology
) -> Nminus1Definition:
    """Create the N-1 definition from the network masks.

    Parameters
    ----------
    network: Network
        The network to create the N-1 definition for.
    network_masks: NetworkMasks
        The network masks to create the N-1 definition from.
    master_topology: MasterAssetTopology
        The asset-topology master data providing the busbars and couplers of the relevant stations.

    Returns
    -------
    Nminus1Definition
        The created N-1 definition.
    """
    contingencies = [Contingency(id="BASECASE", name="BASECASE", elements=[])]

    lines = network.get_lines(attributes=["name"])
    monitored_lines = _create_monitored_elements(
        lines,
        network_masks.line_for_optimized,
        network_masks.line_for_non_worsening,
        "LINE",
        "branch",
        weights=network_masks.line_overload_weight,
    )
    outaged_lines = [
        Contingency(id=idx, name=row["name"], elements=[GridElement(id=idx, name=row["name"], type="LINE", kind="branch")])
        for idx, row in lines[network_masks.line_for_nminus1].iterrows()
    ]

    trafos = sort_powsybl_element_frame_by_id(network.get_2_windings_transformers(attributes=["name"]))
    is_trafo2w = ~trafos.index.str.contains(TRAFO3W_LEG_PATTERN)
    monitored_trafos = _create_monitored_elements(
        trafos,
        is_trafo2w & network_masks.trafo_for_optimized,
        is_trafo2w & network_masks.trafo_for_non_worsening,
        "TWO_WINDINGS_TRANSFORMER",
        "branch",
        weights=network_masks.trafo_overload_weight,
    )
    outaged_trafos = [
        Contingency(
            id=idx,
            name=row["name"],
            elements=[GridElement(id=idx, name=row["name"], type="TWO_WINDINGS_TRANSFORMER", kind="branch")],
        )
        for idx, row in trafos[is_trafo2w & network_masks.trafo_for_nminus1].iterrows()
    ]

    is_trafo3w = trafos.index.str.contains(TRAFO3W_LEG_PATTERN)
    trafos.index = trafos.index.str.replace(TRAFO3W_LEG_PATTERN, "", regex=True)
    if not trafos.empty:
        trafos.name = trafos.name.str.replace(TRAFO3W_LEG_PATTERN, "", regex=True)

    monitored_trafo3w = _create_monitored_elements(
        trafos,
        is_trafo3w & network_masks.trafo_for_optimized,
        is_trafo3w & network_masks.trafo_for_non_worsening,
        "THREE_WINDINGS_TRANSFORMER",
        "branch",
        weights=network_masks.trafo_overload_weight,
        drop_duplicates=True,
    )
    outaged_trafo3w = [
        Contingency(
            id=idx,
            name=row["name"],
            elements=[GridElement(id=idx, name=row["name"], type="THREE_WINDINGS_TRANSFORMER", kind="branch")],
        )
        for idx, row in trafos[is_trafo3w & network_masks.trafo_for_nminus1].drop_duplicates().iterrows()
    ]

    tie_lines = network.get_tie_lines(attributes=["name"])
    monitored_tie_lines = _create_monitored_elements(
        tie_lines,
        network_masks.tie_line_for_optimized,
        network_masks.tie_line_for_non_worsening,
        "TIE_LINE",
        "branch",
        weights=network_masks.tie_line_overload_weight,
    )
    outaged_tie_lines = [
        Contingency(
            id=idx, name=row["name"], elements=[GridElement(id=idx, name=row["name"], type="TIE_LINE", kind="branch")]
        )
        for idx, row in tie_lines[network_masks.tie_line_for_nminus1].iterrows()
    ]

    boundary_lines = network.get_boundary_lines(attributes=["name", "paired"])
    outaged_boundary = [
        Contingency(
            id=idx,
            name=row["name"],
            elements=[
                GridElement(id=idx, name=row["name"], type="BOUNDARY_LINE", kind="injection"),
            ],
        )
        for idx, row in boundary_lines[network_masks.boundary_line_for_nminus1 & ~boundary_lines["paired"]].iterrows()
    ]

    generators = network.get_generators(attributes=["name"])
    outaged_generators = [
        Contingency(
            id=idx, name=row["name"], elements=[GridElement(id=idx, name=row["name"], type="GENERATOR", kind="injection")]
        )
        for idx, row in generators[network_masks.generator_for_nminus1].iterrows()
    ]

    loads = network.get_loads(attributes=["name"])
    outaged_loads = [
        Contingency(
            id=idx, name=row["name"], elements=[GridElement(id=idx, name=row["name"], type="LOAD", kind="injection")]
        )
        for idx, row in loads[network_masks.load_for_nminus1].iterrows()
    ]

    relevant_voltage_level_ids = set(
        network.get_buses(attributes=["voltage_level_id"])["voltage_level_id"][network_masks.relevant_subs]
    )
    monitored_stations = get_monitored_station_elements(
        [bus_group for bus_group in master_topology.bus_groups if bus_group.voltage_level_id in relevant_voltage_level_ids]
    )
    monitored_station_ids = {element.id for element in monitored_stations}

    switches = network.get_switches(attributes=["name"])
    monitored_switches = [
        element
        for element in _create_monitored_elements(
            switches, network_masks.switch_for_optimized, network_masks.switch_for_non_worsening, "SWITCH", "switch"
        )
        if element.id not in monitored_station_ids
    ]
    outaged_switches = [
        Contingency(id=idx, name=row["name"], elements=[GridElement(id=idx, name=row["name"], type="SWITCH", kind="branch")])
        for idx, row in switches[network_masks.switch_for_nminus1].iterrows()
    ]

    busbar_sections = network.get_busbar_sections(attributes=["name"])
    outaged_busbars = [
        Contingency(
            id=idx,
            name=row["name"],
            elements=[GridElement(id=idx, name=row["name"], type="BUSBAR_SECTION", kind="bus")],
        )
        for idx, row in busbar_sections[network_masks.busbar_for_nminus1].iterrows()
    ]

    nminus1_definition = Nminus1Definition(
        monitored_elements=(
            monitored_lines
            + monitored_trafos
            + monitored_trafo3w
            + monitored_tie_lines
            + monitored_switches
            + monitored_stations
        ),
        contingencies=(
            contingencies
            + outaged_lines
            + outaged_trafos
            + outaged_trafo3w
            + outaged_tie_lines
            + outaged_boundary
            + outaged_generators
            + outaged_loads
            + outaged_switches
            + outaged_busbars
        ),
    )
    return nminus1_definition

create_nminus1_definition

create_nminus1_definition(
    network,
    network_masks,
    master_topology,
    input_nminus1_definition=None,
)

Create the N-1 definition that is saved next to the processed grid.

The input N-1 definition is authoritative and not filtered by the area settings; without one, the definition is derived from the network masks. Three-winding transformers are then replaced by their two-winding legs, and an input definition is checked against network again, dropping elements that preprocessing removed.

PARAMETER	DESCRIPTION
network	The processed grid, after the three-winding transformer conversion. TYPE: Network
network_masks	The network masks, used when no input N-1 definition is given. TYPE: NetworkMasks
master_topology	The asset-topology master data, used when no input N-1 definition is given. TYPE: MasterAssetTopology
input_nminus1_definition	The grid-validated input N-1 definition, if any. TYPE: Optional[Nminus1Definition] DEFAULT: None

RETURNS	DESCRIPTION
Nminus1Definition	The N-1 definition referencing only elements of the processed grid.

Source code in packages/importer_pkg/src/toop_engine_importer/pypowsybl_import/preprocessing.py

def create_nminus1_definition(
    network: Network,
    network_masks: NetworkMasks,
    master_topology: MasterAssetTopology,
    input_nminus1_definition: Optional[Nminus1Definition] = None,
) -> Nminus1Definition:
    """Create the N-1 definition that is saved next to the processed grid.

    The input N-1 definition is authoritative and not filtered by the area settings; without one, the definition is
    derived from the network masks. Three-winding transformers are then replaced by their two-winding legs, and an
    input definition is checked against ``network`` again, dropping elements that preprocessing removed.

    Parameters
    ----------
    network : Network
        The processed grid, after the three-winding transformer conversion.
    network_masks : NetworkMasks
        The network masks, used when no input N-1 definition is given.
    master_topology : MasterAssetTopology
        The asset-topology master data, used when no input N-1 definition is given.
    input_nminus1_definition : Optional[Nminus1Definition]
        The grid-validated input N-1 definition, if any.

    Returns
    -------
    Nminus1Definition
        The N-1 definition referencing only elements of the processed grid.
    """
    if input_nminus1_definition is None:
        nminus1_definition = create_nminus1_definition_from_masks(network, network_masks, master_topology)
        logger.warning("No input N-1 definition provided, creating one from network masks.")
        return convert_three_winding_transformers_in_nminus1_definition(nminus1_definition, network)
    nminus1_definition = convert_three_winding_transformers_in_nminus1_definition(input_nminus1_definition, network)
    return filter_nminus1_definition_to_network(nminus1_definition, network)

get_input_nminus1_definition_voltage_level_ids

get_input_nminus1_definition_voltage_level_ids(
    network, input_nminus1_definition
)

Get the voltage levels of all input N-1 elements, including the star voltage levels of converted 3W trafos.

Network reduction keeps them, because the input N-1 definition takes precedence over the area settings.

PARAMETER	DESCRIPTION
network	The grid after the three-winding transformer conversion and before the reduction. TYPE: Network
input_nminus1_definition	The grid-validated input N-1 definition, if any. TYPE: Optional[Nminus1Definition]

RETURNS	DESCRIPTION
list[str]	The voltage level ids to keep, empty if no input N-1 definition is given.

Source code in packages/importer_pkg/src/toop_engine_importer/pypowsybl_import/preprocessing.py

def get_input_nminus1_definition_voltage_level_ids(
    network: Network, input_nminus1_definition: Optional[Nminus1Definition]
) -> list[str]:
    """Get the voltage levels of all input N-1 elements, including the star voltage levels of converted 3W trafos.

    Network reduction keeps them, because the input N-1 definition takes precedence over the area settings.

    Parameters
    ----------
    network : Network
        The grid after the three-winding transformer conversion and before the reduction.
    input_nminus1_definition : Optional[Nminus1Definition]
        The grid-validated input N-1 definition, if any.

    Returns
    -------
    list[str]
        The voltage level ids to keep, empty if no input N-1 definition is given.
    """
    if input_nminus1_definition is None:
        return []
    converted_definition = convert_three_winding_transformers_in_nminus1_definition(input_nminus1_definition, network)
    return get_voltage_level_ids_of_elements(network, get_nminus1_definition_element_ids(converted_definition))

load_and_prepare_network

load_and_prepare_network(
    importer_parameters,
    processed_gridfile_fs,
    unprocessed_gridfile_fs,
    status_update_fn,
)

Copy, load, and normalize the input network before preprocessing.

An input N-1 definition is validated before the three-winding transformer conversion, as it uses their original ids.

PARAMETER	DESCRIPTION
importer_parameters	Parameters describing the input grid file and output folder. TYPE: BaseImporterParameters
processed_gridfile_fs	Filesystem where the original input grid is archived. TYPE: AbstractFileSystem
unprocessed_gridfile_fs	Filesystem from which the input grid is loaded. TYPE: AbstractFileSystem
status_update_fn	Callback used to report preprocessing progress. TYPE: StatusUpdateFn

RETURNS	DESCRIPTION
Network	The loaded and normalized network.
Optional[Nminus1Definition]	The grid-validated input N-1 definition, or None if importer_parameters.input_nminus1_definition_file is None.

Source code in packages/importer_pkg/src/toop_engine_importer/pypowsybl_import/preprocessing.py

def load_and_prepare_network(
    importer_parameters: BaseImporterParameters,
    processed_gridfile_fs: AbstractFileSystem,
    unprocessed_gridfile_fs: AbstractFileSystem,
    status_update_fn: StatusUpdateFn,
) -> tuple[Network, Optional[Nminus1Definition]]:
    """Copy, load, and normalize the input network before preprocessing.

    An input N-1 definition is validated before the three-winding transformer conversion, as it uses their original ids.

    Parameters
    ----------
    importer_parameters : BaseImporterParameters
        Parameters describing the input grid file and output folder.
    processed_gridfile_fs : AbstractFileSystem
        Filesystem where the original input grid is archived.
    unprocessed_gridfile_fs : AbstractFileSystem
        Filesystem from which the input grid is loaded.
    status_update_fn : StatusUpdateFn
        Callback used to report preprocessing progress.

    Returns
    -------
    Network
        The loaded and normalized network.
    Optional[Nminus1Definition]
        The grid-validated input N-1 definition, or None if `importer_parameters.input_nminus1_definition_file` is None.
    """
    copy_file_fs(
        src_fs=unprocessed_gridfile_fs,
        src_path=importer_parameters.grid_model_file.as_posix(),
        dest_fs=processed_gridfile_fs,
        dest_path=(
            importer_parameters.data_folder
            / PREPROCESSING_PATHS["original_gridfile_path"]
            / importer_parameters.grid_model_file.name
        ).as_posix(),
    )

    status_update_fn("load_from_fs", "start loading grid file")
    network = load_powsybl_from_fs(
        filesystem=unprocessed_gridfile_fs,
        file_path=importer_parameters.grid_model_file,
        parameters={"iidm.import.cgmes.post-processors": "cgmesGLImport", "iidm.import.cgmes.cgm-with-subnetworks": "false"},
    )
    network_analysis.remove_branches_with_same_bus(network)
    status_update_fn("load_from_fs", "done loading grid file")

    input_nminus1_definition = None
    if importer_parameters.input_nminus1_definition_file is not None:
        status_update_fn("load_from_fs", "Validating the input N-1 definition against the grid")
        input_nminus1_definition = load_nminus1_definition_for_network(
            network=network,
            file_path=importer_parameters.input_nminus1_definition_file,
            filesystem=unprocessed_gridfile_fs,
        )

    pypowsybl.network.replace_3_windings_transformers_with_3_2_windings_transformers(network)
    if pypowsybl.__version__ <= "1.12.0":
        # Fix the bug, where the operational limits of the 2winding transformers are not set correctly
        op_lim = network.get_operational_limits(all_attributes=True, show_inactive_sets=True)
        trafo3w_lims = op_lim[op_lim.index.str.contains(TRAFO3W_LEG_PATTERN)][["group_name"]].rename(
            columns={"group_name": "selected_limits_group_1"}
        )
        trafo3w_lims.index.name = "id"
        network.update_2_windings_transformers(trafo3w_lims)

    return network, input_nminus1_definition

import_grid_with_security_analysis

import_grid_with_security_analysis(
    importer_parameters,
    status_update_fn=empty_status_update_fn,
    processed_gridfile_fs=None,
    unprocessed_gridfile_fs=None,
    *,
    job_id="",
)

Import the grid file and run its AC security analysis, which sets the artificial limits.

Saves data and network to the output folder. The security analysis runs once. Its results, with the artificial limits attached, are returned as the initial loadflow, so no second analysis is needed.

PARAMETER	DESCRIPTION
importer_parameters	Parameters that are required to import the data from a UCTE or CGMES file. This will utilize powsybl and the powsybl backend to the loadflow solver TYPE: BaseImporterParameters
status_update_fn	A function to call to signal progress in the preprocessing pipeline. Takes a stage, an optional message and network stats as parameters TYPE: StatusUpdateFn DEFAULT: empty_status_update_fn
processed_gridfile_fs	A filesystem where the processed gridfiles are stored. If None, the local filesystem is used TYPE: Optional[AbstractFileSystem] DEFAULT: None
unprocessed_gridfile_fs	A filesystem where the unprocessed gridfiles are stored. If None, the local filesystem is used. TYPE: Optional[AbstractFileSystem] DEFAULT: None
job_id	The job id stored with the initial loadflow. TYPE: str DEFAULT: ''

RETURNS	DESCRIPTION
ImportedGridModel	The statistics of the import, the initial loadflow and its metrics.

RAISES	DESCRIPTION
BaseCaseNotConvergedError	If the base case loadflow does not converge, before or after the preprocessing changes to the network.

Source code in packages/importer_pkg/src/toop_engine_importer/pypowsybl_import/preprocessing.py

def import_grid_with_security_analysis(  # noqa: PLR0915
    importer_parameters: BaseImporterParameters,
    status_update_fn: StatusUpdateFn = empty_status_update_fn,
    processed_gridfile_fs: Optional[AbstractFileSystem] = None,
    unprocessed_gridfile_fs: Optional[AbstractFileSystem] = None,
    *,
    job_id: str = "",
) -> ImportedGridModel:
    """Import the grid file and run its AC security analysis, which sets the artificial limits.

    Saves data and network to the output folder. The security analysis runs once. Its results, with the artificial
    limits attached, are returned as the initial loadflow, so no second analysis is needed.

    Parameters
    ----------
    importer_parameters: BaseImporterParameters
        Parameters that are required to import the data from a UCTE or CGMES file. This will utilize
        powsybl and the powsybl backend to the loadflow solver
    status_update_fn: StatusUpdateFn
        A function to call to signal progress in the preprocessing pipeline. Takes a stage, an
        optional message and network stats as parameters
    processed_gridfile_fs: Optional[AbstractFileSystem]
        A filesystem where the processed gridfiles are stored. If None, the local filesystem is used
    unprocessed_gridfile_fs: Optional[AbstractFileSystem]
        A filesystem where the unprocessed gridfiles are stored. If None, the local filesystem is used.
    job_id: str
        The job id stored with the initial loadflow.

    Returns
    -------
    ImportedGridModel
        The statistics of the import, the initial loadflow and its metrics.

    Raises
    ------
    BaseCaseNotConvergedError
        If the base case loadflow does not converge, before or after the preprocessing changes to the network.
    """
    if unprocessed_gridfile_fs is None:
        unprocessed_gridfile_fs = LocalFileSystem()
    if processed_gridfile_fs is None:
        processed_gridfile_fs = LocalFileSystem()
    network, input_nminus1_definition = load_and_prepare_network(
        importer_parameters=importer_parameters,
        processed_gridfile_fs=processed_gridfile_fs,
        unprocessed_gridfile_fs=unprocessed_gridfile_fs,
        status_update_fn=status_update_fn,
    )

    # Iterate over Loadflow parameters and voltage initialization methods to find a converging loadflow.
    # This is necessary because some grid files do not converge with the
    # default loadflow parameters and voltage initialization method.

    statistics = PreProcessingStatistics(
        import_result=ImportResult(data_folder=importer_parameters.data_folder, grid_type=importer_parameters.data_type),
        import_parameter=importer_parameters,
    )

    # a loadflow is needed for the network set_tie_line_boundary_equivalents and later for the reduction
    lf_params, main_result = _run_initial_loadflow(importer_parameters, network, unprocessed_gridfile_fs)
    if main_result.status != pypowsybl.loadflow.ComponentStatus.CONVERGED:
        raise BaseCaseNotConvergedError("The base case loadflow did not converge, preprocessing stopped.")

    # set_tie_line_boundary_equivalents
    # sets the p0 and q0 of the boundary lines to match the actual flow over the tie line
    # This is needed if the grid is reduced and a tie line is removed -> wrong loadflow if the p0 and q0 is not set
    network_analysis.set_tie_line_boundary_equivalents(net=network)

    if importer_parameters.network_reduction_voltage_level_range >= 0:
        status_update_fn("reduce_network_to_view_area", "Reducing network to view area")
        reduce_network_based_on_area_settings(
            net=network,
            importer_parameters=importer_parameters,
            extra_voltage_level_ids=get_input_nminus1_definition_voltage_level_ids(network, input_nminus1_definition),
        )

    status_update_fn("apply_cb_list", "Applying Whitelists")
    if importer_parameters.data_type == "ucte":
        # TODO: move to UCTE Toolset after all PRs are merged
        apply_preprocessing_changes_to_network(
            network=network,
            statistics=statistics,
            status_update_fn=status_update_fn,
        )

        # apply black and white list
        statistics = network_analysis.apply_cb_lists(
            network=network,
            statistics=statistics,
            white_list_file=importer_parameters.white_list_file,
            black_list_file=importer_parameters.black_list_file,
            fs=unprocessed_gridfile_fs,
        )
    elif importer_parameters.data_type == "cgmes":
        statistics = network_analysis.apply_cb_lists_cgmes(
            statistics=statistics,
            white_list_file=importer_parameters.white_list_file,
            ignore_list_file=importer_parameters.ignore_list_file,
            filesystem=unprocessed_gridfile_fs,
        )

    # Save and reload Network due to powsybl changing order during save
    grid_file_path = importer_parameters.data_folder / PREPROCESSING_PATHS["grid_file_path_powsybl"]
    save_powsybl_to_fs(
        network,
        filesystem=processed_gridfile_fs,
        file_path=grid_file_path,
    )

    # Reload Network because powsybl likes to change order during save
    network = load_powsybl_from_fs(
        filesystem=processed_gridfile_fs,
        file_path=grid_file_path,
    )
    lf_params, main_result = _run_initial_loadflow(importer_parameters, network, unprocessed_gridfile_fs)
    if main_result.status != pypowsybl.loadflow.ComponentStatus.CONVERGED:
        raise BaseCaseNotConvergedError(
            "The base case loadflow did not converge after the preprocessing changes, preprocessing stopped."
        )
    save_lf_params_to_fs(
        lf_params=lf_params,
        filesystem=processed_gridfile_fs,
        file_path=importer_parameters.data_folder / PREPROCESSING_PATHS["loadflow_parameters_file_path"],
    )

    # get N-1 masks
    status_update_fn("get_masks", "Creating Network Masks")
    network_masks = compute_network_masks(
        importer_parameters,
        processed_gridfile_fs,
        unprocessed_gridfile_fs,
        network,
        statistics,
    )

    if (
        importer_parameters.area_settings.dso_trafo_factors is not None
        or importer_parameters.area_settings.border_line_factors is not None
    ):
        status_update_fn("cross_border_current", "Setting cross border current limit")
        create_new_border_limits(network, network_masks, importer_parameters)
        # save new border limits

    status_update_fn("get_topology_model", "Creating canonical asset-topology master data")
    topology_master_data = get_master_asset_topology_artifact(
        network,
        network_masks,
        importer_parameters,
    )
    fill_statistics_for_network_masks(network=network, statistics=statistics, network_masks=network_masks)

    save_masks_to_filesystem(
        data_folder=importer_parameters.data_folder, network_masks=network_masks, filesystem=processed_gridfile_fs
    )

    # get nminus1 definition
    nminus1_definition = create_nminus1_definition(network, network_masks, topology_master_data, input_nminus1_definition)
    fill_statistics_for_nminus1_definition(statistics=statistics, nminus1_definition=nminus1_definition)
    save_pydantic_model_fs(
        filesystem=processed_gridfile_fs,
        file_path=importer_parameters.data_folder / PREPROCESSING_PATHS["nminus1_definition_file_path"],
        pydantic_model=nminus1_definition,
    )

    if importer_parameters.create_artificial_limit_when_missing_permanent:
        # Before the analysis, as the loading of a branch is measured against its permanent limit
        add_default_permanent_limits(
            network,
            [element.id for element in nminus1_definition.monitored_elements if element.kind == "branch"],
            importer_parameters.default_permanent,
        )

    status_update_fn("security_analysis", "Running the AC security analysis, stored as the initial loadflow")
    security_analysis_results = run_contingency_analysis_powsybl(
        net=network,
        n_minus_1_definition=nminus1_definition,
        job_id=job_id,
        timestep=0,
        method="ac",
        polars=True,
        lf_params=lf_params,
        n_processes=importer_parameters.n_contingency_worker,
        batch_size=importer_parameters.contingency_batch_size,
    )
    assert isinstance(security_analysis_results, LoadflowResultsPolars), (
        "Security analysis did not return LoadflowResultsPolars"
    )
    status_update_fn("set_artificial_limits", "Setting the optimization limits of the monitored branches")
    initial_loadflow = attach_artificial_limits_to_results(
        network=network,
        nminus1_definition=nminus1_definition,
        security_analysis_results=security_analysis_results,
        double_limits=importer_parameters.double_limits,
    )
    initial_metrics = compute_initial_metrics(initial_loadflow, nminus1_definition)

    status_update_fn("set_artificial_dc_limits", "Setting the DC optimization limits of the monitored branches")
    dc_metrics = compute_dc_artificial_limit_metrics(
        network=network,
        nminus1_definition=nminus1_definition,
        double_limits=importer_parameters.double_limits,
        lf_params=lf_params,
        job_id=job_id,
        n_processes=importer_parameters.n_contingency_worker,
        batch_size=importer_parameters.contingency_batch_size,
    )
    fill_statistics_for_overload_energy(statistics=statistics, metrics=initial_metrics, dc_metrics=dc_metrics)

    save_powsybl_to_fs(
        network,
        filesystem=processed_gridfile_fs,
        file_path=grid_file_path,
    )

    save_preprocessing_statistics_filesystem(
        statistics=statistics,
        file_path=importer_parameters.data_folder / PREPROCESSING_PATHS["importer_auxiliary_file_path"],
        filesystem=processed_gridfile_fs,
    )

    save_pydantic_model_fs(
        filesystem=processed_gridfile_fs,
        file_path=importer_parameters.data_folder / PREPROCESSING_PATHS["asset_topology_master_data_file_path"],
        pydantic_model=topology_master_data,
        indent=4,
    )
    return ImportedGridModel(
        import_result=statistics.import_result,
        initial_loadflow=initial_loadflow,
        initial_metrics=initial_metrics,
    )

compute_network_masks

compute_network_masks(
    importer_parameters,
    processed_gridfile_fs,
    unprocessed_gridfile_fs,
    network,
    statistics,
)

Create, persist, and return network masks.

PARAMETER	DESCRIPTION
importer_parameters	Import configuration providing the data folder and mask generation settings. TYPE: Union[UcteImporterParameters, CgmesImporterParameters]
processed_gridfile_fs	Filesystem used to persist the generated masks. TYPE: AbstractFileSystem
unprocessed_gridfile_fs	Filesystem used to resolve auxiliary inputs required during mask creation. TYPE: AbstractFileSystem
network	Powsybl network for which masks are computed. TYPE: Network
statistics	Statistics object updated while generating masks. TYPE: PreProcessingStatistics

RETURNS	DESCRIPTION
NetworkMasks	Generated network masks after saving them.

Source code in packages/importer_pkg/src/toop_engine_importer/pypowsybl_import/preprocessing.py

def compute_network_masks(
    importer_parameters: Union[UcteImporterParameters, CgmesImporterParameters],
    processed_gridfile_fs: AbstractFileSystem,
    unprocessed_gridfile_fs: AbstractFileSystem,
    network: Network,
    statistics: PreProcessingStatistics,
) -> NetworkMasks:
    """Create, persist, and return network masks.

    Parameters
    ----------
    importer_parameters : Union[UcteImporterParameters, CgmesImporterParameters]
        Import configuration providing the data folder and mask generation settings.
    processed_gridfile_fs : AbstractFileSystem
        Filesystem used to persist the generated masks.
    unprocessed_gridfile_fs : AbstractFileSystem
        Filesystem used to resolve auxiliary inputs required during mask creation.
    network : Network
        Powsybl network for which masks are computed.
    statistics : PreProcessingStatistics
        Statistics object updated while generating masks.

    Returns
    -------
    NetworkMasks
        Generated network masks after saving them.
    """
    slack_id = network.get_extension("slackTerminal").iloc[0].bus_id
    network_masks = get_network_masks(
        network,
        slack_id,
        importer_parameters,
        statistics,
        filesystem=unprocessed_gridfile_fs,
    )
    save_masks_to_filesystem(
        data_folder=importer_parameters.data_folder, network_masks=network_masks, filesystem=processed_gridfile_fs
    )
    return network_masks

get_slack_ids

get_slack_ids(network)

Get the slack bus ids from the network.

PARAMETER	DESCRIPTION
network	The network to get the slack bus ids from. TYPE: Network

RETURNS	DESCRIPTION
list[str] | None	The list of slack bus ids.

Source code in packages/importer_pkg/src/toop_engine_importer/pypowsybl_import/preprocessing.py

def get_slack_ids(network: Network) -> list[str] | None:
    """Get the slack bus ids from the network.

    Parameters
    ----------
    network: Network
        The network to get the slack bus ids from.

    Returns
    -------
    list[str] | None
        The list of slack bus ids.
    """
    gen_ids_by_prio = network.get_extensions("referencePriorities").sort_values(by="priority").index
    if gen_ids_by_prio.empty:
        # in this case it will pick the most connected
        return None
    gens = network.get_generators(attributes=["bus_id"])
    slack_ids = gens[gens != ""].bus_id.to_list()
    return slack_ids

find_converging_loadflow_params

find_converging_loadflow_params(
    importer_parameters, network
)

Iterate over Loadflow parameters and voltage initialization methods to find a converging loadflow.

This is necessary because some grid files do not converge with the default loadflow parameters and voltage initialization method.

PARAMETER	DESCRIPTION
importer_parameters	The importer parameters to use for the loadflow parameters. TYPE: BaseImporterParameters
network	The network to run the loadflow on. TYPE: Network

RETURNS	DESCRIPTION
Tuple[Parameters, ComponentResult]	The loadflow parameters that converged and the result of the loadflow with those parameters.

Source code in packages/importer_pkg/src/toop_engine_importer/pypowsybl_import/preprocessing.py

def find_converging_loadflow_params(
    importer_parameters: BaseImporterParameters, network: Network
) -> tuple[pypowsybl.loadflow.Parameters, pypowsybl.loadflow.ComponentResult]:
    """Iterate over Loadflow parameters and voltage initialization methods to find a converging loadflow.

    This is necessary because some grid files do not converge with
    the default loadflow parameters and voltage initialization method.

    Parameters
    ----------
    importer_parameters: BaseImporterParameters
        The importer parameters to use for the loadflow parameters.
    network: Network
        The network to run the loadflow on.

    Returns
    -------
    Tuple[pypowsybl.loadflow.Parameters, pypowsybl.loadflow.ComponentResult]
        The loadflow parameters that converged and the result of the loadflow with those parameters.
    """
    lf_params_list = [POWSYBL_LOADFLOW_PARAM_PF, CGMES_DISTRIBUTED_SLACK]
    voltage_methods = [VoltageInitMode.PREVIOUS_VALUES, VoltageInitMode.DC_VALUES, VoltageInitMode.UNIFORM_VALUES]

    for lf_params_base, voltage_method in product(lf_params_list, voltage_methods):
        lf_params = deepcopy(lf_params_base)
        lf_params.provider_parameters = deepcopy(lf_params_base.provider_parameters)
        lf_params.voltage_init_mode = voltage_method
        try:
            main_result, *_ = pypowsybl.loadflow.run_ac(network, parameters=lf_params)
        except pypowsybl.PyPowsyblError:
            continue

        if main_result.status == pypowsybl.loadflow.ComponentStatus.CONVERGED:
            break
    else:
        if importer_parameters.fail_on_non_convergence:
            raise RuntimeError(
                "Loadflow did not converge with any voltage initialization method. "
                "Please check the grid file and the loadflow parameters."
            )
        lf_params = CGMES_DISTRIBUTED_SLACK
        logger.warning(
            "Loadflow did not converge with any voltage initialization method. "
            "Continuing with the DISTRIBUTED SLACK params but the loadflow results should be treated with caution."
        )

    return lf_params, main_result

get_network_masks

get_network_masks(
    network,
    slack_id,
    importer_parameters,
    statistics,
    filesystem,
)

Create network masks and save them.

PARAMETER	DESCRIPTION
network	The network to create the asset topology for TYPE: Network
slack_id	The id of the slack bus TYPE: str
importer_parameters	import parameters that include the datafolder TYPE: Union[UcteImporterParameters, CgmesImporterParameters]
statistics	preprocessing statistics to fill with information TYPE: PreProcessingStatistics
filesystem	The filesystem to load the mask files from. TYPE: AbstractFileSystem

RETURNS	DESCRIPTION
NetworkMasks	The created network masks

Source code in packages/importer_pkg/src/toop_engine_importer/pypowsybl_import/preprocessing.py

def get_network_masks(
    network: Network,
    slack_id: str,
    importer_parameters: Union[UcteImporterParameters, CgmesImporterParameters],
    statistics: PreProcessingStatistics,
    filesystem: AbstractFileSystem,
) -> NetworkMasks:
    """Create network masks and save them.

    Parameters
    ----------
    network: Network
        The network to create the asset topology for
    slack_id: str
        The id of the slack bus
    importer_parameters: Union[UcteImporterParameters, CgmesImporterParameters]
        import parameters that include the datafolder
    statistics: PreProcessingStatistics
        preprocessing statistics to fill with information
    filesystem: AbstractFileSystem
        The filesystem to load the mask files from.

    Returns
    -------
    NetworkMasks
        The created network masks
    """
    network_masks = make_masks(
        network=network,
        slack_id=slack_id,
        importer_parameters=importer_parameters,
        blacklisted_ids=statistics.id_lists["black_list"],
        filesystem=filesystem,
    )
    fill_statistics_for_network_masks(network=network, statistics=statistics, network_masks=network_masks)
    return network_masks

get_master_asset_topology_artifact

get_master_asset_topology_artifact(
    network, network_masks, importer_parameters
)

Return canonical asset-topology master data for preprocessing persistence.

Source code in packages/importer_pkg/src/toop_engine_importer/pypowsybl_import/preprocessing.py

def get_master_asset_topology_artifact(
    network: Network,
    network_masks: NetworkMasks,
    importer_parameters: Union[UcteImporterParameters, CgmesImporterParameters],
) -> MasterAssetTopology:
    """Return canonical asset-topology master data for preprocessing persistence."""
    if importer_parameters.data_type == "ucte":
        return get_bus_breaker_master_asset_topology(
            network=network,
            relevant_stations=network_masks.relevant_subs,
            topology_id=importer_parameters.grid_model_file.name,
            grid_model_file=str(importer_parameters.grid_model_file),
        )

    if importer_parameters.data_type == "cgmes":
        return powsybl_station_to_graph.get_node_breaker_master_asset_topology(
            network=network,
            network_masks=network_masks,
            importer_parameters=importer_parameters,
        )

    raise ValueError(f"Unsupported importer data_type {importer_parameters.data_type}")

apply_preprocessing_changes_to_network

apply_preprocessing_changes_to_network(
    network, statistics, status_update_fn=None
)

Apply the default changes to the network.

These changes include: - removing low impedance lines - removing branches across switches

PARAMETER	DESCRIPTION
network	The network to apply the changes to. Note: This function modifies the network in place. TYPE: Network
statistics	The statistics of the preprocessing. Note: This function modifies the statistics in place. TYPE: PreProcessingStatistics
status_update_fn	A function to call to signal progress in the preprocessing pipeline. Takes a stage and an optional message as parameters TYPE: Optional[StatusUpdateFn] DEFAULT: None

Source code in packages/importer_pkg/src/toop_engine_importer/pypowsybl_import/preprocessing.py

def apply_preprocessing_changes_to_network(
    network: Network,
    statistics: PreProcessingStatistics,
    status_update_fn: Optional[StatusUpdateFn] = None,
) -> None:
    """Apply the default changes to the network.

    These changes include:
    - removing low impedance lines
    - removing branches across switches

    Parameters
    ----------
    network: Network
        The network to apply the changes to.
        Note: This function modifies the network in place.
    statistics: PreprocessingStatistics
        The statistics of the preprocessing.
        Note: This function modifies the statistics in place.
    status_update_fn: Optional[StatusUpdateFn]
        A function to call to signal progress in the preprocessing pipeline. Takes a stage and an
        optional message as parameters

    """
    if status_update_fn is None:
        status_update_fn = empty_status_update_fn
    status_update_fn("modify_low_impedance_lines", "Converting low impedance lines to breakers")
    low_impedance_lines = network_analysis.convert_low_impedance_lines(network, "D8")
    statistics.import_result.n_low_impedance_lines = len(low_impedance_lines)
    statistics.network_changes["low_impedance_lines"] = low_impedance_lines.index.to_list()

    status_update_fn("modify_branches_over_switches", "Removing branches across switches")
    branches_across_switch = network_analysis.remove_branches_across_switch(network)
    statistics.import_result.n_branch_across_switch = len(branches_across_switch)
    statistics.network_changes["branches_across_switch"] = branches_across_switch.index.to_list()

fill_statistics_for_overload_energy

fill_statistics_for_overload_energy(
    statistics, metrics, dc_metrics
)

Fill the statistics with the overload energy against the permanent, the artificial and the artificial DC limits.

PARAMETER	DESCRIPTION
statistics	The statistics to fill. Note: This function modifies the statistics in place. TYPE: PreProcessingStatistics
metrics	The AC metrics of the unsplit grid, see compute_initial_metrics. A missing metric is stored as None. TYPE: Mapping[MetricType, float]
dc_metrics	The DC metrics of the unsplit grid against the artificial DC limits, named like the artificial limit metrics. A missing metric is stored as None. TYPE: Mapping[MetricType, float]

Source code in packages/importer_pkg/src/toop_engine_importer/pypowsybl_import/preprocessing.py

def fill_statistics_for_overload_energy(
    statistics: PreProcessingStatistics, metrics: Mapping[MetricType, float], dc_metrics: Mapping[MetricType, float]
) -> None:
    """Fill the statistics with the overload energy against the permanent, the artificial and the artificial DC limits.

    Parameters
    ----------
    statistics: PreprocessingStatistics
        The statistics to fill.
        Note: This function modifies the statistics in place.
    metrics: Mapping[MetricType, float]
        The AC metrics of the unsplit grid, see compute_initial_metrics. A missing metric is stored as None.
    dc_metrics: Mapping[MetricType, float]
        The DC metrics of the unsplit grid against the artificial DC limits, named like the artificial limit metrics.
        A missing metric is stored as None.
    """
    statistics.import_result.overload_energy_permanent_n_0 = metrics.get("overload_energy_permanent_n_0")
    statistics.import_result.overload_energy_permanent_n_1 = metrics.get("overload_energy_permanent_n_1")
    statistics.import_result.overload_energy_artificial_limit_n_0 = metrics.get("overload_energy_artificial_limit_n_0")
    statistics.import_result.overload_energy_artificial_limit_n_1 = metrics.get("overload_energy_artificial_limit_n_1")
    statistics.import_result.overload_energy_artificial_dc_limit_n_0 = dc_metrics.get("overload_energy_artificial_limit_n_0")
    statistics.import_result.overload_energy_artificial_dc_limit_n_1 = dc_metrics.get("overload_energy_artificial_limit_n_1")

fill_statistics_for_nminus1_definition

fill_statistics_for_nminus1_definition(
    statistics, nminus1_definition
)

Fill the statistics with the monitored branches of the N-1 definition.

PARAMETER	DESCRIPTION
statistics	The statistics to fill. Note: This function modifies the statistics in place. TYPE: PreProcessingStatistics
nminus1_definition	The N-1 definition, generated from the masks or given as input. TYPE: Nminus1Definition

Source code in packages/importer_pkg/src/toop_engine_importer/pypowsybl_import/preprocessing.py

def fill_statistics_for_nminus1_definition(
    statistics: PreProcessingStatistics, nminus1_definition: Nminus1Definition
) -> None:
    """Fill the statistics with the monitored branches of the N-1 definition.

    Parameters
    ----------
    statistics: PreprocessingStatistics
        The statistics to fill.
        Note: This function modifies the statistics in place.
    nminus1_definition: Nminus1Definition
        The N-1 definition, generated from the masks or given as input.
    """
    branches = [element for element in nminus1_definition.monitored_elements if element.kind == "branch"]
    statistics.import_result.n_branches_monitored = len(branches)
    statistics.import_result.n_branches_optimized = sum(element.optimized for element in branches)
    statistics.import_result.n_branches_non_worsening = sum(element.non_worsening for element in branches)

fill_statistics_for_network_masks

fill_statistics_for_network_masks(
    network, statistics, network_masks
)

Fill the statistics with the network masks.

PARAMETER	DESCRIPTION
network	The network to get the id lists from. TYPE: Network
statistics	The statistics to fill. Note: This function modifies the statistics in place. TYPE: PreProcessingStatistics
network_masks	The masks for the network. TYPE: NetworkMasks

Source code in packages/importer_pkg/src/toop_engine_importer/pypowsybl_import/preprocessing.py

def fill_statistics_for_network_masks(
    network: Network, statistics: PreProcessingStatistics, network_masks: NetworkMasks
) -> None:
    """Fill the statistics with the network masks.

    Parameters
    ----------
    network: Network
        The network to get the id lists from.
    statistics: PreprocessingStatistics
        The statistics to fill.
        Note: This function modifies the statistics in place.
    network_masks: NetworkMasks
        The masks for the network.

    """
    statistics.id_lists["relevant_subs"] = network.get_buses(attributes=[])[network_masks.relevant_subs].index.to_list()
    statistics.id_lists["line_for_nminus1"] = network.get_lines(attributes=[])[
        network_masks.line_for_nminus1
    ].index.to_list()
    statistics.id_lists["trafo_for_nminus1"] = sort_powsybl_element_frame_by_id(
        network.get_2_windings_transformers(attributes=[])
    )[network_masks.trafo_for_nminus1].index.to_list()
    statistics.id_lists["tie_line_for_nminus1"] = network.get_tie_lines(attributes=[])[
        network_masks.tie_line_for_nminus1
    ].index.to_list()
    statistics.id_lists["boundary_line_for_nminus1"] = network.get_boundary_lines(attributes=[])[
        network_masks.boundary_line_for_nminus1
    ].index.to_list()
    statistics.id_lists["generator_for_nminus1"] = network.get_generators(attributes=[])[
        network_masks.generator_for_nminus1
    ].index.to_list()
    statistics.id_lists["load_for_nminus1"] = network.get_loads(attributes=[])[
        network_masks.load_for_nminus1
    ].index.to_list()
    statistics.id_lists["switch_for_nminus1"] = network.get_switches(attributes=[])[
        network_masks.switch_for_nminus1
    ].index.to_list()
    statistics.id_lists["line_disconnectable"] = network.get_lines(attributes=[])[
        network_masks.line_disconnectable
    ].index.to_list()
    statistics.id_lists["trafo_disconnectable"] = sort_powsybl_element_frame_by_id(
        network.get_2_windings_transformers(attributes=[])
    )[network_masks.trafo_disconnectable].index.to_list()

    statistics.import_result.n_relevant_subs = int(network_masks.relevant_subs.sum())
    statistics.import_result.n_line_for_nminus1 = int(network_masks.line_for_nminus1.sum())
    statistics.import_result.n_line_for_optimized = int(network_masks.line_for_optimized.sum())
    statistics.import_result.n_line_for_non_worsening = int(network_masks.line_for_non_worsening.sum())
    statistics.import_result.n_line_disconnectable = int(network_masks.line_disconnectable.sum())

    statistics.import_result.n_trafo_for_nminus1 = int(network_masks.trafo_for_nminus1.sum())
    statistics.import_result.n_trafo_for_optimized = int(network_masks.trafo_for_optimized.sum())
    statistics.import_result.n_trafo_for_non_worsening = int(network_masks.trafo_for_non_worsening.sum())
    statistics.import_result.n_trafo_disconnectable = int(network_masks.trafo_disconnectable.sum())
    statistics.import_result.n_tie_line_for_nminus1 = int(network_masks.tie_line_for_nminus1.sum())
    statistics.import_result.n_tie_line_for_optimized = int(network_masks.tie_line_for_optimized.sum())
    statistics.import_result.n_tie_line_for_non_worsening = int(network_masks.tie_line_for_non_worsening.sum())
    statistics.import_result.n_tie_line_disconnectable = int(network_masks.tie_line_disconnectable.sum())
    statistics.import_result.n_boundary_line_for_nminus1 = int(network_masks.boundary_line_for_nminus1.sum())
    statistics.import_result.n_generator_for_nminus1 = int(network_masks.generator_for_nminus1.sum())
    statistics.import_result.n_load_for_nminus1 = int(network_masks.load_for_nminus1.sum())
    statistics.import_result.n_switch_for_nminus1 = int(network_masks.switch_for_nminus1.sum())
    statistics.import_result.n_switch_for_optimized = int(network_masks.switch_for_optimized.sum())
    statistics.import_result.n_switch_for_non_worsening = int(network_masks.switch_for_non_worsening.sum())

toop_engine_importer.pypowsybl_import.artificial_limits

Create the artificial optimization limits of the monitored branches and measure the unsplit grid against them.

The limits are derived from the worst currents of a security analysis, the branch role (optimized or non-worsening) and the double limits. See docs/importer/pypowsybl/artificial_limits.md.

ARTIFICIAL_LIMITS module-attribute

ARTIFICIAL_LIMITS = (
    OperationalLimit.ARTIFICIAL_N_0,
    OperationalLimit.ARTIFICIAL_N_1,
)

ARTIFICIAL_DC_LIMITS module-attribute

ARTIFICIAL_DC_LIMITS = (
    OperationalLimit.ARTIFICIAL_DC_N_0,
    OperationalLimit.ARTIFICIAL_DC_N_1,
)

compute_artificial_limits

compute_artificial_limits(
    flow, limit, optimized, non_worsening, double_limits
)

Compute the optimization limit of every branch and side from its worst flow.

• Branches that are optimized and non-worsening ignore the lower and upper limit. They get the physical limit if overloaded, otherwise their flow plus non_worsening_margin (capped at the physical limit), at least minimum_lower_physical_non_worsening_limit times the physical limit.
• Optimized branches above the upper limit get the upper limit.
• Non-worsening branches get their flow plus non_worsening_margin, at least minimum_lower_physical_non_worsening_limit times the physical limit.
• Other optimized branches get the larger of the lower limit and their flow.

PARAMETER	DESCRIPTION
flow	The worst absolute flow of the case. TYPE: Series
limit	The physical limit of the case, in the unit of the flow. TYPE: Series
optimized	Whether the branch is optimized. TYPE: Series
non_worsening	Whether the branch is non-worsening. TYPE: Series
double_limits	The lower and upper limit and the minimum non-worsening limit, relative to the physical limit. TYPE: DoubleLimits

RETURNS	DESCRIPTION
Series	The optimization limits. NaN for branches that are neither optimized nor non-worsening and for branches without a flow or a physical limit.

Source code in packages/importer_pkg/src/toop_engine_importer/pypowsybl_import/artificial_limits.py

def compute_artificial_limits(
    flow: pd.Series,
    limit: pd.Series,
    optimized: pd.Series,
    non_worsening: pd.Series,
    double_limits: DoubleLimits,
) -> pd.Series:
    """Compute the optimization limit of every branch and side from its worst flow.

    - Branches that are optimized and non-worsening ignore the lower and upper limit.
      They get the physical limit if overloaded, otherwise their flow plus ``non_worsening_margin`` (capped at
      the physical limit), at least ``minimum_lower_physical_non_worsening_limit`` times the physical limit.
    - Optimized branches above the upper limit get the upper limit.
    - Non-worsening branches get their flow plus ``non_worsening_margin``, at least
      ``minimum_lower_physical_non_worsening_limit`` times the physical limit.
    - Other optimized branches get the larger of the lower limit and their flow.

    Parameters
    ----------
    flow: pd.Series
        The worst absolute flow of the case.
    limit: pd.Series
        The physical limit of the case, in the unit of the flow.
    optimized: pd.Series
        Whether the branch is optimized.
    non_worsening: pd.Series
        Whether the branch is non-worsening.
    double_limits: DoubleLimits
        The lower and upper limit and the minimum non-worsening limit, relative to the physical limit.

    Returns
    -------
    pd.Series
        The optimization limits. NaN for branches that are neither optimized nor non-worsening and for branches
        without a flow or a physical limit.
    """
    upper_limit = limit * double_limits.upper
    non_worsening_limit = (flow * (1 + double_limits.non_worsening_margin)).clip(
        lower=limit * double_limits.minimum_lower_physical_non_worsening_limit
    )
    optimized_limit = flow.clip(lower=limit * double_limits.lower)
    optimized_limit = optimized_limit.where(flow <= upper_limit, upper_limit)
    both_limit = non_worsening_limit.clip(upper=limit).where(flow <= limit, limit)

    limits = non_worsening_limit.where(non_worsening)
    limits = limits.mask(optimized, optimized_limit)
    limits = limits.mask(optimized & non_worsening, both_limit)

    return limits.where(flow.notna() & limit.notna())

get_artificial_limits_for_case

get_artificial_limits_for_case(
    branches_df, worst_case, double_limits, limit
)

Get the optimization limits of the monitored branches for one limit.

PARAMETER	DESCRIPTION
branches_df	The branches dataframe with the columns type, i1, i2, optimized, non_worsening, the limits and group names of the case and the boundary lines of the tie lines. TYPE: DataFrame
worst_case	The worst absolute currents with the columns element, side, n0 and n1. TYPE: DataFrame
double_limits	The lower and upper limit, relative to the physical limit. TYPE: DoubleLimits
limit	The limit to create, e.g. ARTIFICIAL_N_0 or ARTIFICIAL_DC_N_1. It defines the case, name and acceptable duration. TYPE: OperationalLimit

RETURNS	DESCRIPTION
list[DataFrame]	A list of dataframes in the required format for create_operational_limits.

Source code in packages/importer_pkg/src/toop_engine_importer/pypowsybl_import/artificial_limits.py

def get_artificial_limits_for_case(
    branches_df: pd.DataFrame,
    worst_case: pd.DataFrame,
    double_limits: DoubleLimits,
    limit: OperationalLimit,
) -> list[pd.DataFrame]:
    """Get the optimization limits of the monitored branches for one limit.

    Parameters
    ----------
    branches_df: pd.DataFrame
        The branches dataframe with the columns type, i1, i2, optimized, non_worsening, the limits and group names
        of the case and the boundary lines of the tie lines.
    worst_case: pd.DataFrame
        The worst absolute currents with the columns element, side, n0 and n1.
    double_limits: DoubleLimits
        The lower and upper limit, relative to the physical limit.
    limit: OperationalLimit
        The limit to create, e.g. ARTIFICIAL_N_0 or ARTIFICIAL_DC_N_1. It defines the case, name and acceptable duration.

    Returns
    -------
    list[pd.DataFrame]
        A list of dataframes in the required format for create_operational_limits.
    """
    case = limit.case
    sides = (BranchSide.ONE, BranchSide.TWO)
    flows = {side: _get_worst_flows(branches_df, worst_case, side, case) for side in sides}

    artificial_limits = []
    for element_type in ("LINE", "TWO_WINDINGS_TRANSFORMER"):
        elements_df = branches_df[branches_df["type"] == element_type]
        if elements_df.empty:
            continue
        for side in sides:
            new_limit = compute_artificial_limits(
                flow=flows[side][elements_df.index],
                limit=elements_df[f"{case}_i{side.value}_max"],
                optimized=elements_df["optimized"],
                non_worsening=elements_df["non_worsening"],
                double_limits=double_limits,
            )
            artificial_limits.append(
                create_current_limits_df(
                    new_limit,
                    element_type=element_type,
                    side=side,
                    limit=limit,
                    group_names=elements_df[f"{case}_group_name_{side.value}"],
                )
            )

    tie_lines_df = branches_df[branches_df["type"] == "TIE_LINE"]
    if not tie_lines_df.empty:
        tie_lines_df = tie_lines_df.assign(
            new_limit=compute_artificial_limits(
                flow=pd.concat(flows, axis=1).max(axis=1)[tie_lines_df.index],
                limit=tie_lines_df[[f"{case}_i1_max", f"{case}_i2_max"]].min(axis=1),
                optimized=tie_lines_df["optimized"],
                non_worsening=tie_lines_df["non_worsening"],
                double_limits=double_limits,
            )
        )
        for side_value, dangling_line_col in zip([1, 2], ["boundary_line1_id", "boundary_line2_id"], strict=True):
            dangling_df = tie_lines_df.set_index(dangling_line_col)
            artificial_limits.append(
                create_current_limits_df(
                    dangling_df["new_limit"],
                    element_type="BOUNDARY_LINE",
                    side=BranchSide.NONE,
                    limit=limit,
                    group_names=dangling_df[f"{case}_group_name_{side_value}"],
                )
            )
    return artificial_limits

add_default_permanent_limits

add_default_permanent_limits(
    network, monitored_branch_ids, default_permanent
)

Add a permanent limit to the monitored branches that have None on a branch SIDE.

The limits of tie lines are set on their boundary lines.

PARAMETER	DESCRIPTION
network	The network to add the limits to. It is modified in place. TYPE: Network
monitored_branch_ids	The ids of the monitored branches. TYPE: list[str]
default_permanent	The permanent limit in A of the added limits. TYPE: float

Source code in packages/importer_pkg/src/toop_engine_importer/pypowsybl_import/artificial_limits.py

def add_default_permanent_limits(network: Network, monitored_branch_ids: list[str], default_permanent: float) -> None:
    """Add a permanent limit to the monitored branches that have None on a branch SIDE.

    The limits of tie lines are set on their boundary lines.

    Parameters
    ----------
    network: Network
        The network to add the limits to. It is modified in place.
    monitored_branch_ids: list[str]
        The ids of the monitored branches.
    default_permanent: float
        The permanent limit in A of the added limits.
    """
    existing_limits = network.get_operational_limits().reset_index()
    has_permanent = set(
        existing_limits.loc[existing_limits["name"] == OperationalLimit.PERMANENT, ["element_id", "side"]].itertuples(
            index=False
        )
    )
    branches = network.get_branches(attributes=["type"])
    branches = branches[branches.index.isin(monitored_branch_ids)]
    tie_lines = network.get_tie_lines(attributes=["boundary_line1_id", "boundary_line2_id"])
    missing = []
    for element_id, element_type in branches["type"].items():
        for side in (BranchSide.ONE, BranchSide.TWO):
            if (element_id, side.name) in has_permanent:
                continue
            if element_type == "TIE_LINE":
                missing.append((tie_lines.at[element_id, f"boundary_line{side.value}_id"], BranchSide.NONE.name))
            elif element_type in ("LINE", "TWO_WINDINGS_TRANSFORMER"):
                missing.append((element_id, side.name))
    if not missing:
        return
    # The dataframe form silently creates nothing for elements without any limit, the keyword form does not
    network.create_loading_limits(
        element_id=[element_id for element_id, _ in missing],
        side=[side for _, side in missing],
        name=[OperationalLimit.DEFAULT_PERMANENT.value] * len(missing),
        type=["CURRENT"] * len(missing),
        value=[default_permanent] * len(missing),
        acceptable_duration=[-1] * len(missing),
    )

create_artificial_limits

create_artificial_limits(
    network,
    nminus1_definition,
    worst_case_currents,
    double_limits,
    limits=ARTIFICIAL_LIMITS,
)

Create the optimization limits of the monitored branches in the network.

The limits replace the limits of the same element, side and name, see compute_artificial_limits.

PARAMETER	DESCRIPTION
network	The network to create the limits for. The loadflow calculation needs to have happened TYPE: Network
nminus1_definition	The N-1 definition with the optimized and non-worsening monitored branches TYPE: Nminus1Definition
worst_case_currents	The worst absolute currents with the columns element, side, n0 and n1, see extract_worst_case_branch_results_polars TYPE: DataFrame
double_limits	The lower and upper limit, relative to the physical limit. TYPE: DoubleLimits
limits	The limits to create, ARTIFICIAL_LIMITS by default. Use ARTIFICIAL_DC_LIMITS for DC results. TYPE: tuple[OperationalLimit, ...] DEFAULT: ARTIFICIAL_LIMITS

RETURNS	DESCRIPTION
DataFrame	The new limits for the network including the already existing ones

Source code in packages/importer_pkg/src/toop_engine_importer/pypowsybl_import/artificial_limits.py

def create_artificial_limits(
    network: Network,
    nminus1_definition: Nminus1Definition,
    worst_case_currents: pl.DataFrame,
    double_limits: DoubleLimits,
    limits: tuple[OperationalLimit, ...] = ARTIFICIAL_LIMITS,
) -> pd.DataFrame:
    """Create the optimization limits of the monitored branches in the network.

    The limits replace the limits of the same element, side and name, see compute_artificial_limits.

    Parameters
    ----------
    network: Network
        The network to create the limits for. The loadflow calculation needs to have happened
    nminus1_definition: Nminus1Definition
        The N-1 definition with the optimized and non-worsening monitored branches
    worst_case_currents: pl.DataFrame
        The worst absolute currents with the columns element, side, n0 and n1,
        see extract_worst_case_branch_results_polars
    double_limits: DoubleLimits
        The lower and upper limit, relative to the physical limit.
    limits: tuple[OperationalLimit, ...]
        The limits to create, ARTIFICIAL_LIMITS by default. Use ARTIFICIAL_DC_LIMITS for DC results.

    Returns
    -------
    pd.DataFrame
        The new limits for the network including the already existing ones
    """
    monitored_branches = [element for element in nminus1_definition.monitored_elements if element.kind == "branch"]
    existing_limits = network.get_operational_limits()
    branches_df = get_branches_including_limits_and_dangling_lines(
        network.get_branches(attributes=["type", "i1", "i2"]),
        existing_limits,
        network.get_tie_lines(attributes=["boundary_line1_id", "boundary_line2_id"]),
    )
    monitored_elements_df = pd.DataFrame(
        {
            "optimized": [element.optimized for element in monitored_branches],
            "non_worsening": [element.non_worsening for element in monitored_branches],
        },
        index=[element.id for element in monitored_branches],
        dtype=bool,
    )
    branches_df[["optimized", "non_worsening"]] = monitored_elements_df.reindex(branches_df.index, fill_value=False)
    worst_case = pd.DataFrame(worst_case_currents.to_dict(as_series=False))

    new_limits = []
    for limit in limits:
        new_limits += get_artificial_limits_for_case(branches_df, worst_case, double_limits, limit)
    if not new_limits:
        return existing_limits
    new_limits_df = pd.concat(new_limits)

    # Exclude tie lines, since they cant be directly updated
    old_limits_df = existing_limits[existing_limits.element_type != "TIE_LINE"]
    key_columns = ["element_id", "side", "name"]
    replaced = pd.MultiIndex.from_frame(old_limits_df.reset_index()[key_columns]).isin(
        pd.MultiIndex.from_frame(new_limits_df.reset_index()[key_columns])
    )
    kept_limits_df = old_limits_df[~replaced]
    updated_limits_df = pd.concat([kept_limits_df, _shift_occupied_durations(new_limits_df, kept_limits_df)])
    updated_limits_df = updated_limits_df.drop(columns=["element_type"])
    updated_limits_df = updated_limits_df[
        updated_limits_df.index.get_level_values("element_id").isin(existing_limits.index.get_level_values("element_id"))
    ]
    network.create_operational_limits(updated_limits_df.reset_index("acceptable_duration"))
    return updated_limits_df

attach_artificial_limits_to_results

attach_artificial_limits_to_results(
    network,
    nminus1_definition,
    security_analysis_results,
    double_limits,
    limits=ARTIFICIAL_LIMITS,
)

Create the artificial limits from the worst currents and attach the new limits to the results.

Works for AC and DC results alike, the limits tell them apart.

Currents do not depend on limits, so the returned results equal those of an analysis run after the limits were created, without running one.

PARAMETER	DESCRIPTION
network	The network to create the artificial limits on. It is modified in place. TYPE: Network
nminus1_definition	The N-1 definition with the optimized and non-worsening monitored branches. TYPE: Nminus1Definition
security_analysis_results	The results of the AC security analysis of the network. TYPE: LoadflowResultsPolars
double_limits	The lower and upper limit, relative to the physical limit. TYPE: DoubleLimits
limits	The limits to create. Use ARTIFICIAL_DC_LIMITS for DC results. TYPE: tuple[OperationalLimit, ...] DEFAULT: ARTIFICIAL_LIMITS

RETURNS	DESCRIPTION
LoadflowResultsPolars	The security analysis results with the branch limits of the monitored branches, including the artificial ones.

Source code in packages/importer_pkg/src/toop_engine_importer/pypowsybl_import/artificial_limits.py

def attach_artificial_limits_to_results(
    network: Network,
    nminus1_definition: Nminus1Definition,
    security_analysis_results: LoadflowResultsPolars,
    double_limits: DoubleLimits,
    limits: tuple[OperationalLimit, ...] = ARTIFICIAL_LIMITS,
) -> LoadflowResultsPolars:
    """Create the artificial limits from the worst currents and attach the new limits to the results.

    Works for AC and DC results alike, the limits tell them apart.

    Currents do not depend on limits, so the returned results equal those of an analysis run after the limits were
    created, without running one.

    Parameters
    ----------
    network: Network
        The network to create the artificial limits on. It is modified in place.
    nminus1_definition: Nminus1Definition
        The N-1 definition with the optimized and non-worsening monitored branches.
    security_analysis_results: LoadflowResultsPolars
        The results of the AC security analysis of the network.
    double_limits: DoubleLimits
        The lower and upper limit, relative to the physical limit.
    limits: tuple[OperationalLimit, ...]
        The limits to create. Use ARTIFICIAL_DC_LIMITS for DC results.

    Returns
    -------
    LoadflowResultsPolars
        The security analysis results with the branch limits of the monitored branches, including the artificial ones.
    """
    worst_case_currents = extract_worst_case_branch_results_polars(security_analysis_results, nminus1_definition, timestep=0)
    create_artificial_limits(
        network,
        nminus1_definition,
        worst_case_currents,
        double_limits,
        limits=limits,
    )
    branch_limits = get_monitored_branch_limits_polars(network, nminus1_definition, timestep=0)
    return security_analysis_results.model_copy(update={"branch_limits": branch_limits})

compute_initial_metrics

compute_initial_metrics(
    loadflow_results, nminus1_definition
)

Compute the metrics of the unsplit grid.

PARAMETER	DESCRIPTION
loadflow_results	The AC N-1 results of the unsplit grid, with the branch limits to measure against. TYPE: LoadflowResultsPolars
nminus1_definition	The N-1 definition, which holds the base case. TYPE: Nminus1Definition

RETURNS	DESCRIPTION
dict[MetricType, float]	The metrics. Metrics of limits that no branch has are left out.

Source code in packages/importer_pkg/src/toop_engine_importer/pypowsybl_import/artificial_limits.py

def compute_initial_metrics(
    loadflow_results: LoadflowResultsPolars, nminus1_definition: Nminus1Definition
) -> dict[MetricType, float]:
    """Compute the metrics of the unsplit grid.

    Parameters
    ----------
    loadflow_results: LoadflowResultsPolars
        The AC N-1 results of the unsplit grid, with the branch limits to measure against.
    nminus1_definition: Nminus1Definition
        The N-1 definition, which holds the base case.

    Returns
    -------
    dict[MetricType, float]
        The metrics. Metrics of limits that no branch has are left out.
    """
    base_case = nminus1_definition.base_case
    metrics = compute_metrics(
        loadflow_results,
        base_case_id=base_case.id if base_case is not None else None,
        displayed_branch_ids=nminus1_definition.displayed_branch_ids,
    )
    return {name: value for name, value in metrics.items() if value is not None}

compute_dc_artificial_limit_metrics

compute_dc_artificial_limit_metrics(
    network,
    nminus1_definition,
    double_limits,
    lf_params,
    job_id,
    n_processes,
    batch_size,
)

Run a DC security analysis, create the artificial DC limits from it and measure the unsplit grid against them.

The DC solver reads these limits instead of the AC based ones, as they match its DC flows.

PARAMETER	DESCRIPTION
network	The network to create the artificial DC limits on. Only its limits are modified. TYPE: Network
nminus1_definition	The N-1 definition with the optimized and non-worsening monitored branches. TYPE: Nminus1Definition
double_limits	The lower and upper limit, relative to the physical limit. TYPE: DoubleLimits
lf_params	The loadflow parameters of the import. TYPE: Parameters
job_id	The job id stored with the results. TYPE: str
n_processes	The number of threads of the security analysis. TYPE: int
batch_size	Maximum number of outage contingencies per sequential analysis. TYPE: Optional[int]

RETURNS	DESCRIPTION
dict[MetricType, float]	The DC metrics of the unsplit grid against the artificial DC limits, named like the artificial limit metrics. Metrics of limits that no branch has are left out.

Source code in packages/importer_pkg/src/toop_engine_importer/pypowsybl_import/artificial_limits.py

def compute_dc_artificial_limit_metrics(
    network: Network,
    nminus1_definition: Nminus1Definition,
    double_limits: DoubleLimits,
    lf_params: pypowsybl.loadflow.Parameters,
    job_id: str,
    n_processes: int,
    batch_size: Optional[int],
) -> dict[MetricType, float]:
    """Run a DC security analysis, create the artificial DC limits from it and measure the unsplit grid against them.

    The DC solver reads these limits instead of the AC based ones, as they match its DC flows.

    Parameters
    ----------
    network: Network
        The network to create the artificial DC limits on. Only its limits are modified.
    nminus1_definition: Nminus1Definition
        The N-1 definition with the optimized and non-worsening monitored branches.
    double_limits: DoubleLimits
        The lower and upper limit, relative to the physical limit.
    lf_params: pypowsybl.loadflow.Parameters
        The loadflow parameters of the import.
    job_id: str
        The job id stored with the results.
    n_processes: int
        The number of threads of the security analysis.
    batch_size: Optional[int]
        Maximum number of outage contingencies per sequential analysis.

    Returns
    -------
    dict[MetricType, float]
        The DC metrics of the unsplit grid against the artificial DC limits, named like the artificial limit metrics.
        Metrics of limits that no branch has are left out.
    """
    # A copy, since the DC loadflow also overwrites the boundary voltages of dangling lines, which are not variant-aware
    # TODO fix deep copy -> run_contingency_analysis_powsybl() modfies network
    dc_results = run_contingency_analysis_powsybl(
        net=deepcopy(network),
        n_minus_1_definition=nminus1_definition,
        job_id=job_id,
        timestep=0,
        method="dc",
        polars=True,
        lf_params=lf_params,
        n_processes=n_processes,
        batch_size=batch_size,
    )
    assert isinstance(dc_results, LoadflowResultsPolars), "Security analysis did not return LoadflowResultsPolars"
    dc_loadflow = attach_artificial_limits_to_results(
        network=network,
        nminus1_definition=nminus1_definition,
        security_analysis_results=dc_results,
        double_limits=double_limits,
        limits=ARTIFICIAL_DC_LIMITS,
    )
    # The metrics measure against the artificial limits, so expose the DC limits under their names
    dc_limit_names = {
        dc_limit.value: limit.value for dc_limit, limit in zip(ARTIFICIAL_DC_LIMITS, ARTIFICIAL_LIMITS, strict=True)
    }
    branch_limits = dc_loadflow.branch_limits.filter(
        ~pl.col("limit_name").is_in(list(dc_limit_names.values()))
    ).with_columns(pl.col("limit_name").replace(dc_limit_names))
    return compute_initial_metrics(dc_loadflow.model_copy(update={"branch_limits": branch_limits}), nminus1_definition)

toop_engine_importer.pypowsybl_import.contingency_from_file.nminus1_definition_input

Grid-file stage of the input N-1 definition: validate a Pydantic dump of Nminus1Definition against a grid.

logger module-attribute

logger = structlog.get_logger(__name__)

KIND_BY_ELEMENT_TYPE module-attribute

KIND_BY_ELEMENT_TYPE = {
    "LINE": "branch",
    "TWO_WINDINGS_TRANSFORMER": "branch",
    "THREE_WINDINGS_TRANSFORMER": "branch",
    "HVDC_LINE": "branch",
    "TIE_LINE": "branch",
    "GENERATOR": "injection",
    "LOAD": "injection",
    "BOUNDARY_LINE": "injection",
    "SHUNT_COMPENSATOR": "injection",
    "BUS": "bus",
    "BUSBAR_SECTION": "bus",
    "SWITCH": "switch",
}

GridElementT module-attribute

GridElementT = TypeVar('GridElementT', bound=GridElement)

filter_nminus1_definition_to_network

filter_nminus1_definition_to_network(definition, network)

Validate an input N-1 definition against the grid, dropping what the grid does not contain.

The network must be the grid before the three-winding transformer conversion, because three-winding transformers are referenced by their original id.

PARAMETER	DESCRIPTION
definition	The input N-1 definition with Powsybl element ids. TYPE: Nminus1Definition
network	The unconverted grid to validate against. TYPE: Network

RETURNS

DESCRIPTION

Nminus1Definition

The definition restricted to the grid. Every change is logged as a warning: elements whose id is not in the grid are dropped, from contingencies and monitored elements; elements whose type or kind differ from the grid are corrected to the grid's values; contingencies that lose all their elements are dropped, while contingencies that were empty (the base case) are kept; SPPS rules are dropped as a whole if their contingency was dropped or if any condition or action element is not in the grid.

Source code in packages/importer_pkg/src/toop_engine_importer/pypowsybl_import/contingency_from_file/nminus1_definition_input.py

def filter_nminus1_definition_to_network(definition: Nminus1Definition, network: Network) -> Nminus1Definition:
    """Validate an input N-1 definition against the grid, dropping what the grid does not contain.

    The network must be the grid **before** the three-winding transformer conversion, because three-winding
    transformers are referenced by their original id.

    Parameters
    ----------
    definition : Nminus1Definition
        The input N-1 definition with Powsybl element ids.
    network : pypowsybl.network.Network
        The unconverted grid to validate against.

    Returns
    -------
    Nminus1Definition
        The definition restricted to the grid. Every change is logged as a warning:

        - elements whose id is not in the grid are dropped, from contingencies and monitored elements;
        - elements whose type or kind differ from the grid are corrected to the grid's values;
        - contingencies that lose all their elements are dropped, while contingencies that were empty
          (the base case) are kept;
        - SPPS rules are dropped as a whole if their contingency was dropped or if any condition or action
          element is not in the grid.
    """
    grid_element_types: dict[str, list[str]] = (
        get_all_element_names(network).groupby("grid_model_id", sort=False)["element_type"].agg(list).to_dict()
    )

    contingencies = []
    for contingency in definition.contingencies:
        elements = _align_elements_with_grid(contingency.elements, grid_element_types, contingency.id)
        if contingency.elements and not elements:
            logger.warning(
                "empty_nminus1_contingency_dropped", contingency_id=contingency.id, contingency_name=contingency.name
            )
        else:
            contingencies.append(contingency.model_copy(update={"elements": elements}))

    spps_rules = None
    if definition.spps_rules is not None:
        contingency_ids = {contingency.id for contingency in contingencies}
        spps_rules = [
            rule
            for rule in definition.spps_rules
            if _spps_rule_is_resolvable(rule, contingency_ids, set(grid_element_types))
        ] or None

    filtered_definition = definition.model_copy(
        update={
            "contingencies": contingencies,
            "monitored_elements": _align_elements_with_grid(definition.monitored_elements, grid_element_types),
            "spps_rules": spps_rules,
        }
    )
    # Re-validate, which model_copy skips, so the SPPS integrity check runs on the result
    return Nminus1Definition.model_validate(filtered_definition.model_dump())

load_nminus1_definition_for_network

load_nminus1_definition_for_network(
    network, file_path, filesystem
)

Load an input N-1 definition JSON and validate it against the grid.

PARAMETER	DESCRIPTION
network	The unconverted grid to validate against, see :func:filter_nminus1_definition_to_network. TYPE: Network
file_path	Path to the Pydantic JSON of an Nminus1Definition. TYPE: str or Path
filesystem	Filesystem from which to read file_path. TYPE: AbstractFileSystem

RETURNS	DESCRIPTION
Nminus1Definition	The grid-validated N-1 definition.

Source code in packages/importer_pkg/src/toop_engine_importer/pypowsybl_import/contingency_from_file/nminus1_definition_input.py

def load_nminus1_definition_for_network(
    network: Network,
    file_path: str | Path,
    filesystem: AbstractFileSystem,
) -> Nminus1Definition:
    """Load an input N-1 definition JSON and validate it against the grid.

    Parameters
    ----------
    network : pypowsybl.network.Network
        The unconverted grid to validate against, see :func:`filter_nminus1_definition_to_network`.
    file_path : str or pathlib.Path
        Path to the Pydantic JSON of an `Nminus1Definition`.
    filesystem : fsspec.AbstractFileSystem
        Filesystem from which to read ``file_path``.

    Returns
    -------
    Nminus1Definition
        The grid-validated N-1 definition.
    """
    definition = load_nminus1_definition_fs(filesystem=filesystem, file_path=file_path)
    return filter_nminus1_definition_to_network(definition, network)

toop_engine_importer.pypowsybl_import.contingency_from_file.nminus1_definition_conversion

Convert three-winding transformers in an N-1 definition to their two-winding legs <id>-Leg1/2/3.

This is the last stage of the N-1 definition pipeline before the definition is saved next to the converted grid.

logger module-attribute

logger = structlog.get_logger(__name__)

SppsItemT module-attribute

SppsItemT = TypeVar('SppsItemT', bound=BaseModel)

convert_three_winding_transformers_in_nminus1_definition

convert_three_winding_transformers_in_nminus1_definition(
    definition, network
)

Convert every three-winding transformer reference in an N-1 definition to its two-winding legs.

Contingency and monitored elements on a converted three-winding transformer are replaced by its three legs. SPPS conditions and actions on it are copied once per leg, regardless of the rule's condition_logic, so an ANY rule triggers as soon as a single leg meets the condition. The conversion is idempotent.

PARAMETER	DESCRIPTION
definition	The N-1 definition, referencing three-winding transformers by their original id. TYPE: Nminus1Definition
network	The grid after replace_3_windings_transformers_with_3_2_windings_transformers. TYPE: Network

RETURNS	DESCRIPTION
Nminus1Definition	The converted definition. A three-winding transformer whose legs are not in the grid is kept and warned about.

Source code in packages/importer_pkg/src/toop_engine_importer/pypowsybl_import/contingency_from_file/nminus1_definition_conversion.py

def convert_three_winding_transformers_in_nminus1_definition(
    definition: Nminus1Definition, network: Network
) -> Nminus1Definition:
    """Convert every three-winding transformer reference in an N-1 definition to its two-winding legs.

    Contingency and monitored elements on a converted three-winding transformer are replaced by its three legs.
    SPPS conditions and actions on it are copied once per leg, regardless of the rule's ``condition_logic``, so an
    ``ANY`` rule triggers as soon as a single leg meets the condition. The conversion is idempotent.

    Parameters
    ----------
    definition : Nminus1Definition
        The N-1 definition, referencing three-winding transformers by their original id.
    network : pypowsybl.network.Network
        The grid **after** ``replace_3_windings_transformers_with_3_2_windings_transformers``.

    Returns
    -------
    Nminus1Definition
        The converted definition. A three-winding transformer whose legs are not in the grid is kept and warned about.
    """
    trafos = network.get_2_windings_transformers(attributes=["name"])
    legs = trafos[trafos.index.str.contains(TRAFO3W_LEG_PATTERN)].sort_index()
    legs_by_trafo3w: dict[str, list[GridElement]] = {}
    for leg_id, leg_name in legs["name"].items():
        legs_by_trafo3w.setdefault(get_trafo3w_id(leg_id), []).append(
            GridElement(id=leg_id, name=leg_name or "", type="TWO_WINDINGS_TRANSFORMER", kind="branch")
        )

    contingencies = [
        contingency.model_copy(
            update={"elements": _replace_by_legs(contingency.elements, legs_by_trafo3w, context="contingency")}
        )
        for contingency in definition.contingencies
    ]
    monitored_elements = _replace_by_legs(definition.monitored_elements, legs_by_trafo3w, context="monitored_element")
    spps_rules = definition.spps_rules and [
        rule.model_copy(
            update={
                "conditions": _copy_per_leg(rule.conditions, "condition_element_unique_id", legs_by_trafo3w),
                "actions": _copy_per_leg(rule.actions, "measure_element_unique_id", legs_by_trafo3w),
            }
        )
        for rule in definition.spps_rules
    ]
    converted_definition = definition.model_copy(
        update={"contingencies": contingencies, "monitored_elements": monitored_elements, "spps_rules": spps_rules}
    )
    return Nminus1Definition.model_validate(converted_definition.model_dump())

get_nminus1_definition_element_ids

get_nminus1_definition_element_ids(definition)

Collect the ids of all contingency, monitored, SPPS condition and SPPS action elements of an N-1 definition.

PARAMETER	DESCRIPTION
definition	The N-1 definition. TYPE: Nminus1Definition

RETURNS	DESCRIPTION
set[str]	The referenced element ids.

Source code in packages/importer_pkg/src/toop_engine_importer/pypowsybl_import/contingency_from_file/nminus1_definition_conversion.py

def get_nminus1_definition_element_ids(definition: Nminus1Definition) -> set[str]:
    """Collect the ids of all contingency, monitored, SPPS condition and SPPS action elements of an N-1 definition.

    Parameters
    ----------
    definition : Nminus1Definition
        The N-1 definition.

    Returns
    -------
    set[str]
        The referenced element ids.
    """
    element_ids = {element.id for contingency in definition.contingencies for element in contingency.elements}
    element_ids.update(element.id for element in definition.monitored_elements)
    for rule in definition.spps_rules or []:
        element_ids.update(condition.condition_element_unique_id for condition in rule.conditions)
        element_ids.update(action.measure_element_unique_id for action in rule.actions)
    return element_ids

toop_engine_importer.pypowsybl_import.network_reduction

Reduce the network based on voltage levels and a depth range.

get_voltage_level_ids_of_elements

get_voltage_level_ids_of_elements(net, element_ids)

Get the sorted, unique voltage levels the given elements are connected to.

PARAMETER	DESCRIPTION
net	The network containing the elements. TYPE: Network
element_ids	Ids of branches (both sides), injections (including busbar sections), HVDC lines (both converter stations), switches, bus-breaker buses or voltage levels. Ids not found in the network are ignored. TYPE: Iterable[str]

RETURNS	DESCRIPTION
list[str]	The voltage level ids.

Source code in packages/importer_pkg/src/toop_engine_importer/pypowsybl_import/network_reduction.py

def get_voltage_level_ids_of_elements(net: Network, element_ids: Iterable[str]) -> list[str]:
    """Get the sorted, unique voltage levels the given elements are connected to.

    Parameters
    ----------
    net : pypowsybl.network.Network
        The network containing the elements.
    element_ids : Iterable[str]
        Ids of branches (both sides), injections (including busbar sections), HVDC lines (both converter stations),
        switches, bus-breaker buses or voltage levels. Ids not found in the network are ignored.

    Returns
    -------
    list[str]
        The voltage level ids.
    """
    element_ids = set(element_ids)
    hvdc_lines = net.get_hvdc_lines(attributes=["converter_station1_id", "converter_station2_id"])
    element_ids.update(hvdc_lines[hvdc_lines.index.isin(element_ids)].to_numpy().ravel())
    branches = net.get_branches(attributes=["voltage_level1_id", "voltage_level2_id"])
    voltage_level_ids = set(branches[branches.index.isin(element_ids)].to_numpy().ravel())
    voltage_level_ids |= element_ids & set(net.get_voltage_levels(attributes=[]).index)
    for elements in (
        net.get_injections(attributes=["voltage_level_id"]),
        net.get_switches(attributes=["voltage_level_id"]),
        net.get_bus_breaker_view_buses(attributes=["voltage_level_id"]),
    ):
        voltage_level_ids.update(elements.loc[elements.index.isin(element_ids), "voltage_level_id"])
    return sorted(voltage_level_ids)

reduce_network_based_on_area_settings

reduce_network_based_on_area_settings(
    net, importer_parameters, extra_voltage_level_ids=()
)

Reduce the network based on area settings and range.

PARAMETER	DESCRIPTION
net	The network to be reduced. Note: The network is modified in place. TYPE: Network
importer_parameters	The importer parameters containing the area settings and range. TYPE: Union[UcteImporterParameters, CgmesImporterParameters]
extra_voltage_level_ids	Additional voltage levels to keep, regardless of the area settings, e.g. the voltage levels of the elements of an input N-1 definition. They are reduced with the same range as the area voltage levels. TYPE: Sequence[str] DEFAULT: ()

Source code in packages/importer_pkg/src/toop_engine_importer/pypowsybl_import/network_reduction.py

def reduce_network_based_on_area_settings(
    net: Network,
    importer_parameters: Union[UcteImporterParameters, CgmesImporterParameters],
    extra_voltage_level_ids: Sequence[str] = (),
) -> None:
    """Reduce the network based on area settings and range.

    Parameters
    ----------
    net : pypowsybl.network.Network
        The network to be reduced.
        Note: The network is modified in place.
    importer_parameters : Union[UcteImporterParameters, CgmesImporterParameters]
        The importer parameters containing the area settings and range.
    extra_voltage_level_ids : Sequence[str]
        Additional voltage levels to keep, regardless of the area settings, e.g. the voltage levels of the elements
        of an input N-1 definition. They are reduced with the same range as the area voltage levels.
    """
    view_area = importer_parameters.area_settings.view_area
    control_area = importer_parameters.area_settings.control_area
    nminus1_area = importer_parameters.area_settings.nminus1_area
    area_codes = list(set(view_area + control_area + nminus1_area))

    if importer_parameters.data_type == "cgmes":
        voltage_level_ids = get_potentially_relevant_voltage_levels_cgmes(
            net=net,
            area_codes=area_codes,
            cutoff_voltage=importer_parameters.area_settings.cutoff_voltage,
            # We need the full area, this short list may only have a few voltage levels
            select_by_voltage_level_id_list=None,
        )
    elif importer_parameters.data_type == "ucte":
        voltage_level_ids = get_potentially_relevant_voltage_levels_ucte(
            net=net,
            area_codes=area_codes,
            cutoff_voltage=importer_parameters.area_settings.cutoff_voltage,
            # We need the full area, this short list may only have a few voltage levels
            select_by_voltage_level_id_list=None,
        )
    else:
        raise ValueError(f"Unsupported data type: {importer_parameters.data_type}")

    voltage_level_ids = list(dict.fromkeys([*voltage_level_ids, *extra_voltage_level_ids]))

    # get vl_depths with importer setting
    vl_depths = [(vl_id, importer_parameters.network_reduction_voltage_level_range) for vl_id in voltage_level_ids]
    # Keep PowSyBl's load equivalents for ordinary lines crossing the
    # reduction boundary.  A generated boundary line contains half of the
    # original line impedance, but PowSyBl initializes its p0/q0 with the
    # retained terminal flow.  That is not an AC-equivalent setpoint: on the
    # next loadflow, losses and shunt consumption of the half-line change the
    # retained-area flows.
    #
    # Tie-line endpoints are already BoundaryLines and remain in the network
    # after their TieLine is removed.  Their p0/q0 values are corrected before
    # reduction by set_tie_line_boundary_equivalents, so enabling creation of
    # additional boundary lines here is neither needed nor flow preserving.
    # set with_boundary_lines=False
    net.reduce_by_ids_and_depths(vl_depths=vl_depths, with_boundary_lines=False)

Importer Network Graph Pandapower

toop_engine_importer.network_graph

Importer-specific network-graph helpers.

__all__ module-attribute

__all__ = [
    "get_branch_df",
    "get_network_graph",
    "get_network_graph_data",
    "get_nodes",
    "get_switches_df",
]

get_branch_df

get_branch_df(net, only_relevant_col=True)

Get the branches data from the pandapower network and return a df ready to use for the BranchSchema.

Note: A star equivalent transformation for three winding trafos is not needed before calling this module. The the graph module only needs the connection and does no calculation.

PARAMETER	DESCRIPTION
net	The pandapower network. TYPE: pandapower
only_relevant_col	Whether to return only the relevant columns, by default True relevant is determined by the default BranchSchema columns TYPE: bool DEFAULT: True

RETURNS	DESCRIPTION
branch_df	The DataFrame containing the branches in the format of the BranchSchema. TYPE: DataFrame[BranchSchema]

Source code in packages/importer_pkg/src/toop_engine_importer/network_graph/pandapower_network_to_graph.py

def get_branch_df(net: pandapowerNet, only_relevant_col: bool = True) -> pat.DataFrame[BranchSchema]:
    """Get the branches data from the pandapower network and return a df ready to use for the BranchSchema.

    Note: A star equivalent transformation for three winding trafos is not needed before calling this module.
    The the graph module only needs the connection and does no calculation.

    Parameters
    ----------
    net : pandapower
        The pandapower network.
    only_relevant_col : bool, optional
        Whether to return only the relevant columns, by default True
        relevant is determined by the default BranchSchema columns

    Returns
    -------
    branch_df : pat.DataFrame[BranchSchema]
        The DataFrame containing the branches in the format of the BranchSchema.
    """
    line_df = net.line.copy()
    line_df = get_edges_data(line_df, asset_type="line", only_relevant_col=only_relevant_col)

    impedances_df = net.impedance.copy()
    impedances_df = get_edges_data(impedances_df, asset_type="impedance", only_relevant_col=only_relevant_col)
    tcsc_df = net.tcsc.copy()
    tcsc_df = get_edges_data(tcsc_df, asset_type="tcsc", only_relevant_col=only_relevant_col)

    dclines = net.dcline.copy()
    dclines = get_edges_data(dclines, asset_type="dcline", only_relevant_col=only_relevant_col)

    transformers = net.trafo.copy()
    transformers = get_edges_data(transformers, asset_type="trafo", only_relevant_col=only_relevant_col)

    trafos3w1 = net.trafo3w.copy()
    trafos3w2 = net.trafo3w.copy()
    trafos3w2["lv_bus"] = trafos3w2["mv_bus"]
    trafos3w = pd.concat([trafos3w1, trafos3w2])
    trafos3w = get_edges_data(trafos3w, asset_type="trafo3w", only_relevant_col=only_relevant_col)

    branches_df = pd.concat([line_df, impedances_df, tcsc_df, dclines, transformers, trafos3w])
    branches_df.reset_index(drop=True, inplace=True)
    return branches_df

get_network_graph

get_network_graph(network_graph_data)

Get the network graph from the NetworkGraphData and run default filter.

PARAMETER	DESCRIPTION
network_graph_data	The NetworkGraphData containing the nodes, switches, branches and node_assets. TYPE: NetworkGraphData

RETURNS	DESCRIPTION
Graph	The network graph.

Source code in packages/importer_pkg/src/toop_engine_importer/network_graph/pandapower_network_to_graph.py

def get_network_graph(network_graph_data: NetworkGraphData) -> nx.Graph:
    """Get the network graph from the NetworkGraphData and run default filter.

    Parameters
    ----------
    network_graph_data : NetworkGraphData
        The NetworkGraphData containing the nodes, switches, branches and node_assets.

    Returns
    -------
    nx.Graph
        The network graph.
    """
    graph = generate_graph(network_graph_data)
    set_substation_id(graph=graph, network_graph_data=network_graph_data)
    run_default_filter_strategy(graph=graph)
    return graph

get_network_graph_data

get_network_graph_data(net, only_relevant_col=True)

Get the network graph from the pandapower network.

PARAMETER	DESCRIPTION
net	The pandapower network. TYPE: pandapower
only_relevant_col	Whether to return only the relevant columns, by default True relevant is determined by the default BranchSchema columns TYPE: bool DEFAULT: True

RETURNS	DESCRIPTION
net_graph	The network graph in the format of the NetworkGraph class. Contains the full network with all substations. TYPE: NetworkGraphData

Source code in packages/importer_pkg/src/toop_engine_importer/network_graph/pandapower_network_to_graph.py

def get_network_graph_data(net: pandapowerNet, only_relevant_col: bool = True) -> NetworkGraphData:
    """Get the network graph from the pandapower network.

    Parameters
    ----------
    net : pandapower
        The pandapower network.
    only_relevant_col : bool, optional
        Whether to return only the relevant columns, by default True
        relevant is determined by the default BranchSchema columns

    Returns
    -------
    net_graph : NetworkGraphData
        The network graph  in the format of the NetworkGraph class.
        Contains the full network with all substations.
    """
    branches_df = get_branch_df(net, only_relevant_col=only_relevant_col)
    nodes_df = get_nodes(net, only_relevant_col=only_relevant_col)
    switches_df = get_switches_df(net, only_relevant_col=only_relevant_col)
    logger.warning("generators are missing in the network graph - they are not implemented yet")
    network_graph_data = NetworkGraphData(nodes=nodes_df, switches=switches_df, branches=branches_df)
    add_graph_specific_data(network_graph_data)
    return network_graph_data

get_nodes

get_nodes(net, only_relevant_col=True)

Get the nodes data from the pandapower network and return a df ready to use for the NodeSchema.

PARAMETER	DESCRIPTION
net	The pandapower network. TYPE: pandapower
only_relevant_col	Whether to return only the relevant columns, by default True relevant is determined by the default NodeSchema columns TYPE: bool DEFAULT: True

RETURNS	DESCRIPTION
nodes_df	The DataFrame containing the nodes in the format of the NodeSchema. TYPE: DataFrame[NodeSchema]

Source code in packages/importer_pkg/src/toop_engine_importer/network_graph/pandapower_network_to_graph.py

def get_nodes(net: pandapowerNet, only_relevant_col: bool = True) -> pat.DataFrame[NodeSchema]:
    """Get the nodes data from the pandapower network and return a df ready to use for the NodeSchema.

    Parameters
    ----------
    net : pandapower
        The pandapower network.
    only_relevant_col : bool, optional
        Whether to return only the relevant columns, by default True
        relevant is determined by the default NodeSchema columns

    Returns
    -------
    nodes_df : pat.DataFrame[NodeSchema]
        The DataFrame containing the nodes in the format of the NodeSchema.
    """
    nodes_df = net.bus.copy()
    if "equipment" in nodes_df.columns:
        nodes_df.rename(columns={"equipment": "foreign_id"}, inplace=True)
    else:
        nodes_df["foreign_id"] = nodes_df["name"]
    nodes_df.rename(
        columns={"name": "grid_model_id", "type": "node_type", "vn_kv": "voltage_level", "zone": "system_operator"},
        inplace=True,
    )
    nodes_df["node_type"] = "node"
    logger.warning(
        "There is a bug in Pandapower where the 'Busbar_id' is not extracted from the CGMES data. See:"
        "https://github.com/e2nIEE/pandapower/issues/2517"
        "The next line will fail if the bug is not fixed."
        "add the following hotfix to pandapower at:"
        "pandapower/converter/cim/cim2pp/converter_classes/connectivitynodes/connectivityNodesCim16.py"
        "line 24:connectivity_nodes, eqssh_terminals = self._prepare_connectivity_nodes_cim16()"
        "new line:connectivity_nodes = connectivity_nodes.rename(columns={'busbar_id': 'Busbar_id', 'busbar_name': 'Busbar_name'})"  # noqa: E501
    )
    nodes_df.loc[~(nodes_df.Busbar_id.isna() | (nodes_df.Busbar_id == "")), "node_type"] = "busbar"

    nodes_df.loc[nodes_df["system_operator"].isna(), "system_operator"] = ""

    if "Substation_id" in nodes_df.columns:
        nodes_df.rename(columns={"Substation_id": "substation_id"}, inplace=True)
    else:
        nodes_df["substation_id"] = ""
    nodes_df.fillna({"foreign_id": ""}, inplace=True)
    nodes_df.fillna({"voltage_level": 0}, inplace=True)
    nodes_df["helper_node"] = False
    nodes_df["voltage_level"] = nodes_df["voltage_level"].astype(float)
    nodes_df["bus_id"] = nodes_df.index.astype(str)
    nodes_df["bus_breaker_bus_id"] = None
    if only_relevant_col:
        needed_col = list(NodeSchema.to_schema().columns.keys())
        nodes_df = nodes_df[needed_col]

    return nodes_df

get_switches_df

get_switches_df(net, only_relevant_col=True)

Get the switches data from the pandapower network and return a df ready to use for the SwitchSchema.

PARAMETER	DESCRIPTION
net	The pandapower network. TYPE: pandapower
only_relevant_col	Whether to return only the relevant columns, by default True relevant is determined by the default SwitchSchema columns TYPE: bool DEFAULT: True

RETURNS	DESCRIPTION
switches_df	The DataFrame containing the switches in the format of the SwitchSchema. TYPE: DataFrame[SwitchSchema]

Source code in packages/importer_pkg/src/toop_engine_importer/network_graph/pandapower_network_to_graph.py

def get_switches_df(net: pandapowerNet, only_relevant_col: bool = True) -> pat.DataFrame[SwitchSchema]:
    """Get the switches data from the pandapower network and return a df ready to use for the SwitchSchema.

    Parameters
    ----------
    net : pandapower
        The pandapower network.
    only_relevant_col : bool, optional
        Whether to return only the relevant columns, by default True
        relevant is determined by the default SwitchSchema columns

    Returns
    -------
    switches_df : pat.DataFrame[SwitchSchema]
        The DataFrame containing the switches in the format of the SwitchSchema.
    """
    switches_df = net.switch.copy()
    # get only switches that interconnect buses
    switches_df = switches_df[switches_df["et"] == "b"]
    # reverse the "closed" column to "open"
    switches_df["closed"] = ~switches_df["closed"]
    switches_df.rename(columns={"closed": "open"}, inplace=True)
    if "in_service" not in switches_df.columns:
        switches_df["in_service"] = True
    switches_df = get_edges_data(switches_df, asset_type="switch", only_relevant_col=only_relevant_col)
    return switches_df

toop_engine_importer.network_graph.pandapower_network_to_graph

Convert a pandapower network to a network graph.

logger module-attribute

logger = structlog.get_logger(__name__)

get_network_graph_data

get_network_graph_data(net, only_relevant_col=True)

Get the network graph from the pandapower network.

PARAMETER	DESCRIPTION
net	The pandapower network. TYPE: pandapower
only_relevant_col	Whether to return only the relevant columns, by default True relevant is determined by the default BranchSchema columns TYPE: bool DEFAULT: True

RETURNS	DESCRIPTION
net_graph	The network graph in the format of the NetworkGraph class. Contains the full network with all substations. TYPE: NetworkGraphData

Source code in packages/importer_pkg/src/toop_engine_importer/network_graph/pandapower_network_to_graph.py

def get_network_graph_data(net: pandapowerNet, only_relevant_col: bool = True) -> NetworkGraphData:
    """Get the network graph from the pandapower network.

    Parameters
    ----------
    net : pandapower
        The pandapower network.
    only_relevant_col : bool, optional
        Whether to return only the relevant columns, by default True
        relevant is determined by the default BranchSchema columns

    Returns
    -------
    net_graph : NetworkGraphData
        The network graph  in the format of the NetworkGraph class.
        Contains the full network with all substations.
    """
    branches_df = get_branch_df(net, only_relevant_col=only_relevant_col)
    nodes_df = get_nodes(net, only_relevant_col=only_relevant_col)
    switches_df = get_switches_df(net, only_relevant_col=only_relevant_col)
    logger.warning("generators are missing in the network graph - they are not implemented yet")
    network_graph_data = NetworkGraphData(nodes=nodes_df, switches=switches_df, branches=branches_df)
    add_graph_specific_data(network_graph_data)
    return network_graph_data

get_network_graph

get_network_graph(network_graph_data)

Get the network graph from the NetworkGraphData and run default filter.

PARAMETER	DESCRIPTION
network_graph_data	The NetworkGraphData containing the nodes, switches, branches and node_assets. TYPE: NetworkGraphData

RETURNS	DESCRIPTION
Graph	The network graph.

Source code in packages/importer_pkg/src/toop_engine_importer/network_graph/pandapower_network_to_graph.py

def get_network_graph(network_graph_data: NetworkGraphData) -> nx.Graph:
    """Get the network graph from the NetworkGraphData and run default filter.

    Parameters
    ----------
    network_graph_data : NetworkGraphData
        The NetworkGraphData containing the nodes, switches, branches and node_assets.

    Returns
    -------
    nx.Graph
        The network graph.
    """
    graph = generate_graph(network_graph_data)
    set_substation_id(graph=graph, network_graph_data=network_graph_data)
    run_default_filter_strategy(graph=graph)
    return graph

get_edges_data

get_edges_data(
    dataframe, asset_type, only_relevant_col=True
)

Get the edges data from the dataframe and return a df ready to use for the BranchSchema.

Can be used for switches, trafos and other branches.

PARAMETER	DESCRIPTION
dataframe	The DataFrame containing the edges. TYPE: DataFrame
asset_type	The type of the asset. TYPE: str
only_relevant_col	Whether to return only the relevant columns, by default True relevant is determined by the default BranchSchema columns TYPE: bool DEFAULT: True

RETURNS	DESCRIPTION
DataFrame	The DataFrame containing the edges in the format of the

Source code in packages/importer_pkg/src/toop_engine_importer/network_graph/pandapower_network_to_graph.py

def get_edges_data(dataframe: pd.DataFrame, asset_type: str, only_relevant_col: bool = True) -> pd.DataFrame:
    """Get the edges data from the dataframe and return a df ready to use for the BranchSchema.

    Can be used for switches, trafos and other branches.

    Parameters
    ----------
    dataframe : pd.DataFrame
        The DataFrame containing the edges.
    asset_type : str
        The type of the asset.
    only_relevant_col : bool, optional
        Whether to return only the relevant columns, by default True
        relevant is determined by the default BranchSchema columns

    Returns
    -------
    pd.DataFrame
        The DataFrame containing the edges in the format of the


    """
    needed_col = list(get_empty_dataframe_from_df_model(BranchSchema).columns)
    if asset_type == "switch":
        dataframe.rename(
            columns={
                "bus": "from_node",
                "element": "to_node",
            },
            inplace=True,
        )
        needed_col.append("open")
    elif asset_type in ["trafo", "trafo3w"]:
        dataframe["type"] = asset_type
        dataframe.rename(
            columns={
                "hv_bus": "from_node",
                "lv_bus": "to_node",
            },
            inplace=True,
        )
    else:
        dataframe["type"] = asset_type
        dataframe.rename(
            columns={
                "from_bus": "from_node",
                "to_bus": "to_node",
            },
            inplace=True,
        )
    if "equipment" in dataframe.columns:  # a parameter from the Frauenhofer script
        dataframe.rename(columns={"equipment": "foreign_id"}, inplace=True)
    else:
        dataframe["foreign_id"] = dataframe["name"]
    dataframe.rename(
        columns={
            "name": "grid_model_id",
            "type": "asset_type",
        },
        inplace=True,
    )
    dataframe.fillna({"foreign_id": ""}, inplace=True)
    cond = dataframe["foreign_id"] == ""
    dataframe.loc[cond, "foreign_id"] = dataframe.loc[cond, "grid_model_id"]
    dataframe["from_node"] = dataframe["from_node"].astype(int)
    dataframe["to_node"] = dataframe["to_node"].astype(int)
    if only_relevant_col:
        dataframe = dataframe[needed_col]
    return dataframe

get_nodes

get_nodes(net, only_relevant_col=True)

Get the nodes data from the pandapower network and return a df ready to use for the NodeSchema.

PARAMETER	DESCRIPTION
net	The pandapower network. TYPE: pandapower
only_relevant_col	Whether to return only the relevant columns, by default True relevant is determined by the default NodeSchema columns TYPE: bool DEFAULT: True

RETURNS	DESCRIPTION
nodes_df	The DataFrame containing the nodes in the format of the NodeSchema. TYPE: DataFrame[NodeSchema]

Source code in packages/importer_pkg/src/toop_engine_importer/network_graph/pandapower_network_to_graph.py

def get_nodes(net: pandapowerNet, only_relevant_col: bool = True) -> pat.DataFrame[NodeSchema]:
    """Get the nodes data from the pandapower network and return a df ready to use for the NodeSchema.

    Parameters
    ----------
    net : pandapower
        The pandapower network.
    only_relevant_col : bool, optional
        Whether to return only the relevant columns, by default True
        relevant is determined by the default NodeSchema columns

    Returns
    -------
    nodes_df : pat.DataFrame[NodeSchema]
        The DataFrame containing the nodes in the format of the NodeSchema.
    """
    nodes_df = net.bus.copy()
    if "equipment" in nodes_df.columns:
        nodes_df.rename(columns={"equipment": "foreign_id"}, inplace=True)
    else:
        nodes_df["foreign_id"] = nodes_df["name"]
    nodes_df.rename(
        columns={"name": "grid_model_id", "type": "node_type", "vn_kv": "voltage_level", "zone": "system_operator"},
        inplace=True,
    )
    nodes_df["node_type"] = "node"
    logger.warning(
        "There is a bug in Pandapower where the 'Busbar_id' is not extracted from the CGMES data. See:"
        "https://github.com/e2nIEE/pandapower/issues/2517"
        "The next line will fail if the bug is not fixed."
        "add the following hotfix to pandapower at:"
        "pandapower/converter/cim/cim2pp/converter_classes/connectivitynodes/connectivityNodesCim16.py"
        "line 24:connectivity_nodes, eqssh_terminals = self._prepare_connectivity_nodes_cim16()"
        "new line:connectivity_nodes = connectivity_nodes.rename(columns={'busbar_id': 'Busbar_id', 'busbar_name': 'Busbar_name'})"  # noqa: E501
    )
    nodes_df.loc[~(nodes_df.Busbar_id.isna() | (nodes_df.Busbar_id == "")), "node_type"] = "busbar"

    nodes_df.loc[nodes_df["system_operator"].isna(), "system_operator"] = ""

    if "Substation_id" in nodes_df.columns:
        nodes_df.rename(columns={"Substation_id": "substation_id"}, inplace=True)
    else:
        nodes_df["substation_id"] = ""
    nodes_df.fillna({"foreign_id": ""}, inplace=True)
    nodes_df.fillna({"voltage_level": 0}, inplace=True)
    nodes_df["helper_node"] = False
    nodes_df["voltage_level"] = nodes_df["voltage_level"].astype(float)
    nodes_df["bus_id"] = nodes_df.index.astype(str)
    nodes_df["bus_breaker_bus_id"] = None
    if only_relevant_col:
        needed_col = list(NodeSchema.to_schema().columns.keys())
        nodes_df = nodes_df[needed_col]

    return nodes_df

get_switches_df

get_switches_df(net, only_relevant_col=True)

Get the switches data from the pandapower network and return a df ready to use for the SwitchSchema.

PARAMETER	DESCRIPTION
net	The pandapower network. TYPE: pandapower
only_relevant_col	Whether to return only the relevant columns, by default True relevant is determined by the default SwitchSchema columns TYPE: bool DEFAULT: True

RETURNS	DESCRIPTION
switches_df	The DataFrame containing the switches in the format of the SwitchSchema. TYPE: DataFrame[SwitchSchema]

Source code in packages/importer_pkg/src/toop_engine_importer/network_graph/pandapower_network_to_graph.py

def get_switches_df(net: pandapowerNet, only_relevant_col: bool = True) -> pat.DataFrame[SwitchSchema]:
    """Get the switches data from the pandapower network and return a df ready to use for the SwitchSchema.

    Parameters
    ----------
    net : pandapower
        The pandapower network.
    only_relevant_col : bool, optional
        Whether to return only the relevant columns, by default True
        relevant is determined by the default SwitchSchema columns

    Returns
    -------
    switches_df : pat.DataFrame[SwitchSchema]
        The DataFrame containing the switches in the format of the SwitchSchema.
    """
    switches_df = net.switch.copy()
    # get only switches that interconnect buses
    switches_df = switches_df[switches_df["et"] == "b"]
    # reverse the "closed" column to "open"
    switches_df["closed"] = ~switches_df["closed"]
    switches_df.rename(columns={"closed": "open"}, inplace=True)
    if "in_service" not in switches_df.columns:
        switches_df["in_service"] = True
    switches_df = get_edges_data(switches_df, asset_type="switch", only_relevant_col=only_relevant_col)
    return switches_df

get_branch_df

get_branch_df(net, only_relevant_col=True)

Get the branches data from the pandapower network and return a df ready to use for the BranchSchema.

Note: A star equivalent transformation for three winding trafos is not needed before calling this module. The the graph module only needs the connection and does no calculation.

PARAMETER	DESCRIPTION
net	The pandapower network. TYPE: pandapower
only_relevant_col	Whether to return only the relevant columns, by default True relevant is determined by the default BranchSchema columns TYPE: bool DEFAULT: True

RETURNS	DESCRIPTION
branch_df	The DataFrame containing the branches in the format of the BranchSchema. TYPE: DataFrame[BranchSchema]

Source code in packages/importer_pkg/src/toop_engine_importer/network_graph/pandapower_network_to_graph.py

def get_branch_df(net: pandapowerNet, only_relevant_col: bool = True) -> pat.DataFrame[BranchSchema]:
    """Get the branches data from the pandapower network and return a df ready to use for the BranchSchema.

    Note: A star equivalent transformation for three winding trafos is not needed before calling this module.
    The the graph module only needs the connection and does no calculation.

    Parameters
    ----------
    net : pandapower
        The pandapower network.
    only_relevant_col : bool, optional
        Whether to return only the relevant columns, by default True
        relevant is determined by the default BranchSchema columns

    Returns
    -------
    branch_df : pat.DataFrame[BranchSchema]
        The DataFrame containing the branches in the format of the BranchSchema.
    """
    line_df = net.line.copy()
    line_df = get_edges_data(line_df, asset_type="line", only_relevant_col=only_relevant_col)

    impedances_df = net.impedance.copy()
    impedances_df = get_edges_data(impedances_df, asset_type="impedance", only_relevant_col=only_relevant_col)
    tcsc_df = net.tcsc.copy()
    tcsc_df = get_edges_data(tcsc_df, asset_type="tcsc", only_relevant_col=only_relevant_col)

    dclines = net.dcline.copy()
    dclines = get_edges_data(dclines, asset_type="dcline", only_relevant_col=only_relevant_col)

    transformers = net.trafo.copy()
    transformers = get_edges_data(transformers, asset_type="trafo", only_relevant_col=only_relevant_col)

    trafos3w1 = net.trafo3w.copy()
    trafos3w2 = net.trafo3w.copy()
    trafos3w2["lv_bus"] = trafos3w2["mv_bus"]
    trafos3w = pd.concat([trafos3w1, trafos3w2])
    trafos3w = get_edges_data(trafos3w, asset_type="trafo3w", only_relevant_col=only_relevant_col)

    branches_df = pd.concat([line_df, impedances_df, tcsc_df, dclines, transformers, trafos3w])
    branches_df.reset_index(drop=True, inplace=True)
    return branches_df

Exporter

toop_engine_importer.exporter

Export data from the AICoE_HPC_RL_Optimizer back to the original format.

• asset_topology_to_dgs.py: Translate asset topology model to a DGS file (PowerFactory).
• asset_topology_to_ucte.py: Translate asset topology model to a UCTE file.
• uct_exporter.py: Translate a RealizedTopology json file to a UCTE file.

__all__ module-attribute

__all__ = [
    "asset_topo_to_uct",
    "load_ucte",
    "process_file",
    "validate_ucte_changes",
]

asset_topo_to_uct

asset_topo_to_uct(
    master_data,
    grid_model_file_output,
    starting_stations=None,
    grid_model_file_input=None,
    station_list=None,
)

Translate asset topology model to UCT and saves the model.

PARAMETER	DESCRIPTION
master_data	Canonical master data describing the exported topology. TYPE: MasterAssetTopology
grid_model_file_output	Path to save the UCTE file. TYPE: Path
starting_stations	Optional runtime-aware station snapshots to export directly. If not provided, they are materialized from master_data and the input grid file via the same network-state path used by the backend. TYPE: Optional[list[RuntimeBusGroup]] DEFAULT: None
grid_model_file_input	Path to the grid model file. If not provided, asset_topology.grid_model_file will be used. TYPE: Optional[Path] DEFAULT: None
station_list	List of station ids to be translated. If not provided, all stations in the asset_topology will be translated. TYPE: Optional[str] DEFAULT: None

RAISES	DESCRIPTION
NotImplementedError	If master_data.asset_setpoints is not None.

Source code in packages/importer_pkg/src/toop_engine_importer/exporter/asset_topology_to_ucte.py

def asset_topo_to_uct(
    master_data: MasterAssetTopology,
    grid_model_file_output: Path,
    starting_stations: Optional[list[RuntimeBusGroup]] = None,
    grid_model_file_input: Optional[Path] = None,
    station_list: Optional[str] = None,
) -> None:
    """Translate asset topology model to UCT and saves the model.

    Parameters
    ----------
    master_data : MasterAssetTopology
        Canonical master data describing the exported topology.
    grid_model_file_output : Path
        Path to save the UCTE file.
    starting_stations : Optional[list[RuntimeBusGroup]]
        Optional runtime-aware station snapshots to export directly. If not provided,
        they are materialized from ``master_data`` and the input grid file via the same
        network-state path used by the backend.
    grid_model_file_input : Optional[Path]
        Path to the grid model file. If not provided, ``asset_topology.grid_model_file`` will be used.
    station_list : Optional[str]
        List of station ids to be translated.
        If not provided, all stations in the asset_topology will be translated.

    Raises
    ------
    NotImplementedError
        If master_data.asset_setpoints is not None.

    """
    if master_data.asset_setpoints is not None:
        raise NotImplementedError("Asset setpoints are not supported yet.")
    if grid_model_file_input is None:
        grid_model_file_input = Path(master_data.grid_model_file)
    if starting_stations is None:
        starting_stations = _get_starting_stations(master_data=master_data, grid_model_file_input=grid_model_file_input)
    preamble, nodes, lines, trafos, trafo_reg, postamble = load_ucte(grid_model_file_input)
    for station in starting_stations:
        if station_list is not None and station.bus_group_id not in station_list:
            continue
        asset_change_df = pd.DataFrame(get_changes_from_switching_table(station))
        if len(asset_change_df) > 0:
            change_trafos_lines_in_ucte(trafos, asset_change_df)
            change_trafos_lines_in_ucte(lines, asset_change_df)
        coupler_state_df = pd.DataFrame(get_coupler_state_ucte(station.couplers))
        change_busbar_coupler_state(lines, coupler_state_df)

    # handle order of elements in the ucte file
    handle_duplicated_grid_ids(trafos)
    handle_duplicated_grid_ids(lines)

    output_ucte_str = make_ucte(preamble, nodes, lines, trafos, trafo_reg, postamble)
    with open(grid_model_file_output, "w") as f:
        f.write(output_ucte_str)

load_ucte

load_ucte(input_uct)

Load UCTE file and return its contents as separate dataframes.

PARAMETER	DESCRIPTION
input_uct	Path to the UCTE file. TYPE: Path

RETURNS	DESCRIPTION
preamble	Preamble of the UCTE file. TYPE: str
nodes	Nodes of the UCTE file. TYPE: DataFrame
lines	Lines of the UCTE file. TYPE: DataFrame
trafos	Transformers of the UCTE file. TYPE: DataFrame
trafo_reg	Transformer regulation of the UCTE file. TYPE: DataFrame
postamble	Postamble of the UCTE file. TYPE: str

Source code in packages/importer_pkg/src/toop_engine_importer/exporter/asset_topology_to_ucte.py

def load_ucte(
    input_uct: Path | str,
) -> tuple[str, pd.DataFrame, pd.DataFrame, pd.DataFrame, pd.DataFrame, str]:
    """Load UCTE file and return its contents as separate dataframes.

    Parameters
    ----------
    input_uct : Path
        Path to the UCTE file.

    Returns
    -------
    preamble : str
        Preamble of the UCTE file.
    nodes : pd.DataFrame
        Nodes of the UCTE file.
    lines : pd.DataFrame
        Lines of the UCTE file.
    trafos : pd.DataFrame
        Transformers of the UCTE file.
    trafo_reg : pd.DataFrame
        Transformer regulation of the UCTE file.
    postamble : str
        Postamble of the UCTE file.
    """
    with open(input_uct, "r") as f:
        ucte_contents = f.read()
    preamble, nodes, lines, trafos, trafo_reg, postamble = parse_ucte(ucte_contents)

    return preamble, nodes, lines, trafos, trafo_reg, postamble

process_file

process_file(
    input_uct,
    input_json,
    output_uct,
    topo_id=0,
    reassign_branches=True,
    reassign_injections=False,
)

Process a UCTE file and a preprocessed json file to include split substations.

PARAMETER	DESCRIPTION
input_uct	The path to the input UCTE file, the original UCTE TYPE: Path
input_json	The preprocessed json holding the split substations and information, use the loadflowsolver's preprocessing notebook to generate this TYPE: Path
output_uct	The path to the output UCTE file, will be overwritten TYPE: Path
topo_id	The id of the topology to use in the json file TYPE: int DEFAULT: 0
reassign_branches	If True, reassign branches to the new busbars TYPE: bool DEFAULT: True
reassign_injections	If True, reassign injections to the new busbars Note: not implemented yet TYPE: bool DEFAULT: False

RETURNS	DESCRIPTION
list[str]	The codes of the fake busbars that were inserted

Source code in packages/importer_pkg/src/toop_engine_importer/exporter/uct_exporter.py

def process_file(
    input_uct: Path,
    input_json: Path,
    output_uct: Path,
    topo_id: int = 0,
    reassign_branches: bool = True,
    reassign_injections: bool = False,
) -> dict:
    """Process a UCTE file and a preprocessed json file to include split substations.

    Parameters
    ----------
    input_uct : Path
        The path to the input UCTE file, the original UCTE
    input_json : Path
        The preprocessed json holding the split substations and information, use the loadflowsolver's
        preprocessing notebook to generate this
    output_uct : Path
        The path to the output UCTE file, will be overwritten
    topo_id : int
        The id of the topology to use in the json file
    reassign_branches : bool
        If True, reassign branches to the new busbars
    reassign_injections : bool
        If True, reassign injections to the new busbars
        Note: not implemented yet

    Returns
    -------
    list[str]
        The codes of the fake busbars that were inserted
    """
    if reassign_injections:
        raise NotImplementedError("Reassigning injections is not implemented yet.")

    with open(input_uct, "r") as f:
        ucte_contents = f.read()
    with open(input_json, "r") as f:
        json_contents = json.load(f)
    topo = json_contents[topo_id]["topology"]["substation_info"]
    split_subs = [s for s in topo if is_split(s)]

    preamble, nodes, lines, trafos, trafo_reg, postamble = parse_ucte(ucte_contents)

    statistics = {"changed_ids": {}}  # type: dict

    for topo_element in split_subs:
        statistics["changed_ids"][topo_element["id"]] = {}
        statistics["changed_ids"][topo_element["id"]]["branches"] = {}
        statistics["changed_ids"][topo_element["id"]]["injections"] = {}

        code = topo_element["id"][0:7]
        switches = find_switches(lines, code)
        switches_dict = group_switches(switches)
        if len(switches_dict) == 0:
            raise ValueError(f"No switches found for substation {code}")
        switch_group_id = get_switch_group_number(switches_dict)
        bus_a, bus_b = get_bus_a_b(switches_dict[switch_group_id])

        if (topo_element["branch_assignments"] is not None) and reassign_branches:
            statistics["changed_ids"][topo_element["id"]]["branches"] = apply_branch_assignment(
                topo_element,
                lines,
                trafos,
                trafo_reg,
                bus_a,
                bus_b,
                statistics["changed_ids"],
            )

        statistics["changed_ids"][topo_element["id"]]["switches"] = open_switches(lines, switches_dict[switch_group_id])

    new_ucte = make_ucte(preamble, nodes, lines, trafos, trafo_reg, postamble)

    with open(output_uct, "w") as f:
        f.write(new_ucte)

    validate_ucte_changes(ucte_contents, new_ucte)

    return statistics

validate_ucte_changes

validate_ucte_changes(ucte_contents, ucte_contents_out)

Validate the changes made to the UCTE file.

PARAMETER	DESCRIPTION
ucte_contents	The original UCTE file TYPE: str
ucte_contents_out	The modified UCTE file TYPE: str

RAISES	DESCRIPTION
RuntimeError	If the changes are not as expected

Source code in packages/importer_pkg/src/toop_engine_importer/exporter/uct_exporter.py

def validate_ucte_changes(ucte_contents: str, ucte_contents_out: str) -> None:
    """Validate the changes made to the UCTE file.

    Parameters
    ----------
    ucte_contents : str
        The original UCTE file
    ucte_contents_out : str
        The modified UCTE file

    Raises
    ------
    RuntimeError
        If the changes are not as expected

    """
    if len(ucte_contents) != len(ucte_contents_out):
        raise RuntimeError(
            f"File sizes are different -> error in applying topology. "
            f"Length of original UCTE: {len(ucte_contents)}, Length of modified UCTE: {len(ucte_contents_out)}. "
            + "Length should not change, due to renaming of branches and opening switches."
        )

toop_engine_importer.exporter.asset_topology_to_ucte

Module containing functions to translate asset topology model to UCT model.

File: asset_topology_to_uct.py Author: Benjamin Petrick Created: 2024-10-22

Note: this module currently ignores the asset_setpoints. Note: this module currently ignores generator and load reassignments.

logger module-attribute

logger = structlog.get_logger(__name__)

UCTE_STATUS_CODES module-attribute

UCTE_STATUS_CODES = {
    0: {"name": "in_service", "opposite": 8},
    1: {"name": "in_service", "opposite": 9},
    2: {"name": "in_service", "opposite": 7},
    7: {"name": "out_of_service", "opposite": 2},
    8: {"name": "out_of_service", "opposite": 0},
    9: {"name": "out_of_service", "opposite": 1},
}

UCTE_STATUS_CODE_SWITCH module-attribute

UCTE_STATUS_CODE_SWITCH = {True: 7, False: 2}

load_ucte

load_ucte(input_uct)

Load UCTE file and return its contents as separate dataframes.

PARAMETER	DESCRIPTION
input_uct	Path to the UCTE file. TYPE: Path

RETURNS	DESCRIPTION
preamble	Preamble of the UCTE file. TYPE: str
nodes	Nodes of the UCTE file. TYPE: DataFrame
lines	Lines of the UCTE file. TYPE: DataFrame
trafos	Transformers of the UCTE file. TYPE: DataFrame
trafo_reg	Transformer regulation of the UCTE file. TYPE: DataFrame
postamble	Postamble of the UCTE file. TYPE: str

Source code in packages/importer_pkg/src/toop_engine_importer/exporter/asset_topology_to_ucte.py

def load_ucte(
    input_uct: Path | str,
) -> tuple[str, pd.DataFrame, pd.DataFrame, pd.DataFrame, pd.DataFrame, str]:
    """Load UCTE file and return its contents as separate dataframes.

    Parameters
    ----------
    input_uct : Path
        Path to the UCTE file.

    Returns
    -------
    preamble : str
        Preamble of the UCTE file.
    nodes : pd.DataFrame
        Nodes of the UCTE file.
    lines : pd.DataFrame
        Lines of the UCTE file.
    trafos : pd.DataFrame
        Transformers of the UCTE file.
    trafo_reg : pd.DataFrame
        Transformer regulation of the UCTE file.
    postamble : str
        Postamble of the UCTE file.
    """
    with open(input_uct, "r") as f:
        ucte_contents = f.read()
    preamble, nodes, lines, trafos, trafo_reg, postamble = parse_ucte(ucte_contents)

    return preamble, nodes, lines, trafos, trafo_reg, postamble

asset_topo_to_uct

asset_topo_to_uct(
    master_data,
    grid_model_file_output,
    starting_stations=None,
    grid_model_file_input=None,
    station_list=None,
)

Translate asset topology model to UCT and saves the model.

PARAMETER	DESCRIPTION
master_data	Canonical master data describing the exported topology. TYPE: MasterAssetTopology
grid_model_file_output	Path to save the UCTE file. TYPE: Path
starting_stations	Optional runtime-aware station snapshots to export directly. If not provided, they are materialized from master_data and the input grid file via the same network-state path used by the backend. TYPE: Optional[list[RuntimeBusGroup]] DEFAULT: None
grid_model_file_input	Path to the grid model file. If not provided, asset_topology.grid_model_file will be used. TYPE: Optional[Path] DEFAULT: None
station_list	List of station ids to be translated. If not provided, all stations in the asset_topology will be translated. TYPE: Optional[str] DEFAULT: None

RAISES	DESCRIPTION
NotImplementedError	If master_data.asset_setpoints is not None.

Source code in packages/importer_pkg/src/toop_engine_importer/exporter/asset_topology_to_ucte.py

def asset_topo_to_uct(
    master_data: MasterAssetTopology,
    grid_model_file_output: Path,
    starting_stations: Optional[list[RuntimeBusGroup]] = None,
    grid_model_file_input: Optional[Path] = None,
    station_list: Optional[str] = None,
) -> None:
    """Translate asset topology model to UCT and saves the model.

    Parameters
    ----------
    master_data : MasterAssetTopology
        Canonical master data describing the exported topology.
    grid_model_file_output : Path
        Path to save the UCTE file.
    starting_stations : Optional[list[RuntimeBusGroup]]
        Optional runtime-aware station snapshots to export directly. If not provided,
        they are materialized from ``master_data`` and the input grid file via the same
        network-state path used by the backend.
    grid_model_file_input : Optional[Path]
        Path to the grid model file. If not provided, ``asset_topology.grid_model_file`` will be used.
    station_list : Optional[str]
        List of station ids to be translated.
        If not provided, all stations in the asset_topology will be translated.

    Raises
    ------
    NotImplementedError
        If master_data.asset_setpoints is not None.

    """
    if master_data.asset_setpoints is not None:
        raise NotImplementedError("Asset setpoints are not supported yet.")
    if grid_model_file_input is None:
        grid_model_file_input = Path(master_data.grid_model_file)
    if starting_stations is None:
        starting_stations = _get_starting_stations(master_data=master_data, grid_model_file_input=grid_model_file_input)
    preamble, nodes, lines, trafos, trafo_reg, postamble = load_ucte(grid_model_file_input)
    for station in starting_stations:
        if station_list is not None and station.bus_group_id not in station_list:
            continue
        asset_change_df = pd.DataFrame(get_changes_from_switching_table(station))
        if len(asset_change_df) > 0:
            change_trafos_lines_in_ucte(trafos, asset_change_df)
            change_trafos_lines_in_ucte(lines, asset_change_df)
        coupler_state_df = pd.DataFrame(get_coupler_state_ucte(station.couplers))
        change_busbar_coupler_state(lines, coupler_state_df)

    # handle order of elements in the ucte file
    handle_duplicated_grid_ids(trafos)
    handle_duplicated_grid_ids(lines)

    output_ucte_str = make_ucte(preamble, nodes, lines, trafos, trafo_reg, postamble)
    with open(grid_model_file_output, "w") as f:
        f.write(output_ucte_str)

change_trafos_lines_in_ucte

change_trafos_lines_in_ucte(ucte_df, change_df)

Change the 'from' and 'to' columns of the trafos or line DataFrame based on the change_df.

PARAMETER	DESCRIPTION
ucte_df	The ucte trafos or line DataFrame Note: The DataFrame should have 'from', 'to', 'order' columns Note: The DataFrame is modified in place TYPE: DataFrame
change_df	The change_df, containing the 'grid_model_id', 'initial_busbar' and 'final_busbar' columns TYPE: DataFrame

RETURNS	DESCRIPTION
None

Source code in packages/importer_pkg/src/toop_engine_importer/exporter/asset_topology_to_ucte.py

def change_trafos_lines_in_ucte(ucte_df: pd.DataFrame, change_df: pd.DataFrame) -> None:
    """Change the 'from' and 'to' columns of the trafos or line DataFrame based on the change_df.

    Parameters
    ----------
    ucte_df : pd.DataFrame
        The ucte trafos or line DataFrame
        Note: The DataFrame should have 'from', 'to', 'order' columns
        Note: The DataFrame is modified in place
    change_df : pd.DataFrame
        The change_df, containing the 'grid_model_id', 'initial_busbar' and 'final_busbar' columns

    Returns
    -------
    None
    """
    # Create a new column 'grid_model_id' in the trafos DataFrame
    ucte_df["grid_model_id"] = ucte_df.apply(lambda row: f"{row['from']} {row['to']} {row['order']}", axis=1)
    ucte_df["index"] = ucte_df.index
    # Merge the ucte_df DataFrame with the change_df DataFrame and apply the update_busbars function
    ucte_df_w_changes = ucte_df.merge(change_df, on="grid_model_id", how="inner", suffixes=("", "_change")).set_index(
        "index"
    )

    ucte_df_w_changes_reassign = ucte_df_w_changes[ucte_df_w_changes["final_busbar"].notnull()]
    ucte_df_w_changes_disconnect = ucte_df_w_changes[ucte_df_w_changes["final_busbar"].isnull()]

    # Update 'from' and 'to' columns based on 'initial_busbar' and 'final_busbar'
    if ucte_df_w_changes_reassign.shape[0] > 0:
        ucte_df_w_changes_reassign = ucte_df_w_changes_reassign.apply(update_busbar_name, axis=1)
    if ucte_df_w_changes_disconnect.shape[0] > 0:
        ucte_df_w_changes_disconnect = ucte_df_w_changes_disconnect.apply(disconnect_line_from_ucte, axis=1)
    ucte_df_w_changes = pd.concat([ucte_df_w_changes_reassign, ucte_df_w_changes_disconnect])
    # update ucte_df
    ucte_df.drop(columns=["grid_model_id", "index"], inplace=True)
    ucte_df.loc[ucte_df_w_changes.index, ["from", "to", "status"]] = ucte_df_w_changes[["from", "to", "status"]]

update_busbar_name

update_busbar_name(row)

Update 'from' and 'to' columns based on 'initial_busbar' and 'final_busbar'.

PARAMETER	DESCRIPTION
row	A row of the DataFrame TYPE: Series

RETURNS	DESCRIPTION
Series	The updated row

Source code in packages/importer_pkg/src/toop_engine_importer/exporter/asset_topology_to_ucte.py

def update_busbar_name(row: pd.Series) -> pd.Series:
    """Update 'from' and 'to' columns based on 'initial_busbar' and 'final_busbar'.

    Parameters
    ----------
    row : pd.Series
        A row of the DataFrame

    Returns
    -------
    pd.Series
        The updated row
    """
    row["from"] = row["from"].replace(row["initial_busbar"], row["final_busbar"])
    row["to"] = row["to"].replace(row["initial_busbar"], row["final_busbar"])
    return row

disconnect_line_from_ucte

disconnect_line_from_ucte(line_row)

Disconnect a line from UCTE.

Note: a switch is modeled as a line in UCTE. To differentiate between a switch and a line, different status codes are used.

PARAMETER	DESCRIPTION
line_row	A row of the line DataFrame from the parse_ucte() function TYPE: Series

RETURNS	DESCRIPTION
Series	The updated row can be used to update the line DataFrame

Source code in packages/importer_pkg/src/toop_engine_importer/exporter/asset_topology_to_ucte.py

def disconnect_line_from_ucte(line_row: pd.Series) -> pd.Series:
    """Disconnect a line from UCTE.

    Note: a switch is modeled as a line in UCTE.
    To differentiate between a switch and a line, different status codes are used.

    Parameters
    ----------
    line_row : pd.Series
        A row of the line DataFrame from the parse_ucte() function

    Returns
    -------
    pd.Series
        The updated row can be used to update the line DataFrame
    """
    if UCTE_STATUS_CODES[int(line_row["status"])]["name"] == "in_service":
        line_row["status"] = str(UCTE_STATUS_CODES[int(line_row["status"])]["opposite"])
    return line_row

change_busbar_coupler_state

change_busbar_coupler_state(lines_df, change_df)

Change the 'status' columns of the lines DataFrame based on the change_df.

PARAMETER	DESCRIPTION
lines_df	The lines DataFrame Note: The DataFrame should have 'from', 'to', 'status' columns Note: The DataFrame is modified in place TYPE: DataFrame
change_df	The change_df, containing the 'grid_model_id' and 'coupler_state_ucte' columns TYPE: DataFrame

Source code in packages/importer_pkg/src/toop_engine_importer/exporter/asset_topology_to_ucte.py

def change_busbar_coupler_state(lines_df: pd.DataFrame, change_df: pd.DataFrame) -> None:
    """Change the 'status' columns of the lines DataFrame based on the change_df.

    Parameters
    ----------
    lines_df : pd.DataFrame
        The lines DataFrame
        Note: The DataFrame should have 'from', 'to', 'status' columns
        Note: The DataFrame is modified in place
    change_df : pd.DataFrame
        The change_df, containing the 'grid_model_id' and 'coupler_state_ucte' columns
    """
    # Create a new column 'grid_model_id' in the lines DataFrame
    lines_df["grid_model_id"] = lines_df.apply(lambda row: f"{row['from']} {row['to']} {row['order']}", axis=1)
    lines_df["index"] = lines_df.index
    # Merge the lines DataFrame with the change_df DataFrame and apply the update_coupler_state function
    lines_df_w_changes = lines_df.merge(change_df, on="grid_model_id", how="inner", suffixes=("", "_change")).set_index(
        "index"
    )
    lines_df_w_changes = lines_df_w_changes.apply(update_coupler_state, axis=1)
    # update lines_df
    lines_df.drop(columns=["grid_model_id", "index"], inplace=True)
    lines_df.loc[lines_df_w_changes.index, "status"] = lines_df_w_changes["status"]

update_coupler_state

update_coupler_state(row)

Update 'status' column based on 'coupler_state_ucte'.

PARAMETER	DESCRIPTION
row	A row of the DataFrame TYPE: Series

RETURNS	DESCRIPTION
Series	The updated row

Source code in packages/importer_pkg/src/toop_engine_importer/exporter/asset_topology_to_ucte.py

def update_coupler_state(row: pd.Series) -> pd.Series:
    """Update 'status' column based on 'coupler_state_ucte'.

    Parameters
    ----------
    row : pd.Series
        A row of the DataFrame

    Returns
    -------
    pd.Series
        The updated row
    """
    if int(row["status"]) not in UCTE_STATUS_CODE_SWITCH.values():
        initial_status_name = UCTE_STATUS_CODES[int(row["status"])]["name"]
        new_status_name = UCTE_STATUS_CODES[int(row["coupler_state_ucte"])]["name"]
        if initial_status_name != new_status_name:
            logger.warning(
                f"Line '{row['grid_model_id']}' has a status different from 2 or 7 with status: "
                + f"{row['status']}. Trying to switch a none busbar coupler. Status will be changed "
                + f"to {UCTE_STATUS_CODES[int(row['status'])]['opposite']}"
            )
            row["status"] = str(UCTE_STATUS_CODES[int(row["status"])]["opposite"])

    else:
        row["status"] = str(row["coupler_state_ucte"])
    return row

get_coupler_state_ucte

get_coupler_state_ucte(couplers)

Get coupler ucte state of from a BusbarCoupler.

PARAMETER	DESCRIPTION
couplers	BusbarCoupler object from the asset topology model TYPE: list[BusbarCoupler]

RETURNS	DESCRIPTION
list[dict[str, Union[str, int]]]	List of dictionaries containing the coupler_name and the state of the coupler in UCTE format 2: busbar coupler in operation (definition: R=0, X=0, B=0) 7: busbar coupler out of operation (definition: R=0, X=0, B=0)

Source code in packages/importer_pkg/src/toop_engine_importer/exporter/asset_topology_to_ucte.py

def get_coupler_state_ucte(couplers: list[BusbarCoupler]) -> list[dict[str, Union[str, int]]]:
    """Get coupler ucte state of from a BusbarCoupler.

    Parameters
    ----------
    couplers : list[BusbarCoupler]
        BusbarCoupler object from the asset topology model

    Returns
    -------
    list[dict[str, Union[str, int]]]
        List of dictionaries containing the coupler_name and the state of the coupler in UCTE format
        2: busbar coupler in operation (definition: R=0, X=0, B=0)
        7: busbar coupler out of operation (definition: R=0, X=0, B=0)
    """
    coupler_state_ucte = [  # TODO: make a dataclass for this (code style)
        {
            "grid_model_id": coupler.grid_model_id,
            "coupler_state_ucte": UCTE_STATUS_CODE_SWITCH[coupler.open],
        }
        for coupler in couplers
    ]
    return coupler_state_ucte

get_changes_from_switching_table

get_changes_from_switching_table(station)

Get changes from switching table.

PARAMETER	DESCRIPTION
station	Station object with switching table, busbars and assets TYPE: Station

RETURNS	DESCRIPTION
list[dict[str, Union[str, None]]]	List of tuples with asset name, initial_busbar and final_busbar Note: initial_busbar and final_busbar can both be None if asset is disconnected

Source code in packages/importer_pkg/src/toop_engine_importer/exporter/asset_topology_to_ucte.py

def get_changes_from_switching_table(
    station: RuntimeBusGroup,
) -> list[dict[str, Union[str, None]]]:
    """Get changes from switching table.

    Parameters
    ----------
    station : Station
        Station object with switching table, busbars and assets

    Returns
    -------
    list[dict[str, Union[str, None]]]
        List of tuples with asset name, initial_busbar and final_busbar
        Note: initial_busbar and final_busbar can both be None if asset is disconnected
    """
    switching_table = np.concatenate([station.branch_switching_table, station.injection_switching_table], axis=1)
    busbar_name_list = [busbar.grid_model_id for busbar in station.busbars]
    asset_connections = [
        *station.branch_connections,
        *station.injection_connections,
    ]
    asset_list = [
        *(asset_connection.asset for asset_connection in station.branch_connections),
        *(asset_connection.asset for asset_connection in station.injection_connections),
    ]
    change_list = []  # TODO: make a dataclass for this (code style)
    # loop over assets -> by column
    for asset_index, asset_in_table in enumerate(switching_table.T):
        asset_name = asset_list[asset_index].grid_model_id
        asset_type = asset_list[asset_index].asset_type
        asset_connection = asset_connections[asset_index]
        busbar_initial = [busbar for busbar in busbar_name_list if busbar in asset_name]
        if asset_in_table.sum() > 1:
            raise ValueError(
                f"Asset {asset_list[asset_index].grid_model_id} is connected to multiple"
                + " busbars. This is not supported for the UCTE format"
            )
        if asset_in_table.sum() == 0:
            if asset_connection.branch_end is None and len(busbar_initial) > 1:
                continue
            # asset is disconnected
            change_list.append(
                {
                    "grid_model_id": asset_name,
                    "initial_busbar": None,
                    "final_busbar": None,
                    "asset_type": asset_type,
                }
            )
            continue
        # asset is connected, check if busbar assignment is changed
        for busbar_index, asset_connected in enumerate(asset_in_table):
            if not asset_connected:
                continue
            busbar_name = busbar_name_list[busbar_index]

            if len(busbar_initial) == 0:
                raise ValueError(f"Asset {asset_name} busbar connection is not found, busbar_name_list: {busbar_name_list}")

            if busbar_name not in asset_name:
                if len(busbar_initial) > 1:
                    raise ValueError(
                        f"Asset {asset_name} is connected to multiple busbars within the same station. "
                        + "Asset can not be reassigned."
                    )
                change_list.append(
                    {
                        "grid_model_id": asset_name,
                        "initial_busbar": busbar_initial[0],
                        "final_busbar": busbar_name,
                        "asset_type": asset_type,
                    }
                )

    return change_list

handle_duplicated_grid_ids

handle_duplicated_grid_ids(ucte_df)

Handle duplicated grid ids in the ucte file.

The function will increment the order of the duplicated grid ids by 1.

PARAMETER	DESCRIPTION
ucte_df	The ucte DataFrame to be updated Note: The DataFrame should have 'from', 'to', 'order' columns Note: The DataFrame is modified in place TYPE: DataFrame

Source code in packages/importer_pkg/src/toop_engine_importer/exporter/asset_topology_to_ucte.py

def handle_duplicated_grid_ids(ucte_df: pd.DataFrame) -> None:
    """Handle duplicated grid ids in the ucte file.

    The function will increment the order of the duplicated grid ids by 1.

    Parameters
    ----------
    ucte_df : pd.DataFrame
        The ucte DataFrame to be updated
        Note: The DataFrame should have 'from', 'to', 'order' columns
        Note: The DataFrame is modified in place
    """
    ucte_df["grid_model_id"] = ucte_df.apply(lambda row: f"{row['from']} {row['to']} {row['order']}", axis=1)
    ucte_df["index"] = ucte_df.index
    duplicated_ids = ucte_df[ucte_df["grid_model_id"].duplicated(keep="first")]
    run_count = 0
    max_runs = 20
    while duplicated_ids.shape[0] > 0:
        # Find duplicated grid_model_ids
        duplicated_ids["order"] = duplicated_ids.apply(lambda row: f"{int(row['order']) + 1}", axis=1)
        duplicated_ids.set_index("index")
        ucte_df.loc[duplicated_ids.index, "order"] = duplicated_ids["order"]

        # set new grid_model_id
        ucte_df["grid_model_id"] = ucte_df.apply(lambda row: f"{row['from']} {row['to']} {row['order']}", axis=1)
        duplicated_ids = ucte_df[ucte_df["grid_model_id"].duplicated(keep="first")]

        run_count += 1
        if run_count > max_runs:
            raise ValueError("Duplicated grid_model_ids could not be resolved. More than 20 iterations have been reached.")
    ucte_df.drop(columns=["grid_model_id", "index"], inplace=True)

toop_engine_importer.exporter.uct_exporter

Module containing functions to translate a RealizedTopology json file to a UCTE file.

DeprecationWarning: This module is deprecated and will be removed in the future, due to deprecation of RealizedTopology. Use Topology (AssetTopology) instead.

File: ucte_exporter.py Author: Benjamin Petrick Created: 2024

Note: this module ignores generator and load reassignments.

logger module-attribute

logger = structlog.get_logger(__name__)

process_file

process_file(
    input_uct,
    input_json,
    output_uct,
    topo_id=0,
    reassign_branches=True,
    reassign_injections=False,
)

Process a UCTE file and a preprocessed json file to include split substations.

PARAMETER	DESCRIPTION
input_uct	The path to the input UCTE file, the original UCTE TYPE: Path
input_json	The preprocessed json holding the split substations and information, use the loadflowsolver's preprocessing notebook to generate this TYPE: Path
output_uct	The path to the output UCTE file, will be overwritten TYPE: Path
topo_id	The id of the topology to use in the json file TYPE: int DEFAULT: 0
reassign_branches	If True, reassign branches to the new busbars TYPE: bool DEFAULT: True
reassign_injections	If True, reassign injections to the new busbars Note: not implemented yet TYPE: bool DEFAULT: False

RETURNS	DESCRIPTION
list[str]	The codes of the fake busbars that were inserted

Source code in packages/importer_pkg/src/toop_engine_importer/exporter/uct_exporter.py

def process_file(
    input_uct: Path,
    input_json: Path,
    output_uct: Path,
    topo_id: int = 0,
    reassign_branches: bool = True,
    reassign_injections: bool = False,
) -> dict:
    """Process a UCTE file and a preprocessed json file to include split substations.

    Parameters
    ----------
    input_uct : Path
        The path to the input UCTE file, the original UCTE
    input_json : Path
        The preprocessed json holding the split substations and information, use the loadflowsolver's
        preprocessing notebook to generate this
    output_uct : Path
        The path to the output UCTE file, will be overwritten
    topo_id : int
        The id of the topology to use in the json file
    reassign_branches : bool
        If True, reassign branches to the new busbars
    reassign_injections : bool
        If True, reassign injections to the new busbars
        Note: not implemented yet

    Returns
    -------
    list[str]
        The codes of the fake busbars that were inserted
    """
    if reassign_injections:
        raise NotImplementedError("Reassigning injections is not implemented yet.")

    with open(input_uct, "r") as f:
        ucte_contents = f.read()
    with open(input_json, "r") as f:
        json_contents = json.load(f)
    topo = json_contents[topo_id]["topology"]["substation_info"]
    split_subs = [s for s in topo if is_split(s)]

    preamble, nodes, lines, trafos, trafo_reg, postamble = parse_ucte(ucte_contents)

    statistics = {"changed_ids": {}}  # type: dict

    for topo_element in split_subs:
        statistics["changed_ids"][topo_element["id"]] = {}
        statistics["changed_ids"][topo_element["id"]]["branches"] = {}
        statistics["changed_ids"][topo_element["id"]]["injections"] = {}

        code = topo_element["id"][0:7]
        switches = find_switches(lines, code)
        switches_dict = group_switches(switches)
        if len(switches_dict) == 0:
            raise ValueError(f"No switches found for substation {code}")
        switch_group_id = get_switch_group_number(switches_dict)
        bus_a, bus_b = get_bus_a_b(switches_dict[switch_group_id])

        if (topo_element["branch_assignments"] is not None) and reassign_branches:
            statistics["changed_ids"][topo_element["id"]]["branches"] = apply_branch_assignment(
                topo_element,
                lines,
                trafos,
                trafo_reg,
                bus_a,
                bus_b,
                statistics["changed_ids"],
            )

        statistics["changed_ids"][topo_element["id"]]["switches"] = open_switches(lines, switches_dict[switch_group_id])

    new_ucte = make_ucte(preamble, nodes, lines, trafos, trafo_reg, postamble)

    with open(output_uct, "w") as f:
        f.write(new_ucte)

    validate_ucte_changes(ucte_contents, new_ucte)

    return statistics

validate_ucte_changes

validate_ucte_changes(ucte_contents, ucte_contents_out)

Validate the changes made to the UCTE file.

PARAMETER	DESCRIPTION
ucte_contents	The original UCTE file TYPE: str
ucte_contents_out	The modified UCTE file TYPE: str

RAISES	DESCRIPTION
RuntimeError	If the changes are not as expected

Source code in packages/importer_pkg/src/toop_engine_importer/exporter/uct_exporter.py

def validate_ucte_changes(ucte_contents: str, ucte_contents_out: str) -> None:
    """Validate the changes made to the UCTE file.

    Parameters
    ----------
    ucte_contents : str
        The original UCTE file
    ucte_contents_out : str
        The modified UCTE file

    Raises
    ------
    RuntimeError
        If the changes are not as expected

    """
    if len(ucte_contents) != len(ucte_contents_out):
        raise RuntimeError(
            f"File sizes are different -> error in applying topology. "
            f"Length of original UCTE: {len(ucte_contents)}, Length of modified UCTE: {len(ucte_contents_out)}. "
            + "Length should not change, due to renaming of branches and opening switches."
        )

is_split

is_split(sub_info)

Check if the substation was split.

Source code in packages/importer_pkg/src/toop_engine_importer/exporter/uct_exporter.py

def is_split(sub_info: dict) -> bool:
    """Check if the substation was split."""
    return any(b["on_bus_b"] for b in sub_info["branch_assignments"])

get_switch_group_number

get_switch_group_number(grouped_switches)

Decide which switch group to open. Selects the group with the fewest unique busbars.

PARAMETER	DESCRIPTION
grouped_switches	The grouped switches data-frame from group_switches(). Each key contains all switches necessary to isolate a busbar TYPE: dict

RETURNS	DESCRIPTION
reassignment_key	The dict key of the switch group to open TYPE: str

Source code in packages/importer_pkg/src/toop_engine_importer/exporter/uct_exporter.py

def get_switch_group_number(grouped_switches: dict) -> str:
    """Decide which switch group to open. Selects the group with the fewest unique busbars.

    Parameters
    ----------
    grouped_switches : dict
        The grouped switches data-frame from group_switches(). Each key contains all switches necessary to isolate a busbar

    Returns
    -------
    reassignment_key : str
        The dict key of the switch group to open

    """
    reassignment_key = ""
    ideal_candidate = False
    n_unique_busbars = 2
    n_switchtes = 1

    for sw, sw_values in grouped_switches.items():
        unique_busbars = get_unique_busbars(sw_values)
        if len(unique_busbars) == n_unique_busbars and len(sw_values) == n_switchtes:
            # ideal candidate for branch assignment
            reassignment_key = sw
            ideal_candidate = True
            break
        if len(unique_busbars) == n_unique_busbars:
            # still only one busbar to reassign, but with multiple switches -> continue searching
            reassignment_key = sw
            ideal_candidate = True
        elif len(unique_busbars) > n_unique_busbars and reassignment_key == "":
            # not ideal candidate for branch assignment -> continue searching
            reassignment_key = sw
            # candidates with only one unique busbar are left out -> doesn't make sense

    if reassignment_key == "":
        raise ValueError(f"No switch group found. Using the first switch group: {grouped_switches}")
    if not ideal_candidate:
        logger.warning(
            f"Switch group {reassignment_key} has more than 2 busbars. "
            + f"Validate if behavior is as expected: {get_unique_busbars(grouped_switches[reassignment_key])}"
        )
    return reassignment_key

find_switches

find_switches(lines, node_id)

Find the all switches on the input node.

PARAMETER	DESCRIPTION
lines	The lines data-frame from UCTE file TYPE: DataFrame
node_id	The node id(s) to search for switches. Note: expects switches to be closed -> status = 2 (closed) Note: this is the first 7 characters of the id (Node), not the full id of busbar that has one additional character TYPE: str | list[str]

RETURNS	DESCRIPTION
switches	All switches found on the input node, with status = 2 (closed) TYPE: DataFrame

Source code in packages/importer_pkg/src/toop_engine_importer/exporter/uct_exporter.py

def find_switches(lines: pd.DataFrame, node_id: str | list[str]) -> pd.DataFrame:
    """Find the all switches on the input node.

    Parameters
    ----------
    lines : pd.DataFrame
        The lines data-frame from UCTE file
    node_id : str | list[str]
        The node id(s) to search for switches. Note: expects switches to be closed -> status = 2 (closed)
        Note: this is the first 7 characters of the id (Node), not the full id of busbar that has one additional character

    Returns
    -------
    switches : pd.DataFrame
        All switches found on the input node, with status = 2 (closed)


    """
    switches = lines[
        ((lines["from"].str.startswith(node_id)) & (lines["to"].str.startswith(node_id))) & (lines["status"] == "2")
    ]
    return switches

get_unique_busbars

get_unique_busbars(switches)

Get the unique busbars from the switches.

PARAMETER	DESCRIPTION
switches	The switches data-frame from find_switches() TYPE: DataFrame

RETURNS	DESCRIPTION
unique_busbars	A list of unique busbars found in the switches TYPE: list[str]

Source code in packages/importer_pkg/src/toop_engine_importer/exporter/uct_exporter.py

def get_unique_busbars(switches: pd.DataFrame) -> list[str]:
    """Get the unique busbars from the switches.

    Parameters
    ----------
    switches : pd.DataFrame
        The switches data-frame from find_switches()

    Returns
    -------
    unique_busbars : list[str]
        A list of unique busbars found in the switches

    """
    unique_busbars = []
    for _, row in switches.iterrows():
        if row["from"] not in unique_busbars:
            unique_busbars.append(row["from"])
        if row["to"] not in unique_busbars:
            unique_busbars.append(row["to"])

    return unique_busbars

group_switches

group_switches(switches)

Group the switches by the busbar id.

There can be multiple switches between the same busbars. This function groups them together. One list element is one group of switches to isolate the bus completely.

PARAMETER	DESCRIPTION
switches	All switches data-frame from UCTE file from a specific node TYPE: DataFrame

RETURNS	DESCRIPTION
switches_sort	A dict of switches data-frames sorted by the busbar id TYPE: dict

Source code in packages/importer_pkg/src/toop_engine_importer/exporter/uct_exporter.py

def group_switches(switches: pd.DataFrame) -> dict:
    """Group the switches by the busbar id.

    There can be multiple switches between the same busbars.
    This function groups them together. One list element is one group of switches to isolate the bus completely.

    Parameters
    ----------
    switches : pd.DataFrame
        All switches data-frame from UCTE file from a specific node

    Returns
    -------
    switches_sort : dict
        A dict of switches data-frames sorted by the busbar id
    """
    unique_switch_ids = get_unique_busbars(switches)

    # sort switches by busbar id of switch
    switches_sort = {}
    for switch_id in unique_switch_ids:
        switches_sort[switch_id] = switches[(switches["from"] == switch_id) | (switches["to"] == switch_id)]
    return switches_sort

get_bus_a_b

get_bus_a_b(switches)

Get the bus A and B from the switches.

PARAMETER	DESCRIPTION
switches	The switches data-frame from find_switches() TYPE: DataFrame

RETURNS	DESCRIPTION
bus_a	The bus A of the substation TYPE: str
bus_b	The bus B of the substation TYPE: str

RAISES	DESCRIPTION
ValueError	If the switches contain switches from multiple nodes.

Source code in packages/importer_pkg/src/toop_engine_importer/exporter/uct_exporter.py

def get_bus_a_b(switches: pd.DataFrame) -> tuple[str, str]:
    """Get the bus A and B from the switches.

    Parameters
    ----------
    switches : pd.DataFrame
        The switches data-frame from find_switches()

    Returns
    -------
    bus_a : str
        The bus A of the substation
    bus_b : str
        The bus B of the substation

    Raises
    ------
    ValueError
        If the switches contain switches from multiple nodes.

    """
    from_values = switches["from"].values
    to_values = switches["to"].values

    if all(x == from_values[0] for x in from_values) and all(x == to_values[0] for x in to_values):
        bus_a = from_values[0]
        bus_b = to_values[0]
    else:
        raise ValueError(
            f"Switches DataFrame contains switches from multiple nodes. Node 'from' {from_values}, Node 'to' {to_values}"
        )

    return bus_a, bus_b

open_switches

open_switches(lines, switches)

Open switches in the UCTE data by changing the status code value.

PARAMETER	DESCRIPTION
lines	The lines data-frame from UCTE file Note: modifies the data-frames in place TYPE: DataFrame
switches	The switches data-frame from find_switches() Note: modifies the data-frames in place TYPE: DataFrame

RETURNS	DESCRIPTION
stats	A dictionary containing the bus A and B of the substation, the number of switches and the from and to busbar of the switches TYPE: dict

Source code in packages/importer_pkg/src/toop_engine_importer/exporter/uct_exporter.py

def open_switches(lines: pd.DataFrame, switches: pd.DataFrame) -> dict:
    """Open switches in the UCTE data by changing the status code value.

    Parameters
    ----------
    lines : pd.DataFrame
        The lines data-frame from UCTE file
        Note: modifies the data-frames in place
    switches : pd.DataFrame
        The switches data-frame from find_switches()
        Note: modifies the data-frames in place


    Returns
    -------
    stats : dict
        A dictionary containing the bus A and B of the substation,
        the number of switches and the from and to busbar of the switches


    """
    switch_idx = switches.index
    bus_a, bus_b = get_bus_a_b(switches)
    lines.loc[switch_idx, "status"] = "7"  # 7 -> open switch
    stats = {
        "bus_a": bus_a,
        "bus_b": bus_b,
        "from": switches.iloc[0]["from"],
        "to": switches.iloc[0]["to"],
        "order_of_switches": list(switches["order"].values),
    }
    return stats

handle_order_of_branch

handle_order_of_branch(branch_df, replacement_id)

Get a unique order number of the branch for new id.

PARAMETER	DESCRIPTION
branch_df	The branch data-frame from UCTE file. e.g. lines or trafos or trafo_reg TYPE: DataFrame
replacement_id	The replacement ID of the branch TYPE: str

RETURNS	DESCRIPTION
order	The order of the branch TYPE: str

Source code in packages/importer_pkg/src/toop_engine_importer/exporter/uct_exporter.py

def handle_order_of_branch(branch_df: pd.DataFrame, replacement_id: str) -> str:
    """Get a unique order number of the branch for new id.

    Parameters
    ----------
    branch_df : pd.DataFrame
        The branch data-frame from UCTE file. e.g. lines or trafos or trafo_reg
    replacement_id : str
        The replacement ID of the branch

    Returns
    -------
    order : str
        The order of the branch

    """
    from_node, to_node, order = replacement_id.split(" ")
    order_int = int(order)
    loop_count = 0
    max_loop = 100
    while len(find_branch_index(branch_df, f"{from_node} {to_node} {order_int}")) > 0:
        order_int += 1

        loop_count += 1
        if loop_count > max_loop:
            raise ValueError("handle_order_of_branch() Loop count exceeded 100")

    updated_replacement_id = f"{from_node} {to_node} {order_int}"
    return updated_replacement_id

find_branch_index

find_branch_index(branch_df, id)

Find the index of the branch in the UCTE data.

PARAMETER	DESCRIPTION
branch_df	The branch data-frame from UCTE file. e.g. lines or trafos or trafo_reg TYPE: DataFrame
id	The ID of the branch TYPE: str

RETURNS	DESCRIPTION
branch_df_idx	The index of the branch in the data-frame TYPE: DataFrame

Source code in packages/importer_pkg/src/toop_engine_importer/exporter/uct_exporter.py

def find_branch_index(branch_df: pd.DataFrame, id: str) -> pd.Index:
    """Find the index of the branch in the UCTE data.

    Parameters
    ----------
    branch_df : pd.DataFrame
        The branch data-frame from UCTE file. e.g. lines or trafos or trafo_reg
    id : str
        The ID of the branch

    Returns
    -------
    branch_df_idx : pd.DataFrame
        The index of the branch in the data-frame

    """
    from_node = id[0:8]
    to_node = id[9:17]
    order = id[18:19]
    branch_df_idx = branch_df[
        (branch_df["from"] == from_node) & (branch_df["to"] == to_node) & (branch_df["order"] == order)
    ].index
    return branch_df_idx

execute_branch_assignment

execute_branch_assignment(
    branch_df, id, replacement_id, statistics_all_stations
)

Apply branch assignment to the UCTE data.

PARAMETER	DESCRIPTION
branch_df	The branch data-frame from UCTE file. e.g. lines or trafos or trafo_reg Note: modifies the data-frames in place TYPE: DataFrame
id	The original ID of the branch TYPE: str
replacement_id	The replacement ID of the branch TYPE: str
statistics_all_stations	The statistics dictionary from process_file() expects as input the statistics["changed_ids"] dictionary TYPE: dict

RETURNS	DESCRIPTION
replaced	True if the branch was replaced, False if the branch was not found in the data-frame TYPE: bool

Source code in packages/importer_pkg/src/toop_engine_importer/exporter/uct_exporter.py

def execute_branch_assignment(
    branch_df: pd.DataFrame,
    id: str,
    replacement_id: str,
    statistics_all_stations: dict[str, dict[str, list[str] | dict[str, list[str]]]],
) -> bool:
    """Apply branch assignment to the UCTE data.

    Parameters
    ----------
    branch_df : pd.DataFrame
        The branch data-frame from UCTE file. e.g. lines or trafos or trafo_reg
        Note: modifies the data-frames in place
    id : str
        The original ID of the branch
    replacement_id : str
        The replacement ID of the branch
    statistics_all_stations : dict
        The statistics dictionary from process_file()
        expects as input the statistics["changed_ids"] dictionary

    Returns
    -------
    replaced : bool
        True if the branch was replaced, False if the branch was not found in the data-frame

    """
    from_node_replacement = replacement_id[0:8]
    to_node_replacement = replacement_id[9:17]
    order_replacement = replacement_id[18:19]
    branch_df_idx = find_branch_index(branch_df, id)
    if len(branch_df_idx) == 0:
        # check if the ID has been replaced in the statistics
        id = update_id_if_has_been_replaced(id, statistics_all_stations)
        branch_df_idx = find_branch_index(branch_df, id)

    if len(branch_df_idx) > 0:
        replaced = True
        branch_df.loc[branch_df_idx, "from"] = from_node_replacement
        branch_df.loc[branch_df_idx, "to"] = to_node_replacement
        branch_df.loc[branch_df_idx, "order"] = order_replacement
    else:
        replaced = False
    return replaced

get_replacement_id

get_replacement_id(element, code, bus_a, bus_b)

Get the replacement ID of the branch.

Decides if the element is on bus A or B and replaces the ID accordingly. Bus A is a logical bus and can be electrically connected to other buses e.g. bus C. Bus B is the new Bus that will be split off.

PARAMETER	DESCRIPTION
element	The element to replace imported from the json TYPE: dict
code	The code of the substation TYPE: str
bus_a	The bus A of the substation TYPE: str
bus_b	The bus B of the substation TYPE: str

RETURNS	DESCRIPTION
replacement_id	The replacement ID of the branch TYPE: str

Source code in packages/importer_pkg/src/toop_engine_importer/exporter/uct_exporter.py

def get_replacement_id(element: dict, code: str, bus_a: str, bus_b: str) -> str:
    """Get the replacement ID of the branch.

    Decides if the element is on bus A or B and replaces the ID accordingly.
    Bus A is a logical bus and can be electrically connected to other buses e.g. bus C.
    Bus B is the new Bus that will be split off.

    Parameters
    ----------
    element : dict
        The element to replace imported from the json
    code : str
        The code of the substation
    bus_a : str
        The bus A of the substation
    bus_b : str
        The bus B of the substation

    Returns
    -------
    replacement_id : str
        The replacement ID of the branch
    """
    id = element["id"]

    if element["on_bus_b"]:
        # element on bus B -> replace with bus B
        replacement_id = re.sub(rf"{re.escape(code)}\d?", bus_b, id)
    elif bus_b in id:
        # element on bus A -> replace with bus A
        replacement_id = re.sub(rf"{re.escape(code)}\d?", bus_a, id)
    else:
        # element on the logical bus A but is not on the new electrically isolated bus B
        # -> keep ID, as it is not affected by the split
        # e.g. lines/switches between bus A and C should still be closed and therefore A and C should be logically connected
        replacement_id = id
    return replacement_id

update_id_if_has_been_replaced

update_id_if_has_been_replaced(id, statistics)

Update the ID if it has been replaced in the statistics.

Each branch has a "from" and "to" bus. It can oocur that e.g. the "from" bus has been already replaced, but the "to" bus not. This function checks if the ID has been replaced and returns the new ID if it has been replaced.

PARAMETER	DESCRIPTION
id	The ID of the branch, which might have been replaced TYPE: str | int
statistics	The statistics dictionary from process_file() expects as input the statistics["changed_ids"] dictionary TYPE: dict

Source code in packages/importer_pkg/src/toop_engine_importer/exporter/uct_exporter.py

def update_id_if_has_been_replaced(id: str | int, statistics: dict[str, Any]) -> str | int:
    """Update the ID if it has been replaced in the statistics.

    Each branch has a "from" and "to" bus.
    It can oocur that e.g. the "from" bus has been already replaced, but the "to" bus not.
    This function checks if the ID has been replaced and returns the new ID if it has been replaced.

    Parameters
    ----------
    id : str | int
        The ID of the branch, which might have been replaced
    statistics : dict
        The statistics dictionary from process_file()
        expects as input the statistics["changed_ids"] dictionary

    """
    for station in statistics.values():
        for element in station["branches"]:
            if element["original_id"] == id:
                if element["replacement_id"] != "":
                    id = element["replacement_id"]
                return id
    return id

apply_branch_assignment

apply_branch_assignment(
    topology_optimizer_results,
    lines,
    trafos,
    trafo_reg,
    bus_a,
    bus_b,
    statistics_all_stations,
)

Apply branch assignment to the UCTE data.

Uses the method to split a busbar into two busbars A and B

PARAMETER	DESCRIPTION
topology_optimizer_results	The substation split to apply from the postprocessed json TYPE: dict
lines	The lines data-frame from UCTE file Note: modifies the data-frames in place TYPE: DataFrame
trafos	The transformers data-frame from UCTE file Note: modifies the data-frames in place TYPE: DataFrame
trafo_reg	The transformer regulation data-frame from UCTE file Note: modifies the data-frames in place TYPE: DataFrame
bus_a	The bus A of the substation TYPE: str
bus_b	The bus B of the substation TYPE: str
statistics_all_stations	The statistics dictionary from process_file() TYPE: dict

RETURNS	DESCRIPTION
statistics	A list of dictionaries containing the original and replacement IDs of the branches that were modified TYPE: list

RAISES	DESCRIPTION
ValueError	If the branch type is not recognized If the branch is not found in the data-frame

Source code in packages/importer_pkg/src/toop_engine_importer/exporter/uct_exporter.py

def apply_branch_assignment(  # noqa: PLR0912, C901
    topology_optimizer_results: dict,
    lines: pd.DataFrame,
    trafos: pd.DataFrame,
    trafo_reg: pd.DataFrame,
    bus_a: str,
    bus_b: str,
    statistics_all_stations: dict[str, dict[str, list[str] | dict[str, list[str]]]],
) -> list:
    """Apply branch assignment to the UCTE data.

    Uses the method to split a busbar into two busbars A and B

    Parameters
    ----------
    topology_optimizer_results : dict
        The substation split to apply from the postprocessed json
    lines : pd.DataFrame
        The lines data-frame from UCTE file
        Note: modifies the data-frames in place
    trafos : pd.DataFrame
        The transformers data-frame from UCTE file
        Note: modifies the data-frames in place
    trafo_reg : pd.DataFrame
        The transformer regulation data-frame from UCTE file
        Note: modifies the data-frames in place
    bus_a : str
        The bus A of the substation
    bus_b : str
        The bus B of the substation
    statistics_all_stations : dict
        The statistics dictionary from process_file()

    Returns
    -------
    statistics : list
        A list of dictionaries containing the original and replacement IDs of the branches that were modified

    Raises
    ------
    ValueError
        - If the branch type is not recognized
        - If the branch is not found in the data-frame

    """
    statistics = []  # type: list
    code = topology_optimizer_results["id"][0:7]

    # replace busbar ID in lines, trafos and trafo_reg df
    for element in topology_optimizer_results["branch_assignments"]:
        replacement_id = get_replacement_id(element, code, bus_a, bus_b)

        id = element["id"]
        if id != replacement_id:
            # replace only if ID is different
            if element["type"] == "LINE":
                replacement_id = handle_order_of_branch(lines, replacement_id)
                replaced = execute_branch_assignment(lines, id, replacement_id, statistics_all_stations)
                if replaced:
                    statistics.append(
                        {
                            "original_id": id,
                            "replacement_id": replacement_id,
                            "type": element["type"],
                        }
                    )
                else:
                    raise ValueError(
                        f"Line not found: bus_a: {bus_a}, bus_b: {bus_b}, id:{id}, replacement_id: {replacement_id}"
                    )

            elif element["type"] == "TWO_WINDINGS_TRANSFORMER":
                replacement_id = handle_order_of_branch(trafos, replacement_id)
                replaced = execute_branch_assignment(trafos, id, replacement_id, statistics_all_stations)
                replaced2 = execute_branch_assignment(trafo_reg, id, replacement_id, statistics_all_stations)
                if replaced or replaced2:
                    statistics.append(
                        {
                            "original_id": id,
                            "replacement_id": replacement_id,
                            "type": element["type"],
                        }
                    )
                else:
                    raise ValueError(
                        f"Transformer not found: bus_a: {bus_a}, bus_b: {bus_b}, id:{id}, replacement_id: {replacement_id}"
                    )
            elif element["type"] == "TIE_LINE":
                # TIE_LINE is a special case, as it is consists of two lines
                # UCTE has only lines -> split TIE_LINE into two lines -> search for both lines and replace
                id1 = id.split(" + ")[0]
                id2 = id.split(" + ")[1]
                replacement_id1 = replacement_id.split(" + ")[0]
                replacement_id2 = replacement_id.split(" + ")[1]
                if bus_a in id1 and bus_b in replacement_id1:
                    replacement_id1 = handle_order_of_branch(lines, replacement_id1)
                    replaced = execute_branch_assignment(lines, id1, replacement_id1, statistics_all_stations)
                elif bus_a in id2 and bus_b in replacement_id2:
                    replacement_id2 = handle_order_of_branch(lines, replacement_id2)
                    replaced = execute_branch_assignment(lines, id2, replacement_id2, statistics_all_stations)
                else:
                    raise ValueError(
                        f"TIE_LINE not found: bus_a: {bus_a}, bus_b: {bus_b}, id:{id}, replacement_id: {replacement_id}"
                    )
                if replaced:
                    statistics.append(
                        {
                            "original_id": id,
                            "replacement_id": replacement_id,
                            "type": element["type"],
                        }
                    )
                else:
                    raise ValueError(
                        f"TIE_LINE not found: bus_a: {bus_a}, bus_b: {bus_b}, id:{id}, replacement_id: {replacement_id}"
                    )

            else:
                raise ValueError(f"Unknown branch type: {element['type']}")
        else:
            statistics.append(
                {
                    "original_id": id,
                    "replacement_id": "",
                    "type": element["type"],
                }
            )

    return statistics