Skip to content

Index

Exposes interfaces and enumerations for the Beachlore Safety framework.

BlockInterface

Bases: ABC

Abstract interface defining the mandatory structure for all logic blocks.

Every block in the system must implement this interface to ensure modularity and nesting capabilities within the safety analysis.

Source code in src/ecc_analyzer/interfaces/block_interface.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
class BlockInterface(ABC):
    """Abstract interface defining the mandatory structure for all logic blocks.

    Every block in the system must implement this interface to ensure modularity
    and nesting capabilities within the safety analysis.
    """

    @abstractmethod
    def compute_fit(self, spfm_rates: dict[FaultType, float], lfm_rates: dict[FaultType, float]) -> tuple[dict[FaultType, float], dict[FaultType, float]]:
        """Transforms the input fault rate dictionaries according to the block's specific logic.

        Args:
            spfm_rates (dict[FaultType, float]): Dictionary mapping fault types to
                their current residual (SPFM) failure rates (FIT).
            lfm_rates (dict[FaultType, float]): Dictionary mapping fault types to
                their current latent (LFM) failure rates (FIT).

        Returns:
            tuple[dict[FaultType, float], dict[FaultType, float]]: A tuple containing:
                - Updated SPFM rates dictionary.
                - Updated LFM rates dictionary.
        """
        pass

compute_fit(spfm_rates, lfm_rates) abstractmethod

Transforms the input fault rate dictionaries according to the block's specific logic.

Parameters:

Name Type Description Default
spfm_rates dict[FaultType, float]

Dictionary mapping fault types to their current residual (SPFM) failure rates (FIT).

 required
lfm_rates dict[FaultType, float]

Dictionary mapping fault types to their current latent (LFM) failure rates (FIT).

 required

Returns:

Type Description
tuple[dict[FaultType, float], dict[FaultType, float]]

tuple[dict[FaultType, float], dict[FaultType, float]]: A tuple containing: - Updated SPFM rates dictionary. - Updated LFM rates dictionary.
Source code in src/ecc_analyzer/interfaces/block_interface.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
@abstractmethod
def compute_fit(self, spfm_rates: dict[FaultType, float], lfm_rates: dict[FaultType, float]) -> tuple[dict[FaultType, float], dict[FaultType, float]]:
    """Transforms the input fault rate dictionaries according to the block's specific logic.

    Args:
        spfm_rates (dict[FaultType, float]): Dictionary mapping fault types to
            their current residual (SPFM) failure rates (FIT).
        lfm_rates (dict[FaultType, float]): Dictionary mapping fault types to
            their current latent (LFM) failure rates (FIT).

    Returns:
        tuple[dict[FaultType, float], dict[FaultType, float]]: A tuple containing:
            - Updated SPFM rates dictionary.
            - Updated LFM rates dictionary.
    """
    pass

FaultType

Bases: Enum

Enumeration of supported fault types within the safety model. Each member represents a specific failure mode used for FIT rate calculations and visualization.

Source code in src/ecc_analyzer/interfaces/fault_type.py 
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
class FaultType(Enum):
    """
    Enumeration of supported fault types within the safety model.
    Each member represents a specific failure mode used for FIT rate calculations and visualization.
    """

    SBE = auto()
    DBE = auto()
    TBE = auto()
    MBE = auto()
    WD = auto()
    AZ = auto()
    SB = auto()
    SDB = auto()
    OTH = auto()
    SBE_IF = auto()

ObservableInterface

Bases: ABC

Abstract interface for objects that need to be monitored by a SafetyObserver.

It defines the mechanism for attaching observers and broadcasting calculation results.

Source code in src/ecc_analyzer/interfaces/observable_interface.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
class ObservableInterface(ABC):
    """Abstract interface for objects that need to be monitored by a SafetyObserver.

    It defines the mechanism for attaching observers and broadcasting calculation results.
    """

    @abstractmethod
    def attach(self, observer: SafetyObserver):
        """Registers a listener (observer) to receive notifications upon computation events.

        Args:
            observer (SafetyObserver): The SafetyObserver instance to be registered.
        """
        pass

    @abstractmethod
    def notify(
        self,
        input_ports: dict,
        spfm_in: dict[FaultType, float],
        lfm_in: dict[FaultType, float],
        spfm_out: dict[FaultType, float],
        lfm_out: dict[FaultType, float],
    ) -> dict:
        """Broadcasts the computation results and visual context to all registered observers.

        Args:
            input_ports (dict): Mapping of fault types to incoming node IDs for visualization.
            spfm_in (dict[FaultType, float]): Incoming residual/SPFM fault rates.
            lfm_in (dict[FaultType, float]): Incoming latent/LFM fault rates.
            spfm_out (dict[FaultType, float]): Transformed outgoing SPFM fault rates.
            lfm_out (dict[FaultType, float]): Transformed outgoing LFM fault rates.

        Returns:
            dict: The visual output ports generated by the observers.
        """
        pass

attach(observer) abstractmethod

Registers a listener (observer) to receive notifications upon computation events.

Parameters:

Name Type Description Default
observer SafetyObserver

The SafetyObserver instance to be registered.

 required
Source code in src/ecc_analyzer/interfaces/observable_interface.py 
17
18
19
20
21
22
23
24
@abstractmethod
def attach(self, observer: SafetyObserver):
    """Registers a listener (observer) to receive notifications upon computation events.

    Args:
        observer (SafetyObserver): The SafetyObserver instance to be registered.
    """
    pass

notify(input_ports, spfm_in, lfm_in, spfm_out, lfm_out) abstractmethod

Broadcasts the computation results and visual context to all registered observers.

Parameters:

Name Type Description Default
input_ports dict

Mapping of fault types to incoming node IDs for visualization.

 required
spfm_in dict[FaultType, float]

Incoming residual/SPFM fault rates.

 required
lfm_in dict[FaultType, float]

Incoming latent/LFM fault rates.

 required
spfm_out dict[FaultType, float]

Transformed outgoing SPFM fault rates.

 required
lfm_out dict[FaultType, float]

Transformed outgoing LFM fault rates.

 required

Returns:

Name Type Description
dict dict

The visual output ports generated by the observers.
Source code in src/ecc_analyzer/interfaces/observable_interface.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
@abstractmethod
def notify(
    self,
    input_ports: dict,
    spfm_in: dict[FaultType, float],
    lfm_in: dict[FaultType, float],
    spfm_out: dict[FaultType, float],
    lfm_out: dict[FaultType, float],
) -> dict:
    """Broadcasts the computation results and visual context to all registered observers.

    Args:
        input_ports (dict): Mapping of fault types to incoming node IDs for visualization.
        spfm_in (dict[FaultType, float]): Incoming residual/SPFM fault rates.
        lfm_in (dict[FaultType, float]): Incoming latent/LFM fault rates.
        spfm_out (dict[FaultType, float]): Transformed outgoing SPFM fault rates.
        lfm_out (dict[FaultType, float]): Transformed outgoing LFM fault rates.

    Returns:
        dict: The visual output ports generated by the observers.
    """
    pass

SafetyObserver

Bases: ABC

Abstract base class for all observers in the safety system.

Allows the separation of calculation logic (Model) from visualization or reporting (View), implementing the Observer design pattern.

Source code in src/ecc_analyzer/interfaces/observer.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
class SafetyObserver(ABC):
    """Abstract base class for all observers in the safety system.

    Allows the separation of calculation logic (Model) from visualization or
    reporting (View), implementing the Observer design pattern.
    """

    @abstractmethod
    def on_block_computed(
        self,
        block: "BlockInterface",
        input_ports: dict,
        spfm_in: dict[FaultType, float],
        lfm_in: dict[FaultType, float],
        spfm_out: dict[FaultType, float],
        lfm_out: dict[FaultType, float],
    ) -> dict:
        """Triggered after a hardware block completes its FIT rate transformation.

        The observer acts upon this event (e.g., drawing the block in a diagram)
        and returns the new visual connection points (ports).

        Args:
            block (BlockInterface): The instance of the logic block (defines shape and type).
            input_ports (dict): Mapping of fault types to incoming node IDs (defines edge origins).
            spfm_in (dict[FaultType, float]): Dictionary of incoming residual/SPFM FIT rates.
            lfm_in (dict[FaultType, float]): Dictionary of incoming latent/LFM FIT rates.
            spfm_out (dict[FaultType, float]): Updated dictionary of outgoing residual/SPFM FIT rates.
            lfm_out (dict[FaultType, float]): Updated dictionary of outgoing latent/LFM FIT rates.

        Returns:
            dict: A dictionary containing the newly created output ports (node IDs) to be
            used as inputs for the next block in the chain.
        """
        pass

on_block_computed(block, input_ports, spfm_in, lfm_in, spfm_out, lfm_out) abstractmethod

Triggered after a hardware block completes its FIT rate transformation.

The observer acts upon this event (e.g., drawing the block in a diagram) and returns the new visual connection points (ports).

Parameters:

Name Type Description Default
block BlockInterface

The instance of the logic block (defines shape and type).

 required
input_ports dict

Mapping of fault types to incoming node IDs (defines edge origins).

 required
spfm_in dict[FaultType, float]

Dictionary of incoming residual/SPFM FIT rates.

 required
lfm_in dict[FaultType, float]

Dictionary of incoming latent/LFM FIT rates.

 required
spfm_out dict[FaultType, float]

Updated dictionary of outgoing residual/SPFM FIT rates.

 required
lfm_out dict[FaultType, float]

Updated dictionary of outgoing latent/LFM FIT rates.

 required

Returns:

Name Type Description
dict dict

A dictionary containing the newly created output ports (node IDs) to be
dict

used as inputs for the next block in the chain.
Source code in src/ecc_analyzer/interfaces/observer.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
@abstractmethod
def on_block_computed(
    self,
    block: "BlockInterface",
    input_ports: dict,
    spfm_in: dict[FaultType, float],
    lfm_in: dict[FaultType, float],
    spfm_out: dict[FaultType, float],
    lfm_out: dict[FaultType, float],
) -> dict:
    """Triggered after a hardware block completes its FIT rate transformation.

    The observer acts upon this event (e.g., drawing the block in a diagram)
    and returns the new visual connection points (ports).

    Args:
        block (BlockInterface): The instance of the logic block (defines shape and type).
        input_ports (dict): Mapping of fault types to incoming node IDs (defines edge origins).
        spfm_in (dict[FaultType, float]): Dictionary of incoming residual/SPFM FIT rates.
        lfm_in (dict[FaultType, float]): Dictionary of incoming latent/LFM FIT rates.
        spfm_out (dict[FaultType, float]): Updated dictionary of outgoing residual/SPFM FIT rates.
        lfm_out (dict[FaultType, float]): Updated dictionary of outgoing latent/LFM FIT rates.

    Returns:
        dict: A dictionary containing the newly created output ports (node IDs) to be
        used as inputs for the next block in the chain.
    """
    pass