317 lines
9.8 KiB
JavaScript
317 lines
9.8 KiB
JavaScript
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 = `<div class="padding">
|
|
<div class="row">
|
|
<div class="col-md-6">
|
|
<div data-jc="textbox" data-jc-path="host" data-jc-config="placeholder:test.mosquitto.org;required:false" class="m">Hostname or IP address</div>
|
|
</div>
|
|
<div class="col-md-6">
|
|
<div data-jc="textbox" data-jc-path="port" data-jc-config="placeholder:1883;required:true" class="m">Port</div>
|
|
</div>
|
|
</div>
|
|
<div class="row">
|
|
<div class="col-md-6">
|
|
<div data-jc="textbox" data-jc-path="clientid">@(Client id)</div>
|
|
</div>
|
|
<div class="col-md-6">
|
|
<div data-jc="textbox" data-jc-path="username" class="m">@(Username)</div>
|
|
</div>
|
|
</div>
|
|
</div>`;
|
|
|
|
|
|
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
|
|
}
|
|
]
|
|
*/
|
|
|
|
|
|
|