bundles/mosquitto: add telegraf stats for tasmota devices

bundles/mosquitto/files/tasmota-telegraf-plugin

@@ -0,0 +1,53 @@
+#!/usr/bin/env python3
+
+from json import loads
+from subprocess import Popen, PIPE
+from sys import argv, stderr
+from time import sleep
+
+while True:
+    try:
+        proc = Popen(['mosquitto_sub', '-h', argv[1], '-v', '-t', argv[2]], stdout=PIPE)
+
+        while True:
+            try:
+                line = proc.stdout.readline().decode().strip()
+
+                if not line:
+                    continue
+
+                topic, message = line.split(' ', 1)
+
+                if not any(message.startswith(i) for i in ['{', '[']):
+                    # probably not json
+                    continue
+
+                json = loads(message)
+                topic_split = topic.strip('/').split('/')
+                sensor_name = '_'.join(topic_split[:-1])
+
+                if topic_split[-1] == 'STATE':
+                    # /switch/wohnzimmer/ambilight/STATE {"Time":"2021-05-15T07:40:16","Uptime":"4T18:48:43","Vcc":3.108,"Heap":19,"SleepMode":"Dynamic","Sleep":50,"LoadAvg":19,"POWER":"OFF","Wifi":{"AP":1,"SSId":"NSA Surveillance Van 2342","BSSId":"DA:CC:46:87:58:F7","Channel":6,"RSSI":100,"LinkCount":2,"Downtime":"0T00:00:27"}}
+                    power_state = bool(json['POWER'] == 'ON')
+                    print(f'tasmota,device={sensor_name} state={power_state}', flush=True)
+                    print(f'tasmota,device={sensor_name} rssi={json["Wifi"]["RSSI"]}i', flush=True)
+
+                if topic_split[-1] == 'SENSOR':
+                    # /switch/wohnzimmer/ambilight/SENSOR {"Time":"2021-05-15T07:40:16","ENERGY":{"TotalStartTime":"2019-08-02T13:56:56","Total":2.324,"Yesterday":0.004,"Today":0.000,"Period":0,"Power":0,"ApparentPower":0,"ReactivePower":0,"Factor":0.00,"Voltage":237,"Current":0.000}}
+                    print(f'tasmota,device={sensor_name} energy_total={json["ENERGY"]["Total"]}', flush=True)
+                    print(f'tasmota,device={sensor_name} energy_today={json["ENERGY"]["Today"]}', flush=True)
+                    print(f'tasmota,device={sensor_name} energy_yesterday={json["ENERGY"]["Yesterday"]}', flush=True)
+                    print(f'tasmota,device={sensor_name} power={json["ENERGY"]["Power"]}', flush=True)
+                    print(f'tasmota,device={sensor_name} apparent_power={json["ENERGY"]["ApparentPower"]}', flush=True)
+                    print(f'tasmota,device={sensor_name} reactive_power={json["ENERGY"]["ReactivePower"]}', flush=True)
+                    print(f'tasmota,device={sensor_name} powerfactor={json["ENERGY"]["Factor"]}', flush=True)
+                    print(f'tasmota,device={sensor_name} voltage={json["ENERGY"]["Voltage"]}i', flush=True)
+                    print(f'tasmota,device={sensor_name} current={json["ENERGY"]["Current"]}', flush=True)
+            except Exception as e:
+                if line:
+                    print('parse error while parsing a line:', file=stderr)
+                    print(line, file=stderr)
+                print(repr(e), file=stderr, flush=True)
+    except Exception as e:
+        print(repr(e), file=stderr, flush=True)
+        sleep(10)

bundles/mosquitto/items.py

@@ -5,6 +5,9 @@ files = {
             'svc_systemd:mosquitto:restart',
         },
     },
+    '/usr/local/bin/tasmota-telegraf-plugin': {
+        'mode': '0755',
+    },
 }
 
 svc_systemd = {

bundles/mosquitto/metadata.py

@@ -40,3 +40,32 @@ def iptables(metadata):
             'port_rules': result,
         },
     }
+
+
+@metadata_reactor.provides(
+    'telegraf/input_plugins/builtin/execd',
+)
+def telegraf(metadata):
+    topic = metadata.get('mosquitto/tasmota-telegraf-topic', None)
+
+    if not topic:
+        return {}
+
+    return {
+        'telegraf': {
+            'input_plugins': {
+                'execd': {
+                    'tasmota': {
+                        'command': [
+                            '/usr/local/bin/tasmota-telegraf-plugin',
+                            '127.0.0.1',
+                            topic
+                        ],
+                        'signal': 'none',
+                        'restart_delay': '1s',
+                        'data_format': 'influx',
+                    },
+                },
+            },
+        },
+    }

bundles/telegraf/items.py

@@ -55,7 +55,7 @@ telegraf_config = {
     },
 }
 
-# Bundlewrap can't merge lists. To work around this, telegraf/input_plugins/exec
+# Bundlewrap can't merge lists. To work around this, telegraf/input_plugins/exec(d)
 # is a dict, of which we only use the value of it. This also allows us
 # to overwrite values set by metadata defaults/reactors in node and group
 # metadata, if needed.
@@ -65,6 +65,12 @@ for config in node.metadata.get('telegraf/input_plugins/exec', {}).values():
 
     telegraf_config['inputs']['exec'].append(config)
 
+for config in node.metadata.get('telegraf/input_plugins/execd', {}).values():
+    if 'execd' not in telegraf_config['inputs']:
+        telegraf_config['inputs']['execd'] = []
+
+    telegraf_config['inputs']['execd'].append(config)
+
 files = {
     '/etc/telegraf/telegraf.conf': {
         'content_type': 'mako',

nodes/home/nas.py

@@ -103,6 +103,7 @@ nodes['home.nas'] = {
                     'protocol': 'websockets',
                 },
             },
+            'tasmota-telegraf-topic': '/switch/#',
             'restrict-to': {
                 '172.19.136.0/25',
                 '172.19.138.0/24',