Node

class ioiocore.node.Node(name=None, **kwargs)

Bases: ProcessingElement

Abstract base class representing a concrete processing node in a pipeline.

Extends ProcessingElement with setup() and step() methods that define the node’s initialization and processing logic.

Parameters:

name (str)

_IMP_CLASS

The implementation class for the Node.

Type:

type

_imp

The instance of the implementation class for the Node.

Type:

_IMP_CLASS

config

The configuration object for the Node.

Type:

Configuration

class Configuration(name=None, **kwargs)

Bases: Configuration

Configuration class for the Node

Parameters:

name (str)

Keys

The configuration keys.

class Keys

Bases: Keys

Configuration keys.

__init__(name=None, **kwargs)

Initialize the Node configuration.

Parameters:

• name (str) – The name of the Node (default is the class name).

• kwargs (additional keyword arguments) – Other configuration parameters passed to the base class.

__init__(name=None, **kwargs)

Initialize the Node.

Parameters:

• name (str) – The name of the Node (default is None).

• kwargs (additional keyword arguments) – Other parameters passed to create the Node configuration.

config: Configuration

is_externally_drained()

Report whether this node’s outputs leave the pipeline.

The outlet counterpart of is_externally_fed(): a chain refuses an IONode at its outlet because its output would be silently dropped. A node that sends its output somewhere outside the graph returns True, and the chain admits it.

Returns:

True if the outputs are consumed externally.

Return type:

bool

is_externally_fed()

Report whether this node’s inputs come from outside the pipeline.

A chain refuses an IONode at its inlet, because an IONode there would leave its input ports with nothing connected to them and the chain could never be fed. That is the right default: it catches a real authoring mistake.

It is wrong for one kind of node – one whose input is satisfied from outside the graph entirely, such as a bridge receiving from a socket. Its unconnected input port is not a dangling port; it is the point where external data arrives. Such a node says so by returning True here, and the chain then admits it.

Declared rather than inferred, and False by default, so nothing existing changes and no mistake is silently accepted.

Returns:

True if the inputs are satisfied externally.

Return type:

bool

pre_setup(data, port_context_in)

Adapt the input contexts before setup() sees them.

Called once, immediately before setup, with the input port contexts that setup is about to receive. Modify them in place; the return value is ignored.

The setup-time counterpart of pre_step(). Together they let a consuming package present every node with uniform inputs: this one settles what the inputs are – how many channels, at what rate, with what timing – and pre_step supplies frames that match on every cycle. Adapting only the frames would not be enough, because a node decides its own output shape here, from exactly these contexts.

The default does nothing.

Parameters:

• data (dict) – Initial data, as passed to setup.

• port_context_in (dict) – Input port contexts keyed by port name, to be modified in place.

Return type:

None

pre_step(data)

Adapt the input frames before step() sees them.

Called once per cycle, immediately before step, with the dictionary that step is about to receive. Modify it in place.

Return True to withhold the cycle: step is not called and nothing is emitted. A hook that holds frames back needs this – without it, the only way to keep the cycle going is to fabricate a frame, which puts invented samples at the head of the stream and shifts every later one by however many were held. Returning None or False runs the cycle as usual, which is what every hook that does not hold anything back does.

This exists so that a consuming package can make the inputs of every node uniform without every node author having to know about it. The engine deliberately holds no opinion about what “uniform” means: it delivers whatever arrived, and this is where a domain layer may reshape it – resample a stream that arrives irregularly onto the grid of one that does not, substitute a value for a port that produced nothing, or reject a combination it cannot express.

The default does nothing, which is the behaviour of every node that does not override it.

Notes

This runs on the per-cycle path, so an override should do as little as possible and decide as much as it can once, in setup, rather than on every frame.

Parameters:

data (dict) – Input frames keyed by port name, to be modified in place. A port that produced nothing this cycle may be absent.

Return type:

bool

process(data)

Process the whole record at once. Override for a batch node.

Called once, with every sample the node will ever see, in place of step(). This is what a non-causal operation needs – a zero-phase filter, a whole-record statistic – and it is why the node needs no buffering of its own: a batch pipeline delivers one block.

Parameters:

data (dict) – The complete input data, keyed by port name.

Returns:

The processed data.

Return type:

dict

Raises:

NotImplementedError – If the node implements only step(). Reaching this means the caller asked for whole-record processing from a node that declares none, which the pipeline should have refused first.

abstractmethod setup(data, port_context_in)

Abstract method to setup the Node.

This method should be implemented to define the setup logic for the Node.

Parameters:

• data (dict) – The input data to configure the Node.

• port_context_in (dict) – Context for the input ports.

Returns:

The configuration of the output ports.

Return type:

dict

start()

Start the Node.

step(data)

Process one frame. Override for a node that runs in realtime.

Not abstract, because a node may instead implement process() and run only in a batch pipeline. Exactly one of the two is required, checked in __init__.

Parameters:

data (dict) – The input data to process.

Returns:

The processed data after applying the step logic.

Return type:

dict

Raises:

NotImplementedError – If the node implements only process(). Reaching this means a batch-only node was driven frame by frame.

stop()

Stop the Node.