Mercurial Repositories > MQTT Monitor / file revision

# -*- coding: utf-8 -*-

# Copyright (c) 2018 - 2024 Detlev Offenbach <detlev@die-offenbachs.de>
#

"""
Module implementing a dialog to enter MQTT connection options.
"""

import copy

from PyQt6.QtCore import QUuid, pyqtSlot
from PyQt6.QtWidgets import QAbstractButton, QDialog, QDialogButtonBox

try:
    from eric7.EricGui import EricPixmapCache
except ImportError:
    from UI import PixmapCache as EricPixmapCache

from eric7.EricWidgets import EricMessageBox
from eric7.EricWidgets.EricPathPicker import EricPathPickerModes
from eric7.Utilities.crypto import pwConvert

from .MqttClient import MqttClient
from .MqttProtocols import MqttProtocols
from .Ui_MqttConnectionOptionsDialog import Ui_MqttConnectionOptionsDialog


class MqttConnectionOptionsDialog(QDialog, Ui_MqttConnectionOptionsDialog):
    """
    Class implementing a dialog to enter MQTT connection options.
    """

    def __init__(self, options=None, parent=None):
        """
        Constructor

        @param options dictionary containing the connection options to
            populate the dialog with. It must have the keys "ClientId",
            "Protocol", "ConnectionTimeout", "Keepalive", "CleanSession",
            "Username", "Password", "WillTopic", "WillMessage", "WillQos",
            "WillRetain", "WillProperties", "TlsEnable", "TlsCaCert",
            "UserProperties".
        @type dict
        @param parent reference to the parent widget
        @type QWidget
        """
        super().__init__(parent)
        self.setupUi(self)

        self.tlsCertsFilePicker.setMode(EricPathPickerModes.OPEN_FILE_MODE)
        self.tlsCertsFilePicker.setFilters(
            self.tr("Certificate Files (*.crt *.pem);;All Files (*)")
        )

        self.willPropertiesButton.setIcon(EricPixmapCache.getIcon("listSelection"))

        self.optionsWidget.setCurrentIndex(0)

        # initialize MQTTv5 related stuff
        self.on_mqttv5Button_toggled(False)

        self.__populateDefaults(options=options)

        self.connectPropertiesButton.clicked[bool].connect(
            self.__propertiesTypeSelected
        )
        self.disconnectPropertiesButton.clicked[bool].connect(
            self.__propertiesTypeSelected
        )

        self.__updateOkButton()

    def __updateOkButton(self):
        """
        Private method to update the enabled state of the OK button.
        """
        if self.clientIdEdit.text() == "" and not self.cleanSessionCheckBox.isChecked():
            enable = False
            EricMessageBox.critical(
                self,
                self.tr("Invalid Connection Parameters"),
                self.tr("""An empty Client ID requires a clean session."""),
            )
        else:
            enable = True

        self.buttonBox.button(QDialogButtonBox.StandardButton.Ok).setEnabled(enable)

    @pyqtSlot()
    def on_generateIdButton_clicked(self):
        """
        Private slot to generate a client ID.
        """
        uuid = QUuid.createUuid()
        self.clientIdEdit.setText(uuid.toString(QUuid.StringFormat.WithoutBraces))

    @pyqtSlot(str)
    def on_clientIdEdit_textChanged(self, clientId):
        """
        Private slot handling a change of the client ID string.

        @param clientId client ID
        @type str
        """
        self.__updateOkButton()

    @pyqtSlot(bool)
    def on_cleanSessionCheckBox_clicked(self, checked):
        """
        Private slot to handle a change of the clean session selection.

        @param checked current state of the clean session selection
        @type bool
        """
        self.__updateOkButton()

    @pyqtSlot(bool)
    def on_mqttv5Button_toggled(self, checked):
        """
        Private slot to handle the selection of the MQTT protocol.

        @param checked state of the button
        @type bool
        """
        self.optionsWidget.setTabEnabled(
            self.optionsWidget.indexOf(self.propertiesTab), checked
        )
        self.willPropertiesButton.setEnabled(checked)
        self.willPropertiesButton.setVisible(checked)

    @pyqtSlot(QAbstractButton)
    def on_buttonBox_clicked(self, button):
        """
        Private slot to handle the press of a button box button.

        @param button button that has been pressed
        @type QAbstractButton
        """
        if button == self.buttonBox.button(
            QDialogButtonBox.StandardButton.RestoreDefaults
        ):
            self.__populateDefaults(options=None)

    @pyqtSlot(bool)
    def on_samePropertiesCheckBox_toggled(self, checked):
        """
        Private slot to handle a change of the properties usage.

        @param checked flag indicating to use the same user properties for
            CONNECT and DISCONNECT
        @type bool
        """
        if checked and not self.connectPropertiesButton.isChecked():
            self.connectPropertiesButton.click()
        self.disconnectPropertiesButton.setEnabled(not checked)

    @pyqtSlot(bool)
    def __propertiesTypeSelected(self, checked):
        """
        Private slot to handle the switching of the user properties type.

        @param checked state of the buttons
        @type bool
        """
        if checked:
            # handle the selection only
            if self.connectPropertiesButton.isChecked():
                self.__userProperties["disconnect"] = (
                    self.propertiesWidget.getProperties()
                )
                self.propertiesWidget.setProperties(self.__userProperties["connect"])
            else:
                self.__userProperties["connect"] = self.propertiesWidget.getProperties()
                self.propertiesWidget.setProperties(self.__userProperties["disconnect"])

    @pyqtSlot()
    def on_willPropertiesButton_clicked(self):
        """
        Private slot to edit the last will user properties.
        """
        from .MqttUserPropertiesEditor import MqttUserPropertiesEditorDialog

        dlg = MqttUserPropertiesEditorDialog(
            self.tr("Last Will User Properties"), self.__willProperties, self
        )
        if dlg.exec() == QDialog.DialogCode.Accepted:
            self.__willProperties = dlg.getProperties()

    def __populateDefaults(self, options=None):
        """
        Private method to populate the dialog.

        If no options dictionary is given, the dialog will be populated with
        default values.

        @param options dictionary containing the connection options to populate
            the dialog with. It must have the keys "ClientId", "Protocol",
            "ConnectionTimeout", "Keepalive", "CleanSession", "Username",
            "Password", "WillTopic", "WillMessage", "WillQos", "WillRetain",
            "WillProperties", "TlsEnable", "TlsCaCert", "UserProperties".
        @type dict
        """
        if options is None:
            options = MqttClient.defaultConnectionOptions()

        # general
        self.clientIdEdit.setText(options["ClientId"])
        self.mqttv31Button.setChecked(options["Protocol"] == MqttProtocols.MQTTv31)
        self.mqttv311Button.setChecked(options["Protocol"] == MqttProtocols.MQTTv311)
        self.mqttv5Button.setChecked(options["Protocol"] == MqttProtocols.MQTTv5)
        self.connectionTimeoutSpinBox.setValue(options["ConnectionTimeout"])
        self.keepaliveSpinBox.setValue(options["Keepalive"])
        self.cleanSessionCheckBox.setChecked(options["CleanSession"])

        # user credentials
        self.usernameEdit.setText(options["Username"])
        self.passwordEdit.setText(pwConvert(options["Password"], encode=False))

        # last will and testament
        self.willQosSpinBox.setValue(options["WillQos"])
        self.willRetainCheckBox.setChecked(options["WillRetain"])
        self.willTopicEdit.setText(options["WillTopic"])
        self.willMessageEdit.setPlainText(options["WillMessage"])
        self.__willProperties = copy.deepcopy(options.get("WillProperties", []))

        # TLS parameters
        self.tlsEnableCheckBox.setChecked(options["TlsEnable"])
        self.tlsCertsFilePicker.setText(options["TlsCaCert"])

        # user properties
        self.__userProperties = copy.deepcopy(options.get("UserProperties", {}))
        if not self.__userProperties:
            self.__userProperties = {
                "connect": [],
                "disconnect": [],
                "use_connect": True,
            }

        if options["Protocol"] == MqttProtocols.MQTTv5:
            self.connectPropertiesButton.setChecked(True)
            self.propertiesWidget.setProperties(self.__userProperties["connect"])
            self.samePropertiesCheckBox.setChecked(self.__userProperties["use_connect"])
            self.disconnectPropertiesButton.setEnabled(
                not self.__userProperties["use_connect"]
            )
        else:
            self.propertiesWidget.clear()

    def getConnectionOptions(self):
        """
        Public method get the entered connection options.

        @return dictionary containing the connection options. It has the keys
            "ClientId", "Protocol", "ConnectionTimeout", "Keepalive",
            "CleanSession", "Username", "Password", "WillTopic", "WillMessage",
            "WillQos", "WillRetain", "WillProperties", "TlsEnable",
            "TlsCaCert", "UserProperties".
        @rtype dict
        """
        if self.mqttv31Button.isChecked():
            protocol = MqttProtocols.MQTTv31
        elif self.mqttv311Button.isChecked():
            protocol = MqttProtocols.MQTTv311
        elif self.mqttv5Button.isChecked():
            protocol = MqttProtocols.MQTTv5
        else:
            # should never happen
            protocol = MqttProtocols.MQTTv311

        if protocol == MqttProtocols.MQTTv5:
            if self.connectPropertiesButton.isChecked():
                self.__userProperties["connect"] = self.propertiesWidget.getProperties()
            else:
                self.__userProperties["disconnect"] = (
                    self.propertiesWidget.getProperties()
                )
            self.__userProperties["use_connect"] = (
                self.samePropertiesCheckBox.isChecked()
            )
        else:
            self.__userProperties = {}
            self.__willProperties = []

        return {
            "ClientId": self.clientIdEdit.text(),
            "Protocol": protocol,
            "ConnectionTimeout": self.connectionTimeoutSpinBox.value(),
            "Keepalive": self.keepaliveSpinBox.value(),
            "CleanSession": self.cleanSessionCheckBox.isChecked(),
            "Username": self.usernameEdit.text(),
            "Password": pwConvert(self.passwordEdit.text(), encode=True),
            "WillTopic": self.willTopicEdit.text(),
            "WillMessage": self.willMessageEdit.toPlainText(),
            "WillQos": self.willQosSpinBox.value(),
            "WillRetain": self.willRetainCheckBox.isChecked(),
            "WillProperties": copy.deepcopy(self.__willProperties),
            "TlsEnable": self.tlsEnableCheckBox.isChecked(),
            "TlsCaCert": self.tlsCertsFilePicker.text(),
            "UserProperties": copy.deepcopy(self.__userProperties),
        }