Skip to content

rupture_id_filter

rupture_id_filter

This module provides a class for filtering solution ruptures.

Classes:

Name Description
FilterRuptureIds

a chainable filter for ruptures, returning qualifying rupture ids.

Examples:

>>> # ruptures within 50km of Hamilton with magnitude between 5.75 and 6.25.
>>> ham50 = solvis.circle_polygon(50000, -37.78, 175.28)  # 50km radius around Hamilton
<POLYGON ((175.849 -37.779, 175.847 -37.823, 175.839 -37.866, 175.825 -37.90...>
>>> solution = solvis.InversionSolution.from_archive(filename)
>>> rupture_ids = FilterRuptureIds(solution)\
        .for_magnitude(min_mag=5.75, max_mag=6.25)\
        .for_polygon(ham50)

>>> # ruptures on any of faults A, B, with magnitude and rupture rate limits
>>> rupture_ids = FilterRuptureIds(solution)\
>>>    .for_parent_fault_names(['Alpine: Jacksons to Kaniere', 'Vernon 1' ])\
>>>    .for_magnitude(7.0, 8.0)\
>>>    .for_rupture_rate(1e-6, 1e-2)

>>> # ruptures on fault A that do not involve fault B:
>>> rupture_ids = FilterRuptureIds(solution)\
>>>    .for_parent_fault_names(['Alpine: Jacksons to Kaniere'])\
>>>    .for_parent_fault_names(['Vernon 1'], join_prior='difference')

FilterRuptureIds

Bases: ChainableSetBase

A helper class to filter solution ruptures, returning the qualifying rupture_ids.

Source code in solvis/filter/rupture_id_filter.py 
 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
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
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
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
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
class FilterRuptureIds(ChainableSetBase):
    """A helper class to filter solution ruptures, returning the qualifying rupture_ids."""

    def __init__(
        self,
        solution: 'InversionSolution',
        drop_zero_rates: bool = True,
    ):
        """Instantiate a new filter.

        Args:
            solution: The solution instance to filter on.
            drop_zero_rates: Exclude ruptures with rupture_rate == 0 (default=True).
        """
        self._solution = solution
        self._drop_zero_rates = drop_zero_rates
        self._filter_subsection_ids = FilterSubsectionIds(solution)
        self._filter_parent_fault_ids = FilterParentFaultIds(solution)
        self.__rupture_sections: Optional['pd.DataFrame'] = None

    def _ruptures_with_or_without_rupture_rates(self, drop_zero_rates: bool = False) -> 'pd.DataFrame':
        """Get ruptures with or without rupture rates.

        Catering for the difference in rates structure of InversionSolution subclasses.

        Args:
            drop_zero_rates: If True, exclude ruptures with zero rupture rate.

        Returns:
            DataFrame containing ruptures with and without rupture rates.
        """
        if isinstance(self._solution, solvis.solution.fault_system_solution.FaultSystemSolution):
            df_rr: pd.DataFrame = self._solution.solution_file.rupture_rates.drop(
                columns=["Rupture Index", "fault_system"]
            )
            df_rr.index = df_rr.index.droplevel(0)  # so we're indexed by "Rupture Index" without " ault_system"
        else:
            df_rr = self._solution.solution_file.rupture_rates.drop(columns=["Rupture Index"])

        if drop_zero_rates:
            rate_column = self._solution.model.rate_column_name()
            nonzero = df_rr[rate_column] > 0
            df_rr = df_rr[nonzero]

        return self._solution.solution_file.ruptures.join(df_rr, on="Rupture Index", rsuffix='_r', how='inner')

    def _adapted_rupture_sections(self) -> 'pd.DataFrame':
        """Get adapted rupture sections based on the drop_zero_rates flag.

        The adaption caches for better performance in client functions

        Returns:
            DataFrame containing adapted rupture sections.
        """
        if self.__rupture_sections is not None:
            return self.__rupture_sections

        df0: pd.DataFrame = self._solution.model.rupture_sections
        rate_column = self._solution.model.rate_column_name()
        if self._drop_zero_rates:
            df1 = self._ruptures_with_or_without_rupture_rates(drop_zero_rates=self._drop_zero_rates)
            df0 = df0.join(df1.set_index("Rupture Index"), on='rupture', how='inner')[
                [rate_column, "rupture", "section"]
            ]

        self.__rupture_sections = df0
        return self.__rupture_sections

    def _get_rupture_ids_for_subsection_ids(self, subsection_ids: Iterable[int]) -> Set[int]:
        """Get rupture ids for given subsection ids.

        Args:
            subsection_ids (Iterable[int]): A collection of subsection ids.

        Returns:
            Set[int]: A set containing the rupture ids that correspond to the provided subsection ids.
        """
        df0 = self._adapted_rupture_sections()
        ids = df0[df0['section'].isin(list(subsection_ids))]['rupture'].tolist()
        return set([int(id) for id in ids])

    def tolist(self) -> List[int]:
        """
        Returns the filtered rupture ids as a list of integers.

        Returns:
            A list of integers representing the filtered rupture ids.
        """
        return list(self)

    def all(self) -> ChainableSetBase:
        """Convenience method returning ids for all solution ruptures.

        NB the usual `join_prior` argument is not implemented as it doesn't seem useful here.

        Returns:
            A chainable set of all the rupture_ids.
        """
        result = set(self._solution.solution_file.ruptures['Rupture Index'].to_list())
        return self.new_chainable_set(result, self._solution)

    def for_named_fault_names(
        self,
        named_fault_names: Iterable[str],
        join_type: Union[SetOperationEnum, str] = 'union',
        join_prior: Union[SetOperationEnum, str] = 'intersection',
    ) -> ChainableSetBase:
        """Filter rupture ids based on named fault names.

        This method filters the rupture ids for given named fault names. It supports
        both intersection and union operations to combine results from multiple named
        faults. The `join_type` parameter determines how the sets of rupture ids are
        combined, while the `join_prior` parameter specifies how these sets are combined
        with any existing filters.

        Args:
            named_fault_names (Iterable[str]): A collection of named fault names.
            join_type (Union[SetOperationEnum, str], optional): The type of set operation to use for
                combining results from multiple named faults. It can be either 'union' or 'intersection'.
            join_prior (Union[SetOperationEnum, str], optional): The type of set operation to use when
                combining the filtered rupture ids with any existing filters. It can be either 'intersection'
                or 'difference'.

        Returns:
            ChainableSetBase: A chainable set containing the filtered rupture ids.

        Raises:
            ValueError: If an unsupported join_type is provided.
        """
        if isinstance(join_type, str):
            join_type = SetOperationEnum[join_type.upper()]

        rupture_id_sets: List[Set[int]] = []
        for named_fault_name in named_fault_names:
            parent_fault_ids = named_fault.named_fault_table().loc[named_fault_name].parent_fault_ids
            rupture_ids = self.for_parent_fault_ids(parent_fault_ids, join_prior=join_prior).chained_set
            rupture_id_sets.append(rupture_ids)

        if join_type == SetOperationEnum.INTERSECTION:
            rupture_ids = set.intersection(*rupture_id_sets)
        elif join_type == SetOperationEnum.UNION:
            rupture_ids = set.union(*rupture_id_sets)
        else:
            raise ValueError("Only INTERSECTION and UNION operations are supported for option 'join_type'")
        return self.new_chainable_set(rupture_ids, self._solution, self._drop_zero_rates, join_prior=join_prior)

    def for_parent_fault_names(
        self,
        parent_fault_names: Iterable[str],
        join_type: Union[SetOperationEnum, str] = 'union',
        join_prior: Union[SetOperationEnum, str] = 'intersection',
    ) -> ChainableSetBase:
        """Filter rupture ids based on parent fault names.

        This method filters the rupture ids for given parent fault names. It supports
        both intersection and union operations to combine results from multiple parent
        faults. The `join_type` parameter determines how the sets of rupture ids are
        combined, while the `join_prior` parameter specifies how these sets are combined
        with any existing filters.

        Args:
            parent_fault_names (Iterable[str]): A collection of parent fault names.
            join_type (Union[SetOperationEnum, str], optional): The type of set operation to use for
                combining results from multiple parent faults. It can be either 'union' or 'intersection'.
            join_prior (Union[SetOperationEnum, str], optional): The type of set operation to use when
                combining the filtered rupture ids with any existing filters. It can be either 'intersection'
                or 'difference'.

        Returns:
            ChainableSetBase: A chainable set containing the filtered rupture ids.

        Raises:
            ValueError: If an unsupported join_type is provided.
        """
        if isinstance(join_type, str):
            join_type = SetOperationEnum[join_type.upper()]

        parent_fault_ids = self._filter_parent_fault_ids.for_parent_fault_names(parent_fault_names)
        return self.for_parent_fault_ids(parent_fault_ids=parent_fault_ids, join_type=join_type, join_prior=join_prior)

    def for_parent_fault_id(
        self,
        parent_fault_id: int,
        join_prior: Union[SetOperationEnum, str] = 'intersection',
    ) -> ChainableSetBase:
        """
        Find ruptures that occur on the given parent fault id.

        Args:
            parent_fault_id: The parent fault id to filter by.
            join_prior: How to join this result with the prior chain (if any) (default = 'intersection').

        Returns:
            A chainable set of rupture_ids matching the filter.
        """
        subsection_ids = self._filter_subsection_ids.for_parent_fault_ids([parent_fault_id])
        rupture_ids = self._get_rupture_ids_for_subsection_ids(subsection_ids)
        return self.new_chainable_set(rupture_ids, self._solution, self._drop_zero_rates, join_prior=join_prior)

    def for_parent_fault_ids(
        self,
        parent_fault_ids: Iterable[int],
        join_type: Union[SetOperationEnum, str] = 'union',
        join_prior: Union[SetOperationEnum, str] = 'intersection',
    ) -> ChainableSetBase:
        """Find ruptures that occur on the given parent_fault ids.

        Args:
            parent_fault_ids: A list of one or more `parent_fault` ids.
            join_type: The type of set operation to perform when combining results from multiple parent_fault_ids.
                Options are 'union' and 'intersection'.
            join_prior: How to join this result with the prior chain (if any) (default = 'intersection').

        Returns:
            A chainable set of rupture_ids matching the filter.
        """
        if isinstance(join_type, str):
            join_type = SetOperationEnum[join_type.upper()]

        rupture_id_sets: List[Set[int]] = []
        for fault_id in parent_fault_ids:
            rupture_id_sets.append(self.for_parent_fault_id(fault_id, join_prior=join_prior).chained_set)

        if join_type == SetOperationEnum.INTERSECTION:
            rupture_ids = set.intersection(*rupture_id_sets)
        elif join_type == SetOperationEnum.UNION:
            rupture_ids = set.union(*rupture_id_sets)
        else:
            raise ValueError("Only INTERSECTION and UNION operations are supported for option 'join_type'")
        return self.new_chainable_set(rupture_ids, self._solution, self._drop_zero_rates, join_prior=join_prior)

    def for_subsection_ids(
        self,
        fault_section_ids: Iterable[int],
        join_type: Union[SetOperationEnum, str] = 'union',
        join_prior: Union[SetOperationEnum, str] = 'intersection',
    ) -> ChainableSetBase:
        """Find ruptures that occur on any of the given fault_section_ids.

        Return the Union of all rupture ids involvcedin te `fault_section_ids`.

        Args:
            fault_section_ids: A list of one or more fault_section ids.
            join_type: The type of set operation to perform when combining results from multiple fault_section_ids.
                Options are 'union' and 'intersection'.
            join_prior: How to join this result with the prior chain (if any) (default = 'intersection').

        Returns:
            A chainable set of rupture_ids matching the filter.
        """
        if isinstance(join_type, str):
            join_type = SetOperationEnum[join_type.upper()]

        rupture_id_sets: List[Set[int]] = []
        for fault_section_id in fault_section_ids:
            rupture_id_sets.append(self._get_rupture_ids_for_subsection_ids([fault_section_id]))
        if join_type == SetOperationEnum.INTERSECTION:
            rupture_ids = set.intersection(*rupture_id_sets)
        elif join_type == SetOperationEnum.UNION:
            rupture_ids = set.union(*rupture_id_sets)
        else:
            raise ValueError("Only INTERSECTION and UNION operations are supported for option 'join_type'")
        return self.new_chainable_set(rupture_ids, self._solution, self._drop_zero_rates, join_prior=join_prior)

    def for_rupture_rate(
        self,
        min_rate: Optional[float] = None,
        max_rate: Optional[float] = None,
        join_prior: Union[SetOperationEnum, str] = 'intersection',
    ) -> ChainableSetBase:
        """Find ruptures that occur within given rates bounds.

        Args:
            min_rate: The minumum rupture _rate bound.
            max_rate: The maximum rupture _rate bound.
            join_prior: How to join this result with the prior chain (if any) (default = 'intersection').


        Returns:
            A chainable set of rupture_ids matching the filter arguments.
        """
        index = "Rupture Index"
        df0 = self._ruptures_with_or_without_rupture_rates(drop_zero_rates=self._drop_zero_rates)

        rate_column = self._solution.model.rate_column_name()

        # rate col is different for InversionSolution
        df0 = df0 if not max_rate else df0[df0[rate_column] <= max_rate]
        df0 = df0 if not min_rate else df0[df0[rate_column] > min_rate]
        result = set(df0[index].tolist())
        return self.new_chainable_set(result, self._solution, self._drop_zero_rates, join_prior=join_prior)

    def for_magnitude(
        self,
        min_mag: Optional[float] = None,
        max_mag: Optional[float] = None,
        join_prior: Union[SetOperationEnum, str] = 'intersection',
    ) -> ChainableSetBase:
        """Find ruptures that occur within given magnitude bounds.

        Args:
            min_mag: The minumum rupture magnitude bound.
            max_mag: The maximum rupture magnitude bound.
            join_prior: How to join this result with the prior chain (if any) (default = 'intersection').

        Returns:
            A chainable set of rupture_ids matching the filter arguments.
        """
        index = "Rupture Index"
        df0 = self._ruptures_with_or_without_rupture_rates(drop_zero_rates=self._drop_zero_rates)

        df0 = df0 if not max_mag else df0[df0.Magnitude <= max_mag]
        df0 = df0 if not min_mag else df0[df0.Magnitude > min_mag]
        result = set(df0[index].tolist())
        return self.new_chainable_set(result, self._solution, self._drop_zero_rates, join_prior=join_prior)

    def for_polygons(
        self,
        polygons: Iterable[shapely.geometry.Polygon],
        join_type: Union[SetOperationEnum, str] = SetOperationEnum.UNION,
        join_prior: Union[SetOperationEnum, str] = 'intersection',
    ) -> ChainableSetBase:
        """Find ruptures involving several polygon areas.

        Each polygon will return a set of matching rupture ids, so the user may choose to override the
        default set operation (UNION) between these using the `join_type' argument.

        This method

        Args:
            polygons: Polygons defining the areas of interest.
            join_type: How to join the polygon results (default = 'union').
            join_prior: How to join this result with the prior chain (if any) (default = 'intersection').

        Returns:
            A chainable set of rupture_ids matching the filter arguments.
        """
        if isinstance(join_type, str):
            try:
                join_type = SetOperationEnum[join_type.upper()]
            except KeyError:
                raise ValueError(f'Unsupported set operation `{join_type}` for `join_type` argument.')

        rupture_id_sets: List[Set[int]] = []
        for polygon in polygons:
            rupture_id_sets.append(self.for_polygon(polygon, join_prior=join_prior).chained_set)

        if join_type == SetOperationEnum.INTERSECTION:
            rupture_ids = set.intersection(*rupture_id_sets)
        elif join_type == SetOperationEnum.UNION:
            rupture_ids = set.union(*rupture_id_sets)
        elif join_type == SetOperationEnum.DIFFERENCE:
            rupture_ids = set.difference(*rupture_id_sets)
        else:
            raise ValueError(
                "Only INTERSECTION, UNION & DIFFERENCE operations are supported for `join_type`"
            )  # pragma: no cover
        return self.new_chainable_set(rupture_ids, self._solution, self._drop_zero_rates, join_prior=join_prior)

    def for_polygon(
        self,
        polygon: shapely.geometry.Polygon,
        join_prior: Union[SetOperationEnum, str] = 'intersection',
    ) -> ChainableSetBase:
        """Find ruptures that involve a single polygon area.

        Args:
            polygon: The polygon defining the area of intersection.
            join_prior: How to join this result with the prior chain (if any) (default = 'intersection').

        Returns:
            A chainable set of rupture_ids matching the filter arguments.
        """
        fault_sections_df: pd.DataFrame = gpd.GeoDataFrame(self._solution.solution_file.fault_sections)

        # filter fault sections using the polygon geometry
        filtered_fault_sections_df = fault_sections_df[fault_sections_df['geometry'].intersects(polygon)]

        # filter ruptures by drop_zero_rates
        ruptures_df = self._ruptures_with_or_without_rupture_rates(drop_zero_rates=self._drop_zero_rates)

        # join filtered ruptures with rupture_sections
        rupture_sections_df = self._solution.model.rupture_sections
        rate_column = self._solution.model.rate_column_name()
        filtered_rupture_sections_df = rupture_sections_df.join(
            ruptures_df.set_index("Rupture Index"), on='rupture', how='inner'
        )[[rate_column, "rupture", "section"]]

        # join filtered rupture sections  and fault sections
        geom_flt_df = filtered_rupture_sections_df.join(filtered_fault_sections_df, 'section', how='inner')
        result = set(geom_flt_df["rupture"].tolist())
        return self.new_chainable_set(result, self._solution, self._drop_zero_rates, join_prior=join_prior)

__rupture_sections: Optional[pd.DataFrame] = None instance-attribute

__init__(solution: InversionSolution, drop_zero_rates: bool = True)

Instantiate a new filter.

Parameters:

Name Type Description Default
solution InversionSolution

The solution instance to filter on.

 required
drop_zero_rates bool

Exclude ruptures with rupture_rate == 0 (default=True).

 True
Source code in solvis/filter/rupture_id_filter.py 
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
def __init__(
    self,
    solution: 'InversionSolution',
    drop_zero_rates: bool = True,
):
    """Instantiate a new filter.

    Args:
        solution: The solution instance to filter on.
        drop_zero_rates: Exclude ruptures with rupture_rate == 0 (default=True).
    """
    self._solution = solution
    self._drop_zero_rates = drop_zero_rates
    self._filter_subsection_ids = FilterSubsectionIds(solution)
    self._filter_parent_fault_ids = FilterParentFaultIds(solution)
    self.__rupture_sections: Optional['pd.DataFrame'] = None

all() -> ChainableSetBase

Convenience method returning ids for all solution ruptures.

NB the usual join_prior argument is not implemented as it doesn't seem useful here.

Returns:

Type Description
ChainableSetBase

A chainable set of all the rupture_ids.
Source code in solvis/filter/rupture_id_filter.py 
139
140
141
142
143
144
145
146
147
148
def all(self) -> ChainableSetBase:
    """Convenience method returning ids for all solution ruptures.

    NB the usual `join_prior` argument is not implemented as it doesn't seem useful here.

    Returns:
        A chainable set of all the rupture_ids.
    """
    result = set(self._solution.solution_file.ruptures['Rupture Index'].to_list())
    return self.new_chainable_set(result, self._solution)

for_magnitude(min_mag: Optional[float] = None, max_mag: Optional[float] = None, join_prior: Union[SetOperationEnum, str] = 'intersection') -> ChainableSetBase

Find ruptures that occur within given magnitude bounds.

Parameters:

Name Type Description Default
min_mag Optional[float]

The minumum rupture magnitude bound.

 None
max_mag Optional[float]

The maximum rupture magnitude bound.

 None
join_prior Union[SetOperationEnum, str]

How to join this result with the prior chain (if any) (default = 'intersection').

 'intersection'

Returns:

Type Description
ChainableSetBase

A chainable set of rupture_ids matching the filter arguments.
Source code in solvis/filter/rupture_id_filter.py 
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
def for_magnitude(
    self,
    min_mag: Optional[float] = None,
    max_mag: Optional[float] = None,
    join_prior: Union[SetOperationEnum, str] = 'intersection',
) -> ChainableSetBase:
    """Find ruptures that occur within given magnitude bounds.

    Args:
        min_mag: The minumum rupture magnitude bound.
        max_mag: The maximum rupture magnitude bound.
        join_prior: How to join this result with the prior chain (if any) (default = 'intersection').

    Returns:
        A chainable set of rupture_ids matching the filter arguments.
    """
    index = "Rupture Index"
    df0 = self._ruptures_with_or_without_rupture_rates(drop_zero_rates=self._drop_zero_rates)

    df0 = df0 if not max_mag else df0[df0.Magnitude <= max_mag]
    df0 = df0 if not min_mag else df0[df0.Magnitude > min_mag]
    result = set(df0[index].tolist())
    return self.new_chainable_set(result, self._solution, self._drop_zero_rates, join_prior=join_prior)

for_named_fault_names(named_fault_names: Iterable[str], join_type: Union[SetOperationEnum, str] = 'union', join_prior: Union[SetOperationEnum, str] = 'intersection') -> ChainableSetBase

Filter rupture ids based on named fault names.

This method filters the rupture ids for given named fault names. It supports both intersection and union operations to combine results from multiple named faults. The join_type parameter determines how the sets of rupture ids are combined, while the join_prior parameter specifies how these sets are combined with any existing filters.

Parameters:

Name Type Description Default
named_fault_names Iterable[str]

A collection of named fault names.

 required
join_type Union[SetOperationEnum, str]

The type of set operation to use for combining results from multiple named faults. It can be either 'union' or 'intersection'.

 'union'
join_prior Union[SetOperationEnum, str]

The type of set operation to use when combining the filtered rupture ids with any existing filters. It can be either 'intersection' or 'difference'.

 'intersection'

Returns:

Name Type Description
ChainableSetBase ChainableSetBase

A chainable set containing the filtered rupture ids.

Raises:

Type Description
ValueError

If an unsupported join_type is provided.
Source code in solvis/filter/rupture_id_filter.py 
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
def for_named_fault_names(
    self,
    named_fault_names: Iterable[str],
    join_type: Union[SetOperationEnum, str] = 'union',
    join_prior: Union[SetOperationEnum, str] = 'intersection',
) -> ChainableSetBase:
    """Filter rupture ids based on named fault names.

    This method filters the rupture ids for given named fault names. It supports
    both intersection and union operations to combine results from multiple named
    faults. The `join_type` parameter determines how the sets of rupture ids are
    combined, while the `join_prior` parameter specifies how these sets are combined
    with any existing filters.

    Args:
        named_fault_names (Iterable[str]): A collection of named fault names.
        join_type (Union[SetOperationEnum, str], optional): The type of set operation to use for
            combining results from multiple named faults. It can be either 'union' or 'intersection'.
        join_prior (Union[SetOperationEnum, str], optional): The type of set operation to use when
            combining the filtered rupture ids with any existing filters. It can be either 'intersection'
            or 'difference'.

    Returns:
        ChainableSetBase: A chainable set containing the filtered rupture ids.

    Raises:
        ValueError: If an unsupported join_type is provided.
    """
    if isinstance(join_type, str):
        join_type = SetOperationEnum[join_type.upper()]

    rupture_id_sets: List[Set[int]] = []
    for named_fault_name in named_fault_names:
        parent_fault_ids = named_fault.named_fault_table().loc[named_fault_name].parent_fault_ids
        rupture_ids = self.for_parent_fault_ids(parent_fault_ids, join_prior=join_prior).chained_set
        rupture_id_sets.append(rupture_ids)

    if join_type == SetOperationEnum.INTERSECTION:
        rupture_ids = set.intersection(*rupture_id_sets)
    elif join_type == SetOperationEnum.UNION:
        rupture_ids = set.union(*rupture_id_sets)
    else:
        raise ValueError("Only INTERSECTION and UNION operations are supported for option 'join_type'")
    return self.new_chainable_set(rupture_ids, self._solution, self._drop_zero_rates, join_prior=join_prior)

for_parent_fault_id(parent_fault_id: int, join_prior: Union[SetOperationEnum, str] = 'intersection') -> ChainableSetBase

Find ruptures that occur on the given parent fault id.

Parameters:

Name Type Description Default
parent_fault_id int

The parent fault id to filter by.

 required
join_prior Union[SetOperationEnum, str]

How to join this result with the prior chain (if any) (default = 'intersection').

 'intersection'

Returns:

Type Description
ChainableSetBase

A chainable set of rupture_ids matching the filter.
Source code in solvis/filter/rupture_id_filter.py 
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
def for_parent_fault_id(
    self,
    parent_fault_id: int,
    join_prior: Union[SetOperationEnum, str] = 'intersection',
) -> ChainableSetBase:
    """
    Find ruptures that occur on the given parent fault id.

    Args:
        parent_fault_id: The parent fault id to filter by.
        join_prior: How to join this result with the prior chain (if any) (default = 'intersection').

    Returns:
        A chainable set of rupture_ids matching the filter.
    """
    subsection_ids = self._filter_subsection_ids.for_parent_fault_ids([parent_fault_id])
    rupture_ids = self._get_rupture_ids_for_subsection_ids(subsection_ids)
    return self.new_chainable_set(rupture_ids, self._solution, self._drop_zero_rates, join_prior=join_prior)

for_parent_fault_ids(parent_fault_ids: Iterable[int], join_type: Union[SetOperationEnum, str] = 'union', join_prior: Union[SetOperationEnum, str] = 'intersection') -> ChainableSetBase

Find ruptures that occur on the given parent_fault ids.

Parameters:

Name Type Description Default
parent_fault_ids Iterable[int]

A list of one or more parent_fault ids.

 required
join_type Union[SetOperationEnum, str]

The type of set operation to perform when combining results from multiple parent_fault_ids. Options are 'union' and 'intersection'.

 'union'
join_prior Union[SetOperationEnum, str]

How to join this result with the prior chain (if any) (default = 'intersection').

 'intersection'

Returns:

Type Description
ChainableSetBase

A chainable set of rupture_ids matching the filter.
Source code in solvis/filter/rupture_id_filter.py 
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
def for_parent_fault_ids(
    self,
    parent_fault_ids: Iterable[int],
    join_type: Union[SetOperationEnum, str] = 'union',
    join_prior: Union[SetOperationEnum, str] = 'intersection',
) -> ChainableSetBase:
    """Find ruptures that occur on the given parent_fault ids.

    Args:
        parent_fault_ids: A list of one or more `parent_fault` ids.
        join_type: The type of set operation to perform when combining results from multiple parent_fault_ids.
            Options are 'union' and 'intersection'.
        join_prior: How to join this result with the prior chain (if any) (default = 'intersection').

    Returns:
        A chainable set of rupture_ids matching the filter.
    """
    if isinstance(join_type, str):
        join_type = SetOperationEnum[join_type.upper()]

    rupture_id_sets: List[Set[int]] = []
    for fault_id in parent_fault_ids:
        rupture_id_sets.append(self.for_parent_fault_id(fault_id, join_prior=join_prior).chained_set)

    if join_type == SetOperationEnum.INTERSECTION:
        rupture_ids = set.intersection(*rupture_id_sets)
    elif join_type == SetOperationEnum.UNION:
        rupture_ids = set.union(*rupture_id_sets)
    else:
        raise ValueError("Only INTERSECTION and UNION operations are supported for option 'join_type'")
    return self.new_chainable_set(rupture_ids, self._solution, self._drop_zero_rates, join_prior=join_prior)

for_parent_fault_names(parent_fault_names: Iterable[str], join_type: Union[SetOperationEnum, str] = 'union', join_prior: Union[SetOperationEnum, str] = 'intersection') -> ChainableSetBase

Filter rupture ids based on parent fault names.

This method filters the rupture ids for given parent fault names. It supports both intersection and union operations to combine results from multiple parent faults. The join_type parameter determines how the sets of rupture ids are combined, while the join_prior parameter specifies how these sets are combined with any existing filters.

Parameters:

Name Type Description Default
parent_fault_names Iterable[str]

A collection of parent fault names.

 required
join_type Union[SetOperationEnum, str]

The type of set operation to use for combining results from multiple parent faults. It can be either 'union' or 'intersection'.

 'union'
join_prior Union[SetOperationEnum, str]

The type of set operation to use when combining the filtered rupture ids with any existing filters. It can be either 'intersection' or 'difference'.

 'intersection'

Returns:

Name Type Description
ChainableSetBase ChainableSetBase

A chainable set containing the filtered rupture ids.

Raises:

Type Description
ValueError

If an unsupported join_type is provided.
Source code in solvis/filter/rupture_id_filter.py 
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
def for_parent_fault_names(
    self,
    parent_fault_names: Iterable[str],
    join_type: Union[SetOperationEnum, str] = 'union',
    join_prior: Union[SetOperationEnum, str] = 'intersection',
) -> ChainableSetBase:
    """Filter rupture ids based on parent fault names.

    This method filters the rupture ids for given parent fault names. It supports
    both intersection and union operations to combine results from multiple parent
    faults. The `join_type` parameter determines how the sets of rupture ids are
    combined, while the `join_prior` parameter specifies how these sets are combined
    with any existing filters.

    Args:
        parent_fault_names (Iterable[str]): A collection of parent fault names.
        join_type (Union[SetOperationEnum, str], optional): The type of set operation to use for
            combining results from multiple parent faults. It can be either 'union' or 'intersection'.
        join_prior (Union[SetOperationEnum, str], optional): The type of set operation to use when
            combining the filtered rupture ids with any existing filters. It can be either 'intersection'
            or 'difference'.

    Returns:
        ChainableSetBase: A chainable set containing the filtered rupture ids.

    Raises:
        ValueError: If an unsupported join_type is provided.
    """
    if isinstance(join_type, str):
        join_type = SetOperationEnum[join_type.upper()]

    parent_fault_ids = self._filter_parent_fault_ids.for_parent_fault_names(parent_fault_names)
    return self.for_parent_fault_ids(parent_fault_ids=parent_fault_ids, join_type=join_type, join_prior=join_prior)

for_polygon(polygon: shapely.geometry.Polygon, join_prior: Union[SetOperationEnum, str] = 'intersection') -> ChainableSetBase

Find ruptures that involve a single polygon area.

Parameters:

Name Type Description Default
polygon shapely.geometry.Polygon

The polygon defining the area of intersection.

 required
join_prior Union[SetOperationEnum, str]

How to join this result with the prior chain (if any) (default = 'intersection').

 'intersection'

Returns:

Type Description
ChainableSetBase

A chainable set of rupture_ids matching the filter arguments.
Source code in solvis/filter/rupture_id_filter.py 
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
def for_polygon(
    self,
    polygon: shapely.geometry.Polygon,
    join_prior: Union[SetOperationEnum, str] = 'intersection',
) -> ChainableSetBase:
    """Find ruptures that involve a single polygon area.

    Args:
        polygon: The polygon defining the area of intersection.
        join_prior: How to join this result with the prior chain (if any) (default = 'intersection').

    Returns:
        A chainable set of rupture_ids matching the filter arguments.
    """
    fault_sections_df: pd.DataFrame = gpd.GeoDataFrame(self._solution.solution_file.fault_sections)

    # filter fault sections using the polygon geometry
    filtered_fault_sections_df = fault_sections_df[fault_sections_df['geometry'].intersects(polygon)]

    # filter ruptures by drop_zero_rates
    ruptures_df = self._ruptures_with_or_without_rupture_rates(drop_zero_rates=self._drop_zero_rates)

    # join filtered ruptures with rupture_sections
    rupture_sections_df = self._solution.model.rupture_sections
    rate_column = self._solution.model.rate_column_name()
    filtered_rupture_sections_df = rupture_sections_df.join(
        ruptures_df.set_index("Rupture Index"), on='rupture', how='inner'
    )[[rate_column, "rupture", "section"]]

    # join filtered rupture sections  and fault sections
    geom_flt_df = filtered_rupture_sections_df.join(filtered_fault_sections_df, 'section', how='inner')
    result = set(geom_flt_df["rupture"].tolist())
    return self.new_chainable_set(result, self._solution, self._drop_zero_rates, join_prior=join_prior)

for_polygons(polygons: Iterable[shapely.geometry.Polygon], join_type: Union[SetOperationEnum, str] = SetOperationEnum.UNION, join_prior: Union[SetOperationEnum, str] = 'intersection') -> ChainableSetBase

Find ruptures involving several polygon areas.

Each polygon will return a set of matching rupture ids, so the user may choose to override the default set operation (UNION) between these using the `join_type' argument.

This method

Parameters:

Name Type Description Default
polygons Iterable[shapely.geometry.Polygon]

Polygons defining the areas of interest.

 required
join_type Union[SetOperationEnum, str]

How to join the polygon results (default = 'union').

 SetOperationEnum.UNION
join_prior Union[SetOperationEnum, str]

How to join this result with the prior chain (if any) (default = 'intersection').

 'intersection'

Returns:

Type Description
ChainableSetBase

A chainable set of rupture_ids matching the filter arguments.
Source code in solvis/filter/rupture_id_filter.py 
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
def for_polygons(
    self,
    polygons: Iterable[shapely.geometry.Polygon],
    join_type: Union[SetOperationEnum, str] = SetOperationEnum.UNION,
    join_prior: Union[SetOperationEnum, str] = 'intersection',
) -> ChainableSetBase:
    """Find ruptures involving several polygon areas.

    Each polygon will return a set of matching rupture ids, so the user may choose to override the
    default set operation (UNION) between these using the `join_type' argument.

    This method

    Args:
        polygons: Polygons defining the areas of interest.
        join_type: How to join the polygon results (default = 'union').
        join_prior: How to join this result with the prior chain (if any) (default = 'intersection').

    Returns:
        A chainable set of rupture_ids matching the filter arguments.
    """
    if isinstance(join_type, str):
        try:
            join_type = SetOperationEnum[join_type.upper()]
        except KeyError:
            raise ValueError(f'Unsupported set operation `{join_type}` for `join_type` argument.')

    rupture_id_sets: List[Set[int]] = []
    for polygon in polygons:
        rupture_id_sets.append(self.for_polygon(polygon, join_prior=join_prior).chained_set)

    if join_type == SetOperationEnum.INTERSECTION:
        rupture_ids = set.intersection(*rupture_id_sets)
    elif join_type == SetOperationEnum.UNION:
        rupture_ids = set.union(*rupture_id_sets)
    elif join_type == SetOperationEnum.DIFFERENCE:
        rupture_ids = set.difference(*rupture_id_sets)
    else:
        raise ValueError(
            "Only INTERSECTION, UNION & DIFFERENCE operations are supported for `join_type`"
        )  # pragma: no cover
    return self.new_chainable_set(rupture_ids, self._solution, self._drop_zero_rates, join_prior=join_prior)

for_rupture_rate(min_rate: Optional[float] = None, max_rate: Optional[float] = None, join_prior: Union[SetOperationEnum, str] = 'intersection') -> ChainableSetBase

Find ruptures that occur within given rates bounds.

Parameters:

Name Type Description Default
min_rate Optional[float]

The minumum rupture _rate bound.

 None
max_rate Optional[float]

The maximum rupture _rate bound.

 None
join_prior Union[SetOperationEnum, str]

How to join this result with the prior chain (if any) (default = 'intersection').

 'intersection'

Returns:

Type Description
ChainableSetBase

A chainable set of rupture_ids matching the filter arguments.
Source code in solvis/filter/rupture_id_filter.py 
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
def for_rupture_rate(
    self,
    min_rate: Optional[float] = None,
    max_rate: Optional[float] = None,
    join_prior: Union[SetOperationEnum, str] = 'intersection',
) -> ChainableSetBase:
    """Find ruptures that occur within given rates bounds.

    Args:
        min_rate: The minumum rupture _rate bound.
        max_rate: The maximum rupture _rate bound.
        join_prior: How to join this result with the prior chain (if any) (default = 'intersection').


    Returns:
        A chainable set of rupture_ids matching the filter arguments.
    """
    index = "Rupture Index"
    df0 = self._ruptures_with_or_without_rupture_rates(drop_zero_rates=self._drop_zero_rates)

    rate_column = self._solution.model.rate_column_name()

    # rate col is different for InversionSolution
    df0 = df0 if not max_rate else df0[df0[rate_column] <= max_rate]
    df0 = df0 if not min_rate else df0[df0[rate_column] > min_rate]
    result = set(df0[index].tolist())
    return self.new_chainable_set(result, self._solution, self._drop_zero_rates, join_prior=join_prior)

for_subsection_ids(fault_section_ids: Iterable[int], join_type: Union[SetOperationEnum, str] = 'union', join_prior: Union[SetOperationEnum, str] = 'intersection') -> ChainableSetBase

Find ruptures that occur on any of the given fault_section_ids.

Return the Union of all rupture ids involvcedin te fault_section_ids.

Parameters:

Name Type Description Default
fault_section_ids Iterable[int]

A list of one or more fault_section ids.

 required
join_type Union[SetOperationEnum, str]

The type of set operation to perform when combining results from multiple fault_section_ids. Options are 'union' and 'intersection'.

 'union'
join_prior Union[SetOperationEnum, str]

How to join this result with the prior chain (if any) (default = 'intersection').

 'intersection'

Returns:

Type Description
ChainableSetBase

A chainable set of rupture_ids matching the filter.
Source code in solvis/filter/rupture_id_filter.py 
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
def for_subsection_ids(
    self,
    fault_section_ids: Iterable[int],
    join_type: Union[SetOperationEnum, str] = 'union',
    join_prior: Union[SetOperationEnum, str] = 'intersection',
) -> ChainableSetBase:
    """Find ruptures that occur on any of the given fault_section_ids.

    Return the Union of all rupture ids involvcedin te `fault_section_ids`.

    Args:
        fault_section_ids: A list of one or more fault_section ids.
        join_type: The type of set operation to perform when combining results from multiple fault_section_ids.
            Options are 'union' and 'intersection'.
        join_prior: How to join this result with the prior chain (if any) (default = 'intersection').

    Returns:
        A chainable set of rupture_ids matching the filter.
    """
    if isinstance(join_type, str):
        join_type = SetOperationEnum[join_type.upper()]

    rupture_id_sets: List[Set[int]] = []
    for fault_section_id in fault_section_ids:
        rupture_id_sets.append(self._get_rupture_ids_for_subsection_ids([fault_section_id]))
    if join_type == SetOperationEnum.INTERSECTION:
        rupture_ids = set.intersection(*rupture_id_sets)
    elif join_type == SetOperationEnum.UNION:
        rupture_ids = set.union(*rupture_id_sets)
    else:
        raise ValueError("Only INTERSECTION and UNION operations are supported for option 'join_type'")
    return self.new_chainable_set(rupture_ids, self._solution, self._drop_zero_rates, join_prior=join_prior)

tolist() -> List[int]

Returns the filtered rupture ids as a list of integers.

Returns:

Type Description
List[int]

A list of integers representing the filtered rupture ids.
Source code in solvis/filter/rupture_id_filter.py 
130
131
132
133
134
135
136
137
def tolist(self) -> List[int]:
    """
    Returns the filtered rupture ids as a list of integers.

    Returns:
        A list of integers representing the filtered rupture ids.
    """
    return list(self)