Source code for brainaccess_board

from .database import ReadDB
from .message_queue import BoardControl
from .stream import Stimulation


__author__: str = """neurotechnology"""
__email__: str = '[email protected]'
__version__: str = '1.0.0'




def stimulation_connect(name: str = "BrainAccessMarkers") -> Stimulation:
    """ Create a new Stimulation object

    Parameters
    ----------
    name : str
        Name of the stimulation stream.

    Returns
    -------
    Stimulation
        Stimulation object to send markers to LSL stream.

    """

    return Stimulation(name=name)





def msg_connect() -> tuple[BoardControl, dict[str, str], bool]:
    """Create connection to BrainAccess Board for communication.

    Returns
    -------
    tuple: (BoardControl, dict, bool)
        BoardControl object, commands dictionary and connection status.

    """
    board_control = BoardControl(request_timeout=1000)
    response = board_control.get_commands()
    if "data" not in response:
        return None, None, True
    commands = response["data"]
    command = commands["test"]
    reply = board_control.command(command)
    if reply["message"] == "Connection successful":
        return board_control, commands, True
    return board_control, commands, False





def db_connect(filename: str = "current") -> tuple[ReadDB, bool]:
    """Connect to the database.

    Parameters
    ----------
    filename : str
        Name of the database file.

    Returns
    -------
    tuple : (ReadDB, bool)
        ReadDB object and connection status.
    """
    db_status = False
    db = None
    try:
        db = ReadDB(filename)
        if db.handle:
            db_status = True
    except Exception:
        db = None
        db_status = False
    return db, db_status