Skip to content

Index

Exposes the core logic blocks for the Beachlore Safety framework.

AsilBlock

Evaluates final system metrics and determines the achieved ASIL level.

Calculates Single-Point Fault Metric (SPFM) and Latent Fault Metric (LFM) according to ISO 26262 requirements.

Source code in src/ecc_analyzer/core/asil_block.py 
 10
 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
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
class AsilBlock:
    """Evaluates final system metrics and determines the achieved ASIL level.

    Calculates Single-Point Fault Metric (SPFM) and Latent Fault Metric (LFM)
    according to ISO 26262 requirements.
    """

    # Standardized ASIL requirements
    # Format: [Min SPFM, Min LFM, Max Residual FIT]
    ASIL_REQUIREMENTS = {
        "D": [0.99, 0.90, 10.0],
        "C": [0.97, 0.80, 100.0],
        "B": [0.90, 0.60, 100.0],
        "A": [0.00, 0.00, 1000.0],
    }

    def __init__(self, name: str):
        """Initializes the ASIL calculation block.

        Args:
            name (str): The descriptive name of the calculation block.
        """
        self.name = name

    def _determine_asil(self, spfm: float, lfm: float, lambda_rf_sum: float) -> str:
        """Determines the achieved ASIL level based on calculated metrics.

        Args:
            spfm (float): Single-Point Fault Metric value (0.0 to 1.0).
            lfm (float): Latent Fault Metric value (0.0 to 1.0).
            lambda_rf_sum (float): Total sum of residual FIT rates.

        Returns:
            str: A string representing the achieved ASIL level (e.g., "ASIL D")
            or "QM" (Quality Management).
        """
        for asil_level in ["D", "C", "B"]:
            req = self.ASIL_REQUIREMENTS[asil_level]
            spfm_min, lfm_min, rf_max = req
            if spfm >= spfm_min and lfm >= lfm_min and lambda_rf_sum < rf_max:
                return f"ASIL {asil_level}"

        if lambda_rf_sum < self.ASIL_REQUIREMENTS["A"][2]:
            return "ASIL A"

        return "QM (Quality Management)"

    def compute_metrics(
        self,
        lambda_total: float,
        final_spfm_dict: dict[FaultType, float],
        final_lfm_dict: dict[FaultType, float],
    ) -> dict[str, Any]:
        """Calculates final ISO 26262 metrics using result dictionaries.

        Args:
            lambda_total (float): The total FIT rate of the entire system.
            final_spfm_dict (dict[FaultType, float]): Dictionary containing final
                residual and dangerous FIT rates.
            final_lfm_dict (dict[FaultType, float]): Dictionary containing final
                latent FIT rates.

        Returns:
            dict[str, Any]: A dictionary containing:
                - "SPFM" (float): Single-Point Fault Metric.
                - "LFM" (float): Latent Fault Metric.
                - "Lambda_RF_Sum" (float): Residual FIT Rate Sum.
                - "ASIL_Achieved" (str): The determined ASIL level.
        """
        lambda_dangerous_sum = sum(final_spfm_dict.values())
        lambda_latent_sum = sum(final_lfm_dict.values())
        lambda_rf_sum = lambda_dangerous_sum

        spfm = 0.0
        lfm = 0.0

        if lambda_total > 0:
            spfm = 1.0 - (lambda_dangerous_sum / lambda_total)

        lambda_safe_and_covered = lambda_total - lambda_dangerous_sum

        if lambda_safe_and_covered > 0:
            lfm = 1.0 - (lambda_latent_sum / lambda_safe_and_covered)

        achieved_asil = self._determine_asil(spfm, lfm, lambda_rf_sum)

        return {
            "SPFM": spfm,
            "LFM": lfm,
            "Lambda_RF_Sum": lambda_rf_sum,
            "ASIL_Achieved": achieved_asil,
        }

__init__(name)

Initializes the ASIL calculation block.

Parameters:

Name Type Description Default
name str

The descriptive name of the calculation block.

 required
Source code in src/ecc_analyzer/core/asil_block.py 
26
27
28
29
30
31
32
def __init__(self, name: str):
    """Initializes the ASIL calculation block.

    Args:
        name (str): The descriptive name of the calculation block.
    """
    self.name = name

compute_metrics(lambda_total, final_spfm_dict, final_lfm_dict)

Calculates final ISO 26262 metrics using result dictionaries.

Parameters:

Name Type Description Default
lambda_total float

The total FIT rate of the entire system.

 required
final_spfm_dict dict[FaultType, float]

Dictionary containing final residual and dangerous FIT rates.

 required
final_lfm_dict dict[FaultType, float]

Dictionary containing final latent FIT rates.

 required

Returns:

Type Description
dict[str, Any]

dict[str, Any]: A dictionary containing: - "SPFM" (float): Single-Point Fault Metric. - "LFM" (float): Latent Fault Metric. - "Lambda_RF_Sum" (float): Residual FIT Rate Sum. - "ASIL_Achieved" (str): The determined ASIL level.
Source code in src/ecc_analyzer/core/asil_block.py 
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
def compute_metrics(
    self,
    lambda_total: float,
    final_spfm_dict: dict[FaultType, float],
    final_lfm_dict: dict[FaultType, float],
) -> dict[str, Any]:
    """Calculates final ISO 26262 metrics using result dictionaries.

    Args:
        lambda_total (float): The total FIT rate of the entire system.
        final_spfm_dict (dict[FaultType, float]): Dictionary containing final
            residual and dangerous FIT rates.
        final_lfm_dict (dict[FaultType, float]): Dictionary containing final
            latent FIT rates.

    Returns:
        dict[str, Any]: A dictionary containing:
            - "SPFM" (float): Single-Point Fault Metric.
            - "LFM" (float): Latent Fault Metric.
            - "Lambda_RF_Sum" (float): Residual FIT Rate Sum.
            - "ASIL_Achieved" (str): The determined ASIL level.
    """
    lambda_dangerous_sum = sum(final_spfm_dict.values())
    lambda_latent_sum = sum(final_lfm_dict.values())
    lambda_rf_sum = lambda_dangerous_sum

    spfm = 0.0
    lfm = 0.0

    if lambda_total > 0:
        spfm = 1.0 - (lambda_dangerous_sum / lambda_total)

    lambda_safe_and_covered = lambda_total - lambda_dangerous_sum

    if lambda_safe_and_covered > 0:
        lfm = 1.0 - (lambda_latent_sum / lambda_safe_and_covered)

    achieved_asil = self._determine_asil(spfm, lfm, lambda_rf_sum)

    return {
        "SPFM": spfm,
        "LFM": lfm,
        "Lambda_RF_Sum": lambda_rf_sum,
        "ASIL_Achieved": achieved_asil,
    }

Base

Bases: BlockInterface, ABC

Abstract base class for hardware components.

Provides a structured way to define internal logic hierarchies by wrapping complex logic into a single modular unit.

Source code in src/ecc_analyzer/core/base.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
48
49
50
51
52
53
54
class Base(BlockInterface, ABC):
    """Abstract base class for hardware components.

    Provides a structured way to define internal logic hierarchies by wrapping
    complex logic into a single modular unit.
    """

    def __init__(self, name: str):
        """Initializes the component and triggers the internal block configuration.

        Args:
            name (str): The descriptive name of the hardware component.
        """
        self.name = name
        self.root_block: Optional[BlockInterface] = None
        self.configure_blocks()

    @abstractmethod
    def configure_blocks(self):
        """Abstract method to define the internal logic structure (root block).

        Must be implemented by subclasses to specify the internal tree of blocks
        (e.g., using SumBlock, PipelineBlock).
        """
        pass

    def compute_fit(self, spfm_rates: dict[FaultType, float], lfm_rates: dict[FaultType, float]) -> tuple[dict[FaultType, float], dict[FaultType, float]]:
        """Delegates the FIT rate transformation to the internal root block.

        This allows the component to be treated as a single modular unit within the system,
        hiding its internal complexity.

        Args:
            spfm_rates (dict[FaultType, float]): Current residual failure rates.
            lfm_rates (dict[FaultType, float]): Current latent failure rates.

        Returns:
            tuple[dict[FaultType, float], dict[FaultType, float]]: Updated FIT rates
            processed by the internal root block.
        """
        if self.root_block is None:
            return spfm_rates.copy(), lfm_rates.copy()

        return self.root_block.compute_fit(spfm_rates, lfm_rates)

__init__(name)

Initializes the component and triggers the internal block configuration.

Parameters:

Name Type Description Default
name str

The descriptive name of the hardware component.

 required
Source code in src/ecc_analyzer/core/base.py 
18
19
20
21
22
23
24
25
26
def __init__(self, name: str):
    """Initializes the component and triggers the internal block configuration.

    Args:
        name (str): The descriptive name of the hardware component.
    """
    self.name = name
    self.root_block: Optional[BlockInterface] = None
    self.configure_blocks()

compute_fit(spfm_rates, lfm_rates)

Delegates the FIT rate transformation to the internal root block.

This allows the component to be treated as a single modular unit within the system, hiding its internal complexity.

Parameters:

Name Type Description Default
spfm_rates dict[FaultType, float]

Current residual failure rates.

 required
lfm_rates dict[FaultType, float]

Current latent failure rates.

 required

Returns:

Type Description
dict[FaultType, float]

tuple[dict[FaultType, float], dict[FaultType, float]]: Updated FIT rates
dict[FaultType, float]

processed by the internal root block.
Source code in src/ecc_analyzer/core/base.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
def compute_fit(self, spfm_rates: dict[FaultType, float], lfm_rates: dict[FaultType, float]) -> tuple[dict[FaultType, float], dict[FaultType, float]]:
    """Delegates the FIT rate transformation to the internal root block.

    This allows the component to be treated as a single modular unit within the system,
    hiding its internal complexity.

    Args:
        spfm_rates (dict[FaultType, float]): Current residual failure rates.
        lfm_rates (dict[FaultType, float]): Current latent failure rates.

    Returns:
        tuple[dict[FaultType, float], dict[FaultType, float]]: Updated FIT rates
        processed by the internal root block.
    """
    if self.root_block is None:
        return spfm_rates.copy(), lfm_rates.copy()

    return self.root_block.compute_fit(spfm_rates, lfm_rates)

configure_blocks() abstractmethod

Abstract method to define the internal logic structure (root block).

Must be implemented by subclasses to specify the internal tree of blocks (e.g., using SumBlock, PipelineBlock).

Source code in src/ecc_analyzer/core/base.py 
28
29
30
31
32
33
34
35
@abstractmethod
def configure_blocks(self):
    """Abstract method to define the internal logic structure (root block).

    Must be implemented by subclasses to specify the internal tree of blocks
    (e.g., using SumBlock, PipelineBlock).
    """
    pass

BasicEvent

Bases: BlockInterface

Represents a source of a fault (Basic Event) that injects a specific FIT rate.

This class handles the mathematical addition of failure rates to the fault dictionaries.

Source code in src/ecc_analyzer/core/basic_event.py 
 8
 9
10
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
class BasicEvent(BlockInterface):
    """Represents a source of a fault (Basic Event) that injects a specific FIT rate.

    This class handles the mathematical addition of failure rates to the fault dictionaries.
    """

    def __init__(self, fault_type: FaultType, rate: float, is_spfm: bool = True):
        """Initializes the BasicEvent fault source.

        Args:
            fault_type (FaultType): The type of fault (Enum) this event produces.
            rate (float): The FIT rate of this basic event.
            is_spfm (bool, optional): Whether this rate counts towards SPFM (True)
                or LFM (False). Defaults to True.
        """
        self.fault_type = fault_type
        self.lambda_BE = rate
        self.is_spfm = is_spfm

    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 by injecting the defined FIT rate.

        Args:
            spfm_rates (dict[FaultType, float]): Dictionary containing current SPFM/residual fault rates.
            lfm_rates (dict[FaultType, float]): Dictionary containing current LFM/latent fault rates.

        Returns:
            tuple[dict[FaultType, float], dict[FaultType, float]]: A tuple containing:
                - Updated SPFM rates dictionary.
                - Updated LFM rates dictionary.
        """
        new_spfm = spfm_rates.copy()
        new_lfm = lfm_rates.copy()

        target_dict = new_spfm if self.is_spfm else new_lfm
        target_dict[self.fault_type] = target_dict.get(self.fault_type, 0.0) + self.lambda_BE

        return new_spfm, new_lfm

__init__(fault_type, rate, is_spfm=True)

Initializes the BasicEvent fault source.

Parameters:

Name Type Description Default
fault_type FaultType

The type of fault (Enum) this event produces.

 required
rate float

The FIT rate of this basic event.

 required
is_spfm bool

Whether this rate counts towards SPFM (True) or LFM (False). Defaults to True.

 True
Source code in src/ecc_analyzer/core/basic_event.py 
14
15
16
17
18
19
20
21
22
23
24
25
def __init__(self, fault_type: FaultType, rate: float, is_spfm: bool = True):
    """Initializes the BasicEvent fault source.

    Args:
        fault_type (FaultType): The type of fault (Enum) this event produces.
        rate (float): The FIT rate of this basic event.
        is_spfm (bool, optional): Whether this rate counts towards SPFM (True)
            or LFM (False). Defaults to True.
    """
    self.fault_type = fault_type
    self.lambda_BE = rate
    self.is_spfm = is_spfm

compute_fit(spfm_rates, lfm_rates)

Transforms the input fault rate dictionaries by injecting the defined FIT rate.

Parameters:

Name Type Description Default
spfm_rates dict[FaultType, float]

Dictionary containing current SPFM/residual fault rates.

 required
lfm_rates dict[FaultType, float]

Dictionary containing current LFM/latent fault rates.

 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/core/basic_event.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
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 by injecting the defined FIT rate.

    Args:
        spfm_rates (dict[FaultType, float]): Dictionary containing current SPFM/residual fault rates.
        lfm_rates (dict[FaultType, float]): Dictionary containing current LFM/latent fault rates.

    Returns:
        tuple[dict[FaultType, float], dict[FaultType, float]]: A tuple containing:
            - Updated SPFM rates dictionary.
            - Updated LFM rates dictionary.
    """
    new_spfm = spfm_rates.copy()
    new_lfm = lfm_rates.copy()

    target_dict = new_spfm if self.is_spfm else new_lfm
    target_dict[self.fault_type] = target_dict.get(self.fault_type, 0.0) + self.lambda_BE

    return new_spfm, new_lfm

CoverageBlock

Bases: BlockInterface

Applies diagnostic coverage (DC) to a fault type.

Splits FIT rates into residual and latent components based on the defined coverage values (c_R, c_L).

Source code in src/ecc_analyzer/core/coverage_block.py 
10
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
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
class CoverageBlock(BlockInterface):
    """Applies diagnostic coverage (DC) to a fault type.

    Splits FIT rates into residual and latent components based on the defined
    coverage values (c_R, c_L).
    """

    def __init__(
        self,
        target_fault: FaultType,
        dc_rate_c_or_cR: float,
        dc_rate_latent_cL: Optional[float] = None,
        is_spfm: bool = True,
    ):
        """Initializes the CoverageBlock with specific diagnostic coverage parameters.

        Args:
            target_fault (FaultType): The fault type (Enum) to which coverage is applied.
            dc_rate_c_or_cR (float): The diagnostic coverage for residual faults
                (typically denoted as K_DC or c_R).
            dc_rate_latent_cL (Optional[float]): Optional specific coverage for latent
                faults (c_L). If None, standard ISO 26262 logic (1 - c_R) is assumed.
            is_spfm (bool, optional): Indicates if this block processes the SPFM/residual
                path. Defaults to True.
        """
        self.target_fault = target_fault
        self.is_spfm = is_spfm
        if dc_rate_latent_cL is not None:
            self.c_R = dc_rate_c_or_cR
            self.c_L = dc_rate_latent_cL
        else:
            self.c_R = dc_rate_c_or_cR
            self.c_L = 1.0 - dc_rate_c_or_cR

    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 by applying diagnostic coverage logic.

        Args:
            spfm_rates (dict[FaultType, float]): Current residual failure rates.
            lfm_rates (dict[FaultType, float]): Current latent failure rates.

        Returns:
            tuple[dict[FaultType, float], dict[FaultType, float]]: A tuple containing:
                - Updated SPFM rates dictionary.
                - Updated LFM rates dictionary.
        """
        new_spfm = spfm_rates.copy()
        new_lfm = lfm_rates.copy()

        if self.is_spfm:
            if self.target_fault in new_spfm:
                lambda_in = new_spfm.pop(self.target_fault)
                lambda_rf = lambda_in * (1.0 - self.c_R)
                if lambda_rf > 0:
                    new_spfm[self.target_fault] = new_spfm.get(self.target_fault, 0.0) + lambda_rf
                lambda_mpf_l = lambda_in * (1.0 - self.c_L)
                if lambda_mpf_l > 0:
                    new_lfm[self.target_fault] = new_lfm.get(self.target_fault, 0.0) + lambda_mpf_l
        else:
            if self.target_fault in new_lfm:
                lambda_in = new_lfm.pop(self.target_fault)
                lambda_rem = lambda_in * (1.0 - self.c_R)
                if lambda_rem > 0:
                    new_lfm[self.target_fault] = lambda_rem

        return new_spfm, new_lfm

__init__(target_fault, dc_rate_c_or_cR, dc_rate_latent_cL=None, is_spfm=True)

Initializes the CoverageBlock with specific diagnostic coverage parameters.

Parameters:

Name Type Description Default
target_fault FaultType

The fault type (Enum) to which coverage is applied.

 required
dc_rate_c_or_cR float

The diagnostic coverage for residual faults (typically denoted as K_DC or c_R).

 required
dc_rate_latent_cL Optional[float]

Optional specific coverage for latent faults (c_L). If None, standard ISO 26262 logic (1 - c_R) is assumed.

 None
is_spfm bool

Indicates if this block processes the SPFM/residual path. Defaults to True.

 True
Source code in src/ecc_analyzer/core/coverage_block.py 
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
def __init__(
    self,
    target_fault: FaultType,
    dc_rate_c_or_cR: float,
    dc_rate_latent_cL: Optional[float] = None,
    is_spfm: bool = True,
):
    """Initializes the CoverageBlock with specific diagnostic coverage parameters.

    Args:
        target_fault (FaultType): The fault type (Enum) to which coverage is applied.
        dc_rate_c_or_cR (float): The diagnostic coverage for residual faults
            (typically denoted as K_DC or c_R).
        dc_rate_latent_cL (Optional[float]): Optional specific coverage for latent
            faults (c_L). If None, standard ISO 26262 logic (1 - c_R) is assumed.
        is_spfm (bool, optional): Indicates if this block processes the SPFM/residual
            path. Defaults to True.
    """
    self.target_fault = target_fault
    self.is_spfm = is_spfm
    if dc_rate_latent_cL is not None:
        self.c_R = dc_rate_c_or_cR
        self.c_L = dc_rate_latent_cL
    else:
        self.c_R = dc_rate_c_or_cR
        self.c_L = 1.0 - dc_rate_c_or_cR

compute_fit(spfm_rates, lfm_rates)

Transforms the input fault rate dictionaries by applying diagnostic coverage logic.

Parameters:

Name Type Description Default
spfm_rates dict[FaultType, float]

Current residual failure rates.

 required
lfm_rates dict[FaultType, float]

Current latent failure rates.

 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/core/coverage_block.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
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 by applying diagnostic coverage logic.

    Args:
        spfm_rates (dict[FaultType, float]): Current residual failure rates.
        lfm_rates (dict[FaultType, float]): Current latent failure rates.

    Returns:
        tuple[dict[FaultType, float], dict[FaultType, float]]: A tuple containing:
            - Updated SPFM rates dictionary.
            - Updated LFM rates dictionary.
    """
    new_spfm = spfm_rates.copy()
    new_lfm = lfm_rates.copy()

    if self.is_spfm:
        if self.target_fault in new_spfm:
            lambda_in = new_spfm.pop(self.target_fault)
            lambda_rf = lambda_in * (1.0 - self.c_R)
            if lambda_rf > 0:
                new_spfm[self.target_fault] = new_spfm.get(self.target_fault, 0.0) + lambda_rf
            lambda_mpf_l = lambda_in * (1.0 - self.c_L)
            if lambda_mpf_l > 0:
                new_lfm[self.target_fault] = new_lfm.get(self.target_fault, 0.0) + lambda_mpf_l
    else:
        if self.target_fault in new_lfm:
            lambda_in = new_lfm.pop(self.target_fault)
            lambda_rem = lambda_in * (1.0 - self.c_R)
            if lambda_rem > 0:
                new_lfm[self.target_fault] = lambda_rem

    return new_spfm, new_lfm

ObservableBlock

Bases: ObservableInterface

A wrapper class that encapsulates a logic block.

Manages both mathematical results (via the wrapped block) and visual output ports (via observers).

Source code in src/ecc_analyzer/core/observable_block.py 
 8
 9
10
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
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
class ObservableBlock(ObservableInterface):
    """A wrapper class that encapsulates a logic block.

    Manages both mathematical results (via the wrapped block) and visual output
    ports (via observers).
    """

    def __init__(self, logic_block: BlockInterface):
        """Initializes the observable wrapper.

        Args:
            logic_block (BlockInterface): The pure mathematical block to be wrapped.
        """
        self.block = logic_block
        self._observers = []

    def attach(self, observer: SafetyObserver):
        """Registers an observer.

        Args:
            observer (SafetyObserver): The SafetyObserver instance to be registered.
        """
        if observer not in self._observers:
            self._observers.append(observer)

    def run(
        self,
        spfm_in: dict[FaultType, float],
        lfm_in: dict[FaultType, float],
        input_ports: dict,
    ) -> tuple[dict[FaultType, float], dict[FaultType, float], dict]:
        """Executes calculation and collects output ports from the observer.

        Args:
            spfm_in (dict[FaultType, float]): Incoming SPFM fault rates.
            lfm_in (dict[FaultType, float]): Incoming LFM fault rates.
            input_ports (dict): Mapping of incoming node IDs for visualization.

        Returns:
            tuple[dict[FaultType, float], dict[FaultType, float], dict]: A tuple containing:
                - Updated SPFM rates.
                - Updated LFM rates.
                - Output ports dictionary from the observer.
        """
        spfm_out, lfm_out = self.block.compute_fit(spfm_in, lfm_in)

        output_ports = self.notify(input_ports, spfm_in, lfm_in, spfm_out, lfm_out)

        return spfm_out, lfm_out, output_ports

    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 results and returns the visual ports created by the observer.

        Args:
            input_ports (dict): Incoming visual ports.
            spfm_in (dict[FaultType, float]): Incoming SPFM rates.
            lfm_in (dict[FaultType, float]): Incoming LFM rates.
            spfm_out (dict[FaultType, float]): Outgoing SPFM rates.
            lfm_out (dict[FaultType, float]): Outgoing LFM rates.

        Returns:
            dict: The visual output ports created by the observers.
        """
        last_created_ports = {}
        for observer in self._observers:
            ports = observer.on_block_computed(self.block, input_ports, spfm_in, lfm_in, spfm_out, lfm_out)
            if ports:
                last_created_ports = ports

        return last_created_ports

__init__(logic_block)

Initializes the observable wrapper.

Parameters:

Name Type Description Default
logic_block BlockInterface

The pure mathematical block to be wrapped.

 required
Source code in src/ecc_analyzer/core/observable_block.py 
15
16
17
18
19
20
21
22
def __init__(self, logic_block: BlockInterface):
    """Initializes the observable wrapper.

    Args:
        logic_block (BlockInterface): The pure mathematical block to be wrapped.
    """
    self.block = logic_block
    self._observers = []

attach(observer)

Registers an observer.

Parameters:

Name Type Description Default
observer SafetyObserver

The SafetyObserver instance to be registered.

 required
Source code in src/ecc_analyzer/core/observable_block.py 
24
25
26
27
28
29
30
31
def attach(self, observer: SafetyObserver):
    """Registers an observer.

    Args:
        observer (SafetyObserver): The SafetyObserver instance to be registered.
    """
    if observer not in self._observers:
        self._observers.append(observer)

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

Broadcasts results and returns the visual ports created by the observer.

Parameters:

Name Type Description Default
input_ports dict

Incoming visual ports.

 required
spfm_in dict[FaultType, float]

Incoming SPFM rates.

 required
lfm_in dict[FaultType, float]

Incoming LFM rates.

 required
spfm_out dict[FaultType, float]

Outgoing SPFM rates.

 required
lfm_out dict[FaultType, float]

Outgoing LFM rates.

 required

Returns:

Name Type Description
dict dict

The visual output ports created by the observers.
Source code in src/ecc_analyzer/core/observable_block.py 
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
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 results and returns the visual ports created by the observer.

    Args:
        input_ports (dict): Incoming visual ports.
        spfm_in (dict[FaultType, float]): Incoming SPFM rates.
        lfm_in (dict[FaultType, float]): Incoming LFM rates.
        spfm_out (dict[FaultType, float]): Outgoing SPFM rates.
        lfm_out (dict[FaultType, float]): Outgoing LFM rates.

    Returns:
        dict: The visual output ports created by the observers.
    """
    last_created_ports = {}
    for observer in self._observers:
        ports = observer.on_block_computed(self.block, input_ports, spfm_in, lfm_in, spfm_out, lfm_out)
        if ports:
            last_created_ports = ports

    return last_created_ports

run(spfm_in, lfm_in, input_ports)

Executes calculation and collects output ports from the observer.

Parameters:

Name Type Description Default
spfm_in dict[FaultType, float]

Incoming SPFM fault rates.

 required
lfm_in dict[FaultType, float]

Incoming LFM fault rates.

 required
input_ports dict

Mapping of incoming node IDs for visualization.

 required

Returns:

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

tuple[dict[FaultType, float], dict[FaultType, float], dict]: A tuple containing: - Updated SPFM rates. - Updated LFM rates. - Output ports dictionary from the observer.
Source code in src/ecc_analyzer/core/observable_block.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
def run(
    self,
    spfm_in: dict[FaultType, float],
    lfm_in: dict[FaultType, float],
    input_ports: dict,
) -> tuple[dict[FaultType, float], dict[FaultType, float], dict]:
    """Executes calculation and collects output ports from the observer.

    Args:
        spfm_in (dict[FaultType, float]): Incoming SPFM fault rates.
        lfm_in (dict[FaultType, float]): Incoming LFM fault rates.
        input_ports (dict): Mapping of incoming node IDs for visualization.

    Returns:
        tuple[dict[FaultType, float], dict[FaultType, float], dict]: A tuple containing:
            - Updated SPFM rates.
            - Updated LFM rates.
            - Output ports dictionary from the observer.
    """
    spfm_out, lfm_out = self.block.compute_fit(spfm_in, lfm_in)

    output_ports = self.notify(input_ports, spfm_in, lfm_in, spfm_out, lfm_out)

    return spfm_out, lfm_out, output_ports

PipelineBlock

Bases: BlockInterface

Executes a sequence of blocks where the output of one block becomes the input of the next.

This block type is used to model serial hardware paths or sequential processing steps (e.g., Source -> ECC -> Trim).

Source code in src/ecc_analyzer/core/pipeline_block.py 
 8
 9
10
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
class PipelineBlock(BlockInterface):
    """Executes a sequence of blocks where the output of one block becomes the input of the next.

    This block type is used to model serial hardware paths or sequential processing steps
    (e.g., Source -> ECC -> Trim).
    """

    def __init__(self, name: str, blocks: list[BlockInterface]):
        """Initializes the PipelineBlock with a sequence of sub-blocks.

        Args:
            name (str): The descriptive name of the pipeline.
            blocks (list[BlockInterface]): A list of blocks implementing BlockInterface
                to be executed in strict sequential order.
        """
        self.name = name
        self.blocks = blocks

    def compute_fit(self, spfm_rates: dict[FaultType, float], lfm_rates: dict[FaultType, float]) -> tuple[dict[FaultType, float], dict[FaultType, float]]:
        """Sequentially processes all blocks in the pipeline.

        Passes the output rates of one block as the input to the next block in the list.

        Args:
            spfm_rates (dict[FaultType, float]): Initial residual failure rates entering the pipeline.
            lfm_rates (dict[FaultType, float]): Initial latent failure rates entering the pipeline.

        Returns:
            tuple[dict[FaultType, float], dict[FaultType, float]]: A tuple containing:
                - Final SPFM rates after the last block.
                - Final LFM rates after the last block.
        """
        current_spfm = spfm_rates.copy()
        current_lfm = lfm_rates.copy()

        for block in self.blocks:
            current_spfm, current_lfm = block.compute_fit(current_spfm, current_lfm)

        return current_spfm, current_lfm

__init__(name, blocks)

Initializes the PipelineBlock with a sequence of sub-blocks.

Parameters:

Name Type Description Default
name str

The descriptive name of the pipeline.

 required
blocks list[BlockInterface]

A list of blocks implementing BlockInterface to be executed in strict sequential order.

 required
Source code in src/ecc_analyzer/core/pipeline_block.py 
15
16
17
18
19
20
21
22
23
24
def __init__(self, name: str, blocks: list[BlockInterface]):
    """Initializes the PipelineBlock with a sequence of sub-blocks.

    Args:
        name (str): The descriptive name of the pipeline.
        blocks (list[BlockInterface]): A list of blocks implementing BlockInterface
            to be executed in strict sequential order.
    """
    self.name = name
    self.blocks = blocks

compute_fit(spfm_rates, lfm_rates)

Sequentially processes all blocks in the pipeline.

Passes the output rates of one block as the input to the next block in the list.

Parameters:

Name Type Description Default
spfm_rates dict[FaultType, float]

Initial residual failure rates entering the pipeline.

 required
lfm_rates dict[FaultType, float]

Initial latent failure rates entering the pipeline.

 required

Returns:

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

tuple[dict[FaultType, float], dict[FaultType, float]]: A tuple containing: - Final SPFM rates after the last block. - Final LFM rates after the last block.
Source code in src/ecc_analyzer/core/pipeline_block.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
def compute_fit(self, spfm_rates: dict[FaultType, float], lfm_rates: dict[FaultType, float]) -> tuple[dict[FaultType, float], dict[FaultType, float]]:
    """Sequentially processes all blocks in the pipeline.

    Passes the output rates of one block as the input to the next block in the list.

    Args:
        spfm_rates (dict[FaultType, float]): Initial residual failure rates entering the pipeline.
        lfm_rates (dict[FaultType, float]): Initial latent failure rates entering the pipeline.

    Returns:
        tuple[dict[FaultType, float], dict[FaultType, float]]: A tuple containing:
            - Final SPFM rates after the last block.
            - Final LFM rates after the last block.
    """
    current_spfm = spfm_rates.copy()
    current_lfm = lfm_rates.copy()

    for block in self.blocks:
        current_spfm, current_lfm = block.compute_fit(current_spfm, current_lfm)

    return current_spfm, current_lfm

SplitBlock

Bases: BlockInterface

Distributes the FIT rate of a specific fault type across multiple other fault types.

The distribution is based on a defined percentage mapping. This is typically used to model how a generic fault (like "DRAM Error") manifests as specific sub-types (e.g., SBE, DBE) based on physical probabilities.

Source code in src/ecc_analyzer/core/split_block.py 
 8
 9
10
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
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
class SplitBlock(BlockInterface):
    """Distributes the FIT rate of a specific fault type across multiple other fault types.

    The distribution is based on a defined percentage mapping. This is typically used
    to model how a generic fault (like "DRAM Error") manifests as specific sub-types
    (e.g., SBE, DBE) based on physical probabilities.
    """

    def __init__(
        self,
        name: str,
        fault_to_split: FaultType,
        distribution_rates: dict[FaultType, float],
        is_spfm: bool = True,
    ):
        """Initializes the SplitBlock with a distribution mapping.

        Args:
            name (str): The descriptive name of the split operation.
            fault_to_split (FaultType): The source fault type (Enum) to be distributed.
            distribution_rates (dict[FaultType, float]): Dictionary mapping target
                FaultTypes to their probability (0.0 - 1.0).
            is_spfm (bool, optional): Indicates if this split occurs on the SPFM/residual
                path. Defaults to True.

        Raises:
            ValueError: If the sum of the provided distribution rates exceeds 1.0.
        """
        sum_of_rates = sum(distribution_rates.values())
        if sum_of_rates > 1.0 + 1e-9:
            raise ValueError(f"Sum of distribution rates ({sum_of_rates:.4f}) must not exceed 1.0.")

        self.name = name
        self.fault_to_split = fault_to_split
        self.distribution_rates = distribution_rates
        self.is_spfm = is_spfm

    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 by redistributing the source fault rate.

        Args:
            spfm_rates (dict[FaultType, float]): Current residual failure rates.
            lfm_rates (dict[FaultType, float]): Current latent failure rates.

        Returns:
            tuple[dict[FaultType, float], dict[FaultType, float]]: A tuple containing:
                - Updated SPFM rates.
                - Updated LFM rates.
        """
        new_spfm = spfm_rates.copy()
        new_lfm = lfm_rates.copy()
        target_dict = new_spfm if self.is_spfm else new_lfm

        if self.fault_to_split in target_dict:
            original_rate = target_dict.pop(self.fault_to_split)
            for target_fault, probability in self.distribution_rates.items():
                split_rate = original_rate * probability
                target_dict[target_fault] = target_dict.get(target_fault, 0.0) + split_rate

        return new_spfm, new_lfm

__init__(name, fault_to_split, distribution_rates, is_spfm=True)

Initializes the SplitBlock with a distribution mapping.

Parameters:

Name Type Description Default
name str

The descriptive name of the split operation.

 required
fault_to_split FaultType

The source fault type (Enum) to be distributed.

 required
distribution_rates dict[FaultType, float]

Dictionary mapping target FaultTypes to their probability (0.0 - 1.0).

 required
is_spfm bool

Indicates if this split occurs on the SPFM/residual path. Defaults to True.

 True

Raises:

Type Description
ValueError

If the sum of the provided distribution rates exceeds 1.0.
Source code in src/ecc_analyzer/core/split_block.py 
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
def __init__(
    self,
    name: str,
    fault_to_split: FaultType,
    distribution_rates: dict[FaultType, float],
    is_spfm: bool = True,
):
    """Initializes the SplitBlock with a distribution mapping.

    Args:
        name (str): The descriptive name of the split operation.
        fault_to_split (FaultType): The source fault type (Enum) to be distributed.
        distribution_rates (dict[FaultType, float]): Dictionary mapping target
            FaultTypes to their probability (0.0 - 1.0).
        is_spfm (bool, optional): Indicates if this split occurs on the SPFM/residual
            path. Defaults to True.

    Raises:
        ValueError: If the sum of the provided distribution rates exceeds 1.0.
    """
    sum_of_rates = sum(distribution_rates.values())
    if sum_of_rates > 1.0 + 1e-9:
        raise ValueError(f"Sum of distribution rates ({sum_of_rates:.4f}) must not exceed 1.0.")

    self.name = name
    self.fault_to_split = fault_to_split
    self.distribution_rates = distribution_rates
    self.is_spfm = is_spfm

compute_fit(spfm_rates, lfm_rates)

Transforms the input fault rate dictionaries by redistributing the source fault rate.

Parameters:

Name Type Description Default
spfm_rates dict[FaultType, float]

Current residual failure rates.

 required
lfm_rates dict[FaultType, float]

Current latent failure rates.

 required

Returns:

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

tuple[dict[FaultType, float], dict[FaultType, float]]: A tuple containing: - Updated SPFM rates. - Updated LFM rates.
Source code in src/ecc_analyzer/core/split_block.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
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 by redistributing the source fault rate.

    Args:
        spfm_rates (dict[FaultType, float]): Current residual failure rates.
        lfm_rates (dict[FaultType, float]): Current latent failure rates.

    Returns:
        tuple[dict[FaultType, float], dict[FaultType, float]]: A tuple containing:
            - Updated SPFM rates.
            - Updated LFM rates.
    """
    new_spfm = spfm_rates.copy()
    new_lfm = lfm_rates.copy()
    target_dict = new_spfm if self.is_spfm else new_lfm

    if self.fault_to_split in target_dict:
        original_rate = target_dict.pop(self.fault_to_split)
        for target_fault, probability in self.distribution_rates.items():
            split_rate = original_rate * probability
            target_dict[target_fault] = target_dict.get(target_fault, 0.0) + split_rate

    return new_spfm, new_lfm

SumBlock

Bases: BlockInterface

Parallel block that aggregates FIT rates from multiple sub-blocks.

Manages path junctions by executing sub-blocks in parallel (starting from the same input state) and calculating the sum of their individual contributions (deltas) to the total system rates.

Source code in src/ecc_analyzer/core/sum_block.py 
 8
 9
10
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
48
49
50
51
52
53
class SumBlock(BlockInterface):
    """Parallel block that aggregates FIT rates from multiple sub-blocks.

    Manages path junctions by executing sub-blocks in parallel (starting from the
    same input state) and calculating the sum of their individual contributions
    (deltas) to the total system rates.
    """

    def __init__(self, name: str, sub_blocks: list[BlockInterface]):
        """Initializes the SumBlock with a list of parallel sub-blocks.

        Args:
            name (str): The descriptive name of the aggregation block.
            sub_blocks (list[BlockInterface]): List of blocks whose results will be summed.
        """
        self.name = name
        self.sub_blocks = sub_blocks

    def compute_fit(self, spfm_rates: dict[FaultType, float], lfm_rates: dict[FaultType, float]) -> tuple[dict[FaultType, float], dict[FaultType, float]]:
        """Aggregates the FIT rate transformations from all internal parallel blocks.

        Calculates the delta contribution of each block relative to the input state
        and sums these deltas to produce the final output state.

        Args:
            spfm_rates (dict[FaultType, float]): Current residual failure rates (Input state).
            lfm_rates (dict[FaultType, float]): Current latent failure rates (Input state).

        Returns:
            tuple[dict[FaultType, float], dict[FaultType, float]]: A tuple containing:
                - Final aggregated SPFM rates.
                - Final aggregated LFM rates.
        """
        total_spfm = spfm_rates.copy()
        total_lfm = lfm_rates.copy()
        for block in self.sub_blocks:
            res_spfm, res_lfm = block.compute_fit(spfm_rates, lfm_rates)
            for fault in set(res_spfm.keys()) | set(spfm_rates.keys()):
                delta = res_spfm.get(fault, 0.0) - spfm_rates.get(fault, 0.0)
                if delta != 0:
                    total_spfm[fault] = total_spfm.get(fault, 0.0) + delta
            for fault in set(res_lfm.keys()) | set(lfm_rates.keys()):
                delta = res_lfm.get(fault, 0.0) - lfm_rates.get(fault, 0.0)
                if delta != 0:
                    total_lfm[fault] = total_lfm.get(fault, 0.0) + delta
        return total_spfm, total_lfm

__init__(name, sub_blocks)

Initializes the SumBlock with a list of parallel sub-blocks.

Parameters:

Name Type Description Default
name str

The descriptive name of the aggregation block.

 required
sub_blocks list[BlockInterface]

List of blocks whose results will be summed.

 required
Source code in src/ecc_analyzer/core/sum_block.py 
16
17
18
19
20
21
22
23
24
def __init__(self, name: str, sub_blocks: list[BlockInterface]):
    """Initializes the SumBlock with a list of parallel sub-blocks.

    Args:
        name (str): The descriptive name of the aggregation block.
        sub_blocks (list[BlockInterface]): List of blocks whose results will be summed.
    """
    self.name = name
    self.sub_blocks = sub_blocks

compute_fit(spfm_rates, lfm_rates)

Aggregates the FIT rate transformations from all internal parallel blocks.

Calculates the delta contribution of each block relative to the input state and sums these deltas to produce the final output state.

Parameters:

Name Type Description Default
spfm_rates dict[FaultType, float]

Current residual failure rates (Input state).

 required
lfm_rates dict[FaultType, float]

Current latent failure rates (Input state).

 required

Returns:

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

tuple[dict[FaultType, float], dict[FaultType, float]]: A tuple containing: - Final aggregated SPFM rates. - Final aggregated LFM rates.
Source code in src/ecc_analyzer/core/sum_block.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
def compute_fit(self, spfm_rates: dict[FaultType, float], lfm_rates: dict[FaultType, float]) -> tuple[dict[FaultType, float], dict[FaultType, float]]:
    """Aggregates the FIT rate transformations from all internal parallel blocks.

    Calculates the delta contribution of each block relative to the input state
    and sums these deltas to produce the final output state.

    Args:
        spfm_rates (dict[FaultType, float]): Current residual failure rates (Input state).
        lfm_rates (dict[FaultType, float]): Current latent failure rates (Input state).

    Returns:
        tuple[dict[FaultType, float], dict[FaultType, float]]: A tuple containing:
            - Final aggregated SPFM rates.
            - Final aggregated LFM rates.
    """
    total_spfm = spfm_rates.copy()
    total_lfm = lfm_rates.copy()
    for block in self.sub_blocks:
        res_spfm, res_lfm = block.compute_fit(spfm_rates, lfm_rates)
        for fault in set(res_spfm.keys()) | set(spfm_rates.keys()):
            delta = res_spfm.get(fault, 0.0) - spfm_rates.get(fault, 0.0)
            if delta != 0:
                total_spfm[fault] = total_spfm.get(fault, 0.0) + delta
        for fault in set(res_lfm.keys()) | set(lfm_rates.keys()):
            delta = res_lfm.get(fault, 0.0) - lfm_rates.get(fault, 0.0)
            if delta != 0:
                total_lfm[fault] = total_lfm.get(fault, 0.0) + delta
    return total_spfm, total_lfm

TransformationBlock

Bases: BlockInterface

Transfers a portion of one fault type's rate to another fault type.

This operation adds a calculated rate to the target fault type based on the source fault type, without removing the rate from the source (unlike SplitBlock).

Source code in src/ecc_analyzer/core/transformation_block.py 
 8
 9
10
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
class TransformationBlock(BlockInterface):
    """Transfers a portion of one fault type's rate to another fault type.

    This operation adds a calculated rate to the target fault type based on the
    source fault type, without removing the rate from the source (unlike SplitBlock).
    """

    def __init__(self, source_fault: FaultType, target_fault: FaultType, factor: float):
        """Initializes the transformation block.

        Args:
            source_fault (FaultType): The fault type from which the rate is calculated.
            target_fault (FaultType): The fault type to which the calculated rate is added.
            factor (float): The multiplication factor applied to the source rate.
        """
        self.source = source_fault
        self.target = target_fault
        self.factor = factor

    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 by transferring a portion of the source rate.

        Args:
            spfm_rates (dict[FaultType, float]): Current residual failure rates.
            lfm_rates (dict[FaultType, float]): Current latent failure rates.

        Returns:
            tuple[dict[FaultType, float], dict[FaultType, float]]: A tuple containing:
                - Updated SPFM rates (target fault increased).
                - Unchanged LFM rates.
        """
        new_spfm = spfm_rates.copy()
        if self.source in new_spfm:
            transfer_rate = new_spfm[self.source] * self.factor
            new_spfm[self.target] = new_spfm.get(self.target, 0.0) + transfer_rate

        return new_spfm, lfm_rates.copy()

__init__(source_fault, target_fault, factor)

Initializes the transformation block.

Parameters:

Name Type Description Default
source_fault FaultType

The fault type from which the rate is calculated.

 required
target_fault FaultType

The fault type to which the calculated rate is added.

 required
factor float

The multiplication factor applied to the source rate.

 required
Source code in src/ecc_analyzer/core/transformation_block.py 
15
16
17
18
19
20
21
22
23
24
25
def __init__(self, source_fault: FaultType, target_fault: FaultType, factor: float):
    """Initializes the transformation block.

    Args:
        source_fault (FaultType): The fault type from which the rate is calculated.
        target_fault (FaultType): The fault type to which the calculated rate is added.
        factor (float): The multiplication factor applied to the source rate.
    """
    self.source = source_fault
    self.target = target_fault
    self.factor = factor

compute_fit(spfm_rates, lfm_rates)

Transforms the input fault rate dictionaries by transferring a portion of the source rate.

Parameters:

Name Type Description Default
spfm_rates dict[FaultType, float]

Current residual failure rates.

 required
lfm_rates dict[FaultType, float]

Current latent failure rates.

 required

Returns:

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

tuple[dict[FaultType, float], dict[FaultType, float]]: A tuple containing: - Updated SPFM rates (target fault increased). - Unchanged LFM rates.
Source code in src/ecc_analyzer/core/transformation_block.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
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 by transferring a portion of the source rate.

    Args:
        spfm_rates (dict[FaultType, float]): Current residual failure rates.
        lfm_rates (dict[FaultType, float]): Current latent failure rates.

    Returns:
        tuple[dict[FaultType, float], dict[FaultType, float]]: A tuple containing:
            - Updated SPFM rates (target fault increased).
            - Unchanged LFM rates.
    """
    new_spfm = spfm_rates.copy()
    if self.source in new_spfm:
        transfer_rate = new_spfm[self.source] * self.factor
        new_spfm[self.target] = new_spfm.get(self.target, 0.0) + transfer_rate

    return new_spfm, lfm_rates.copy()