Collector

class gpype.backend.sinks.collector.Collector

Bases: INode

Terminates a pipeline in memory instead of in a file.

The point of an offline run is usually to get data back into Python, not into a second file. Pipeline.run() returns every Collector’s Result, keyed by node name.

In a batch run this receives exactly one block – the whole recording – and keeping it costs nothing beyond the data itself. It also works in a realtime pipeline, where it accumulates frames and therefore grows for as long as the pipeline runs; that is useful for a short measurement and wrong for a long one.

The input takes the timing of whatever is connected to it. A continuous stream is joined along the time axis, (time, channel). Epochs from an asynchronous node such as Trigger are stacked into (time, channel, trial), one trial per epoch in the order they arrived, even when only one arrived. The trial axis carries no condition labels. Epochs of differing shape are refused when the result is read.

Parameters:

**kwargs – Additional arguments for the parent INode.

COLLECTS = True

Declares that Pipeline.run() should gather this node’s result. Declared rather than inferred from having a result member, so a node that happens to expose one is not collected by accident – and so a user’s own sink can opt in deliberately.

__init__(**kwargs)

Initialize the INode with input port configurations.

Parameters:

• input_ports – List of input port configurations or None.

• **kwargs – Additional arguments passed to parent classes. A sink or a widget passes its edge_id here, and it is checked here because every one of them reaches this constructor; a node that only touches data passes none.

Raises:

• ValueError – If an edge_id is empty or padded with whitespace.

• TypeError – If an edge_id is not a string.

setup(data, port_context_in)

Record what describes the incoming stream.

Parameters:

• data (dict[str, ndarray]) – Initial data dictionary.

• port_context_in (dict[str, dict]) – Input port contexts.

Return type:

dict[str, dict]

Returns:

An empty dict – a Collector has no output ports.

step(data)

Keep the incoming block.

Parameters:

data (dict[str, ndarray]) – Input data dictionary.

Return type:

dict

Returns:

An empty dict.

property block_count

How many blocks arrived.

One after a batch run. More says the run was paced, which is worth being able to see from a test. On an asynchronous input it is the number of trials.

property result

What was collected, or None if nothing arrived.

A continuous stream is concatenated along the time axis, so a paced run and a batch run of the same recording give the same array. Epochs are stacked along a third, trial axis.

Raises:

ValueError – If epochs of differing shape arrived, naming the first shape and the first that differs.