Java (Android)
This tutorial explains how to interact with the NeuroScale Platform API from Android. This tutorial is meant to be read alongside the source code of the demo app DemoAndroidApp.java (link).
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).
Instantiate Neuroscale Client¶
Our client will allow interaction with the NeuroScale API using HTTP requests (and parse HTTP responses). There are several HTTP libraries that one can use but in this example an asynchronous callback-based HTTP client for Android developed by James Smith is used. This client is built on Apache’s HTTPClient libraries which are also fine to use in this case. The advantages of the library we use is that all requests are made outside your applications’s main UI thread. Further, the callbacks are executed on the same thread where they were created (using Android’s Handler message passing). For more information please see this link. Here we instantiate the NeuroScale client:
```java NeuroScaleClient.java NeuroScaleClient ( String apiUrl, String apiVersion, String accessToken ) { this.apiUrl = apiUrl; this.apiVersion = apiVersion; this.accessToken = accessToken; }
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.
```java DemoAndroidAppSettings.java
public static final String API_URL = "https://api.neuroscale.io";
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.
```java DemoAndroidAppSettings.java
public static final String ACCESS_TOKEN = "## Instantiating a Pipeline
The API is designed according to [REST](../advanced-developer-topics/the-restful-nature-of-the-neuroscale-api.md) 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](http://www.json.org/). All interactions need to be authorized by including an additional HTTP `Authorization` header whose value should be `Bearer your-access-token`:
```java NeuroScaleClient.java
public String getAuthorizationHeader() {
return "Bearer " + this.getAccessToken();
}
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 "Relaxation" pipeline, which returns an index of relaxation from incoming EEG data:
```java DemoAndroidAppSettings.java public static final String PIPELINE = "Relaxation (Generic)";
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:
```java NeuroScaleClient.java
Header[] headers = {
new BasicHeader("Authorization", this.getAuthorizationHeader())
};
AsyncHttpClient client = new AsyncHttpClient();
client.setTimeout(30000);
client.get(ctx, this.getPipelinesUrl(), headers, null, responseHandler);
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:
```java DemoAndroidApp.java try { // iterate through and select id needed // we need to extract all pipeline ids from the // response in order to select the id that has a // matching name pipelines = response.getJSONArray("data"); for (int i = 0; i < pipelines.length(); i++) { JSONObject pipeline = pipelines.getJSONObject(i); Log.d(TAG, pipeline.toString()); if (pipeline.getString("name").equals(DemoAndroidAppSettings.PIPELINE)) { pipelineId = pipeline.getString("id"); Log.i(TAG, "Using Pipeline with id:" + pipelineId); } } // allow user to create instance createInstanceButton.setEnabled(true); } catch (JSONException e) { handleError(e); }
Also, before we can launch this pipeline we need to describe the data that we want to stream up; this [metadata](../advanced-developer-topics/instance-metadata.md) 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 use the following meta data:
```java DemoAndroidAppSettings.java
public static final String METADATA = "" +
"{\n" +
" \"nodes\": {\n" +
" \"out\": [\n" +
" {\n" +
" \"name\": \"default\",\n" +
" \"streams\": [\n" +
" {\n" +
" \"type\": \"EEG\",\n" +
" \"channels\": [\n" +
" {\n" +
" \"label\": \"concindex\"\n" +
" }\n" +
" ],\n" +
" \"name\": \"myeeg\",\n" +
" \"sampling_rate\": 200.0\n" +
" }\n" +
" ]\n" +
" }\n" +
" ],\n" +
" \"in\": [\n" +
" {\n" +
" \"name\": \"default\",\n" +
" \"streams\": [\n" +
" {\n" +
" \"type\": \"EEG\",\n" +
" \"channels\": [\n" +
" {\n" +
" \"label\": \"Ch0\"\n" +
" },\n" +
" {\n" +
" \"label\": \"Ch1\"\n" +
" },\n" +
" {\n" +
" \"label\": \"Ch2\"\n" +
" },\n" +
" {\n" +
" \"label\": \"Ch3\"\n" +
" },\n" +
" {\n" +
" \"label\": \"Ch4\"\n" +
" },\n" +
" {\n" +
" \"label\": \"Ch5\"\n" +
" }\n" +
" ],\n" +
" \"name\": \"myeeg\",\n" +
" \"sampling_rate\": 200.0\n" +
" }\n" +
" ]\n" +
" }\n" +
" ]\n" +
" }\n" +
" }";
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. A pipeline instance can be launched as follows:
```java NeuroScaleClient.java // create headers auth + content-type Header[] headers = { new BasicHeader( "Content-Type", "application/json" ), new BasicHeader( "Authorization", this.getAuthorizationHeader() ) };
// create body
String requestBody = "" +
"{\n" +
" \"pipeline\": \"%s\",\n" +
" \"description\": \"%s\",\n" +
" \"metadata\": %s\n" +
"}";
requestBody = String.format(
requestBody,
pipelineId,
description,
metadata
);
Log.d(TAG, requestBody);
// set up http client
AsyncHttpClient client = new AsyncHttpClient();
client.setTimeout( 30000 );
// send request
try {
client.post( ctx, endpoint, headers, new StringEntity( requestBody ), null, responseHandler );
} catch( UnsupportedEncodingException e ) {
throw new NeuroScaleException( e );
}
```
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:
java DemoAndroidApp.java
if (instance != null) {
String id = instance.getString("id");
String state = instance.getString("state");
....
if (state.equals("running")) {
scheduler.removeCallbacks(poller);
// get broker url from endpoint (hack)
try {
URL url = new URL(writeEndpoint.replace("mqtt", "http"));
String uri = "tcp://" + url.getHost() + ":" + url.getPort();
String writeTopic = url.getPath();
String readTopic = new URL(readEndpoint.replace("mqtt", "http")).getPath() + "/#";
this.mqttClientSession = new MqttClientSession(this, uri, writeTopic, readTopic, this);
} catch (MalformedURLException e) {
handleError(e);
}
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):
```java NeuroScaleClient.java if (instance != null) { String id = instance.getString("id"); String state = instance.getString("state");
String readEndpoint = this.neuroScaleClient.getEndpointURL(
instance, "read"
);
String writeEndpoint = this.neuroScaleClient.getEndpointURL(
instance, "write"
);
java NeuroScaleClient.java
try {
JSONObject endpoints = instance.getJSONObject("endpoints");
JSONArray data = endpoints.getJSONArray("data");
String endpointURL = null;
String readTopicName = null, writeTopicName = null;
for (int i = 0; i < data.length(); i++) {
JSONObject endpoint = data.getJSONObject(i);
if (endpoint.getString("mode").equals(type) ){
endpointURL = endpoint.getString("url");
}
}
return endpointURL;
} catch (Exception e) {
throw new NeuroScaleException(e);
}
```
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:
```java MqttClientSession.java public MqttClientSession(Context ctx, String uri, String writeTopic, String readTopic, IMqttHandler iMqttHandler) { this.state = State.DISCONNECTED; this.ctx = ctx; this.uri = uri; this.readTopic = readTopic; this.writeTopic = writeTopic;
this.iMqttHandler = iMqttHandler;
// Prepare our MQTT connection options.
this.options = new MqttConnectOptions();
this.options.setConnectionTimeout(0);
this.options.setCleanSession(true); // Transient MQTT connection.
// Try to connect.
this.connect();
}
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):java MqttClientSession.java
@Override
public void messageArrived(String topic, MqttMessage message) {
// Called when a message arrives from the server that matches any
// subscription made by the client
// String time = new Timestamp(System.currentTimeMillis()).toString();
// System.out.println("Time:\t" +time +
// " Topic:\t" + topic +
// " Message:\t" + new String(message.getPayload()) +
// " QoS:\t" + message.getQos());
IMqttHandler handler = (IMqttHandler) this.ctx;
handler.onMessageArrived(topic, new String(message.getPayload()));
}
```
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:
```java DemoAndroidApp.java // continuously send data to a pipeline running in NeuroScale public void sendData(View view) {
if (sendingData) {
// Stop sending data
sendingData = false;
dataSenderThread.interrupt();
sendDataButton.setText(R.string.send_data);
} else {
//Continually send data to running instance
dataSenderThread = new Thread(new Runnable() {
@Override
public void run() {
while (true) {
Log.d(TAG, "sending data");
try {
// note that using a shorter sleep interval here will lead to more
// messages per second being transmitted; there is an upper limit
// to how many messages per second the backend will pick up,
// and if you find that the output starts to lag behind by several
// seconds, most likely your transmission interval is too short
Thread.sleep(100); // samplingRate / numChunks
try {
// create message to send up to running
// instance for processing
JSONObject message = createMessage();
// send the message
mqttClientSession.publish(message.toString().getBytes());
} catch (MqttException e) {
Log.e(TAG, e.toString());
}
} catch (InterruptedException e) {
Log.e(TAG, e.toString());
break;
}
if (Thread.currentThread().isInterrupted()) {
break;
}
}
}
}, "dataSenderThread");
sendingData = true;
dataSenderThread.start();
sendDataButton.setText(R.string.stop_data);
}
}
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 6. The generated chunk is a list of samples, each of which is a list of values (10 per channel):java DemoAndroidApp.java
private JSONObject createMessage() {
/**
* Create a dummy message to send to pipeline
* See template in settings file
*/
Integer numSamples = 10;
Integer numChannels = 6;
Double samplingRate = 100.0;
JSONArray timestamps = new JSONArray();
JSONArray streams = new JSONArray();
JSONArray samples = new JSONArray();
JSONObject message = new JSONObject();
JSONObject stream = new JSONObject();
JSONArray sample;
Random r = new Random();
try {
// create timestamps and random samples
for (int i = 0; i < numSamples; i++) {
sample = new JSONArray();
for (int j = 0; j < numChannels; j++) {
sample.put(r.nextDouble());
}
samples.put(sample);
timestamps.put(System.currentTimeMillis() / 1000 + i / samplingRate);
}
stream.put("name", "myeeg");
stream.put("timestamps", timestamps);
stream.put("samples", samples);
streams.put(stream);
message.put("streams", streams);
// Log.d(TAG, message.toString());
} catch (JSONException e) {
Log.e(TAG, e.toString());
}
return message;
}
```
Streaming Data Down¶
Whenever a message is received back from NeuroScale containing real-time results, the MQTT client will cause the following handler function to be called. Here we are just picking up the message payload and passing it on to another function of ours in order to have it parsed and ultimately displayed. The format of the returned message is the same as the format that you're using for sending data up.
@Override
public void messageArrived(String topic, MqttMessage message) {
// Called when a message arrives from the server that matches any
// subscription made by the client
// String time = new Timestamp(System.currentTimeMillis()).toString();
// System.out.println("Time:\t" +time +
// " Topic:\t" + topic +
// " Message:\t" + new String(message.getPayload()) +
// " QoS:\t" + message.getQos());
IMqttHandler handler = (IMqttHandler) this.ctx;
handler.onMessageArrived(topic, new String(message.getPayload()));
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:
```java NeuroScaleClient.java public void deleteInstance( Context ctx, String id, ResponseHandlerInterface responseHandler ) throws NeuroScaleException { /** * Deletes a Instance * https://neuroscale.io/docs/api-reference/delete-an-existing-instance.html */
Header[] headers = {
new BasicHeader( "Authorization", this.getAuthorizationHeader() )
};
AsyncHttpClient client = new AsyncHttpClient();
client.setTimeout( 30000 );
client.delete( ctx, this.getInstanceUrl( id ), headers, responseHandler );
}
```