Merge pull request #1460 from mddub/wip/mmconnect-use-devicestatus

Make MM Connect plugin use devicestatus collection

lib/bootevent.js

@@ -91,7 +91,7 @@ function boot (env) {
   }
 
   function setupMMConnect (ctx, next) {
-    ctx.mmconnect = require('./plugins/mmconnect').init(env, ctx.entries);
+    ctx.mmconnect = require('./plugins/mmconnect').init(env, ctx.entries, ctx.devicestatus);
     if (ctx.mmconnect) {
       ctx.mmconnect.run();
     }

lib/plugins/mmconnect.js

@@ -4,22 +4,22 @@
 var _ = require('lodash'),
   connect = require('minimed-connect-to-nightscout');
 
-function init (env, entries) {
+function init (env, entries, devicestatus) {
   if (env.extendedSettings.mmconnect && env.extendedSettings.mmconnect.userName && env.extendedSettings.mmconnect.password) {
-    return {run: makeRunner(env, entries)};
+    return {run: makeRunner(env, entries, devicestatus)};
   } else {
     console.info('MiniMed Connect not enabled');
     return null;
   }
 }
 
-function makeRunner (env, entries) {
+function makeRunner (env, entries, devicestatus) {
   var options = getOptions(env);
 
   var client = connect.carelink.Client(options);
   connect.logger.setVerbose(options.verbose);
 
-  var handleData = makeHandler_(entries, options.sgvLimit, makeRecentSgvFilter(), options.storeRawData);
+  var handleData = makeHandler_(entries, devicestatus, options.sgvLimit, options.storeRawData);
 
   return function run () {
     setInterval(function() {
@@ -40,50 +40,52 @@ function getOptions (env) {
   };
 }
 
-function makeHandler_ (entries, sgvLimit, filter, storeRawData) {
+function makeHandler_ (entries, devicestatus, sgvLimit, storeRawData) {
+  var filterSgvs = connect.filter.makeRecencyFilter(function(item) {
+    return item['date'];
+  });
+  var filterDevicestatus = connect.filter.makeRecencyFilter(function(item) {
+    return new Date(item['created_at']).getTime();
+  });
+
   return function handleCarelinkData (err, data) {
     if (err) {
       console.error('MiniMed Connect error: ' + err);
     } else {
       var transformed = connect.transform(data, sgvLimit);
 
-      if (storeRawData && transformed.length) {
-        transformed.push(rawDataEntry(data));
+      if (storeRawData && (transformed.entries.length || transformed.devicestatus.length)) {
+        transformed.entries.push(rawDataEntry(data));
       }
 
       // If we blindly upsert the SGV entries, we will lose trend data for
       // entries we've already stored, since all SGVs from CareLink except
       // the most recent are missing trend data.
-      var filtered = filter(transformed);
+      var filteredSgvs = filterSgvs(transformed.entries);
+
+      // The devicestatus collection doesn't upsert, so we need to avoid
+      // duplicates here
+      var filteredStatus = filterDevicestatus(transformed.devicestatus);
 
-      entries.create(filtered, function afterCreate (err) {
-        if (err) {
-          console.error('MiniMed Connect storage error: ' + err);
-        }
+      // Can't do "bulk" insert, must be done serially
+      createMaybe_(entries, filteredSgvs, function() {
+        createMaybe_(devicestatus, filteredStatus, _.noop);
       });
     }
   };
 }
 
-function makeRecentSgvFilter () {
-  var lastSgvDate = 0;
-
-  return function filter (entries) {
-    var out = [];
-
-    entries.forEach(function(entry) {
-      if (entry['type'] !== 'sgv' || entry['date'] > lastSgvDate) {
-        out.push(entry);
+function createMaybe_ (collection, items, callback) {
+  if (items.length === 0) {
+    callback();
+  } else {
+    collection.create(items, function afterCreate (err) {
+      if (err) {
+        console.error('MiniMed Connect storage error: ' + err);
       }
+      callback();
     });
-
-    out.filter(function(e) { return e['type'] === 'sgv'; })
-      .forEach(function(e) {
-        lastSgvDate = Math.max(lastSgvDate, e['date']);
-      });
-
-    return out;
-  };
+  }
 }
 
 function rawDataEntry (data) {
@@ -110,6 +112,5 @@ module.exports = {
   init: init
   // exposed for testing
   , getOptions: getOptions
-  , makeRecentSgvFilter: makeRecentSgvFilter
   , rawDataEntry: rawDataEntry
 };

package.json

@@ -64,7 +64,7 @@
     "jquery": "^2.1.4",
     "lodash": "^3.9.1",
     "long": "~2.2.3",
-    "minimed-connect-to-nightscout": "git://github.com/mddub/minimed-connect-to-nightscout#v0.2.4",
+    "minimed-connect-to-nightscout": "git://github.com/mddub/minimed-connect-to-nightscout#v1.0.0",
     "moment": "2.10.6",
     "moment-timezone": "^0.4.0",
     "parse-duration": "^0.1.1",

tests/mmconnect.test.js

@@ -57,44 +57,6 @@ describe('mmconnect', function () {
 
   });
 
-  describe('makeRecentSgvFilter()', function () {
-    function sgv(date) {
-      return {type: 'sgv', date: date};
-    }
-    function pumpStatus(date) {
-      return {type: 'pump_status', date: date};
-    }
-
-    it('should return a stateful filter which discards sgvs older than the most recent one seen', function() {
-      var filter = mmconnect.makeRecentSgvFilter();
-
-      filter([2, 3, 4].map(sgv)).length.should.equal(3);
-
-      filter([2, 3, 4].map(sgv)).length.should.equal(0);
-
-      var filtered = filter([1, 2, 3, 4, 5, 6].map(sgv));
-      filtered.length.should.equal(2);
-      _.pluck(filtered, 'date').should.containEql(5);
-      _.pluck(filtered, 'date').should.containEql(6);
-    });
-
-    it('should return a stateful filter which allows non-sgv entries to be old', function() {
-      var filter = mmconnect.makeRecentSgvFilter();
-
-      filter([2, 3, 4].map(sgv)).length.should.equal(3);
-
-      filter([1, 2, 3, 4, 5].map(pumpStatus)).length.should.equal(5);
-
-      var filtered = filter(
-        [1, 2, 3, 4, 5].map(pumpStatus).concat(
-          [3, 4, 5, 6, 7].map(sgv)
-        )
-      );
-      _.filter(filtered, {type: 'pump_status'}).length.should.equal(5);
-      _.filter(filtered, {type: 'sgv'}).length.should.equal(3);
-    });
-  });
-
   describe('rawDataEntry()', function () {
     it('should generate a "carelink_raw" entry with sgs truncated and PII redacted', function () {
       var data = {