pipeline_block

Executes a sequence of logic blocks for FIT rate transformations.

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

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, sub_blocks: list[BlockInterface]):
        """Initializes the PipelineBlock with a sequence of sub-blocks.

        Args:
            name (str): The descriptive name of the pipeline.
            sub_blocks (list[BlockInterface]): A list of blocks implementing BlockInterface
                to be executed in strict sequential order.
        """
        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]]:
        """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.sub_blocks:
            current_spfm, current_lfm = block.compute_fit(current_spfm, current_lfm)

        return current_spfm, current_lfm

    def to_dict(self):
        """Serializes the PipelineBlock into a dictionary for configuration export.

        Returns:
            dict: A dictionary containing the block type and all parameters
                needed to reconstruct this PipelineBlock via the BlockFactory.
        """

        return {"type": "PipelineBlock", "name": self.name, "sub_blocks": [block.to_dict() for block in self.sub_blocks]}

__init__(name, sub_blocks)

Initializes the PipelineBlock with a sequence of sub-blocks.

Parameters:

Name	Type	Description	Default
name	str	The descriptive name of the pipeline.	required
sub_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

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

    Args:
        name (str): The descriptive name of the pipeline.
        sub_blocks (list[BlockInterface]): A list of blocks implementing BlockInterface
            to be executed in strict sequential order.
    """
    self.name = name
    self.sub_blocks = sub_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

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.sub_blocks:
        current_spfm, current_lfm = block.compute_fit(current_spfm, current_lfm)

    return current_spfm, current_lfm

to_dict()

Serializes the PipelineBlock into a dictionary for configuration export.

Returns:

Name	Type	Description
dict		A dictionary containing the block type and all parameters needed to reconstruct this PipelineBlock via the BlockFactory.

Source code in src/ecc_analyzer/core/pipeline_block.py

def to_dict(self):
    """Serializes the PipelineBlock into a dictionary for configuration export.

    Returns:
        dict: A dictionary containing the block type and all parameters
            needed to reconstruct this PipelineBlock via the BlockFactory.
    """

    return {"type": "PipelineBlock", "name": self.name, "sub_blocks": [block.to_dict() for block in self.sub_blocks]}