MQTT Monitor: comparison MqttMonitor/MqttClient.py

-:17ef10819773
+:40582e448c4b
 Module implementing a PyQt wrapper around the paho MQTT client.
 """
 from __future__ import unicode_literals
-from PyQt5.QtCore import pyqtSignal, QObject, QCoreApplication
+from PyQt5.QtCore import pyqtSignal, pyqtSlot, QObject, QCoreApplication, \
+QTimer
 import paho.mqtt.client as mqtt
 from Utilities.crypto import pwConvert
 @signal onConnect(flags, rc) emitted after the client has connected to the
 broker
 @signal onDisconnected(rc) emitted after the client has disconnected from
 the broker
+@signal onLog(level, message) emitted to send client log data
 @signal onMessage(topic, payload, qos, retain) emitted after a message has
 been received by the client
 @signal onPublish(mid) emitted after a message has been published
 @signal onSubscribe(mid, grantedQos) emitted after the client has
 subscribed to some topics
 @signal onUnsubscribe(mid) emitted after the client has unsubscribed from
 some topics
 """
 onConnect = pyqtSignal(dict, int)
 onDisconnected = pyqtSignal(int)
+onLog = pyqtSignal(int, str)
 onMessage = pyqtSignal(str, bytes, int, bool)
 onPublish = pyqtSignal(int)
 onSubscribe = pyqtSignal(int, tuple)
 onUnsubscribe = pyqtSignal(int)
+connectTimeout = pyqtSignal()
+DefaultConnectTimeout = 15      # connect timeout in seconds
+LogDebug = 0x01
+LogInfo = 0x02
+LogNotice = 0x04
+LogWarning = 0x08
+LogError = 0x10
+LogLevelMap = {
+mqtt.MQTT_LOG_DEBUG: LogDebug,
+mqtt.MQTT_LOG_INFO: LogInfo,
+mqtt.MQTT_LOG_NOTICE: LogNotice,
+mqtt.MQTT_LOG_WARNING: LogWarning,                        # __NO-TASK__
+mqtt.MQTT_LOG_ERR: LogError,
+}
 def __init__(self, clientId="", cleanSession=True, userdata=None,
 protocol=mqtt.MQTTv311, transport="tcp", parent=None):
 """
 Constructor
 """
 QObject.__init__(self, parent=parent)
 self.__loopStarted = False
+self.__connectTimeoutTimer = QTimer(self)
+self.__connectTimeoutTimer.setSingleShot(True)
+self.__connectTimeoutTimer.setInterval(
+MqttClient.DefaultConnectTimeout * 1000)
+self.__connectTimeoutTimer.timeout.connect(self.__connectTimeout)
+self.onConnect.connect(self.__connectTimeoutTimer.stop)
 self.__mqttClient = mqtt.Client(
 client_id=clientId, clean_session=cleanSession, userdata=None,
 protocol=mqtt.MQTTv311, transport="tcp")
 self.__initCallbacks()
 self.__mqttClient.on_connect = \
 lambda client, userdata, flags, rc: self.onConnect.emit(
 flags, rc)
 self.__mqttClient.on_disconnect = \
 lambda client, userdata, rc: self.onDisconnected.emit(rc)
+self.__mqttClient.on_log = \
+lambda client, userdata, level, buf: self.onLog.emit(level, buf)
 self.__mqttClient.on_message = \
 lambda client, userdata, message: self.onMessage.emit(
 message.topic, message.payload, message.qos, message.retain)
 self.__mqttClient.on_publish = \
 lambda client, userdata, mid: self.onPublish.emit(mid)
 lambda client, userdata, mid, grantedQos: self.onSubscribe.emit(
 mid, grantedQos)
 self.__mqttClient.on_unsubscribe = \
 lambda client, userdata, mid: self.onUnsubscribe.emit(mid)
+@pyqtSlot()
+def __connectTimeout(self):
+"""
+Privat slot handling a failed connection attempt.
+"""
+self.stopLoop()
+self.connectTimeout.emit()
 def reinitialise(self, clientId="", cleanSession=True, userdata=None):
 """
 Public method to reinitialize the client with given data.
 @param clientId ID to be used for the client
 """
 self.__mqttClient.reinitialise(
 client_id=clientId, clean_session=cleanSession, userdata=userdata)
 self.__initCallbacks()
+def setConnectionTimeout(self, timeout):
+"""
+Public method to set the connection timeout value.
+@param timeout timeout value to be set in seconds
+@type int
+"""
+self.__connectTimeoutTimer.setInterval(timeout * 1000)
 def setMaxInflightMessages(self, inflight=20):
 """
 Public method to set the maximum number of messages with QoS > 0 that
 can be part way through their network flow at once.
 Public method to stop the MQTT client loop.
 """
 self.__mqttClient.loop_stop()
 self.__loopStarted = False
-def connectToServer(self, host, port=1883, keepalive=60, bindAddress=""):
+def connectToServer(self, host, port=1883, keepalive=60, bindAddress="",
+reinit=True):
 """
 Public method to connect to a remote MQTT broker.
 @param host host name or IP address of the remote broker
 @type str
 @type int
 @param bindAddress IP address of a local network interface to bind
 this client to
 @type str
 """
-# TODO: get this fixed or allow to interrupt
+if reinit:
-self.__mqttClient.reconnect_delay_set(max_delay=16)
+self.reinitialise()
 self.__mqttClient.connect_async(
 host, port=port, keepalive=keepalive, bind_address=bindAddress)
+self.__connectTimeoutTimer.start()
 if not self.__loopStarted:
 self.startLoop()
 def connectToServerWithOptions(self, host, port=1883, bindAddress="",
 @type str
 @param options dictionary containing the connection options. This
 dictionary should contain the keys "ClientId", "Keepalive",
 "CleanSession", "Username", "Password", "WillTopic", "WillMessage",
 "WillQos", "WillRetain", "TlsEnable", "TlsCaCert", "TlsClientCert",
-"TlsClientKey"
+"TlsClientKey", "ConnectionTimeout"
 @type dict
 """
 if options:
 parametersDict = self.defaultConnectionOptions()
 parametersDict.update(options)
 # step 1: reinitialize to set the client ID and clean session flag
 self.reinitialise(
 clientId=parametersDict["ClientId"],
 cleanSession=parametersDict["CleanSession"]
 )
+self.setConnectionTimeout(parametersDict["ConnectionTimeout"])
 # step 2: set username and password
 if parametersDict["Username"]:
 if parametersDict["Password"]:
 self.setUserCredentials(
 # use default TLS configuration
 self.setTLS()
 # step 5: connect to server
 self.connectToServer(host, port=port,
-keepalive=parametersDict["Keepalive"])
+keepalive=parametersDict["Keepalive"],
+reinit=False)
 else:
 keepalive = self.defaultConnectionOptions["Keepalive"]
 self.connectToServer(host, port=port, keepalive=keepalive,
 bindAddress=bindAddress)
 values.
 @return dictionary containing the default connection options. It has
 the keys "ClientId", "Keepalive", "CleanSession", "Username",
 "Password", "WillTopic", "WillMessage", "WillQos", "WillRetain",
-"TlsEnable", "TlsCaCert", "TlsClientCert", "TlsClientKey".
+"TlsEnable", "TlsCaCert", "TlsClientCert", "TlsClientKey",
+"ConnectionTimeout".
 @rtype dict
 """
 return {
 "ClientId": "ERIC6_MQTT_MONITOR_CLIENT",
+"ConnectionTimeout": MqttClient.DefaultConnectTimeout,
 "Keepalive": 60,
 "CleanSession": True,
 "Username": "",
 "Password": "",
 "WillTopic": "",
 def reconnectToServer(self):
 """
 Public method to reconnect the client with the same parameters.
 """
+self.__connectTimeoutTimer.start()
 self.__mqttClient.reconnect()
 if not self.__loopStarted:
 self.startLoop()
 def disconnectFromServer(self):
 """
 Public method to disconnect the client from the remote broker.
 """
+self.__connectTimeoutTimer.stop()
 self.__mqttClient.disconnect()
 def subscribe(self, topic, qos=0):
 """
 Public method to subscribe to topics with quality of service.
 retain=retain)
 def mqttConnackMessage(connackCode):
 """
-Public method to get the string associated with a CONNACK result.
+Module function to get the string associated with a CONNACK result.
 @param connackCode result code of the connection request
 @type int
 @return textual representation for the result code
 @rtype str
 "Connection Refused: unknown reason.")
 def mqttErrorMessage(mqttErrno):
 """
-Public method to get the error string associated with an MQTT error
+Module function to get the error string associated with an MQTT error
 number.
 @param mqttErrno result code of a MQTT request
 @type int
 @return textual representation of the result code
 "Error defined by errno.")
 else:
 return QCoreApplication.translate(
 "MqttErrorMessage",
 "Unknown error.")
+def mqttLogLevelString(mqttLogLevel, isMqttLogLevel=True):
+"""
+Module function to get the log level string associated with a log level.
+@param mqttLogLevel log level of the paho-mqtt client
+@type int
+@param isMqttLogLevel flag indicating a MQTT log level is given (if
+False it is the MqttClient variant, i.e. Debug being lowest)
+@type bool
+@return textual representation of the log level
+@rtype str
+"""
+if isMqttLogLevel:
+try:
+logLevel = MqttClient.LogLevelMap[mqttLogLevel]
+except KeyError:
+return QCoreApplication.translate("MqttLogLevelString", "Unknown")
+else:
+logLevel = mqttLogLevel
+if logLevel == MqttClient.LogInfo:
+return QCoreApplication.translate("MqttLogLevelString", "Info")
+elif logLevel == MqttClient.LogNotice:
+return QCoreApplication.translate("MqttLogLevelString", "Notice")
+elif logLevel == MqttClient.LogWarning:
+return QCoreApplication.translate("MqttLogLevelString", "Warning")
+elif logLevel == MqttClient.LogError:
+return QCoreApplication.translate("MqttLogLevelString", "Error")
+elif logLevel == MqttClient.LogDebug:
+return QCoreApplication.translate("MqttLogLevelString", "Debug")
+else:
+return QCoreApplication.translate("MqttLogLevelString", "Unknown")

Mercurial Repositories > MQTT Monitor / file comparison

comparison: MqttMonitor/MqttClient.py

MqttMonitor/MqttClient.py