Skip to content

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 
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
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 
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
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:
                importer_results = import_grid_model(
                    start_command=command,
                    status_update_fn=heartbeat_fn,
                    unprocessed_gridfile_fs=unprocessed_gridfile_fs,
                    processed_gridfile_fs=processed_gridfile_fs,
                )

                result = preprocess(
                    start_command=command,
                    import_results=importer_results,
                    status_update_fn=heartbeat_fn,
                    loadflow_result_fs=loadflow_result_fs,
                    processed_gridfile_fs=processed_gridfile_fs,
                )

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

import_grid_model #

import_grid_model(
    start_command,
    unprocessed_gridfile_fs,
    processed_gridfile_fs,
    status_update_fn,
)

Run the import procedure.

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

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
ImportResult

A result dataclass from the importer, mainly including the grid folder and some stats
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 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
def import_grid_model(
    start_command: StartPreprocessingCommand,
    unprocessed_gridfile_fs: AbstractFileSystem,
    processed_gridfile_fs: AbstractFileSystem,
    status_update_fn: StatusUpdateFn,
) -> ImportResult:
    """Run the import procedure.

    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 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.
    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
    -------
    ImportResult
        A result dataclass from the importer, mainly including the grid folder and some stats

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

run_initial_loadflow #

run_initial_loadflow(
    start_command,
    processed_gridfile_dirfs,
    status_update_fn,
    loadflow_result_fs,
    lf_params=None,
)

Run the initial AC contingency analysis

PARAMETER DESCRIPTION
start_command

The command that was sent to the worker

TYPE: StartPreprocessingCommand

processed_gridfile_dirfs

A filesystem where the processed gridfiles are stored. This is assumed to be a dirfs pointing to the data folder for this import job, where the preprocessed gridfiles are stored

TYPE: AbstractFileSystem

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

loadflow_result_fs

A filesystem where the loadflow results are stored - this should be a NFS share together with the backend and optimizer. The importer needs this to store the initial loadflows

TYPE: AbstractFileSystem

lf_params

The loadflow parameters to use for the runner, if any. This is passed in the preprocessing results and can be used to run the loadflows with the same parameters as the initial loadflow in the preprocessing step. If None, the runner will use default parameters.

TYPE: Optional[Parameters] DEFAULT: None

RETURNS DESCRIPTION
StoredLoadflowReference

A reference to the stored loadflow results
dict[MetricType, float]

A dictionary containing the computed metrics
Source code in packages/importer_pkg/src/toop_engine_importer/worker/preprocessor.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
def run_initial_loadflow(
    start_command: StartPreprocessingCommand,
    processed_gridfile_dirfs: AbstractFileSystem,
    status_update_fn: StatusUpdateFn,
    loadflow_result_fs: AbstractFileSystem,
    lf_params: Optional[pypowsybl.loadflow.Parameters] = None,
) -> tuple[StoredLoadflowReference, dict[MetricType, float]]:
    """Run the initial AC contingency analysis

    Parameters
    ----------
    start_command: StartPreprocessingCommand
        The command that was sent to the worker
    processed_gridfile_dirfs: AbstractFileSystem
        A filesystem where the processed gridfiles are stored. This is assumed to be a dirfs pointing to the data folder for
        this import job, where the preprocessed gridfiles are stored
    status_update_fn: StatusUpdateFn
        A function to call to signal progress in the preprocessing pipeline. Takes a stage and an
        optional message as parameters
    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 importer needs this to store the initial loadflows
    lf_params: Optional[pypowsybl.loadflow.Parameters]
        The loadflow parameters to use for the runner, if any. This is passed in the preprocessing results
        and can be used to run the loadflows with the same parameters as the initial loadflow in the preprocessing step.
        If None, the runner will use default parameters.

    Returns
    -------
    StoredLoadflowReference
        A reference to the stored loadflow results
    dict[MetricType, float]
        A dictionary containing the computed metrics
    """
    status_update_fn("prepare_contingency_analysis", "Preparing initial loadflow contingency analysis")
    n_minus_1_definition = load_pydantic_model_fs(
        filesystem=processed_gridfile_dirfs,
        file_path=Path(PREPROCESSING_PATHS["nminus1_definition_file_path"]),
        model_class=Nminus1Definition,
    )
    net = load_base_grid_fs(processed_gridfile_dirfs, Path(PREPROCESSING_PATHS["grid_file_path_powsybl"]), "powsybl")
    status_update_fn("run_contingency_analysis", "Running initial loadflow contingency analysis")
    timestep_result_polars = get_ac_loadflow_results(
        net=net,
        n_minus_1_definition=n_minus_1_definition,
        timestep=0,
        job_id=start_command.preprocess_id,
        n_processes=start_command.preprocess_parameters.initial_loadflow_processes,
        lf_params=lf_params,
    )
    ref_polars = save_loadflow_results_polars(
        loadflow_result_fs, f"initial_loadflow_{start_command.preprocess_id}", timestep_result_polars
    )
    metrics = compute_metrics(
        timestep_result_polars,
        base_case_id=n_minus_1_definition.base_case.id if n_minus_1_definition.base_case is not None else None,
    )
    return ref_polars, metrics

preprocess #

preprocess(
    start_command,
    import_results,
    status_update_fn,
    loadflow_result_fs,
    processed_gridfile_fs,
)

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

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

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

RETURNS DESCRIPTION
PreprocessingSuccessResult

A result dataclass for the entire preprocessing, including paths to the ready static_information and network_data dataclasses.
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 
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
def preprocess(
    start_command: StartPreprocessingCommand,
    import_results: ImportResult,
    status_update_fn: StatusUpdateFn,
    loadflow_result_fs: AbstractFileSystem,
    processed_gridfile_fs: AbstractFileSystem,
) -> 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
    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.
    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.


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

    Raises
    ------
    Exception
        Any exception raised will be caught by the worker and sent back
    """
    logger.info("Starting preprocessing", preprocess_id=start_command.preprocess_id)
    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,
    )

    initial_loadflow, lf_metrics = run_initial_loadflow(
        start_command=start_command,
        processed_gridfile_dirfs=output_dirfs,
        status_update_fn=status_update_fn,
        loadflow_result_fs=loadflow_result_fs,
        lf_params=lf_params,
    )

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

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

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=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=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=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=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 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
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 
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
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",
    "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=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=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=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=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 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
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 
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
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 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
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 
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
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 
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
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 
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
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 
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
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 
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
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 
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
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 
176
177
178
179
180
181
182
183
184
185
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.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 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
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.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 
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
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 
239
240
241
242
243
244
245
246
247
248
249
250
251
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.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 
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
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 
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
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.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 
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
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).

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
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 
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
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.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 
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
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 
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
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 
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
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. 

1
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 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
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 
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
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 
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
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 
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
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 
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
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.replace_xward_by_internal_elements(net)
    pp.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 
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
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 
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
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_file",
    "convert_low_impedance_lines",
    "create_default_network_masks",
    "get_branches_df_with_element_name",
    "get_bus_breaker_master_asset_topology",
    "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 
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
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 
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
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"] = "N-1"
    # 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 
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
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 
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
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 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
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 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
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 
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
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 
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
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 = update_reward_masks_to_include_border_branches(network_masks, importer_parameters)
    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 
954
955
956
957
958
959
960
961
962
963
964
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 
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
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
    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 
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
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()

convert_file #

convert_file(
    importer_parameters,
    status_update_fn=empty_status_update_fn,
    processed_gridfile_fs=None,
    unprocessed_gridfile_fs=None,
)

Convert the grid file to a format that can be used by the preprocessing.

Saves data and network to the output folder.

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

RETURNS DESCRIPTION
ImportResult

The result of the import process.
Source code in packages/importer_pkg/src/toop_engine_importer/pypowsybl_import/preprocessing.py 
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
def convert_file(
    importer_parameters: BaseImporterParameters,
    status_update_fn: StatusUpdateFn = empty_status_update_fn,
    processed_gridfile_fs: Optional[AbstractFileSystem] = None,
    unprocessed_gridfile_fs: Optional[AbstractFileSystem] = None,
) -> ImportResult:
    """Convert the grid file to a format that can be used by the preprocessing.

    Saves data and network to the output folder.

    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.

    Returns
    -------
    ImportResult
        The result of the import process.
    """
    if unprocessed_gridfile_fs is None:
        unprocessed_gridfile_fs = LocalFileSystem()
    if processed_gridfile_fs is None:
        processed_gridfile_fs = LocalFileSystem()
    network = 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
    if importer_parameters.loadflow_parameters_file:
        lf_params = load_lf_params_from_fs(
            filesystem=unprocessed_gridfile_fs,
            file_path=importer_parameters.loadflow_parameters_file,
        )
        main_result, *_ = pypowsybl.loadflow.run_ac(network, parameters=lf_params)
    else:
        lf_params, main_result = find_converging_loadflow_params(importer_parameters, network)

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

    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,
    )
    if importer_parameters.loadflow_parameters_file:
        lf_params = load_lf_params_from_fs(
            filesystem=unprocessed_gridfile_fs,
            file_path=importer_parameters.loadflow_parameters_file,
        )
        main_result, *_ = pypowsybl.loadflow.run_ac(network, parameters=lf_params)
    else:
        lf_params, main_result = find_converging_loadflow_params(importer_parameters, network)
    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_and_n_1_definition(
        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")
        if main_result.status != pypowsybl.loadflow.ComponentStatus.CONVERGED:
            pypowsybl.loadflow.run_dc(network, parameters=lf_params)
        create_new_border_limits(network, network_masks, importer_parameters)
        # save new border limits
        save_powsybl_to_fs(
            network,
            filesystem=processed_gridfile_fs,
            file_path=grid_file_path,
        )

    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_from_masks(network, network_masks)
    save_pydantic_model_fs(
        filesystem=processed_gridfile_fs,
        file_path=importer_parameters.data_folder / PREPROCESSING_PATHS["nminus1_definition_file_path"],
        pydantic_model=nminus1_definition,
    )

    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 statistics.import_result

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 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
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 
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
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__)

CONVERTED_TRAFO3W_ENDING module-attribute #

CONVERTED_TRAFO3W_ENDING = '-Leg[123]$'

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 
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
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 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
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
)

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

RETURNS DESCRIPTION
Nminus1Definition

The created N-1 definition.
Source code in packages/importer_pkg/src/toop_engine_importer/pypowsybl_import/preprocessing.py 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
def create_nminus1_definition_from_masks(network: Network, network_masks: NetworkMasks) -> 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.

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

    lines = network.get_lines(attributes=["name"])
    monitored_lines = [
        MonitoredElement(id=idx, name=row["name"], type="LINE", kind="branch")
        for idx, row in lines[network_masks.line_for_reward].iterrows()
    ]
    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(CONVERTED_TRAFO3W_ENDING)
    monitored_trafos = [
        MonitoredElement(id=idx, name=row["name"], type="TWO_WINDINGS_TRANSFORMER", kind="branch")
        for idx, row in trafos[is_trafo2w & network_masks.trafo_for_reward].iterrows()
    ]
    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(CONVERTED_TRAFO3W_ENDING)
    trafos.index = trafos.index.str.replace(CONVERTED_TRAFO3W_ENDING, "", regex=True)
    if not trafos.empty:
        trafos.name = trafos.name.str.replace(CONVERTED_TRAFO3W_ENDING, "", regex=True)

    monitored_trafo3w = [
        MonitoredElement(id=idx, name=row["name"], type="THREE_WINDINGS_TRANSFORMER", kind="branch")
        for idx, row in trafos[is_trafo3w & network_masks.trafo_for_reward].drop_duplicates().iterrows()
    ]
    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 = [
        MonitoredElement(id=idx, name=row["name"], type="TIE_LINE", kind="branch")
        for idx, row in tie_lines[network_masks.tie_line_for_reward].iterrows()
    ]
    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()
    ]

    switches = network.get_switches(attributes=["name"])
    monitored_switches = [
        MonitoredElement(id=idx, name=row["name"], type="SWITCH", kind="branch")
        for idx, row in switches[network_masks.switch_for_reward].iterrows()
    ]
    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()
    ]

    buses = network.get_buses()
    relevant_bus_ids = buses.index[network_masks.relevant_subs].to_list()
    busbar_sections = network.get_busbar_sections(attributes=["name", "bus_id"])
    monitored_busbars = [
        MonitoredElement(id=idx, name=row["name"], type="BUSBAR_SECTION", kind="bus")
        for idx, row in busbar_sections[busbar_sections.index.isin(relevant_bus_ids)].iterrows()
    ]
    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()
    ]
    busbreaker_buses = network.get_bus_breaker_view_buses(attributes=["name", "bus_id"])
    monitored_busbreakers = [
        MonitoredElement(id=idx, name=row["name"], type="BUS_BREAKER_BUS", kind="bus")
        for idx, row in busbreaker_buses[busbreaker_buses.index.isin(relevant_bus_ids)].iterrows()
    ]

    nminus1_definition = Nminus1Definition(
        monitored_elements=(
            monitored_lines
            + monitored_trafos
            + monitored_trafo3w
            + monitored_tie_lines
            + monitored_switches
            + monitored_busbars
            + monitored_busbreakers
        ),
        contingencies=(
            contingencies
            + outaged_lines
            + outaged_trafos
            + outaged_trafo3w
            + outaged_tie_lines
            + outaged_boundary
            + outaged_generators
            + outaged_loads
            + outaged_switches
            + outaged_busbars
        ),
    )
    return nminus1_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.

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.
Source code in packages/importer_pkg/src/toop_engine_importer/pypowsybl_import/preprocessing.py 
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
def load_and_prepare_network(
    importer_parameters: BaseImporterParameters,
    processed_gridfile_fs: AbstractFileSystem,
    unprocessed_gridfile_fs: AbstractFileSystem,
    status_update_fn: StatusUpdateFn,
) -> Network:
    """Copy, load, and normalize the input network before preprocessing.

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

    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("-Leg")][["group_name"]].rename(
            columns={"group_name": "selected_limits_group_1"}
        )
        trafo3w_lims.index.name = "id"
        network.update_2_windings_transformers(trafo3w_lims)

    return network

convert_file #

convert_file(
    importer_parameters,
    status_update_fn=empty_status_update_fn,
    processed_gridfile_fs=None,
    unprocessed_gridfile_fs=None,
)

Convert the grid file to a format that can be used by the preprocessing.

Saves data and network to the output folder.

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

RETURNS DESCRIPTION
ImportResult

The result of the import process.
Source code in packages/importer_pkg/src/toop_engine_importer/pypowsybl_import/preprocessing.py 
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
def convert_file(
    importer_parameters: BaseImporterParameters,
    status_update_fn: StatusUpdateFn = empty_status_update_fn,
    processed_gridfile_fs: Optional[AbstractFileSystem] = None,
    unprocessed_gridfile_fs: Optional[AbstractFileSystem] = None,
) -> ImportResult:
    """Convert the grid file to a format that can be used by the preprocessing.

    Saves data and network to the output folder.

    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.

    Returns
    -------
    ImportResult
        The result of the import process.
    """
    if unprocessed_gridfile_fs is None:
        unprocessed_gridfile_fs = LocalFileSystem()
    if processed_gridfile_fs is None:
        processed_gridfile_fs = LocalFileSystem()
    network = 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
    if importer_parameters.loadflow_parameters_file:
        lf_params = load_lf_params_from_fs(
            filesystem=unprocessed_gridfile_fs,
            file_path=importer_parameters.loadflow_parameters_file,
        )
        main_result, *_ = pypowsybl.loadflow.run_ac(network, parameters=lf_params)
    else:
        lf_params, main_result = find_converging_loadflow_params(importer_parameters, network)

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

    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,
    )
    if importer_parameters.loadflow_parameters_file:
        lf_params = load_lf_params_from_fs(
            filesystem=unprocessed_gridfile_fs,
            file_path=importer_parameters.loadflow_parameters_file,
        )
        main_result, *_ = pypowsybl.loadflow.run_ac(network, parameters=lf_params)
    else:
        lf_params, main_result = find_converging_loadflow_params(importer_parameters, network)
    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_and_n_1_definition(
        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")
        if main_result.status != pypowsybl.loadflow.ComponentStatus.CONVERGED:
            pypowsybl.loadflow.run_dc(network, parameters=lf_params)
        create_new_border_limits(network, network_masks, importer_parameters)
        # save new border limits
        save_powsybl_to_fs(
            network,
            filesystem=processed_gridfile_fs,
            file_path=grid_file_path,
        )

    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_from_masks(network, network_masks)
    save_pydantic_model_fs(
        filesystem=processed_gridfile_fs,
        file_path=importer_parameters.data_folder / PREPROCESSING_PATHS["nminus1_definition_file_path"],
        pydantic_model=nminus1_definition,
    )

    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 statistics.import_result

compute_network_masks_and_n_1_definition #

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

Create, persist, and return network masks plus the derived N-1 definition.

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 and N-1 definition.

TYPE: AbstractFileSystem

unprocessed_gridfile_fs

Filesystem used to resolve auxiliary inputs required during mask creation.

TYPE: AbstractFileSystem

network

Powsybl network for which masks and contingencies are computed.

TYPE: Network

statistics

Statistics object updated while generating masks.

TYPE: PreProcessingStatistics

RETURNS DESCRIPTION
NetworkMasks

Generated network masks after saving them and the derived N-1 definition.
Source code in packages/importer_pkg/src/toop_engine_importer/pypowsybl_import/preprocessing.py 
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
def compute_network_masks_and_n_1_definition(
    importer_parameters: Union[UcteImporterParameters, CgmesImporterParameters],
    processed_gridfile_fs: AbstractFileSystem,
    unprocessed_gridfile_fs: AbstractFileSystem,
    network: Network,
    statistics: PreProcessingStatistics,
) -> NetworkMasks:
    """Create, persist, and return network masks plus the derived N-1 definition.

    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 and N-1 definition.
    unprocessed_gridfile_fs : AbstractFileSystem
        Filesystem used to resolve auxiliary inputs required during mask creation.
    network : Network
        Powsybl network for which masks and contingencies are computed.
    statistics : PreProcessingStatistics
        Statistics object updated while generating masks.

    Returns
    -------
    NetworkMasks
        Generated network masks after saving them and the derived N-1 definition.
    """
    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
    )

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

    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 
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
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 
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
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 
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
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 
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
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 
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
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_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 
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
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_reward = int(network_masks.line_for_reward.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_reward = int(network_masks.trafo_for_reward.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_reward = int(network_masks.tie_line_for_reward.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_reward = int(network_masks.switch_for_reward.sum())

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 
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
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 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
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 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
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 
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
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(int)
    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 
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
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 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
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 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
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 
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
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 
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
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(int)
    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 
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
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 
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
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 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
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 
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
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 
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
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 
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
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 
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
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 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
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 
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
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 
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
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 
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
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 
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
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 
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
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 
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
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 
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
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 
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
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 
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
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 
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
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 
139
140
141
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 
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
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 
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
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 
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
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 
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
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 
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
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 
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
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 
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
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 
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
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 
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
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 
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
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 
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
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 
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
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