Upstreaming data from a wearable device to a server: an Android app for a data lab

9 Aug 2016

sisifospage@gmail.com 300 139 Sisifo's Page getting data from a wearable upstream, Android app, data lab, accelerometer data analysis 31 Oct 2016

Introduction & purpose

This post is more related to the data engineering side of things, rather than the analytics, at least first sight. At the end, indeed it is about an Android app! The point is, we've been for quite some time now preparing a data laboratory, and our intention is to buy pretty soon one or two dozens of wristbands, have some people wear them for a few hours a day, upstream all the data to a database in a server, and then analyze the data to find insights.

Everything you would need for building such a lab is pretty much available: there's several providers for the hardware of the wristband, which also give an Android app to communicate with it. And Android itself has open and well documented libraries for nearly everything you need, thus it should be fairly easy to enhance the provider's app to your purpose. However, the sensors in the wristband produce an amount of data that is not negligible, i.e. if you want to process the raw data (commercial products pre-process the sensors' data in the hardware of the wristband, and only accumulated results go to the app in your smartphone). Besides, upstreaming is not really so common in any Android development, at least not in a fast and continuous manner. Thus, finally it was not so easy to prepare the app for the lab, and that's why we thought it could make a hopefully interesting and useful post.

This post aims to explain the details of such an app, with code snippets and links to the full solution, and being agnostic about the hardware.

A cache object and an upstream service

In this post you will find the description of two main objects of the app:

• A cache object that stores the data from the wearable in an array in memory, so that the access is very fast. Periodically, the cache takes the elements from its buffers and stores them in a SQLite in-memory database
• An upstream service, which takes care of the communication with the server, regularly uploading the data from the SQLite database into the server via a REST API

The complete app (see here for the full code), contains several Android Activities which handle the interaction with the user and the connection with the device. The cache is tightly linked to one of the activities, while the communication with the server is implemented as an Android Service. (Note that, in Android development, an Activity runs in the foreground and serves as the user's GUI, while a Service implements something that the user does not see directly, but it is important for the app, see here for more details.)

The cache in detail, including code snippets

You can find the full implementation of the object here. Main features follow as bullet list.

• It is implemented as a singleton, which means there is a unique instance of the object in the app -thus the unique and same instance may be retrieved from either the Activity handling the interaction with the user and the hardware, or from the Upstream Service. However, note that only the Activity will actually create the new instance; the Service will just fetch it. Also note ahta the context of the Android Activity is stored -will be used for establishing the connection with the SQLite database, and also in the Service

    public synchronized static AccDataCacheSingleton getInstance(Context context) {
          if (accDataCacheSingleton == null && context != null) {
              Log.d(TAG, "Creating cache singleton");
              accDataCacheSingleton = new AccDataCacheSingleton();
              accDataCacheSingleton.context = context;
              accDataCacheSingleton.initBuffer();
          }
          return accDataCacheSingleton;
      }
  

• Additionally, there is a call to a method to inizialize the buffers

    private void initBuffer() {
            togglingBuffer = new Buffer[TOGGLING_BUFFER_SIZE];
            bufferPointer = 0;
            unitsPointer = 0;
            bufferBackupPending = new HashMap<>();
            togglingBuffer[accDataCacheSingleton.bufferPointer] = new Buffer();
        }
    

• The cache receives requests from the main activity to add data from the sensor. Note that these data come one by one, and at a relatively high rate. For example, in the case of an accelerometer, it may be between 10 and 50 samples per second, each of them with the 3 position coordinates. The cache stores the data in a buffer and responds fast

    public synchronized void add(FraaStreamDataUnit unit) {
          if (unitsPointer == LENGTH_OF_BUFFER) {
              Log.e(TAG, "Error, index has been skipped!!!");
              toggleBuffer();
          }
          togglingBuffer[bufferPointer].units[unitsPointer++] = unit;
          if (unitsPointer == LENGTH_OF_BUFFER) {
              Log.d(TAG, "Recreating singleton buffer");
              UUID id = toggleBuffer();
              new InsertIntoDatabaseTask().execute(id.toString());
          }
      }
  

• Additionally, when the buffer is full, the cache toggles to a second buffer for adding new data, and flushes the data of the first buffer into the SQLite database. This is the method that toggles the buffers when the active buffer gets full. It uses the database semaphore to guarantee that no other process accesses to bufferBackupPending at the same time (being the async insertions to the database the other process that may access it, see below)

    private static class Buffer {
            public FraaStreamDataUnit[] units;

            public Buffer() {
                this.units = new FraaStreamDataUnit[LENGTH_OF_BUFFER];
            }
        }

        private UUID toggleBuffer() {
            if (bufferBackupPending.get(bufferPointer) != null) {
                Log.e(TAG, "Data would get lost in this case... consider adding a semaphore?");
            }
            UUID id = UUID.randomUUID();
            // use database semaphore...
            getDbHelperWhenAvailable();
            bufferBackupPending.put(bufferPointer++, id);
            release();
            bufferPointer = (bufferPointer == TOGGLING_BUFFER_SIZE) ? 0 : bufferPointer;
            togglingBuffer[bufferPointer] = new Buffer();
            unitsPointer = 0;
            return id;
        }
    

• While the following method, invoked after toggling, inserts the data of the full buffer into SQLite as an asynchronous task (thus the Activity does not need to wait for it to finish):

    private class InsertIntoDatabaseTask extends AsyncTask<String, Integer, Integer> {
            @Override
            protected Integer doInBackground(String... params) {
                UUID id = UUID.fromString(params[0]);
                Integer bufferToBackup = null;
                FraaDbHelper fraaDbHelper = getDbHelperWhenAvailable();
                SQLiteDatabase db = fraaDbHelper.getWritableDatabase();
                for (Integer key : bufferBackupPending.keySet()) {
                    UUID candidate = bufferBackupPending.get(key);
                    if (candidate != null && candidate.equals(id)) {
                        bufferToBackup = key;
                    }
                }
                if (bufferToBackup == null) {
                    Log.e(TAG, "Data got lost... Don't find " + id.toString() + " anymore");
                }

                for (FraaStreamDataUnit unit : togglingBuffer[bufferToBackup].units) {
                    //Log.d(StreamingActivity.TAG, "count:" + count++);
                    ContentValues values = new ContentValues();
                    values.put(AccDataContract.AccDataEntry.COLUMN_NAME_HEADER_ID, getHeaderId());
                    values.put(AccDataContract.AccDataEntry.COLUMN_NAME_INDEX, unit.getIndex());
                    values.put(AccDataContract.AccDataEntry.COLUMN_NAME_X, unit.getX());
                    values.put(AccDataContract.AccDataEntry.COLUMN_NAME_Y, unit.getY());
                    values.put(AccDataContract.AccDataEntry.COLUMN_NAME_Z, unit.getZ());
                    db.insert(AccDataContract.AccDataEntry.TABLE_NAME,
                            null,
                            values);
                }

                //db.endTransaction();
                bufferBackupPending.put(bufferToBackup, null);
                Log.d(TAG, "copy to SQLite ok (" + bufferToBackup + ")");
                db.close();
                release();
                return 1;
            }
        }
    

• Finally, it is important to note that the cache object handles every access to the SQLite database, also the ones that are required by the upstream service. That way, since all accesses are handled in a single object, contention can be avoided (since SQLite implementation in Android may fail in concurrent accesses). The access to the database is protected through a simple semaphore in our current implementation, so that only one process can access it at a time:

    private final Semaphore available = new Semaphore(1);

    private FraaDbHelper getDbHelperWhenAvailable() {
        available.acquireUninterruptibly();
        return new FraaDbHelper(this.context);
    }
  

The upstream service in detail, including code snippets

The upstream service periodically uploads the available data into the server. As stated above, every time it needs to access the SQLite database, it asks it to the cache object (a unique singleton instance). Note that:

• The object extends an Android Service, and only implements a few methods with the actual functionality. Hint: when you create a service, do it with Android Studio, so that the tools take care of the modifications to the manifest and gradle scripts
• The requests are done using volley, a very handy Android library for REST accesses; however, since most of the common accesses are downstream (not upstream), the library is not as straightforward to use as in other cases. Still, it makes it pretty easy
• Since the cache is a singleton, the communication between the service and the cache is direct. The upstream service will get the instance of the cache that the Activity has created -anyhow only the Activity, not the service, will initialize the cache. (Recall that the Activity handles the interaction with the user an the hardware. Usually in Android, when communicating an Activity with a Service, a Service Intent object is used, but in this case it is not needed)

You can find the full implementation of the object here. Main features follow as bullet list.

• The following method sends a bunch of data upstream to the server. It uses Google's gson for serialization. Most of the logic is implemented as a volley custom request (since it is upstream and with particular data type), following this guidelines and implemented in a generic object called GsonRequest. Once this object is implemented, each of the requests just needs to describe what to do when the request is either successful or fails

    private void sendToServer(FraaStreamData data) {
        RequestQueue queue = Volley.newRequestQueue(this);
        String url = server_url + "data";
        GsonRequest postRequest = new GsonRequest<FraaStreamData, UUID>(Request.Method.POST, url, data, null,
                new Response.Listener<UUID>() {
                    @Override
                    public void onResponse(UUID response) {
                        Log.i(StreamingActivity.TAG, "Response id to be removed: " + response + "(" + pendingRequests.get(response).getHeaderId() + ")");
                        // update new serverHeaderId
                        AccDataCacheSingleton obj = AccDataCacheSingleton.getInstance();
                        obj.removeFromDatabase(pendingRequests.get(response));
                        Log.i(StreamingActivity.TAG, "Response id finished removing: " + response + "(" + pendingRequests.get(response).getHeaderId() + ")");
                    }
                }, new Response.ErrorListener() {
            @Override
            public void onErrorResponse(VolleyError error) {
                Log.i(StreamingActivity.TAG, "sendToServer didn't work!");
                Log.i(StreamingActivity.TAG, error.toString());
            }
        }, UUID.class);
        // avoid sending the data twice (big message over a slow network)
        postRequest.setRetryPolicy(new DefaultRetryPolicy(0,
                DefaultRetryPolicy.DEFAULT_MAX_RETRIES,
                DefaultRetryPolicy.DEFAULT_BACKOFF_MULT));
        // Add the request to the RequestQueue.
        queue.add(postRequest);
    }
  

• Finally for the upstream service, find below the task that regularly checks if there is new data in the cache, and if there is, it adds a new item to the list of pending requests. (Note that there is another scheduled task that actually sends the data upstream to the server)

    public void checkCacheDataStream() {
          final Runnable checkData = new Runnable() {
              public void run() {
                  AccDataCacheSingleton obj = AccDataCacheSingleton.getInstance();
                  Collection<FraaStreamData> list = convertWithMaxDataSize(obj.selectRowsHeaderEqualTo(getHeaderId()));
                  boolean dataPending = false;
                  for (FraaStreamData data : list) {
                      if (data.getDataUnits().length == MAX_NUMBER_DATA_UNITS_UPSTREAM) {
                          addDataToPendingRequests(data);
                          dataPending = true;
                      }
                  }
                  if (dataPending && dataProcessHandle.isCancelled()) {
                      Log.d(StreamingActivity.TAG, "restarting data process");
                      checkDataEveryInterval();
                  }
              }
          };
          // runs every 2 minutes, never stops
          // TODO adjust time interval depending on size of buffers & sampling rate
          cacheScheduler.scheduleAtFixedRate(checkData, 0, 2, TimeUnit.MINUTES);
      }
  

Note on the server

You can find the full code here. The server just does enough to handle the REST requests and flush the data into a database. It is implemented in Java, using Hibernate for handling the database. It uses maven to produce a war file, which may be deployed e.g. using tomcat.