<commit_before>if(!process.env.MQTT_CLIENT_ID) {
	console.log('Please define your MQTT_CLIENT_ID.');
	return;
}

console.log('Starting up');

var async = require('async');
var pollingInterval = 1000 * (process.env.POLLING_INTERVAL || 5); 

function initializeSensor(cb) {
	var sensorLib = null;
	try {
		sensorLib = require('node-dht-sensor');
		sensorLib.initialize(11, process.env.SENSOR_GPIO || 4);
	} 
	catch(err) {
		console.log('You are not running on a BCM2835 hardware.');
		cb(err);
	}
	cb(null, sensorLib);
}

function loadProtoBuf(sensorLib, cb) {	
	var request = require('request');
	request.get('https://raw.githubusercontent.com/tabman83/phm-messages/master/sensor-message.proto', function (err, response, body) {
		if(err) {
			cb(err);
			return;
		}
		if(response.statusCode !== 200) {
			cb(new Error('Error while loading protobuf definition file (' + response.statusCode + ')'));
			return;
		}
		var ProtoBuf = require('protobufjs');
		var builder = ProtoBuf.loadProto(body);
		if(!builder) {
			cb(new Error('Error while decoding protobuf definition file'));
			return;
		}
		var SensorMessage = builder.build('SensorMessage');
		cb(null, sensorLib, SensorMessage);
	});	
}

function initializeMqtt(sensorLib, SensorMessage, cb) {
	
	var terminate = function() {
		if(handle) {
			clearTimeout(handle);
		}
		if(client) {
			client.end(true);
		}
	}	
	
	var publishSomething = function() {
		//var mean = 25;
		//var offset = Math.random() * 5 - 2.5;
		
		var readout = sensorLib.read();
		
		var payload = {
			timestamp: Date.now(),
			location: process.env.MQTT_CLIENT_ID,
			temperature: readout.temperature, //mean + offset
			humidity: readout.humidity
		};
		
		var sensorMessage = new SensorMessage(payload);
		
		var buffer = sensorMessage.toBuffer();
		
		client.publish('phm/sensors', buffer, function(err) {
			if(err) {
				terminate();
				cb(err);
				return;
			}
			console.log('Published: ', payload);
		});
	}
	
	var mqtt    = require('mqtt');
	var mqttUrl = 'mqtt://' + (process.env.MQTT_HOST || 'localhost');
	if(process.env.MQTT_PORT) {
		mqttUrl += ':' + process.env.MQTT_PORT;
	}
	var client  = mqtt.connect(mqttUrl, {
		clientId: process.env.MQTT_CLIENT_ID,
		username: process.env.MQTT_USERNAME,
		password: <PASSWORD>
	});

	var handle = null;

	client.on('connect', function () {
		handle = setInterval(publishSomething, pollingInterval);
	});

	client.on('error', function (err) {
		terminate();
		cb(err);
	});
	
	process.on('SIGINT', function () {
		terminate();
		cb();
	});
}

async.waterfall([initializeSensor, loadProtoBuf, initializeMqtt], function(err) {
	if(err) {
		console.log(err);
	}
});<commit_msg>Improved logging.<commit_after>if(!process.env.MQTT_CLIENT_ID) {
	console.log('Please define your MQTT_CLIENT_ID.');
	return;
}

console.log('Starting up');

var async = require('async');
var pollingInterval = 1000 * (process.env.POLLING_INTERVAL || 5); 

function initializeSensor(cb) {
	var sensorLib = null;
	try {
		sensorLib = require('node-dht-sensor');
		sensorLib.initialize(11, process.env.SENSOR_GPIO || 4);
	} 
	catch(err) {
		console.log('You are not running on a BCM2835 hardware.');
		cb(err);
	}
	cb(null, sensorLib);
}

function loadProtoBuf(sensorLib, cb) {	
	var request = require('request');
	request.get('https://raw.githubusercontent.com/tabman83/phm-messages/master/sensor-message.proto', function (err, response, body) {
		if(err) {
			cb(err);
			return;
		}
		if(response.statusCode !== 200) {
			cb(new Error('Error while loading protobuf definition file (' + response.statusCode + ')'));
			return;
		}
		var ProtoBuf = require('protobufjs');
		var builder = ProtoBuf.loadProto(body);
		if(!builder) {
			cb(new Error('Error while decoding protobuf definition file'));
			return;
		}
		var SensorMessage = builder.build('SensorMessage');
		cb(null, sensorLib, SensorMessage);
	});	
}

function initializeMqtt(sensorLib, SensorMessage, cb) {
	
	var terminate = function() {
		if(handle) {
			clearTimeout(handle);
		}
		if(client) {
			client.end(true);
		}
	}	
	
	var publishSomething = function() {
		//var mean = 25;
		//var offset = Math.random() * 5 - 2.5;
		
		var readout = sensorLib.read();
		
		var payload = {
			timestamp: Date.now(),
			location: process.env.MQTT_CLIENT_ID,
			temperature: readout.temperature, //mean + offset
			humidity: readout.humidity
		};
		
		var sensorMessage = new SensorMessage(payload);
		
		var buffer = sensorMessage.toBuffer();
		
		client.publish('phm/sensors', buffer, function(err) {
			if(err) {
				terminate();
				cb(err);
				return;
			}
			console.log('Published: ', payload);
		});
	}
	
	var mqtt    = require('mqtt');
	var mqttUrl = 'mqtt://' + (process.env.MQTT_HOST || 'localhost');
	if(process.env.MQTT_PORT) {
		mqttUrl += ':' + process.env.MQTT_PORT;
	}
	
	console.log('Connecting to ' + mqttUrl);
	
	var client  = mqtt.connect(mqttUrl, {
		clientId: process.env.MQTT_CLIENT_ID,
		username: process.env.MQTT_USERNAME,
		password: <PASSWORD>
	});
	
	var handle = null;

	client.on('connect', function () {
		handle = setInterval(publishSomething, pollingInterval);
	});

	client.on('error', function (err) {
		terminate();
		cb(err);
	});
	
	process.on('SIGINT', function () {
		terminate();
		cb();
	});
}

async.waterfall([initializeSensor, loadProtoBuf, initializeMqtt], function(err) {
	if(err) {
		console.error(err);
	}
});
{'fraction_non_alphanumeric': 0.1599404872605542, 'fraction_numerical': 0.008183001673795797, 'mean_word_length': 3.92942254812099, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5776000', 'n_tokens_mistral': 2131, 'n_tokens_neox': 1862, 'n_words': 366}