Mqtt Paho subscribe und Daten vom drei Sensoren bekommen.

jesuisla · Nov 29th 2022

Guten Tag zusammen,

ich möchte mich mit 3 MikroControllern subscriben und daten bekommen, bis jetzt klappt es mit einen, ich habe aber noch 2 andere, die mir Daten vom Sensoren über Mqtt schicken.

kann mir bitte jemand helfen, wie ich vom drei verschiedene Sensoren Daten bekomme und in Mysql die Daten speichere?

Code

import datetime
import threading
import mysql.connector
import paho.mqtt.client as mqtt_client
import tables

ID_machinen = {"CoEx1": 1, "CoEx2": 2, "MonoLinie": 3}

def connect_mqtt(ipadresse, portmachine):

    def on_connect(client, userdata, flags, rc):

        if rc == 0:
            print("Connected to MQTT Broker!")
        else:
            print("Failed to connect, return code %d\n", rc)
    client = mqtt_client.Client('user')
    client.on_connect = on_connect
    client.connect(ipadresse, portmachine)
    return client

def subscribe(client: mqtt_client, topic):

    def on_message(client, userdata, msg):
        db = mysql.connector.connect(
            host="xxxx",
            user="xxxx",
            password="xxxxxxx",
            port='3306',
            database="xxxx"
        )
        print(datetime.datetime.now(), msg.topic, msg.payload.decode())
        value1 = msg.payload.decode()
        mycursor = db.cursor(buffered=True)
        #mycursor.execute("INSERT into `extrusion`.`wassersensoren` (id,name) VALUES (%s,%s)", (1, 'hier_name_der_machine'))
        mycursor.execute("INSERT into `extrusion`.`verbrauch` (machine,timestamp, verbrauch) VALUES (%s,%s,%s)", (ID_machinen[msg.topic], datetime.datetime.now(), value1))
        db.commit()
    client.subscribe('CoEx2')
    client.on_message = on_message
    #client.loop_forever()

def insert_into_table(client, topic, db):

    mycursor = db.cursor(buffered=True)
    mycursor.execute("""select count(*) from extrusion.verbrauch WHERE timestamp between '2022-11-25 08:00:00' and '2022-11-25 16:00:00'; """)
    summeRows = mycursor.fetchall()
    if len(summeRows):
        summe = summeRows[0][0]
        start_time = '2022-11-25 08:00:00'                              # bitte das Start Time auch hier einragen
        end_time = '2022-11-25 16:00:00'                                # bitte das Ende Time auch hier einragen
        mycursor.execute("INSERT into `extrusion`.`verbrauchproeinheit` (machine,Start,Ende, summe) VALUES (%s,%s,%s,%s)", (2, start_time, end_time, summe))
        db.commit()
        print('das Wasserverbrauch in diesen Intervall von ' + start_time + ' bis %s wurde erfolgreich gespeichert ' + end_time)
    else:
        print('kein Wasserverbrauch in dem angegebenen Zeitintervall')

def main():

    db = mysql.connector.connect(
        host="localhost",
        user="xxxx",
        password="xxxx",
        port='3306',
        database="xxxx"
    )
    listAdrress = [("10.28.1.119", 1883, "CoEx2")]
    listClient = []
    for i in listAdrress:
        listClient.append((connect_mqtt(ipadresse=i[0], portmachine=i[1]), i[2]))
    thread_list = []
    for i in listClient:
        client = i[0]
        topic = i[1]
        thread = threading.Thread(target=subscribe, args=(client, topic))
        thread_intervall = threading.Thread(target=insert_into_table, args=(client, topic, db))
        thread_list.append(thread)
        thread_list.append(thread_intervall)
        thread.start()
        thread_intervall.start()
    #subscribe(client, topic)
    client.loop_forever()

if __name__ == '__main__':

    tables.init()
    main()

Display More

DeaD_EyE · Nov 29th 2022

Das geht auch ohne Threads mit nur 2 Klassen:

Python

import time

from paho.mqtt.client import Client as mqtt_client


class Mqtt2Sql(mqtt_client):
    def __init__(self, database, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.database = database

    def on_connect(self, userdata, flags, reasonCode, properties):
        print("Connected")
        for topic in self.database.id_maschinen:
            print(f"Subscribe {topic}")
            self.subscribe(topic)

    def on_disconnect(self, userdata, reasonCode, properties):
        print("Disconnected")

    def on_message(self, client, userdata, message):
        self.database.insert(message)


class DB:
    id_maschinen = {"CoEx1": 1, "CoEx2": 2, "MonoLinie": 3}

    def __init__(self):
        """
        Verbindung zur DB aufbauen...
        """
        self.connection = ...
        raise NotImplementedError(
            "Code um eine Verbindung zur DB aufzubauen fehlt noch."
        )

    def insert(self, message):
        machine_id = self.id_maschinen.get(message.topic, -1)
        payload = message.payload.decode()

        # hier irgendwas mit machine_id und payload (die Nachricht) machen
        print(machine_id, payload)
        raise NotImplementedError(
            "Hier wird noch nichts in die DB geschrieben. Code vervollständigen."
        )


db = DB()
client = Mqtt2Sql(client_id="Desktop-Client", database=db)
client.connect("localhost")
client.loop_forever()

Display More

Du musst nur noch bei der Klasse DB die Verbindung in der __init__ Methode aufbauen und in der insert Methode die Logik vervollständigen.

Wenn die MQTT-Verbindung getrennt wird, baut er automatisch eine neue auf. Bei der DB müsstest du das selbst machen.

Was nicht enthalten ist, ist die Ausgabe, wenn längere Zeit nichts passiert. Ist das notwendig?

Mqtt Paho subscribe und Daten vom drei Sensoren bekommen.

Tags

Users Viewing This Thread