factored data loading and AR2 forecasts out of websockets

lib/bootevent.js

@@ -31,11 +31,8 @@ function boot (env) {
       store.ensureIndexes(ctx.profile( ), ctx.profile.indexedFields);
 
       ctx.heartbeat = require('./ticker')(env, ctx);
-      ctx.heartbeat.uptime( );
 
-      ctx.heartbeat.on('tick', function(tick) {
-        console.info('tick', tick)
-      });
+      ctx.data = require('./data')(env, ctx);
 
       next( );
     })

lib/data.js

@@ -0,0 +1,141 @@
+'use strict';
+
+var async = require('async');
+
+function init (env, ctx) {
+
+  ctx.data = {
+    sgvs: []
+    , treatments: []
+    , mbgs: []
+    , cals: []
+    , profile: []
+    , devicestatus: {}
+    , lastUpdated: 0
+  };
+
+  var ONE_DAY = 86400000
+    , TWO_DAYS = 172800000
+    ;
+
+  var dir2Char = {
+    'NONE': '⇼',
+    'DoubleUp': '⇈',
+    'SingleUp': '↑',
+    'FortyFiveUp': '↗',
+    'Flat': '→',
+    'FortyFiveDown': '↘',
+    'SingleDown': '↓',
+    'DoubleDown': '⇊',
+    'NOT COMPUTABLE': '-',
+    'RATE OUT OF RANGE': '↮'
+  };
+
+  function directionToChar(direction) {
+    return dir2Char[direction] || '-';
+  }
+
+  ctx.data.update = function update (done) {
+
+    //save some chars
+    var d = ctx.data;
+
+    console.log('running data.update');
+    var now = d.lastUpdated = Date.now();
+
+    var earliest_data = now - TWO_DAYS;
+    var treatment_earliest_data = now - (ONE_DAY * 8);
+
+    function sort (values) {
+      values.sort(function sorter (a, b) {
+        return a.x - b.x;
+      });
+    }
+
+    async.parallel({
+      entries: function (callback) {
+        var q = { find: {"date": {"$gte": earliest_data}} };
+        ctx.entries.list(q, function (err, results) {
+          if (!err && results) {
+            results.forEach(function (element) {
+              if (element) {
+                if (element.mbg) {
+                  d.mbgs.push({
+                    y: element.mbg, x: element.date, d: element.dateString, device: element.device
+                  });
+                } else if (element.sgv) {
+                  d.sgvs.push({
+                    y: element.sgv, x: element.date, d: element.dateString, device: element.device, direction: directionToChar(element.direction), filtered: element.filtered, unfiltered: element.unfiltered, noise: element.noise, rssi: element.rssi
+                  });
+                }
+              }
+            });
+          }
+
+          //FIXME: sort in mongo
+          sort(d.mbgs);
+          sort(d.sgvs);
+          callback();
+        })
+      }, cal: function (callback) {
+        //FIXME: date $gte?????
+        var cq = { count: 1, find: {"type": "cal"} };
+        ctx.entries.list(cq, function (err, results) {
+          if (!err && results) {
+            results.forEach(function (element) {
+              if (element) {
+                d.cals.push({
+                  x: element.date, d: element.dateString, scale: element.scale, intercept: element.intercept, slope: element.slope
+                });
+              }
+            });
+          }
+          callback();
+        });
+      }, treatments: function (callback) {
+        var tq = { find: {"created_at": {"$gte": new Date(treatment_earliest_data).toISOString()}} };
+        ctx.treatments.list(tq, function (err, results) {
+          d.treatments = results.map(function (treatment) {
+            var timestamp = new Date(treatment.timestamp || treatment.created_at);
+            treatment.x = timestamp.getTime();
+            return treatment;
+          });
+
+          //FIXME: sort in mongo
+          d.treatments.sort(function(a, b) {
+            return a.x - b.x;
+          });
+
+          callback();
+        });
+      }, profile: function (callback) {
+        ctx.profile.list(function (err, results) {
+          if (!err && results) {
+            // There should be only one document in the profile collection with a DIA.  If there are multiple, use the last one.
+            results.forEach(function (element, index, array) {
+              if (element) {
+                if (element.dia) {
+                  d.profile[0] = element;
+                }
+              }
+            });
+          }
+          callback();
+        });
+      }, devicestatus: function (callback) {
+        ctx.devicestatus.last(function (err, result) {
+          if (!err && result) {
+            d.devicestatus.uploaderBattery = result.uploaderBattery;
+          }
+          callback();
+        })
+      }
+    }, done);
+
+  };
+
+  return ctx.data;
+
+}
+
+module.exports = init;

lib/entries.js

@@ -121,7 +121,7 @@ function storage(env, ctx) {
         });
 
         ctx.plugins.eachEnabledPlugin(function eachEnabled(plugin) {
-          if (plugin.processEntry) plugin.processEntry(doc, ctx, env)
+          if (plugin.processEntry) plugin.processEntry(doc, ctx, env);
         });
       });
     });

lib/plugins/ar2.js

@@ -0,0 +1,67 @@
+'use strict';
+
+function init() {
+
+  function ar2() {
+    return ar2;
+  }
+
+  ar2.label = 'AR2';
+  ar2.pluginType = 'forecast';
+
+  var ONE_HOUR = 3600000;
+  var ONE_MINUTE = 60000;
+  var FIVE_MINUTES = 300000;
+
+  ar2.forecast = function forecast(env, ctx) {
+
+    var actual = ctx.data.sgvs;
+    var actualLength = actual.length - 1;
+    var lastUpdated = ctx.data.lastUpdated;
+
+    var result = {
+      predicted: []
+      , avgLoss: 0
+    };
+
+    if (actualLength > 1) {
+      // predict using AR model
+      var lastValidReadingTime = actual[actualLength].x;
+      var elapsedMins = (actual[actualLength].x - actual[actualLength - 1].x) / ONE_MINUTE;
+      var BG_REF = 140;
+      var BG_MIN = 36;
+      var BG_MAX = 400;
+      var y = Math.log(actual[actualLength].y / BG_REF);
+      if (elapsedMins < 5.1) {
+        y = [Math.log(actual[actualLength - 1].y / BG_REF), y];
+      } else {
+        y = [y, y];
+      }
+      var n = Math.ceil(12 * (1 / 2 + (lastUpdated - lastValidReadingTime) / ONE_HOUR));
+      var AR = [-0.723, 1.716];
+      var dt = actual[actualLength].x;
+      for (var i = 0; i <= n; i++) {
+        y = [y[1], AR[0] * y[0] + AR[1] * y[1]];
+        dt = dt + FIVE_MINUTES;
+        result.predicted[i] = {
+          x: dt,
+          y: Math.max(BG_MIN, Math.min(BG_MAX, Math.round(BG_REF * Math.exp(y[1]))))
+        };
+      }
+
+      // compute current loss
+      var size = Math.min(result.predicted.length - 1, 6);
+      for (var j = 0; j <= size; j++) {
+        result.avgLoss += 1 / size * Math.pow(log10(result.predicted[j].y / 120), 2);
+      }
+    }
+
+    return result;
+  };
+
+  return ar2();
+}
+
+function log10(val) { return Math.log(val) / Math.LN10; }
+
+module.exports = init;