--no commit message

--no commit message

plugins/dt/dt-agritronics/index.js

+var util = require('util');
+var DTPlugin = require('../dt-plugin');
+
+function DTTask(context,request){
+  DTPlugin.call(this,context,request);
+  this.name = "noop";
+  this.output_type = "";
+}
+util.inherits(DTTask,DTPlugin);
+
+DTTask.prototype.perform  =  require('./perform');
+
+module.exports = DTTask;

plugins/dt/dt-agritronics/package.json

+{
+  "name": "dt-agritronics-ibitz",
+  "version": "0.0.1",
+  "description": "",
+  "main": "index.js",
+  "scripts": {
+    "test": "echo \"Error: no test specified\" && exit 1"
+  },
+  "keywords": [
+    "bigstream"
+  ],
+  "author": "lsr.bigstream",
+  "license": "ISC",
+  "dependencies": {
+    "async": "^2.1.2",
+    "xml2json": "^0.10.0"
+  }
+}

plugins/dt/dt-agritronics/parser/agri_parser_factory.js

+var simpleParser = require('./simple_parser');
+var windDirParser = require('./wind_direction_parser');
+var imageParser = require('./image_parser');
+
+exports.getParser = function(type){
+	if(type === "Camera") return imageParser;
+	else if(type === "Wind Direction Degree") return windDirParser;
+	else return simpleParser;
+}
\ No newline at end of file

plugins/dt/dt-agritronics/parser/image_parser.js

+var fs = require('fs');
+var request = require('request');
+var async = require('async');
+
+exports.getValues = function(json, cb){
+
+	let dataTemplate = {
+		"type": json.xhr.IO.Type,
+		"unit": json.xhr.IO.Unit,
+		"value_type": json.xhr.IO.ValueType,
+		"image_type": json.xhr.IO.Name,
+		"values": []
+	};
+
+	let values = [];
+	
+	if(typeof json.xhr.IO.Data.length === "undefined"){
+
+		getImage(json.xhr.IO.Data.Value).then((data) =>{
+			values.push({"observeddatetime": json.xhr.IO.Data.IODateTime, "value": data});
+			if(values.length > 0){
+				dataTemplate.values = values;
+				cb(dataTemplate);
+			}
+			else cb(null);
+		}).catch((err) => {
+			cb(err);
+		})
+	}
+	else{
+		var idx = 0;
+		async.whilst(function() { return idx < json.xhr.IO.Data.length;}, function(callback) {
+			let d = json.xhr.IO.Data[idx];
+			idx++;
+
+			getImage(d.Value).then((data) =>{
+				values.push({"observeddatetime": d.IODateTime, "value": data});
+				callback();
+				
+			}).catch((err) => {
+				callback(err);
+			});
+
+		}, function(err) {
+		    if( err ) {
+		      console.log(err);
+		    } else {
+		    	if(values.length > 0){
+					dataTemplate.values = values;
+					cb(dataTemplate);
+				}
+				else cb(null);
+		    }
+		});
+		
+	}
+}
+
+
+
+function getImage(url) {
+	return new Promise((resolve, reject) => {
+		request(url, function (error, resp, body) {
+		    if (!error && resp.statusCode == 200) {
+		      resolve("data:" + resp.headers["content-type"] + ";base64," + new Buffer(body).toString('base64'));
+		    }else{
+		      return reject(error);
+		    }
+	  	})
+	})
+}
\ No newline at end of file

plugins/dt/dt-agritronics/parser/simple_parser.js

+exports.getValues = function(json, callback){
+	let dataTemplate = {
+		"type": json.xhr.IO.Type,
+		"unit": json.xhr.IO.Unit,
+		"value_type": json.xhr.IO.ValueType,
+		"values": []
+	};
+
+	let values = [];
+	
+	if(typeof json.xhr.IO.Data.length === "undefined"){
+		values.push({"observeddatetime": json.xhr.IO.Data.IODateTime, "value": json.xhr.IO.Data.Value});
+	}
+	else{
+		for (var i = 0; i < json.xhr.IO.Data.length; i++) {
+			let d = json.xhr.IO.Data[i];
+			values.push({"observeddatetime": d.IODateTime, "value": d.Value});
+		}
+	}
+	
+	//console.log("values.langth >>>" + values.length);
+	if(values.length > 0){
+		dataTemplate.values = values;
+		callback(dataTemplate);
+	}
+	else callback(null);
+}
\ No newline at end of file

plugins/dt/dt-agritronics/parser/wind_direction_parser.js

+exports.getValues = function(json, callback){
+	let dataTemplate = {
+		"type": json.xhr.IO.Type,
+		"unit": json.xhr.IO.Unit,
+		"value_type": json.xhr.IO.ValueType,
+		"values": []
+	};
+
+	let values = [];
+	
+	if(typeof json.xhr.IO.Data.length === "undefined"){
+		values.push({"observeddatetime": json.xhr.IO.Data.IODateTime, "value": json.xhr.IO.Data.Value});
+	}
+	else{
+		for (var i = 0; i < json.xhr.IO.Data.length; i++) {
+			let d = json.xhr.IO.Data[i];
+			values.push({"observeddatetime": d.IODateTime, "value": d.Value, "direction": d.Direction});
+		}
+	}
+	
+	//console.log("values.langth >>>" + values.length);
+	if(values.length > 0){
+		dataTemplate.values = values;
+		callback(dataTemplate);
+	}
+	else callback(null);
+}
\ No newline at end of file

plugins/dt/dt-agritronics/perform.js

+function perform_function(context,request,response){
+  var job_id = context.jobconfig.job_id;
+  var transaction_id = context.transaction.id;
+  var param = context.jobconfig.data_transform.param;
+  var memstore = context.task.memstore
+
+  var output_type = request.input_type;
+  var di_data = request.data;
+
+  var agriParser = require('./parser/agri_parser_factory');
+  var async = require('async');
+  var parser = require('xml2json');
+  var fs = require('fs');
+
+  let idx = 0;
+  let result = {
+    "station_id": di_data.station_id,
+    "latitude": "",
+    "longitude": "",
+    "data":[]
+  };
+    
+  //console.log(json_table.length);
+  async.whilst(function() { return idx < di_data.data.length;}, function(callback) {
+    let json = parser.toJson(di_data.data[idx].value, {object: true});
+    idx++;
+    if(typeof json.xhr.IO.Data !== 'undefined') {
+      agriParser.getParser(json.xhr.IO.Type).getValues(json, (values) => {
+        if(values !== null){
+            result.latitude = json.xhr.IO.Latitude;
+            result.longitude = json.xhr.IO.Longitude;
+            result.data.push(values);
+            callback();
+        }
+      });
+    }
+      
+  }, function(err) {
+    console.log("CB");
+      if( err ) {
+        console.log(err);
+      } else {
+        console.log('data out...');
+        fs.writeFileSync("./result.json", JSON.stringify(result));
+        console.log(JSON.stringify(result));
+      }
+  });
+
+
+  // memstore.setItem('lasttransaction',transaction_id,function(err){
+  //   response.success(data);
+  // });
+
+  // memstore.getItem('lasttransaction',function(err,value){
+  //   response.success(value);
+  // });
+  //data = data + "--DT--"
+
+  response.success(result,output_type);
+  //response.reject();
+  //response.error("error message")
+
+}
+
+module.exports = perform_function;