Tasmota – MQTT Daten mit Grafana visualisieren Python + InfluxDB

Ich habe nach einer einfachen Möglichkeit gesucht die Daten von einem Tasmota Gerät mit Grafana zu visualisieren. Einfach ist hier immer so eine Sache, denn hier muss man immer etwas dazwischen schalten, was die Daten mittels MQTT abgreift und dann in eine Datenbank schiebt. In der Regel wird immer zu einer fertigen Lösung wie Homeassistant, ioBroker, OpenHab oder NodeRED geraten. Ich hatte aber kein Bock auf die Lösungen und hab nach einem einfachen Skript gesucht welches genau diese Anforderungen erfüllt. Tadaaa… Python Skript steht unten.

#	Vorschau	Produkt	Bewertung	Preis
1		Tasmota Steckdose NOUS A1T, Kleinste Wlan ESP8266 smart plug, 16A mini wifi steckdosen mit...	Aktuell keine Bewertungen	49,99 EUR	Bei Amazon ansehen
2		TASMOTA Steckdose 16A, CURRYSMARTER V1.1 Mini Stecker mit Stromzähler | Wlan ESP8285 Plug |...	Aktuell keine Bewertungen		Bei Amazon ansehen
3		Tasmota Steckdose mit Stromzähler, EIGHTREE Alexa WLAN Smart Steckdose, Smart Home, ESP8685 mit...	Aktuell keine Bewertungen	45,99 EUR	Bei Amazon ansehen

Voraussetzungen für diese Skript Lösung

Da deine Tasmota Dose über das MQTT Protokoll spricht, brauchst du einen laufenden Mosquitto Server. Dieser wird für die Kommunikation zwischen Tasmota Dose und den Skript benötigt. Und natürlich muss eine Datenbank vorhanden sein, wo die Daten abgelegt werden. In meinem Falle habe ich mich für eine InfluxDB entschieden, da diese einfach in der Handhabung ist, aber auch mit großen Datenmengen super umgehen kann.

• Mosquitto Server – Port 1883
• InfluxDB – Port 8086

Im folgenden Bild könnt Ihr auch erkennen, dass ich bei mir das MQTT Topic auf “shellies” umgestellt habe. Also alle Shellys sollen zukünftig in diesem Topic senden, damit auch alle in die Datenbank aufgenommen werden. Natürlich sollte sich der TopicNAME / Gerätename sich zwischen euren verschiedenen Dosen unterscheiden. Wenn Ihr ein anderes Full Topic wählt, müsst Ihr dies natürlich im Skript anpassen.

Das Skript muss natürlich auf einem System ausgeführt werden wo Python3 läuft. Also z.B. ein Raspberry Pi oder ein kleines Linux System, wie ein LXC Container unter Proxmox. Benötigt werden folgende Pakete.

apt install python3 python3-pip -y

pip3 install paho-mqtt
pip3 install influxdb

import re
import json
from typing import NamedTuple

import paho.mqtt.client as mqtt
from influxdb import InfluxDBClient

INFLUXDB_ADDRESS = 'DEINE IP VON DER INFLUXDB'
INFLUXDB_USER = ''
INFLUXDB_PASSWORD = ''
INFLUXDB_DATABASE = 'DB NAME'

MQTT_ADDRESS = '192.168.10.20'
MQTT_USER = ''
MQTT_PASSWORD = ''
MQTT_TOPIC = 'shellies/+/SENSOR'
MQTT_REGEX = 'shellies/([^/]+)/([^/]+)'
MQTT_CLIENT_ID = 'MQTT_InfluxDB_Bridge'

MQTT_FIELDS = [
    'ApparentPower',
    'Current',
    'Voltage',
    'Power',
    'ReactivePower',
    'Factor'
    # hier weitere relevante auflisten
]

influxdb_client = InfluxDBClient(INFLUXDB_ADDRESS, 8086, INFLUXDB_USER, INFLUXDB_PASSWORD, None)

class SensorData(NamedTuple):
    location: str
    measurement: str
    value: float

def on_connect(client, userdata, flags, rc):
    """ The callback for when the client receives a CONNACK response from the server."""
    print('Connected with result code ' + str(rc))
    client.subscribe(MQTT_TOPIC)

def _parse_mqtt_message(topic, payload):
    #print(payload)
    match = re.match(MQTT_REGEX, topic)
    if match:
        location = match.group(1)
        measurement = match.group(2)
        if measurement == 'status':
            return None
        #return SensorData(location, measurement, float(payload))
        print(payload)
        payload = json.loads(payload)
        for field in MQTT_FIELDS:
            yield SensorData(location, field, float(payload["ENERGY"][field]))
    else:
        return None

def _send_sensor_data_to_influxdb(sensor_data):
    json_body = [
        {
            'measurement': sensor_data.measurement,
            'tags': {
                'location': sensor_data.location
            },
            'fields': {
                'value': sensor_data.value
            }
        }
    ]
    influxdb_client.write_points(json_body)

def on_message(client, userdata, msg):
    """The callback for when a PUBLISH message is received from the server."""
    print(msg.topic + ' ' + str(msg.payload))
    sensor_data_sets = _parse_mqtt_message(msg.topic, msg.payload.decode('utf-8'))
    if sensor_data_sets is None:
        print("Couldn't parse sensor data!")
        return
    for sensor_data in sensor_data_sets:
        _send_sensor_data_to_influxdb(sensor_data)

def _init_influxdb_database():
    databases = influxdb_client.get_list_database()
    if len(list(filter(lambda x: x['name'] == INFLUXDB_DATABASE, databases))) == 0:
        influxdb_client.create_database(INFLUXDB_DATABASE)
    influxdb_client.switch_database(INFLUXDB_DATABASE)

def main():
    _init_influxdb_database()

    mqtt_client = mqtt.Client(MQTT_CLIENT_ID)
    mqtt_client.username_pw_set(MQTT_USER, MQTT_PASSWORD)
    mqtt_client.on_connect = on_connect
    mqtt_client.on_message = on_message

    mqtt_client.connect(MQTT_ADDRESS, 1883)
    mqtt_client.loop_forever()


if __name__ == '__main__':
    print('MQTT to InfluxDB bridge')
    main()