Apply

class gpype.backend.transform.apply.Apply[source]

Bases: IONode

Runs a plain Python function over a whole recording.

The escape hatch for offline work: a transform that is three lines of numpy needs no class, no Configuration holder and no port declaration – gp.Apply(np.abs) is the whole node.

Batch only; a realtime pipeline is refused by start() with this node named (D-BATCH-56). An arbitrary callable has no bound on its execution time, and a closure quietly accumulating state across frames looks identical to a pure transform at the call site. Offline neither applies: there is no deadline, and the whole recording arrives in one call.

The function must return an array of the same shape and dtype it was given. Anything else is a change to the contract with the next node, which a function has no way to declare, so it is refused rather than misconfiguring whatever is downstream – write a node class for that. A function is not a configuration value either, so a pipeline holding an Apply cannot be written to a document: serialize() refuses it, naming this node.

A channel’s declared unit is carried through unchanged: the shape and dtype check already refuses a function that resizes the output, and a bare callable has no way to say what its scale did to the unit, so it is not this node’s place to guess (D-BATCH-43). Wrap a scale-changing function in a proper node class to declare the new unit.

Parameters:

  • function (Callable) – Called once with the whole recording as a numpy.ndarray of shape (samples, channels). Must return an array of the same shape and dtype.

  • **kwargs – Passed to IONode – name in particular, which is what a refusal will call this node.

Raises:

TypeError – If function is not callable.

BATCH_ONLY = True

Refused outside a batch run, by Pipeline._validate. A class attribute rather than a check inside step, so the refusal arrives from start() with the node named – and not on the first cycle, by which point a source may already hold a device.

SCRIPT_ONLY = True

Refused by Pipeline.serialize(). The function is instance state, so a document naming this node would rebuild it without one (D-BATCH-70).

__init__(function, **kwargs)[source]

Initialize the IONode with input and output port configurations.

Parameters:

  • input_ports – List of input port configurations or None.

  • output_ports – List of output port configurations or None.

  • **kwargs – Additional arguments passed to parent classes.

process(data)[source]

Hand the whole recording to the function.

process and not step: this node exists only for batch runs, and a batch run delivers the recording in one call. There is deliberately no step, so a realtime pipeline that somehow reached a cycle would fail loudly rather than silently doing nothing.

Parameters:

data (dict[str, ndarray]) – The complete recording, under key PORT_IN.

Return type:

dict[str, ndarray]

Returns:

What the function returned, under key PORT_OUT.

Raises:

ValueError – If the function returned something of a different shape or dtype, or something that is not an array at all.