exports.readme = `# send all data to projects.worksys.io, required to monitor status of controller(unipi)`;
-const fs = require('fs');
-var path = require('path');
+exports.install = function(instance) {
-exports.install = async function(instance) {
-
+ let id;
let allValues = {};
let sendAllValuesInterval;
+ let configured = false;
let now = new Date();
console.log(exports.title, "INSTALLED", now.toLocaleString("sk-SK"));
@@ -47,23 +43,19 @@ exports.install = async function(instance) {
}
}
}
-
function sendValues()
{
- const id = FLOW.OMS_projects_id;
+ if(!configured) return;
if(Object.keys(allValues).length > 0)
{
- if(id !== undefined)
+ if(id)
{
delete allValues.__force__;
let dataToSend = {...allValues};
dataToSend.id = id;
dataToSend.ipAddresses = ipAddresses;
- //dataToSend.notify_date = new Date().toISOString().slice(0, 19).replace('T', ' ');
-
- //console.log(exports.title, "------------>sendValues", dataToSend);
instance.send(0, dataToSend);
@@ -71,7 +63,7 @@ exports.install = async function(instance) {
}
else
{
- console.log(exports.title, "unable to send data, id is undefined");
+ console.log(exports.title, "unable to send data, no id");
}
}
@@ -81,10 +73,14 @@ exports.install = async function(instance) {
clearInterval(sendAllValuesInterval);
})
- instance.on("data", (flowdata) => {
+ instance.on("0", _ => {
+ id = FLOW.GLOBALS.settings.project_id;
+ configured = true;
+ })
+
+ instance.on("1", flowdata => {
allValues = { ...allValues, ...flowdata.data};
-
//console.log("DATA RECEIVED", flowdata.data);
//__force__
diff --git a/flow/modbus_citysys.js b/flow/modbus_citysys.js
new file mode 100644
index 0000000..d038b69
--- /dev/null
+++ b/flow/modbus_citysys.js
@@ -0,0 +1,1127 @@
+exports.id = 'modbus_citysys';
+exports.title = 'Modbus_citysys';
+exports.version = '1.0.0';
+exports.group = 'Worksys';
+exports.color = '#2134B0';
+exports.input = 1;
+exports.output = ["red", "white", "blue", "orange"];
+exports.click = false;
+exports.author = 'Jakub Klena';
+exports.icon = 'bolt';
+exports.options = { edge: "undefined" };
+
+exports.html = `
`;
+
+exports.readme = `# Energomonitor
+## Outputs
+
+ - *Red* - ERROR output (can connect to filewriter or something)
+ - *White* - STATUS output (answers to your commands, ERRORS and WARNINGS from your commands go both to this and to their own outputs, so they get logged)
+ - *Blue* - TB output (pure data for TB)
+`;
+
+
+const instanceSendTo = {
+ error: 0,
+ debug: 1,
+ tb: 2,
+ di_do_controller: 3
+}
+
+const DataToTbHandler = require('./helper/DataToTbHandler.js');
+const { sendNotification } = require('./helper/notification_reporter.js');
+const dbRelays = TABLE("relays");
+const { promisifyBuilder, makeMapFromDbResult } = require('./helper/db_helper.js');
+let tbname;
+
+async function loadSettings()
+{
+ //todo global FLOW.OMS_edgeName is making problem, so we load it here as well, it should not be
+ let responseRelays = await promisifyBuilder(dbRelays.find());
+ FLOW.OMS_edgeName = responseRelays[0]["tbname"];
+ tbname = FLOW.OMS_edgeName;
+}
+
+loadSettings();
+
+exports.install = function(instance) {
+ const SerialPort = require('serialport');
+ const { exec } = require('child_process');
+ const fs = require("fs");
+ const filepath = F.path.root("saved_data/modbus_settings");
+ const backup_filepath = F.path.root("saved_data/modbus_settings_backup");
+
+ const ErrorToServiceHandler = require('./helper/ErrorToServiceHandler.js');
+ const errorHandler = new ErrorToServiceHandler();
+
+ let receivedDataArray = [];
+
+
+ instance.CONFIG = {
+ "isRunning": false,
+ "debug": true,
+ "timeoutTime": 10000,
+ "msgWaitTime": 1000,
+ "port": "/dev/ttymxc1",
+ //"port_options": "stty -F /dev/ttymxc1 115200 min 1 time 5 ignbrk -brkint -icrnl -imaxbel -opost -isig -icanon -iexten -echo -echoe -echok -echoctl -echoke"
+ "port_options": "stty -F /dev/ttymxc1 9600 min 1 time 5 ignbrk -brkint -icrnl -imaxbel -opost -isig -icanon -iexten -echo -echoe -echok -echoctl -echoke"
+ };
+
+ let PRIVATEVARS = {
+ "errBuffer": [], // Buffer for error messages
+ "tbBuffer": [], // Buffer for TB push messages
+ "device_index": 0,
+ "cmd_index": -1,
+ "devices": [
+ /*{
+ "name": "Elektrometer 1",
+ "tb_name": "EOzNMgZ9n43qPbjXmy7zwdA2DKdYvW5e6pxGRrVa",
+ "type": "EM111",
+ "address": 1,
+ "data":[],
+ "cmd":[],
+ "timeoutcount":0,
+ "status":"virtual"
+ },*/
+ // {
+ // "name": "Elektrometer 2",
+ // "tb_name": "pJX1ObgmqGZ54DMyYL7aDdkEVdve38WKRzwjNrQ9",
+ // "type": "EM111",
+ // "address": 2,
+ // "data":[],
+ // "cmd":[],
+ // "timeoutcount":0,
+ // "status":"virtual"
+ // },
+ // {
+ // "name": "Elektrometer 3",
+ // "tb_name": "XRvmwNz8QPblKp41GD7lKVkJrLVYoBO92dMegn6W",
+ // "type": "EM111",
+ // "address": 3,
+ // "data":[],
+ // "cmd":[],
+ // "timeoutcount":0,
+ // "status":"virtual"
+ // },
+ // {
+ // "name": "Elektrometer 4",
+ // "tb_name": "oRO8rjaBDy21qPQJzW7oD9ApK3xmNleVZg9Ed4Gw",
+ // "type": "EM111",
+ // "address": 4,
+ // "data":[],
+ // "cmd":[],
+ // "timeoutcount":0,
+ // "status":"virtual"
+ // },
+ {
+ "name": "Elektrometer 1",
+ "tb_name": "KjbN4q7JPZmexgdnz2yKQ98YAWwO0Q3BMX6ERLoV",
+ "type": "EM340",
+ "address": 1,
+ "data":[],
+ "cmd":[],
+ "timeoutcount":0,
+ "status":"virtual"
+ }
+ ],
+ "cmd_tables": [
+ {
+ "type":"EM340",
+ "cmd":[
+ {
+ "name": "Voltage L1",
+ "tb_name": "a",
+ "register": 0,
+ "size": 2,
+ "multiplier": 0.1
+ },
+ {
+ "name": "Voltage L2",
+ "tb_name": "b",
+ "register": 2,
+ "size": 2,
+ "multiplier": 0.1
+ },
+ {
+ "name": "Voltage L3",
+ "tb_name": "c",
+ "register": 4,
+ "size": 2,
+ "multiplier": 0.1
+ },
+ {
+ "name": "Current L1",
+ "tb_name": "d",
+ "register": 12,
+ "size": 2,
+ "multiplier": 0.001
+ },
+ {
+ "name": "Current L2",
+ "tb_name": "e",
+ "register": 14,
+ "size": 2,
+ "multiplier": 0.001
+ },
+ {
+ "name": "Current L3",
+ "tb_name": "f",
+ "register": 16,
+ "size": 2,
+ "multiplier": 0.001
+ }
+
+
+ // {
+ // "name": "Power factor",
+ // "tb_name": "power_factor",
+ // "register": 14,
+ // "size": 1,
+ // "multiplier": 0.001
+ // },
+ // {
+ // "name": "Frequency",
+ // "tb_name": "frequency",
+ // "register": 15,
+ // "size": 1,
+ // "multiplier": 0.1
+ // },
+ // {
+ // "name": "Energy",
+ // "tb_name": "consumption",
+ // "register": 16,
+ // "size": 2,
+ // "multiplier": 0.1
+ // }
+ ]
+ }
+ ]
+ };
+
+ let ERRWEIGHT = {
+ EMERGENCY: "emergency", // System unusable
+ ALERT: "alert", // Action must be taken immidiately
+ CRITICAL: "critical", // Component unable to function
+ ERROR: "error", // Error, but component able to recover from it
+ WARNING: "warning", // Possibility of error, system running futher
+ NOTICE: "notice", // Significant message but not an error, things user might want to know about
+ INFO: "informational", // Info
+ DEBUG: "debug" // Debug - only if CONFIG.debug is enabled
+ };
+
+ instance.currentData = function(){
+ let resp = [];
+ for (let f = 0; f < PRIVATEVARS.devices.length; f++){
+ let dev = PRIVATEVARS.devices[f];
+ for (let e = 0; e < dev.data.length; e++){
+ let d = dev.data[e];
+ resp.push({
+ "name": dev.name+" - "+d.name,
+ "value": d.value
+ });
+ }
+ }
+ return resp;
+ };
+
+ instance.configList = function(){
+ let resp = [];
+ /*let data = PRIVATEVARS.feeds;
+ for (let a = 0; a < data.length; a++){
+ for (let i = 0; i < instance.CONFIG.feeds.length; i++){
+ let feed = instance.CONFIG.feeds[i];
+ if (feed.name === data[a].id){
+ for (let b = 0; b < data[a].streams.length; b++){
+ for (let j = 0; j < feed.streams.length; j++){
+ let stream = feed.streams[j];
+ if (stream.name === data[a].streams[b].id){
+ data[a].streams[b]["exists"] = true;
+ data[a].streams[b]["currently"] = stream;
+ }
+ }
+ }
+ }
+ }
+ }
+ resp.push({
+ "name":"Device manager",
+ "icon":"tasks",
+ "_show":false,
+ "js_func":"energoDevManager",
+ "data": data
+ });*/
+
+ return resp;
+ }
+
+ let timeoutInterval = null;
+ let msgWaitInterval = null;
+ let port = null;
+ let myEdge = "undefined";
+ let starter = null;
+ instance.status("Loading...", "red");
+
+
+ instance.availableCommands = [
+ {
+ "name": "Status",
+ "cmd": "qStatus",
+ "icon": "stream",
+ "func": function(body){
+ let a = true;
+ if (timeoutInterval === null){
+ a = false;
+ }
+ let b = true;
+ if (msgWaitInterval === null){
+ b = false;
+ }
+ let st = {
+ "isRunning":instance.CONFIG.isRunning,
+ "timeoutInterval":a,
+ "msgWaitInterval":b,
+ "CONFIG":instance.CONFIG
+ };
+ return {
+ "type": "ok",
+ "timestamp": humanReadableTimeAndDate(),
+ "resp": st
+ };
+ }
+ },
+ {
+ "name": "Start Reading",
+ "cmd": "sStart",
+ "icon": "play",
+ "func": function(body){
+ /*if (running === false){
+ startCmdWaitInterval();
+ running = true;
+ return "Reading started !";
+ } else {
+ return "Reading already active !";
+ }*/
+ return {
+ "type": "ok",
+ "timestamp": humanReadableTimeAndDate(),
+ "resp": "WIP"
+ };
+ }
+ },
+ {
+ "name": "Stop Reading",
+ "cmd": "sStop",
+ "icon": "stop",
+ "func": function(body){
+ /*if (running === true){
+ stopCmdWaitInterval();
+ stopTimeoutInterval();
+ running = false;
+ return "Reading stopped !";
+ } else {
+ return "Reading already inactive !";
+ }*/
+ return {
+ "type": "ok",
+ "timestamp": humanReadableTimeAndDate(),
+ "resp": "WIP"
+ };
+ }
+ },
+ {
+ "name": "Read current data",
+ "cmd": "qCurrentData",
+ "icon": "chart-bar",
+ "func": function(body){
+ let resp = instance.currentData();
+ return {
+ "type": "ok",
+ "timestamp": humanReadableTimeAndDate(),
+ "resp": resp
+ };
+ }
+ },
+ {
+ "name": "Save current config",
+ "cmd": "saveConfig",
+ "icon": "save",
+ "func": function(body){
+
+ instance.set("config", JSON.stringify(instance.CONFIG));
+ instance.set("private", JSON.stringify(PRIVATEVARS));
+ return {
+ "type": "ok",
+ "timestamp": humanReadableTimeAndDate(),
+ "resp": "done"
+ };
+ }
+ },
+ {
+ "name": "Toggle debug",
+ "cmd": "sDebug",
+ "icon": "comment-dots",
+ "func": function(body){
+
+ if (instance.CONFIG.debug){
+ instance.CONFIG.debug = false;
+ instance.set("config", JSON.stringify(instance.CONFIG));
+
+ return {
+ "type": "ok",
+ "timestamp": humanReadableTimeAndDate(),
+ "resp": "debug OFF"
+ };
+ } else {
+ instance.CONFIG.debug = true;
+ instance.set("config", JSON.stringify(instance.CONFIG));
+
+ return {
+ "type": "ok",
+ "timestamp": humanReadableTimeAndDate(),
+ "resp": "debug ON"
+ };
+ }
+
+ }
+ }
+ ];
+
+
+
+ function saveData(){
+ if (checkFile(filepath)){
+ let content = undefined;
+ try {
+ content = fs.readFileSync(filepath);
+ } catch (err){
+ console.log("saveData", myEdge, ERRWEIGHT.ERROR, "Unable to read original configuration !", {"name":instance.name, "id":instance.id, "file":filepath, "err":err.message});
+
+ //sendError("saveData", myEdge, ERRWEIGHT.ERROR, "Unable to read original configuration !", {"name":instance.name, "id":instance.id, "file":filepath, "err":err.message});
+ }
+
+ if (content !== undefined){
+ try {
+ fs.writeFileSync(backup_filepath, content, "utf8");
+ } catch (err) {
+ //sendError("saveData", myEdge, ERRWEIGHT.ERROR, "Unable to save backup of configuration !", {"name":instance.name, "id":instance.id, "file":backup_filepath, "err":err.message});
+ console.log("saveData", myEdge, ERRWEIGHT.ERROR, "Unable to save backup of configuration !", {"name":instance.name, "id":instance.id, "file":backup_filepath, "err":err.message});
+ }
+ }
+ }
+
+ let a = {
+ "config":instance.CONFIG,
+ "private":PRIVATEVARS
+ };
+
+ try {
+ fs.writeFileSync(filepath, JSON.stringify(a), "utf8");
+ } catch (err) {
+ //sendError("saveData", myEdge, ERRWEIGHT.CRITICAL, "Unable to save configuration !", {"name":instance.name, "id":instance.id, "file":filepath, "err":err.message});
+ console.log("saveData", myEdge, ERRWEIGHT.CRITICAL, "Unable to save configuration !", {"name":instance.name, "id":instance.id, "file":filepath, "err":err.message});
+ }
+ }
+
+ function loadData(){
+ let content = undefined;
+ //console.log(filepath);
+ if (checkFile(filepath)){
+ try {
+ content = fs.readFileSync(filepath);
+ } catch (err){
+ //sendError("loadData", myEdge, ERRWEIGHT.ERROR, "Unable to read original configuration !", {"name":instance.name, "id":instance.id, "file":filepath, "err":err.message});
+ console.log("loadData", myEdge, ERRWEIGHT.ERROR, "Unable to read original configuration !", {"name":instance.name, "id":instance.id, "file":filepath, "err":err.message});
+ }
+ } else {
+ if (checkFile(backup_filepath)){
+ try {
+ content = fs.readFileSync(backup_filepath);
+ } catch (err){
+ //sendError("loadData", myEdge, ERRWEIGHT.CRITICAL, "Unable to read backup configuration !", {"name":instance.name, "id":instance.id, "file":backup_filepath, "err":err.message});
+ console.log("loadData", myEdge, ERRWEIGHT.CRITICAL, "Unable to read backup configuration !", {"name":instance.name, "id":instance.id, "file":backup_filepath, "err":err.message});
+ }
+ if (content !== undefined){
+ //sendError("loadData", myEdge, ERRWEIGHT.WARNING, "No configuration, loading from backup !", {"name":instance.name, "id":instance.id});
+ console.log("loadData", myEdge, ERRWEIGHT.WARNING, "No configuration, loading from backup !", {"name":instance.name, "id":instance.id});
+ }
+ } else {
+ //sendError("loadData", myEdge, ERRWEIGHT.CRITICAL, "No configuration, not even backup !", {"name":instance.name, "id":instance.id});
+ console.log("loadData", myEdge, ERRWEIGHT.CRITICAL, "No configuration, not even backup !", {"name":instance.name, "id":instance.id, "filepath": filepath});
+ }
+ }
+
+
+
+ if (content !== undefined){
+ let a = JSON.parse(content);
+ instance.send(instanceSendTo.debug, a);
+ let c = a.config;
+ let p = a.private;
+
+
+ if (c === undefined){
+ //sendError("loadData", myEdge, ERRWEIGHT.CRITICAL, "Configuration not found !", {"name":instance.name, "id":instance.id});
+ console.log("loadData", myEdge, ERRWEIGHT.CRITICAL, "Configuration not found !", {"name":instance.name, "id":instance.id});
+ instance.status("Error - no config", "red");
+ } else if (p === undefined){
+ //sendError("loadData", myEdge, ERRWEIGHT.CRITICAL, "Privatevars not found !", {"name":instance.name, "id":instance.id});
+ console.log("loadData", myEdge, ERRWEIGHT.CRITICAL, "Privatevars not found !", {"name":instance.name, "id":instance.id});
+ instance.status("Error - no private vars", "red");
+ } else {
+ instance.CONFIG = c;
+ PRIVATEVARS = p;
+
+ // Daj kazdemu device jeho tabulku prikazu
+ for (let i = 0; i < PRIVATEVARS.devices.length; i++){
+ let device = PRIVATEVARS.devices[i];
+ for (let j = 0; j < PRIVATEVARS.cmd_tables.length; j++){
+ let table = PRIVATEVARS.cmd_tables[j];
+
+ if (device.type === table.type){
+ PRIVATEVARS.devices[i].cmd = table.cmd;
+ }
+ }
+ }
+
+ if (myEdge === "undefined"){
+ instance.status("Unconfigured", "red");
+ } else {
+ instance.status("Running", "green");
+ startCmdWaitInterval();
+
+ instance.CONFIG.isRunning = true;
+ console.log("modbus loaded: ", PRIVATEVARS.devices);
+ }
+ }
+ }
+ }
+
+ function checkFile(name){
+ try {
+ fs.accessSync(name, fs.constants.F_OK | fs.constants.R_OK | fs.constants.W_OK);
+ return true;
+ } catch (err) {
+ return false;
+ }
+ }
+
+ //Zapina slucku vycitavania dat
+ function readDeviceData(){
+ stopCmdWaitInterval();
+
+ // let tbname = FLOW.OMS_edgeName;
+
+ // Check port existance
+ if (port === null)
+ {
+ port = new SerialPort(instance.CONFIG.port);
+
+ port.on('error', function(err) {
+ //logger.debug("rsPort opened error - failed", err.message);
+ //instance.send(instanceSendTo.debug, err.message);
+
+ errorHandler.sendMessageToService( exports.title + " MODBUS RS485 open - failed: " + err.message);
+ })
+
+ port.on('open', function() {
+
+ console.log("--->MODBUS RS485 READY - port opened");
+
+ exec("sudo halfduplex /dev/ttymxc1", (error, stdout, stderr) => {
+ instance.send(instanceSendTo.debug, {"err": error});
+
+ if (error) {
+ console.log("--->MODBUS RS485", error, stderr);
+ errorHandler.sendMessageToService( exports.title + " sudo halfduplex /dev/ttymxc1 - failed: " + error);
+ }
+
+ });
+
+ exec(instance.CONFIG.port_options, (error, stdout, stderr) => {
+ instance.send(instanceSendTo.debug, {"stdout":stdout,"stderr":stderr,"err":error});
+
+ if (error) {
+ console.log("--->MODBUS RS485", error, stderr);
+ errorHandler.sendMessageToService( exports.title + " " + instance.CONFIG.port_options + " - failed: " + error);
+ }
+
+ });
+
+ });
+
+ port.on('data', receivedData);
+
+ //sendError("readDeviceData", myEdge, ERRWEIGHT.DEBUG, "Serial port open!", {});
+ //console.log("-->MODBUS readDeviceData", myEdge, ERRWEIGHT.DEBUG, "Serial port open!", {});
+
+ startCmdWaitInterval();
+ return; // Cakame na port aby sa spravne otvoril a rozbehol
+ }
+
+
+ // Skontroluj existenciu device listu
+ if (PRIVATEVARS.devices.length > 0){
+ // Ponastavuj indexy
+ PRIVATEVARS.cmd_index++;
+ if (PRIVATEVARS.cmd_index >= PRIVATEVARS.devices[PRIVATEVARS.device_index].cmd.length){
+ // Kedže všetky príkazy pre daný device sú vybavené, je treba odoslat vyčítané data do TB
+ updateDataInTB();
+
+ PRIVATEVARS.cmd_index = 0;
+ PRIVATEVARS.device_index++;
+
+ if (PRIVATEVARS.device_index >= PRIVATEVARS.devices.length){
+ PRIVATEVARS.device_index = 0;
+ }
+ }
+
+ let device = PRIVATEVARS.devices[PRIVATEVARS.device_index];
+
+ // Skontroluj existenciu príkazú pre daný device type
+ if (device.cmd.length < 1){
+ //sendError("readDeviceData", tbname, ERRWEIGHT.ERROR, "No commands for this device type !", {"type": device.type});
+ console.log("readDeviceData", tbname, ERRWEIGHT.ERROR, "No commands for this device type !", {"type": device.type});
+ startCmdWaitInterval();
+ return;
+ }
+
+ // Odešli nasledujúci príkaz
+ sendCommand();
+
+ } else {
+ instance.CONFIG.isRunning = false;
+ //sendError("readDeviceData", myEdge, ERRWEIGHT.CRITICAL, "Modbus has no devices registered!", {});
+ console.log("modbus_citys: readDeviceData", myEdge, ERRWEIGHT.CRITICAL, "Modbus has no devices registered!", {});
+ }
+ }
+
+ function readingTimeouted(){
+ stopCmdWaitInterval();
+ stopTimeoutInterval();
+
+ // let tbname = FLOW.OMS_edgeName;
+
+ let device = PRIVATEVARS.devices[PRIVATEVARS.device_index];
+ let com = device.cmd[PRIVATEVARS.cmd_index];
+ //sendError("readingTimeouted", tbame, ERRWEIGHT.WARNING, "Reading timeouted !", {"device": device.address, "cmd": com.register});
+ console.log("modbus_citys: readingTimeouted", tbname, ERRWEIGHT.WARNING, "Reading timeouted !", {"device": device.address, "cmd": com.register});
+
+ device.timeoutcount++;
+ //console.log("device.timeoutcount", device.timeoutcount);
+ if (device.timeoutcount === 16)
+ {
+
+ //sendError("modbus_citys: readingTimeouted", tbname, ERRWEIGHT.CRITICAL, "Electrometer is not responding - reading timeouted", "");
+ sendNotification("modbus_citys: readingTimeouted", tbname, "electrometer_is_not_responding", {}, "", instanceSendTo.tb, instance );
+
+ if (device.status === "OK"){
+ device.status === "NOK";
+ }
+ }
+
+ startCmdWaitInterval();
+ }
+
+ function receivedData(data){
+
+ //let array = [...data];
+ //console.log("received data", array);
+
+ // let tbname = FLOW.OMS_edgeName;
+
+ //!if received data are less than 9 bytes, we store it in array variable and return. than we concatenate second incoming
+ // data and then length of array is 9
+ receivedDataArray = [...receivedDataArray, ...data];
+ //if (array.length < 9) return;
+ let l = receivedDataArray.length;
+ //console.log("received",receivedDataArray, l)
+
+ if ( l < 7 || l > 9 || l == 8 ) return;
+
+ let device = PRIVATEVARS.devices[PRIVATEVARS.device_index];
+ let com = device.cmd[PRIVATEVARS.cmd_index];
+
+ if (device.timeoutcount > 16) {
+ //sendError("Modbus_citysys: receivedData", tbname, ERRWEIGHT.NOTICE, "Electrometer is responding again", "");
+ sendNotification("modbus_citys: receivedData", tbname, "electrometer_is_responding_again", {}, "", instanceSendTo.tb, instance );
+ }
+
+ device.timeoutcount = 0;
+
+ if ((l == 7 && com.size != 1) || ( l == 9 && com.size != 2)) return;
+
+ stopTimeoutInterval();
+
+ //sendError("receivedData", tbname, ERRWEIGHT.DEBUG, "Received data !", {"cmd": receivedDataArray});
+ //console.log("receivedData", tbname, ERRWEIGHT.DEBUG, "Received data !", {"cmd": receivedDataArray});
+
+ // Skontroluj či sedí počet bytú v správe
+ //console.log("com size", com.size*2, "array2", receivedDataArray[2]);
+ if (receivedDataArray[2] !== (com.size*2)){
+ //sendError("receivedData", tbname, ERRWEIGHT.ERROR, "Received data of incorrect size !", {"expected": (com.size*2), "received": receivedDataArray[2], "cmd": com.register, "whole_msg": receivedDataArray});
+ console.log("modbus_citys: receivedData", tbname, ERRWEIGHT.ERROR, "Received data of incorrect size !", {"expected": (com.size*2), "received": receivedDataArray[2], "cmd": com.register, "whole_msg": receivedDataArray});
+ startTimeoutInterval();
+ } else {
+ // Konvertuj raw data na human readable
+
+ let v = (receivedDataArray[3] << 8) | receivedDataArray[4];
+ if (com.size == 2){
+ v = v | (receivedDataArray[5] << 24) | (receivedDataArray[6] << 16);
+ }
+ v = Math.round((v * com.multiplier) * 100) / 100;
+
+
+ // Pokad device nemá ešte žádné hodnoty vyčítané, pushni túto hodnotu do pola
+ if (device.data.length < 1){
+ device.data.push({ // Vždy ked správne zakomunikuje obnov status na OK
+ "changed": true,
+ "name": "status",
+ "value": "OK"
+ });
+ device.data.push({
+ "changed": true,
+ "name": com.tb_name,
+ "value": v
+ });
+ } else {
+ // Kedže už neco v poli má, kukni sa či je tam aj táto hodnota
+ let found = false;
+ for (let i = 0; i < device.data.length; i++){
+ let d = device.data[i];
+ if (d.name == "status"){ // Ked natrefíš na status (vždy tam musí byt) prepíš ho na OK lebo zakomunikoval správne
+ device.data[i].changed = true;
+ device.data[i].value = "OK";
+ }
+
+ if (d.name == com.tb_name){
+ found = true;
+ device.data[i].changed = true;
+ device.data[i].value = v;
+ }
+ }
+
+ // Pole existuje, ale táto hodnota tam neni, pridaj ju
+ if (found === false){
+ device.data.push({
+ "changed": true,
+ "name": com.tb_name,
+ "value": v
+ });
+ }
+ }
+
+ //Správne sme prijali odpoveď, je čas na další msg
+ startCmdWaitInterval();
+ }
+ //console.log('received data array', receivedDataArray);
+ receivedDataArray = [];
+ }
+
+ function updateDataInTB(){
+ let device = PRIVATEVARS.devices[PRIVATEVARS.device_index];
+
+ // console.log("---- MB device", device);
+ // console.log("---MB device data", device.data);
+
+ let values = "";
+ for (let i = 0; i < device.data.length; i++){
+ let data = device.data[i];
+ if (data.changed){
+ if (values !== ""){
+ values += ", ";
+ }
+
+ if (data.name === "status"){
+ values += "\""+data.name+"\":\""+data.value+"\"";
+ // This makes sure, that if this device doesn’t respond even once in next reading cycle, it will be marked as NOK
+ device.data[i].changed = true;
+ device.data[i].value = "NOK";
+ } else {
+ values += "\""+data.name+"\":"+data.value;
+ device.data[i].changed = false;
+ }
+ }
+
+ }
+
+ //console.log("values modbus", values);
+
+
+ if (values !== ""){
+
+ // let tbname = FLOW.OMS_edgeName;
+ // if(tbname == "" || tbname === undefined )
+ // {
+ // console.log("!!!!!!FLOW.OMS_edgeName is empty - 1");
+ // return;
+ // }
+
+ let tbmsg = "{\"" + tbname + "\":[{\"ts\":"+Date.now()+", \"values\":{"+values+"}}] }";
+ tbmsg = JSON.parse(tbmsg);
+
+ values = tbmsg[tbname][0]["values"];
+
+ //console.log("modbus", Object.keys(values));
+
+ //sum Phase_1_power, Phase_2_power, Phase_3_power (if one of them is undefined, we handle it)
+ const numOr0 = n => isNaN(n) ? 0 : n;
+ let calculated_total_power = [values["Phase_1_power"], values["Phase_2_power"], values["Phase_3_power"]].reduce((a, b) => numOr0(a) + numOr0(b));
+ values["total_power"] = parseFloat(calculated_total_power.toFixed(2));
+ tbmsg[tbname][0]["values"] = values;
+
+ Object.keys(values).map(singleValue => {
+ if (["Phase_1_voltage", "Phase_2_voltage", "Phase_3_voltage"].includes(singleValue))
+ {
+
+ let l = singleValue.split("_");
+ let phase = parseInt(l[1]);
+
+ if(FLOW.OMS_no_voltage == undefined) FLOW.OMS_no_voltage = new Set();
+
+ if(values[singleValue] == 0)
+ {
+ sendNotification("modbus_citys: updateDataInTB", tbname, "no_voltage_detected_on_phase", {phase: phase}, "", instanceSendTo.tb, instance, "voltage" + phase );
+
+ FLOW.OMS_no_voltage.add(phase);
+ }
+ else
+ {
+ FLOW.OMS_no_voltage.delete(phase);
+ sendNotification("modbus_citys: updateDataInTB", tbname, "voltage_on_phase_has_been_restored", {phase: phase}, "", instanceSendTo.tb, instance, "voltage" + phase);
+ }
+
+ }
+ })
+
+ sendThingsBoard(tbmsg);
+ }
+ }
+
+
+ let electrometerNotResponding = 0;
+ function sendCommand(){
+ let device = PRIVATEVARS.devices[PRIVATEVARS.device_index];
+ let com = device.cmd[PRIVATEVARS.cmd_index];
+ let array = [device.address, 3, ((com.register >> 8) & 0xFF), (com.register & 0xFF), ((com.size >> 8) & 0xFF), (com.size & 0xFF)];
+ array = modbusCRC(array);
+
+ //console.log('---device--', device);
+ //console.log('---device type--', device.type);
+
+ //sendError("sendCommand", device.tb_name, ERRWEIGHT.DEBUG, "Sending command !", {"cmd": array});
+ //console.log("sendCommand", device.tb_name, ERRWEIGHT.DEBUG, "Sending command !", {"cmd": array});
+
+ // let tbname = FLOW.OMS_edgeName;
+ // if(tbname == "" || tbname === undefined )
+ // {
+ // console.log("!!!!!!FLOW.OMS_edgeName is empty - 2");
+ // return;
+ // }
+
+ startTimeoutInterval();
+ port.write(Buffer.from(array), function(err) {
+
+ //! poslany command
+ //console.log("poslany tento commant", array, err, device.type);
+
+ if (err) {
+ stopTimeoutInterval();
+ stopCmdWaitInterval();
+
+
+ // elektromer neodpoveda viac ako 5 minut (15 commands za minutu sa posiela)
+ if (device.type === "EM111" || device.type === "EM340")
+ {
+ electrometerNotResponding++;
+
+ if (electrometerNotResponding > 15 && electrometerNotResponding < 17)
+ {
+
+ //sendError("Modbus_citys: sendCommand", tbname, ERRWEIGHT.CRITICAL, "Electrometer is not responding", {"err": err.message, "info": "No response more than 5 minutes"});
+ sendNotification("modbus_citys: sendCommand", tbname, "electrometer_is_not_responding", {}, {"err": err.message, "info": "No response more than 5 minutes"}, instanceSendTo.tb, instance );
+
+ let tbmsg = {
+ [tbname]: [
+ {
+ "ts": Date.now(),
+ "values": {
+ "status": "NOK"
+ }
+ }
+ ]
+ }
+
+ sendThingsBoard(tbmsg)
+ }
+ }
+
+
+ return;
+ }
+
+ if (device.type === "EM111")
+ {
+ if (electrometerNotResponding > 15)
+ {
+ //sendError("Modbus_citys: sendCommand", tbname, ERRWEIGHT.NOTICE, "Electrometer is responding again", "");
+ sendNotification("modbus_citys: sendCommand", tbname, "electrometer_is_responding_again", {}, "", instanceSendTo.tb, instance );
+ }
+ electrometerNotResponding = 0;
+
+ }
+ });
+ }
+
+ function modbusCRC(array){
+ let crc = 0xFFFF;
+ for (let i = 0; i < array.length; i++){
+ let b = array[i];
+ crc = crc ^ b;
+
+ for (let j = 8; j>0; j--){
+ if ((crc & 0x0001) != 0){
+ crc = crc >> 1;
+ crc = crc ^ 0xA001;
+ } else {
+ crc = crc >> 1;
+ }
+ }
+ }
+
+ array.push(crc & 0xFF);
+ array.push((crc >> 8) & 0xFF);
+
+ return array;
+ }
+
+
+ instance.on('data', function(flowdata) {
+
+ console.log("flowdata on data", flowdata);
+ sendStatus({"CONFIG": instance.CONFIG, "PRIVATEVARS":PRIVATEVARS})
+
+ });
+
+ instance.reconfigure = function() {
+
+ //TODO remove ftom options
+ myEdge = instance.options.edge;
+ if (starter === null){
+ starter = setInterval(function(){
+ loadData();
+ clearInterval(starter);
+ starter = null;
+ }, 5000);
+ }
+ };
+
+ instance.close = function() {
+ // close sockets and such
+ };
+
+ function sendError(func, device, weight, str, extra){
+ if ((weight === ERRWEIGHT.DEBUG) && (instance.CONFIG.debug === false)){
+ return; // Allow debug messages only if CONFIG.debug is active
+ }
+
+ let content = {
+ "type": weight,
+ "status": "new",
+ "source": {
+ "func":func,
+ "component":instance.id,
+ "component_name":instance.name,
+ "edge":myEdge
+ },
+ "message":str,
+ "message_data": extra
+ };
+ let msg = {};
+ msg[device] = [
+ {
+ "ts": Date.now(),
+ "values": {
+ "_event":content
+ }
+ }
+ ];
+
+ // Msg can be outputted from components only after configuration
+ /*if (canSendErrData()){
+ sendBufferedErrors();
+ } else {
+ bufferError(msg);
+ }*/
+ instance.send(instanceSendTo.tb, msg); // Even if error server is unavailable, send this message to output, for other possible component connections
+
+
+
+ function sendBufferedErrors(){
+ if (PRIVATEVARS.errBuffer === undefined){
+ console.log("errBuffer undefined");
+ console.log("private: ", PRIVATEVARS);
+ }
+ console.log("errBuffer size: ", PRIVATEVARS.errBuffer.length);
+ if (PRIVATEVARS.errBuffer.length > 0){
+ for (let i = 0; i < PRIVATEVARS.errBuffer.length; i++){
+ instance.send(instanceSendTo.error, PRIVATEVARS.errBuffer[i]);
+ }
+ PRIVATEVARS.errBuffer = []; //Clear the buffer
+ saveData();
+ }
+ }
+
+ function bufferError(msg){
+ PRIVATEVARS.errBuffer.push(msg);
+ saveData();
+ }
+ }
+
+ function canSendErrData(){
+ //if (FLOW.errServerAvailable)
+ return true;
+ //else
+ // return false;
+ }
+
+ function sendStatus(str){
+ instance.send(instanceSendTo.debug, str);
+ }
+
+ function sendThingsBoard(obj){
+ // Msg can be outputted from components only after configuration
+ /*if (canSendTbData()){
+ sendBufferedTB();
+ } else {
+ console.log("cant send data");
+ bufferTB(str);
+ }*/
+ //console.log("send thingsboard", str);
+
+ //console.log("FLOW.OMS_edgeName", FLOW.OMS_edgeName, obj);
+
+ if(obj.hasOwnProperty(FLOW.OMS_edgeName) && FLOW.OMS_edgeName != "")
+ {
+ //send it to di_do_controller
+ instance.send(instanceSendTo.di_do_controller, {sender: "modbus_citysys", tbdata: obj});
+ }
+ // else
+ {
+ instance.send(instanceSendTo.tb, obj); // Even if TB server is unavailable, send this message to output, for other possible component connections
+ }
+
+ //instance.send(2, str); // Even if TB server is unavailable, send this message to output, for other possible component connections
+
+
+ function sendBufferedTB(){
+ if (PRIVATEVARS.tbBuffer.length > 0){
+ console.log("sending buffered: ", PRIVATEVARS.tbBuffer.length );
+ for (let i = 0; i < PRIVATEVARS.tbBuffer.length; i++){
+ instance.send(instanceSendTo.tb, PRIVATEVARS.tbBuffer[i]);
+ }
+ PRIVATEVARS.tbBuffer = []; //Clear the buffer
+ saveData();
+ }
+ }
+
+ function bufferTB(str){
+ PRIVATEVARS.tbBuffer.push(str);
+ saveData();
+ }
+ }
+
+ function canSendTbData(){
+ //if (FLOW.tbAvailable)
+ return true;
+ //else
+ // return false;
+ }
+
+ function startTimeoutInterval(){
+ if (!timeoutInterval){
+ timeoutInterval = setInterval(readingTimeouted, instance.CONFIG.timeoutTime);
+ }
+ }
+
+ function stopTimeoutInterval(){
+ if (timeoutInterval){
+ clearInterval(timeoutInterval);
+ timeoutInterval = null;
+ }
+ }
+
+ function startCmdWaitInterval(){
+ if (!msgWaitInterval){
+ msgWaitInterval = setInterval(readDeviceData, instance.CONFIG.msgWaitTime);
+ }
+ }
+
+ function stopCmdWaitInterval(){
+ if (msgWaitInterval){
+ clearInterval(msgWaitInterval);
+ msgWaitInterval = null;
+ }
+ }
+
+ instance.on('options', instance.reconfigure);
+ instance.reconfigure();
+
+ // LAST SECTION FOR COMMON FUNCTIONS
+ function humanReadableTimeAndDate(){
+ let date_ob = new Date();
+
+ let date = ("0" + date_ob.getDate()).slice(-2);
+ let month = ("0" + (date_ob.getMonth() + 1)).slice(-2);
+ let year = date_ob.getFullYear();
+
+ let hours = ("0" + date_ob.getHours()).slice(-2);
+ let minutes = ("0" + date_ob.getMinutes()).slice(-2);
+ let seconds = ("0" + date_ob.getSeconds()).slice(-2);
+
+ return date+"."+month+"."+year+" "+hours+":"+minutes+":"+seconds;
+ }
+
+ if (starter === null){
+ starter = setInterval(function(){
+ loadData();
+ clearInterval(starter);
+ starter = null;
+ }, 5000);
+ }
+
+ //setTimeout(loadData, 5000);
+};
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
diff --git a/flow/modbus_reader.js b/flow/modbus_reader.js
deleted file mode 100644
index f34ec9a..0000000
--- a/flow/modbus_reader.js
+++ /dev/null
@@ -1,348 +0,0 @@
-exports.id = 'modbus_reader';
-exports.title = 'Modbus reader';
-exports.version = '2.0.0';
-exports.group = 'Worksys';
-exports.color = '#2134B0';
-exports.output = ["red", "white", "yellow"];
-exports.click = false;
-exports.author = 'Rastislav Kovac';
-exports.icon = 'bolt';
-exports.readme = `
- Modbus requests to modbus devices (electromer, twilight sensor, thermometer.
- Component keeps running arround deviceConfig array in "timeoutInterval" intervals. Array items are objects with single modbus devices.
- Everything is sent to dido_controller. If requests to device fail (all registers must fail to send NOK status) , we send "NOK-'device'" status to dido_controller.
- This device needs to be configured in dido_controller!!! Double check if it is. In dido_controller we calculate final status and all values with status are pushed to tb.
-`;
-
-const modbus = require('jsmodbus');
-const SerialPort = require('serialport');
-const { timeoutInterval, deviceConfig } = require("../databases/modbus_config");
-const { sendNotification } = require('./helper/notification_reporter');
-
-const DELAY_BETWEEN_DEVICES = 10000;
-
-const SEND_TO = {
- debug: 0,
- dido_controller: 1,
- tb: 2
-};
-
-//to handle NOK and OK sendNotifications s
-const numberOfNotResponding = {};
-let tbName = null;
-let mainSocket;
-
-
-exports.install = function(instance) {
-
- class SocketWithClients {
-
- constructor () {
- this.stream = null;
- this.socket = null;
- this.clients = {};
- this.allValues = {};
- this.errors = 0;
- this.index = 0;
- this.timeoutInterval = 5000;
-
- // we need to go always around for all devices. So we need index value, device address, as well as number of registers for single device
- this.deviceAddress = null; // device address (1 - EM340 and 2 for twilight_sensor)
- this.indexInDeviceConfig = 0; // first item in deviceConfig
- this.lengthOfActualDeviceStream = null;
- this.device = null;
-
- // lampSwitchNotification helper variables
- this.onNotificationSent = false;
- this.offNotificationSent = false;
-
- this.startSocket();
- }
-
- startSocket = () => {
-
- let obj = this;
-
- this.socket = new SerialPort("/dev/ttymxc0", {
- baudRate: 9600,
- })
-
- // we create a client for every deviceAddress ( = address) in list and push them into dictionary
- for( let i = 0; i < deviceConfig.length; i++)
- {
- this.clients[deviceConfig[i].deviceAddress] = new modbus.client.RTU(this.socket, deviceConfig[i].deviceAddress, 2000); // 2000 is timeout in register request, default is 5000, which is too long
- }
-
- this.socket.on('error', function(e) {
- console.log('socket connection error', e);
- if(e.code == 'ECONNREFUSED' || e.code == 'ECONNRESET') {
- console.log(exports.title + ' Waiting 10 seconds before trying to connect again');
- setTimeout(obj.startSocket, 10000);
- }
- });
-
- this.socket.on('close', function() {
- console.log('Socket connection closed ' + exports.title + ' Waiting 10 seconds before trying to connect again');
- setTimeout(obj.startSocket, 10000);
- });
-
- this.socket.on('open', function () {
- console.log("socket connected");
- obj.getActualStreamAndDevice();
- obj.timeoutInterval = timeoutInterval - DELAY_BETWEEN_DEVICES; // to make sure readout always runs in timeoutinterval we substract DELAY_BETWEEN_DEVICES
- })
-
- };
-
- getActualStreamAndDevice = () => {
- const dev = deviceConfig[this.indexInDeviceConfig];
- this.index = 0;
- this.errors = 0;
- this.stream = dev.stream;
- this.lengthOfActualDeviceStream = dev.stream.length;
- this.deviceAddress = dev.deviceAddress; // 1 or 2 or any number
- this.device = dev.device; //em340, twilight_sensor
-
- if(this.indexInDeviceConfig == 0) setTimeout(this.readRegisters, this.timeoutInterval);
- else setTimeout(this.readRegisters, DELAY_BETWEEN_DEVICES);
- }
-
- readRegisters = () => {
-
- const str = this.stream[this.index];
- const register = str.register;
- const size = str.size;
- const tbAttribute = str.tbAttribute;
-
- let obj = this;
-
- this.clients[this.deviceAddress].readHoldingRegisters(register, size)
- .then( function (resp) {
-
- resp = resp.response._body.valuesAsArray; //resp is array of length 1 or 2, f.e. [2360,0]
- // console.log(deviceAddress, register, tbAttribute, resp);
-
- //device is responding again after NOK status
- if(numberOfNotResponding.hasOwnProperty(obj.device))
- {
- let message = "";
- if(obj.device == "em340")
- {
- message = "electrometer_ok";
- }
- else if(obj.device == "twilight_sensor")
- {
- message = "twilight_sensor_ok";
- }
- message && sendNotification("modbus_reader: readRegisters", tbName, message, {}, "", SEND_TO.tb, instance);
- delete numberOfNotResponding[obj.device];
- }
-
- obj.transformResponse(resp, register);
-
- //obj.errors = 0;
- obj.index++;
- obj.readAnotherRegister();
-
- }).catch (function () {
-
- console.log("errors pri citani modbus registra", register, obj.indexInDeviceConfig, tbName, tbAttribute);
-
- obj.errors++;
- if(obj.errors == obj.lengthOfActualDeviceStream)
- {
- instance.send(SEND_TO.dido_controller, {status: "NOK-" + obj.device}); // NOK-em340, NOK-em111, NOK-twilight_sensor, NOK-thermometer
-
- //todo - neposlalo notification, ked sme vypojili twilight a neposle to do tb, ale do dido ??
- if(!numberOfNotResponding.hasOwnProperty(obj.device))
- {
- let message = "";
- if(obj.device == "twilight_sensor")
- {
- message = "twilight_sensor_nok";
- }
- else if(obj.device == "em340")
- {
- message = "electrometer_nok";
- }
- message && sendNotification("modbus_reader: readingTimeouted", tbName, message, {}, "", SEND_TO.tb, instance);
- numberOfNotResponding[obj.device] = 1;
- }
-
- obj.errors = 0;
- numberOfNotResponding[obj.device] += 1;
- }
-
- console.error(require('util').inspect(arguments, {
- depth: null
- }))
-
- // if reading out of device's last register returns error, we send accumulated allValues to dido_controller (if allValues are not an empty object)
- if(obj.index + 1 >= obj.lengthOfActualDeviceStream)
- {
- if(!isObjectEmpty(obj.allValues)) instance.send(SEND_TO.dido_controller, {values: obj.allValues});
- obj.allValues = {};
- }
- obj.index++;
- obj.readAnotherRegister();
- })
-
- };
-
- readAnotherRegister = () => {
- if(this.index < this.lengthOfActualDeviceStream) setTimeout(this.readRegisters, 0);
- else this.setNewStream();
- }
-
- transformResponse = (response, register) => {
-
- for (let i = 0; i < this.lengthOfActualDeviceStream; i++) {
-
- let a = this.stream[i];
- if (a.register === register)
- {
- let tbAttribute = a.tbAttribute;
- let multiplier = a.multiplier;
-
- let value = this.calculateValue(response, multiplier);
- // console.log(register, tbName, tbAttribute, response, a.multiplier, value);
-
- // if(tbName == undefined) return;
-
- if(this.index + 1 < this.lengthOfActualDeviceStream)
- {
- this.allValues[tbAttribute] = value;
- return;
- }
-
- const values = {
- ...this.allValues,
- [tbAttribute]: value,
- };
-
- this.checkNullVoltage(values);
- this.lampSwitchNotification(values);
-
- instance.send(SEND_TO.dido_controller, {values: values});
-
- this.allValues = {};
- break;
- }
-
- }
-
- }
-
- setNewStream = () =>
- {
- if(this.lengthOfActualDeviceStream == this.index)
- {
- if(this.indexInDeviceConfig + 1 == deviceConfig.length)
- {
- this.indexInDeviceConfig = 0;
- }
- else
- {
- this.indexInDeviceConfig += 1;
- }
-
- this.getActualStreamAndDevice();
- }
- }
-
- calculateValue = (response, multiplier) =>
- {
- let value = 0;
-
- let l = response.length;
- if (l === 2)
- {
- value = (response[1]*(2**16) + response[0]);
-
- if(value >= (2**31)) // ak je MSB bit nastavený, eventuálne sa dá použiť aj (value & 0x80000000), ak vieš robiť logický súčin
- {
- value = value - "0xFFFFFFFF" + 1;
- }
- }
- else if (l === 1)
- {
- value = response[0];
-
- if(value >= (2**15)) // ak je MSB bit nastavený, eventuálne sa dá použiť aj (value & 0x8000), ak vieš robiť logický súčin
- {
- value = value - "0xFFFF" + 1;
- }
- }
-
- return Math.round(value * multiplier * 10) / 10;
- }
-
- checkNullVoltage = (values) => {
-
- if(!(values.hasOwnProperty("Phase_1_voltage") || values.hasOwnProperty("Phase_2_voltage") || values.hasOwnProperty("Phase_3_voltage"))) return;
-
- Object.keys(values).map(singleValue => {
- if (["Phase_1_voltage", "Phase_2_voltage", "Phase_3_voltage"].includes(singleValue))
- {
- let l = singleValue.split("_");
- let phase = parseInt(l[1]);
-
- if(FLOW.OMS_no_voltage == undefined) FLOW.OMS_no_voltage = new Set();
- // console.log(values[singleValue], tbName);
-
- if(values[singleValue] == 0)
- {
- FLOW.OMS_no_voltage.add(phase);
- sendNotification("modbus_reader: checkNullVoltage", tbName, "no_voltage_on_phase", {phase: phase}, "", SEND_TO.tb, instance, "voltage" + phase );
- // console.log('no voltage')
- }
- else
- {
- FLOW.OMS_no_voltage.delete(phase);
- // console.log('voltage detected')
- sendNotification("modbus_reader: checkNullVoltage", tbName, "voltage_on_phase_restored", {phase: phase}, "", SEND_TO.tb, instance, "voltage" + phase);
- }
- }
- })
- }
-
- /**
- * function sends notification to slack and to tb, if EM total_power value changes more than 500. This should show, that RVO lamps has been switched on or off
- */
- lampSwitchNotification = (values) => {
-
- if(!values.hasOwnProperty("total_power")) return;
-
- const actualTotalPower = values.total_power;
- if(actualTotalPower > 600 && this.onNotificationSent == false)
- {
- sendNotification("modbus_reader: lampSwitchNotification", tbName, "lamps_have_turned_on", {}, "", SEND_TO.tb, instance);
- this.onNotificationSent = true;
- this.offNotificationSent = false;
- }
- else if(actualTotalPower <= 600 && this.offNotificationSent == false)
- {
- sendNotification("modbus_reader: lampSwitchNotification", tbName, "lamps_have_turned_off", {}, "", SEND_TO.tb, instance);
- this.onNotificationSent = false;
- this.offNotificationSent = true;
- }
- }
-
- }
-
- const isObjectEmpty = (objectName) => {
- return Object.keys(objectName).length === 0 && objectName.constructor === Object;
- }
-
- setTimeout(() => {
-
- mainSocket = new SocketWithClients();
- tbName = FLOW.OMS_rvo_tbname;
-
- // this notification is to show, that flow (unipi) has been restarted
- sendNotification("modbus_reader", tbName, "flow_restart", {}, "", SEND_TO.slack, instance);
-
- }, 25000);
-
-}
-
diff --git a/flow/monitorcpu.js b/flow/monitorcpu.js
new file mode 100644
index 0000000..0e906d1
--- /dev/null
+++ b/flow/monitorcpu.js
@@ -0,0 +1,107 @@
+exports.id = 'monitorcpu';
+exports.title = 'CPU';
+exports.version = '1.0.0';
+exports.author = 'Peter Širka';
+exports.group = 'Monitoring';
+exports.color = '#F6BB42';
+exports.output = 1;
+exports.icon = 'microchip';
+exports.options = { enabled: true };
+exports.click = true;
+exports.readme = `# CPU monitoring
+
+This component monitors CPU \`% percentage\` consumption in Linux systems. It uses \`mpstat\` command.
+
+__Data Example__:
+
+\`\`\`javascript
+{
+ cpu: 30, // percentage
+ cores: [4, 60, 0], // percentage
+ count: 3 // count of cores
+}
+\`\`\``;
+
+exports.html = `
+
+
+
@(Interval in milliseconds)
+
+
+
`;
+
+exports.install = function(instance) {
+
+ var fields = ['CPU', '%idle'];
+ var current = { cores: [], cpu: 0, count: 0 };
+ var proc = null;
+ var tproc = null;
+
+ instance.custom.kill = function() {
+ if (proc) {
+ proc.kill('SIGKILL');
+ proc = null;
+ }
+ };
+
+ instance.custom.run = function() {
+
+ if (tproc) {
+ clearTimeout(tproc);
+ tproc = null;
+ }
+
+ instance.custom.kill();
+ proc = require('child_process').spawn('mpstat', ['-P', 'ALL', 10]);
+ proc.stdout.on('data', U.streamer('\n\n', instance.custom.process));
+ proc.stdout.on('error', function(e) {
+ instance.error(e);
+ instance.custom.kill();
+ tproc = setTimeout(instance.custom.run, instance.options.interval || 5000);
+ });
+ };
+
+ instance.custom.process = function(chunk) {
+ current.cpu = 0;
+ chunk.toString('utf8').parseTerminal(fields, instance.custom.parse);
+ current.count = current.cores.length;
+ if (current.count) {
+ instance.send2(current);
+ instance.custom.status();
+ }
+ };
+
+ instance.custom.parse = function(values) {
+ var val = 100 - values[1].parseFloat2();
+ if (values[0] === 'all')
+ current.cpu = val;
+ else
+ current.cores[+values[0]] = val;
+ };
+
+ instance.custom.status = function() {
+ if (instance.options.enabled)
+ instance.status(current.cpu.floor(1) + '%');
+ else
+ instance.status('Disabled', 'red');
+ };
+
+ instance.on('click', function() {
+ instance.options.enabled = !instance.options.enabled;
+ instance.custom.status();
+ if (instance.options.enabled)
+ instance.custom.run();
+ else
+ instance.custom.kill();
+ });
+
+ instance.on('close', function() {
+ instance.custom.kill();
+ if (tproc) {
+ clearTimeout(tproc);
+ tproc = null;
+ }
+ });
+
+ setTimeout(instance.custom.run, 1000);
+};
\ No newline at end of file
diff --git a/flow/mqtt.js b/flow/mqtt.js
new file mode 100644
index 0000000..e34c940
--- /dev/null
+++ b/flow/mqtt.js
@@ -0,0 +1,441 @@
+exports.id = 'mqtt';
+exports.title = 'MQTT broker';
+exports.group = 'MQTT';
+exports.color = '#888600';
+exports.version = '1.0.1';
+exports.icon = 'exchange';
+exports.input = true;
+exports.output = 0;
+exports.author = 'Martin Smola';
+exports.variables = true;
+exports.options = { host: '127.0.0.1', port: 1883 };
+exports.traffic = false;
+exports.npm = ['mqtt'];
+
+exports.html = `
+
+
+
Hostname or IP address
+
+
+
+
+
+
@(Client id)
+
@(Supports variables, example: \`client_{device-id}\`)
+
@(Secure (ssl))
+
+
+
+
+
+
@(Require Authorization)
+
+
+
+
+
+
+
+
@(Last will topic)
+
@(Supports variables, example: \`lwt/{device-id}\`)
+
+
+
@(Last will message)
+
@(Supports variables, example: \`{device-id} is offline\`)
+
+
+
+`;
+
+exports.readme = `
+# MQTT Broker
+
+## Input
+Allows to change connection programaticaly
+\`\`\`javascipt
+{
+ host: '1.2.3.4',
+ port: '',
+ secure: true/false,
+ username: 'john',
+ password: 'X',
+ lwttopic: '',
+ lwtmessage: '',
+ clientid: ''
+}
+\`\`\`
+`;
+
+var MQTT_BROKERS = [];
+var mqtt;
+
+global.MQTT = {};
+
+exports.install = function(instance) {
+
+ var broker;
+
+ mqtt = require('mqtt');
+
+ instance.on('data', function(flowdata){
+ var data= flowdata.data;
+ var options = instance.options;
+
+ if (data.host && data.port)
+ return instance.custom.reconfigure(data, options);
+
+ if (data.close === true)
+ instance.close(NOOP);
+ });
+
+ instance.custom.reconfigure = function(o, old_options) {
+
+ if (old_options)
+ MQTT_BROKERS = MQTT_BROKERS.remove(function(b){
+ return b.id === old_options.id;
+ });
+
+ var options = instance.options;
+
+ if (!options.host || !options.port) {
+ instance.status('Not configured', 'red');
+ return;
+ }
+
+ options.id = (options.username || '') + '@' + options.host + ':' + options.port;
+
+ if (broker) {
+ broker.close();
+ EMIT('mqtt.brokers.status', 'reconfigured', old_options.id, options.id);
+ }
+
+ instance.custom.createBroker();
+ };
+
+ instance.custom.createBroker = function() {
+
+ ON('mqtt.brokers.status', brokerstatus);
+
+ var o = instance.options;
+ var opts = {
+ host: o.host,
+ port: o.port,
+ id: o.id,
+ secure: o.secure,
+ rejectUnauthorized: false,
+ reconnectPeriod: 3000,
+ resubscribe: false
+ };
+
+ if (o.auth) {
+ opts.username = o.username;
+ opts.password = o.password;
+ }
+
+ if (o.lwt) {
+ opts.will = {
+ topic: instance.arg(o.lwttopic),
+ payload: instance.arg(o.lwtmessage)
+ }
+ }
+
+ if (o.clientid)
+ opts.clientId = instance.arg(o.clientid);
+
+ broker = new Broker(opts);
+ MQTT_BROKERS.push(broker);
+
+ instance.status('Ready');
+ };
+
+ instance.close = function(done) {
+
+ broker && broker.close(function() {
+ MQTT_BROKERS = MQTT_BROKERS.remove('id', instance.options.id);
+ EMIT('mqtt.brokers.status', 'removed', instance.options.id);
+ });
+
+ OFF('mqtt.brokers.status', brokerstatus);
+
+ done();
+ };
+
+ function brokerstatus(status, brokerid, err) {
+ if (brokerid !== instance.options.id)
+ return;
+
+ switch (status) {
+ case 'connecting':
+ instance.status('Connecting', '#a6c3ff');
+ break;
+ case 'connected':
+ instance.status('Connected', 'green');
+ break;
+ case 'disconnected':
+ instance.status('Disconnected', 'red');
+ break;
+ case 'connectionfailed':
+ instance.status('Connection failed', 'red');
+ break;
+ case 'error':
+ instance.error('MQTT Error, ID: ' + instance.id + '\n ' + err);
+ break;
+ }
+ }
+
+ instance.on('options', instance.custom.reconfigure);
+ instance.custom.reconfigure();
+};
+
+FLOW.trigger('mqtt.brokers', function(next) {
+ var brokers = [''];
+ MQTT_BROKERS.forEach(n => brokers.push(n.id));
+ next(brokers);
+});
+
+MQTT.add = function(brokerid, componentid) {
+ var broker = MQTT_BROKERS.findItem('id', brokerid);
+
+ if (broker)
+ broker.add(componentid);
+};
+
+MQTT.remove = function(brokerid, componentid) {
+ var broker = MQTT_BROKERS.findItem('id', brokerid);
+ broker && broker.remove(componentid);
+};
+
+MQTT.publish = function(brokerid, topic, data, options) {
+ var broker = MQTT_BROKERS.findItem('id', brokerid);
+ if (broker)
+ broker.publish(topic, data, options);
+ else
+ EMIT('mqtt.brokers.status', 'error', brokerid, 'No such broker');
+};
+
+MQTT.subscribe = function(brokerid, componentid, topic, qos) {
+ var broker = MQTT_BROKERS.findItem('id', brokerid);
+
+ if (!broker)
+ return;
+
+ broker.add(componentid);
+ broker.subscribe(componentid, topic, qos);
+};
+
+MQTT.unsubscribe = function(brokerid, componentid, topic, qos) {
+ var broker = MQTT_BROKERS.findItem('id', brokerid);
+ if (!broker)
+ return;
+
+ broker.unsubscribe(componentid, topic);
+ broker.remove(componentid);
+};
+
+MQTT.broker = function(brokerid) {
+ return MQTT_BROKERS.findItem('id', brokerid);
+};
+
+/*
+
+ https://github.com/mqttjs/MQTT.js/blob/master/examples/client/secure-client.js
+
+*/
+
+/*
+ TODO
+
+ - add `birth` and `last will and testament` messages
+ - add options to self.client.connect(broker [,options]); - credentials, certificate etc.
+
+
+*/
+
+function Broker(options) {
+ var self = this;
+
+ if (!options.host || !options.port)
+ return false;
+
+ self.connecting = false;
+ self.connected = false;
+ self.closing = false;
+ self.components = [];
+ self.subscribtions = {};
+ self.id = options.id;
+ self.options = options;
+ setTimeout(function() {
+ EMIT('mqtt.brokers.status', 'new', self.id);
+ }, 500);
+ return self;
+}
+
+Broker.prototype.connect = function() {
+
+ var self = this;
+ if (self.connected || self.connecting)
+ return EMIT('mqtt.brokers.status', self.connected ? 'connected' : 'connecting', self.id);
+
+ self.connecting = true;
+ var broker = self.options.secure ? 'mqtts://' : 'mqtt://' + self.options.host + ':' + self.options.port;
+
+ EMIT('mqtt.brokers.status', 'connecting', self.id);
+
+ self.client = mqtt.connect(broker, self.options);
+
+ self.client.on('connect', function() {
+ self.connecting = false;
+ self.connected = true;
+ if (self.reconnecting) {
+ EMIT('mqtt.brokers.status', 'reconnected', self.id);
+ self.reconnecting = false;
+ self.resubscribe();
+ }
+ EMIT('mqtt.brokers.status', 'connected', self.id);
+ });
+
+ self.client.on('reconnect', function() {
+ self.connecting = true;
+ self.connected = false;
+ self.reconnecting = true;
+ EMIT('mqtt.brokers.status', 'connecting', self.id);
+ });
+
+ self.client.on('message', function(topic, message) {
+ message = message.toString();
+ if (message[0] === '{') {
+ TRY(function() {
+ message = JSON.parse(message);
+ }, () => FLOW.debug('MQTT: Error parsing data', message));
+ }
+ EMIT('mqtt.brokers.message', self.id, topic, message);
+ });
+
+ self.client.on('close', function(err) {
+ if (err && err.toString().indexOf('Error')) {
+ self.connecting = false;
+ self.connected = false;
+ EMIT('mqtt.brokers.status', 'error', self.id, err.code);
+ }
+
+ if (self.connected || !self.connecting) {
+ self.connected = false;
+ EMIT('mqtt.brokers.status', 'disconnected', self.id);
+ } else if (self.connecting) {
+ self.connecting = false;
+ EMIT('mqtt.brokers.status', 'connectionfailed', self.id);
+ }
+ });
+
+ self.client.on('error', function(err) {
+
+ if (self.connecting) {
+ self.client.end();
+ self.connecting = false;
+ EMIT('mqtt.brokers.status', 'error', self.id, err);
+ }
+ });
+
+};
+
+Broker.prototype.disconnect = function(reconnect) {
+ var self = this;
+ if (!self.closing)
+ self.close(function(){
+ reconnect && self.connect();
+ });
+};
+
+Broker.prototype.close = function(callback) {
+ var self = this;
+ self.closing = true;
+
+ if ((self.connected || self.connecting) && self.client && self.client.end)
+ self.client.end(true, cb);
+ else
+ cb();
+
+ function cb() {
+ EMIT('mqtt.brokers.status', 'disconnected', self.id);
+ self.client && self.client.removeAllListeners();
+ self.components = [];
+ self.client = null;
+ callback && callback();
+ }
+};
+
+Broker.prototype.subscribe = function(componentid, topic) {
+ var self = this;
+ self.subscribtions[topic] = self.subscribtions[topic] || [];
+ if (self.subscribtions[topic].indexOf(componentid) > -1)
+ return;
+ self.client.subscribe(topic);
+ self.subscribtions[topic].push(componentid);
+};
+
+Broker.prototype.resubscribe = function() {
+ var self = this;
+ var topics = Object.keys(self.subscribtions);
+ for (var i = 0; i < topics.length; i++)
+ self.client.subscribe(topics[i]);
+};
+
+Broker.prototype.unsubscribe = function(componentid, topic) {
+ var self = this;
+ var sub = self.subscribtions[topic];
+ if (sub) {
+ self.subscribtions[topic] = sub.remove(componentid);
+ self.client.connected && !self.subscribtions[topic].length && self.client.unsubscribe(topic);
+ }
+};
+
+Broker.prototype.publish = function(topic, data, options) {
+ var self = this;
+ if (!self.connected)
+ return;
+
+ if (typeof(data) === 'object') {
+ options.qos = parseInt(data.qos || options.qos);
+ options.retain = data.retain || options.retain;
+ topic = data.topic || topic;
+ data.payload && (data = typeof(data.payload) === 'string' ? data.payload : JSON.stringify(data.payload));
+ }
+
+ if (options.qos !== 0 || options.qos !== 1 || options.qos !== 2)
+ options.qos = null;
+
+ if (typeof(data) !== 'string')
+ data = JSON.stringify(data);
+
+ self.client.publish(topic, data || '', options);
+};
+
+Broker.prototype.add = function(componentid) {
+ var self = this;
+ self.components.indexOf(componentid) === -1 && self.components.push(componentid);
+ self.connect();
+};
+
+Broker.prototype.remove = function(componentid) {
+ var self = this;
+ self.components = self.components.remove(componentid);
+ !self.components.length && self.disconnect();
+};
diff --git a/flow/mqtt_listener.js b/flow/mqtt_listener.js
new file mode 100644
index 0000000..b9d778d
--- /dev/null
+++ b/flow/mqtt_listener.js
@@ -0,0 +1,171 @@
+exports.id = 'mqttlistener';
+exports.title = 'MQTT listener';
+exports.group = 'MQTT';
+exports.color = '#888600';
+exports.version = '1.0.0';
+exports.icon = 'sign-out';
+exports.input = 1;
+exports.output = ["red", "white"];
+exports.author = 'Rastislav Kovac';
+exports.options = { host: "", port: 1883, clientid: "", username: "" };
+//exports.npm = ['mqtt', 'streamroller'];
+
+
+exports.html = `
+
+
+
Hostname or IP address
+
+
+
+
+
`;
+
+
+exports.readme = `
+# MQTT processor
+
+Version 1.0.0
+
+It serves as a client and subscribes to 'rpc' topic. It receives rpc calls from kovobel-prod01, which receives it from platform.
+
+`;
+
+const instanceSendTo = {
+ debug: 0,
+ rpcCall: 1,
+}
+
+
+exports.install = function(instance) {
+
+
+ const mqtt = require("mqtt");
+
+ instance.on('options', loadNodes);
+
+
+ function loadNodes()
+ {
+
+ if(instance.options.host == "")
+ {
+ instance.status('No configuration', 'red');
+ }
+ else
+ {
+ var o = instance.options;
+ opts = {
+ host: o.host,
+ port: o.port,
+ clientId: o.clientid,
+ rejectUnauthorized: false,
+ resubscribe: true
+ };
+
+ connectToServer();
+
+ }
+
+ }
+
+ function connectToServer()
+ {
+ var url = "mqtt://" + opts.host + ":" + opts.port;
+ console.log("MQTT URL: ", url);
+
+ client = mqtt.connect(url, opts);
+
+ client.on('connect', function() {
+ instance.status("Connected", "green");
+ client.subscribe('rpc', function (err) {
+ if (!err) {
+ client.publish('rpc', 'Hello mqtt');
+ console.log('message published');
+ }
+ });
+ });
+
+ client.on('reconnect', function() {
+ instance.status("Reconnecting", "yellow");
+ });
+
+ client.on('message', function(topic, message) {
+ // message is type of buffer
+ message = message.toString();
+
+ // instance.send(1, message);
+ // return;
+
+ if (message[0] === '{') {
+
+ try {
+ message = JSON.parse(message);
+
+ if (Object.keys(message).length < 2 || !Object.keys(message).includes("message")) return;
+
+ message.message = JSON.parse(message.message);
+ instance.send(instanceSendTo.rpcCall, message);
+
+ } catch (e) {
+ instance.debug(`MQTT: Error parsing data, ${e}`);
+ }
+ }
+
+ });
+
+ client.on('close', function(err) {
+ if (err && err.toString().indexOf('Error')) {
+ instance.status("Err: "+err.code, "red");
+ instance.send(instanceSendTo.debug, {"message":"Broker CLOSE signal received !", "error":err, "opt":opts });
+ } else {
+ instance.status("Disconnected", "red");
+ instance.send(instanceSendTo.debug, {"message":"Broker CLOSE signal received !", "error":err, "opt":opts });
+ }
+ });
+
+ client.on('error', function(err) {
+ instance.status("Err: "+ err.code, "red");
+ instance.send(instanceSendTo.debug, {"message":"Broker ERROR signal received !", "error":err, "opt":opts });
+
+ });
+ }
+
+
+ //set opts accortding to options
+ instance.reconfigure = function() {
+
+ var o = instance.options;
+ opts = {
+ host: o.host,
+ port: o.port,
+ clientId: o.clientid,
+ rejectUnauthorized: false,
+ resubscribe: true
+ };
+
+ connectToServer();
+ };
+
+ instance.close = function(done) {
+ client.end();
+ };
+
+
+ loadNodes();
+
+ instance.on('options', instance.reconfigure);
+ instance.reconfigure();
+};
+
+
+
diff --git a/flow/mqttprocessor.js b/flow/mqttprocessor.js
new file mode 100644
index 0000000..1bf050c
--- /dev/null
+++ b/flow/mqttprocessor.js
@@ -0,0 +1,317 @@
+exports.id = 'mqttprocessor';
+exports.title = 'MQTT processor';
+exports.group = 'MQTT';
+exports.color = '#888600';
+exports.version = '1.0.0';
+exports.icon = 'sign-out';
+exports.input = 1;
+exports.output = ["red", "white", "blue"];
+exports.author = 'Rastislav Kovac';
+exports.options = { host: "", port: 1883, clientid: "", username: "" };
+//exports.npm = ['mqtt', 'streamroller'];
+
+
+exports.html = `
+
+
+
Hostname or IP address
+
+
+
+
+
`;
+
+
+exports.readme = `
+# MQTT processor
+
+Version 1.0.0
+
+It serves as a client, listens and subscribes to nodes in citysys configuration.
+Is able to send messages to flow (as rpc calls from platform)
+
+Added:
+- database collections,
+- rpc response
+`;
+
+const instanceSendTo = {
+ debug: 0,
+ rpcCall: 1,
+ cmdManager: 2
+}
+
+const { promisifyBuilder, makeMapFromDbResult } = require('./helper/db_helper.js');
+const mqtt = require('mqtt');
+
+
+//topics = {joqRYBVL30k9eQWOlZ5qwpD2KJpNEmA6gPxXzwaM: 'lamp_697/brightness'} --> to get topic from dataToTb tb name.
+const topics = {};
+//tbNames = {'lamp_697/brightness': joqRYBVL30k9eQWOlZ5qwpD2KJpNEmA6gPxXzwaM }
+const tbNames = {};
+
+
+exports.install = function(instance) {
+
+ let client = null;
+ let opts = null;
+
+ instance.on('options', loadNodes);
+
+
+ async function loadNodes()
+ {
+ const nodes = TABLE("nodes");
+ let nodesData = await promisifyBuilder(nodes.find());
+
+ nodesData.map(item => {
+ const value = 'lamp_' + item.node + '/brightness';
+ const key = item.tbname;
+ topics[key] = value;
+ tbNames[value] = key;
+ })
+
+ // console.log('******* -------- ', topics);
+ // console.log('******* -------- ', tbNames);
+
+ if(instance.options.host == "")
+ {
+ instance.status('No configuration', 'red');
+ }
+ else
+ {
+ var o = instance.options;
+ opts = {
+ host: o.host,
+ port: o.port,
+ clientId: o.clientid,
+ rejectUnauthorized: false,
+ resubscribe: true
+ };
+
+ connectToServer();
+
+ }
+
+ }
+
+ function connectToServer()
+ {
+ var url = "mqtt://" + opts.host + ":" + opts.port;
+ console.log("MQTT URL: ", url);
+
+ client = mqtt.connect(url, opts);
+
+ client.on('connect', function() {
+ instance.status("Connected", "green");
+ // array of subscribed topics ==> Object.values(topics)
+ client.subscribe(Object.values(topics), function (err) {
+ if (!err) {
+ client.publish('presence', 'Hello mqtt');
+ console.log('message published');
+ }
+ });
+ });
+
+ client.on('reconnect', function() {
+ instance.status("Reconnecting", "yellow");
+ });
+
+ client.on('message', function(topic, message) {
+ // message is type of buffer
+ message = message.toString();
+
+ if (message[0] === '{') {
+
+ try {
+ message = JSON.parse(message);
+
+ // we ensure to process messages only with 1 key ==> {'lamp_698/brightness': 20 } and if topic is in topics
+ if (Object.keys(message).length > 1 || !Object.values(topics).includes(topic)) return;
+
+ instance.send(0, {message:message, topic:topic})
+
+ const date = Date.now();
+ const tbName = tbNames[topic];
+
+
+ const transformToRpc = {
+ "topic": "v1/gateway/rpc",
+ "content": {
+ // 'device' is not needed here in 3rd party systems
+ "device": "KjbN4q7JPZmexgdnz2yKQ98YAWwO0Q3BMX6ERLoV",
+ "data": {
+ // 'id' is not needed here
+ "id": 5,
+ "method": "set_command",
+ "params": {
+ "entities": [
+ {
+ "entity_type": "street_luminaire",
+ "tb_name": tbName
+ }
+ ],
+ "command": "dimming",
+ "payload": {
+ "value": message[topic]
+ }
+ }
+ }
+ }
+ }
+
+
+ instance.send(instanceSendTo.cmdManager, transformToRpc);
+
+ } catch (e) {
+ instance.debug(`MQTT: Error parsing data, ${e}`);
+ }
+ }
+
+ });
+
+ client.on('close', function(err) {
+ if (err && err.toString().indexOf('Error')) {
+ instance.status("Err: "+err.code, "red");
+ instance.send(instanceSendTo.debug, {"message":"Broker CLOSE signal received !", "error":err, "opt":opts });
+ } else {
+ instance.status("Disconnected", "red");
+ instance.send(instanceSendTo.debug, {"message":"Broker CLOSE signal received !", "error":err, "opt":opts });
+ }
+ });
+
+ client.on('error', function(err) {
+ instance.status("Err: "+ err.code, "red");
+ instance.send(instanceSendTo.debug, {"message":"Broker ERROR signal received !", "error":err, "opt":opts });
+
+ });
+ }
+
+
+ //set opts accortding to options
+ instance.reconfigure = function() {
+
+ var o = instance.options;
+ opts = {
+ host: o.host,
+ port: o.port,
+ clientId: o.clientid,
+ rejectUnauthorized: false,
+ resubscribe: true
+ };
+
+ connectToServer();
+ };
+
+
+ instance.on('data', function(data) {
+ const value = data.data;
+ const nodeTbName = Object.keys(value)[0];
+
+ // tbnames of all lamps
+ const nodes = Object.values(tbNames);
+
+ if(nodes.includes(nodeTbName))
+ {
+
+ if(value[nodeTbName][0].values.hasOwnProperty('dimming'))
+ {
+ const key = nodeTbName;
+ const tbValues = value[key][0].values;
+
+ Object.keys(tbValues).map(item => {
+ if(item != 'dimming')
+ {
+ delete tbValues[item];
+ }
+ })
+
+ const topic = topics[key];
+ // console.log('---- ***********', key, topic)
+ // console.log('^^^^^^^^^^^^^^', value);
+
+ // transform data to send to 3rd party
+ const v = value[key][0];
+ v[topic] = v.values['dimming'];
+ delete v.values;
+
+ client.publish(topic, JSON.stringify(v));
+ }
+ else if(value[nodeTbName][0].values.hasOwnProperty('status'))
+ {
+ const key = nodeTbName;
+ const tbValues = value[key][0].values;
+
+ Object.keys(tbValues).map(item => {
+ if(item != 'status')
+ {
+ delete tbValues[item];
+ }
+ })
+
+ // we expect topic to be 'lamp_000/brightness'. we must make 'lamp_000/status'
+ const topic = topics[key].slice(0, 8) + '/status';
+ // console.log('---- ***********', key, topic)
+ // console.log('^^^^^^^^^^^^^^', value);
+
+ // transform data to send to 3rd party
+ const v = value[key][0];
+ v[topic] = v.values['status'];
+ delete v.values;
+
+ client.publish(topic, JSON.stringify(v));
+ }
+
+ }
+
+
+
+
+ });
+
+ instance.close = function(done) {
+ client.end();
+ };
+
+
+
+ loadNodes();
+
+ instance.on('options', instance.reconfigure);
+ instance.reconfigure();
+};
+
+
+
+
+/*
+ [ {
+ node: 683,
+ tbname: 'XKQbz3WAwY21dGa0R453rWyJm9PZOjqlvpr6Nkeo',
+ line: 3,
+ profile: '{"intervals":[{"cct":3000,"value":0,"end_time":"20:00","start_time":"13:00"},{"cct":3000,"value":100,"end_time":"21:20","start_time":"20:00"},{"cct":3000,"value":0,"end_time":"13:00","start_time":"05:30"},{"cct":3000,"value":80,"end_time":"22:20","start_time":"21:50"},{"cct":3000,"value":70,"end_time":"22:50","start_time":"22:20"},{"cct":3000,"value":60,"end_time":"23:20","start_time":"22:50"},{"cct":3000,"value":50,"end_time":"23:50","start_time":"23:20"},{"cct":3000,"value":40,"end_time":"00:20","start_time":"23:50"},{"cct":3000,"value":100,"end_time":"05:30","start_time":"03:20"},{"cct":3000,"value":30,"end_time":"00:50","start_time":"00:20"},{"cct":3000,"value":20,"end_time":"01:20","start_time":"00:50"},{"cct":3000,"value":90,"end_time":"21:50","start_time":"21:20"},{"cct":3000,"value":30,"end_time":"01:50","start_time":"01:20"},{"cct":3000,"value":40,"end_time":"02:20","start_time":"01:50"},{"cct":3000,"value":50,"end_time":"02:50","start_time":"02:20"},{"cct":3000,"value":60,"end_time":"03:20","start_time":"02:50"}],"astro_clock":true,"dawn_lux_sensor":false,"dusk_lux_sensor":false,"dawn_lux_sensor_value":5,"dusk_lux_sensor_value":5,"dawn_astro_clock_offset":0,"dusk_astro_clock_offset":0,"dawn_lux_sensor_time_window":30,"dusk_lux_sensor_time_window":30,"dawn_astro_clock_time_window":60,"dusk_astro_clock_time_window":60}',
+ processed: true,
+ status: true
+ },
+ {
+ node: 688,
+ tbname: 'PaGbQ3wBAZWOmRvK9VDpvz5endLJYopEqlkzNMxX',
+ line: 3,
+ profile: '{"intervals":[{"cct":3000,"value":0,"end_time":"20:00","start_time":"13:00"},{"cct":3000,"value":100,"end_time":"21:20","start_time":"20:00"},{"cct":3000,"value":0,"end_time":"13:00","start_time":"05:30"},{"cct":3000,"value":80,"end_time":"22:20","start_time":"21:50"},{"cct":3000,"value":70,"end_time":"22:50","start_time":"22:20"},{"cct":3000,"value":60,"end_time":"23:20","start_time":"22:50"},{"cct":3000,"value":50,"end_time":"23:50","start_time":"23:20"},{"cct":3000,"value":40,"end_time":"00:20","start_time":"23:50"},{"cct":3000,"value":100,"end_time":"05:30","start_time":"03:20"},{"cct":3000,"value":30,"end_time":"00:50","start_time":"00:20"},{"cct":3000,"value":20,"end_time":"01:20","start_time":"00:50"},{"cct":3000,"value":90,"end_time":"21:50","start_time":"21:20"},{"cct":3000,"value":30,"end_time":"01:50","start_time":"01:20"},{"cct":3000,"value":40,"end_time":"02:20","start_time":"01:50"},{"cct":3000,"value":50,"end_time":"02:50","start_time":"02:20"},{"cct":3000,"value":60,"end_time":"03:20","start_time":"02:50"}],"astro_clock":true,"dawn_lux_sensor":false,"dusk_lux_sensor":false,"dawn_lux_sensor_value":5,"dusk_lux_sensor_value":5,"dawn_astro_clock_offset":0,"dusk_astro_clock_offset":0,"dawn_lux_sensor_time_window":30,"dusk_lux_sensor_time_window":30,"dawn_astro_clock_time_window":60,"dusk_astro_clock_time_window":60}',
+ processed: true,
+ status: true
+ }
+]
+*/
+
+
+
diff --git a/flow/mqttpublish.js b/flow/mqttpublish.js
new file mode 100644
index 0000000..3f6aef0
--- /dev/null
+++ b/flow/mqttpublish.js
@@ -0,0 +1,134 @@
+exports.id = 'mqttpublish';
+exports.title = 'MQTT publish';
+exports.group = 'MQTT';
+exports.color = '#888600';
+exports.version = '1.1.0';
+exports.icon = 'sign-out';
+exports.input = true;
+exports.output = 1;
+exports.author = 'Martin Smola';
+exports.options = {};
+
+exports.html = `
+
@(Brokers)
+
Topic
+
@(Supports variables, example: \`device/{device-id}\`)
+
Static message(string)
+
@(Supports variables), @(If specified then incoming data are ignored and this message is sent instead. Topic is required if static message is defined.)
+
@(QoS)
+
@(Retain)
+
+`;
+
+exports.readme = `# MQTT publish
+
+If the topic field is left empty and the data object does not have a 'topic' property then nothing is send.
+Also if data object has a valid topic property it is assumed the object also have data property which is send as a payload;
+Example:
+\`\`\`javacsript
+{
+ topic: '/topic',
+ data: {
+ hello: 'world'
+ }
+}
+// in above case only { hello: 'world' } is published
+\`\`\`
+
+If the topic field is not empty then the entire incomming data object is passed to the output.`;
+
+
+exports.install = function(instance) {
+
+ var PUBLISH_OPTIONS = {};
+
+ var ready = false;
+
+ instance.custom.reconfigure = function() {
+
+ ready = false;
+
+ if (!MQTT.broker(instance.options.broker))
+ return instance.status('No broker', 'red');
+
+ if (instance.options.broker) {
+
+ MQTT.add(instance.options.broker, instance.id);
+ ready = true;
+ PUBLISH_OPTIONS.retain = instance.options.retain || false;
+ PUBLISH_OPTIONS.qos = parseInt(instance.options.qos || 0);
+ return;
+ }
+
+ instance.status('Not configured', 'red');
+ };
+
+ instance.on('options', instance.custom.reconfigure);
+
+ instance.on('data', function(flowdata) {
+ if (!ready)
+ return;
+ var msg = instance.options.staticmessage ? instance.arg(instance.options.staticmessage) : flowdata.data;
+ var topic = instance.arg(instance.options.topic || msg.topic);
+ if (topic) {
+ if (msg.topic)
+ msg = msg.data;
+ MQTT.publish(instance.options.broker, topic, msg, PUBLISH_OPTIONS);
+ } else
+ instance.debug('MQTT publish no topic');
+
+ instance.send(flowdata);
+ });
+
+ instance.on('close', function() {
+ MQTT.remove(instance.options.broker, instance.id);
+ OFF('mqtt.brokers.status', instance.custom.brokerstatus);
+ });
+
+
+ instance.custom.brokerstatus = function(status, brokerid, msg) {
+ if (brokerid !== instance.options.broker)
+ return;
+
+ switch (status) {
+ case 'connecting':
+ instance.status('Connecting', '#a6c3ff');
+ break;
+ case 'connected':
+ instance.status('Connected', 'green');
+ break;
+ case 'disconnected':
+ instance.status('Disconnected', 'red');
+ break;
+ case 'connectionfailed':
+ instance.status('Connection failed', 'red');
+ break;
+ case 'new':
+ !ready && instance.custom.reconfigure();
+ break;
+ case 'removed':
+ instance.custom.reconfigure();
+ break;
+ case 'error':
+ instance.status(msg, 'red');
+ break;
+ case 'reconfigured':
+ instance.options.broker = msg;
+ instance.reconfig();
+ instance.custom.reconfigure();
+ break;
+ }
+ }
+
+ ON('mqtt.brokers.status', instance.custom.brokerstatus);
+
+ instance.custom.reconfigure();
+};
diff --git a/flow/mqttsubscribe.js b/flow/mqttsubscribe.js
new file mode 100644
index 0000000..e866178
--- /dev/null
+++ b/flow/mqttsubscribe.js
@@ -0,0 +1,168 @@
+exports.id = 'mqttsubscribe';
+exports.title = 'MQTT subscribe';
+exports.group = 'MQTT';
+exports.color = '#888600';
+exports.version = '1.1.0';
+exports.icon = 'sign-in';
+exports.output = 1;
+exports.variables = true;
+exports.author = 'Martin Smola';
+exports.options = {};
+
+exports.html = `
+
@(Select a broker)
+
Topic
+
@(Supports variables, example: \`device/{device-id}\`)
+
@(QoS)
+
+`;
+
+exports.readme = `
+# MQTT subscribe
+
+The data recieved are passed to the output as follows:
+\`\`\`javascript
+{
+ topic: '/lights/on',
+ data: 'kitchen'
+}
+\`\`\`
+
+If the topic is wildcard then there's an array of matches in flowdata repository which can be used in Function component like so:
+\`\`\`javascript
+// wildcard -> /+/status
+// topic -> /devicename/status
+
+var match = flowdata.get('mqtt_wildcard');
+// match === ['devicename']
+\`\`\`
+
+More on wildcard topics [here](https://mosquitto.org/man/mqtt-7.html)
+`;
+
+exports.install = function(instance) {
+
+ var old_topic;
+ var ready = false;
+
+ instance.custom.reconfigure = function(o, old_options) {
+
+
+ ready = false;
+
+ if (!MQTT.broker(instance.options.broker))
+ return instance.status('No broker', 'red');
+
+ if (instance.options.broker && instance.options.topic) {
+
+ if (old_topic)
+ MQTT.unsubscribe(instance.options.broker, instance.id, old_topic);
+
+ old_topic = instance.arg(instance.options.topic);
+ MQTT.subscribe(instance.options.broker, instance.id, old_topic);
+ ready = true;
+ return;
+ }
+
+ instance.status('Not configured', 'red');
+ };
+
+ instance.on('options', instance.custom.reconfigure);
+
+ instance.on('close', function() {
+ MQTT.unsubscribe(instance.options.broker, instance.id, instance.options.topic);
+ OFF('mqtt.brokers.message', instance.custom.message);
+ OFF('mqtt.brokers.status', instance.custom.brokerstatus);
+ });
+
+ instance.custom.brokerstatus = function(status, brokerid, msg) {
+ if (brokerid !== instance.options.broker)
+ return;
+
+ switch (status) {
+ case 'connecting':
+ instance.status('Connecting', '#a6c3ff');
+ break;
+ case 'connected':
+ instance.status('Connected', 'green');
+ break;
+ case 'disconnected':
+ instance.status('Disconnected', 'red');
+ break;
+ case 'connectionfailed':
+ instance.status('Connection failed', 'red');
+ break;
+ case 'new':
+ !ready && instance.custom.reconfigure();
+ break;
+ case 'removed':
+ instance.custom.reconfigure();
+ break;
+ case 'error':
+ instance.status(msg, 'red');
+ break;
+ case 'reconfigured':
+ instance.options.broker = msg;
+ instance.reconfig();
+ instance.custom.reconfigure();
+ break;
+ }
+ }
+
+ instance.custom.message = function(brokerid, topic, message) {
+ if (brokerid !== instance.options.broker)
+ return;
+
+ var match = mqttWildcard(topic, old_topic);
+ if (match) {
+ var flowdata = instance.make({ topic: topic, data: message })
+ flowdata.set('mqtt_wildcard', match);
+ instance.send2(flowdata);
+ }
+ }
+
+ ON('mqtt.brokers.message', instance.custom.message);
+ ON('mqtt.brokers.status', instance.custom.brokerstatus);
+
+ instance.custom.reconfigure();
+};
+
+// https://github.com/hobbyquaker/mqtt-wildcard
+function mqttWildcard(topic, wildcard) {
+ if (topic === wildcard) {
+ return [];
+ } else if (wildcard === '#') {
+ return [topic];
+ }
+
+ var res = [];
+
+ var t = String(topic).split('/');
+ var w = String(wildcard).split('/');
+
+ var i = 0;
+ for (var lt = t.length; i < lt; i++) {
+ if (w[i] === '+') {
+ res.push(t[i]);
+ } else if (w[i] === '#') {
+ res.push(t.slice(i).join('/'));
+ return res;
+ } else if (w[i] !== t[i]) {
+ return null;
+ }
+ }
+
+ if (w[i] === '#') {
+ i += 1;
+ }
+
+ return (i === w.length) ? res : null;
+}
diff --git a/flow/nodesdb_changecheck.js b/flow/nodesdb_changecheck.js
new file mode 100644
index 0000000..be1e700
--- /dev/null
+++ b/flow/nodesdb_changecheck.js
@@ -0,0 +1,70 @@
+exports.id = 'nodesdb_change_check';
+exports.title = 'Nodes DB change check';
+exports.group = 'Worksys';
+exports.color = '#888600';
+exports.version = '1.0.2';
+exports.icon = 'sign-out';
+exports.input = 1;
+exports.output = 1;
+exports.readme = `Check, if nodes.table db changed compared to original database`;
+
+const fs = require('fs');
+const path = require('path');
+const { sendNotification } = require('./helper/notification_reporter');
+const nodesOriginalFile = path.join(__dirname, '../databases/nodes_original/', 'nodes_original.table');
+
+
+exports.install = function(instance) {
+
+ function compareArrays(array1, array2) {
+ let message = "";
+ let areEqual = true;
+ let zmenene = []
+
+ if (array1.length !== array2.length) {
+ message += "Nezhoda v pocte nodov. "
+ }
+
+ const set1 = new Set(array1.map(obj => JSON.stringify(obj)));
+ const set2 = new Set(array2.map(obj => JSON.stringify(obj)));
+
+ for (const objStr of set1) {
+
+ if (!set2.has(objStr)) {
+ zmenene.push(objStr)
+ areEqual = false;
+ } else {
+ set2.delete(objStr);
+ }
+ }
+
+ if(!areEqual) {
+ message += `Aktualne nody: ${zmenene.toString()}. Zmenene proti originalu: ${Array.from(set2).join(' ')}`;
+ sendNotification("Nodesdb_changecheck", FLOW.GLOBALS.settings.rvoTbName, "nodes_db_changed", "", message, 0, instance);
+ }
+ else console.log("Arrays are equal.");
+
+ console.log(message)
+ }
+
+
+ instance.on("data", _ => {
+
+ let nodesData = FLOW.GLOBALS.nodesData;
+
+ // we check if nodes.table has changed compared to nodes_original.table (we have array of nodes e.g. [{node:255, tbname: "agruhuwhgursuhgo34hgsdiguhrr"}]
+ const nodes_actual = Object.keys(nodesData).map(node => ({[node]: nodesData[node].tbname}))
+ let nodes_original = fs.readFileSync(nodesOriginalFile, { encoding: 'utf8', flag: 'r' });
+
+ try {
+ nodes_original = JSON.parse(nodes_original);
+ } catch(e) {
+ console.log(e)
+ }
+
+
+ setTimeout(() => compareArrays(nodes_actual, nodes_original),10000);
+ })
+
+}
+
diff --git a/flow/nosql.js b/flow/nosql.js
new file mode 100644
index 0000000..db6b7c4
--- /dev/null
+++ b/flow/nosql.js
@@ -0,0 +1,191 @@
+exports.id = 'nosql';
+exports.title = 'NoSQL';
+exports.version = '1.2.1';
+exports.group = 'Databases';
+exports.author = 'Martin Smola';
+exports.color = '#D770AD';
+exports.icon = 'database';
+exports.input = true;
+exports.output = 2;
+exports.options = {};
+exports.readme = `# NoSQL embedded
+
+## Outputs
+
+First output is response from nosql engine and second is the data passed in.
+
+## Collection
+
+if the collection field is left empty, then we try to look at \`flowdata.get('collection')\`, to set this value you need to use \`flowdata.set('collection', '
')\` in previous component (currently only \`function\` can be used)
+
+## Insert
+
+- will insert recieved data
+- expects data to be an Object
+- returns error, success, id
+
+## Read
+
+- will read a document by id
+- expects data to be an Object with an \`id\` property
+- returns error, response
+
+## Update
+
+- will update document by id
+- expects data to be an Object with \`id\` property and all the props to be updated
+- returns error, response
+- if response is 0 then update failed
+
+## Remove
+
+- will remove document by id
+- expects data to be an Object with an \`id\` property
+- returns error, response
+- if response is 0 then remove failed
+
+## Query
+
+- will query DB
+- expects data to be an Array as shown bellow
+- returns error, response
+
+\`\`\`javascript
+[
+ ['where', 'sensor', 'temp'], // builder.where('sensor', 'temp');
+ ['limit', 2] // builder.limit(2);
+]
+\`\`\``;
+
+exports.html = `
+
+
DB collection name
+
@(Method)
+
+
Add unique ID to data before insert
+
+
+
Insert document if it doesn't exist
+
Add unique ID to data before insert (only if it doesn't exist)
+
+
-
-
-
@(Name of this server)
-
-
-
@(Slack channel to receive the alerts)
-
-
-
@(Watch these types, comma separated names)
-
-
-
@(Watch messages that include any of the following strings)
-
-
-
@(Tag people if message includes something)
-
-
-
-
-
-
Hostname or IP address (if not empty - setting will override db setting)
-
-
-
-
+
+
+
Hostname or IP address (if not empty - setting will override db setting)
+
+
+
+