Writing sensor values to an InfluxDB database with macchina.io EDGE

macchina-edge-demo.js

var net = require('net');

//
// An object for writing measurements to InfluxDB
//
const InfluxDB = {
	formatValue: function(value) {
		if (typeof value === 'number')
		{
			return value.toString();
		}
		else if (typeof value === 'boolean')
		{
			return value.toString();
		}
		else if (typeof value === 'string')
		{
			return JSON.stringify(value);
		}
	},

	formatMeasurement: function(measurement) {
		var result = measurement.name;
		for (var tag in measurement.tags)
		{
			result += ',';
			result += tag;
			result += '=';
			result += InfluxDB.formatValue(measurement.tags[tag]);
		}
		result = result.concat(' ');
		var firstField = true;
		for (var field in measurement.fields)
		{
			if (firstField)
			{
				firstField = false;
			}
			else
			{
				result += ',';
			}
			result += field;
			result += '=';
			result += InfluxDB.formatValue(measurement.fields[field]);
		}
		var ts;
		if (measurement.hasOwnProperty('ts'))
		{
			ts = measurement.ts;
		}
		else
		{
			ts = DateTime().timestamp;
		}
		result += ' ';
		result += ts.valueOf();

		return result;
	},

	postMeasurements: function(uri, org, bucket, token, measurements) {
		var httpRequest = new net.HTTPRequest('POST', uri + '?org=' + encodeURIComponent(org) + '&bucket=' + encodeURIComponent(bucket) + '&precision=us');
		httpRequest.timeout = 5.0;
		httpRequest.set('Authorization', 'Token ' + token);
		httpRequest.contentType = 'text/plain; charset=utf-8';
		var content = '';
		for (var m of measurements)
		{
			content += InfluxDB.formatMeasurement(m) + '\n';
		}
		httpRequest.content = content;
		console.log(content);
		httpRequest.send(result => {
			if (result.error)
			{
				console.error('Error writing data to InfluxDB: %s', result.error);
			}
			else if (result.response.status != 204)
			{
				console.error('Error writing data to InfluxDB: %d %s', result.response.status, result.response.reason);
			}
		});
	}
};


//
// Find available sensors.
//
// We look for temperature, humidity, illuminance (light) and acceleration sensors.
// The way we lookup sensors, the code should be independent of the actual sensor
// connected to the device running macchina.io EDGE.
//
var sensors = {};

var temperatureRefs = serviceRegistry.find('io.macchina.deviceType == "io.macchina.sensor" && io.macchina.physicalQuantity == "temperature"');
if (temperatureRefs.length > 0)
{
    console.log('Temperature sensor found.');
	sensors.temperature = temperatureRefs[0].instance();
}

var humidityRefs = serviceRegistry.find('io.macchina.deviceType == "io.macchina.sensor" && io.macchina.physicalQuantity == "humidity"');
if (humidityRefs.length > 0)
{
    console.log('Humidity sensor found.');
	sensors.humidity = humidityRefs[0].instance();
}

var lightRefs = serviceRegistry.find('io.macchina.deviceType == "io.macchina.sensor" && io.macchina.physicalQuantity == "illuminance"');
if (lightRefs.length > 0)
{
    console.log('Light sensor found.');
	sensors.light = lightRefs[0].instance();
}

var accelerometerRefs = serviceRegistry.find('io.macchina.deviceType == "io.macchina.accelerometer"');
if (accelerometerRefs.length > 0)
{
    console.log('Accelerometer found.');
    sensors.accelerometer = accelerometerRefs[0].instance();
}


//
// Get InfluxDB credentials from configuration settings.
//
const influxURL    = application.config.getString('influxdb.url', 'http://influxdb:8086/api/v2/write');
const influxOrg    = application.config.getString('influxdb.org');
const influxBucket = application.config.getString('influxdb.bucket');
const influxToken  = application.config.getString('influxdb.token');


//
// Additional configuration values, with sensible defaults.
//
const environmentalInterval = application.config.getInt('datalog.environmental.interval', 10000);
const accelerationInterval  = application.config.getInt('datalog.acceleration.interval', 50);
const accelerationChunkSize = application.config.getInt('datalog.acceleration.chunkSize', 100);


//
// Write temperature, humidity and illuminance every 10 seconds.
//
setInterval(
    () => {
        var measurement = {
            name: 'environmental',
            tags: {
                sensor: 'ciss'
            },
            fields: {
            }
        };
        if (sensors.temperature)
        {
            measurement.fields.temperature = sensors.temperature.value();
        }
        if (sensors.humidity)
        {
            measurement.fields.humidity = sensors.humidity.value();
        }
        if (sensors.light)
        {
            measurement.fields.illuminance = sensors.light.value();
        }
        InfluxDB.postMeasurements(influxURL, influxOrg, influxBucket, influxToken, [measurement]);
    },
    environmentalInterval
);


//
// Acquire acceleration samples every accelerationInterval (50) milliseconds.
// Collect accelerationChunkSize (100) measurements and write them at once to
// avoid the overhead of too many HTTP requests.
//
if (sensors.accelerometer)
{
    var measurements = [];
    setInterval(
        () => {
            const accel = sensors.accelerometer.acceleration();
            const measurement = {
                name: 'acceleration',
                tags: {
                    sensor: 'ciss'
                },
                fields: {
                    x: accel.x,
                    y: accel.y,
                    z: accel.z
                },
                ts: DateTime().timestamp
            };
            measurements.push(measurement);
            if (measurements.length == accelerationChunkSize)
            {
                 InfluxDB.postMeasurements(influxURL, influxOrg, influxBucket, influxToken, measurements);
                 measurements = [];
            }
        },
        accelerationInterval
    );
}