Python
This tutorial explains in detail how to interact with the NeuroScale Platform API from Python (which is also separately documented here). This tutorial is meant to be read alongside the source code of the Python example client_basic.py (link), and later sections in this tutorial series cover also more advanced clients that add some features here and there, including client_lsl.py (link). These scripts are found in the public NeuroScale-Users GitHub repository.
Before actually running any of these examples, be sure to get an overview of the scripts that are there in the README.md file for that repository.
Note
If you cannot access this repository, you may have to create a GitHub account first, and then get your GitHub username added to the NeuroScale-Users organization page on GitHub (you should have received instructions about this in your welcome email, and if not, just contact support).
Installation¶
Before you start, you need to make sure that you have the right packages installed. The recommended way to install packages for Python without affecting your operating system is to do so in a Python virtual environment (venv) that you create for each specific project (see here for more information).
Tip
On Windows, the activate command that is typically used to turn on a Python virtual environment is located at "myvenv/Scripts/activate.bat" instead.
Unless you chose to skip the venv creation, once you have activated the venv, you can proceed to install the requirements in the requirements.txt file (link) that is bundled with the example scripts, as follows:
Prerequisites¶
Follows client_basic.py from here on.
To interact with the NeuroScale Platform, you need to have a package to emit HTTP requests (and parse HTTP responses), for which we use here the requests package. Furthermore, you need a library that implements the MQTT protocol (on the client side), for which we use the Paho MQTT client. Lastly, you need to be able to parse and generate the actual messages, for which you can either use JSON or MessagePack (MessagePack is recommended for production use). We use the corresponding Python packages for this purpose:
The first thing you need when connecting to NeuroScale is the URL at which the API lives. Besides the default endpoint (used in the below snippet), you may have received one or more endpoints for testing or customer-specific features, which you can use here. The below default corresponds to the "production" platform. There are also staging and testing platforms to which you may be granted access if your are using cutting-edge pipelines.
The other piece of information you need is your personal access token, explained on the Security in NeuroScale page here. Access tokens are credentials specific to a customer and should be treated with the same secrecy as passwords for security reasons. They should never be committed to repositories, and should also not be entered verbatim as command-line arguments to prevent them from leaking into your shell history file. One way to manage this key is to read it from a file that only authorized users have access to, or to hold it in a system environment variable. Access tokens are usually strings of hexadecimal digits of the form 01234567-89ab-cdef-0123-456789abcdef.
Note
If you do not have an access token yet, check your welcome email or our online documentation on how to generate one yourself using your username and password combination.
Instantiating a Pipeline¶
The API is designed according to REST principles, meaning that interaction takes place by performing HTTP requests with standard verbs (POST, GET, PATCH, DELETE) against certain URLs. The data that is being transmitted in the requests/responses represent objects encoded in JSON. All interactions need to be authorized by including an additional HTTP Authorization header whose value should be Bearer your-access-token:
Now we are ready to connect to the NeuroScale Platform API. NeuroScale provides a number of processing pipelines, which we can launch via the API and then stream data to/from a running pipeline instance. In this example, we use the "Echo" pipeline, which returns the upstream data without performing any processing (it does check your input for correct formatting, though):
The pipeline refers to a description of the processing pipeline, and not an actual running instance of it. The API then allows you to spawn an instance of a pipeline of your choice. It should be noted that spawning an instance is a costly operation on the backend (and also can take several dozen seconds), and to minimize churn it is best to reuse your running instances when running the same pipeline, or when you are testing and debugging your software. Once you have started an instance, you will get its unique identifier back, which is a string starting with "in_". You can then choose to destroy the instance after you are done with it, or alternatively you can reuse and reconfigure it if required. The example code allows you to set the name of an existing instance to connect to here (if empty, the example code will spawn a new instance):
Before we can instantiate the pipeline, we need to look up its global unique identifier (a string that starts with "pl_" followed by a hexadecimal string) from the list of available pipelines. One can obtain a list of the available pipelines by issuing an HTTP GET request to the <API-URL>/v1/pipelines resource (v1 being the API version to use) including the Authorization header that we defined earlier; using the Python requests library the call is:
Next we should check the status code to ensure that the call went through; if either the API URL or authorization credentials were invalid, this call would fail. The status code corresponding to success is 200. Upon success, we can extract the JSON-encoded response body, which contains the list of pipelines in the "data" field of the resulting object:
The result is a list of pipelines with the human-readable names, descriptions, and unique resource identifiers, which we iterate over next to find the pipeline with the desired name:
for p in pipelines:
if p['name'] == pipeline_name:
pipeline_id = p['id']
break
else:
print("ERROR: Could not find a pipeline named '%s'" % pipeline_name)
exit(1)
Also, before we can launch this pipeline we need to describe the data that we want to stream up; this metadata is part of the launch configuration for the pipeline instance. First and foremost, a pipeline instance can accept one or more independent "data streams". Each stream is a multi-channel time series with a fixed or irregular sampling rate, and the channel values may be numbers (numeric time series) or strings (e.g. event markers or other data structures). Each of the streams has a name (no duplicates allowed), a content-type (for example, EEG, eye tracking, video, etc.) and a list of channel labels (also no duplicates allowed). In this example, we will send two streams; an 8-channel EEG time series and a stream of sporadic event markers. If you have no preference for your channel labels, it is customary to name them ch1 to chN:
channel_labels = ['ch1', 'ch2', 'ch3', 'ch4', 'ch5', 'ch6', 'ch7', 'ch8']
sampling_rate = 100 # in Hz
The content-type here is named "modality" (as in measurement modality). While you can in principle use any string here, it is best to check first if there is a standard name for the modality that you are using because there may be nodes both now and in the future that can treat your data more intelligently if the modality is known. For naming we follow the conventions of the XDF multi-modal file format. Also, note that modality strings are case-sensitive.
Next we build the configuration data structure that will go into our JSON-encoded HTTP request from the bottom up, starting with the previously defined properties of the EEG stream object. In addition to the type, sampling rate and list of channel labels, each stream also has an arbitrary name (which must be unique among the streams that you send up if multiple):
eeg_stream = {"name": "myeeg", "type": modality, "sampling_rate": sampling_rate,
"channels": [{"label": c} for c in channel_labels]}
Next we define a marker stream. A marker stream is not required for many simple use cases where all that is being streamed up is a continuous time series. However, in cases where there are particular events happening at various times (e.g. button pressed, image presented, etc.) and when data needs to be analyzed in correspondence with these events, you need to have a marker stream. To identify a stream as holding markers, its type must be set to "Markers" for the pipeline instance to recognize it. In general, a sampling rate of 0 indicates that the data stream is irregularly sampled; besides markers, this also allows you to stream up irregularly sampled time series. A marker stream usually has only a single channel, and the channel label is irrelevant:
marker_stream = {"name": "mymarkers", "type": "Markers", "sampling_rate": 0,
"channels": {"label": "value"}}
The list of streams we transmit is, unsurprisingly, just a list of stream descriptors. If you (and the pipeline that you use) do not need a marker stream, you can simply omit it from this list:
Having declared what streams we will transmit, we need to declare where we want to direct them. A pipeline can have multiple intakes (here called nodes), even though most pipelines have just one node, called the "default" node. This is expressed in the following JSON object (our node descriptor):
Now we have all information at hand to write out the full metadata that we need to launch our pipeline instance. The metadata object has a list of input node descriptors and a list of output node descriptors (since pipelines can also have multiple output nodes). Since we expect that the pipeline just passes our inputs through without changes to either the number of streams and/or the number of channels, we can just reuse our input node descriptor for the output as well:
With the metadata at hand, we can now form the HTTP request body that we use to bring up a pipeline instance. We need our pipeline ID as well as the metadata to configure our instance. Also, we may choose between either JSON or MessagePack encoding for our messages:
json_fallback = False
params = {"pipeline": pipeline_id, "metadata": metadata,
"encoding": 'json' if json_fallback else 'msgpack'}
For initial tests and debugging it is fine to use JSON, but for production use you will want to use the much more efficient (faster and smaller) MessagePack encoding, especially if you stream data at moderately high rates (more than 10 channels x 100 Hz).
The request URL to launch a new instance is <API-URL>/v1/instances, request body is the object we just constructed in JSON format, and the needed headers are the Authorization header that we constructed previously, as well as the Content-Type: application/json header. Also, the HTTP verb is POST. Using the requests package, a pipeline instance is launched as follows:
The Content-Type header is implicit in the above call since we used json= to define the body. If the request succeeded you are now consuming resources in the cloud that you will need to release at some point, therefore please read on!
Note that the above request is used to create a new instance if we do not already have one that we want to reuse. If we do however want to recycle an existing instance, we need a slightly different request. For this, the HTTP verb is PATCH, and the URL is <API-URL>/v1/instances/<instance-id>. The headers and body are the same, but using the requests package things look a bit more complicated since the patch() function has no built-in JSON support (this more closely resembles the call that one would make in another programming language):
else:
r = requests.patch(api_url + '/v1/instances/' + instance_id,
headers={'Authorization': auth_header['Authorization'],
'Content-Type': 'application/json'},
data=json.dumps(params))
Reconfiguring a running instance can be orders of magnitude faster than destroying the old instance and launching a new one (depending on the pipeline used to launch the instance). To complete the instance creation, we will want to check the status code of the HTTP response, for instance in case our URL or body was malformed in some way. Generally, a response code in the 200 range will indicate success, but note that the response code signaling success is 201 if you created a new instance, and 200 if you recycled an existing one:
Important!
You cannot reconfigure an instance to use a different pipeline (any attempt to do so will result in a HTTP 400 Bad Request error). You can only reset an existing instance or update an existing instance by changing its runtime parameters.
If instance creation was unsuccessful there is no need to clean up; otherwise you need to look up the instance ID from the response body, and use that later to destroy the instance again. To prevent accidental leakage of resources, instances will destroy themselves by default after running continuously without having received data for a few hours (this default can be changed in the instance creation HTTP request body). The instance ID of interest is simply the "id" field of the response body:
You may find it instructive to inspect the remaining data that you get in the response body, which includes various other information about the instance and its input/output endpoints.
Waiting Until the Instance is Ready¶
The instance will not be immediately available to process data, and therefore you will want to wait until it has fully launched. This can be done by periodically polling the state of the instance until that state has changed to running (initially it will be configuring).
The HTTP request to check the state is a GET with URL <API-URL>/v1/instances/<instance-id>, empty body, and Authorization header as before. The response will be again a JSON-encoded object of the same form as what was returned after instance creation, and the instance state of interest is in the "state" field:
last_state = ''
while last_state != 'running':
r = requests.get(api_url + '/v1/instances/' + instance_id,
headers=auth_header)
state = r.json()['state']
if state != last_state:
print(state + '...')
last_state = state
time.sleep(1)
Obtaining the MQTT Input/Output Endpoints¶
At this point, you have a running instance, but you do not yet know the concrete endpoints over which to send/receive data to communicate with the instance. These endpoints can be looked up from the response body that corresponded to the <API-URL>/v1/instances/<instance-id> resource (same for the POST, GET and PATCH request cases). The list of endpoint URLs is in the field "endpoints", under "data". Each endpoint has a "url" field, as well as a "mode" field (which is either "read" for pipeline instance outputs or "write" for pipeline instance inputs):
def get_endpoint(x, mode='read'):
url = [e['url'] for e in x['endpoints']['data'] if e['mode'] == mode]
return urlparse(url[0])
read_endpoint = get_endpoint(body, mode='read')
write_endpoint = get_endpoint(body, mode='write')
Setting up Communication With the Instance¶
Now that we have the endpoints, we can establish communication with the instance. Data transmission is done via MQTT, so we first create an MQTT client (the userdata is just for our own use in our callback function implementations). Since we have two endpoints, we need to create two clients, one per endpoint:
Using the Paho MQTT client, the actual work is implemented in callback handler functions, which we go over next. Once our reader has connected to the host, we need to subscribe to a "topic", which is the MQTT way of specifying what messages shall be received. The topic is part of the URL of the read endpoint (specifically, the path of the URL) and begins with a '/' character. However, the topic namespace is hierarchical (like directories), and the read endpoint gives only the super-topic under which the pipeline instance generates all its data. Specifically, for every output node of the pipeline instance there will be a subtopic of that name (typically this will be a single subtopic named "default"). So if the read endpoint was, for instance, mqtt://myhost.com:1883/testing then the default node would publish its data under the topic /testing/default. If at some point we chose to listen to other nodes of a pipeline instance, we would have to subscribe to those subtopics as well. However, to make things simpler we can just subscribe to all subtopics, using the '#' (multi-level wildcard) subtopic as done below. We can then filter the received messages upon receipt according to their topic:
def on_connect(client, userdata, flags, rc):
if userdata == 'reader':
client.subscribe('/' + str(read_endpoint.path[1:]) + '/#')
We will also need a handler that interprets the messages that we receive from the pipeline instance. Messages typically carry chunks of time series data, both in the upstream (raw data) and in the downstream (processed data):
Lastly, we can increase the robustness of our client by making it automatically reconnect to the MQTT broker if it ever gets disconnected. The disconnection can happen, for example, during debugging when the program is paused for too long; in those cases there may be a backlog of messages accumulating in the broker which can cause it to disconnect the offending client:
def on_disconnect(client, userdata, *args):
if userdata == 'reader':
reader.connect(read_endpoint.hostname, read_endpoint.port)
else:
writer.connect(write_endpoint.hostname, write_endpoint.port)
Most MQTT implementations also support other handlers, such as on_subscribe(), on_publish(), on_disconnect(), and on_unsubscribe() for diagnostics. Having defined the handlers, we can complete the MQTT setup:
reader.on_connect = on_connect
reader.on_message = on_message
reader.on_disconnect = on_disconnect
writer.on_connect = on_connect
writer.on_disconnect = on_disconnect
The next step is to actually connect the reader and the writer to their respective endpoints:
reader.connect(read_endpoint.hostname, read_endpoint.port)
writer.connect(write_endpoint.hostname, write_endpoint.port)
The actual MQTT processing using Paho needs to be managed by either calling a function periodically from our main thread, or by launching a background thread to handle messages, which is what we do here. Using a background thread is usually the better idea since when manually advancing the MQTT message pump from the main thread, each function call will only pick up a single message, and so the program can start to fall behind. Note also that we run a thread not only for the reader but also the writer, because even the writer may need to be updated to perform its connect/subscribe/publish tasks:
This concludes the connection establishment.
Streaming Data Up¶
Once all is set up, we are ready to stream data to the cloud. Since we have already declared the metadata of our streams, all we need to do is periodically send up chunks of the actual time series data which we might be getting from some sensor hardware (such as an EEG headset). The following example will skip the hardware aspect and stream up randomly generated data (there is a separate section dedicated to alternatively getting the data from actual hardware).
First off, the following steps will be performed in a main loop:
Next we generate a chunk of random "samples". Each sample is a multi-channel measurement at a particular point in time. For efficiency, you will usually want to transmit multiple samples at a time (unless the data source has a very low or irregular sampling rate), here 10. The generated chunk is a list of samples, each of which is a list of values (one per channel):
chunk_samples = 10
samples = [[random.random() for _ in range(len(channel_labels))]
for _ in range(chunk_samples)]
In NeuroScale, each transmitted sample has an associated timestamp, which indicates the time at which the underlying data was originally measured. Since we are generating samples, we also need to timestamp them. These timestamps should be floating-point numbers (double-precision) in seconds. If you do not transmit multiple streams that shall be synchronized, it does not matter what clock you read them off of and how accurate they are, so you can use the system clock with no repercussion. Generally if you send a chunk, you need a timestamp for each sample, and a simple way to do that is to use the current clock reading for the last sample in the chunk, and successively smaller timestamps (by 1/sampling_rate) for the preceding samples, like we do it here:
If you do need to synchronize multiple streams with each other, you should make sure that the timestamps for each stream are read off the sample clock. Also, if precision better than \~50ms is needed, you need to consider that by the time you get the sample from the hardware, it will already be a few milliseconds old (having travelled through the device, USB, various buffers, the driver stack, and the OS). If you know the average delay of your device, you can subtract this number from your timestamps to achieve accurate synchronization. For some hardware vendors and/or data acquisition systems (such as the Lab Streaming Layer), you may be able to inquire about the device latency if needed, or alternatively you can stimulate a signal in both of your devices, record the data, and then manually determine the relative delay between the two (which is all that matters).
When you attempt to achieve millisecond-level synchronization between devices, you may be worried about jitter in your timestamp measurements — it is important to note that NeuroScale will internally remove the jitter using a Kalman filter, so you do not (and should not) have to worry about it. Attempting to filter the jitter by hand will usually make it (much) worse.
The array of samples and timestamps is all we need to construct the chunk object for transmission. The chunk has its samples in a field named "samples", the timestamps in a field named "timestamps", and a name that needs to match the one that was declared in the metadata for that stream in the field "name". This object will later be encoded using either JSON or MessagePack:
For completeness we now also generate a dummy chunk containing some event markers. This chunk also has an array of samples, each of which represents a single event marker, and is stored as a list of one string (it is technically a 1-channel sample with string values instead of numeric values). The string defines the type of event (and/or its metadata if so desired) and the naming should match the way in which the pipeline expects these events to be formatted (e.g. a pipeline might require that button presses are indicated by a marker named "button"). The timestamps are simply the times at which the events happened, again read off the same clock as the signal of course. In this example, we are simply using three different time points that each lies in the past by a fraction of a second. In a real application, these events will have been generated elsewhere in the system and will have been time-stamped already at the time when they occurred, and then would be transported to the part of the program that is responsible for streaming markers up to the cloud. Also, while it is possible to have multiple marker streams, most pipelines will expect that there is only one marker stream which has all markers in it:
markers = [['mrk1'], ['mrk2'], ['mrk3']]
marker_stamps = [now-0.025, now-0.01, now-0.05]
marker_chunk = {'name': 'mymarkers', 'samples': markers,
'timestamps': marker_stamps}
Having defined all the data, we can define the object that shall be serialized into a message. This is simply a dict/struct with the field "streams" holding a list of chunks (one per stream):
It is not necessary that a stream be present in every message (e.g. if no event happened, the marker chunk might be omitted). Also, it is not strictly necessary that the chunks in a single message overlap in time (e.g. the markers could be older than the EEG). It is important to realize that, as long as all the data is correctly time-stamped, any sort of transmission delays of one or more streams do not change the output values of the pipeline instance (within accuracy limitations), just their delivery time; however, it is best to not unnecessarily delay the transmission of data, and to consult the documentation of the specific pipeline that you use to see if there are any limitations on how much streams may be delayed with respect to each other at transmission time (for instance, one pipeline may have to delay its output until data from a straggler stream has arrived or another pipeline may discard delayed information in order to make a delivery deadline).
Once the message has been defined, we need to encode it either into JSON or MessagePack (the encoding used needs to match the one that was declared in the pipeline instance's launch parameters):
# depending on the instance configuration, we need to encode either
# via JSON or MessagePack
if json_fallback:
msg = json.dumps(msg)
else:
msg = bytearray(msgpack.dumps(msg))
Lastly, we can send off the message bytes by publishing it via MQTT. Same as for the read endpoint, the write endpoint contains the topic on which data shall be published as part of the URL (beginning with a '/'). Any messages that are published directly under the super-topic for the instance will go into the first input node that was declared when the pipeline was instantiated. If a pipeline has multiple input nodes, then messages can be addressed to different nodes by publishing them on a subtopic named the same as the respective node (e.g. "default" for the default node):
Lastly, we end the main loop by sleeping for the interval between any two successive chunks, which is chunk_samples/sampling_rate:
Streaming Data Down¶
Once the pipeline instance is producing output, the data will be received by the MQTT reader client, and will arrive at the on_message() handler function introduced earlier. For pipelines with multiple outputs this function should first check that the topic matches the one of interest (unless that is already taken care of by subscribing to only the desired output node's topic):
Next, the payload of the message needs to be parsed. The message format is the same as for sending data, and is encoded using JSON or MessagePack, according to what was declared when the pipeline was first instantiated:
Once decoded the list of chunks (one per output stream) can be obtained from the "streams" field of the decoded object. Just like a single node can take in multiple streams simultaneously, an output node can also produce multiple output streams. In the case of the Echo pipeline, the output streams are the same as the input streams, and, in fact, the messages should have the same content. In the following Python code, the 'b' prefix is only necessary because the message decoders happen to produce byte strings rather than Unicode strings:
Next we go over the list of stream chunks and print the contents. As in the upstream, each chunk has a name, an array of multi-channel samples (each an array of values), and an array of timestamps (one per sample):
for stream in streams:
name = stream[b'name']
samples = stream[b'samples']
stamps = stream[b'timestamps']
print(' stream %s: %i samples' % (name, len(stamps)))
Instance Teardown¶
The example client performs a few additional steps prior to the teardown to offer the user a prompt to destroy or retain the instance for later reuse. A similar prompt should probably be implemented in custom clients that are used during research and development. It might be beneficial to more intelligently carry over the instance ID from one program run to the next as done here (for simplicity).
Tearing down an instance amounts to emitting a HTTP DELETE request with the URL set to <API-URL>/v1/instances/<instance-id> and passing in the Authorization header, as in:
r = requests.delete(api_url + '/v1/instances/' + instance_id,
headers={'Authorization': 'Bearer ' + access_token})
When performing a deletion, the status code should be checked to ensure that the instance was actually deleted. The status code should be 204 for success:
Getting Data from the Sensor¶
Follows client_lsl.py from here on.
This file builds on top of client_basic.py and differs from it only in how the raw data is obtained (and how the received data is further handled); it is otherwise identical.
To get the data from your sensor, you will typically have to use the SDK or API provided by the hardware vendor. Most devices give you access to the raw data stream, and many vendors also provide some simple processing of their own, which we are not interested in here.
To make life simpler, there is a general-purpose API/SDK to get data from a wide range of sensor hardware which supports the majority of the EEG devices on the market. This is the Lab Streaming Layer (LSL). In a production environment, LSL is basically an inter-process communication system that allows you to receive time series from another process (namely, the one that talks to the vendor-specific hardware); however, it also works across the local network without any change (except on WiFis in public areas). Here we are looking for a data source on the network that has a specific name (this is typically the name of the vendor or the name of the sensor model). You will most likely have to change this name to match whatever hardware you happen to have; the name used here is the default output stream name of the NeuroPype authoring environment (which can be used for playback and other testing purposes, but which we won't discuss in this document).
The example program also shows how to stream the processed EEG data out over LSL for use by other applications (by declaring an output_stream and setting up the corresponding LSL objects), which you most likely will not need in your code. Therefore we do not explain this section here.
The first thing that we do is discover the desired data source on the network by its name:
If you do not know the name you can also find a data source by content-type (i.e. a measurement modality like "EEG"), by using 'type', 'EEG' as query parameters instead. However, there is some potential for confusion because in practice you may have more than one EEG stream on the network at a time (e.g. the LSL output stream of another running instance of this program). This call will block until a stream has been found (though you can give it a timeout, and there is also another facility for non-blocking resolving in the background called the ContinuousResolver).
Now we have found a list of one or more matching streams (which you could go over and inspect to find the one that you wanted if your query was not specific enough). Next we create a stream inlet to read from it:
Since we need to tell NeuroScale the metadata of our stream, like the sampling rate, content-type, and channel labels, it is good practice to look up these properties from the stream inlet that we are connected to:
As for the channel labels, not all data sources provide proper channel labels (especially some that were written in a hurry), so we should be prepared for the following code to fail (note in other programming languages you will want to check the return values instead of try/catch):
desc = info.desc()
try:
channel_labels = []
ch = desc.child('channels').child('channel')
while not ch.empty():
channel_labels.append(ch.child_value('label') or '(no label)')
ch = ch.next_sibling('channel')
except:
# not all data sources provide proper channel labels, but NeuroScale
# requires them
channel_labels = ['Ch' + str(k) for k in range(info.channel_count())]
The last step is to get the next available data chunk from the sensor, replacing the code that randomly generates the samples and their timestamps. The pull_chunk() function as used here is non-blocking and will give you all new data since the last call (though you can use a timeout). It is quite possible that there is no new data since the last call, in which case you will want to skip the part that sends the chunk to NeuroScale:
Note
If you are recording from multiple devices via LSL and need them to be time-synchronized, you need to add the value of inlet.time_correction() to each returned timestamp.
Lastly, since we are no longer producing chunks of predetermined length like in the basic client, our sleep interval is now a constant. Keep in mind that if you make this interval too short, you may be sending too many messages per second to the cloud for the pipeline instance to pick up, in which case it will start to lag behind:
Using Basic Prediction Pipelines¶
Follows client_concentration_lsl.py.
This file builds on top of client_lsl.py and the main difference is that the return stream has a different number of channels than the upstream.
To run non-trivial pipelines, the first thing to keep in mind is that the channel labels and channel types for the data that you stream up matter, and need to be accurate where possible (i.e., try to avoid using 'Ch1', 'Ch2', if you can do better). For EEG, the canonical channel labels are in the so-called 10-20 labeling system, but not all headsets provide them (some have custom labels or no labels at all). It is also important that the labels /must/ be unique, so if they aren't, you need to alter them.
Also, if you have a mix of channels of different types that you got from your hardware and you want to stream them all up, for instance EEG, EMG, ECG, EOG, AUX, Trigger, Accelerometer, Gyroscope, Respiration, Unknown, and so on, make sure that you set their types as such (you can use the previous labels as a guide, or at least don't set their type to 'EEG' if they aren't actually EEG since that would likely throw off the pipeline).
The following code snippet shows how to run a non-trivial pipeline that computes a concentration measure from a data stream collected with the InteraXon Muse.
This headset has 6 channels, the first 4 of which are EEG channels in the 10-20 system, and the last 2 are both called 'n/a' (so they aren't unique). In this case we run the muse-io command-line program provided by InteraXon to stream the data out on the local network using LSL, and we name our LSL stream MyMuse1.
Then we pick up a stream with the same name in our client:
Now it's time to declare our pipeline meta-data, and we define an EEG stream going up. Here we just use all the 6 channels provided by the LSL stream and we somewhat incorrectly set their type to 'EEG', but that happens to be fine since the pipeline we selected was purpose-built for Muse (as indicated by the name) and it knows that only the first 4 channels contain useful EEG.
eeg_stream = {"name": "myeeg", "type": modality, "sampling_rate": sampling_rate,
"channels": [{"label": c} for c in channel_labels]}
The interesting bit is now when we declare the return stream that we expect to get back from NeuroScale. Most importantly, we expect this stream to have only a single channel (and that channel contains the estimated concentration level of the user). The channel label happens to be not checked, so we could put anything (here we use 'concindex'). This pipeline, like most others, will not rename the stream that you are sending up, so if you called it 'myeeg' in the upstream, it will still be called this way in the downstream, and consequently our definition looks as follows:
conc_stream = {"name": "myeeg", "type": modality, "sampling_rate": sampling_rate,
"channels": [{"label": "concindex"}]}
Technically, the sampling_rate of the returned stream for this particular pipeline is lower than what we stream up, since the pipeline will compute only one value for each chunk that we send up, instead of for each sample (so if we send up 10 chunks per second, the return sampling rate will be 10Hz), however, the actual processing pipeline generally does not take into account what sampling rate you declare on the return stream (this is only useful if you write, say, a second program that reads from the returned stream, and you want to communicate the correct sampling rate to it).
Now we are ready to buld the overall pipeline meta-data, which requires us to declare the name and stream declaration for both the input nodes ('inlets') of the pipeline as well as the output nodes ('outlets'). These names must match what the pipeline actually provides, and you can always inspect the description field in the pipeline resource (the one you receive when you query all pipelines with the desired name), but if a pipeline has multiple input/output nodes and you only want to read/write to a subset of them, it is legal to omit the declaration (in this case, no output will be transmitted over that node, which can save you network bandwidth). Almost all pipelines have a single input node called 'default' by convention, and most pipelines have an output node called 'default' that carries the primary output of the pipeline (here the concentration measure), plus perhaps some miscellaneous outputs carrying intermediate processing results (like some partially filtered data for debugging/inspection). The only exception are pipelines that have no single primary output node, so there is no 'default' output. So thus the declaration almost always looks like this:
node_decl_in = {"name": "default", "streams": [eeg_stream]}
node_decl_out = {"name": "default", "streams": [conc_stream]}
# the complete meta-data is now just the list of input and ouput nodes
metadata = {"nodes": {"in": [node_decl_in], "out": [node_decl_out]}}
This meta-data then becomes part of the overall creation parameters JSON data structure that goes into the REST API call, same as for the other clients described in previous sections:
params = {"pipeline": pipeline_id, "metadata": metadata,
"encoding": 'json' if json_fallback else 'msgpack'}
If this seems to tedious, there is some relief for you: many pipelines that have been customized for a particular headset (e.g., the Muse), already have the correct input/output meta-data declared, so you don't need to override it, and can omit the field. However, this does not apply to all pipelines as of this writing, and you may have to declare it, after all.
The handling of the MQTT streams looks basically the same as for client_lsl, so is not repeated here.
Overriding Pipeline Parameters¶
Follows client_parametric_lsl.py.
This file builds on top of client_concentration_lsl.py and the main difference is that parameter overrides are included in the pipeline launch configuration.
Some of the pipelines on NeuroScale expose parameters which can be set at launch time, which alter the processing in certain ways; for instance, the list of channels, frequency bands, time windows, or event markers of interest can be specified for some pipelines, and some pipelines offer parameters that allow for tweaking of the algorithms, like change the dimensionality of the data, or the time constant of some filters, and so on.
A given pipeline may expose a list of named parameters whose values can be overridden; the data type accepted by pipelines is almost always string, unless specified otherwise.
The following code snippet declares two parameters of a pipeline; in this case this is a pipeline that predicts whether the user is imagining left or right hand movements (called Motor Imagery (Generic)). This pipeline requires calibration data to be streamed up (explained in detail in a later section), that calibration data is expected to contain embedded event marker strings of different types that mark up signal portions that contain example data of the different types of movement.
One of the two exposed parameters is a list of strings called class_markers, which allows you to override which of the markers encodes the first class (kind of movement), the second class, and so on. This parameter happens to be an exception in that it is not a string but actually a list (array) of strings. The other parameter is the list of channels to keep, which is a string that is a Python slice or list expression (e.g., '0:5'). Here we set it so that it keeps all but the last two channels. The following code snippet declares these parameters:
The next snippet forms a dictionary that will later be inserted into the JSON launch parameters:
# now build the parameter structure
parameters = {"keep_channels": keep_channels, "class_markers": class_markers}
And lastly, we include those parameters in the launch configuration, together with the other options that you should be familiar with by now:
params = {"pipeline": pipeline_id, "metadata": metadata,
"parameters": parameters,
"encoding": 'json' if json_fallback else 'msgpack'}
That's all that needs to happen to override parameters. By the way, do not confuse pipeline parameters (which govern the processing) with pipeline properties (which allow you to annotate your pipeline with additional information), discussed later.
Sending up Calibration Data¶
Follows client_calibrated_lsl.py.
This file builds on top of client_parametric_lsl.py and the main difference is that the first MQTT message is a whole calibration recording. Other differences include the fact that a different kind of pipeline is being used (one for detecting user errors called Error Detection (Generic)).
Sending up calibration data is in principle a relatively simple business. Fundamentally, there are three main ways in a pipelines can consume calibration data (and each pipeline implements exactly one of them, or doesn't involve any calibration at all).
On-The-Fly Calibration The trivial case is when a pipeline calibrates itself on regular streaming data 'on the fly', and produces outputs right away (although the initial few outputs may be less accurate while the pipeline is not yet fully adapted); in this case, you may not even have to think about the fact that the pipeline is calibrating itself (although it will typically be documented in the pipeline description).
Fixed Amount of Calibration Data A slightly more interesting case is when a pipeline requires a certain minimum amount of calibration data until it is ready to produce output (e.g., 20 seconds; this will generally be documented in the pipeline description): in this case you just need to be aware of the fact that everything is fine even though you're not yet getting any data. Quite a few pipelines that include artifact removal are doing things that way.
Demarcated Calibration Data And the third case is when a pipeline will only produce output after it has seen calibration data, and you need to specify when the calibration data starts and when it ends using event markers (see initial example at the top of the page for how to transmit event markers and how to declare that you have a marker stream). Typically, this will require you to insert one event marker when you calibration period begins, which is most commonly called 'calib-begin' unless documented otherwise, and another marker when you calibration ends, which is usually called 'calib-end'. Some pipelines may expose parameters (see previous section) that allow you to change the event strings that encode these two conditions.
In some cases you might choose to reuse calibration data that you have already recorded before and perhaps written to a file. Using this is as simple as streaming it up before any real data follows, which you can do faster than real time (i.e., at your max upload bandwidth), and you can even send it all up in a single MQTT message as long as it doesn't exceed maxium MQTT message size (256MB).
The example script for this section implements the latter option, where calibration data is explicitly demarcated, and where all the data is sent up in a single message. This example script happens to load the calibration data from a .xdf file (Extensible Data Format), which you can record, for instance, using the LabRecorder application that comes with the lab streaming layer (LSL) distribution; this works especially well if your hardware supports LSL.
We won't go into the details of parsing the XDF file format here since you can use any file format you like, since at the end of the day all that NeuroScale expects you to do is to send one or more MQTT messages up of the same format as your regular streaming data.
When the import code finishes, it leaves behind a list of marker strings and a list of marker timestamps, as well as a list of EEG samples and a list of EEG time-stamps. Our script here ensures that calibration markers are inserted at the beginning and the end of the data, respectively, as shown in the following code snippet:
# optionally add calibration markers
if add_calib_markers:
# we insert the begin marker a second after the beginning of the EEG
calib_markers.insert(0, ['calib-begin'])
calib_marker_stamps.insert(0, calib_eeg_stamps[0] + 1.0)
# ...and the end marker a second before the end of the EEG
calib_markers.append(['calib-end'])
calib_marker_stamps.append(calib_eeg_stamps[-1] - 1.0)
The reason for the "if" statement is that the imported data might already contain these begin/end markers depending on how you originally created these data, and you should never send up multiple repeated calib-begin and/or calib-end markers as that might produce unexpected behavior.
The other calibration-specific bit about this example is where a message if being formed that contains the calibration data. First, we form the chunk of EEG data from what we imported, and then the chunk of marker data:
# construct the EEG chunk of the message
eeg_chunk = {'name': 'myeeg', 'samples': calib_eeg,
'timestamps': calib_eeg_stamps}
# construct the marker chunk of the message
marker_chunk = {'name': 'mymarkers', 'samples': calib_markers,
'timestamps': calib_marker_stamps}
... and then we build the message data structure that bundles the two parallel data streams:
Then we serialize it either in JSON or msgpack (congruent with the encoding that we declared in the launch parameters):
... and finally we publish it just like any other message.
# send the message (note: max size per message is 256MB, so if yours
# is longer, you need to spread it out over multiple messages)
writer.publish('/' + str(write_endpoint.path[1:]), msg)
Calibration Data for use with Supervised Machine Learning¶
When you use a pipeline that implements supervised machine learning, this means that your calibration data must include examples of EEG (or other raw data) that was produced while the person was in a particular known state, so that the system can learn how to predict the state from the examples. Typically a person may be in more than one possible state (and the goal is to predict which one it is), and so you need examples for each state.
In general, unless you know exactly what you are doing, we suggest that you lay our your calibration data according to the following sequencing: first, you almost always need more than one example per state (e.g., at least 10, or even as much as 50 or more repetitions if you want great calibration, though how much you need depends on the type of state you want to predict). Second, you must alternate between the different states, i.e., you cannot have a recording that is first all examples of state A and then all examples of state B, and a great way to do that is to randomize the order.
Now, since you have a mix of examples of the two states, you need to mark up the examples with labels, which would be encoded in event markers, so that the pipeline knows which is which. Typically, you place a marker either at the beginning or the end of the period during which the person is in a particular state (or sometimes in the middle). So in addition to the markers, the pipeline that is supposed to process the data needs to 'know' where, relative to your markers, the relevant portion of the EEG lies where the person was in the annotated state, i.e., the time window. A pipeline may either assume a hard-coded time window (and then it would also prescribe in advance where you need to place the marker), or the time window is a parameter of the pipeline.
Lastly, since you need to collect markers and store them with the data, implement randomization, and put the user into a series of known states, the calibration data will usually be collected while the person is interacting with a program you wrote that performs these tasks (e.g., it may show the person happy/sad pictures to get them into these two states, or instruct verbally or textually to imagine a happy or sad scenario, or play happy/sad music, and so on..). When you design this program it helps to google the psychology or cognitive science literature on experiment designs in which people were put in the states that you are interested in (e.g., emotion, attention, focus, etc.), and to scan for any valuable info on what pitfalls to avoid ('confounds') or how to stimulate the person accordingly. Oftentimes of course the stimulation in such literature is extremely simplistic and not of the quality in which, say, a video game or modern phone app would be able to present calibration stimuli.
For instance, the Motor Imagery pipeline is usually calibrated using a program in which the person is textually instructed to imagine a movement of either the right or left hand (or, e.g., the feet, and so on); the simulus is typically a single letter that's displayed in the center of the screen (e.g., 'L' or 'R') in a comfortably large font size, the background is neutral and non-distracting, the pacing is comfortable for the user (e.g., a letter every 3 or 4 seconds), and the total number of trials is between 20 and 50 per state ('per class' in the machine learning terminology), there may be one or more scheduled pauses for the user to recover, and there may be an initial user training period where the user can familiarize herself with the task for some time without polluting the calibration data with bad examples. Then, for each letter shown the program emits and logs a string marker (like 'left' or 'right') that goes along with the EEG data (markers can, for instance, be emitted using the lab streaming layer and recorded time-synced with the EEG data into a .xdf file, or your calibration program could be managing the marker generation and collection, EEG data acquisition, time-stamping, and NeuroScale streaming all at once).