IChain

class ioiocore.i_chain.IChain(input_ports=None, **kwargs)

Bases: Chain

A chain that receives input and processes it through a sequence of internal nodes.

Structure: <input> => IONode => … => IONode => INode

The input ports are exposed from the first internal node, making this chain usable anywhere an input element is expected.

Inheriting classes must implement create_internal_nodes(), which returns the list of internal nodes.

Example

class MyProcessor(IChain):
    def create_internal_nodes(self) -> List[Node]:
        return [
            FilterNode(),    # IONode
            TransformNode(), # IONode
            SinkNode()       # INode (terminal)
        ]

Parameters:

input_ports (list)

class Configuration(input_ports=None, **kwargs)

Bases: Configuration

Configuration class for IChain.

Parameters:

input_ports (list)

class Keys

Bases: Keys

Keys for the IChain configuration.

INPUT_PORTS = 'input_ports'

__init__(input_ports=None, **kwargs)

Initialize the IChain configuration.

Parameters:

• input_ports (list) – A list of input port configurations. If None, ports will be inherited from the first internal node.

• **kwargs (additional keyword arguments) – Other configuration options.

__init__(input_ports=None, **kwargs)

Initialize the IChain.

Parameters:

• input_ports (list) – A list of input port configurations (default is None).

• **kwargs (additional keyword arguments) – Other configuration options.

config: Configuration

abstractmethod create_internal_nodes()

Create and return the list of internal nodes.

The first node must have input ports (IONode or INode). The last node must be an INode. Intermediate nodes should be IONodes.

Returns:

List of internal nodes forming the processing chain.

Return type:

List[Node]

get_input_port(port_name)

Get an input port by name.

Parameters:

port_name (str)

get_port(port_name)

Deprecated: use get_input_port() instead.

Parameters:

port_name (str)