observable_interface

Defines the contract for observable components within the safety system.

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

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

@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

@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