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
Port
@(Client id)
@(Username)
`; 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 } ] */