Add MQTT exporting

danielwippermann · Nov 29, 2018 · 366d426 · 366d426
1 parent 7c78cde
commit 366d426
Show file tree

Hide file tree

Showing 4 changed files with 607 additions and 0 deletions.
diff --git a/examples/vbustouch-proxy/config.js.example b/examples/vbustouch-proxy/config.js.example
@@ -72,6 +72,46 @@ module.exports = {
      */
     rewriteWebHeaderSets: false,
 
+    /**
+     * Interval (milliseconds) in which data will be uploaded to MQTT. A value of zero disables this functionality.
+     */
+    mqttInterval: 5000,
+
+    /**
+     * MQTT connect parameters, https://github.com/mqttjs/MQTT.js#connect
+     */
+    mqttConnect: {
+        host: 'mqtt://localhost',
+        username: 'mosquitto',
+        password: 'secret',
+    },
+
+    /**
+     * MQTT topic to publish to.
+     */
+    mqttTopic: 'home/resol',
+
+    /**
+     * A map of MQTT message attributes to VBus packet field IDs.
+     *
+     * An example sensor in Home Assistant would be:
+     * - platform: mqtt
+     *    name: "Resol Collector Temp"
+     *    state_topic: "home/resol"
+     *    unit_of_measurement: '°C'
+     *    value_template: "{{ value_json.temp1 }}"
+     */
+    mqttPacketFieldMap: {
+        temp1: '00_0010_427B_10_0100_000_2_0',
+        temp2: '00_0010_427B_10_0100_002_2_0',
+        temp3: '00_0010_427B_10_0100_004_2_0',
+        temp4: '00_0010_427B_10_0100_006_2_0',
+        relay1: '00_0010_427B_10_0100_008_1_0',
+        relay2: '00_0010_427B_10_0100_012_1_0',
+        relay1hrs: '00_0010_427B_10_0100_010_2_0',
+        relay2hrs: '00_0010_427B_10_0100_014_2_0'
+    },
+
     /**
      * Interval (milliseconds) in which data will be uploaded to PvOutput.org. A value of zero disables this functionality.
      */

diff --git a/examples/vbustouch-proxy/index.js b/examples/vbustouch-proxy/index.js
@@ -12,6 +12,7 @@ const express = require('express');
 const morgan = require('morgan');
 const request = require('request');
 const winston = require('winston');
+const mqtt = require('mqtt');
 
 
 const {
@@ -422,6 +423,64 @@ const startHeaderSetConsolidatorTimer = async () => {
 };
 
 
+const startMqttLogging = async () => {
+    const onHeaderSet = async (headerSet, client) => {
+        const headers = headerSet.getSortedHeaders();
+        const packetFields = specification.getPacketFieldsForHeaders(headers);
+
+        const valuesById = packetFields.reduce((memo, pf) => {
+            const precision = pf.packetFieldSpec.type.precision;
+
+            const roundedRawValue = pf.rawValue.toFixed(precision);
+
+            //logger.debug('ID = ' + JSON.stringify(pf.id) + ', Name = ' + JSON.stringify(pf.name) + ', Value = ' + pf.rawValue + ', RoundedValue = ' + roundedRawValue);
+
+            memo [pf.id] = roundedRawValue;
+            return memo;
+        }, {});
+
+        const params = Object.keys(config.mqttPacketFieldMap).reduce((memo, key) => {
+            const packetFieldId = config.mqttPacketFieldMap [key];
+
+            let value;
+            if (typeof packetFieldId === 'function') {
+                value = packetFieldId(valuesById);
+            } else {
+                value = valuesById [packetFieldId];
+            }
+            if (typeof value === 'number') {
+                value = value.toString();
+            }
+            if (typeof value === 'string') {
+                memo [key] = value;
+            }
+            return memo;
+        }, {
+            key: 'hi'
+        });
+        client.publish(config.mqttTopic, JSON.stringify(params));
+    };
+
+    if (config.mqttInterval) {
+        logger.debug('Starting MQTT logging');
+        const client  = mqtt.connect(config.mqttConnect)
+
+        client.on('connect', function () {
+            const hsc = new HeaderSetConsolidator({
+                interval: config.mqttInterval,
+            });
+
+            hsc.on('headerSet', () => {
+                onHeaderSet(headerSetConsolidator, client).then(null, err => {
+                    logger.error(err);
+                });
+            });
+
+            hsc.startTimer();
+        });
+    }
+};
+
 const startPvOutputOrgLogging = async () => {
     const onHeaderSet = async (headerSet) => {
         const headers = headerSet.getSortedHeaders();
@@ -568,6 +627,8 @@ const main = async () => {
 
     await startHeaderSetConsolidatorTimer();
 
+    await startMqttLogging();
+
     await startPvOutputOrgLogging();
 
     await startTextLogging();