var Bacon = require('baconjs');
var Qty = require('js-quantities');
var signalkSchema = require('signalk-schema');

var conversions = {
  "rad": Qty.swiftConverter('rad', 'deg'),
  "K": Qty.swiftConverter("tempK", "tempC")
}

function StreamBundle() {
  this.buses = {};
  this.streams = {};
  this.pathValues = new Bacon.Bus();
}

StreamBundle.prototype.push = function(pathValue) {
  this.pathValues.push(pathValue);
  this.getBus(pathValue.path).push(pathValue.value);
}

StreamBundle.prototype.getBus = function(path) {
  var result = this.buses[path];
  if (!result) {
    result = this.buses[path] = new Bacon.Bus();
    this.streams[path] = result.debounceImmediate(200);
  }
  return result;
}

StreamBundle.prototype.getStream = function(path) {
  var result = this.streams[path];
  if (!result) {
    this.buses[path] = new Bacon.Bus();
    result = this.streams[path] = this.buses[path].debounceImmediate(200);
    if (signalkSchema.metadata[path] && conversions[signalkSchema.metadata[path].units]) {
      result = this.streams[path] = this.streams[path].map(conversions[signalkSchema.metadata[path].units]);
    }

  }
  return result;
}

module.exports = StreamBundle;

@@ -1,12 +1,787 @@
+/*%0A%0AStreamBundle is a %22bundle of streams%22: Bacon.js streams, two%0Afor each Signal K (source, path) combination. You can get a stream%0Afor the raw data with getBusForSourcePath and a stream where data%0Ais debounced (200 ms) and some units converted with%0AgetStreamForSourcePath.%0A%0AAll incoming Signal K deltas should be passed in via handleDelta.%0A%0AIn addition to the individual streams StreamBundle provides a%0Astream that contains notifications of new (source, path)%0Acombinations appearing in the incoming data. The data in%0Athat stream contains %7BsourceId, path, key, stream%7D objects.%0AInstrumentPanel discovers new data items by listening to this%0Astream.%0A%0AKey is a 'standard' single string representation  of source id%0Aand path produced with signalkSchema.keyForSourceIdPath.%0A%0A*/%0A%0A%0A
 var Bacon = 

Add comment