bundles/mosquitto: rewrite tasmota-telegraf-plugin using paho-mqtt library

bundles/mosquitto/files/tasmota-telegraf-plugin

@@ -1,53 +1,84 @@
 #!/usr/bin/env python3
 
 from json import loads
-from subprocess import Popen, PIPE
+from os import getpid
 from sys import argv, stderr
 from time import sleep
 
-while True:
+import paho.mqtt.client as mqtt
+
+
+BROKER_HOST = argv[1]
+BROKER_TOPIC = argv[2]
+
+
+def log(msg):
+    print(msg, file=stderr, flush=True)
+
+
+def out(msg):
+    print(msg, flush=True)
+
+
+def on_connect(client, userdata, flags, rc):
+    log(f'connected to mqtt broker at {BROKER_HOST} with status: {mqtt.connack_string(rc)}')
+
+    client.subscribe(BROKER_TOPIC)
+
+
+def on_disconnect(client, userdata, rc):
+    log(f'disconnected from mqtt broker at {BROKER_HOST} with status: {mqtt.error_string(rc)}')
+
+
+def on_message(client, userdata, msg):
     try:
-        proc = Popen(['mosquitto_sub', '-h', argv[1], '-v', '-t', argv[2]], stdout=PIPE)
+        message = msg.payload.decode()
-
-        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
+            return
 
         json = loads(message)
-                topic_split = topic.strip('/').split('/')
+        topic_split = msg.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)
+            out(f'tasmota,device={sensor_name} state={power_state}')
-                    print(f'tasmota,device={sensor_name} rssi={json["Wifi"]["RSSI"]}i', flush=True)
+            out(f'tasmota,device={sensor_name} rssi={json["Wifi"]["RSSI"]}i')
 
         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)
+            out(f'tasmota,device={sensor_name} energy_total={json["ENERGY"]["Total"]}')
-                    print(f'tasmota,device={sensor_name} energy_today={json["ENERGY"]["Today"]}', flush=True)
+            out(f'tasmota,device={sensor_name} energy_today={json["ENERGY"]["Today"]}')
-                    print(f'tasmota,device={sensor_name} energy_yesterday={json["ENERGY"]["Yesterday"]}', flush=True)
+            out(f'tasmota,device={sensor_name} energy_yesterday={json["ENERGY"]["Yesterday"]}')
-                    print(f'tasmota,device={sensor_name} power={json["ENERGY"]["Power"]}', flush=True)
+            out(f'tasmota,device={sensor_name} power={json["ENERGY"]["Power"]}')
-                    print(f'tasmota,device={sensor_name} apparent_power={json["ENERGY"]["ApparentPower"]}', flush=True)
+            out(f'tasmota,device={sensor_name} apparent_power={json["ENERGY"]["ApparentPower"]}')
-                    print(f'tasmota,device={sensor_name} reactive_power={json["ENERGY"]["ReactivePower"]}', flush=True)
+            out(f'tasmota,device={sensor_name} reactive_power={json["ENERGY"]["ReactivePower"]}')
-                    print(f'tasmota,device={sensor_name} powerfactor={json["ENERGY"]["Factor"]}', flush=True)
+            out(f'tasmota,device={sensor_name} powerfactor={json["ENERGY"]["Factor"]}')
-                    print(f'tasmota,device={sensor_name} voltage={json["ENERGY"]["Voltage"]}i', flush=True)
+            out(f'tasmota,device={sensor_name} voltage={json["ENERGY"]["Voltage"]}i')
-                    print(f'tasmota,device={sensor_name} current={json["ENERGY"]["Current"]}', flush=True)
+            out(f'tasmota,device={sensor_name} current={json["ENERGY"]["Current"]}')
     except Exception as e:
-                if line:
+        log(f'parse error while parsing a message on topic {msg.topic}:')
-                    print('parse error while parsing a line:', file=stderr)
+        log(repr(e))
-                    print(line, file=stderr)
+        log(repr(msg.payload))
-                print(repr(e), file=stderr, flush=True)
+
-    except Exception as e:
+
-        print(repr(e), file=stderr, flush=True)
+def on_subscribe(client, userdata, mid, granted_qos):
-        sleep(10)
+    log(f'subscription {mid} was successful')
+
+
+if __name__ == '__main__':
+    client = mqtt.Client(
+        client_id=f'tasmota-telegraf-plugin_{getpid()}'
+    )
+
+    client.on_connect = on_connect
+    client.on_disconnect = on_disconnect
+    client.on_message = on_message
+    client.on_subscribe = on_subscribe
+
+    client.connect(BROKER_HOST, 1883, 30)
+
+    client.loop_forever()

bundles/mosquitto/items.py

@@ -7,6 +7,9 @@ files = {
     },
     '/usr/local/bin/tasmota-telegraf-plugin': {
         'mode': '0755',
+        'needs': {
+            'pkg_apt:python3-paho-mqtt',
+        },
     },
 }
 
@@ -18,3 +21,8 @@ svc_systemd = {
         },
     },
 }
+
+if node.has_bundle('telegraf'):
+    files['/usr/local/bin/tasmota-telegraf-plugin']['triggers'] = {
+        'svc_systemd:telegraf:restart',
+    }

bundles/mosquitto/metadata.py

@@ -6,6 +6,7 @@ defaults = {
         'packages': {
             'mosquitto': {},
             'mosquitto-clients': {},
+            'python3-paho-mqtt': {}, # for telegraf plugin
         },
     },
     'icinga2_api': {