brainaccess_board package#

Submodules#

brainaccess_board.database module#

class brainaccess_board.database.ReadDB(filename='current')[source]#

Bases: object

Get current database file to read from it

Parameters:

filename (str)

get_mne(device=None, duration=None, time_range=None, only_lsl=True, marker_devices_include=None)[source]#

Get data from the database as mne objects

If no parameters are given, dictionary with all devices full data are returned.

Parameters:

  • device (str, optional) – The device identifier. Usually the LSL uid. If empty all devices are returned.

  • duration (int, Optional) – The duration in seconds from the end.

  • time_range (tuple, Optional) – The time range to get data from. If duration is set time_range is ignored.

  • only_lsl (bool, Optional) – If True, only LSL devices are included. With default templates non LSL devices are not recorded.

  • marker_devices_include (list, Optional) – List of marker devices to include. If left empty all marker devices are included.

Returns:

Dictionary with mne objects.

Return type:

dict

brainaccess_board.message_queue module#

class brainaccess_board.message_queue.BoardControl(logger=None, request_timeout=5000)[source]#

Bases: SocketClient

Control BrainAccess Board via messages

Parameters:

  • logger (Logger | None)

  • request_timeout (int)

class brainaccess_board.message_queue.SocketClient(port, commands, mode='json', request_timeout=5000, logger=None)[source]#

Bases: object

Client to send messages to server

Parameters:

  • port (int)

  • commands (dict)

  • mode (str)

  • request_timeout (int)

  • logger (Logger | None)

command(command)[source]#

Send command to board

Parameters:

command (dict) – message to send

Returns:

reply from board

Return type:

reply (dict)

get_commands()[source]#

Get all possible commands

Returns:

all possible commands

Return type:

dict

log(message, level='info')[source]#

Log message

Parameters:

  • message (str) – message to log

  • level (str, optional) – log level. Defaults to

  • values ("info". Possible) – “debug”, “info”, “warning”, “error”, “critical”

Return type:

None

brainaccess_board.sq module#

exception brainaccess_board.sq.InvalidDirectionError[source]#

Bases: Exception

brainaccess_board.sq.adapt_array(arr)[source]#

Converts a NumPy array to a binary format for storing in SQLite.

Parameters:

arr (ndarray)

Return type:

memoryview

brainaccess_board.sq.close_db(handle)[source]#
Parameters:

handle (Dict)

Return type:

None

brainaccess_board.sq.convert_array(text)[source]#

Converts a binary format back to a NumPy array.

Parameters:

text (bytes)

Return type:

ndarray

brainaccess_board.sq.get_data(handle, device=None, direction='last', count=10)[source]#

Retrieves data from a table based on the direction and count of records.

Parameters: handle (Dict): The database handle. device (Optional[str]): The device identifier. direction (str): Direction of retrieval (‘all’, ‘last’, ‘first’). count (int): Number of records to retrieve.

Returns: List: Data records.

Parameters:

  • handle (Dict)

  • device (str | None)

  • direction (str)

  • count (int)

Return type:

list

brainaccess_board.sq.get_data_after(handle, start, column, device)[source]#

Retrieves data records that have a timestamp greater than the specified start time.

Parameters: handle (Dict): The database handle. start (float): The start time. column (str): The column to compare the time. device (str): The device identifier.

Returns: List: Data records.

Parameters:

  • handle (Dict)

  • start (float)

  • column (str)

  • device (str)

Return type:

List

brainaccess_board.sq.get_devices(handle)[source]#

Lists all devices based on the tables available in the database.

Parameters: handle (Dict): The database handle.

Returns: List[str]: A list of device identifiers.

Parameters:

handle (Dict)

Return type:

List[str]

brainaccess_board.sq.get_first_timestamp(handle, device)[source]#

Gets the earliest timestamp from a data table for a specific device.

Parameters: handle (Dict): The database handle. device (str): The device identifier.

Returns: Optional[float]: The earliest timestamp if found, else None.

Parameters:

  • handle (Dict)

  • device (str)

Return type:

float | None

brainaccess_board.sq.get_handle(name, uri=False)[source]#

Establishes a connection to an SQLite database and sets up the cursor.

Parameters: name (Union[pathlib.Path, str]): Path to the SQLite database file. uri (bool): Whether to treat the name as a URI.

Returns: Dict: A dictionary containing the cursor and connection objects.

Parameters:

  • name (Path | str)

  • uri (bool)

Return type:

Dict

brainaccess_board.sq.get_last_seconds_data(handle, device, duration)[source]#

Retrieves data from the last given number of seconds.

Parameters: handle (Dict): The database handle. device (str): The device identifier. duration (int): The duration in seconds.

Returns: List: Data records.

Parameters:

  • handle (Dict)

  • device (str)

  • duration (int)

Return type:

List

brainaccess_board.sq.get_metadata(handle, device)[source]#

Fetches metadata for a given device from the database.

Parameters: handle (Dict): The database handle. device (str): The device identifier.

Returns: List: Metadata records.

Parameters:

  • handle (Dict)

  • device (str)

Return type:

List

brainaccess_board.sq.get_range_data(handle, device, time_range)[source]#

Retrieves data within a given time range.

Parameters:

  • (Dict) (handle)

  • (str) (device)

  • (tuple) (time_range)

  • handle (Dict)

  • device (str)

  • time_range (tuple)

Returns:

List

Return type:

Data records.

brainaccess_board.sq.get_table(handle, name, name2=None)[source]#

Finds a specific table based on given name patterns.

Parameters: handle (Dict): The database handle. name (str): Primary name pattern. name2 (Optional[str]): Secondary name pattern.

Returns: Optional[str]: The table name if found, else None.

Parameters:

  • handle (Dict)

  • name (str)

  • name2 (str | None)

Return type:

str | None

brainaccess_board.sq.get_tables(handle)[source]#

Retrieves the list of all tables in the database.

Parameters: handle (Dict): The database handle.

Returns: List[str]: A list of table names.

Parameters:

handle (Dict)

Return type:

List[str]

brainaccess_board.sq.query(handle, sql_query)[source]#

Executes a given SQL query and fetches all results.

Parameters: handle (Dict): The database handle containing cursor and connection. sql_query (str): The SQL query to execute.

Returns: List: Query results.

Parameters:

  • handle (Dict)

  • sql_query (str)

Return type:

List

brainaccess_board.stream module#

class brainaccess_board.stream.Stimulation(name='BrainAccessMarkers', source_id='BrainAccessMarkers')[source]#

Bases: object

Initialize LSL marker device.

Parameters:

  • name (str, optional) – Name of the LSL stream, by default “BrainAccessMarkers”

  • source_id (str, optional) – Source ID of the LSL stream, by default “BrainAccessMarkers”

annotate(msg)[source]#

Send a marker to the LSL stream

Parameters:

msg (str) – Marker message to send to the LSL stream.

Return type:

None

have_consumers()[source]#

Check if there are any consumers connected to the LSL stream

Returns:

True if there are consumers connected to the LSL stream, False otherwise.

Return type:

bool

brainaccess_board.utils module#

brainaccess_board.utils.find_free_port()[source]#

Find a free port on the localhost

Returns:

Port number

Return type:

int

Module contents#

brainaccess_board.db_connect(filename='current')[source]#

Connect to the database.

Parameters:

filename (str) – Name of the database file.

Returns:

tuple – ReadDB object and connection status.

Return type:

(ReadDB, bool)

brainaccess_board.msg_connect()[source]#

Create connection to BrainAccess Board for communication.

Returns:

tuple – BoardControl object, commands dictionary and connection status.

Return type:

(BoardControl, dict, bool)

brainaccess_board.stimulation_connect(name='BrainAccessMarkers')[source]#

Create a new Stimulation object

Parameters:

name (str) – Name of the stimulation stream.

Returns:

Stimulation object to send markers to LSL stream.

Return type:

Stimulation