Trigger

class gpype.backend.flow.trigger.Trigger

Bases: IONode

Event-triggered data extraction node for BCI applications.

Monitors trigger events and extracts time-locked data segments around trigger occurrences. Maintains a rolling buffer of input data and outputs complete data epochs when target trigger values are detected. Commonly used in event-related potential (ERP) analysis.

DEFAULT_TIME_PRE = 0.7

Default pre-trigger window in seconds

DEFAULT_TIME_POST = 0.2

Default post-trigger window in seconds

PORT_TRIGGER = 'trigger'

Port name for trigger input

class Configuration

Bases: Configuration

Configuration class for Trigger parameters.

class Keys

Bases: Keys

Configuration key constants for the Trigger.

TIME_PRE = 'time_pre'

Pre-trigger time configuration key

TIME_POST = 'time_post'

Post-trigger time configuration key

TARGET = 'target'

Target trigger value configuration key

__init__(time_pre=None, time_post=None, target=None, **kwargs)

Initialize the Trigger node with timing and target configurations.

Parameters:

• time_pre (float) – Time in seconds before trigger to include in epoch. Must be > 0. Defaults to 0.7 seconds.

• time_post (float) – Time in seconds after trigger to include in epoch. Must be > 0. Defaults to 0.2 seconds.

• target (float) – Trigger value(s) that cause epoch extraction. Can be single value or list. Defaults to [1].

• **kwargs – Additional configuration parameters passed to IONode.

Raises:

ValueError – If time_pre or time_post is <= 0.

setup(data, port_context_in)

Set up the Trigger node and initialize internal buffers.

Validates input requirements, calculates buffer sizes based on sampling rate and timing parameters, and initializes the rolling input buffer for data collection.

Parameters:

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

• port_context_in (dict[str, dict]) – Input port context with sampling rates, frame sizes, and channel counts.

Return type:

dict[str, dict]

Returns:

Output port context with updated frame size and timing information for the extracted epochs.

Raises:

ValueError – If input frame size is not 1 or sampling rate is not provided.

step(data)

Process one frame of data and check for trigger events.

Updates the rolling input buffer with new data, monitors trigger changes, and extracts complete epochs when countdown timers expire. Multiple triggers can be active simultaneously.

Parameters:

data (dict[str, ndarray]) – Dictionary containing input data arrays. Must include both the main data port and trigger port data.

Return type:

dict[str, ndarray]

Returns:

Dictionary containing extracted epoch data when a trigger countdown completes, None otherwise. Epoch has shape (samples_pre + samples_post, channel_count).