Stream Results from NeuroScale (API)

This tutorial will cover how to stream results from an Instance of a streaming Pipeline with the NeuroScale API using Python and assumes:

Your Python environment is installed and setup.
You have completed the Streaming Data to NeuroScale (API) tutorial and data is being streamed up to the Instance.
You are reading alongside the source code of the Python example client_basic.py (link),

Please look over the Python documentation, go through the Streaming Data to NeuroScale (API) tutorial and client_basic.py, if this is not the case.

Outline¶

Review: Obtaining the MQTT Input/Output Endpoints
Review: Setting up Communication With the Instance
Handling Return Messages From Instance
Managing Returned Results

Review: Obtaining the MQTT Input/Output Endpoints¶

At this point, you should 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')

Review: 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:

reader = mqtt.Client(userdata='reader')
writer = mqtt.Client(userdata='writer')

Handling Return Messages From Instance¶

In order to use the streaming results of the pipeline we need a handler that interprets the messages that we receive from the pipeline instance. This is conducted in the on_message function in the python clients:

def on_message(client, userdata, msg):
  # ...

We will use the following code excerpt to illustrate how to handle the results:

def on_message(client, userdata, msg):
            # this callback will be invoked whenever we get a message back;
            # some pipelines have multiple output nodes; these messages from
            # these nodes generally come in on sub-topics with the same name as
            # the respective node; to distinguish messages from non-default
            # nodes, we may compare the topic to the one that we expect, and
            # discard irrelevant messages
            topic = msg.topic.split('/')[-1]
            if topic != 'default':
                print("received message on non-default topic (%s); ignoring."
                      % topic)
            # the payload must first be decoded using either JSON or msgpack
            # (depending on pipeline settings)
            payload = msg.payload
            data = json.loads(payload) if json_fallback else msgpack.loads(payload)
            # the message has a 'streams' field that is a list of one chunk per
            # stream for which data was received (note that a stream that
            # updates less frequently than others may not be included in every
            # message)
            streams = data[b'streams']
            if streams:
                print('received message on topic %s: ' % msg.topic)
            else:
                print('received empty message on topic %s: ' % msg.topic)
            for stream in streams:
                # the name of the stream: many pipelines will pass the input
                # names through unchanged
                name = stream[b'echo']
                # this is a list of samples, each of which is a list of channels
                samples = stream[b'samples']
                # a list of time stamps, one per sample (same length as samples)
                stamps = stream[b'timestamps']
                print('  stream %s: %i samples' % (name, len(stamps)))

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. 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):

topic = msg.topic.split('/')[-1]
            if topic != 'default':
                # ...

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:

payload = msg.payload
            data = json.loads(payload) if json_fallback else msgpack.lo

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:

streams = data[b'streams']

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'echo']
                samples = stream[b'samples']
                stamps = stream[b'timestamps']
                print('  stream %s: %i samples' % (name, len(stamps)))

By default a pipeline would not change the name of the stream as it passes through the pipeline. This means that the stream name used when configuring the packet being streamed to the Neuroscale pipeline would be the same name to use here in order to parse the return data. Since echo was used in the previous example, it is also used here, though you can use any name desired (so long as they match in the in and out sections of the metadata used to launch the pipeline). If a Neuroscale pipeline renames the incoming data stream (requiring a name change here), or creates additional streams, that should be defined in the pipeline documentation. (If in doubt, you can print the incoming stream fields (i.e., print(stream.keys()) in the above for loop) which would then tell you the contents of each stream.

Managing Returned Results¶

Now that our handler is interpreting the messages received from the pipeline instance, we can manage what happens to the results in on_message. A powerful option is to stream the processed data out in real-time to your application(s) -- we recommend doing this over the Lab Streaming Layer (LSL) network, shown in client_lsl.py 's code snippet below.

for stream in streams:
                # the name of the stream: many pipelines will pass the input
                # names through unchanged
                name = stream[b'name']
                # this is a list of samples, each of which is a list of channels
                samples = stream[b'samples']
                # a list of time stamps, one per sample (same length as samples)
                stamps = stream[b'timestamps']
                # here we send our received data back to LSL
                if name == b'myeeg':
                    outlet.push_chunk(samples)
                print('  stream %s: %i samples' %
                      (name, len(stamps)))

Please checkout the Stream Results from NeuroScale (API) with LSL tutorial, LSL examples on git, and client_lsl.py for more details.