{"size": 3685, "ext": "js", "max_stars_repo_path": "samples/IoTBench/ETL/CsvToSenML.js", "max_stars_repo_name": "DependableSystemsLab/ThingsJS", "max_stars_repo_stars_event_min_datetime": "2018-10-13T02:54:22.000Z", "max_stars_repo_stars_event_max_datetime": "2022-02-25T09:27:52.000Z", "max_issues_repo_path": "samples/IoTBench/ETL/CsvToSenML.js", "max_issues_repo_name": "DependableSystemsLab/ThingsJS", "max_issues_repo_issues_event_min_datetime": "2018-09-17T17:19:48.000Z", "max_issues_repo_issues_event_max_datetime": "2021-05-28T23:32:29.000Z", "max_forks_repo_path": "samples/IoTBench/ETL/CsvToSenML.js", "max_forks_repo_name": "karthikp-ubc/ThingsJS", "max_forks_repo_forks_event_min_datetime": "2018-01-06T04:48:23.000Z", "max_forks_repo_forks_event_max_datetime": "2018-08-06T21:41:56.000Z"}
{"max_stars_count": 13, "max_issues_count": 24, "max_forks_count": 5, "avg_line_length": 23.1761006289, "max_line_length": 85, "alphanum_fraction": 0.620624152}
/* CsvToSenML adapted from RIotBench
 * modified for measuring processing rate
 */
var things = require('things-js');
var fs = require('fs');
var mongoUrl = 'mongodb://localhost:27017/things-js-fs';
var GFS = require('things-js').GFS(mongoUrl);
var readline = require('readline');

/* configurable variables */
var gfsFlag = true;
var pubsubUrl = 'mqtt://test.mosquitto.org';
var processingTopic = 'iotbench/processing';
var subscribeTopic = processingTopic + '/annotate';
var publishTopic = processingTopic + '/csvtosenml';
var propertiesPath = './TAXI_properties.json';

var pubsub = new things.Pubsub(pubsubUrl);

/* csv to senml properties */
var schemaFields = {};
var schemaTypes = [];
var schemaValues = [];

function beginComponent(properties) {
	var schemaFile = properties['PARSE.CSV_SCHEMA_WITH_ANNOTATEDFIELDS_FILEPATH'];
	if (!schemaFile) {
		console.log('Schema file does not exist');
		process.exit();
	}
	parseEntireSchema(schemaFile).then(function() {
		console.log('Beginning conversion to SenML format');
		pubsub.subscribe(subscribeTopic, convertToSenML);	
	});
}

function setup() {
	var properties;
	if (gfsFlag) {
		GFS.readFile(propertiesPath, function(err, data) {
			if (err) {
				console.log('Could not fetch properties: ' + err);
				process.exit();
			}
			properties = JSON.parse(data);
			beginComponent(properties);
		});
	} else {
		try {
			properties = JSON.parse(fs.readFileSync(propertiesPath, 'utf-8'));
		} catch(e) {
			console.log('Problem with reading properties file: ' + e);
			process.exit();
		}
		beginComponent(properties);
	}
}

function parseEntireSchema(file) {
	var lineNumber = 0;
	var lineReader;
	return new Promise(function(resolve, reject) {
		if (gfsFlag) {
			GFS.readFile(file, function(err, data) {
				if (err) {
					console.log('Problem reading schema: ' + err);
					process.exit();
				}
				var lines = data.split('\n');
				for(var i = 0; i < lines.length; i++) {
					switch (i) {
						case 0:
							var fields = line.split(',');
							var i = 0;
							fields.forEach(function(field) {
								schemaFields[field] = i; 
								i++;
							});
							break;
						case 1:
							schemaTypes = line.split(',');
							break;
						case 2:
							schemaValues = line.split(',');
							break;
					}
				}
				resolve();
			});
		} else {
			try {
				lineReader = readline.createInterface({
					input: fs.createReadStream(file)
				});
			} catch(e) {
				console.log('Problem reading schema: ' + e);
				process.exit();
			}
			lineReader.on('line', function(line) {
				switch (lineNumber) {
					case 0: 
						var fields = line.split(',');
						var i = 0;
						fields.forEach(function(field) {
							schemaFields[field] = i;
							i++;
						});
						break;
					case 1:
						schemaTypes = line.split(',');
						break;
					case 2:
						schemaValues = line.split(',');
						break;
				}
				lineNumber++;
			});

			lineReader.on('close', resolve);	
			}
	});
}

function convertToSenML(msg) {
	var start = Date.now();

	var data = JSON.parse(msg);
	content = data.content;
	id = data.id; 

	var arr = [];
	var keys = Object.keys(content);

	keys.forEach(function(field) {
		var senml = {};
		senml['n'] = field;
		var index = schemaFields[field];
		var type = schemaTypes[index];
		var val = schemaValues[index];

		senml['u'] = type;
		senml[val] = content[field];

		arr.push(senml);
	});

	var end = Date.now();
	var elapsed = end - start;
	pubsub.publish(processingTopic, { id: id, component: 'csvtosenml', time: elapsed });
	pubsub.publish(publishTopic, JSON.stringify({ id: id, content: { 'e': arr } }));

	console.log(JSON.stringify(arr));
}

pubsub.on('ready', function() {
	setup();
});

