Use Paho-mqtt With Pyqt

How I can use callback function from QThread? Callback function on_message does not print any data. In run() I connect to mqtt-broker and subscribe on topic. on_message must work w

Solution 1:

It is not necessary to use threads, just follow the logic of Qt: let's use signals. In the following code I show an advance of the library of an MQTT client that I am implementing that at least up to now implements what you want, in next days I will add more functionalities.

from PyQt5 import QtCore, QtWidgets
import paho.mqtt.client as mqtt


classMqttClient(QtCore.QObject):
    Disconnected = 0
    Connecting = 1
    Connected = 2

    MQTT_3_1 = mqtt.MQTTv31
    MQTT_3_1_1 = mqtt.MQTTv311

    connected = QtCore.pyqtSignal()
    disconnected = QtCore.pyqtSignal()

    stateChanged = QtCore.pyqtSignal(int)
    hostnameChanged = QtCore.pyqtSignal(str)
    portChanged = QtCore.pyqtSignal(int)
    keepAliveChanged = QtCore.pyqtSignal(int)
    cleanSessionChanged = QtCore.pyqtSignal(bool)
    protocolVersionChanged = QtCore.pyqtSignal(int)

    messageSignal = QtCore.pyqtSignal(str)

    def__init__(self, parent=None):
        super(MqttClient, self).__init__(parent)

        self.m_hostname = ""
        self.m_port = 1883
        self.m_keepAlive = 60
        self.m_cleanSession = True
        self.m_protocolVersion = MqttClient.MQTT_3_1

        self.m_state = MqttClient.Disconnected

        self.m_client =  mqtt.Client(clean_session=self.m_cleanSession,
            protocol=self.protocolVersion)

        self.m_client.on_connect = self.on_connect
        self.m_client.on_message = self.on_message
        self.m_client.on_disconnect = self.on_disconnect


    @QtCore.pyqtProperty(int, notify=stateChanged)defstate(self):
        return self.m_state

    @state.setterdefstate(self, state):
        if self.m_state == state: return
        self.m_state = state
        self.stateChanged.emit(state) 

    @QtCore.pyqtProperty(str, notify=hostnameChanged)defhostname(self):
        return self.m_hostname

    @hostname.setterdefhostname(self, hostname):
        if self.m_hostname == hostname: return
        self.m_hostname = hostname
        self.hostnameChanged.emit(hostname)

    @QtCore.pyqtProperty(int, notify=portChanged)defport(self):
        return self.m_port

    @port.setterdefport(self, port):
        if self.m_port == port: return
        self.m_port = port
        self.portChanged.emit(port)

    @QtCore.pyqtProperty(int, notify=keepAliveChanged)defkeepAlive(self):
        return self.m_keepAlive

    @keepAlive.setterdefkeepAlive(self, keepAlive):
        if self.m_keepAlive == keepAlive: return
        self.m_keepAlive = keepAlive
        self.keepAliveChanged.emit(keepAlive)

    @QtCore.pyqtProperty(bool, notify=cleanSessionChanged)defcleanSession(self):
        return self.m_cleanSession

    @cleanSession.setterdefcleanSession(self, cleanSession):
        if self.m_cleanSession == cleanSession: return
        self.m_cleanSession = cleanSession
        self.cleanSessionChanged.emit(cleanSession)

    @QtCore.pyqtProperty(int, notify=protocolVersionChanged)defprotocolVersion(self):
        return self.m_protocolVersion

    @protocolVersion.setterdefprotocolVersion(self, protocolVersion):
        if self.m_protocolVersion == protocolVersion: returnif protocolVersion in (MqttClient.MQTT_3_1, MQTT_3_1_1):
            self.m_protocolVersion = protocolVersion
            self.protocolVersionChanged.emit(protocolVersion)

    #################################################################    @QtCore.pyqtSlot()defconnectToHost(self):
        if self.m_hostname:
            self.m_client.connect(self.m_hostname, 
                port=self.port, 
                keepalive=self.keepAlive)

            self.state = MqttClient.Connecting
            self.m_client.loop_start()

    @QtCore.pyqtSlot()defdisconnectFromHost(self):
        self.m_client.disconnect()

    defsubscribe(self, path):
        if self.state == MqttClient.Connected:
            self.m_client.subscribe(path)

    ################################################################## callbacksdefon_message(self, mqttc, obj, msg):
        mstr = msg.payload.decode("ascii")
        # print("on_message", mstr, obj, mqttc)
        self.messageSignal.emit(mstr)

    defon_connect(self, *args):
        # print("on_connect", args)
        self.state = MqttClient.Connected
        self.connected.emit()

    defon_disconnect(self, *args):
        # print("on_disconnect", args)
        self.state = MqttClient.Disconnected
        self.disconnected.emit()


classWidget(QtWidgets.QWidget):
    def__init__(self, parent=None):
        super(Widget, self).__init__(parent)

        lay = QtWidgets.QVBoxLayout(self)
        self.lcd_number = QtWidgets.QLCDNumber()
        lay.addWidget(self.lcd_number)

        self.client = MqttClient(self)
        self.client.stateChanged.connect(self.on_stateChanged)
        self.client.messageSignal.connect(self.on_messageSignal)

        self.client.hostname = "broker.hivemq.com"
        self.client.connectToHost()

    @QtCore.pyqtSlot(int)defon_stateChanged(self, state):
        if state == MqttClient.Connected:
            print(state)
            self.client.subscribe("shok2")

    @QtCore.pyqtSlot(str)defon_messageSignal(self, msg):
        try:
            val = float(msg)
            self.lcd_number.display(val)
        except ValueError:
            print("error: Not is number")


if __name__ == '__main__':
    import sys

    app = QtWidgets.QApplication(sys.argv)
    w = Widget()
    w.show()
    sys.exit(app.exec_())

Based on this answer I have created a library to use paho with PyQt5 and PySide2: https://github.com/eyllanesc/qtmqtt

Solution 2:

Do you have to create the method on_connect use the subscribe on on_connect method not on run.

defon_connect(self, client, userdata, flags, rc):
    print"Connected", rc
    client.subscribe("shok2")

defrun(self):
    msg = subscribe.simple("shok2", hostname="broker.hivemq.com")
    print("%s %s" % (msg.topic, msg.payload))
    mqttc = mqtt.Client()
    mqttc.on_subscribe = self.on_subscribe
    mqttc.on_message = self.on_message
    mqttc.on_connect = on_connect
    mqttc.connect("broker.hivemq.com")
    mqttc.loop_forever()