Skip to content

ThinkGear Connector

ThinkGearConnector 

ThinkGearConnector(host: str = 'localhost', port: int = 13854, enable_raw_output: bool = True, output_format: str = 'Json')

Manage connection and communication with the ThinkGear Connector service.

This class provides methods to establish, maintain, and close connections with the ThinkGear Connector service, as well as read and write data to/from the device.

The host and port for the ThinkGear Connector service are configured with default values of "localhost" and 13854, respectively. In most cases, you will not need to modify these settings.

Parameters:

Name Type Description Default
host str

The hostname of the ThinkGear Connector service.

 'localhost'
port int

The port number of the ThinkGear Connector service.

 13854
enable_raw_output bool

Whether to enable raw EEG output.

 True
output_format str

The format of the output data. Can be either "Json" or "BinaryPacket".

 'Json'
Source code in mindwave/connector.py 
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
def __init__(
    self,
    host: str = "localhost",
    port: int = 13854,
    enable_raw_output: bool = True,
    output_format: str = "Json",
) -> None:
    """Initialize the ThinkGearConnector.

    The host and port for the ThinkGear Connector service are
    configured with default values of "localhost" and 13854, respectively.
    In most cases, you will not need to modify these settings.

    Args:
        host (str): The hostname of the ThinkGear Connector service.
        port (int): The port number of the ThinkGear Connector service.
        enable_raw_output (bool): Whether to enable raw EEG output.
        output_format (str): The format of the output data. Can be either "Json" or "BinaryPacket".
    """
    self.host: str = host
    self.port: int = port
    self.enable_raw_output: bool = enable_raw_output
    self.output_format: str = output_format
    self.st_writer: asyncio.StreamWriter = None
    self.st_reader: asyncio.StreamReader = None
    self._logger = Logger.get_logger(self.__class__.__name__)

write 

write(data: bytes) -> None

Write data to the ThinkGear Connector.

Parameters:

Name Type Description Default
data bytes

The data to be sent to the ThinkGear Connector.

 required

Raises:

Type Description
ConnectionError

If the ThinkGear Connector is not connected
Source code in mindwave/connector.py 
84
85
86
87
88
89
90
91
92
93
94
95
96
def write(self, data: bytes) -> None:
    """Write data to the ThinkGear Connector.

    Args:
        data (bytes): The data to be sent to the ThinkGear Connector.

    Raises:
        ConnectionError: If the ThinkGear Connector is not connected
    """
    if not self.is_connected():
        raise ConnectionError("The device is not connected, please connect first using the connect() method.")

    self.st_writer.write(data)

read async 

read() -> bytes

Read data from the ThinkGear Connector.

Returns:

Name Type Description
bytes bytes

The data received from the ThinkGear Connector.

Raises:

Type Description
ConnectionError

If the ThinkGear Connector is not connected.
Source code in mindwave/connector.py 
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
async def read(self) -> bytes:
    """Read data from the ThinkGear Connector.

    Returns:
        bytes: The data received from the ThinkGear Connector.

    Raises:
        ConnectionError: If the ThinkGear Connector is not connected.
    """
    if not self.is_connected():
        raise ConnectionError("The device is not connected, please connect first using the connect() method.")

    out = await self.st_reader.readuntil(b"\r")
    return out

connect async 

connect() -> bool

Establish connection with the ThinkGear Connector service.

Attempts to connect to the ThinkGear Connector service using the specified host and port. If successful, configures the connection with the specified settings.

Returns:

Name Type Description
bool bool

True if connection is successful or already connected, False otherwise.
Source code in mindwave/connector.py 
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
async def connect(self) -> bool:
    """Establish connection with the ThinkGear Connector service.

    Attempts to connect to the ThinkGear Connector service using the specified host
    and port. If successful, configures the connection with the specified settings.

    Returns:
        bool: True if connection is successful or already connected, False otherwise.
    """
    self._logger.debug("Initializing connection to ThinkGear Connector...")
    if self.is_connected():
        self._logger.warning(
            "ThinkGear connector is already connected, to reconnect with different settings, "
            "please disconnect first using the disconnect() method."
        )
        return True

    try:
        self.st_reader, self.st_writer = await asyncio.open_connection(self.host, self.port)
        self._logger.debug("Connected to ThinkGear Connector")
    except ConnectionRefusedError:
        self._logger.error(
            f"Connection to ThinkGear Connector at {self.host}:{self.port} refused!, "
            "Check if the ThinkGear Connector is running."
        )
        await asyncio.sleep(3)
        return False

    self.write(
        json.dumps(
            {
                "enableRawOutput": self.enable_raw_output,
                "format": self.output_format,
            }
        ).encode("utf-8")
    )
    return True

disconnect 

disconnect() -> bool

Disconnect from the ThinkGear Connector service.

Returns:

Name Type Description
bool bool

True if disconnection is successful or already disconnected.
Source code in mindwave/connector.py 
151
152
153
154
155
156
157
158
159
160
161
162
163
164
def disconnect(self) -> bool:
    """Disconnect from the ThinkGear Connector service.

    Returns:
        bool: True if disconnection is successful or already disconnected.
    """
    self._logger.info("Disconnecting ThinkGear Connector...")

    if not self.is_connected():
        self._logger.debug("The device is already disconnected!")
        return True

    self.st_writer.close()  # Closing st_writer automatically cleans up st_reader as well.
    return True

is_connected 

is_connected() -> bool

Check if the connection to ThinkGear Connector is active.

Returns:

Name Type Description
bool bool

True if connected, False otherwise.
Source code in mindwave/connector.py 
166
167
168
169
170
171
172
173
174
175
def is_connected(self) -> bool:
    """Check if the connection to ThinkGear Connector is active.

    Returns:
        bool: True if connected, False otherwise.
    """
    if not self.st_writer:
        return False

    return not self.st_writer.is_closing()