gistfile1.js

/*
 *
 * (C) 2013, MangoRaft.
 *
 */
require('sugar');
var mongoose = require('mongoose');
var Schema = mongoose.Schema;
var Mixed = mongoose.Schema.Types.Mixed;

var roundMinute = function(d) {
	var t = new Date(d.getFullYear(), d.getMonth(), d.getDate(), d.getHours(), d.getMinutes());
	return t;
};

function getInitializer() {
	var updates = {};

	updates.values = [];

	for (var k = 0; k < 60; k++) {
		updates.values[k] = 0;
	};

	return updates;
}

function getUpdates(timestamp, value, inc) {

	var updates = {};

	//statistics
	updates['updatedAt'] = new Date();

	updates['$inc'] = {
		'num_samples' : 1,
		'total_samples' : value
	};

	if (inc) {
		updates['$inc']['values.' + timestamp.getSeconds()] = value;
	} else {
		updates['values.' + timestamp.getSeconds()] = value;
	}
	return updates;
}

/**
 * Schema definition
 */
var TimeSeries = new Schema({
	minute : {
		type : Date,
		index : true,
		required : true,
		unique : true
	},
	createdAt : {
		type : Date,
		'default' : Date
	},
	updatedAt : {
		type : Date,
		'default' : Date
	},
	num_samples : {
		type : Number,
		'default' : 0
	},
	total_samples : {
		type : Number,
		'default' : 0
	},
	values : [Number]
});

TimeSeries.static('push', function(timestamp, value, metadata, cb) {

	var minute = roundMinute(timestamp);

	var condition = {
		'minute' : minute
	};

	var updates = getUpdates(timestamp, value);

	var self = this;

	this.findOneAndUpdate(condition, updates, {
		select : '_id'
	}, function(error, doc) {

		if (error) {
			if (cb)
				cb(error);
		} else if (doc) {
			if (cb)
				cb(null, doc);
		} else {
			//console.log('Create new');
			var datainit = getInitializer(null);
			var doc = new self({
				minute : minute
			});
			doc.set(datainit);
			doc.set(updates);
			doc.save(cb);

		}
	});
});
TimeSeries.static('inc', function(timestamp, value, metadata, cb) {

	var minute = roundMinute(timestamp);

	var condition = {
		'minute' : minute
	};

	var updates = getUpdates(timestamp, value, true);

	var self = this;

	this.findOneAndUpdate(condition, updates, {
		select : '_id'
	}, function(error, doc) {
		console.log(error, doc)
		if (error) {
			if (cb)
				cb(error);
		} else if (doc) {
			if (cb)
				cb(null, doc);
		} else {
			//console.log('Create new');
			var datainit = getInitializer(null);
			var doc = new self({
				minute : minute
			});
			doc.set(datainit);
			doc.set(updates);
			doc.save(cb);

		}
	});
});

TimeSeries.static('findData', function(request, callback) {

	var condition = {
		'$and' : []
	};
	if (!request.to)
		request.to = new Date();
	else
		request.to = new Date(request.to);

	if (!request.dir)
		request.dir = 1;

	condition['$and'].push({
		'minute' : {
			'$gte' : request.from,
		}
	});
	condition['$and'].push({
		'minute' : {
			'$lte' : request.to,
		}
	});

	this.find(condition).sort({
		'minute' : request.dir
	}).exec(function(error, docs) {
		if (error) {
			callback(error);
		} else {
			var data = [];

			docs.forEach(function(doc) {
				doc.getData(request.to).forEach(function(row) {
					data.push(row);
				});
			});
			callback(null, data);
		}
	});
});

/**
 * Virtual methods
 */
TimeSeries.method('getData', function(to) {
	var data = [];
	var year = this.minute.getFullYear();
	var month = this.minute.getMonth();
	var day = this.minute.getDate();
	var hour = this.minute.getHours();
	var minute = this.minute.getMinutes();

	var self = this;

	function getSeconds(seconds) {
		var arry = [];
		for (var second = 0; second < seconds.length; second++) {
			if (roundMinute(to).getTime() == roundMinute(self.minute).getTime()) {
				if (hour == to.getHours() && minute == to.getMinutes() && second == to.getSeconds()) {
					return arry;
				}
			}
			var d = seconds[second];
			console.log(d)
			var timestamp = new Date(year, month, day, hour, minute, second);
			arry.push([timestamp.getTime(), d]);

		};
		return arry;
	}

	return getSeconds(this.values, minute, hour);
});
var TimeSeriesModel = function(collection, options) {

	var model, schema;
	var isNew = false

	/**
	 * Methods
	 */

	/**
	 * Model initialization
	 */
	function init(collection, options) {
		if (mongoose.connection.modelNames().indexOf(collection) >= 0) {
			model = connection.model(collection);

		} else {
			isNew = true
			model = mongoose.model(collection, TimeSeries);
		}

	}

	/**
	 * Push new value to collection
	 */
	var push = function push(timestamp, value, metadata, cb) {
		model.push(timestamp, value, metadata, cb);
	};
	/**
	 * inc value to collection
	 */
	var inc = function inc(timestamp, value, metadata, cb) {
		model.inc(timestamp, value, metadata, cb);
	};

	/**
	 *  Find data of given period
	 */
	var findData = function(options, cb) {
		model.findData(options, cb);
	};

	var getModel = function() {
		return model;
	};

	init(collection, options);

	/* Return model api */
	return {
		push : push,
		inc : inc,
		findData : findData,
		model : model
	};
};
module.exports = TimeSeriesModel;