MqttMonitor/MqttConnectionOptionsDialog.py

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

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

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

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

from EricWidgets import EricMessageBox
from EricWidgets.EricPathPicker import EricPathPickerModes

from .Ui_MqttConnectionOptionsDialog import Ui_MqttConnectionOptionsDialog

from .MqttClient import MqttClient, MqttProtocols

from Utilities.crypto import pwConvert


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", "TlsEnable", "TlsCaCert".
        @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.__populateDefaults(options=options)
        
        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(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)
    
    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", "Keepalive",
            "CleanSession", "Username", "Password", "WillTopic", "WillMessage",
            "WillQos", "WillRetain", "TlsEnable", "TlsCaCert",
            "ConnectionTimeout".
        @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"])
        
        # TLS parameters
        self.tlsEnableCheckBox.setChecked(options["TlsEnable"])
        self.tlsCertsFilePicker.setText(options["TlsCaCert"])
    
    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", "TlsEnable", "TlsCaCert".
        @rtype dict
        """
        if self.mqttv31Button.isChecked():
            protocol = MqttProtocols.MQTTv31
        elif self.mqttv311Button.isChecked():
            protocol = MqttProtocols.MQTTv311
        elif self.mqttv5Button.isChecked():
            protocol = MqttProtocols.MQTTv5
        else:
            protocol = MqttProtocols.MQTTv311
        
        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(),
            "TlsEnable": self.tlsEnableCheckBox.isChecked(),
            "TlsCaCert": self.tlsCertsFilePicker.text()
        }
    
    @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()