Cyton Base

class openbci_stream.acquisition.cyton_base.CytonBase(daisy: Literal['auto', True, False], montage: Optional[Union[list, dict]], streaming_package_size: int, capture_stream: bool, board_id: str = '', parallel_boards: int = 1)

The Cyton data format and SDK define all interactions and capabilities of the board, the full instructions can be found in the official documentation.

• https://docs.openbci.com/Cyton/CytonDataFormat/

• https://docs.openbci.com/Cyton/CytonSDK/

daisy

Daisy board can be detected on runtime or declare it specifically.

montage

A list means consecutive channels e.g. [‘Fp1’, ‘Fp2’, ‘F3’, ‘Fz’, ‘F4’] and a dictionary means specific channels {1: ‘Fp1’, 2: ‘Fp2’, 3: ‘F3’, 4: ‘Fz’, 5: ‘F4’}.

streaming_package_size

The streamer will try to send packages of this size, this is NOT the sampling rate for data acquisition.

capture_stream

Indicates if the data from the stream will be captured in asynchronous mode.

__getattribute__(attr: str) → Any

Some attributes must be acceded from RPyC.

activate_channel(channels: List[int]) → None

Activate the channels specified.

Parameters

channels – 1-based indexing channels.

property aux_buffer: openbci_stream.acquisition.cyton_base.Queue

Return the deserialized data buffer for auxiliar data.

property aux_time_series: numpy.ndarray

Return auxiliar data acquired in shape (AUX, time).

capture_stream() → None

Create a process for connecting to the stream and capture data from it.

channel_settings(channels: List[int], power_down: Optional[bytes] = b'0', gain: Optional[bytes] = b'6', input_type: Optional[bytes] = b'0', bias: Optional[bytes] = b'1', srb2: Optional[bytes] = b'1', srb1: Optional[bytes] = b'0') → None

Channel Settings commands.

Parameters

• channels – 1-based indexing channels list that will share the configuration specified.

• power_down – POWER_DOWN_ON (default), POWER_DOWN_OFF.

• gainoptional – GAIN_24 (default), GAIN_12, GAIN_8, GAIN_6, GAIN_4, GAIN_2, GAIN_1.

• input_type – Select the ADC channel input source: ADSINPUT_NORMAL (default), ADSINPUT_SHORTED, ADSINPUT_BIAS_MEAS, ADSINPUT_MVDD, ADSINPUT_TEMP, ADSINPUT_TESTSIG, ADSINPUT_BIAS_DRP, ADSINPUT_BIAS_DRN,

• bias – Select to include the channel input in BIAS generation: BIAS_INCLUDE (default), BIAS_REMOVE.

• srb2 – Select to connect this channel’s P input to the SRB2 pin. This closes a switch between P input and SRB2 for the given channel, and allows the P input also remain connected to the ADC: SRB2_CONNECT (default), SRB2_DISCONNECT.

• srb1 – Select to connect all channels’ N inputs to SRB1. This effects all pins, and disconnects all N inputs from the ADC: SRB1_DISCONNECT (default), SRB1_CONNECT.

Returns

• On success –

  • If streaming, no confirmation of success. Note: WiFi Shields will always get a response, even if streaming.

  • If not streaming, returns Success: Channel set for 3$$$, where 3 is the channel that was requested to be set.

• On failure –

  • If not streaming, NOTE: WiFi shield always sends the following responses without $$$

    • Not enough characters received, Failure: too few chars$$$ (example user sends x102000X)

    • 9th character is not the upper case X, Failure: 9th char not X$$$ (example user sends x1020000V)

    • Too many characters or some other issue, Failure: Err: too many chars$$$

  • If not all commands are not received within 1 second, Timeout processing multi byte message - please send all commands at once as of v2$$$

abstract close()

Stops data stream.

command(c: Union[str, bytes]) → str

Send a command to device.

Before send the commmand the input buffer is cleared, and after that waits 300 ms for a response. Is possible to send a raw bytes, a CytonConstants attribute or the constant name e.g.

>>> command(b'~4')
>>> command(CytonConstants.SAMPLE_RATE_1KSPS)
>>> command('SAMPLE_RATE_1KSPS')

Parameters

c – Command to send.

Returns

If the command generate a response this will be returned.

Return type

str

daisy_attached() → bool

Check if a Daisy module is attached.

This command will activate the Daisy module is this is available.

Returns

Daisy module activated.

Return type

bool

deactivate_channel(channels: List[int]) → None

Deactivate the channels specified.

Parameters

channels – 1-based indexing channels.

property eeg_buffer: openbci_stream.acquisition.cyton_base.Queue

Return the deserialized data buffer for EEG.

property eeg_time_series: numpy.ndarray

Return data acquired in shape (channels, time).

leadoff_impedance(channels: List[int], pchan: Optional[bytes] = b'0', nchan: Optional[bytes] = b'1') → None

LeadOff Impedance Commands

Parameters

• channels – 1-based indexing channels list that will share the configuration specified.

• pchan – TEST_SIGNAL_NOT_APPLIED (default), TEST_SIGNAL_APPLIED.

• nchan – TEST_SIGNAL_APPLIED (default), TEST_SIGNAL_NOT_APPLIED.

Returns

• On success –

  • If streaming, no confirmation of success. Note: WiFi Shields will always get a response, even if streaming.

  • If not streaming, returns Success: Lead off set for 4$$$, where 4 is the channel that was requested to be set.

• On failure –

  • If not streaming, NOTE: WiFi shield always sends the following responses without $$$

    • Not enough characters received, Failure: too few chars$$$ (example user sends z102000Z)

    • 5th character is not the upper case ‘Z’, Failure: 5th char not Z$$$ (example user sends z1020000X)

    • Too many characters or some other issue, Failure: Err: too many chars$$$

  • If not all commands are not received within 1 second, Timeout processing multi byte message - please send all commands at once as of v2$$$

listen_stream_markers(host: Optional[str] = 'localhost', topics: Optional[List[str]] = ['markers']) → None

Redirect markers form Kafka stream to board, this feature needs markers boardmode configured.

property markers: Dict[str, list]

Return a dictionary with markes as keys and a list of timestamps as values e.g

>>> {'LEFT': [1603150888.752062, 1603150890.752062, 1603150892.752062],
     'RIGHT': [1603150889.752062, 1603150891.752062, 1603150893.752062],}

property markers_buffer: openbci_stream.acquisition.cyton_base.Queue

Return the deserialized data buffer for markers.

property montage: Union[List, Dict]

The current montage configured on initialization.

abstract read()

Read binary data.

reset_buffers() → None

Discard buffers.

reset_input_buffer()

Flush input data.

save(filename: str, montage_name: str, sample_rate: Optional[int] = None) → None

Create a hdf file with acquiered data.

Parameters

• filename – Path with the destination of the hdf file.

• montage_name – Montage name for MNE e.g ‘standard_1020’.

• sample_rate – The sampling rate for acquired data.

send_marker(marker: Union[str, bytes, int], burst: int = 4) → None

Send marker to device.

The marker sended will be added to the AUX bytes in the next data input.

The OpenBCI markers does not work as well as expected, so this module implement an strategy for make it works. A burst markers are sended but just one are readed, this add a limitation: No more that one marker each 300 ms are permitted.

Parameters

• marker – A single value with the desired marker. Only can be a capitalized letter, or an integer between 65 and 90. These limitations are imposed by this library and not by the SDK

• burst – How many times the marker will be send.

start_stream() → None

Create the binary stream channel.

stop_stream() → None

Stop the acquisition daemon if exists.

stream(duration: int) → None

Start a synchronous process for start stream data.

This call will hangs until durations be completed.

Parameters

duration – Seconds for data collection.

property timestamp_buffer: openbci_stream.acquisition.cyton_base.Queue

Return the deserialized data buffer for timestamps.

property timestamp_time_series: numpy.ndarray

Return timestamps acquired.

Since there is only one timestamp for package, the others values are interpolated.

abstract write()

Write bytes.

class openbci_stream.acquisition.cyton_base.CytonConstants

Default constants defined in the Cyton SDK