sum_block

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

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

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

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

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