connection_string_builder.py

import sys
from PyQt5.QtWidgets import (
    QWidget, QLabel, QLineEdit, QVBoxLayout, QFormLayout, QComboBox,
    QPushButton, QCheckBox, QApplication, QFileDialog, QMessageBox, QStackedLayout, QHBoxLayout
)
from PyQt5.QtCore import Qt, pyqtSignal


class ConnectionStringBuilder(QWidget):
    """
    A simple tool to generate PLC4X connection strings for different protocols, such as OPCUA, Modbus, and S7.
    """

    connectionStringGenerated = pyqtSignal(str)

    def __init__(self, initial_connection_string="", connection_type='production', parent=None):
        super().__init__(parent)
        self.connection_type = connection_type
        self.setWindowTitle(f"Connection String Builder ({connection_type.capitalize()})")
        self.setGeometry(100, 100, 600, 600)

        self.init_ui()
        if initial_connection_string:
            self.load_connection_string(initial_connection_string)

    def init_ui(self):
        # Main Layout
        main_layout = QVBoxLayout(self)

        mode_layout = QFormLayout()
        self.mode_dropdown = QComboBox()
        self.mode_dropdown.addItems(["OPCUA", "Modbus", "S7"])
        self.mode_dropdown.currentTextChanged.connect(self.update_mode)

        self.expert_mode_checkbox = QCheckBox("Expert Mode")
        self.expert_mode_checkbox.stateChanged.connect(self.toggle_expert_mode)

        mode_layout.addRow("Mode:", self.mode_dropdown)
        mode_layout.addRow("", self.expert_mode_checkbox)

        self.manual_connection_edit = QLineEdit()
        self.manual_connection_edit.setPlaceholderText("Enter custom connection string here")
        self.manual_connection_edit.setVisible(False)

        main_layout.addLayout(mode_layout)
        main_layout.addWidget(self.manual_connection_edit)

        self.stacked_layout = QStackedLayout()
        main_layout.addLayout(self.stacked_layout)

        self.init_opcua_form()
        self.init_modbus_form()
        self.init_s7_form()

        generate_button = QPushButton("Generate Connection String")
        generate_button.clicked.connect(self.generate_connection_string)
        main_layout.addWidget(generate_button)

        self.generated_connection_edit = QLineEdit()
        self.generated_connection_edit.setReadOnly(True)
        main_layout.addWidget(QLabel("Generated Connection String:"))
        main_layout.addWidget(self.generated_connection_edit)

        self.update_mode(self.mode_dropdown.currentText())

    def load_connection_string(self, connection_str):
        """
        Parses the connection string and populates the UI fields accordingly.
        """

        try:
            if connection_str.startswith("opcua:"):
                mode = "OPCUA"
                stripped_conn = connection_str[len("opcua:"):]
            elif connection_str.startswith("opc.tcp://"):
                mode = "OPCUA"
                stripped_conn = connection_str 
            elif connection_str.startswith("modbus-"):
                mode = "Modbus"
                stripped_conn = connection_str[len("modbus-tcp://"):]
            elif connection_str.startswith("s7://"):
                mode = "S7"
                stripped_conn = connection_str[len("s7://"):]
            else:
                self.expert_mode_checkbox.setChecked(True)
                self.manual_connection_edit.setText(connection_str)
                return

            self.mode_dropdown.setCurrentText(mode)

            if mode == "OPCUA":
                self.parse_opcua_connection_string(stripped_conn)
            elif mode == "Modbus":
                self.parse_modbus_connection_string(connection_str)
            elif mode == "S7":
                self.parse_s7_connection_string(connection_str)

        except Exception as e:
            QMessageBox.critical(self, "Error", f"Failed to parse connection string: {e}")
            self.expert_mode_checkbox.setChecked(True)
            self.manual_connection_edit.setText(connection_str)

    def parse_opcua_connection_string(self, connection_str):
        """
        Parses an OPCUA connection string and populates the OPCUA form fields.
        """

        try:
            # Handle both "opc.tcp://" and "tcp://" prefixes
            if connection_str.startswith("opc.tcp://"):
                conn = connection_str[len("opc.tcp://"):]
            elif connection_str.startswith("tcp://"):
                conn = connection_str[len("tcp://"):]
            else:
                conn = connection_str

            protocol, rest = conn.split("://", 1) if "://" in conn else ("opc.tcp", conn)
            address_port, _, params = rest.partition("?")
            ip, port = address_port.split(":") if ":" in address_port else (address_port, "4840") # as defult port = 4840

            self.opcua_ip_edit.setText(ip)
            self.opcua_port_edit.setText(port)

            params_dict = {}
            if params:
                for param in params.split("&"):
                    if '=' in param:
                        key, value = param.split("=", 1)
                        params_dict[key] = value

            self.opcua_discovery_checkbox.setChecked(params_dict.get("discovery", "false").lower() == "true")

            security = params_dict.get("security", "NONE")
            self.opcua_security_dropdown.setCurrentText(security)

            self.set_opcua_security_fields_visible(security)

            security_policy = params_dict.get("security-policy", "NONE")
            if security_policy != "NONE":
                self.opcua_security_policy_dropdown.setCurrentText(security_policy)

            if security in ["SIGN", "SIGN_ENCRYPT"]:
                self.opcua_username_edit.setText(params_dict.get("username", ""))
                self.opcua_password_edit.setText(params_dict.get("password", ""))

            if security in ["SIGN", "SIGN_ENCRYPT"]:
                self.opcua_key_store_file_edit.setText(params_dict.get("key-store-file", ""))
                self.opcua_key_store_password_edit.setText(params_dict.get("key-store-password", ""))
                self.opcua_server_certificate_file_edit.setText(params_dict.get("server-certificate-file", ""))

        except Exception as e:
            raise ValueError(f"Invalid OPCUA connection string format: {e}")

    def parse_modbus_connection_string(self, connection_str):
        """
        Parses a Modbus connection string and populates the Modbus form fields.
        """

        try:
            conn = connection_str[len("modbus-"):]
            protocol, address = conn.split("://", 1)
            self.modbus_protocol_dropdown.setCurrentText(protocol.upper())
            if protocol.lower() == "tcp":
                ip = address
                self.modbus_ip_edit.setText(ip)
            elif protocol.lower() == "serial":
                port = address
                self.modbus_ip_edit.setText(port)
        except Exception as e:
            raise ValueError(f"Invalid Modbus connection string format: {e}")

    def parse_s7_connection_string(self, connection_str):
        """
        Parses an S7 connection string and populates the S7 form fields.
        """

        try:
            conn = connection_str[len("s7://"):]
            address, _, params = conn.partition("?")
            ip, port = address.split(":") if ":" in address else (address, "102")
            self.s7_ip_edit.setText(ip)
            self.s7_port_edit.setText(port)

            params_dict = {}
            if params:
                for param in params.split("&"):
                    if '=' in param:
                        key, value = param.split("=", 1)
                        params_dict[key] = value

        except Exception as e:
            raise ValueError(f"Invalid S7 connection string format: {e}")

    def init_opcua_form(self):
        """
        Initialize the OPCUA configuration form.
        """

        self.opcua_widget = QWidget()
        opcua_layout = QFormLayout(self.opcua_widget)

        self.opcua_ip_edit = QLineEdit()
        self.opcua_port_edit = QLineEdit()
        self.opcua_port_edit.setPlaceholderText("Default: 4840")
        self.opcua_discovery_checkbox = QCheckBox("Enable Discovery")
        self.opcua_security_dropdown = QComboBox()
        self.opcua_security_dropdown.addItems(["NONE", "SIGN", "SIGN_ENCRYPT"])
        self.opcua_security_dropdown.currentTextChanged.connect(self.update_opcua_security_fields)

        # New Security Policy Dropdown
        self.opcua_security_policy_dropdown = QComboBox()
        self.opcua_security_policy_dropdown.addItems([
            "NONE",
            "Basic128Rsa15",
            "Basic256",
            "Basic256Sha256",
            "Aes128_Sha256_RsaOaep",
            "Aes256_Sha256_RsaPss"
        ])
        self.opcua_security_policy_dropdown.setCurrentText("NONE")

        self.opcua_username_edit = QLineEdit()
        self.opcua_password_edit = QLineEdit()
        self.opcua_password_edit.setEchoMode(QLineEdit.Password)

        # New Fields
        self.opcua_key_store_file_edit = QLineEdit()
        self.opcua_key_store_password_edit = QLineEdit()
        self.opcua_key_store_password_edit.setEchoMode(QLineEdit.Password)
        self.opcua_server_certificate_file_edit = QLineEdit()

        # Buttons to browse files
        self.opcua_browse_key_store_file = QPushButton("Browse")
        self.opcua_browse_key_store_file.clicked.connect(lambda: self.browse_file(self.opcua_key_store_file_edit))
        self.opcua_browse_server_cert_file = QPushButton("Browse")
        self.opcua_browse_server_cert_file.clicked.connect(lambda: self.browse_file(self.opcua_server_certificate_file_edit))

        # Layout
        opcua_layout.addRow("IP:", self.opcua_ip_edit)
        opcua_layout.addRow("Port:", self.opcua_port_edit)
        opcua_layout.addRow("", self.opcua_discovery_checkbox)
        opcua_layout.addRow("Security:", self.opcua_security_dropdown)

        # For Production Connection String, include all security options
        opcua_layout.addRow("Security Policy:", self.opcua_security_policy_dropdown)

        # Username and Password Fields
        self.opcua_username_label = QLabel("Username:")
        self.opcua_password_label = QLabel("Password:")

        # Certificate fields
        self.opcua_key_store_file_label = QLabel("Key Store File:")
        self.opcua_key_store_password_label = QLabel("Key Store Password:")
        self.opcua_server_certificate_file_label = QLabel("Server Certificate File:")

        # Username and Password Fields
        opcua_layout.addRow(self.opcua_username_label, self.opcua_username_edit)
        opcua_layout.addRow(self.opcua_password_label, self.opcua_password_edit)

        # Key Store File
        key_store_file_layout = QHBoxLayout()
        key_store_file_layout.addWidget(self.opcua_key_store_file_edit)
        key_store_file_layout.addWidget(self.opcua_browse_key_store_file)
        opcua_layout.addRow(self.opcua_key_store_file_label, key_store_file_layout)

        # Key Store Password
        opcua_layout.addRow(self.opcua_key_store_password_label, self.opcua_key_store_password_edit)

        # Server Certificate File
        server_cert_file_layout = QHBoxLayout()
        server_cert_file_layout.addWidget(self.opcua_server_certificate_file_edit)
        server_cert_file_layout.addWidget(self.opcua_browse_server_cert_file)
        opcua_layout.addRow(self.opcua_server_certificate_file_label, server_cert_file_layout)

        self.set_opcua_security_fields_visible("NONE")

        self.stacked_layout.addWidget(self.opcua_widget)

    def browse_file(self, line_edit):
        """
        Opens a file dialog to select a file and sets the selected file path to the provided QLineEdit.
        """

        options = QFileDialog.Options()
        file_path, _ = QFileDialog.getOpenFileName(
            self,
            "Select File",
            "",
            "All Files (*)",
            options=options
        )
        if file_path:
            line_edit.setText(file_path)

    def init_modbus_form(self):
        """
        Initialize the Modbus configuration form.
        """

        self.modbus_widget = QWidget()
        modbus_layout = QFormLayout(self.modbus_widget)

        self.modbus_ip_edit = QLineEdit()
        self.modbus_protocol_dropdown = QComboBox()
        self.modbus_protocol_dropdown.addItems(["TCP", "Serial"])

        modbus_layout.addRow("IP/Port:", self.modbus_ip_edit)
        modbus_layout.addRow("Protocol:", self.modbus_protocol_dropdown)

        self.stacked_layout.addWidget(self.modbus_widget)

    def init_s7_form(self):
        """Initialize the S7 configuration form."""
        self.s7_widget = QWidget()
        s7_layout = QFormLayout(self.s7_widget)

        self.s7_ip_edit = QLineEdit()
        self.s7_port_edit = QLineEdit()

        s7_layout.addRow("IP:", self.s7_ip_edit)
        s7_layout.addRow("Port:", self.s7_port_edit)

        self.stacked_layout.addWidget(self.s7_widget)

    def toggle_expert_mode(self, state):
        is_expert = state == Qt.Checked
        self.manual_connection_edit.setVisible(is_expert)
        self.stacked_layout.setCurrentIndex(-1)

        for i in range(self.stacked_layout.count()):
            widget = self.stacked_layout.widget(i)
            widget.setEnabled(not is_expert)
        if not is_expert:
            self.update_mode(self.mode_dropdown.currentText())

    def update_mode(self, mode):
        if self.expert_mode_checkbox.isChecked():
            return
        if mode == "OPCUA":
            self.stacked_layout.setCurrentWidget(self.opcua_widget)
        elif mode == "Modbus":
            self.stacked_layout.setCurrentWidget(self.modbus_widget)
        elif mode == "S7":
            self.stacked_layout.setCurrentWidget(self.s7_widget)
        else:
            self.stacked_layout.setCurrentIndex(-1)  # Hide all

    def update_opcua_security_fields(self, security_type):
        """
        Show or hide OPCUA security options based on security_type.
        """
        self.set_opcua_security_fields_visible(security_type)

    def set_opcua_security_fields_visible(self, security_type):
        """
        Helper method to set visibility of OPCUA security fields.
        """

        self.opcua_key_store_file_label.hide()
        self.opcua_key_store_file_edit.hide()
        self.opcua_browse_key_store_file.hide()
        self.opcua_key_store_password_label.hide()
        self.opcua_key_store_password_edit.hide()
        self.opcua_server_certificate_file_label.hide()
        self.opcua_server_certificate_file_edit.hide()
        self.opcua_browse_server_cert_file.hide()

        if security_type == "SIGN":
            self.opcua_username_label.show()
            self.opcua_username_edit.show()
            self.opcua_password_label.show()
            self.opcua_password_edit.show()
            self.opcua_security_policy_dropdown.show()
            self.opcua_security_policy_dropdown.setEnabled(True)

        elif security_type == "SIGN_ENCRYPT":
            self.opcua_username_label.show()
            self.opcua_username_edit.show()
            self.opcua_password_label.show()
            self.opcua_password_edit.show()
            self.opcua_security_policy_dropdown.show()
            self.opcua_security_policy_dropdown.setEnabled(True)
            self.opcua_key_store_file_label.show()
            self.opcua_key_store_file_edit.show()
            self.opcua_browse_key_store_file.show()
            self.opcua_key_store_password_label.show()
            self.opcua_key_store_password_edit.show()
            self.opcua_server_certificate_file_label.show()
            self.opcua_server_certificate_file_edit.show()
            self.opcua_browse_server_cert_file.show()
        else:
            self.opcua_username_label.hide()
            self.opcua_username_edit.hide()
            self.opcua_password_label.hide()
            self.opcua_password_edit.hide()
            self.opcua_security_policy_dropdown.hide()
            self.opcua_security_policy_dropdown.setCurrentText("NONE")
            self.opcua_key_store_file_label.hide()
            self.opcua_key_store_file_edit.hide()
            self.opcua_browse_key_store_file.hide()
            self.opcua_key_store_password_label.hide()
            self.opcua_key_store_password_edit.hide()
            self.opcua_server_certificate_file_label.hide()
            self.opcua_server_certificate_file_edit.hide()
            self.opcua_browse_server_cert_file.hide()

    def generate_connection_string(self):
        if self.expert_mode_checkbox.isChecked():
            connection_str = self.manual_connection_edit.text().strip()
            if not connection_str:
                QMessageBox.warning(self, "Input Error", "Please enter a connection string.")
                return
            self.generated_connection_edit.setText(connection_str)

            # Emit the generated connection string
            self.connectionStringGenerated.emit(connection_str)
            return

        mode = self.mode_dropdown.currentText()
        connection_str = ""

        if mode == "OPCUA":
            ip = self.opcua_ip_edit.text().strip()
            port = self.opcua_port_edit.text().strip() or "4840"
            if not ip or not port:
                QMessageBox.warning(self, "Input Error", "Please enter both IP and Port for OPCUA.")
                return
            
            params = []

            if self.connection_type == 'scrape':
                connection_str = f"opc.tcp://{ip}:{port}"
                security = self.opcua_security_dropdown.currentText()
                if security != "NONE":
                    params.append(f"security={security}")

                    # Security Policy
                    security_policy = self.opcua_security_policy_dropdown.currentText()
                    if security_policy != "NONE":
                        params.append(f"security-policy={security_policy}")

                    # Username and Password
                    if security in ["SIGN", "SIGN_ENCRYPT"]:
                        username = self.opcua_username_edit.text().strip()
                        password = self.opcua_password_edit.text().strip()
                        if not username or not password:
                            QMessageBox.warning(self, "Input Error", "Username and Password are required for the selected security level.")
                            return
                        params.append(f"username={username}")
                        params.append(f"password={password}")

                    # Key Store and Server Certificate Files
                    if security in ["SIGN", "SIGN_ENCRYPT"]:
                        key_store_file = self.opcua_key_store_file_edit.text().strip()
                        key_store_password = self.opcua_key_store_password_edit.text().strip()
                        server_certificate_file = self.opcua_server_certificate_file_edit.text().strip()

                        if not key_store_file or not key_store_password or not server_certificate_file:
                            QMessageBox.warning(self, "Input Error", "Key Store File, Key Store Password, and Server Certificate File are required for the selected security level.")
                            return

                        params.append(f"key-store-file={key_store_file}")
                        params.append(f"key-store-password={key_store_password}")
                        params.append(f"server-certificate-file={server_certificate_file}")

            else:
                connection_str = f"opcua:tcp://{ip}:{port}"
                # For production connection string, include all params
                if self.opcua_discovery_checkbox.isChecked():
                    params.append("discovery=true")

                security = self.opcua_security_dropdown.currentText()
                if security != "NONE":
                    params.append(f"security={security}")

                    # Security Policy
                    security_policy = self.opcua_security_policy_dropdown.currentText()
                    if security_policy != "NONE":
                        params.append(f"security-policy={security_policy}")

                    # Username and Password
                    if security in ["SIGN", "SIGN_ENCRYPT"]:
                        username = self.opcua_username_edit.text().strip()
                        password = self.opcua_password_edit.text().strip()
                        if not username or not password:
                            QMessageBox.warning(self, "Input Error", "Username and Password are required for the selected security level.")
                            return
                        params.append(f"username={username}")
                        params.append(f"password={password}")

                    # Key Store and Server Certificate Files
                    if security in ["SIGN", "SIGN_ENCRYPT"]:
                        key_store_file = self.opcua_key_store_file_edit.text().strip()
                        key_store_password = self.opcua_key_store_password_edit.text().strip()
                        server_certificate_file = self.opcua_server_certificate_file_edit.text().strip()

                        if not key_store_file or not key_store_password or not server_certificate_file:
                            QMessageBox.warning(self, "Input Error", "Key Store File, Key Store Password, and Server Certificate File are required for the selected security level.")
                            return

                        params.append(f"key-store-file={key_store_file}")
                        params.append(f"key-store-password={key_store_password}")
                        params.append(f"server-certificate-file={server_certificate_file}")

            if params:
                connection_str += "?" + "&".join(params)

        elif mode == "Modbus":
            ip_or_port = self.modbus_ip_edit.text().strip()
            protocol = self.modbus_protocol_dropdown.currentText().lower()
            if not ip_or_port:
                QMessageBox.warning(self, "Input Error", "Please enter the IP or Port for Modbus.")
                return
            connection_str = f"modbus-{protocol}://{ip_or_port}"

        elif mode == "S7":
            ip = self.s7_ip_edit.text().strip()
            if not ip:
                QMessageBox.warning(self, "Input Error", "Please enter IP for S7.")
                return
            port = self.s7_port_edit.text().strip()

            if port:
                connection_str = f"s7://{ip}:{port}"
            else:
                connection_str = f"s7://{ip}"

        self.generated_connection_edit.setText(connection_str)
        self.connectionStringGenerated.emit(connection_str)

def main():
    app = QApplication(sys.argv)
    app.setStyle("Fusion")
    window = ConnectionStringBuilder()
    window.show()
    sys.exit(app.exec_())


if __name__ == "__main__":
    main()