Skip to content

Topology Perturbations

This module provides classes for generating perturbed network topologies.

TopologyGenerator

Bases: ABC

Abstract base class for generating perturbed network topologies.

Source code in gridfm_datakit/perturbations/topology_perturbation.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
class TopologyGenerator(ABC):
    """Abstract base class for generating perturbed network topologies."""

    def __init__(self) -> None:
        """Initialize the topology generator."""
        pass

    @abstractmethod
    def generate(
        self,
        net: Network,
    ) -> Union[Generator[Network, None, None], List[Network]]:
        """Generate perturbed topologies.

        Args:
            net: The power network to perturb.

        Yields:
            A perturbed network topology.
        """
        pass
__init__()

Initialize the topology generator.

Source code in gridfm_datakit/perturbations/topology_perturbation.py 
15
16
17
def __init__(self) -> None:
    """Initialize the topology generator."""
    pass
generate(net) abstractmethod

Generate perturbed topologies.

Parameters:

Name Type Description Default
net Network

The power network to perturb.

 required

Yields:

Type Description
Union[Generator[Network, None, None], List[Network]]

A perturbed network topology.
Source code in gridfm_datakit/perturbations/topology_perturbation.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
@abstractmethod
def generate(
    self,
    net: Network,
) -> Union[Generator[Network, None, None], List[Network]]:
    """Generate perturbed topologies.

    Args:
        net: The power network to perturb.

    Yields:
        A perturbed network topology.
    """
    pass

NoPerturbationGenerator

Bases: TopologyGenerator

Generator that yields the original network without any perturbations.

Source code in gridfm_datakit/perturbations/topology_perturbation.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
class NoPerturbationGenerator(TopologyGenerator):
    """Generator that yields the original network without any perturbations."""

    def generate(
        self,
        net: Network,
    ) -> Generator[Network, None, None]:
        """Yield the original network without any perturbations.

        Args:
            net: The power network.

        Yields:
            The original power network.
        """
        yield copy.deepcopy(net)
generate(net)

Yield the original network without any perturbations.

Parameters:

Name Type Description Default
net Network

The power network.

 required

Yields:

Type Description
Network

The original power network.
Source code in gridfm_datakit/perturbations/topology_perturbation.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
def generate(
    self,
    net: Network,
) -> Generator[Network, None, None]:
    """Yield the original network without any perturbations.

    Args:
        net: The power network.

    Yields:
        The original power network.
    """
    yield copy.deepcopy(net)

NMinusKGenerator

Bases: TopologyGenerator

Generate perturbed topologies for N-k contingency analysis.

Only considers lines and transformers. Generates ALL possible topologies with at most k components set out of service (lines and transformers).

Only topologies that are feasible (= no unsupplied buses) are yielded.

Attributes:

Name Type Description
k

Maximum number of components to drop.
components_to_drop

List of tuples containing component indices and types.
component_combinations

List of all possible combinations of components to drop.
Source code in gridfm_datakit/perturbations/topology_perturbation.py 
 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
113
114
115
116
117
118
119
class NMinusKGenerator(TopologyGenerator):
    """Generate perturbed topologies for N-k contingency analysis.

    Only considers lines and transformers. Generates ALL possible topologies with at most k
    components set out of service (lines and transformers).

    Only topologies that are feasible (= no unsupplied buses) are yielded.

    Attributes:
        k: Maximum number of components to drop.
        components_to_drop: List of tuples containing component indices and types.
        component_combinations: List of all possible combinations of components to drop.
    """

    def __init__(self, k: int, base_net: dict) -> None:
        """Initialize the N-k generator.

        Args:
            k: Maximum number of components to drop.
            base_net: The base power network.

        Raises:
            ValueError: If k is 0.
            Warning: If k > 1, as this may result in slow data generation.
        """
        super().__init__()
        if k > 1:
            warnings.warn("k>1. This may result in slow data generation process.")
        if k == 0:
            raise ValueError(
                'k must be greater than 0. Use "none" as argument for the generator_type if you don\'t want to generate any perturbation',
            )
        self.k = k

        # Prepare the list of components to drop
        self.components_to_drop = base_net.idx_branches_in_service

        # Generate all combinations of at most k components
        self.component_combinations = []
        for r in range(self.k + 1):
            self.component_combinations.extend(combinations(self.components_to_drop, r))

        print(
            f"Number of possible topologies with at most {self.k} dropped components: {len(self.component_combinations)}",
        )

    def generate(
        self,
        net: Network,
    ) -> Generator[Network, None, None]:
        """Generate perturbed topologies by dropping components.
        Does not change the original network.

        Args:
            net: The power network.

        Yields:
            A perturbed network topology with at most k components removed.
        """
        for selected_components in self.component_combinations:
            perturbed_topology = copy.deepcopy(net)

            perturbed_topology.deactivate_branches(selected_components)

            # Check network feasibility and yield the topology
            if perturbed_topology.check_single_connected_component():
                yield perturbed_topology
__init__(k, base_net)

Initialize the N-k generator.

Parameters:

Name Type Description Default
k int

Maximum number of components to drop.

 required
base_net dict

The base power network.

 required

Raises:

Type Description
ValueError

If k is 0.
Warning

If k > 1, as this may result in slow data generation.
Source code in gridfm_datakit/perturbations/topology_perturbation.py 
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
def __init__(self, k: int, base_net: dict) -> None:
    """Initialize the N-k generator.

    Args:
        k: Maximum number of components to drop.
        base_net: The base power network.

    Raises:
        ValueError: If k is 0.
        Warning: If k > 1, as this may result in slow data generation.
    """
    super().__init__()
    if k > 1:
        warnings.warn("k>1. This may result in slow data generation process.")
    if k == 0:
        raise ValueError(
            'k must be greater than 0. Use "none" as argument for the generator_type if you don\'t want to generate any perturbation',
        )
    self.k = k

    # Prepare the list of components to drop
    self.components_to_drop = base_net.idx_branches_in_service

    # Generate all combinations of at most k components
    self.component_combinations = []
    for r in range(self.k + 1):
        self.component_combinations.extend(combinations(self.components_to_drop, r))

    print(
        f"Number of possible topologies with at most {self.k} dropped components: {len(self.component_combinations)}",
    )
generate(net)

Generate perturbed topologies by dropping components. Does not change the original network.

Parameters:

Name Type Description Default
net Network

The power network.

 required

Yields:

Type Description
Network

A perturbed network topology with at most k components removed.
Source code in gridfm_datakit/perturbations/topology_perturbation.py 
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
def generate(
    self,
    net: Network,
) -> Generator[Network, None, None]:
    """Generate perturbed topologies by dropping components.
    Does not change the original network.

    Args:
        net: The power network.

    Yields:
        A perturbed network topology with at most k components removed.
    """
    for selected_components in self.component_combinations:
        perturbed_topology = copy.deepcopy(net)

        perturbed_topology.deactivate_branches(selected_components)

        # Check network feasibility and yield the topology
        if perturbed_topology.check_single_connected_component():
            yield perturbed_topology

RandomComponentDropGenerator

Bases: TopologyGenerator

Generate perturbed topologies by randomly setting components out of service.

Generates perturbed topologies by randomly setting out of service at most k components among the selected element types. Only topologies that are feasible (= no unsupplied buses) are yielded.

Attributes:

Name Type Description
n_topology_variants

Number of topology variants to generate.
k

Maximum number of components to drop.
components_to_drop

List of tuples containing component indices and types.
Source code in gridfm_datakit/perturbations/topology_perturbation.py 
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
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
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
class RandomComponentDropGenerator(TopologyGenerator):
    """Generate perturbed topologies by randomly setting components out of service.

    Generates perturbed topologies by randomly setting out of service at most k components among the selected element types.
    Only topologies that are feasible (= no unsupplied buses) are yielded.

    Attributes:
        n_topology_variants: Number of topology variants to generate.
        k: Maximum number of components to drop.
        components_to_drop: List of tuples containing component indices and types.
    """

    def __init__(
        self,
        n_topology_variants: int,
        k: int,
        base_net: Network,
        elements: List[str] = ["branch", "gen"],
        outage_count_probabilities: Sequence[float] | Mapping[int, float] | None = None,
        max_generation_attempts: int | None = None,
    ) -> None:
        """Initialize the random component drop generator.

        Args:
            n_topology_variants: Number of topology variants to generate.
            k: Maximum number of components to drop.
            base_net: The base power network.
            elements: List of element types to consider for dropping.
            outage_count_probabilities: Optional probabilities over outage counts.
                Index or key `i` means probability of sampling `i` outages.
            max_generation_attempts: Optional cap on sampled topology attempts
                before failing if too many sampled topologies are infeasible.
                If not provided, a default limit of
                ``max(500, 50 * n_topology_variants)`` is used.
        """
        super().__init__()
        self.n_topology_variants = n_topology_variants
        self.k = k

        # Create a list of all components that can be dropped
        self.components_to_drop = []
        if "branch" in elements:
            self.components_to_drop.extend(
                (idx, "branch") for idx in base_net.idx_branches_in_service
            )
        if "gen" in elements:
            self.components_to_drop.extend(
                (idx, "gen")
                for idx in base_net.idx_gens_in_service
                if base_net.gens[idx, GEN_BUS] != base_net.ref_bus_idx
            )

        # Preserve the current 1..k uniform sampling when no explicit count
        # probabilities are provided, but allow configurable sampling over 0..k.
        self.outage_count_values, self.outage_count_probabilities = (
            self._normalize_outage_count_probabilities(k, outage_count_probabilities)
        )
        self.max_generation_attempts = self._resolve_max_generation_attempts(
            n_topology_variants,
            max_generation_attempts,
        )

    def generate(
        self,
        net: Network,
    ) -> Generator[Network, None, None]:
        """Generate perturbed topologies by randomly setting components out of service.

        Args:
            net: The power network.

        Yields:
            A perturbed network topology.
        """
        n_generated_topologies = 0
        n_attempts = 0

        # Stop after we generated n_topology_variants
        while n_generated_topologies < self.n_topology_variants:
            if n_attempts >= self.max_generation_attempts:
                raise RuntimeError(
                    "Unable to generate "
                    f"{self.n_topology_variants} feasible topologies within "
                    f"{self.max_generation_attempts} attempts. "
                    "Consider reducing k or relaxing outage_count_probabilities.",
                )
            n_attempts += 1
            perturbed_topology = copy.deepcopy(net)

            r = self._sample_outage_count()

            # Randomly select r<=k components to drop
            components = tuple(
                np.random.choice(range(len(self.components_to_drop)), r, replace=False),
            )

            # Convert indices back to actual components
            selected_components = tuple(
                self.components_to_drop[idx] for idx in components
            )

            # Separate lines, transformers, generators, and static generators
            branches_to_drop = [
                idx for idx, element in selected_components if element == "branch"
            ]
            gens_to_drop = [
                idx for idx, element in selected_components if element == "gen"
            ]

            # Drop selected lines and transformers, turn off generators and static generators
            perturbed_topology.deactivate_branches(branches_to_drop)
            perturbed_topology.deactivate_gens(gens_to_drop)

            # Check network feasibility and yield the topology
            if perturbed_topology.check_single_connected_component():
                yield perturbed_topology
                n_generated_topologies += 1

    def _sample_outage_count(self) -> int:
        if self.outage_count_probabilities is None:
            return int(np.random.randint(1, self.k + 1))
        return int(
            np.random.choice(
                self.outage_count_values,
                p=self.outage_count_probabilities,
            ),
        )

    @staticmethod
    def _resolve_max_generation_attempts(
        n_topology_variants: int,
        max_generation_attempts: int | None,
    ) -> int:
        if max_generation_attempts is None:
            return max(500, 50 * max(1, int(n_topology_variants)))
        if int(max_generation_attempts) <= 0:
            raise ValueError("max_generation_attempts must be greater than 0.")
        return int(max_generation_attempts)

    @staticmethod
    def _normalize_outage_count_probabilities(
        k: int,
        probabilities: Sequence[float] | Mapping[int, float] | None,
    ) -> tuple[np.ndarray | None, np.ndarray | None]:
        """Validate optional outage-count probabilities for random topology sampling."""

        if probabilities is None:
            return None, None

        if isinstance(probabilities, Mapping):
            allowed_counts = np.arange(k + 1, dtype=int)
            probability_values = np.zeros(k + 1, dtype=float)
            for raw_count, raw_probability in probabilities.items():
                count = int(raw_count)
                if count < 0 or count > k:
                    raise ValueError(
                        f"Outage count {count} is outside the supported range 0..{k}.",
                    )
                probability_values[count] = float(raw_probability)
            return RandomComponentDropGenerator._validate_outage_count_probabilities(
                allowed_counts,
                probability_values,
            )

        probability_values = np.asarray(probabilities, dtype=float)
        if probability_values.ndim != 1:
            raise ValueError(
                "outage_count_probabilities must be a one-dimensional sequence.",
            )
        if len(probability_values) != k + 1:
            raise ValueError(
                "outage_count_probabilities sequence must have length k + 1 so index i maps to i outages.",
            )
        allowed_counts = np.arange(k + 1, dtype=int)
        return RandomComponentDropGenerator._validate_outage_count_probabilities(
            allowed_counts,
            probability_values,
        )

    @staticmethod
    def _validate_outage_count_probabilities(
        counts: np.ndarray,
        probability_values: np.ndarray,
    ) -> tuple[np.ndarray, np.ndarray]:
        if np.any(probability_values < 0.0):
            raise ValueError("outage_count_probabilities must be non-negative.")
        total_probability = float(probability_values.sum())
        if not np.isclose(total_probability, 1.0, atol=1e-8):
            raise ValueError(
                "outage_count_probabilities must sum to 1.0 within numerical tolerance.",
            )
        if not np.any(probability_values > 0.0):
            raise ValueError(
                "outage_count_probabilities must contain at least one positive value.",
            )
        return counts.astype(int, copy=True), probability_values.astype(
            float,
            copy=True,
        )
__init__(n_topology_variants, k, base_net, elements=['branch', 'gen'], outage_count_probabilities=None, max_generation_attempts=None)

Initialize the random component drop generator.

Parameters:

Name Type Description Default
n_topology_variants int

Number of topology variants to generate.

 required
k int

Maximum number of components to drop.

 required
base_net Network

The base power network.

 required
elements List[str]

List of element types to consider for dropping.

 ['branch', 'gen']
outage_count_probabilities Sequence[float] | Mapping[int, float] | None

Optional probabilities over outage counts. Index or key i means probability of sampling i outages.

 None
max_generation_attempts int | None

Optional cap on sampled topology attempts before failing if too many sampled topologies are infeasible. If not provided, a default limit of max(500, 50 * n_topology_variants) is used.

 None
Source code in gridfm_datakit/perturbations/topology_perturbation.py 
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 __init__(
    self,
    n_topology_variants: int,
    k: int,
    base_net: Network,
    elements: List[str] = ["branch", "gen"],
    outage_count_probabilities: Sequence[float] | Mapping[int, float] | None = None,
    max_generation_attempts: int | None = None,
) -> None:
    """Initialize the random component drop generator.

    Args:
        n_topology_variants: Number of topology variants to generate.
        k: Maximum number of components to drop.
        base_net: The base power network.
        elements: List of element types to consider for dropping.
        outage_count_probabilities: Optional probabilities over outage counts.
            Index or key `i` means probability of sampling `i` outages.
        max_generation_attempts: Optional cap on sampled topology attempts
            before failing if too many sampled topologies are infeasible.
            If not provided, a default limit of
            ``max(500, 50 * n_topology_variants)`` is used.
    """
    super().__init__()
    self.n_topology_variants = n_topology_variants
    self.k = k

    # Create a list of all components that can be dropped
    self.components_to_drop = []
    if "branch" in elements:
        self.components_to_drop.extend(
            (idx, "branch") for idx in base_net.idx_branches_in_service
        )
    if "gen" in elements:
        self.components_to_drop.extend(
            (idx, "gen")
            for idx in base_net.idx_gens_in_service
            if base_net.gens[idx, GEN_BUS] != base_net.ref_bus_idx
        )

    # Preserve the current 1..k uniform sampling when no explicit count
    # probabilities are provided, but allow configurable sampling over 0..k.
    self.outage_count_values, self.outage_count_probabilities = (
        self._normalize_outage_count_probabilities(k, outage_count_probabilities)
    )
    self.max_generation_attempts = self._resolve_max_generation_attempts(
        n_topology_variants,
        max_generation_attempts,
    )
generate(net)

Generate perturbed topologies by randomly setting components out of service.

Parameters:

Name Type Description Default
net Network

The power network.

 required

Yields:

Type Description
Network

A perturbed network topology.
Source code in gridfm_datakit/perturbations/topology_perturbation.py 
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
def generate(
    self,
    net: Network,
) -> Generator[Network, None, None]:
    """Generate perturbed topologies by randomly setting components out of service.

    Args:
        net: The power network.

    Yields:
        A perturbed network topology.
    """
    n_generated_topologies = 0
    n_attempts = 0

    # Stop after we generated n_topology_variants
    while n_generated_topologies < self.n_topology_variants:
        if n_attempts >= self.max_generation_attempts:
            raise RuntimeError(
                "Unable to generate "
                f"{self.n_topology_variants} feasible topologies within "
                f"{self.max_generation_attempts} attempts. "
                "Consider reducing k or relaxing outage_count_probabilities.",
            )
        n_attempts += 1
        perturbed_topology = copy.deepcopy(net)

        r = self._sample_outage_count()

        # Randomly select r<=k components to drop
        components = tuple(
            np.random.choice(range(len(self.components_to_drop)), r, replace=False),
        )

        # Convert indices back to actual components
        selected_components = tuple(
            self.components_to_drop[idx] for idx in components
        )

        # Separate lines, transformers, generators, and static generators
        branches_to_drop = [
            idx for idx, element in selected_components if element == "branch"
        ]
        gens_to_drop = [
            idx for idx, element in selected_components if element == "gen"
        ]

        # Drop selected lines and transformers, turn off generators and static generators
        perturbed_topology.deactivate_branches(branches_to_drop)
        perturbed_topology.deactivate_gens(gens_to_drop)

        # Check network feasibility and yield the topology
        if perturbed_topology.check_single_connected_component():
            yield perturbed_topology
            n_generated_topologies += 1