API reference

Auto-generated from the docstrings of the public API (everything exported from the top-level repyability package).

System models

RBD

RBD(edges, nodes=None, k=None, input_node=None, output_node=None, on_infeasible_rbd='raise')

Creates and returns a Reliability Block Diagram object.

The positional order mirrors the subclasses (:class:~repyability.NonRepairableRBD, :class:~repyability.RepairableRBD): the structure -- edges then nodes -- comes first, and k (k-out-of-n) is an optional modifier after it.

Parameters:

Name	Type	Description	Default
edges	Iterable[tuple[Hashable, Hashable]]	The collection of node edges, e.g. [(1, 2), (2, 3)] would correspond to the edges 1-2 and 2-3	required
nodes	Iterable	Extra node names to include beyond those implied by edges (e.g. isolated nodes), by default None. The RBD subclasses pass their set of component names here.	None
k	dict[Any, int]	A dictionary mapping nodes to k-out-of-n (koon) values; by default every node's koon value is 1.	None
on_infeasible_rbd	{'raise', warn, ignore}	Specifies what to do upon encountering a bad line (a line with too many fields). Allowed values are : - 'raise', raise an Exception when an infeasible RBD is detected. - 'warn', raise a warning when an infeasible RBD is detected, but return RBD anyway. - 'ignore', return the RBD without raising any warnings.	{'raise'

Raises:

Type	Description
ValueError	A node is not in the node list or edge list

Source code in repyability/rbd/rbd.py

def __init__(
    self,
    edges: Iterable[tuple[Hashable, Hashable]],
    nodes: Optional[Iterable] = None,
    k: Optional[dict[Any, int]] = None,
    input_node: Optional[Any] = None,
    output_node: Optional[Any] = None,
    on_infeasible_rbd: str = "raise",
):
    """Creates and returns a Reliability Block Diagram object.

    The positional order mirrors the subclasses
    (:class:`~repyability.NonRepairableRBD`,
    :class:`~repyability.RepairableRBD`): the structure -- ``edges`` then
    ``nodes`` -- comes first, and ``k`` (k-out-of-n) is an optional
    modifier after it.

    Parameters
    ----------
    edges : Iterable[tuple[Hashable, Hashable]]
        The collection of node edges, e.g. [(1, 2), (2, 3)] would
        correspond to the edges 1-2 and 2-3
    nodes : Iterable, optional
        Extra node names to include beyond those implied by ``edges`` (e.g.
        isolated nodes), by default None. The RBD subclasses pass their set
        of component names here.
    k : dict[Any, int], optional
        A dictionary mapping nodes to k-out-of-n (koon) values; by default
        every node's koon value is 1.
    on_infeasible_rbd : {{'raise', 'warn', 'ignore'}}, default 'raise'
        Specifies what to do upon encountering a bad line (a line with too
        many fields). Allowed values are :
            - 'raise', raise an Exception when an infeasible RBD is
            detected.
            - 'warn', raise a warning when an infeasible RBD is detected,
            but return RBD anyway.
            - 'ignore', return the RBD without raising any warnings.

    Raises
    ------
    ValueError
        A node is not in the node list or edge list
    """

    # Create RBD graph
    self.G = RBDGraph()
    self.G.add_edges_from(edges)
    if nodes is not None:
        self.G.add_nodes_from(nodes)

    if input_node is not None:
        if input_node not in self.G.nodes:
            raise ValueError("'input_node' not in RBD structure.")
        else:
            self.input_node = input_node

    if output_node is not None:
        if output_node not in self.G.nodes:
            raise ValueError("'output_node' not in RBD structure.")
        else:
            self.output_node = output_node

    # Set whether k for KooN nodes
    has_excess_koon_nodes = False
    excess_koon_nodes = []
    valid_rbd = True
    if k is not None:
        for node, k_val in k.items():
            if node in self.G.nodes:
                self.G.nodes[node]["k"] = k_val
            else:
                valid_rbd = False
                has_excess_koon_nodes = True
                excess_koon_nodes.append(node)

    # Finally, check valid RBD structure
    structure_check = self.G.is_valid_RBD_structure(
        nodes=nodes, input_node=input_node, output_node=output_node
    )

    structure_check["excess_koon_nodes"] = excess_koon_nodes
    structure_check["has_excess_koon_nodes"] = has_excess_koon_nodes
    if structure_check["is_valid"]:
        structure_check["is_valid"] = valid_rbd

    if has_excess_koon_nodes:
        structure_check["koon_errors"].append(
            "Check if you have repeated KooN nodes"
        )

    if not structure_check["is_valid"]:
        if on_infeasible_rbd == "warn":
            warnings.warn(
                "Strucutral Errors in RBD:\n"
                + pprint.pformat(structure_check),
                stacklevel=2,
            )
        elif on_infeasible_rbd == "raise":
            raise ValueError("RBD not correctly structured")
        elif on_infeasible_rbd == "ignore":
            pass
        else:
            raise ValueError(
                "'on_infeasible_rbd' must be one of"
                + " {'raise', 'warn', 'ignore'}"
            )

    self.structure_check = structure_check
    self.input_node = structure_check["input_node"]
    self.output_node = structure_check["output_node"]
    self.in_or_out = [self.input_node, self.output_node]
    self.nodes = [n for n in self.G.nodes if n not in self.in_or_out]
    self.structure_check["has_irrelevant_nodes"] = False
    self.structure_check["irrelevant_nodes"] = set()

    if (
        not structure_check["has_cycles"]
        and not structure_check["has_nodes_with_no_successor"]
    ):
        self.get_min_path_sets()
        irrelevant_nodes = self.find_irrelevant_components()
        if len(irrelevant_nodes) != 0:
            self.structure_check["has_irrelevant_nodes"] = True

        self.structure_check["irrelevant_nodes"] = irrelevant_nodes

get_all_path_sets

get_all_path_sets()

Gets all path sets from input_node to output_node

Really just wraps networkx.all_simple_paths(). This is an expensive operation, so be careful using for very large RBDs.

Returns:

Type	Description
Iterator[list[Hashable]]	The iterator of paths

Source code in repyability/rbd/rbd.py

def get_all_path_sets(self) -> Iterator[list[Hashable]]:
    """Gets all path sets from input_node to output_node

    Really just wraps networkx.all_simple_paths(). This is an expensive
    operation, so be careful using for very large RBDs.

    Returns
    -------
    Iterator[list[Hashable]]
        The iterator of paths
    """
    return nx.all_simple_paths(
        self.G, source=self.input_node, target=self.output_node
    )

get_min_path_sets

get_min_path_sets(include_in_out_nodes=True)

Gets the minimal path-sets of the RBD

Parameters:

Name	Type	Description	Default
include_in_out_nodes	bool	If false, excludes the input and output nodes in the return, by default True	True

Returns:

Type	Description
set[frozenset[Hashable]]	The set of minimal path-sets

Source code in repyability/rbd/rbd.py

def get_min_path_sets(
    self, include_in_out_nodes=True
) -> set[frozenset[Hashable]]:
    """Gets the minimal path-sets of the RBD

    Parameters
    ----------
    include_in_out_nodes : bool, optional
        If false, excludes the input and output nodes
        in the return, by default True

    Returns
    -------
    set[frozenset[Hashable]]
        The set of minimal path-sets
    """
    # Run min_path_sets() but convert all the inner sets to frozensets
    # and remove the input/output nodes if requested
    if hasattr(self, "_min_path_sets"):
        min_path_sets = self._min_path_sets
    else:
        min_path_sets = find_min_path_sets(
            rbd_graph=self.G,
            curr_node=self.output_node,
            solns={},
        )
        self._min_path_sets: list[set[Hashable]] = min_path_sets

    if min_path_sets == []:
        raise ValueError(
            "RBD has no paths through! Need to re-evaluate the KooN nodes."
        )

    ret_set = set()
    for min_path_set in min_path_sets:
        min_path_set = set(min_path_set)
        if not include_in_out_nodes:
            min_path_set.remove(self.input_node)
            min_path_set.remove(self.output_node)
        ret_set.add(frozenset(min_path_set))

    return ret_set

is_system_working

is_system_working(component_status, method)

Returns a boolean as to whether the system is working given the status of the components

Parameters:

Name	Type	Description	Default
component_status	dict[Any, bool]	Dictionary with all components where component_status[component] = True only if the component is working, and = False if not working.	required
method	str	Either "p" (path-set) or "c" (cut-set). Both return the same result; "p" is typically faster as it avoids deriving the cut sets.	required

Returns:

Type	Description
bool	True if the system is working, otherwise False.

Source code in repyability/rbd/rbd.py

def is_system_working(
    self, component_status: dict[Any, bool], method: str
) -> bool:
    """Returns a boolean as to whether the system is working given the
    status of the components

    Parameters
    ----------
    component_status : dict[Any, bool]
        Dictionary with all components where
        component_status[component] = True only if the component is
        working, and = False if not working.
    method : str
        Either "p" (path-set) or "c" (cut-set). Both return the same
        result; "p" is typically faster as it avoids deriving the cut
        sets.

    Returns
    -------
    bool
        True if the system is working, otherwise False.
    """
    # The system structure function is evaluated directly from the minimal
    # path/cut sets, which is plenty fast for the rate at which this is
    # called in the simulations (no Binary Decision Diagram required).
    #
    # - path-set ("p"): the system works iff at least one minimal path set
    #   has all of its components working.
    # - cut-set ("c"): the system works iff every minimal cut set has at
    #   least one of its components working.
    #
    # The path/cut sets (excluding the input/output nodes) are cached on
    # first use so repeated calls during a simulation are cheap.
    if method == "p":
        if not hasattr(self, "_eval_path_sets"):
            self._eval_path_sets = [
                tuple(path_set)
                for path_set in self.get_min_path_sets(
                    include_in_out_nodes=False
                )
            ]
        return any(
            all(component_status[component] for component in path_set)
            for path_set in self._eval_path_sets
        )
    elif method == "c":
        if not hasattr(self, "_eval_cut_sets"):
            self._eval_cut_sets = [
                tuple(cut_set)
                for cut_set in self.get_min_cut_sets(
                    include_in_out_nodes=False
                )
            ]
        return all(
            any(component_status[component] for component in cut_set)
            for cut_set in self._eval_cut_sets
        )
    else:
        raise ValueError("`method` must be either 'p' or 'c'")

get_min_cut_sets

get_min_cut_sets(include_in_out_nodes=False)

Returns the set of frozensets of minimal cut sets of the RBD. The outer set contains the frozenset of nodes. frozensets were used so the inner set elements could be hashable.

The minimal cut sets are the minimal transversals (hitting sets) of the minimal path sets, computed with Berge's algorithm. See minimal_cut_sets_from_path_sets() for details.

Source code in repyability/rbd/rbd.py

def get_min_cut_sets(
    self, include_in_out_nodes=False
) -> set[frozenset[Hashable]]:
    """
    Returns the set of frozensets of minimal cut sets of the RBD. The outer
    set contains the frozenset of nodes. frozensets were used so the inner
    set elements could be hashable.

    The minimal cut sets are the minimal transversals (hitting sets) of the
    minimal path sets, computed with Berge's algorithm. See
    minimal_cut_sets_from_path_sets() for details.
    """
    path_sets = self.get_min_path_sets(
        include_in_out_nodes=include_in_out_nodes
    )
    return minimal_cut_sets_from_path_sets(path_sets)

system_probability

system_probability(node_probabilities, method='p')

Returns the system probability/ies given the probability of each node.

Parameters:

Name	Type	Description	Default
node_probabilities	Dict	Dictionary containing the probabilities of the event for every node in the RBDGraph. Probability is to be either the reliability or the availability (or some other probability that I can't conceive).	required
method	str	Input either "c" or "p" for the function to use the cut set or path set methods respectively, by default "p". Both methods return the same (exact) result; the path set method is the default as it avoids deriving the cut sets.	'p'

Returns:

Type	Description
ndarray	Probability values for all events in nodes_probabilities

Raises:

Type	Description
ValueError	Probability arrays of differing lengths

Source code in repyability/rbd/rbd.py

def system_probability(
    self,
    node_probabilities: Dict,
    method: str = "p",
) -> np.ndarray:
    """Returns the system probability/ies given the probability of each
    node.

    Parameters
    ----------
    node_probabilities: Dict
        Dictionary containing the probabilities of the event for every node
        in the RBDGraph. Probability is to be either the reliability or the
        availability (or some other probability that I can't conceive).
    method: str, optional
        Input either "c" or "p" for the function to use the cut set or
        path set methods respectively, by default "p". Both methods
        return the same (exact) result; the path set method is the default
        as it avoids deriving the cut sets.

    Returns
    -------
    np.ndarray
        Probability values for all events in nodes_probabilities

    Raises
    ------
    ValueError
        Probability arrays of differing lengths
    """

    node_probabilities = copy(node_probabilities)
    lengths = np.array([], dtype=np.int64)
    for node in self.nodes:
        node_array = np.atleast_1d(node_probabilities[node])
        node_probabilities[node] = node_array
        lengths = np.append(lengths, len(node_array))

    if np.any(lengths[0] != lengths[1:]):
        raise ValueError("Probability arrays must be same length")
    else:
        # get shape of input array
        array_shape = lengths[0]

    if method == "p":
        # The system reliability is the probability that at least one
        # minimal path set has all of its components working.
        path_sets = self.get_min_path_sets(include_in_out_nodes=False)
        return probability_any_set_satisfied(
            path_sets, node_probabilities, array_shape
        )

    # method == "c": work with cut sets and node unreliabilities. The
    # system unreliability is the probability that at least one minimal
    # cut set has all of its components failed.
    cut_sets = self.get_min_cut_sets(include_in_out_nodes=False)
    node_unreliability = {k: 1 - v for k, v in node_probabilities.items()}
    system_unreliability = probability_any_set_satisfied(
        cut_sets, node_unreliability, array_shape
    )
    return 1 - system_unreliability

node_names

node_names()

Returns the list of (intermediate) node names of the RBD.

Source code in repyability/rbd/rbd.py

def node_names(self) -> list[Hashable]:
    """Returns the list of (intermediate) node names of the RBD."""
    return list(self.nodes)

structural_importance

structural_importance(working_nodes=None, broken_nodes=None)

Structural (probability-free) importance of every node.

The fraction of the states of the other nodes in which the node is pivotal -- the system works when the node works and fails when it fails, holding the others fixed. Equivalently it is the Birnbaum importance with every node reliability at 1/2, so it depends only on the RBD's structure and not on any failure model -- useful at design time, before any life data exists. It is a structural property, so it is the same for a NonRepairableRBD and a RepairableRBD with the same diagram.

working_nodes/broken_nodes condition on those nodes being forced up/down (validated as elsewhere).

Returns:

Type	Description
dict[Any, float]	Node name -> structural importance, in [0, 1].

Examples:

In a two-component parallel system each node is pivotal in half of the other node's states, independent of any failure model:

>>> from surpyval import FixedEventProbability
>>> from repyability import NonRepairableRBD
>>> rbd = NonRepairableRBD(
...     [("s", "a"), ("s", "b"), ("a", "t"), ("b", "t")],
...     {
...         "a": FixedEventProbability.from_params(0.1),
...         "b": FixedEventProbability.from_params(0.4),
...     },
... )
>>> si = rbd.structural_importance()
>>> {k: round(v, 4) for k, v in sorted(si.items())}
{'a': 0.5, 'b': 0.5}

Source code in repyability/rbd/rbd.py

def structural_importance(
    self,
    working_nodes: Optional[Iterable[Hashable]] = None,
    broken_nodes: Optional[Iterable[Hashable]] = None,
) -> dict[Any, float]:
    """Structural (probability-free) importance of every node.

    The fraction of the states of the *other* nodes in which the node is
    pivotal -- the system works when the node works and fails when it
    fails, holding the others fixed. Equivalently it is the Birnbaum
    importance with every node reliability at 1/2, so it depends only on
    the RBD's structure and not on any failure model -- useful at design
    time, before any life data exists. It is a structural property, so it
    is the same for a ``NonRepairableRBD`` and a ``RepairableRBD`` with the
    same diagram.

    ``working_nodes``/``broken_nodes`` condition on those nodes being
    forced up/down (validated as elsewhere).

    Returns
    -------
    dict[Any, float]
        Node name -> structural importance, in ``[0, 1]``.

    Examples
    --------
    In a two-component parallel system each node is pivotal in half of the
    other node's states, independent of any failure model:

    >>> from surpyval import FixedEventProbability
    >>> from repyability import NonRepairableRBD
    >>> rbd = NonRepairableRBD(
    ...     [("s", "a"), ("s", "b"), ("a", "t"), ("b", "t")],
    ...     {
    ...         "a": FixedEventProbability.from_params(0.1),
    ...         "b": FixedEventProbability.from_params(0.4),
    ...     },
    ... )
    >>> si = rbd.structural_importance()
    >>> {k: round(v, 4) for k, v in sorted(si.items())}
    {'a': 0.5, 'b': 0.5}
    """
    node_probabilities: dict[Any, ArrayLike] = {
        node: np.full(1, 0.5) for node in self.nodes
    }
    node_probabilities = self._probabilities_with_overrides(
        node_probabilities, working_nodes, broken_nodes
    )
    importance = self._birnbaum_importance(node_probabilities)
    return {
        node: float(np.asarray(value).reshape(-1)[0])
        for node, value in importance.items()
    }

to_dict

to_dict()

Serialise the RBD to a JSON-friendly dict (round-trips via :meth:from_dict). Node models are serialised structurally; fitted non-parametric models are not supported.

Source code in repyability/rbd/rbd.py

def to_dict(self) -> dict:
    """Serialise the RBD to a JSON-friendly dict (round-trips via
    :meth:`from_dict`). Node models are serialised structurally; fitted
    non-parametric models are not supported."""
    from repyability.rbd.serialisation import rbd_to_dict

    return rbd_to_dict(self)

to_json

to_json(**json_kwargs)

Serialise the RBD to a JSON string (kwargs pass through to json.dumps).

Source code in repyability/rbd/rbd.py

def to_json(self, **json_kwargs) -> str:
    """Serialise the RBD to a JSON string (kwargs pass through to
    ``json.dumps``)."""
    from repyability.rbd.serialisation import rbd_to_json

    return rbd_to_json(self, **json_kwargs)

from_dict classmethod

from_dict(d)

Reconstruct an RBD from :meth:to_dict output.

Called on a specific subclass, the document's type must match; called on RBD it dispatches to whichever type the document names.

Examples:

The structure round-trips through a plain dict, so reliability is preserved:

>>> import surpyval as surv
>>> from repyability import NonRepairableRBD
>>> rbd = NonRepairableRBD(
...     [("s", "c"), ("c", "t")],
...     {"c": surv.Weibull.from_params([100, 2])},
... )
>>> restored = NonRepairableRBD.from_dict(rbd.to_dict())
>>> round(restored.sf(50), 4) == round(rbd.sf(50), 4)
True

Source code in repyability/rbd/rbd.py

@classmethod
def from_dict(cls, d: dict) -> "RBD":
    """Reconstruct an RBD from :meth:`to_dict` output.

    Called on a specific subclass, the document's ``type`` must match;
    called on ``RBD`` it dispatches to whichever type the document names.

    Examples
    --------
    The structure round-trips through a plain dict, so reliability is
    preserved:

    >>> import surpyval as surv
    >>> from repyability import NonRepairableRBD
    >>> rbd = NonRepairableRBD(
    ...     [("s", "c"), ("c", "t")],
    ...     {"c": surv.Weibull.from_params([100, 2])},
    ... )
    >>> restored = NonRepairableRBD.from_dict(rbd.to_dict())
    >>> round(restored.sf(50), 4) == round(rbd.sf(50), 4)
    True
    """
    from repyability.rbd.serialisation import rbd_from_dict

    if cls is not RBD and cls.__name__ != d.get("type"):
        raise ValueError(
            f"{cls.__name__}.from_dict got a {d.get('type')!r} document; "
            f"use {d.get('type')}.from_dict or RBD.from_dict."
        )
    return rbd_from_dict(d)

from_json classmethod

from_json(s)

Reconstruct an RBD from a JSON string (see :meth:from_dict).

Source code in repyability/rbd/rbd.py

@classmethod
def from_json(cls, s: str) -> "RBD":
    """Reconstruct an RBD from a JSON string (see :meth:`from_dict`)."""
    import json

    return cls.from_dict(json.loads(s))

NonRepairableRBD

NonRepairableRBD(edges, reliabilities, k=None, input_node=None, output_node=None, on_infeasible_rbd='raise', ccf_groups=None)

Bases: RBD

Source code in repyability/rbd/non_repairable_rbd.py

def __init__(
    self,
    edges: Iterable[tuple[Hashable, Hashable]],
    reliabilities: dict[Any, Any],
    k: Optional[dict[Any, int]] = None,
    input_node: Optional[Any] = None,
    output_node: Optional[Any] = None,
    on_infeasible_rbd: str = "raise",
    ccf_groups: Optional[Iterable[CCFGroup]] = None,
):
    if on_infeasible_rbd not in ["raise", "warn", "ignore"]:
        raise ValueError(
            "'on_infeasible_rbd' must be one of"
            + " {'raise', 'warn', 'ignore'}"
        )
    # Capture the constructor inputs verbatim (before any mutation) so the
    # RBD can be faithfully serialised via to_dict()/to_json().
    edges = list(edges)
    ccf_groups = list(ccf_groups) if ccf_groups else []
    self._init_args = {
        "edges": [tuple(e) for e in edges],
        "reliabilities": dict(reliabilities),
        "k": dict(k) if k else None,
        "input_node": input_node,
        "output_node": output_node,
        "on_infeasible_rbd": on_infeasible_rbd,
        "ccf_groups": ccf_groups,
    }
    reliabilities = copy(reliabilities)
    for key, value in reliabilities.items():
        if key == value:
            raise ValueError(
                "Reliability dict cannot point to a node to itself"
            )
    # repeated checks if something was referenced from another node
    repeated = {
        k: v for k, v in reliabilities.items() if v in reliabilities.keys()
    }

    reliabilities = {
        k: v
        for k, v in reliabilities.items()
        if v not in reliabilities.keys()
    }

    if repeated == {}:
        super().__init__(
            edges,
            set(reliabilities.keys()),
            k,
            input_node,
            output_node,
            on_infeasible_rbd,
        )
        self.structure_check["has_repeated_node_in_cycle"] = False
    else:
        new_edges = []
        for start, stop in edges:
            if start in repeated:
                start = repeated[start]
            if stop in repeated:
                stop = repeated[stop]
            new_edges.append((start, stop))
        super().__init__(
            new_edges,
            set(reliabilities.keys()),
            k,
            input_node,
            output_node,
            on_infeasible_rbd,
        )
        self.structure_check["has_repeated_node_in_cycle"] = False
        if self.structure_check["has_cycles"]:
            # Need to find if cycles are due to repeated components.
            G = nx.DiGraph()
            G.add_edges_from(edges)
            cycles = {
                frozenset(cycle) for cycle in list(nx.simple_cycles(G))
            }
            non_repeated_node_cycles = copy(self.structure_check["cycles"])
            for cycle in self.structure_check["cycles"]:
                if cycle not in cycles:
                    non_repeated_node_cycles.remove(cycle)
                    self.structure_check["has_repeated_node_in_cycle"] = (
                        True
                    )
            if len(non_repeated_node_cycles) == 0:
                self.structure_check["has_cycles"] = False
            self.structure_check["cycles"] = non_repeated_node_cycles

    # Check for repeated cycles or non-repeated cycles
    if self.structure_check["has_unique_input_node"]:
        reliabilities[self.input_node] = PerfectReliability
    if self.structure_check["has_unique_output_node"]:
        reliabilities[self.output_node] = PerfectReliability

    # Check that all nodes in graph were in the reliabilities dict
    # Checking that all in the reliabilities dict are in the graph
    # is done in RBD initialisation since the RBD adds nodes from the
    # reliabilities dict and checks if they are connected.
    self.structure_check["is_missing_distributions"] = False
    self.structure_check["nodes_with_no_reliability_distribution"] = []
    for n in self.G.nodes:
        if n not in reliabilities:
            self.structure_check["is_valid"] = False
            self.structure_check["is_missing_distributions"] = True
            self.structure_check[
                "nodes_with_no_reliability_distribution"
            ].append(n)

    if not self.structure_check["is_valid"]:
        if on_infeasible_rbd == "warn":
            warnings.warn(
                "Strucutral Errors in RBD:\n"
                + pprint.pformat(self.structure_check),
                stacklevel=2,
            )
        elif on_infeasible_rbd == "raise":
            raise ValueError("RBD not correctly structured")
        elif on_infeasible_rbd == "ignore":
            pass

    self.reliabilities = reliabilities
    self.repeated = repeated
    self.ccf_groups = self._validate_ccf_groups(ccf_groups)

    fixed_flags = []
    for _, node in self.reliabilities.items():
        if isinstance(node, NonParametric):
            fixed_flags = [False]
            break
        elif isinstance(node, NonRepairableRBD):
            fixed_flags.append(node.is_fixed)
        elif node == PerfectReliability:
            continue
        elif node == PerfectUnreliability:
            continue
        else:
            # when node is a Parametric model
            if isinstance(node, (StandbyModel, LoadSharingModel)):
                fixed_flags = [False]
                break
            elif isinstance(node, RepeatedNode):
                fixed_flags.append(is_fixed_probability(node.model))
            else:
                fixed_flags.append(is_fixed_probability(node))

    self._fixed_probs: bool = all(fixed_flags)
    self.structure_check["all_distributions_fixed"] = self._fixed_probs

    # Record whether the system reliability can be solved analytically
    # (equivalently with the BDD), or whether it requires simulation
    # because one or more nodes are simulation-based (e.g. standby nodes).
    non_analytic_nodes = self.get_non_analytic_nodes()
    self.structure_check["is_analytically_solvable"] = (
        len(non_analytic_nodes) == 0
    )
    self.structure_check["non_analytic_nodes"] = non_analytic_nodes

is_fixed property

is_fixed

True if every node's reliability is a fixed (time-invariant) probability, so the system reliability does not vary with time.

is_time_varying property

is_time_varying

True if any node's reliability varies with time (the complement of :attr:is_fixed).

sf

sf(x=None, working_nodes=None, broken_nodes=None, method='p')

Returns the system reliability for time/s x.

Parameters:

Name	Type	Description	Default
x	ArrayLike	Time/s as a number or iterable	None
working_nodes	Collection[Hashable]	Marks these nodes as perfectly reliable, by default None	None
broken_nodes	Collection[Hashable]	Marks these nodes as perfectly unreliable, by default None	None
method	str	Input either "c" or "p" for the function to use the cut set or path set methods respectively. Both methods return the same (exact) result. By default the path set method ("p") is used, as it avoids deriving the cut sets.	'p'

Returns:

Type	Description
float or ndarray	The system reliability: a float for scalar x, an array for array x.

Raises:

Type	Description
ValueError	If a working/broken node is unknown, is the input/output node, is in both sets, or is a repeat of another node.

Examples:

Two components in parallel, each 90% reliable, so the system is 1 - 0.1 * 0.1 = 0.99 reliable:

>>> from surpyval import FixedEventProbability
>>> from repyability import NonRepairableRBD
>>> rbd = NonRepairableRBD(
...     [("s", "a"), ("s", "b"), ("a", "t"), ("b", "t")],
...     {
...         "a": FixedEventProbability.from_params(0.1),
...         "b": FixedEventProbability.from_params(0.1),
...     },
... )
>>> round(rbd.sf(), 4)
0.99

Conditioning on node "a" having failed leaves only "b":

>>> round(rbd.sf(broken_nodes=["a"]), 4)
0.9

Source code in repyability/rbd/non_repairable_rbd.py

@check_x
def sf(
    self,
    x: Optional[ArrayLike] = None,
    working_nodes: Optional[Collection[Hashable]] = None,
    broken_nodes: Optional[Collection[Hashable]] = None,
    method: str = "p",
) -> Union[float, np.ndarray]:
    """Returns the system reliability for time/s x.

    Parameters
    ----------
    x : ArrayLike
        Time/s as a number or iterable
    working_nodes : Collection[Hashable], optional
        Marks these nodes as perfectly reliable, by default None
    broken_nodes : Collection[Hashable], optional
        Marks these nodes as perfectly unreliable, by default None
    method: str, optional
        Input either "c" or "p" for the function to use the cut set or
        path set methods respectively. Both methods return the same
        (exact) result. By default the path set method ("p") is used, as
        it avoids deriving the cut sets.

    Returns
    -------
    float or np.ndarray
        The system reliability: a float for scalar ``x``, an array for
        array ``x``.

    Raises
    ------
    ValueError
        If a working/broken node is unknown, is the input/output node, is
        in both sets, or is a repeat of another node.

    Examples
    --------
    Two components in parallel, each 90% reliable, so the system is
    ``1 - 0.1 * 0.1 = 0.99`` reliable:

    >>> from surpyval import FixedEventProbability
    >>> from repyability import NonRepairableRBD
    >>> rbd = NonRepairableRBD(
    ...     [("s", "a"), ("s", "b"), ("a", "t"), ("b", "t")],
    ...     {
    ...         "a": FixedEventProbability.from_params(0.1),
    ...         "b": FixedEventProbability.from_params(0.1),
    ...     },
    ... )
    >>> round(rbd.sf(), 4)
    0.99

    Conditioning on node ``"a"`` having failed leaves only ``"b"``:

    >>> round(rbd.sf(broken_nodes=["a"]), 4)
    0.9
    """
    # Normalise the (optional) node overrides into sets for O(1) lookup.
    working_nodes = set() if working_nodes is None else set(working_nodes)
    broken_nodes = set() if broken_nodes is None else set(broken_nodes)
    self._validate_node_overrides(working_nodes, broken_nodes)

    node_probabilities = self._base_node_probabilities(
        x, working_nodes, broken_nodes
    )
    if self.ccf_groups:
        return self._ccf_system_probability(
            node_probabilities, working_nodes, broken_nodes, method
        )
    return self.system_probability(node_probabilities, method=method)

ff

ff(x=None, *args, **kwargs)

Returns the system unreliability for time/s x.

Parameters:

Name	Type	Description	Default
x	ArrayLike	Time/s as a number or iterable	None
*args		Any sf() arguments	()
**kwargs		Any sf() arguments	()

Returns:

Type	Description
float or ndarray	The system unreliability: a float for scalar x, an array for array x.

Examples:

>>> from surpyval import FixedEventProbability
>>> from repyability import NonRepairableRBD
>>> rbd = NonRepairableRBD(
...     [("s", "a"), ("s", "b"), ("a", "t"), ("b", "t")],
...     {
...         "a": FixedEventProbability.from_params(0.1),
...         "b": FixedEventProbability.from_params(0.1),
...     },
... )
>>> round(rbd.ff(), 4)
0.01

Source code in repyability/rbd/non_repairable_rbd.py

def ff(
    self, x: Optional[ArrayLike] = None, *args, **kwargs
) -> Union[float, np.ndarray]:
    """Returns the system unreliability for time/s x.

    Parameters
    ----------
    x : ArrayLike
        Time/s as a number or iterable
    *args, **kwargs :
        Any sf() arguments

    Returns
    -------
    float or np.ndarray
        The system unreliability: a float for scalar ``x``, an array for
        array ``x``.

    Examples
    --------
    >>> from surpyval import FixedEventProbability
    >>> from repyability import NonRepairableRBD
    >>> rbd = NonRepairableRBD(
    ...     [("s", "a"), ("s", "b"), ("a", "t"), ("b", "t")],
    ...     {
    ...         "a": FixedEventProbability.from_params(0.1),
    ...         "b": FixedEventProbability.from_params(0.1),
    ...     },
    ... )
    >>> round(rbd.ff(), 4)
    0.01
    """
    return 1 - self.sf(x, *args, **kwargs)

unreliability

unreliability(x=None, *args, **kwargs)

Returns the system unreliability for time/s x.

Parameters:

Name	Type	Description	Default
x	ArrayLike	Time/s as a number or iterable	None
*args		Any sf() arguments	()
**kwargs		Any sf() arguments	()

Returns:

Type	Description
ndarray	Unreliability values for all nodes at all times x

Source code in repyability/rbd/non_repairable_rbd.py

def unreliability(self, x: Optional[ArrayLike] = None, *args, **kwargs):
    """Returns the system unreliability for time/s x.

    Parameters
    ----------
    x : ArrayLike
        Time/s as a number or iterable
    *args, **kwargs :
        Any sf() arguments

    Returns
    -------
    np.ndarray
        Unreliability values for all nodes at all times x
    """
    return 1 - self.sf(x, *args, **kwargs)

reliability

reliability(x=None, *args, **kwargs)

Returns the system reliability for time/s x.

Parameters:

Name	Type	Description	Default
x	ArrayLike	Time/s as a number or iterable	None
*args		Any sf() arguments	()
**kwargs		Any sf() arguments	()

Returns:

Type	Description
ndarray	Reliability values for all nodes at all times x

Source code in repyability/rbd/non_repairable_rbd.py

def reliability(self, x: Optional[ArrayLike] = None, *args, **kwargs):
    """Returns the system reliability for time/s x.

    Parameters
    ----------
    x : ArrayLike
        Time/s as a number or iterable
    *args, **kwargs :
        Any sf() arguments

    Returns
    -------
    np.ndarray
        Reliability values for all nodes at all times x
    """
    return self.sf(x, *args, **kwargs)

cs

cs(x, X, *args, **kwargs)

Returns the conditional survival of the system.

That is, the probability the system survives a further x given it has already survived to X: R(x | X) = sf(X + x) / sf(X).

Parameters:

Name	Type	Description	Default
x	ArrayLike	The further duration/s at which conditional survival is evaluated.	required
X	ArrayLike	The age/s the system is known to have survived to.	required
*args		Any sf() arguments (e.g. working_nodes, broken_nodes, method).	()
**kwargs		Any sf() arguments (e.g. working_nodes, broken_nodes, method).	()

Returns:

Type	Description
ndarray	The conditional survival probability/ies.

Source code in repyability/rbd/non_repairable_rbd.py

def cs(self, x: ArrayLike, X: ArrayLike, *args, **kwargs) -> np.ndarray:
    """Returns the conditional survival of the system.

    That is, the probability the system survives a *further* ``x`` given it
    has already survived to ``X``: ``R(x | X) = sf(X + x) / sf(X)``.

    Parameters
    ----------
    x : ArrayLike
        The further duration/s at which conditional survival is evaluated.
    X : ArrayLike
        The age/s the system is known to have survived to.
    *args, **kwargs :
        Any sf() arguments (e.g. working_nodes, broken_nodes, method).

    Returns
    -------
    np.ndarray
        The conditional survival probability/ies.
    """
    return conditional_survival(self, x, X, *args, **kwargs)

Hf

Hf(x=None, **kwargs)

Returns the system cumulative hazard function H(x) = -ln R(x).

This is exact given the (exact) system reliability R(x); it is +inf wherever the system reliability has reached zero.

Parameters:

Name	Type	Description	Default
x	ArrayLike	Time/s as a number or iterable.	None
**kwargs		Any sf() arguments (e.g. working_nodes, broken_nodes, method).	{}

Source code in repyability/rbd/non_repairable_rbd.py

@check_x
def Hf(self, x: Optional[ArrayLike] = None, **kwargs) -> np.ndarray:
    """Returns the system cumulative hazard function H(x) = -ln R(x).

    This is exact given the (exact) system reliability R(x); it is +inf
    wherever the system reliability has reached zero.

    Parameters
    ----------
    x : ArrayLike
        Time/s as a number or iterable.
    **kwargs :
        Any sf() arguments (e.g. working_nodes, broken_nodes, method).
    """
    sf = np.asarray(self.sf(x, **kwargs), dtype=float)
    with np.errstate(divide="ignore"):
        return -np.log(sf)

df

df(x=None, dx=1e-06, **kwargs)

Returns the system failure density f(x) = -dR/dx.

Computed by central finite differences of the system reliability (which is composed of the nodes' reliabilities, so it is smooth in x); dx sets the relative step size. The step never crosses into negative time.

Parameters:

Name	Type	Description	Default
x	ArrayLike	Time/s as a number or iterable.	None
dx	float	Relative finite-difference step, by default 1e-6.	1e-06
**kwargs		Any sf() arguments (e.g. working_nodes, broken_nodes, method).	{}

Source code in repyability/rbd/non_repairable_rbd.py

@check_x
def df(
    self, x: Optional[ArrayLike] = None, dx: float = 1e-6, **kwargs
) -> np.ndarray:
    """Returns the system failure density f(x) = -dR/dx.

    Computed by central finite differences of the system reliability
    (which is composed of the nodes' reliabilities, so it is smooth in x);
    ``dx`` sets the relative step size. The step never crosses into
    negative time.

    Parameters
    ----------
    x : ArrayLike
        Time/s as a number or iterable.
    dx : float, optional
        Relative finite-difference step, by default 1e-6.
    **kwargs :
        Any sf() arguments (e.g. working_nodes, broken_nodes, method).
    """
    x = np.atleast_1d(np.asarray(x, dtype=float))
    h = dx * np.maximum(np.abs(x), 1.0)
    x_hi = x + h
    x_lo = np.maximum(x - h, 0.0)
    density = (
        np.asarray(self.sf(x_lo, **kwargs), dtype=float)
        - np.asarray(self.sf(x_hi, **kwargs), dtype=float)
    ) / (x_hi - x_lo)
    return np.clip(density, 0.0, None)

hf

hf(x=None, dx=1e-06, **kwargs)

Returns the system hazard rate h(x) = f(x) / R(x).

Uses the (numerical) failure density df() over the (exact) system reliability. It is +inf wherever the system reliability has reached zero.

Parameters:

Name	Type	Description	Default
x	ArrayLike	Time/s as a number or iterable.	None
dx	float	Relative finite-difference step passed to df(), by default 1e-6.	1e-06
**kwargs		Any sf() arguments (e.g. working_nodes, broken_nodes, method).	{}

Source code in repyability/rbd/non_repairable_rbd.py

@check_x
def hf(
    self, x: Optional[ArrayLike] = None, dx: float = 1e-6, **kwargs
) -> np.ndarray:
    """Returns the system hazard rate h(x) = f(x) / R(x).

    Uses the (numerical) failure density df() over the (exact) system
    reliability. It is +inf wherever the system reliability has reached
    zero.

    Parameters
    ----------
    x : ArrayLike
        Time/s as a number or iterable.
    dx : float, optional
        Relative finite-difference step passed to df(), by default 1e-6.
    **kwargs :
        Any sf() arguments (e.g. working_nodes, broken_nodes, method).
    """
    sf = np.asarray(self.sf(x, **kwargs), dtype=float)
    density = self.df(x, dx=dx, **kwargs)
    return np.divide(
        density,
        sf,
        out=np.full_like(density, np.inf),
        where=sf > 0,
    )

node_sf

node_sf(x=None, *args, **kwargs)

Returns each node's reliability at time/s x (a dict keyed by node name). Floats for scalar x, arrays for array x.

Source code in repyability/rbd/non_repairable_rbd.py

@check_x
def node_sf(
    self, x: Optional[ArrayLike] = None, *args, **kwargs
) -> Dict[Any, Union[float, np.ndarray]]:
    """Returns each node's reliability at time/s x (a dict keyed by node
    name). Floats for scalar ``x``, arrays for array ``x``."""
    node_sf: Dict[Any, Union[float, np.ndarray]] = {}
    for node_name, node in self.reliabilities.items():
        node_sf[node_name] = node.sf(x)
    return node_sf

node_ff

node_ff(x=None, *args, **kwargs)

Returns each node's unreliability at time/s x (a dict keyed by node name). Floats for scalar x, arrays for array x.

Source code in repyability/rbd/non_repairable_rbd.py

@check_x
def node_ff(
    self, x: Optional[ArrayLike] = None, *args, **kwargs
) -> Dict[Any, Union[float, np.ndarray]]:
    """Returns each node's unreliability at time/s x (a dict keyed by node
    name). Floats for scalar ``x``, arrays for array ``x``."""
    node_ff: Dict[Any, Union[float, np.ndarray]] = {}
    for node_name, node in self.reliabilities.items():
        node_ff[node_name] = node.ff(x)

    return node_ff

get_non_analytic_nodes

get_non_analytic_nodes()

Returns the nodes that prevent an analytic / BDD solution.

Returns:

Type	Description
dict[Any, str]	A mapping of node name -> the offending model's type name for every node whose reliability requires Monte-Carlo simulation (e.g. a StandbyModel). Empty if the RBD is analytically solvable.

Source code in repyability/rbd/non_repairable_rbd.py

def get_non_analytic_nodes(self) -> dict[Any, str]:
    """Returns the nodes that prevent an analytic / BDD solution.

    Returns
    -------
    dict[Any, str]
        A mapping of node name -> the offending model's type name for
        every node whose reliability requires Monte-Carlo simulation (e.g.
        a StandbyModel). Empty if the RBD is analytically solvable.
    """
    non_analytic: dict[Any, str] = {}
    for node_name, model in self.reliabilities.items():
        if not self._node_is_analytic(model):
            non_analytic[node_name] = type(model).__name__
    return non_analytic

is_analytically_solvable

is_analytically_solvable()

Returns whether the system reliability can be solved analytically.

The analytic methods (the inclusion-exclusion in system_probability(), and equivalently a BDD evaluation) require every node to expose a reliability sf(t) that does not itself depend on Monte-Carlo simulation. This holds for parametric and non-parametric distributions, fixed-probability nodes, repeated nodes of such models, and nested RBDs that are themselves analytically solvable.

It does NOT hold when any node is a standby arrangement (StandbyModel or RepeatedStandbyNode): a standby node is sequence-dependent and its sf(t) is estimated by simulation, so while sf()/system_probability() will still return a value, that value is only as good as the underlying Monte-Carlo + Kaplan-Meier fit (a step function bounded by the sampled support) rather than a closed-form result. Such systems are better evaluated by simulation (e.g. random()/mean()).

Returns:

Type	Description
bool	True if the RBD can be solved analytically / with a BDD, False if it requires simulation. Use get_non_analytic_nodes() to see which nodes are responsible.

Source code in repyability/rbd/non_repairable_rbd.py

def is_analytically_solvable(self) -> bool:
    """Returns whether the system reliability can be solved analytically.

    The analytic methods (the inclusion-exclusion in system_probability(),
    and equivalently a BDD evaluation) require every node to expose a
    reliability sf(t) that does not itself depend on Monte-Carlo
    simulation. This holds for parametric and non-parametric distributions,
    fixed-probability nodes, repeated nodes of such models, and nested RBDs
    that are themselves analytically solvable.

    It does NOT hold when any node is a standby arrangement (StandbyModel
    or RepeatedStandbyNode): a standby node is sequence-dependent and its
    sf(t) is estimated by simulation, so while sf()/system_probability()
    will still return a value, that value is only as good as the underlying
    Monte-Carlo + Kaplan-Meier fit (a step function bounded by the sampled
    support) rather than a closed-form result. Such systems are better
    evaluated by simulation (e.g. random()/mean()).

    Returns
    -------
    bool
        True if the RBD can be solved analytically / with a BDD, False if
        it requires simulation. Use get_non_analytic_nodes() to see which
        nodes are responsible.
    """
    return len(self.get_non_analytic_nodes()) == 0

random

random(size, seed=None)

Monte-Carlo simulate size system failure times.

Parameters:

Name	Type	Description	Default
size	int	Number of system-lifetime samples to draw.	required
seed	int or None	If given, seeds numpy's global RNG for the duration of the draw so the result is reproducible (surpyval's .random uses the global RNG); the caller's RNG state is restored afterwards. By default None (non-reproducible).	None

Source code in repyability/rbd/non_repairable_rbd.py

def random(self, size, seed=None):
    """Monte-Carlo simulate ``size`` system failure times.

    Parameters
    ----------
    size : int
        Number of system-lifetime samples to draw.
    seed : int or None, optional
        If given, seeds numpy's global RNG for the duration of the draw so
        the result is reproducible (surpyval's ``.random`` uses the global
        RNG); the caller's RNG state is restored afterwards. By default
        None (non-reproducible).
    """
    out = np.zeros(size)
    with numpy_seed(seed):
        for i in range(size):
            event_queue: PriorityQueue = PriorityQueue()
            for node in self.G.nodes:
                # .random(1) returns a 1-element array; take the scalar so
                # the event time orders the PriorityQueue and assigns into
                # ``out`` (NumPy >= 2 rejects assigning a 1-element array
                # to a scalar).
                draw = np.asarray(self.reliabilities[node].random(1))
                time = float(draw.reshape(-1)[0])
                event_queue.put(NodeFailure(time, node))

            working_nodes = {k: True for k in self.G.nodes}
            system_working = True
            while system_working:
                failure = event_queue.get()
                time = failure.time
                working_nodes[failure.node] = False
                system_working = self.is_system_working(
                    working_nodes, method="p"
                )
            out[i] = time

    return out

mean

mean(mc_samples=100000, seed=None)

Returns the Mean Time To Failure of the RBD This is necessary for recursive calls which will only use the mean

Source code in repyability/rbd/non_repairable_rbd.py

def mean(self, mc_samples: int = 100_000, seed=None):
    """Returns the Mean Time To Failure of the RBD
    This is necessary for recursive calls which will only use the `mean`
    """
    return self.random(mc_samples, seed=seed).mean().item()

mean_time_to_failure

mean_time_to_failure(mc_samples=100000, seed=None)

User friendly way to get MTTF

Source code in repyability/rbd/non_repairable_rbd.py

def mean_time_to_failure(self, mc_samples: int = 100_000, seed=None):
    """
    User friendly way to get MTTF
    """
    return self.mean(mc_samples, seed=seed)

mean_time_to_failure_interval

mean_time_to_failure_interval(mc_samples=100000, confidence=0.95, seed=None)

Returns the Monte-Carlo MTTF estimate with its sampling uncertainty.

The MTTF is the mean of mc_samples simulated system lifetimes; by the central limit theorem its sampling error is normal with standard error sample std / sqrt(mc_samples), from which the confidence interval is built.

Parameters:

Name	Type	Description	Default
mc_samples	int	Number of Monte-Carlo samples, by default 100_000.	100000
confidence	float	The confidence level, by default 0.95.	0.95
seed	int or None	Seed for reproducibility (see :meth:random).	None

Returns:

Type	Description
ConfidenceInterval	The estimate, bounds, standard error and sample count.

Source code in repyability/rbd/non_repairable_rbd.py

def mean_time_to_failure_interval(
    self,
    mc_samples: int = 100_000,
    confidence: float = 0.95,
    seed=None,
) -> ConfidenceInterval:
    """Returns the Monte-Carlo MTTF estimate with its sampling
    uncertainty.

    The MTTF is the mean of ``mc_samples`` simulated system lifetimes; by
    the central limit theorem its sampling error is normal with standard
    error ``sample std / sqrt(mc_samples)``, from which the confidence
    interval is built.

    Parameters
    ----------
    mc_samples : int, optional
        Number of Monte-Carlo samples, by default 100_000.
    confidence : float, optional
        The confidence level, by default 0.95.
    seed : int or None, optional
        Seed for reproducibility (see :meth:`random`).

    Returns
    -------
    ConfidenceInterval
        The estimate, bounds, standard error and sample count.
    """
    if not 0.0 < confidence < 1.0:
        raise ValueError("confidence must be between 0 and 1.")
    samples = self.random(mc_samples, seed=seed)
    estimate = float(samples.mean())
    standard_error = float(samples.std(ddof=1) / np.sqrt(len(samples)))
    z = float(norm.ppf(0.5 + confidence / 2.0))
    return ConfidenceInterval(
        estimate=estimate,
        lower=max(0.0, estimate - z * standard_error),
        upper=estimate + z * standard_error,
        confidence=confidence,
        standard_error=standard_error,
        n_samples=mc_samples,
    )

time_to_reliability

time_to_reliability(target, upper_bound=None, **kwargs)

Returns the time at which system reliability equals target.

Solves R(t) = target for t (the inverse of :meth:sf). System reliability is monotonically non-increasing in time, so the solution is unique.

Parameters:

Name	Type	Description	Default
target	float	The reliability level to solve for, in (0, 1).	required
upper_bound	float	An upper bound for the search; found automatically (by doubling) if None.	None
**kwargs		Any sf() arguments (e.g. working_nodes, broken_nodes, method).	{}

Returns:

Type	Description
float	The time at which R(t) == target.

Raises:

Type	Description
ValueError	If target is not in (0, 1), if the RBD is fixed-probability (reliability is constant in time), or if target exceeds the system reliability at t = 0 (so it is never reached).

Examples:

>>> import surpyval as surv
>>> from repyability import NonRepairableRBD
>>> rbd = NonRepairableRBD(
...     [("s", "c"), ("c", "t")],
...     {"c": surv.Weibull.from_params([100, 2])},
... )
>>> round(rbd.time_to_reliability(0.9), 2)
32.46

Source code in repyability/rbd/non_repairable_rbd.py

def time_to_reliability(
    self,
    target: float,
    upper_bound: Optional[float] = None,
    **kwargs,
) -> float:
    """Returns the time at which system reliability equals ``target``.

    Solves ``R(t) = target`` for ``t`` (the inverse of :meth:`sf`). System
    reliability is monotonically non-increasing in time, so the solution
    is unique.

    Parameters
    ----------
    target : float
        The reliability level to solve for, in (0, 1).
    upper_bound : float, optional
        An upper bound for the search; found automatically (by doubling)
        if None.
    **kwargs :
        Any sf() arguments (e.g. working_nodes, broken_nodes, method).

    Returns
    -------
    float
        The time at which ``R(t) == target``.

    Raises
    ------
    ValueError
        If ``target`` is not in (0, 1), if the RBD is fixed-probability
        (reliability is constant in time), or if ``target`` exceeds the
        system reliability at ``t = 0`` (so it is never reached).

    Examples
    --------
    >>> import surpyval as surv
    >>> from repyability import NonRepairableRBD
    >>> rbd = NonRepairableRBD(
    ...     [("s", "c"), ("c", "t")],
    ...     {"c": surv.Weibull.from_params([100, 2])},
    ... )
    >>> round(rbd.time_to_reliability(0.9), 2)
    32.46
    """
    return self._invert_reliability(
        lambda t: float(self.sf(t, **kwargs)), target, upper_bound
    )

bx_life

bx_life(x, **kwargs)

Returns the B\ :sub:X life: the time by which x percent of systems have failed, i.e. the time at which R(t) = 1 - x/100.

For example bx_life(10) is the B10 life (10% failed / 90% reliability).

Parameters:

Name	Type	Description	Default
x	float	The percentage failed, in (0, 100).	required
**kwargs		Any sf() arguments (e.g. working_nodes, broken_nodes, method).	{}

Examples:

The B10 life (10% failed) of a single Weibull component:

>>> import surpyval as surv
>>> from repyability import NonRepairableRBD
>>> rbd = NonRepairableRBD(
...     [("s", "c"), ("c", "t")],
...     {"c": surv.Weibull.from_params([100, 2])},
... )
>>> round(rbd.bx_life(10), 2)
32.46

Source code in repyability/rbd/non_repairable_rbd.py

def bx_life(self, x: float, **kwargs) -> float:
    """Returns the B\\ :sub:`X` life: the time by which ``x`` percent of
    systems have failed, i.e. the time at which ``R(t) = 1 - x/100``.

    For example ``bx_life(10)`` is the B10 life (10% failed / 90%
    reliability).

    Parameters
    ----------
    x : float
        The percentage failed, in (0, 100).
    **kwargs :
        Any sf() arguments (e.g. working_nodes, broken_nodes, method).

    Examples
    --------
    The B10 life (10% failed) of a single Weibull component:

    >>> import surpyval as surv
    >>> from repyability import NonRepairableRBD
    >>> rbd = NonRepairableRBD(
    ...     [("s", "c"), ("c", "t")],
    ...     {"c": surv.Weibull.from_params([100, 2])},
    ... )
    >>> round(rbd.bx_life(10), 2)
    32.46
    """
    if not 0.0 < x < 100.0:
        raise ValueError("x must be a percentage in (0, 100).")
    return self.time_to_reliability(1.0 - x / 100.0, **kwargs)

sf_given_state

sf_given_state(x=None, state=None, method='p')

System reliability a further x into the future, given each node's current state.

The condition-based ("digital twin") generalisation of :meth:sf: instead of assuming every component is new, each component conditions on its own current life X_i (streamed from sensors) and the conditioned per-node reliabilities are propagated exactly through the system::

R_i(x | X_i)     = R_i(X_i + x) / R_i(X_i)
R_sys(x | {X_i}) = system_probability({ R_i(x | X_i) })

Parameters:

Name	Type	Description	Default
x	ArrayLike	The further duration/s at which reliability is evaluated (measured from now, so x = 0 is the present).	None
state	dict[Hashable, NodeState]	The current state of each component. A node omitted from the mapping is treated as new (age 0), so an empty state reproduces :meth:sf.	None
method	str	"p" (path-set, default) or "c" (cut-set); both are exact.	'p'

Returns:

Type	Description
float or ndarray	System reliability given the state: a float for scalar x, an array for array x.

Notes

Only lifetime (time-varying) distributions age; a fixed-probability component conditioned on being alive contributes reliability 1 going forward (its per-demand uncertainty is resolved by observing it alive). Composite / dynamic nodes (standby, repeated, nested RBD) are not supported here in this release and raise if given a state.

Examples:

A component 40 hours into life is less reliable over the next 50 than a new one would be:

>>> import surpyval as surv
>>> from repyability import NonRepairableRBD, NodeState
>>> rbd = NonRepairableRBD(
...     [("s", "c"), ("c", "t")],
...     {"c": surv.Weibull.from_params([100, 2])},
... )
>>> round(rbd.sf_given_state(50, {"c": NodeState(age=40)}), 4)
0.522

Source code in repyability/rbd/non_repairable_rbd.py

@check_x
def sf_given_state(
    self,
    x: Optional[ArrayLike] = None,
    state: Optional[Dict[Hashable, NodeState]] = None,
    method: str = "p",
) -> Union[float, np.ndarray]:
    """System reliability a further ``x`` into the future, given each
    node's current state.

    The condition-based ("digital twin") generalisation of :meth:`sf`:
    instead of assuming every component is new, each component conditions
    on its own current life ``X_i`` (streamed from sensors) and the
    conditioned per-node reliabilities are propagated exactly through the
    system::

        R_i(x | X_i)     = R_i(X_i + x) / R_i(X_i)
        R_sys(x | {X_i}) = system_probability({ R_i(x | X_i) })

    Parameters
    ----------
    x : ArrayLike
        The further duration/s at which reliability is evaluated (measured
        from *now*, so ``x = 0`` is the present).
    state : dict[Hashable, NodeState]
        The current state of each component. A node omitted from the
        mapping is treated as new (age 0), so an empty state reproduces
        :meth:`sf`.
    method : str, optional
        "p" (path-set, default) or "c" (cut-set); both are exact.

    Returns
    -------
    float or np.ndarray
        System reliability given the state: a float for scalar ``x``, an
        array for array ``x``.

    Notes
    -----
    Only lifetime (time-varying) distributions age; a fixed-probability
    component conditioned on being alive contributes reliability 1 going
    forward (its per-demand uncertainty is resolved by observing it
    alive). Composite / dynamic nodes (standby, repeated, nested RBD) are
    not supported here in this release and raise if given a state.

    Examples
    --------
    A component 40 hours into life is less reliable over the next 50 than
    a new one would be:

    >>> import surpyval as surv
    >>> from repyability import NonRepairableRBD, NodeState
    >>> rbd = NonRepairableRBD(
    ...     [("s", "c"), ("c", "t")],
    ...     {"c": surv.Weibull.from_params([100, 2])},
    ... )
    >>> round(rbd.sf_given_state(50, {"c": NodeState(age=40)}), 4)
    0.522
    """
    if state is None:
        state = {}
    node_probabilities = self._state_node_probabilities(x, state)
    return self.system_probability(node_probabilities, method=method)

remaining_life

remaining_life(target, state=None, upper_bound=None)

Remaining useful life (RUL): the further time until system reliability falls to target, given each node's current state.

The condition-based analogue of :meth:time_to_reliability: it solves sf_given_state(t, state) == target for t. Because sf_given_state is measured from now, the result is the time remaining from the current state. remaining_life(1 - x/100, state) is the conditional B\ :sub:X life.

Parameters:

Name	Type	Description	Default
target	float	The system reliability level to solve for, in (0, 1).	required
state	dict[Hashable, NodeState]	The current state of each component (see :meth:sf_given_state).	None
upper_bound	float	An upper bound for the search; found automatically if None.	None

Returns:

Type	Description
float	The remaining time until R_sys(t | state) == target.

Examples:

>>> import surpyval as surv
>>> from repyability import NonRepairableRBD, NodeState
>>> rbd = NonRepairableRBD(
...     [("s", "c"), ("c", "t")],
...     {"c": surv.Weibull.from_params([100, 2])},
... )
>>> round(rbd.remaining_life(0.9, {"c": NodeState(age=40)}), 2)
11.51

Source code in repyability/rbd/non_repairable_rbd.py

def remaining_life(
    self,
    target: float,
    state: Optional[Dict[Hashable, NodeState]] = None,
    upper_bound: Optional[float] = None,
) -> float:
    """Remaining useful life (RUL): the further time until system
    reliability falls to ``target``, given each node's current state.

    The condition-based analogue of :meth:`time_to_reliability`: it solves
    ``sf_given_state(t, state) == target`` for ``t``. Because
    ``sf_given_state`` is measured from now, the result is the time
    *remaining* from the current state. ``remaining_life(1 - x/100,
    state)`` is the conditional B\\ :sub:`X` life.

    Parameters
    ----------
    target : float
        The system reliability level to solve for, in (0, 1).
    state : dict[Hashable, NodeState]
        The current state of each component (see :meth:`sf_given_state`).
    upper_bound : float, optional
        An upper bound for the search; found automatically if None.

    Returns
    -------
    float
        The remaining time until ``R_sys(t | state) == target``.

    Examples
    --------
    >>> import surpyval as surv
    >>> from repyability import NonRepairableRBD, NodeState
    >>> rbd = NonRepairableRBD(
    ...     [("s", "c"), ("c", "t")],
    ...     {"c": surv.Weibull.from_params([100, 2])},
    ... )
    >>> round(rbd.remaining_life(0.9, {"c": NodeState(age=40)}), 2)
    11.51
    """
    if state is None:
        state = {}
    return self._invert_reliability(
        lambda t: float(self.sf_given_state(t, state)),
        target,
        upper_bound,
    )

importances_given_state

importances_given_state(x=None, state=None)

Live, state-dependent node importance over a forward horizon x -- how each component's importance shifts once the current wear on every component is accounted for.

Evaluates the Birnbaum and criticality importance measures at the conditioned per-node reliabilities R_i(x | X_i) (see :meth:sf_given_state) rather than at the as-new reliabilities, so the rankings reflect the current state. Both use the same conventions as :meth:birnbaum_importance and :meth:criticality_importance: they measure how much the system reliability depends on each node now, not which node is most likely to have failed.

Parameters:

Name	Type	Description	Default
x	ArrayLike	The forward horizon/s over which importance is evaluated (from now).	None
state	dict[Hashable, NodeState]	The current state of each component (see :meth:sf_given_state).	None

Returns:

Type	Description
dict[str, dict[Any, float | ndarray]]	{"birnbaum": {node: value}, "criticality": {node: value}}. Values are floats for scalar x and arrays for array x.

Source code in repyability/rbd/non_repairable_rbd.py

def importances_given_state(
    self,
    x: Optional[ArrayLike] = None,
    state: Optional[Dict[Hashable, NodeState]] = None,
) -> Dict[str, Dict[Any, Union[float, np.ndarray]]]:
    """Live, state-dependent node importance over a forward horizon ``x``
    -- how each component's importance shifts once the current wear on
    every component is accounted for.

    Evaluates the Birnbaum and criticality importance measures at the
    conditioned per-node reliabilities ``R_i(x | X_i)`` (see
    :meth:`sf_given_state`) rather than at the as-new reliabilities, so the
    rankings reflect the current state. Both use the same conventions as
    :meth:`birnbaum_importance` and :meth:`criticality_importance`: they
    measure how much the system reliability depends on each node *now*, not
    which node is most likely to have failed.

    Parameters
    ----------
    x : ArrayLike
        The forward horizon/s over which importance is evaluated (from
        now).
    state : dict[Hashable, NodeState]
        The current state of each component (see :meth:`sf_given_state`).

    Returns
    -------
    dict[str, dict[Any, float | np.ndarray]]
        ``{"birnbaum": {node: value}, "criticality": {node: value}}``.
        Values are floats for scalar ``x`` and arrays for array ``x``.
    """
    if state is None:
        state = {}
    if x is None:
        if self.is_fixed:
            x = 1.0
        else:
            raise ValueError(
                "x is required: this RBD is time-varying (at least one "
                "node model's probability depends on time)."
            )
    scalar_in = np.ndim(x) == 0
    x_arr = np.atleast_1d(np.asarray(x, dtype=float))
    node_probabilities = self._state_node_probabilities(x_arr, state)
    birnbaum = super()._birnbaum_importance(node_probabilities)
    criticality = super()._criticality_importance(node_probabilities)

    def _squeeze(measure):
        return {
            node: (
                float(np.asarray(value).reshape(-1)[0])
                if scalar_in
                else np.asarray(value)
            )
            for node, value in measure.items()
        }

    return {
        "birnbaum": _squeeze(birnbaum),
        "criticality": _squeeze(criticality),
    }

node_mttf

node_mttf(mc_samples=100000, seed=None)

Returns each node's mean time to failure (a dict keyed by node name). Simulation-based node models use mc_samples Monte-Carlo draws; fixed-probability nodes have no time dimension and return 0.

Source code in repyability/rbd/non_repairable_rbd.py

def node_mttf(
    self, mc_samples: int = 100_000, seed=None
) -> dict[Any, float]:
    """Returns each node's mean time to failure (a dict keyed by node
    name). Simulation-based node models use ``mc_samples`` Monte-Carlo
    draws; fixed-probability nodes have no time dimension and return 0."""
    out: dict[Any, float] = {}
    with numpy_seed(seed):
        for node in self.nodes:
            model = self.reliabilities[node]
            if isinstance(
                model,
                (
                    StandbyModel,
                    LoadSharingModel,
                    NonRepairableRBD,
                    RepeatedNode,
                ),
            ):
                out[node] = float(np.atleast_1d(model.mean(mc_samples))[0])
            elif is_fixed_probability(model):
                out[node] = 0.0
            else:
                out[node] = float(np.atleast_1d(model.mean())[0])
    return out

birnbaum_importance

birnbaum_importance(x=None, working_nodes=None, broken_nodes=None)

Returns the Birnbaum measure of importance for all nodes.

Note: Birnbaum's measure of importance assumes all nodes are independent.

Parameters:

Name	Type	Description	Default
x	ArrayLike	Time/s as a number or iterable	None
working_nodes	Collection[Hashable]	Condition on these nodes being perfectly reliable, by default None	None
broken_nodes	Collection[Hashable]	Condition on these nodes having failed, by default None	None

Returns:

Type	Description
dict[Any, float | ndarray]	Dictionary with node names as keys and Birnbaum importances as values (floats for scalar x, arrays for array x)

Examples:

In a two-component parallel system each node's Birnbaum importance is the probability the other node has failed:

>>> from surpyval import FixedEventProbability
>>> from repyability import NonRepairableRBD
>>> rbd = NonRepairableRBD(
...     [("s", "a"), ("s", "b"), ("a", "t"), ("b", "t")],
...     {
...         "a": FixedEventProbability.from_params(0.1),
...         "b": FixedEventProbability.from_params(0.1),
...     },
... )
>>> bi = rbd.birnbaum_importance()
>>> {k: round(v, 4) for k, v in sorted(bi.items())}
{'a': 0.1, 'b': 0.1}

Source code in repyability/rbd/non_repairable_rbd.py

@check_x
def birnbaum_importance(
    self,
    x: Optional[ArrayLike] = None,
    working_nodes: Optional[Collection[Hashable]] = None,
    broken_nodes: Optional[Collection[Hashable]] = None,
) -> dict[Any, Union[float, np.ndarray]]:
    """Returns the Birnbaum measure of importance for all nodes.

    Note: Birnbaum's measure of importance assumes all nodes are
    independent.

    Parameters
    ----------
    x : ArrayLike
        Time/s as a number or iterable
    working_nodes : Collection[Hashable], optional
        Condition on these nodes being perfectly reliable, by default None
    broken_nodes : Collection[Hashable], optional
        Condition on these nodes having failed, by default None

    Returns
    -------
    dict[Any, float | np.ndarray]
        Dictionary with node names as keys and Birnbaum importances as
        values (floats for scalar ``x``, arrays for array ``x``)

    Examples
    --------
    In a two-component parallel system each node's Birnbaum importance is
    the probability the *other* node has failed:

    >>> from surpyval import FixedEventProbability
    >>> from repyability import NonRepairableRBD
    >>> rbd = NonRepairableRBD(
    ...     [("s", "a"), ("s", "b"), ("a", "t"), ("b", "t")],
    ...     {
    ...         "a": FixedEventProbability.from_params(0.1),
    ...         "b": FixedEventProbability.from_params(0.1),
    ...     },
    ... )
    >>> bi = rbd.birnbaum_importance()
    >>> {k: round(v, 4) for k, v in sorted(bi.items())}
    {'a': 0.1, 'b': 0.1}
    """
    self._require_no_ccf()
    node_probabilities = self._probabilities_with_overrides(
        self.node_sf(x), working_nodes, broken_nodes
    )
    return cast(
        Dict[Any, Union[float, np.ndarray]],
        super()._birnbaum_importance(node_probabilities),
    )

improvement_potential

improvement_potential(x=None, working_nodes=None, broken_nodes=None)

Returns the improvement potential of all nodes.

Parameters:

Name	Type	Description	Default
x	ArrayLike	Time/s as a number or iterable	None
working_nodes	Collection[Hashable]	Condition on these nodes being perfectly reliable, by default None	None
broken_nodes	Collection[Hashable]	Condition on these nodes having failed, by default None	None

Returns:

Type	Description
dict[Any, float | ndarray]	Dictionary with node names as keys and improvement potentials as values (floats for scalar x, arrays for array x)

Source code in repyability/rbd/non_repairable_rbd.py

@check_x
def improvement_potential(
    self,
    x: Optional[ArrayLike] = None,
    working_nodes: Optional[Collection[Hashable]] = None,
    broken_nodes: Optional[Collection[Hashable]] = None,
) -> dict[Any, Union[float, np.ndarray]]:
    """Returns the improvement potential of all nodes.

    Parameters
    ----------
    x : ArrayLike
        Time/s as a number or iterable
    working_nodes : Collection[Hashable], optional
        Condition on these nodes being perfectly reliable, by default None
    broken_nodes : Collection[Hashable], optional
        Condition on these nodes having failed, by default None

    Returns
    -------
    dict[Any, float | np.ndarray]
        Dictionary with node names as keys and improvement potentials as
        values (floats for scalar ``x``, arrays for array ``x``)
    """
    self._require_no_ccf()
    node_probabilities = self._probabilities_with_overrides(
        self.node_sf(x), working_nodes, broken_nodes
    )
    return cast(
        Dict[Any, Union[float, np.ndarray]],
        super()._improvement_potential(node_probabilities),
    )

risk_achievement_worth

risk_achievement_worth(x=None, working_nodes=None, broken_nodes=None)

Returns the RAW importance per Modarres & Kaminskiy. That is RAW_i = (unreliability of system given i failed) / (nominal system unreliability).

Parameters:

Name	Type	Description	Default
x	ArrayLike	Time/s as a number or iterable	None
working_nodes	Collection[Hashable]	Condition on these nodes being perfectly reliable, by default None	None
broken_nodes	Collection[Hashable]	Condition on these nodes having failed, by default None	None

Returns:

Type	Description
dict[Any, float | ndarray]	Dictionary with node names as keys and RAW importances as values (floats for scalar x, arrays for array x)

Source code in repyability/rbd/non_repairable_rbd.py

@check_x
def risk_achievement_worth(
    self,
    x: Optional[ArrayLike] = None,
    working_nodes: Optional[Collection[Hashable]] = None,
    broken_nodes: Optional[Collection[Hashable]] = None,
) -> dict[Any, Union[float, np.ndarray]]:
    """Returns the RAW importance per Modarres & Kaminskiy. That is RAW_i =
    (unreliability of system given i failed) /
    (nominal system unreliability).

    Parameters
    ----------
    x : ArrayLike
        Time/s as a number or iterable
    working_nodes : Collection[Hashable], optional
        Condition on these nodes being perfectly reliable, by default None
    broken_nodes : Collection[Hashable], optional
        Condition on these nodes having failed, by default None

    Returns
    -------
    dict[Any, float | np.ndarray]
        Dictionary with node names as keys and RAW importances as values
        (floats for scalar ``x``, arrays for array ``x``)
    """
    self._require_no_ccf()
    node_probabilities = self._probabilities_with_overrides(
        self.node_sf(x), working_nodes, broken_nodes
    )
    return cast(
        Dict[Any, Union[float, np.ndarray]],
        super()._risk_achievement_worth(node_probabilities),
    )

risk_reduction_worth

risk_reduction_worth(x=None, working_nodes=None, broken_nodes=None)

Returns the RRW importance per Modarres & Kaminskiy. That is RRW_i = (nominal unreliability of system) / (unreliability of system given i is working).

Parameters:

Name	Type	Description	Default
x	ArrayLike	Time/s as a number or iterable	None
working_nodes	Collection[Hashable]	Condition on these nodes being perfectly reliable, by default None	None
broken_nodes	Collection[Hashable]	Condition on these nodes having failed, by default None	None

Returns:

Type	Description
dict[Any, float | ndarray]	Dictionary with node names as keys and RRW importances as values (floats for scalar x, arrays for array x)

Source code in repyability/rbd/non_repairable_rbd.py

@check_x
def risk_reduction_worth(
    self,
    x: Optional[ArrayLike] = None,
    working_nodes: Optional[Collection[Hashable]] = None,
    broken_nodes: Optional[Collection[Hashable]] = None,
) -> dict[Any, Union[float, np.ndarray]]:
    """Returns the RRW importance per Modarres & Kaminskiy. That is RRW_i =
    (nominal unreliability of system) /
    (unreliability of system given i is working).

    Parameters
    ----------
    x : ArrayLike
        Time/s as a number or iterable
    working_nodes : Collection[Hashable], optional
        Condition on these nodes being perfectly reliable, by default None
    broken_nodes : Collection[Hashable], optional
        Condition on these nodes having failed, by default None

    Returns
    -------
    dict[Any, float | np.ndarray]
        Dictionary with node names as keys and RRW importances as values
        (floats for scalar ``x``, arrays for array ``x``)
    """
    self._require_no_ccf()
    node_probabilities = self._probabilities_with_overrides(
        self.node_sf(x), working_nodes, broken_nodes
    )
    return cast(
        Dict[Any, Union[float, np.ndarray]],
        super()._risk_reduction_worth(node_probabilities),
    )

criticality_importance

criticality_importance(x=None, working_nodes=None, broken_nodes=None)

Returns the criticality importance of all nodes at time/s x.

Parameters:

Name	Type	Description	Default
x	ArrayLike	Time/s as a number or iterable	None
working_nodes	Collection[Hashable]	Condition on these nodes being perfectly reliable, by default None	None
broken_nodes	Collection[Hashable]	Condition on these nodes having failed, by default None	None

Returns:

Type	Description
dict[Any, float | ndarray]	Dictionary with node names as keys and criticality importances as values (floats for scalar x, arrays for array x)

Source code in repyability/rbd/non_repairable_rbd.py

@check_x
def criticality_importance(
    self,
    x: Optional[ArrayLike] = None,
    working_nodes: Optional[Collection[Hashable]] = None,
    broken_nodes: Optional[Collection[Hashable]] = None,
) -> dict[Any, Union[float, np.ndarray]]:
    """Returns the criticality importance of all nodes at time/s x.

    Parameters
    ----------
    x : ArrayLike
        Time/s as a number or iterable
    working_nodes : Collection[Hashable], optional
        Condition on these nodes being perfectly reliable, by default None
    broken_nodes : Collection[Hashable], optional
        Condition on these nodes having failed, by default None

    Returns
    -------
    dict[Any, float | np.ndarray]
        Dictionary with node names as keys and criticality importances as
        values (floats for scalar ``x``, arrays for array ``x``)
    """
    self._require_no_ccf()
    node_probabilities = self._probabilities_with_overrides(
        self.node_sf(x), working_nodes, broken_nodes
    )
    return cast(
        Dict[Any, Union[float, np.ndarray]],
        super()._criticality_importance(node_probabilities),
    )

fussell_vesely

fussell_vesely(x=None, fv_type='c', working_nodes=None, broken_nodes=None)

Calculate Fussell-Vesely importance of all nodes at time/s x.

Briefly, the Fussell-Vesely importance measure for node i = (sum of probabilities of cut-sets including node i occuring/failing) / (the probability of the system failing).

Typically this measure is implemented using cut-sets as mentioned above, although the measure can be implemented using path-sets. Both are implemented here.

fv_type dictates the method: "c" - cut-set "p" - path-set

Parameters:

Name	Type	Description	Default
x	ArrayLike	Time/s as a number or iterable	None
fv_type	str	Dictates the method of calculation, 'c' = cut-set and 'p' = path-set, by default "c"	'c'
working_nodes	Collection[Hashable]	Condition on these nodes being perfectly reliable, by default None	None
broken_nodes	Collection[Hashable]	Condition on these nodes having failed, by default None	None

Returns:

Type	Description
dict[Any, float | ndarray]	Dictionary with node names as keys and Fussell-Vesely importances as values (floats for scalar x, arrays for array x)

Raises:

Type	Description
ValueError	If fv_type is not 'c' (cut-set) or 'p' (path-set).

Source code in repyability/rbd/non_repairable_rbd.py

@check_x
def fussell_vesely(
    self,
    x: Optional[ArrayLike] = None,
    fv_type: str = "c",
    working_nodes: Optional[Collection[Hashable]] = None,
    broken_nodes: Optional[Collection[Hashable]] = None,
) -> dict[Any, Union[float, np.ndarray]]:
    """Calculate Fussell-Vesely importance of all nodes at time/s x.

    Briefly, the Fussell-Vesely importance measure for node i =
    (sum of probabilities of cut-sets including node i occuring/failing) /
    (the probability of the system failing).

    Typically this measure is implemented using cut-sets as mentioned
    above, although the measure can be implemented using path-sets. Both
    are implemented here.

    fv_type dictates the method:
        "c" - cut-set
        "p" - path-set

    Parameters
    ----------
    x : ArrayLike
        Time/s as a number or iterable
    fv_type : str, optional
        Dictates the method of calculation, 'c' = cut-set and
        'p' = path-set, by default "c"
    working_nodes : Collection[Hashable], optional
        Condition on these nodes being perfectly reliable, by default None
    broken_nodes : Collection[Hashable], optional
        Condition on these nodes having failed, by default None

    Returns
    -------
    dict[Any, float | np.ndarray]
        Dictionary with node names as keys and Fussell-Vesely importances
        as values (floats for scalar ``x``, arrays for array ``x``)

    Raises
    ------
    ValueError
        If ``fv_type`` is not 'c' (cut-set) or 'p' (path-set).
    """
    self._require_no_ccf()
    rel_dict = {}
    for node_name, node in self.reliabilities.items():
        rel_dict[node_name] = node.sf(x)
    rel_dict = self._probabilities_with_overrides(
        rel_dict, working_nodes, broken_nodes
    )
    return cast(
        Dict[Any, Union[float, np.ndarray]],
        super()._fussell_vesely(rel_dict, fv_type),
    )

fussel_vesely

fussel_vesely(x=None, fv_type='c')

Deprecated alias for :meth:fussell_vesely (corrected spelling).

Source code in repyability/rbd/non_repairable_rbd.py

def fussel_vesely(
    self, x: Optional[ArrayLike] = None, fv_type: str = "c"
) -> dict[Any, Union[float, np.ndarray]]:
    """Deprecated alias for :meth:`fussell_vesely` (corrected spelling)."""
    warnings.warn(
        "fussel_vesely() is deprecated; use fussell_vesely() "
        "(Fussell-Vesely). This alias will be removed in a future "
        "release.",
        DeprecationWarning,
        stacklevel=2,
    )
    return self.fussell_vesely(x, fv_type)

parameter_sensitivity

parameter_sensitivity(x=None, working_nodes=None, broken_nodes=None, rel_step=1e-05)

Sensitivity of the system reliability to each node's distribution parameters at time/s x.

For node i with parameter theta, the sensitivity is

d R_sys / d theta = B_i(x) * d sf_i(x; theta) / d theta

where B_i is the Birnbaum importance of node i (how much the system reliability moves per unit change in that node's reliability) and d sf_i / d theta is how much the node's reliability moves per unit change in the parameter. The parameter derivative is taken numerically (a central finite difference, rebuilding the distribution with from_params), so it applies to any parametric surpyval model without per-distribution formulae. It answers "which fitted parameter, if it were a little different, would move system reliability the most" -- e.g. to target data collection or to gauge the impact of estimation uncertainty.

Only nodes with reconstructable distribution parameters are included; composite nodes (a nested RBD, a standby arrangement, a repeated node) and fitted non-parametric models have no parameters to perturb and are omitted. A node forced via working_nodes/broken_nodes is pinned independently of its parameters, so its sensitivities are reported as zero.

Parameters:

Name	Type	Description	Default
x	ArrayLike	Time/s as a number or iterable.	None
working_nodes	Collection[Hashable]	Condition on these nodes being perfectly reliable, by default None.	None
broken_nodes	Collection[Hashable]	Condition on these nodes having failed, by default None.	None
rel_step	float	Relative step used for the finite difference, by default 1e-5.	1e-05

Returns:

Type	Description
dict[Any, dict[str, float | ndarray]]	{node_name: {parameter_name: sensitivity}}. Sensitivities are floats for scalar x and arrays for array x.

Source code in repyability/rbd/non_repairable_rbd.py

def parameter_sensitivity(
    self,
    x: Optional[ArrayLike] = None,
    working_nodes: Optional[Collection[Hashable]] = None,
    broken_nodes: Optional[Collection[Hashable]] = None,
    rel_step: float = 1e-5,
) -> Dict[Any, Dict[str, Union[float, np.ndarray]]]:
    """Sensitivity of the system reliability to each node's distribution
    parameters at time/s ``x``.

    For node ``i`` with parameter ``theta``, the sensitivity is

    ``d R_sys / d theta = B_i(x) * d sf_i(x; theta) / d theta``

    where ``B_i`` is the Birnbaum importance of node ``i`` (how much the
    system reliability moves per unit change in that node's reliability)
    and ``d sf_i / d theta`` is how much the node's reliability moves per
    unit change in the parameter. The parameter derivative is taken
    numerically (a central finite difference, rebuilding the distribution
    with ``from_params``), so it applies to any parametric surpyval model
    without per-distribution formulae. It answers "which fitted parameter,
    if it were a little different, would move system reliability the most"
    -- e.g. to target data collection or to gauge the impact of estimation
    uncertainty.

    Only nodes with reconstructable distribution parameters are included;
    composite nodes (a nested RBD, a standby arrangement, a repeated node)
    and fitted non-parametric models have no parameters to perturb and are
    omitted. A node forced via ``working_nodes``/``broken_nodes`` is pinned
    independently of its parameters, so its sensitivities are reported as
    zero.

    Parameters
    ----------
    x : ArrayLike
        Time/s as a number or iterable.
    working_nodes : Collection[Hashable], optional
        Condition on these nodes being perfectly reliable, by default None.
    broken_nodes : Collection[Hashable], optional
        Condition on these nodes having failed, by default None.
    rel_step : float, optional
        Relative step used for the finite difference, by default ``1e-5``.

    Returns
    -------
    dict[Any, dict[str, float | np.ndarray]]
        ``{node_name: {parameter_name: sensitivity}}``. Sensitivities are
        floats for scalar ``x`` and arrays for array ``x``.
    """
    if x is None:
        if self.is_fixed:
            x = 1.0
        else:
            raise ValueError(
                "x is required: this RBD is time-varying (at least one "
                "node model's probability depends on time)."
            )
    scalar_in = np.ndim(x) == 0
    x_arr = np.atleast_1d(np.asarray(x, dtype=float))

    # Birnbaum importance validates the override sets and honours them.
    birnbaum = self.birnbaum_importance(x_arr, working_nodes, broken_nodes)
    forced = set(working_nodes or ()) | set(broken_nodes or ())

    def _out(value) -> Union[float, np.ndarray]:
        arr = np.asarray(value, dtype=float)
        return float(arr.reshape(-1)[0]) if scalar_in else arr

    sensitivities: Dict[Any, Dict[str, Union[float, np.ndarray]]] = {}
    for node_name, model in self.reliabilities.items():
        spec = parametric_spec(model)
        if spec is None:
            # Composite / non-parametric node: no parameters to perturb.
            continue
        cls, params, names = spec
        node_out: Dict[str, Union[float, np.ndarray]] = {}
        if node_name in forced:
            # Pinned regardless of its parameters -> zero sensitivity.
            zero = np.zeros_like(x_arr)
            for name in names:
                node_out[name] = _out(zero)
            sensitivities[node_name] = node_out
            continue
        b_i = np.asarray(birnbaum[node_name], dtype=float)
        for j, name in enumerate(names):
            dsf = _dsf_dparam(cls, params, j, x_arr, rel_step)
            node_out[name] = _out(b_i * dsf)
        sensitivities[node_name] = node_out
    return sensitivities

RepairableRBD

RepairableRBD(edges, components, k=None, input_node=None, output_node=None, on_infeasible_rbd='raise')

Bases: RBD

Source code in repyability/rbd/repairable_rbd.py

def __init__(
    self,
    edges: Iterable[tuple[Hashable, Hashable]],
    components: dict[Any, Any],
    k: Optional[dict[Any, int]] = None,
    input_node: Optional[Any] = None,
    output_node: Optional[Any] = None,
    on_infeasible_rbd: str = "raise",
):
    # Capture the constructor inputs verbatim (before any mutation) so the
    # RBD can be faithfully serialised via to_dict()/to_json().
    edges = list(edges)
    self._init_args = {
        "edges": [tuple(e) for e in edges],
        "components": dict(components),
        "k": dict(k) if k else None,
        "input_node": input_node,
        "output_node": output_node,
        "on_infeasible_rbd": on_infeasible_rbd,
    }
    components = copy(components)
    reliability = {}
    repairability = {}
    for name, component in components.items():
        if isinstance(component, dict):
            components[name] = NonRepairable(
                component["reliability"], component["repairability"]
            )
            reliability[name] = component["reliability"]
            repairability[name] = component["repairability"]
        elif isinstance(component, RepairableRBD):
            reliability[name] = component
            repairability[name] = None
        elif isinstance(component, NonRepairable):
            reliability[name] = component.reliability
            repairability[name] = component.time_to_replace

    super().__init__(
        edges,
        set(components.keys()),
        k,
        input_node,
        output_node,
        on_infeasible_rbd,
    )

    # Every intermediate graph node needs a component definition (the
    # input/output nodes do not). Surface missing ones now with a clear
    # error rather than a KeyError mid-simulation.
    missing = [
        n
        for n in self.G.nodes
        if n not in components and n not in self.in_or_out
    ]
    self.structure_check["is_missing_components"] = bool(missing)
    self.structure_check["nodes_with_no_component"] = missing
    if missing:
        self.structure_check["is_valid"] = False
        if on_infeasible_rbd == "raise":
            raise ValueError(
                f"Node(s) {sorted(missing, key=str)} have no entry in "
                "the components dict."
            )
        elif on_infeasible_rbd == "warn":
            warnings.warn(
                "Nodes with no component definition: "
                + pprint.pformat(missing),
                stacklevel=2,
            )

    self.components = components
    self.repairability = copy(repairability)

mean_unavailability

mean_unavailability(*args, **kwargs)

Returns the system long run UNavailability

Parameters:

Name	Type	Description	Default
*args		Any mean_availability() arguments	()
**kwargs		Any mean_availability() arguments	()

Returns:

Type	Description
float	Long run unavailability of the system

Source code in repyability/rbd/repairable_rbd.py

def mean_unavailability(self, *args, **kwargs) -> float:
    """Returns the system long run UNavailability

    Parameters
    ----------
    *args, **kwargs :
        Any mean_availability() arguments

    Returns
    -------
    float
        Long run unavailability of the system
    """
    return 1 - self.mean_availability(*args, **kwargs)

mean_availability

mean_availability(working_nodes=None, broken_nodes=None, method='p')

Returns the system long run availability

Parameters:

Name	Type	Description	Default
working_nodes	Collection[Hashable]	Marks these nodes as always available, by default None	None
broken_nodes	Collection[Hashable]	Marks these nodes as failed, by default None	None
method	str	Input either "c" or "p" for the function to use cut sets or path sets respectively. Defaults to path sets.	'p'

Returns:

Type	Description
float	Long run availability of the system

Raises:

Type	Description
ValueError	If a working/broken node is unknown, is the input/output node, or is in both sets.

Examples:

A single component with mean time to failure 10 and mean time to repair 1 has long-run availability 10 / (10 + 1):

>>> import surpyval as surv
>>> from repyability import RepairableRBD
>>> rbd = RepairableRBD(
...     [("s", "c"), ("c", "t")],
...     {
...         "c": {
...             "reliability": surv.Exponential.from_params([0.1]),
...             "repairability": surv.Exponential.from_params([1.0]),
...         }
...     },
... )
>>> round(rbd.mean_availability(), 4)
0.9091

Source code in repyability/rbd/repairable_rbd.py

def mean_availability(
    self,
    working_nodes: Optional[Collection[Hashable]] = None,
    broken_nodes: Optional[Collection[Hashable]] = None,
    method: str = "p",
) -> float:
    """Returns the system long run availability

    Parameters
    ----------
    working_nodes : Collection[Hashable], optional
        Marks these nodes as always available, by default None
    broken_nodes : Collection[Hashable], optional
        Marks these nodes as failed, by default None
    method : str, optional
        Input either "c" or "p" for the function to use cut sets or path
        sets respectively. Defaults to path sets.

    Returns
    -------
    float
        Long run availability of the system

    Raises
    ------
    ValueError
        If a working/broken node is unknown, is the input/output node, or
        is in both sets.

    Examples
    --------
    A single component with mean time to failure 10 and mean time to
    repair 1 has long-run availability ``10 / (10 + 1)``:

    >>> import surpyval as surv
    >>> from repyability import RepairableRBD
    >>> rbd = RepairableRBD(
    ...     [("s", "c"), ("c", "t")],
    ...     {
    ...         "c": {
    ...             "reliability": surv.Exponential.from_params([0.1]),
    ...             "repairability": surv.Exponential.from_params([1.0]),
    ...         }
    ...     },
    ... )
    >>> round(rbd.mean_availability(), 4)
    0.9091
    """
    # Good reference on the Availability of a system
    # https://www.diva-portal.org/smash/get/diva2:986067/FULLTEXT01.pdf
    working_nodes = set() if working_nodes is None else set(working_nodes)
    broken_nodes = set() if broken_nodes is None else set(broken_nodes)
    self._validate_node_overrides(working_nodes, broken_nodes)

    # Cache all component availabilities for efficiency
    component_availability: dict[Hashable, float] = {}
    for comp in self.components:
        if comp in working_nodes:
            component_availability[comp] = 1.0
        elif comp in broken_nodes:
            component_availability[comp] = 0.0
        else:
            component_availability[comp] = float(
                np.atleast_1d(self.components[comp].mean_availability())[0]
            )

    for comp in self.in_or_out:
        component_availability[comp] = 1.0

    mean_availability = self.system_probability(
        component_availability, method=method
    )
    return mean_availability.item()

availability

availability(t_simulation, working_nodes=None, broken_nodes=None, method='p', N=10000, verbose=False, seed=None)

Returns the times, and availability for those times, as numpy arrays

Parameters:

Name	Type	Description	Default
t_simulation	float	Units of time to run each simulation for	required
N	int	Number of simulations, by default 10_000	10000
verbose	bool	If True, displays progress bar of simulations, by default False	False

Returns:

Type	Description
tuple[ndarray, ndarray]	times, availabilities

Source code in repyability/rbd/repairable_rbd.py

def availability(
    self,
    t_simulation: float,
    working_nodes: Optional[Collection[Hashable]] = None,
    broken_nodes: Optional[Collection[Hashable]] = None,
    method: str = "p",
    N: int = 10_000,
    verbose: bool = False,
    seed: Optional[int] = None,
) -> AvailabilityResult:
    """Returns the times, and availability for those times, as numpy
    arrays

    Parameters
    ----------
    t_simulation : float
        Units of time to run each simulation for
    N : int, optional
        Number of simulations, by default 10_000
    verbose : bool, optional
        If True, displays progress bar of simulations, by default False

    Returns
    -------
    tuple[np.ndarray, np.ndarray]
        times, availabilities
    """
    working_nodes = set() if working_nodes is None else set(working_nodes)
    broken_nodes = set() if broken_nodes is None else set(broken_nodes)
    self._validate_node_overrides(working_nodes, broken_nodes)

    # aggregate_timeline keeps track of how many of the simulated systems
    # turn on and off at time t.
    # e.g. aggregate_timeline[t] = +2 means two out of the N simulated
    # systems began working again at time t, while
    # aggregate_timeline[t] = -1 means one out of the N simulated systems
    # stopped working at time t.
    # There is a very strong expectation that due to the random sampling
    # that the generall aggregate_timeline[t] would be only -1 or +1.
    aggregate_timeline: dict[float, int] = defaultdict(lambda: 0)
    # The below two assigments ensure the results have data at 0 and time
    # t_simulation regardless of whether it was sampled at these times.
    # Set the end of the timeline to be 0 (i.e. unchanged if no event
    # falls) exactly at time t_simulation.
    aggregate_timeline[t_simulation] = 0
    # The initial system state is the same for every simulation (the forced
    # working/broken sets are fixed): all components start working except
    # those forced broken, which can make the system start down (e.g. a
    # broken component in series). Seed time 0 accordingly.
    initial_status = {c: c not in broken_nodes for c in self.components}
    initial_system_up = bool(
        self.is_system_working(initial_status, method)
    )
    aggregate_timeline[0] = N if initial_system_up else 0

    # Restoration Criticality Index
    RCI: defaultdict = defaultdict(lambda: defaultdict(lambda: 0))
    system_restorations = 0
    system_downtime = 0

    # Failure Criticality Index
    FCI: defaultdict = defaultdict(lambda: defaultdict(lambda: 0))
    system_failures = 0
    system_uptime = 0

    node_downtime: defaultdict = defaultdict(lambda: 0)
    node_uptime: defaultdict = defaultdict(lambda: 0)
    intersection_uptime: defaultdict = defaultdict(lambda: 0)
    intersection_downtime: defaultdict = defaultdict(lambda: 0)
    union_uptime: defaultdict = defaultdict(lambda: 0)
    union_downtime: defaultdict = defaultdict(lambda: 0)

    # Perform N simulations. surpyval's ``.random`` draws from numpy's
    # global RNG, so seed it here (if a seed was given) to make the run
    # reproducible, restoring the caller's RNG state once the randomised
    # simulations below have finished.
    _rng_state = None
    if seed is not None:
        _rng_state = np.random.get_state()
        np.random.seed(seed)

    for _ in tqdm(
        range(N), disable=not verbose, desc="Running simulations"
    ):
        # Initialize the event queue and the system/component statuses
        self.initialize_event_queue(
            t_simulation,
            working_nodes,
            broken_nodes,
            method,
        )

        # Seed each timeline from the actual initial status so that
        # forced-broken components (and a system they start down) are
        # accounted as down from t=0, not assumed up.
        component_timelines: dict = {
            comp: [(0.0, 1 if self.component_status[comp] else 0)]
            for comp in self.components
        }
        system_timeline = [(0.0, 1 if self.system_state else 0)]

        # Implemented ensure that no events that occur after the
        # end-time of the simulation are added to the queue; so we just
        # need to keep going through the queue until it's empty
        while not self._event_queue.empty():
            # Get the next event and update the component's status

            event = self._event_queue.get()
            # Update the component's status
            self.component_status[event.component] = event.status
            if event.status:
                RCI[event.component]["component_restorations"] += 1
            else:
                FCI[event.component]["component_failures"] += 1

            status = 1 if event.status else -1
            component_timelines[event.component].append(
                (event.time, status)
            )

            # Record new system state, it could still be the same as
            # system_state in which case we don't bother changing
            # aggregate_timeline, but if it is different, we need to +/-1
            # to aggregate_timeline if the system has gone on/off-line
            new_system_state = self.is_system_working(
                self.component_status, method
            )
            if new_system_state != self.system_state:
                status = 1 if new_system_state else -1
                system_timeline.append((event.time, status))
                if new_system_state:
                    # System restored
                    aggregate_timeline[event.time] += 1
                    system_restorations += 1
                    RCI[event.component]["system_restorations"] += 1
                else:
                    aggregate_timeline[event.time] -= 1
                    system_failures += 1
                    FCI[event.component]["system_failures"] += 1

                # Set the system_state to the new state
                self.system_state = new_system_state

            # Now we need to get the component's next event
            # If the component just got repaired then we need it's next
            # failure event, otherwise it just broke and we need it's
            # repair event
            next_event_t, next_event_type = self.components[
                event.component
            ].next_event()

            next_event = Event(
                # The next event time is the current time [event.time]
                # plus the time to next event
                event.time + next_event_t,
                event.component,
                next_event_type,
            )
            # But only queue up the event if it occurs before the end
            # of the simulation
            if next_event.time < t_simulation:
                self._event_queue.put(next_event)

            # Then move on to the next event... until there's no more
            # events in the queue

        system_timeline.append((t_simulation, 0))

        for component in self.components.keys():
            component_timelines[component].append((t_simulation, 0))
            # This simulation's uptime for the component; the downtime is
            # the remainder of the window. (Use the per-simulation value,
            # not the running cumulative node_uptime[component].)
            component_ut = time_at_status(
                component_timelines[component], 1
            )
            node_uptime[component] += component_ut
            node_downtime[component] += t_simulation - component_ut
            joint_t, joint_events = combined_timeline(
                component_timelines[component], system_timeline
            )
            intersection_uptime[component] += intersection(
                joint_t, joint_events
            )
            intersection_downtime[component] += intersection(
                joint_t, 2 - joint_events
            )
            union_uptime[component] += union(joint_t, joint_events)
            union_downtime[component] += union(joint_t, 2 - joint_events)

        simulation_system_ut = time_at_status(system_timeline, 1)
        system_uptime += simulation_system_ut
        system_downtime += t_simulation - simulation_system_ut

    # Randomised simulations are done; restore the caller's RNG state.
    if _rng_state is not None:
        np.random.set_state(_rng_state)

    # Collect Importance/Criticality measures from the simulation
    # reference: https://www.weibull.com/pubs/2004rm_05B_02.pdf
    # Operational Criticality Index
    oci_down = {
        k: _safe_ratio(v, system_downtime)
        for k, v in dict(intersection_downtime).items()
    }
    oci_up = {
        k: _safe_ratio(v, system_uptime)
        for k, v in dict(intersection_uptime).items()
    }
    # Intersection Over Union Importance
    iou_up = {
        k: _safe_ratio(intersection_uptime[k], union_uptime[k])
        for k in dict(intersection_uptime).keys()
    }
    iou_down = {
        k: _safe_ratio(intersection_downtime[k], union_downtime[k])
        for k in dict(intersection_downtime).keys()
    }
    # Failure Criticality Index Importance
    fci_sys = failure_criticality_index_per_system_failures(
        FCI, system_failures
    )
    fci_comp = failure_criticality_index_per_component_failures(FCI)
    # Restoration Criticality Index Importance
    rci_sys = restoration_criticality_index_by_system(
        RCI, system_restorations
    )
    rci_comp = restoration_criticality_index_by_component(RCI)
    criticalities = Criticalities(
        operational_criticality_index=UpDownImportance(
            up=oci_up, down=oci_down
        ),
        iou=UpDownImportance(up=iou_up, down=iou_down),
        failure_criticality_index=FailureCriticalityIndex(
            per_system_failure=fci_sys, per_component_failure=fci_comp
        ),
        restoration_criticality_index=RestorationCriticalityIndex(
            by_system=rci_sys, by_component=rci_comp
        ),
    )

    # Now we need to return the system availability from t=0..t_simulation
    # Using numpy arrays for efficiency
    timeline_arr: np.ndarray = np.array(list(aggregate_timeline.items()))

    # Sort the array by event time
    timeline_arr = timeline_arr[timeline_arr[:, 0].argsort()]
    time = timeline_arr[:, 0]

    # Take the cumulative sum, this is basically calculating for each
    # t just how many systems are working, and divide by N to get
    # availability the as a percentage
    system_availability = timeline_arr[:, 1].cumsum() / N

    # Clean up the interim variables of the simulation
    del self._event_queue
    del self.system_state
    del self.t_simulation
    del self.component_status

    simulation_results = AvailabilityResult(
        timeline=time,
        availability=system_availability,
        system_uptime=system_uptime,
        time_simulated_to=t_simulation,
        criticalities=criticalities,
        node_uptime=dict(node_uptime),
        node_downtime=dict(node_downtime),
        system_downtime=system_downtime,
        system_failures=system_failures,
        system_restorations=system_restorations,
        n_simulations=N,
    )

    return simulation_results

node_availability

node_availability()

Returns each node's long-run availability (a dict keyed by node name); the input/output nodes are 1.0.

Source code in repyability/rbd/repairable_rbd.py

def node_availability(self) -> dict[Hashable, float]:
    """Returns each node's long-run availability (a dict keyed by node
    name); the input/output nodes are 1.0."""
    node_av: dict[Hashable, float] = {}
    for node_name, component in self.components.items():
        node_av[node_name] = float(
            np.atleast_1d(component.mean_availability())[0]
        )

    for node_name in self.in_or_out:
        node_av[node_name] = 1.0

    return node_av

system_failure_frequency

system_failure_frequency(working_nodes=None, broken_nodes=None)

Returns the system's long-run failure frequency (failures per unit time), by the Birnbaum/Vesely formula.

In steady state the system failure frequency is

.. math:: \omega = \sum_i I_B^i \cdot \omega_i

where :math:I_B^i is node i's Birnbaum importance evaluated at the nodes' availabilities and :math:\omega_i = 1/(MTTF_i + MTTR_i) is node i's failure frequency. Exact for independent repairable nodes.

Parameters:

Name	Type	Description	Default
working_nodes	Collection[Hashable]	Condition on these nodes always working (they contribute no failures), by default None	None
broken_nodes	Collection[Hashable]	Condition on these nodes being failed (they contribute no failures), by default None	None

Source code in repyability/rbd/repairable_rbd.py

def system_failure_frequency(
    self,
    working_nodes: Optional[Collection[Hashable]] = None,
    broken_nodes: Optional[Collection[Hashable]] = None,
) -> float:
    """Returns the system's long-run failure frequency (failures per unit
    time), by the Birnbaum/Vesely formula.

    In steady state the system failure frequency is

    .. math::
        \\omega = \\sum_i I_B^i \\cdot \\omega_i

    where :math:`I_B^i` is node i's Birnbaum importance evaluated at the
    nodes' availabilities and :math:`\\omega_i = 1/(MTTF_i + MTTR_i)` is
    node i's failure frequency. Exact for independent repairable nodes.

    Parameters
    ----------
    working_nodes : Collection[Hashable], optional
        Condition on these nodes always working (they contribute no
        failures), by default None
    broken_nodes : Collection[Hashable], optional
        Condition on these nodes being failed (they contribute no
        failures), by default None
    """
    availability = self._probabilities_with_overrides(
        self.node_availability(), working_nodes, broken_nodes
    )
    forced = (set() if working_nodes is None else set(working_nodes)) | (
        set() if broken_nodes is None else set(broken_nodes)
    )
    birnbaum = super()._birnbaum_importance(availability)
    omega = 0.0
    for node, component in self.components.items():
        if node in forced:
            # A forced node never changes state, so it contributes no
            # system failures.
            continue
        omega += np.asarray(birnbaum[node]).item() * (
            self._node_failure_frequency(component)
        )
    return omega

mean_time_between_failures

mean_time_between_failures(working_nodes=None, broken_nodes=None)

Returns the system's long-run Mean Time Between Failures, MTBF = 1 / failure frequency — the mean length of one full up-down cycle, i.e. MTBF = MUT + MDT. Infinite if the system never fails.

Source code in repyability/rbd/repairable_rbd.py

def mean_time_between_failures(
    self,
    working_nodes: Optional[Collection[Hashable]] = None,
    broken_nodes: Optional[Collection[Hashable]] = None,
) -> float:
    """Returns the system's long-run Mean Time Between Failures,
    ``MTBF = 1 / failure frequency`` — the mean length of one full
    up-down cycle, i.e. ``MTBF = MUT + MDT``. Infinite if the system
    never fails."""
    omega = self.system_failure_frequency(working_nodes, broken_nodes)
    return 1.0 / omega if omega > 0.0 else float("inf")

mean_up_time

mean_up_time(working_nodes=None, broken_nodes=None)

Returns the system's long-run Mean Up Time (mean duration of an uninterrupted working period; sometimes called the repairable-system MTTF), MUT = availability / failure frequency.

Source code in repyability/rbd/repairable_rbd.py

def mean_up_time(
    self,
    working_nodes: Optional[Collection[Hashable]] = None,
    broken_nodes: Optional[Collection[Hashable]] = None,
) -> float:
    """Returns the system's long-run Mean Up Time (mean duration of an
    uninterrupted working period; sometimes called the repairable-system
    MTTF), ``MUT = availability / failure frequency``."""
    availability = self.mean_availability(working_nodes, broken_nodes)
    omega = self.system_failure_frequency(working_nodes, broken_nodes)
    if omega > 0.0:
        return float(availability) / omega
    return float("inf") if availability > 0.0 else 0.0

mean_down_time

mean_down_time(working_nodes=None, broken_nodes=None)

Returns the system's long-run Mean Down Time (mean duration of an outage; the repairable-system MTTR), MDT = unavailability / failure frequency.

Source code in repyability/rbd/repairable_rbd.py

def mean_down_time(
    self,
    working_nodes: Optional[Collection[Hashable]] = None,
    broken_nodes: Optional[Collection[Hashable]] = None,
) -> float:
    """Returns the system's long-run Mean Down Time (mean duration of an
    outage; the repairable-system MTTR),
    ``MDT = unavailability / failure frequency``."""
    availability = self.mean_availability(working_nodes, broken_nodes)
    omega = self.system_failure_frequency(working_nodes, broken_nodes)
    if omega > 0.0:
        return (1.0 - float(availability)) / omega
    return 0.0 if availability >= 1.0 else float("inf")

birnbaum_importance

birnbaum_importance(working_nodes=None, broken_nodes=None)

Returns the Birnbaum measure of importance for all nodes, evaluated at the nodes' long-run availabilities.

Note: Birnbaum's measure of importance assumes all nodes are independent.

Parameters:

Name	Type	Description	Default
working_nodes	Collection[Hashable]	Condition on these nodes being always available, by default None	None
broken_nodes	Collection[Hashable]	Condition on these nodes being failed, by default None	None

Returns:

Type	Description
dict[Any, float]	Dictionary with node names as keys and Birnbaum importances as values

Source code in repyability/rbd/repairable_rbd.py

def birnbaum_importance(
    self,
    working_nodes: Optional[Collection[Hashable]] = None,
    broken_nodes: Optional[Collection[Hashable]] = None,
) -> dict[Any, float]:
    """Returns the Birnbaum measure of importance for all nodes,
    evaluated at the nodes' long-run availabilities.

    Note: Birnbaum's measure of importance assumes all nodes are
    independent.

    Parameters
    ----------
    working_nodes : Collection[Hashable], optional
        Condition on these nodes being always available, by default None
    broken_nodes : Collection[Hashable], optional
        Condition on these nodes being failed, by default None

    Returns
    -------
    dict[Any, float]
        Dictionary with node names as keys and Birnbaum importances as
        values
    """
    node_probabilities = self._probabilities_with_overrides(
        self.node_availability(), working_nodes, broken_nodes
    )
    return _squeeze_values(
        super()._birnbaum_importance(node_probabilities)
    )

improvement_potential

improvement_potential(working_nodes=None, broken_nodes=None)

Returns the improvement potential of all nodes, evaluated at the nodes' long-run availabilities.

Returns:

Type	Description
dict[Any, float]	Dictionary with node names as keys and improvement potentials as values

Source code in repyability/rbd/repairable_rbd.py

def improvement_potential(
    self,
    working_nodes: Optional[Collection[Hashable]] = None,
    broken_nodes: Optional[Collection[Hashable]] = None,
) -> dict[Any, float]:
    """Returns the improvement potential of all nodes, evaluated at the
    nodes' long-run availabilities.

    Returns
    -------
    dict[Any, float]
        Dictionary with node names as keys and improvement potentials as
        values
    """
    node_probabilities = self._probabilities_with_overrides(
        self.node_availability(), working_nodes, broken_nodes
    )
    return _squeeze_values(
        super()._improvement_potential(node_probabilities)
    )

risk_achievement_worth

risk_achievement_worth(working_nodes=None, broken_nodes=None)

Returns the RAW importance per Modarres & Kaminskiy, evaluated at the nodes' long-run availabilities. That is RAW_i = (unavailability of system given i failed) / (nominal system unavailability).

Returns:

Type	Description
dict[Any, float]	Dictionary with node names as keys and RAW importances as values

Source code in repyability/rbd/repairable_rbd.py

def risk_achievement_worth(
    self,
    working_nodes: Optional[Collection[Hashable]] = None,
    broken_nodes: Optional[Collection[Hashable]] = None,
) -> dict[Any, float]:
    """Returns the RAW importance per Modarres & Kaminskiy, evaluated at
    the nodes' long-run availabilities. That is RAW_i =
    (unavailability of system given i failed) /
    (nominal system unavailability).

    Returns
    -------
    dict[Any, float]
        Dictionary with node names as keys and RAW importances as values
    """
    node_probabilities = self._probabilities_with_overrides(
        self.node_availability(), working_nodes, broken_nodes
    )
    return _squeeze_values(
        super()._risk_achievement_worth(node_probabilities)
    )

risk_reduction_worth

risk_reduction_worth(working_nodes=None, broken_nodes=None)

Returns the RRW importance per Modarres & Kaminskiy, evaluated at the nodes' long-run availabilities. That is RRW_i = (nominal unavailability of system) / (unavailability of system given i is working).

Returns:

Type	Description
dict[Any, float]	Dictionary with node names as keys and RRW importances as values

Source code in repyability/rbd/repairable_rbd.py

def risk_reduction_worth(
    self,
    working_nodes: Optional[Collection[Hashable]] = None,
    broken_nodes: Optional[Collection[Hashable]] = None,
) -> dict[Any, float]:
    """Returns the RRW importance per Modarres & Kaminskiy, evaluated at
    the nodes' long-run availabilities. That is RRW_i =
    (nominal unavailability of system) /
    (unavailability of system given i is working).

    Returns
    -------
    dict[Any, float]
        Dictionary with node names as keys and RRW importances as values
    """
    node_probabilities = self._probabilities_with_overrides(
        self.node_availability(), working_nodes, broken_nodes
    )
    return _squeeze_values(
        super()._risk_reduction_worth(node_probabilities)
    )

criticality_importance

criticality_importance(working_nodes=None, broken_nodes=None)

Returns the criticality importance of all nodes, evaluated at the nodes' long-run availabilities.

Returns:

Type	Description
dict[Any, float]	Dictionary with node names as keys and criticality importances as values

Source code in repyability/rbd/repairable_rbd.py

def criticality_importance(
    self,
    working_nodes: Optional[Collection[Hashable]] = None,
    broken_nodes: Optional[Collection[Hashable]] = None,
) -> dict[Any, float]:
    """Returns the criticality importance of all nodes, evaluated at the
    nodes' long-run availabilities.

    Returns
    -------
    dict[Any, float]
        Dictionary with node names as keys and criticality importances as
        values
    """
    node_probabilities = self._probabilities_with_overrides(
        self.node_availability(), working_nodes, broken_nodes
    )
    return _squeeze_values(
        super()._criticality_importance(node_probabilities)
    )

fussell_vesely

fussell_vesely(fv_type='c', working_nodes=None, broken_nodes=None)

Calculate Fussell-Vesely importance of all nodes, evaluated at the nodes' long-run availabilities.

Briefly, the Fussell-Vesely importance measure for node i = (sum of probabilities of cut-sets including node i occuring/failing) / (the probability of the system failing).

Typically this measure is implemented using cut-sets as mentioned above, although the measure can be implemented using path-sets. Both are implemented here.

fv_type dictates the method: "c" - cut-set "p" - path-set

Parameters:

Name	Type	Description	Default
fv_type	str	Dictates the method of calculation, 'c' = cut-set and 'p' = path-set, by default "c"	'c'
working_nodes	Collection[Hashable]	Condition on these nodes being always available, by default None	None
broken_nodes	Collection[Hashable]	Condition on these nodes being failed, by default None	None

Returns:

Type	Description
dict[Any, float]	Dictionary with node names as keys and Fussell-Vesely importances as values

Raises:

Type	Description
ValueError	If fv_type is not 'c' (cut-set) or 'p' (path-set).

Source code in repyability/rbd/repairable_rbd.py

def fussell_vesely(
    self,
    fv_type: str = "c",
    working_nodes: Optional[Collection[Hashable]] = None,
    broken_nodes: Optional[Collection[Hashable]] = None,
) -> dict[Any, float]:
    """Calculate Fussell-Vesely importance of all nodes, evaluated at the
    nodes' long-run availabilities.

    Briefly, the Fussell-Vesely importance measure for node i =
    (sum of probabilities of cut-sets including node i occuring/failing) /
    (the probability of the system failing).

    Typically this measure is implemented using cut-sets as mentioned
    above, although the measure can be implemented using path-sets. Both
    are implemented here.

    fv_type dictates the method:
        "c" - cut-set
        "p" - path-set

    Parameters
    ----------
    fv_type : str, optional
        Dictates the method of calculation, 'c' = cut-set and
        'p' = path-set, by default "c"
    working_nodes : Collection[Hashable], optional
        Condition on these nodes being always available, by default None
    broken_nodes : Collection[Hashable], optional
        Condition on these nodes being failed, by default None

    Returns
    -------
    dict[Any, float]
        Dictionary with node names as keys and Fussell-Vesely importances
        as values

    Raises
    ------
    ValueError
        If ``fv_type`` is not 'c' (cut-set) or 'p' (path-set).
    """
    node_probabilities = self._probabilities_with_overrides(
        self.node_availability(), working_nodes, broken_nodes
    )
    return _squeeze_values(
        super()._fussell_vesely(node_probabilities, fv_type)
    )

fussel_vesely

fussel_vesely(fv_type='c')

Deprecated alias for :meth:fussell_vesely (corrected spelling).

Source code in repyability/rbd/repairable_rbd.py

def fussel_vesely(self, fv_type: str = "c") -> dict[Any, float]:
    """Deprecated alias for :meth:`fussell_vesely` (corrected spelling)."""
    warnings.warn(
        "fussel_vesely() is deprecated; use fussell_vesely() "
        "(Fussell-Vesely). This alias will be removed in a future "
        "release.",
        DeprecationWarning,
        stacklevel=2,
    )
    return self.fussell_vesely(fv_type)

Component models

NonRepairable

NonRepairable(reliability, time_to_replace=ExactEventTime.from_params(0))

A component renewed by replacement ("as good as new").

Pairs a lifetime (reliability) model with a time-to-replace model for a unit that cannot be repaired in place: every failure or planned replacement fits a new unit, so each cycle is a statistical renewal.

It plays two roles:

• Standalone, it prices the classic age-replacement policy — replace preventively at age t (planned, cost cp) or on failure (unplanned, cost cu > cp) — via find_optimal_replacement() and optimal_replacement_policy().
• Inside RepairableRBD, it is the per-component representation used by the availability engine (components are "repaired" by as-new replacement, which is exactly the renewal this class models).

Contrast with Repairable, which models minimal repair ("as bad as old") and the overhaul-interval policy.

Source code in repyability/non_repairable.py

def __init__(
    self, reliability, time_to_replace=ExactEventTime.from_params(0)
):
    if isinstance(reliability, Parametric):
        self.model_parameterization = "parametric"
        self.reliability_function = reliability.sf
    elif isinstance(reliability, NonParametric):
        # TODO: Allow non-interpolated?
        self.model_parameterization = "non-parametric"
        self.reliability_function = interp1d(
            reliability.x, 1 - reliability.F, fill_value="extrapolate"
        )
    elif isinstance(reliability, StandbyModel):
        self.model_parameterization = "non-parametric"
        self.reliability_function = interp1d(
            reliability.model.x,
            1 - reliability.model.F,
            fill_value="extrapolate",
        )
    else:
        raise ValueError("Unknown reliability function")

    self.reliability = reliability
    self.time_to_replace = time_to_replace
    self.cost_rate = np.vectorize(
        self._cost_rate,
        doc=(
            "Long-run cost per unit time of an age-replacement policy "
            "with replacement age t:\n"
            "(cp * R(t) + cu * F(t)) / integral_0^t R(u) du.\n"
            "Vectorised over t."
        ),
    )
    self.__next_event_type = FAILURE

mean_availability

mean_availability()

Long-run availability, MTTF / (MTTF + MTTR).

Source code in repyability/non_repairable.py

def mean_availability(self) -> float:
    """Long-run availability, MTTF / (MTTF + MTTR)."""
    if isinstance(self.reliability, NonParametric):
        raise ValueError(
            "Mean Availability requires a parametric reliability model"
        )
    mttf = model_mean(self.reliability)
    mttr = model_mean(self.time_to_replace)
    return mttf / (mttr + mttf)

failure_frequency

failure_frequency()

Long-run failure frequency (failures per unit time).

For an alternating renewal process (fail, repair, fail, ...) this is 1 / (MTTF + MTTR) — one failure per mean up-down cycle.

Source code in repyability/non_repairable.py

def failure_frequency(self) -> float:
    """Long-run failure frequency (failures per unit time).

    For an alternating renewal process (fail, repair, fail, ...) this is
    ``1 / (MTTF + MTTR)`` — one failure per mean up-down cycle.
    """
    if isinstance(self.reliability, NonParametric):
        raise ValueError(
            "Failure frequency requires a parametric reliability model"
        )
    mttf = model_mean(self.reliability)
    mttr = model_mean(self.time_to_replace)
    return 1.0 / (mttf + mttr)

optimal_replacement_policy

optimal_replacement_policy()

The optimal age-replacement policy as a typed result.

Returns a MaintenancePolicy whose interval is the cost-optimal replacement age and whose cost_rate is the long-run cost per unit time under it. When preventive replacement never pays (no aging — e.g. an exponential lifetime, or a Weibull with shape <= 1) the interval is inf and the cost rate is the run-to-failure rate cu / MTTF.

Source code in repyability/non_repairable.py

def optimal_replacement_policy(self) -> MaintenancePolicy:
    """The optimal age-replacement policy as a typed result.

    Returns a ``MaintenancePolicy`` whose ``interval`` is the
    cost-optimal replacement age and whose ``cost_rate`` is the
    long-run cost per unit time under it. When preventive replacement
    never pays (no aging — e.g. an exponential lifetime, or a Weibull
    with shape <= 1) the interval is ``inf`` and the cost rate is the
    run-to-failure rate ``cu / MTTF``.
    """
    if not hasattr(self, "cp"):
        raise ValueError(
            "costs not set: call set_costs_planned_and_unplanned"
            "(cp, cu) first"
        )
    interval = self.find_optimal_replacement()
    if np.isinf(interval):
        rate = self.cu / model_mean(self.reliability)
    else:
        rate = float(self._cost_rate(interval))
    return MaintenancePolicy(interval=float(interval), cost_rate=rate)

StandbyModel

StandbyModel(reliabilities, k=1, n_sims=10000, lower=-np.inf, switching_probability=1.0, seed=None)

Source code in repyability/rbd/standby_node.py

def __init__(
    self,
    reliabilities,
    k=1,
    n_sims=10_000,
    lower=-np.inf,
    switching_probability=1.0,
    seed=None,
):
    if k > len(reliabilities):
        raise ValueError(
            "Must be more nodes in the standby arrangement"
            + " than are required (k)"
        )
    self.reliabilities = reliabilities
    self.k = k
    self.N = len(reliabilities)
    self.n_sims = n_sims
    self.switching_probability = switching_probability

    rate = _identical_exponential_rate(reliabilities)
    if rate is not None and is_perfect_switching(switching_probability):
        # Identical exponential units: the cold standby lifetime is exactly
        # Erlang(N-k+1, k*rate) for any k, by the memorylessness of the
        # exponential. Use that closed form directly.
        self._sf_model = _ExponentialStandbySurvival(rate, self.N, k)
        self.model = None
    elif k == 1:
        # Cold standby (k=1): the lifetime is the sum of the components'
        # lifetimes (or a mixture of partial sums under imperfect
        # switching), whose survival function is computed deterministically
        # by numerical convolution rather than from Monte-Carlo samples.
        self._sf_model = ConvolvedSurvival(
            reliabilities, switching_probability=switching_probability
        )
        self.model = None
    else:
        # For k >= 2 with general (non-exponential) lifetimes the lifetime
        # is not a simple sum (it depends on the order in which components
        # fail), so fall back to the Monte-Carlo + Kaplan-Meier
        # approximation. Imperfect switching is not modelled in that case
        # yet.
        if not is_perfect_switching(switching_probability):
            raise NotImplementedError(
                "switching_probability is only supported for k=1 cold"
                " standby; for k>=2 leave it at 1.0 (perfect switching)."
            )
        x_random = self.random(n_sims, seed=seed)
        self.model = KaplanMeier.fit(x_random, set_lower_limit=lower)
        self._sf_model = None

cs

cs(x, X)

Conditional survival R(x | X) = sf(X + x) / sf(X).

Source code in repyability/rbd/standby_node.py

def cs(self, x, X):
    """Conditional survival ``R(x | X) = sf(X + x) / sf(X)``."""
    from repyability.utils.wrappers import conditional_survival

    return conditional_survival(self, x, X)

LoadSharingModel

LoadSharingModel(models, load, k=1, n_sims=10000, lower=-np.inf, seed=None)

A load-sharing arrangement of coupled AFT units as one RBD node.

Parameters:

Name	Type	Description	Default
models	sequence of fitted surpyval AFT models	The units. Each must be an accelerated-failure-time model fitted with the (scalar) load as its covariate, exposing phi(load) and an AFT baseline distribution.	required
load	float	The total load L shared by the active units. Each of s survivors carries L / s.	required
k	int	The minimum number of surviving units for the group to work, by default 1. The group fails at the (N - k + 1)-th unit failure.	1
n_sims	int	Monte-Carlo replicates for the Kaplan-Meier fit when no closed form applies, by default 10_000.	10000
lower	float	set_lower_limit passed to the Kaplan-Meier fit, by default -inf.	-inf
seed	int or None	Seed for the Monte-Carlo fit (reproducible), by default None.	None

Notes

phi with no load effect (phi == 1) reproduces the ordinary k-out-of-n parallel result exactly, since the units then neither share stress nor age faster as siblings fail.

Source code in repyability/rbd/load_sharing_node.py

def __init__(
    self,
    models,
    load,
    k=1,
    n_sims=10_000,
    lower=-np.inf,
    seed=None,
):
    models = list(models)
    if len(models) == 0:
        raise ValueError("LoadSharingModel needs at least one unit.")
    if k < 1:
        raise ValueError(f"k must be >= 1, got {k!r}.")
    if k > len(models):
        raise ValueError(
            f"k ({k}) cannot exceed the number of units ({len(models)})."
        )
    for m in models:
        if getattr(m, "kind", None) != _AFT_KIND:
            raise ValueError(
                "LoadSharingModel units must be fitted accelerated-"
                "failure-time (AFT) models exposing phi(load); got "
                f"{type(m).__name__} with "
                f"kind={getattr(m, 'kind', None)!r}."
            )
    self.models = models
    self.load = float(load)
    self.k = int(k)
    self.N = len(models)
    self.n_sims = n_sims

    self._baselines = [_baseline(m) for m in models]
    # phi_table[i, s-1] = unit i's aging rate when s units share the load.
    self._phi_table = np.array(
        [
            [
                float(np.ravel(m.phi(np.atleast_2d(self.load / s)))[0])
                for s in range(1, self.N + 1)
            ]
            for m in models
        ]
    )

    rates = _identical_exponential_stage_rates(
        models, self.load, self.k, self._baselines, self._phi_table
    )
    # The partial-fraction hypoexponential needs distinct rates; if a load
    # effect collides two stage rates, fall back to the simulation path.
    if rates is not None and _all_distinct(rates):
        self._sf_model: object = _HypoexponentialSurvival(rates)
        self.model = None
    else:
        x_random = self.random(n_sims, seed=seed)
        self.model = KaplanMeier.fit(x_random, set_lower_limit=lower)
        self._sf_model = None

is_simulated property

is_simulated

True if the survival function is a Monte-Carlo (Kaplan-Meier) fit rather than the exact hypoexponential closed form.

random

random(size, seed=None)

Monte-Carlo simulate size group lifetimes via the cumulative- exposure event loop.

Source code in repyability/rbd/load_sharing_node.py

def random(self, size, seed=None):
    """Monte-Carlo simulate ``size`` group lifetimes via the cumulative-
    exposure event loop."""
    n, k, phi_tab = self.N, self.k, self._phi_table
    with numpy_seed(seed):
        # Baseline exposure-to-failure thresholds: (N, size).
        tau = np.empty((n, size))
        for i, base in enumerate(self._baselines):
            tau[i] = np.asarray(base.random(size), dtype=float).reshape(-1)

        out = np.empty(size)
        for j in range(size):
            thr = tau[:, j]
            e = np.zeros(n)
            alive = np.ones(n, dtype=bool)
            t = 0.0
            s = n
            while s >= k:
                idx = np.flatnonzero(alive)
                phi = phi_tab[idx, s - 1]
                remaining = (thr[idx] - e[idx]) / phi
                w = int(np.argmin(remaining))
                dt = remaining[w]
                t += dt
                e[idx] += phi * dt
                alive[idx[w]] = False
                s -= 1
            out[j] = t
    return out

cs

cs(x, X)

Conditional survival R(x | X) = sf(X + x) / sf(X).

Source code in repyability/rbd/load_sharing_node.py

def cs(self, x, X):
    """Conditional survival ``R(x | X) = sf(X + x) / sf(X)``."""
    return conditional_survival(self, x, X)

RepeatedNode

RepeatedNode(model, repeats, kind)

Source code in repyability/rbd/repeated_node.py

def __init__(self, model, repeats, kind):
    if kind not in REPEATED_NODE_TYPES:
        raise ValueError("'kind' must be either 'parallel' or 'series'")

    self.model = model
    if kind == "parallel":
        self.kind = PARALLEL
    else:
        self.kind = SERIES
    self.repeats = repeats

cs

cs(x, X)

Conditional survival R(x | X) = sf(X + x) / sf(X).

Source code in repyability/rbd/repeated_node.py

def cs(self, x, X):
    """Conditional survival ``R(x | X) = sf(X + x) / sf(X)``."""
    from repyability.utils.wrappers import conditional_survival

    return conditional_survival(self, x, X)

RepeatedStandbyNode

RepeatedStandbyNode(model, repeats, N=10000, lower=-np.inf, switching_probability=1.0)

Source code in repyability/rbd/repeated_standby_node.py

def __init__(
    self,
    model,
    repeats,
    N=10_000,
    lower=-np.inf,
    switching_probability=1.0,
):
    # N and lower are kept for backwards compatibility (a Kaplan-Meier fit
    # was previously made here); they are no longer used.
    self.model = model
    self.repeats = repeats
    self.switching_probability = switching_probability

    # Repeated cold standby: the lifetime is the sum of `repeats`
    # independent copies of `model` (a mixture of partial sums under
    # imperfect switching). Its survival function is computed
    # deterministically by numerical convolution.
    self._sf_model = ConvolvedSurvival(
        [model] * repeats, switching_probability=switching_probability
    )

cs

cs(x, X)

Conditional survival R(x | X) = sf(X + x) / sf(X).

Source code in repyability/rbd/repeated_standby_node.py

def cs(self, x, X):
    """Conditional survival ``R(x | X) = sf(X + x) / sf(X)``."""
    from repyability.utils.wrappers import conditional_survival

    return conditional_survival(self, x, X)

Repairable

Repairable(model)

Repairable component with an optimal overhaul/replacement policy.

Parameters:

Name	Type	Description	Default
model	object	A recurrent-event model exposing the expected number of failures by age t, E[N(t)], as either: cif(t) — an analytic cumulative intensity (minimal repair), e.g. a surpyval CrowAMSAA/Duane/HPP (fitted or from params); or mcf(t, items=..., seed=...) — a simulation-estimated mean cumulative function (imperfect repair), e.g. a fitted surpyval GeneralizedRenewal (Kijima I/II). If both are present cif is used (analytic, exact).	required

Source code in repyability/repairable.py

def __init__(self, model):
    self._analytic = hasattr(model, "cif")
    if not (self._analytic or hasattr(model, "mcf")):
        raise ValueError(
            "model must expose cif() (analytic cumulative intensity, e.g. "
            "a surpyval recurrence model such as CrowAMSAA) or mcf() (a "
            "simulation-estimated mean cumulative function, e.g. a fitted "
            "GeneralizedRenewal)"
        )
    self.model = model

is_simulated property

is_simulated

True if E[N(t)] is estimated by simulation (imperfect repair), so the policy methods honour seed/n_simulations.

set_repair_and_overhaul_costs

set_repair_and_overhaul_costs(cr, co)

Set the repair cost cr and overhaul/replacement cost co.

Requires 0 < cr < co: an overhaul/replacement (a full renewal) must cost more than a repair, otherwise one would simply renew at every failure.

Source code in repyability/repairable.py

def set_repair_and_overhaul_costs(self, cr: float, co: float) -> None:
    """Set the repair cost ``cr`` and overhaul/replacement cost ``co``.

    Requires ``0 < cr < co``: an overhaul/replacement (a full renewal) must
    cost more than a repair, otherwise one would simply renew at every
    failure.
    """
    if cr <= 0:
        raise ValueError("repair cost, cr, must be positive.")
    if cr >= co:
        raise ValueError(
            "repair cost, cr, must be less than overhaul cost, co."
        )
    self.cr = cr
    self.co = co

cost

cost(t, seed=None, n_simulations=_DEFAULT_N_SIMULATIONS)

Expected cost of one overhaul/replacement cycle of length t: cr * E[N(t)] + co.

Scalar t returns a float; array t returns an array. seed/n_simulations apply only to a simulation-backed (imperfect-repair) model.

Source code in repyability/repairable.py

def cost(
    self,
    t,
    seed: Optional[int] = None,
    n_simulations: int = _DEFAULT_N_SIMULATIONS,
) -> Union[float, np.ndarray]:
    """Expected cost of one overhaul/replacement cycle of length ``t``:
    ``cr * E[N(t)] + co``.

    Scalar ``t`` returns a float; array ``t`` returns an array.
    ``seed``/``n_simulations`` apply only to a simulation-backed
    (imperfect-repair) model.
    """
    self._require_costs()
    scalar_in = np.ndim(t) == 0
    tt = np.atleast_1d(np.asarray(t, dtype=float))
    lam = self._expected_failures(tt, seed, n_simulations)
    out = self.cr * lam + self.co
    return out.item() if scalar_in else out

cost_rate

cost_rate(t, seed=None, n_simulations=_DEFAULT_N_SIMULATIONS)

Long-run cost per unit time when renewing every t: (cr * E[N(t)] + co) / t.

Scalar t returns a float; array t returns an array. seed/n_simulations apply only to a simulation-backed (imperfect-repair) model.

Source code in repyability/repairable.py

def cost_rate(
    self,
    t,
    seed: Optional[int] = None,
    n_simulations: int = _DEFAULT_N_SIMULATIONS,
) -> Union[float, np.ndarray]:
    """Long-run cost per unit time when renewing every ``t``:
    ``(cr * E[N(t)] + co) / t``.

    Scalar ``t`` returns a float; array ``t`` returns an array.
    ``seed``/``n_simulations`` apply only to a simulation-backed
    (imperfect-repair) model.
    """
    self._require_costs()
    scalar_in = np.ndim(t) == 0
    tt = np.atleast_1d(np.asarray(t, dtype=float))
    lam = self._expected_failures(tt, seed, n_simulations)
    with np.errstate(divide="ignore"):
        out = (self.cr * lam + self.co) / tt
    return out.item() if scalar_in else out

find_optimal_overhaul_interval

find_optimal_overhaul_interval(seed=None, n_simulations=_DEFAULT_N_SIMULATIONS, max_interval=None)

The overhaul/replacement interval minimising the long-run cost rate.

For an analytic (minimal-repair) model, returns inf when renewal never pays: the unit does not wear out (E[N(t)] grows at most linearly — e.g. HPP, or Crow-AMSAA with beta <= 1), so the cost rate keeps falling as the interval grows.

For a simulation-backed (imperfect-repair) model, pass a seed for a reproducible result; n_simulations sets the Monte-Carlo sample size and max_interval the search horizon (default ~15x the baseline mean-time-to-first-failure).

Source code in repyability/repairable.py

def find_optimal_overhaul_interval(
    self,
    seed: Optional[int] = None,
    n_simulations: int = _DEFAULT_N_SIMULATIONS,
    max_interval: Optional[float] = None,
) -> float:
    """The overhaul/replacement interval minimising the long-run cost rate.

    For an analytic (minimal-repair) model, returns ``inf`` when renewal
    never pays: the unit does not wear out (``E[N(t)]`` grows at most
    linearly — e.g. HPP, or Crow-AMSAA with ``beta <= 1``), so the cost
    rate keeps falling as the interval grows.

    For a simulation-backed (imperfect-repair) model, pass a ``seed`` for a
    reproducible result; ``n_simulations`` sets the Monte-Carlo sample size
    and ``max_interval`` the search horizon (default ~15x the baseline
    mean-time-to-first-failure).
    """
    return self._optimise(seed, n_simulations, max_interval)[0]

optimal_overhaul_policy

optimal_overhaul_policy(seed=None, n_simulations=_DEFAULT_N_SIMULATIONS, max_interval=None)

The optimal overhaul/replacement policy as a typed result.

Returns a MaintenancePolicy carrying the optimal interval and the long-run cost rate under it. For an analytic model an inf interval (never renew) reports the limiting rate of repairs alone.

For a simulation-backed (imperfect-repair) model, pass a seed for a reproducible result; n_simulations sets the Monte-Carlo sample size and max_interval the search horizon.

Source code in repyability/repairable.py

def optimal_overhaul_policy(
    self,
    seed: Optional[int] = None,
    n_simulations: int = _DEFAULT_N_SIMULATIONS,
    max_interval: Optional[float] = None,
) -> MaintenancePolicy:
    """The optimal overhaul/replacement policy as a typed result.

    Returns a ``MaintenancePolicy`` carrying the optimal interval and the
    long-run cost rate under it. For an analytic model an ``inf`` interval
    (never renew) reports the limiting rate of repairs alone.

    For a simulation-backed (imperfect-repair) model, pass a ``seed`` for a
    reproducible result; ``n_simulations`` sets the Monte-Carlo sample size
    and ``max_interval`` the search horizon.
    """
    interval, rate = self._optimise(seed, n_simulations, max_interval)
    return MaintenancePolicy(interval=interval, cost_rate=rate)

expected_time_to_nth_failure

expected_time_to_nth_failure(n, seed=None, n_simulations=_DEFAULT_N_SIMULATIONS)

Expected time to the n-th failure, E[T_n].

Estimated by a seeded simulation for an imperfect-repair model. For a minimal-repair (power-law) process the closed form :func:minimal_repair_time_to_nth_failure is exact and cheaper.

Source code in repyability/repairable.py

def expected_time_to_nth_failure(
    self,
    n: int,
    seed: Optional[int] = None,
    n_simulations: int = _DEFAULT_N_SIMULATIONS,
) -> float:
    """Expected time to the ``n``-th failure, ``E[T_n]``.

    Estimated by a seeded simulation for an imperfect-repair model. For a
    minimal-repair (power-law) process the closed form
    :func:`minimal_repair_time_to_nth_failure` is exact and cheaper.
    """
    if n < 1:
        raise ValueError("n must be a positive integer.")
    curve = self._expected_times_to_failures(n, seed, n_simulations)
    if len(curve) < n:
        raise ValueError(
            f"the simulator could not reach {n} failures for this model "
            "(near-minimal repair); use "
            "minimal_repair_time_to_nth_failure() instead."
        )
    return float(curve[n - 1])

find_optimal_replacement_failure_count

find_optimal_replacement_failure_count(seed=None, n_simulations=_DEFAULT_N_SIMULATIONS, max_failures=_DEFAULT_MAX_FAILURES)

The number of failures per cycle minimising the long-run cost rate of a replace-at-N-th-failure policy.

Simulation-based; pass a seed for a reproducible result. max_failures bounds the search (raise it if the optimum sits at the bound).

Source code in repyability/repairable.py

def find_optimal_replacement_failure_count(
    self,
    seed: Optional[int] = None,
    n_simulations: int = _DEFAULT_N_SIMULATIONS,
    max_failures: int = _DEFAULT_MAX_FAILURES,
) -> int:
    """The number of failures per cycle minimising the long-run cost rate
    of a replace-at-N-th-failure policy.

    Simulation-based; pass a ``seed`` for a reproducible result.
    ``max_failures`` bounds the search (raise it if the optimum sits at the
    bound).
    """
    return self._optimise_failure_limit(seed, n_simulations, max_failures)[
        0
    ]

optimal_failure_limit_policy

optimal_failure_limit_policy(seed=None, n_simulations=_DEFAULT_N_SIMULATIONS, max_failures=_DEFAULT_MAX_FAILURES)

The optimal replace-at-N-th-failure policy as a typed result.

Repair on each failure (cost cr) and replace on the failure_count-th (cost co); the long-run cost rate is (cr*(n-1) + co) / E[T_n] minimised over n.

Source code in repyability/repairable.py

def optimal_failure_limit_policy(
    self,
    seed: Optional[int] = None,
    n_simulations: int = _DEFAULT_N_SIMULATIONS,
    max_failures: int = _DEFAULT_MAX_FAILURES,
) -> FailureLimitPolicy:
    """The optimal replace-at-N-th-failure policy as a typed result.

    Repair on each failure (cost ``cr``) and replace on the
    ``failure_count``-th (cost ``co``); the long-run cost rate is
    ``(cr*(n-1) + co) / E[T_n]`` minimised over ``n``.
    """
    count, rate = self._optimise_failure_limit(
        seed, n_simulations, max_failures
    )
    return FailureLimitPolicy(failure_count=count, cost_rate=rate)

RegressionNode

RegressionNode(model, covariates)

An RBD node backed by a fitted surpyval regression model at fixed covariates.

Parameters:

Name	Type	Description	Default
model	surpyval regression model	A fitted regression model whose sf(x, Z) gives survival at a covariate matrix Z (e.g. surpyval.WeibullAFT.fit(...), surpyval.CoxPH.fit(...)).	required
covariates	array_like	The component's covariate vector Z (the operating conditions), matching the covariates the model was fitted with.	required

Examples:

>>> import numpy as np
>>> import surpyval as surv
>>> from repyability.rbd.regression_node import RegressionNode
>>> rng = np.random.default_rng(0)
>>> Z = rng.normal(size=(300, 1))
>>> x = rng.weibull(2.0, size=300) * 100 + 1e-3
>>> model = surv.WeibullAFT.fit(x, Z=Z)
>>> node = RegressionNode(model, covariates=[0.5])
>>> bool(0.0 < node.sf(np.array([50.0]))[0] < 1.0)
True

Source code in repyability/rbd/regression_node.py

def __init__(self, model: Any, covariates: ArrayLike):
    self.model = model
    self.covariates = np.atleast_1d(np.asarray(covariates, dtype=float))
    # Probe the regression interface so a misuse (a non-regression model,
    # or covariates of the wrong width) fails clearly at construction.
    try:
        probe = self._sf_at(np.array([1.0]))
        if not np.all(np.isfinite(probe)):
            raise ValueError("sf(x, Z) returned non-finite values")
    except Exception as e:
        raise ValueError(
            "RegressionNode requires a fitted surpyval regression model "
            "whose sf(x, Z) accepts a covariate matrix, and covariates "
            f"matching the model's fitted width. Probing sf failed: "
            f"{type(e).__name__}: {e}."
        ) from e
    # Cached (t, sf(t)) grid for mean()/random() (built lazily).
    self._grid: Any = None

sf

sf(x)

Reliability at the stored covariates: model.sf(x, Z).

Source code in repyability/rbd/regression_node.py

def sf(self, x: ArrayLike) -> np.ndarray:
    """Reliability at the stored covariates: ``model.sf(x, Z)``."""
    return self._sf_at(np.atleast_1d(np.asarray(x, dtype=float)))

ff

ff(x)

Unreliability at the stored covariates: 1 - sf(x).

Source code in repyability/rbd/regression_node.py

def ff(self, x: ArrayLike) -> np.ndarray:
    """Unreliability at the stored covariates: ``1 - sf(x)``."""
    return 1.0 - self.sf(x)

mean

mean()

Mean time to failure at the stored covariates.

For a non-negative lifetime E[T] = integral of R(t), integrated numerically over the survival curve.

Source code in repyability/rbd/regression_node.py

def mean(self) -> float:
    """Mean time to failure at the stored covariates.

    For a non-negative lifetime ``E[T] = integral of R(t)``, integrated
    numerically over the survival curve.
    """
    t, s = self._survival_grid()
    return float(np.trapezoid(s, t))

random

random(size)

Draw size failure times at the stored covariates.

Inverse-transform sampling on the survival curve. Uses numpy's global RNG, so wrap the call in :func:~repyability.utils.wrappers.numpy_seed to reproduce.

Source code in repyability/rbd/regression_node.py

def random(self, size: int) -> np.ndarray:
    """Draw ``size`` failure times at the stored covariates.

    Inverse-transform sampling on the survival curve. Uses numpy's global
    RNG, so wrap the call in
    :func:`~repyability.utils.wrappers.numpy_seed` to reproduce.
    """
    t, s = self._survival_grid()
    u = np.random.uniform(size=size)
    # s decreases in t; np.interp needs an increasing sample-point array.
    return np.interp(u, s[::-1], t[::-1])

to_dict

to_dict()

Serialise to a JSON-friendly dict (the fitted model + covariates). See :meth:from_dict.

Source code in repyability/rbd/regression_node.py

def to_dict(self) -> dict:
    """Serialise to a JSON-friendly dict (the fitted model + covariates).
    See :meth:`from_dict`."""
    return {
        "model": self.model.to_dict(),
        "covariates": [float(v) for v in self.covariates],
    }

from_dict classmethod

from_dict(d)

Reconstruct from :meth:to_dict (the fitted model round-trips through surpyval.from_dict).

Source code in repyability/rbd/regression_node.py

@classmethod
def from_dict(cls, d: dict) -> "RegressionNode":
    """Reconstruct from :meth:`to_dict` (the fitted model round-trips
    through ``surpyval.from_dict``)."""
    import surpyval

    return cls(surpyval.from_dict(d["model"]), d["covariates"])

Helpers

NodeState dataclass

NodeState(age=0.0, alive=True)

The current condition of a single RBD node.

A component's forward reliability is conditioned on the life it has already survived: R_i(x | age) = R_i(age + x) / R_i(age). A failed component (alive=False) contributes zero reliability regardless of its age.

Parameters:

Name	Type	Description	Default
age	float	The component's current age / accumulated operating time (>= 0), by default 0.0 (a brand-new component, equivalent to no conditioning).	0.0
alive	bool	Whether the component is currently working, by default True.	True

Notes

Only lifetime (time-varying) distributions age; a fixed-probability component's reliability does not depend on age. Covariate/load dependence is a property of the component, not its transient state: a :class:~repyability.rbd.regression_node.RegressionNode stores the covariates, so age keeps a single meaning (operating time) for every node type.

CCFGroup

CCFGroup(members, model)

A common-cause group: member nodes coupled by a shared failure cause.

Parameters:

Name	Type	Description	Default
members	collection of node names	The RBD nodes that share the common cause (at least two, distinct). Standard CCF theory is for symmetric groups, so the members should carry identical component models.	required
model	BetaFactor or MGL	The common-cause model coupling the members. An :class:MGL model fixes the group size (m - 1 letters for m members).	required

Source code in repyability/rbd/ccf.py

def __init__(self, members: Collection[Hashable], model: Any):
    members = tuple(members)
    if len(members) < 2:
        raise ValueError(
            f"A CCF group needs at least 2 members, got {len(members)}."
        )
    if len(set(members)) != len(members):
        raise ValueError(
            f"CCF group members must be distinct, got {list(members)}."
        )
    if not isinstance(model, (BetaFactor, MGL)):
        raise ValueError(
            "CCFGroup model must be a BetaFactor or MGL; alpha-factor is "
            "not supported yet."
        )
    required = model.required_group_size()
    if required is not None and required != len(members):
        raise ValueError(
            f"{type(model).__name__} describes a group of {required} "
            f"members, but this group has {len(members)}."
        )
    self.members = members
    self.model = model

BetaFactor

BetaFactor(beta)

The beta-factor common-cause model (all-or-nothing).

A fraction beta of each component's total failure probability is attributed to a cause shared across the whole group (which fails every member simultaneously); the remaining 1 - beta is independent.

Parameters:

Name	Type	Description	Default
beta	float	The common-cause fraction, in [0, 1]. beta = 0 is ordinary independence; beta = 1 makes the group fail entirely in unison.	required

Source code in repyability/rbd/ccf.py

def __init__(self, beta: float):
    if not (0.0 <= beta <= 1.0):
        raise ValueError(f"beta must be in [0, 1], got {beta!r}.")
    self.beta = float(beta)

MGL

MGL(*letters)

The Multiple Greek Letter common-cause model.

The parameters are the conditional probabilities of a common-cause failure escalating to the next level: beta = P(shared by >= 2 | failed), gamma = P(>= 3 | >= 2), delta = P(>= 4 | >= 3), and so on. The number of parameters fixes the group size: n letters describe a group of n + 1 members. MGL(beta) is exactly :class:BetaFactor on a two-member group.

The probability that a common cause fails a specific set of k of the m members is the standard MGL basic-event probability

Q_k = [1 / C(m-1, k-1)] * (rho_1 * ... * rho_k) * (1 - rho_{k+1}) * Q

with rho_1 = 1, rho_2 = beta, rho_3 = gamma, ..., and rho_{m+1} = 0. These partition each component's total failure probability Q exactly.

Parameters:

Name	Type	Description	Default
*letters	float	beta, gamma, delta, ..., each in [0, 1]; at least one. A group of m members needs m - 1 letters.	()

Source code in repyability/rbd/ccf.py

def __init__(self, *letters: float):
    if len(letters) < 1:
        raise ValueError("MGL needs at least one parameter (beta).")
    for value in letters:
        if not (0.0 <= value <= 1.0):
            raise ValueError(
                f"MGL parameters must be in [0, 1], got {value!r}."
            )
    self.letters = tuple(float(v) for v in letters)

PerfectReliability

PerfectUnreliability

Result types

AvailabilityResult dataclass

AvailabilityResult(timeline, availability, system_uptime, time_simulated_to, criticalities, node_uptime, node_downtime, system_downtime, system_failures, system_restorations, n_simulations)

Bases: _ResultMapping

The result of RepairableRBD.availability().

The properties mean_up_time, mean_down_time and failure_frequency are simulation estimates derived from the fields below; their exact steady-state counterparts are the RepairableRBD.mean_up_time(), mean_down_time() and system_failure_frequency() methods.

Attributes:

Name	Type	Description
timeline	ndarray	Event times at which the (mean) system availability changes.
availability	ndarray	Mean system availability at each time in timeline.
system_uptime	float	Total system uptime summed over all simulations.
time_simulated_to	float	The t_simulation the simulations were run to.
criticalities	Criticalities	The importance/criticality measures (see Criticalities).
node_uptime	dict	Total uptime per node summed over all simulations.
node_downtime	dict	Total downtime per node summed over all simulations; a node's uptime plus downtime equals n_simulations * time_simulated_to.
system_downtime	float	Total system downtime summed over all simulations.
system_failures	int	Number of system failures observed across all simulations.
system_restorations	int	Number of system restorations observed across all simulations.
n_simulations	int	The number of simulations run (N).

availability_se property

availability_se

Pointwise standard error of the availability estimate.

At each time in timeline the availability is the proportion of the n_simulations systems that were up, so its sampling standard error is the binomial sqrt(A (1 - A) / n).

mean_up_time property

mean_up_time

Simulation estimate of the Mean Up Time, system uptime / system failures. Infinite if no failure was observed.

Note: estimated from a finite window, so each simulation's final (unfinished) up period is censored; for windows that are short relative to the up-down cycle this biases the estimate. Prefer the exact RepairableRBD.mean_up_time() for steady-state values.

mean_down_time property

mean_down_time

Simulation estimate of the Mean Down Time, system downtime / system restorations. Infinite if downtime was observed but never restored.

Note: estimated from a finite window (final unfinished down periods are censored), so short windows bias the estimate. Prefer the exact RepairableRBD.mean_down_time() for steady-state values.

failure_frequency property

failure_frequency

Simulation estimate of the system failure frequency (failures per unit time), system failures / total simulated time.

availability_interval

availability_interval(confidence=0.95)

Pointwise confidence band for the availability curve.

Uses the Wilson score interval for a binomial proportion, which remains well-behaved when the estimated availability is at or near 0 or 1 (where the plain normal interval collapses to zero width).

Parameters:

Name	Type	Description	Default
confidence	float	The confidence level, by default 0.95.	0.95

Returns:

Type	Description
(ndarray, ndarray)	The lower and upper bounds at each time in timeline.

Source code in repyability/rbd/results.py

def availability_interval(
    self, confidence: float = 0.95
) -> Tuple[np.ndarray, np.ndarray]:
    """Pointwise confidence band for the availability curve.

    Uses the Wilson score interval for a binomial proportion, which
    remains well-behaved when the estimated availability is at or near 0
    or 1 (where the plain normal interval collapses to zero width).

    Parameters
    ----------
    confidence : float, optional
        The confidence level, by default 0.95.

    Returns
    -------
    (numpy.ndarray, numpy.ndarray)
        The lower and upper bounds at each time in ``timeline``.
    """
    if not 0.0 < confidence < 1.0:
        raise ValueError("confidence must be between 0 and 1.")
    z = float(norm.ppf(0.5 + confidence / 2.0))
    p = np.asarray(self.availability, dtype=float)
    n = self.n_simulations
    denominator = 1.0 + z**2 / n
    centre = (p + z**2 / (2.0 * n)) / denominator
    half_width = (z / denominator) * np.sqrt(
        p * (1.0 - p) / n + z**2 / (4.0 * n**2)
    )
    lower = np.clip(centre - half_width, 0.0, 1.0)
    upper = np.clip(centre + half_width, 0.0, 1.0)
    return lower, upper

ConfidenceInterval dataclass

ConfidenceInterval(estimate, lower, upper, confidence, standard_error, n_samples)

Bases: _ResultMapping

A Monte-Carlo estimate with its sampling uncertainty.

Attributes:

Name	Type	Description
estimate	float	The point estimate (the sample mean).
lower	float	Lower bound of the confidence interval.
upper	float	Upper bound of the confidence interval.
confidence	float	The confidence level the bounds correspond to (e.g. 0.95).
standard_error	float	The standard error of the estimate.
n_samples	int	The number of Monte-Carlo samples the estimate was computed from.

Criticalities dataclass

Criticalities(operational_criticality_index, iou, failure_criticality_index, restoration_criticality_index)

Bases: _ResultMapping

The importance/criticality measures from an availability simulation.

Attributes:

Name	Type	Description
operational_criticality_index	UpDownImportance	Time the system and a node were jointly up/down, over the system's up/down time.
iou	UpDownImportance	Intersection-over-union of a node's and the system's up/down intervals.
failure_criticality_index	FailureCriticalityIndex	How much each node drives system failures.
restoration_criticality_index	RestorationCriticalityIndex	How much each node drives system restorations.

UpDownImportance dataclass

UpDownImportance(up, down)

Bases: _ResultMapping

An importance measure split by system state.

Attributes:

Name	Type	Description
up	dict	Per-node measure while the system is up.
down	dict	Per-node measure while the system is down.

FailureCriticalityIndex dataclass

FailureCriticalityIndex(per_system_failure, per_component_failure)

Bases: _ResultMapping

Fractions relating a node's failures to system failures.

Attributes:

Name	Type	Description
per_system_failure	dict	For each node, the fraction of system failures that the node caused.
per_component_failure	dict	For each node, the fraction of the node's own failures that caused a system failure.

RestorationCriticalityIndex dataclass

RestorationCriticalityIndex(by_system, by_component)

Bases: _ResultMapping

Fractions relating a node's restorations to system restorations.

Attributes:

Name	Type	Description
by_system	dict	For each node, the fraction of system restorations that the node caused.
by_component	dict	For each node, the fraction of the node's own restorations that restored the system.

MaintenancePolicy dataclass

MaintenancePolicy(interval, cost_rate)

An optimal preventive-maintenance policy for a single component.

Returned by NonRepairable.optimal_replacement_policy() (age replacement) and Repairable.optimal_overhaul_policy() (overhaul under minimal repair).

Attributes:

Name	Type	Description
interval	float	The optimal preventive interval: the replacement age for an age-replacement policy, or the overhaul interval for a minimal-repair component. inf when preventive action never pays (run to failure / never overhaul).
cost_rate	float	The long-run cost per unit time under the policy. When interval is inf this is the limiting cost rate of running without preventive action.