Skip to content

Modify Parameters in Real-Time

This tutorial covers how to change the parameters of streaming instance in real-time using the NeuroScale API.

In order to follow along with this tutorial and test out its code on your end, your Python environment should be installed and setup; please look over the Python documentation if this is not the case. Also see NeuroScale Platform and Run a Streaming Pipeline (API) for information on streaming instances.

Importance

Real-time parametric updates of running pipelines is a power feature of NeuroScale and critical in BCI applications.

Extract Pipeline Information

In order to update parameters at any time, the pipeline of interest's ID, parameter format (names, values), and metadata must first be extracted from NeuroScale. This can be done getting all available pipelines by name with the requests.get REST command, matching the pipeline of interest, and saving the ID, parameters and metadata.

Prerequisite Information

Lets start our code with some prerequisite user information:

import requests

# we need to define the Neuroscale API URL to use
api_url = 'https://api.neuroscale.io'

# you also need to provide your own access token; for security
# reasons it is best to keep this string out of your code commits and instead
# read it from an environment variable or file (if you do not have a token,
# check our online documentation on how to generate one from your
# username/password combination)
access_token = '<your-access-token>'

# construct the HTTP header that will be used for authorization
auth_header = {'Authorization': 'Bearer ' + access_token}

# next we need to identify the pipeline that we want to launch on Neuroscale
pipeline_name = 'Simple Bandpower'

The access_token will be set to one of the authorization tokens for your account. (See Create Access Tokens for more info on access tokens.) The auth_header creates the HTTP header using the access_token. The api_url should be set the NeuroScale API URL shown in the above code snippet (unless instructed otherwise under special circumstances). The pipeline_name is the name of the pipeline we want run.

Find Pipeline and Extract Information

This prerequisite information will be used to find the pipeline of interest and extract the pipeline details:

# init pipeline details
pipeline_id = None
pipeline_parameters = None
pipeline_metadata = None

# next we will look up the pipeline of interest by name from the list
# of available pipelines
r = requests.get(api_url + '/v1/pipelines', headers=auth_header)
if r.status_code == 200:  # 200: ok
    body = r.json()
    pipelines = body['data']
    # now go through all pipelines and pick the one that has a matching name
    for p in pipelines:
        if p['name'] == pipeline_name:
            pipeline_id = p['id']
            pipeline_parameters = p['parameters']
            pipeline_metadata = p['metadata']
            break
    else:
        print("ERROR: Could not find a pipeline named '%s'" % pipeline_name)
        exit(1)
else:
    print("ERROR: Could not query available pipelines (HTTP %s); check "
          "your API URL and credentials." % r.status_code)

if pipeline_id is not None and pipeline_metadata is not None and pipeline_parameters is not None:
    print("pipeline_id = %s" % pipeline_id)
    print("pipeline_parameters = %s" % pipeline_parameters)
    print("pipeline_metadata = %s" % pipeline_metadata)

Here all available pipelines are found with the requests.get REST command, the pipeline of interest is matched by name, and the pipeline of interest's ID, parameters and metadataare extracted.

Running this code will result in the following output:

pipeline_id = pl_LuDcaNQm6eAfnHabnKZGKJ
pipeline_parameters = {'win_len': {'value': '1', 'type': 'string', 'verbose_name': None, 'expert': False, 'is_filename': False, 'category': '', 'description': 'Length of Moving Average window in seconds'}}
pipeline_metadata = {'nodes': {'out': [{'streams': [{'channels': [{'label': 'dummy'}], 'type': 'Control', 'name': 'signal', 'sampling_rate': 10.0}], 'name': 'default'}], 'in': [{'streams': [{'channels': [{'label': 'dummy'}], 'type': 'EEG', 'name': 'signal', 'sampling_rate': 0.0}], 'name': 'default'}]}}

Updating Parameters

Lets take a closer look at the pipeline parameters pulled from your pipeline of interest:

pipeline_parameters = {'win_len': {'value': '1', 'type': 'string', 'verbose_name': None, 'expert': False, 'is_filename': False, 'category': '', 'description': 'Length of Moving Average window in seconds'}}

Here you can see Simple Bandpower's pipeline parameter win_len, (Length of Moving Average window in seconds) and its accompanying details are accessible, which means you can adjust its value to your liking before each launch. For example, if you want a longer moving average window you can set the win_len value to 60 seconds:

pipeline_parameters = {'win_len': {'value': '60', 'type': 'string', 'verbose_name': None, 'expert': False, 'is_filename': False, 'category': '', 'description': 'Length of Moving Average window in seconds'}}

Multi Parameter

Lets check out another pipeline's parameters. This can be done by changing the name of our pipeline:

# next we need to identify the pipeline that we want to launch on Neuroscale
pipeline_name = 'Artifact Removal (Streaming)'

And running our same code. Here is the Artifact Removal (Streaming) pipeline's output:

pipeline_parameters = {
  'calib_duration': 
{'is_filename': False, 'description': 'The amount of calibration data to gather at the beginning of the session, in seconds. Note that the pipeline will not emit any data for this many seconds at the beginning, since it first has to gather calibration data.', 'type': 'float', 'expert': False, 'category': 'Data cleaning', 'verbose_name': 'Calibration duration', 'value': '20.0'}, 
  'keep_channels': 
 {'is_filename': False, 'description': 'The list of channels to keep, e.g., [0,1,2,3,4] or a range such as 0...4. This is important if you have things like flat-line channels, trigger channels, accelerometer channels or etc. in your data as the pipeline may likely not work with these (default: all channels). You can also use : slice syntax as in Python to count from the end.', 'type': 'string', 'expert': False, 'category': 'Core Parameters', 'verbose_name': 'Channels to keep', 'value': ':'}, 
  'clean_cutoff': 
 {'is_filename': False, 'description': 'The threshold for removing artifacts. This is given in standard deviations. 7 is fairly gentle on the data, 5 a good balance, 3.5 is quite aggressive, and 2.5 is extremely aggressive.', 'type': 'float', 'expert': False, 'category': 'Data Cleaning', 'verbose_name': 'Cleaning cutoff', 'value': '7.0'}
}

Here you can see three parameters: calib_duration, keep_channels, and clean_cutoff.

Create a Streaming Instance with Updated Parameters

In order to update parameters of a streaming instance it must first exist. Lets assemble our launch options for the Simple Bandpower pipeline (including the parameter update):

# update parameter value
pipeline_parameters['win_len'].update({'value': '60'})
# assemble the final launch options for the pipeline
new_params = {"pipeline": pipeline_id,
              "metadata": pipeline_metadata,
              "parameters": pipeline_parameters,
              "encoding": 'msgpack'}

And use the requests.post REST command to create a streaming instance and save the instance_id:

# to launch our processing pipeline now we need to create a new "instance"
# of the desired pipeline (a running piece of code)
# IMPORTANT: note that, once the instance has been created, you are
# consuming resources in the cloud until you destroy the instance again
r = requests.post(api_url + '/v1/instances', headers=auth_header, json=params)
instance_id = None
if r.status_code == 201 or r.status_code == 200:  # 201: created / 200: ok
    # we need to extract the instance id from the response so we can shut it
    # down again later (note: it's instructive to print out the JSON response)
    body = r.json()
    instance_id = body['id']
else:
    print("ERROR: Could not post (HTTP %s); check "
          "your API URL and credentials." % r.status_code)

If you have not received a status code error your streaming instance is now launched.

Modify Parameters of an Existing Streaming Instance

Now that a streaming instance of your pipeline exists, you can update the parameters to your liking on the fly and see the results in real-time. This can be done by updating the pipeline_parameters dictionary with the new values, assembling the new pipeline configuration, and using the requests.patch REST command to updated the pipeline with new parameter(s):

# update parameter value
pipeline_parameters['win_len'].update({'value': '100'})
# assemble new pipeline configuration 
new_params = {"pipeline": pipeline_id,
              "metadata": pipeline_metadata,
              "parameters": pipeline_parameters,
              "encoding": 'msgpack'}

if instance_id is not None:
    r = requests.patch(api_url + '/v1/instances/' + instance_id, headers=auth_header, json=new_params)
    if r.status_code == 201 or r.status_code == 200:  # 201: created / 200: ok
        body = r.json()
        pipeline_parameters = body['parameters']
        print("New pipeline_parameters = %s" % pipeline_parameters)
    else:
        print("ERROR: Could not patch (HTTP %s); check "
              "your API URL and credentials." % r.status_code)

If you have not received a status code error the following will print confirming the parameter updates were successful:

New pipeline_parameters = {'win_len': {'value': '100', 'type': 'string', 'verbose_name': None, 'expert': False, 'is_filename': False, 'category': '', 'description': 'Length of Moving Average window in seconds'}}