Qt OPC UA Viewer Example

The Qt OPC UA Viewer example uses the model/view approach to display all nodes of an OPC UA server in a QTreeView.

The simulation server from the C++ OPC UA water pump example can be used for this.

Note

On Windows, the OpenSSL libraries need to be in the path for the libopen6254-1 based backend to work.

Qt OPC UA Viewer Example Screenshot

Download this example

# Copyright (C) 2021 The Qt Company Ltd.
# Copyright (C) 2018 Unified Automation GmbH
# SPDX-License-Identifier: LicenseRef-Qt-Commercial OR BSD-3-Clause

"""PySide6 OpcUa Viewer Example"""

import sys
from mainwindow import MainWindow
from PySide6.QtCore import QCoreApplication
from PySide6.QtWidgets import QApplication


if __name__ == '__main__':
    app = QApplication(sys.argv)
    QCoreApplication.setApplicationName('Qt for Python OpcUa Viewer')
    initial_url = sys.argv[1] if len(sys.argv) > 1 else 'opc.tcp://localhost:48010'
    main_win = MainWindow(initial_url)
    main_win.show()
    sys.exit(app.exec())
# Copyright (C) 2021 The Qt Company Ltd.
# Copyright (C) 2018 Unified Automation GmbH
# SPDX-License-Identifier: LicenseRef-Qt-Commercial OR BSD-3-Clause

from ui_certificatedialog import Ui_CertificateDialog

from PySide6.QtCore import QFile, QIODevice
from PySide6.QtNetwork import QSsl, QSslCertificate
from PySide6.QtGui import QTextCursor
from PySide6.QtWidgets import QDialog


class CertificateDialog(QDialog):

    def __init__(self, parent=None):
        super(CertificateDialog, self).__init__(parent)
        self._cert = QSslCertificate()
        self._trustListDirectory = ''
        self._ui = Ui_CertificateDialog()
        self._ui.setupUi(self)
        self._ui.btnTrust.clicked.connect(self.saveCertificate)

    # Returns 0 if the connect should be aborted, 1 if it should be resumed.
    def showCertificate(self, message, der, trustListDirectory):
        certs = QSslCertificate.fromData(der, QSsl.Der)
        self._trustListDirectory = trustListDirectory

        # If it is an untrusted self-signed certificate we can allow to
        # trust it.
        if len(certs) == 1 and certs[0].isSelfSigned():
            self._cert = certs[0]
            self._ui.btnTrust.setEnabled(True)
        else:
            self._ui.btnTrust.setEnabled(False)

        for cert in certs:
            self._ui.certificate.appendPlainText(cert.toText())

        self._ui.message.setText(message)
        self._ui.certificate.moveCursor(QTextCursor.Start)
        self._ui.certificate.ensureCursorVisible()

        return self.exec()

    def saveCertificate(self):
        digest = self._cert.digest()
        hex_digest = digest.toHex()
        path = f"{self._trustListDirectory}/{hex_digest}.der"

        file = QFile(path)
        if file.open(QIODevice.WriteOnly):
            file.write(self._cert.toDer())
            file.close()
<?xml version="1.0" encoding="UTF-8"?>
<ui version="4.0">
 <class>CertificateDialog</class>
 <widget class="QDialog" name="CertificateDialog">
  <property name="geometry">
   <rect>
    <x>0</x>
    <y>0</y>
    <width>720</width>
    <height>423</height>
   </rect>
  </property>
  <property name="windowTitle">
   <string>Dialog</string>
  </property>
  <layout class="QVBoxLayout" name="verticalLayout">
   <item>
    <widget class="QLabel" name="message">
     <property name="text">
      <string>TextLabel</string>
     </property>
     <property name="alignment">
      <set>Qt::AlignLeading|Qt::AlignLeft|Qt::AlignTop</set>
     </property>
     <property name="wordWrap">
      <bool>true</bool>
     </property>
    </widget>
   </item>
   <item>
    <widget class="QPlainTextEdit" name="certificate">
     <property name="readOnly">
      <bool>true</bool>
     </property>
    </widget>
   </item>
   <item>
    <layout class="QHBoxLayout" name="horizontalLayout">
     <item>
      <spacer name="horizontalSpacer">
       <property name="orientation">
        <enum>Qt::Horizontal</enum>
       </property>
       <property name="sizeHint" stdset="0">
        <size>
         <width>40</width>
         <height>20</height>
        </size>
       </property>
      </spacer>
     </item>
     <item>
      <widget class="QPushButton" name="btnIgnore">
       <property name="text">
        <string>Ignore Error</string>
       </property>
      </widget>
     </item>
     <item>
      <widget class="QPushButton" name="btnTrust">
       <property name="text">
        <string>Trust Certificate</string>
       </property>
      </widget>
     </item>
     <item>
      <widget class="QPushButton" name="btnCancel">
       <property name="text">
        <string>Cancel</string>
       </property>
       <property name="default">
        <bool>true</bool>
       </property>
      </widget>
     </item>
    </layout>
   </item>
  </layout>
 </widget>
 <resources/>
 <connections>
  <connection>
   <sender>btnIgnore</sender>
   <signal>clicked()</signal>
   <receiver>CertificateDialog</receiver>
   <slot>accept()</slot>
   <hints>
    <hint type="sourcelabel">
     <x>475</x>
     <y>402</y>
    </hint>
    <hint type="destinationlabel">
     <x>374</x>
     <y>397</y>
    </hint>
   </hints>
  </connection>
  <connection>
   <sender>btnTrust</sender>
   <signal>clicked()</signal>
   <receiver>CertificateDialog</receiver>
   <slot>accept()</slot>
   <hints>
    <hint type="sourcelabel">
     <x>563</x>
     <y>404</y>
    </hint>
    <hint type="destinationlabel">
     <x>287</x>
     <y>395</y>
    </hint>
   </hints>
  </connection>
  <connection>
   <sender>btnCancel</sender>
   <signal>clicked()</signal>
   <receiver>CertificateDialog</receiver>
   <slot>reject()</slot>
   <hints>
    <hint type="sourcelabel">
     <x>670</x>
     <y>406</y>
    </hint>
    <hint type="destinationlabel">
     <x>717</x>
     <y>395</y>
    </hint>
   </hints>
  </connection>
 </connections>
</ui>
# Copyright (C) 2021 The Qt Company Ltd.
# SPDX-License-Identifier: LicenseRef-Qt-Commercial OR BSD-3-Clause

from textwrap import dedent

from PySide6.QtCore import QCoreApplication, Qt
from PySide6.QtWidgets import (QDialog, QDialogButtonBox, QLabel,
                               QVBoxLayout)


_DOC_URL = 'https://doc.qt.io/QtOPCUA/index.html'

_HELP_FORMAT = dedent("""
    <html></head><body><p>
    The {} can be used to browse
    servers from the <a href="{}">QtOpcUa</A> module, for example the
    waterpump simulation server, which runs on port 43344.</p></body></html>
    """)


class HelpDialog(QDialog):

    def __init__(self, parent=None):
        global _HELP_FORMAT, _DOC_URL

        super(HelpDialog, self).__init__(parent)
        name = QCoreApplication.applicationName()
        self.setWindowTitle('{} Help'.format(name))
        vbox_layout = QVBoxLayout(self)
        help_text = _HELP_FORMAT.format(name, _DOC_URL)
        display = QLabel(help_text, self, wordWrap=True,
                         openExternalLinks=True,
                         textInteractionFlags=Qt.TextBrowserInteraction)
        vbox_layout.addWidget(display)
        vbox_layout.addStretch()
        button_box = QDialogButtonBox(QDialogButtonBox.Close)
        vbox_layout.addWidget(button_box)
        button_box.rejected.connect(self.reject)
# Copyright (C) 2021 The Qt Company Ltd.
# Copyright (C) 2018 Unified Automation GmbH
# SPDX-License-Identifier: LicenseRef-Qt-Commercial OR BSD-3-Clause

import os
from pathlib import Path
from textwrap import dedent

from ui_mainwindow import Ui_MainWindow
from opcuamodel import OpcUaModel
from certificatedialog import CertificateDialog
from helpdialog import HelpDialog

from PySide6.QtCore import (QCoreApplication, QDir, qDebug, QMetaObject,
                            Qt, QtMsgType, Slot, QUrl, qInstallMessageHandler,
                            qWarning, Q_ARG)
from PySide6.QtGui import QColor, QFontDatabase, QKeySequence
from PySide6.QtWidgets import (QHeaderView, QMainWindow, QMessageBox)
from PySide6.QtOpcUa import (QOpcUa, QOpcUaApplicationDescription,
                             QOpcUaAuthenticationInformation, QOpcUaProvider,
                             QOpcUaErrorState, QOpcUaPkiConfiguration,
                             QOpcUaUserTokenPolicy)

_main_window = None
_MESSAGE_TYPES = {QtMsgType.QtWarningMsg: "Warning",
                  QtMsgType.QtCriticalMsg: "Critical",
                  QtMsgType.QtFatalMsg: "Fatal", QtMsgType.QtInfoMsg: "Info",
                  QtMsgType.QtDebugMsg: "Debug"}


_MESSAGE_COLORS = {QtMsgType.QtWarningMsg: Qt.GlobalColor.darkYellow,
                   QtMsgType.QtCriticalMsg: Qt.GlobalColor.darkRed,
                   QtMsgType.QtFatalMsg: Qt.GlobalColor.darkRed,
                   QtMsgType.QtInfoMsg: Qt.GlobalColor.black,
                   QtMsgType.QtDebugMsg: Qt.GlobalColor.black}


def _messageLogContext(context):
    """Return a short, readable string from a QMessageLogContext."""
    result = " ("
    if context.file:
        result += Path(context.file).name + ":" + str(context.line)
    if context.function and "(" in context.function:
        function = context.function[:context.function.index("(")] + "()"
        space = function.rfind(" ")
        result += ", " + (function[space + 1:] if space != -1 else function)
    result += ")"
    return result


def _messageHandler(msg_type, context, message):
    global _main_window

    if _main_window:
        type = _MESSAGE_TYPES[msg_type]
        text = f'{type:<7}: {message}'
        color = QColor(_MESSAGE_COLORS[msg_type])
        QMetaObject.invokeMethod(_main_window, "log", Qt.QueuedConnection,
                                 Q_ARG(str, text),
                                 Q_ARG(str, _messageLogContext(context)),
                                 Q_ARG(QColor, color))


class MainWindow(QMainWindow):

    def __init__(self, initial_url, parent=None):
        global _main_window

        super(MainWindow, self).__init__(parent)

        _main_window = self

        self._end_point_list = []
        self._opcua_client = None
        self._client_connected = False
        self._identity = None
        self._pki_config = QOpcUaPkiConfiguration()
        self._endpoint = None
        self._old_messagehandler = qInstallMessageHandler(_messageHandler)

        self._ui = Ui_MainWindow()
        self._ui.setupUi(self)
        self._ui.log.setFont(QFontDatabase.systemFont(QFontDatabase.SystemFont.FixedFont))
        self.setWindowTitle(QCoreApplication.applicationName())
        self._ui.host.setText(initial_url)
        self._ui.quitAction.setShortcut(QKeySequence(Qt.CTRL | Qt.Key_Q))
        self._ui.quitAction.triggered.connect(self.close)
        self._ui.helpAction.setShortcut(QKeySequence(QKeySequence.HelpContents))
        self._ui.helpAction.triggered.connect(self.showHelpDialog)
        self._ui.aboutAction.triggered.connect(qApp.aboutQt)  # noqa: F821

        self._opcua_model = OpcUaModel(self._ui.centralwidget)
        self.mOpcUaProvider = QOpcUaProvider(self._ui.centralwidget)

        self.updateUiState()

        self._ui.opcUaPlugin.addItems(self.mOpcUaProvider.availableBackends())
        self._ui.treeView.setModel(self._opcua_model)

        if not self._ui.opcUaPlugin.count():
            self._ui.opcUaPlugin.setDisabled(True)
            self._ui.connectButton.setDisabled(True)
            error = "The list of available OPCUA plugins is empty. No connection possible."
            QMessageBox.critical(self, "No OPCUA plugins available", error)

        self._ui.host.returnPressed.connect(self.animateFindServersClick)
        self._ui.findServersButton.clicked.connect(self.findServers)
        self._ui.getEndpointsButton.clicked.connect(self.getEndpoints)
        self._ui.connectButton.clicked.connect(self.connectToServer)

        self.setupPkiConfiguration()

        self._identity = self._pki_config.applicationIdentity()

    def __del__(self):
        qInstallMessageHandler(self._old_messagehandler)

    @Slot()
    def animateFindServersClick(self):
        self._ui.findServersButton.animateClick()

    def setupPkiConfiguration(self):
        pkidir = os.path.join(os.path.dirname(__file__), 'pki')
        self._pki_config.setClientCertificateFile(pkidir + "/own/certs/opcuaviewer.der")
        self._pki_config.setPrivateKeyFile(pkidir + "/own/private/opcuaviewer.pem")
        self._pki_config.setTrustListDirectory(pkidir + "/trusted/certs")
        self._pki_config.setRevocationListDirectory(pkidir + "/trusted/crl")
        self._pki_config.setIssuerListDirectory(pkidir + "/issuers/certs")
        self._pki_config.setIssuerRevocationListDirectory(pkidir + "/issuers/crl")

        # create the folders if they don't exist yet
        self.createPkiFolders()

    def createClient(self):
        if not self._opcua_client:
            client_name = self._ui.opcUaPlugin.currentText()
            self._opcua_client = self.mOpcUaProvider.createClient(client_name)
        if not self._opcua_client:
            message = "Connecting to the given server failed. See the log for details."
            self.log(message, "", Qt.red)
            QMessageBox.critical(self, "Failed to connect to server", message)
            return

        self._opcua_client.connectError.connect(self.showErrorDialog)
        self._opcua_client.setApplicationIdentity(self._identity)
        self._opcua_client.setPkiConfiguration(self._pki_config)

        if (QOpcUaUserTokenPolicy.TokenType.Certificate
                in self._opcua_client.supportedUserTokenTypes()):
            authInfo = QOpcUaAuthenticationInformation()
            authInfo.setCertificateAuthentication()
            self._opcua_client.setAuthenticationInformation(authInfo)

        self._opcua_client.connected.connect(self.clientConnected)
        self._opcua_client.disconnected.connect(self.clientDisconnected)
        self._opcua_client.errorChanged.connect(self.clientError)
        self._opcua_client.stateChanged.connect(self.clientState)
        self._opcua_client.endpointsRequestFinished.connect(self.getEndpointsComplete)
        self._opcua_client.findServersFinished.connect(self.findServersComplete)

    @Slot()
    def findServers(self):
        locale_ids = []
        server_uris = []
        url = QUrl(self._ui.host.text())

        self.updateUiState()

        self.createClient()
        # set default port if missing
        if url.port() == -1:
            url.setPort(4840)

        if self._opcua_client:
            self._opcua_client.findServers(url, locale_ids, server_uris)
            qDebug(f"Discovering servers on {url.toString()}")

    @Slot(list, QOpcUa.UaStatusCode)
    def findServersComplete(self, servers, status_code):
        server = QOpcUaApplicationDescription()
        if QOpcUa.isSuccessStatus(status_code):
            self._ui.servers.clear()
            for server in servers:
                self._ui.servers.addItems(server.discoveryUrls())
        self.updateUiState()

    @Slot()
    def getEndpoints(self):
        self._ui.endpoints.clear()
        self.updateUiState()
        if self._ui.servers.currentIndex() >= 0:
            self.createClient()
            self._opcua_client.requestEndpoints(self._ui.servers.currentText())

    @Slot(list, QOpcUa.UaStatusCode)
    def getEndpointsComplete(self, endpoints, status_code):
        index = 0
        if QOpcUa.isSuccessStatus(status_code):
            self._end_point_list = endpoints
            for endpoint in endpoints:
                securityMode = str(endpoint.securityMode())
                securityMode = securityMode[securityMode.index('.') + 1:]
                securityPolicy = endpoint.securityPolicy()
                self._ui.endpoints.addItem(f"{securityPolicy} ({securityMode})", index)
                index = index + 1

        self.updateUiState()

    @Slot()
    def connectToServer(self):
        if self._client_connected:
            self._opcua_client.disconnectFromEndpoint()
            return

        if self._ui.endpoints.currentIndex() >= 0:
            self._endpoint = self._end_point_list[self._ui.endpoints.currentIndex()]
            self.createClient()
            self._opcua_client.connectToEndpoint(self._endpoint)

    @Slot()
    def clientConnected(self):
        self._client_connected = True
        self.updateUiState()

        self._opcua_client.namespaceArrayUpdated.connect(self.namespacesArrayUpdated)
        self._opcua_client.updateNamespaceArray()

    @Slot()
    def clientDisconnected(self):
        self._client_connected = False
        self._opcua_client = None
        self._opcua_model.setOpcUaClient(None)
        self.updateUiState()

    @Slot(list)
    def namespacesArrayUpdated(self, namespace_array):
        if not namespace_array:
            qWarning("Failed to retrieve the namespaces array")
            return

        self._opcua_client.namespaceArrayUpdated.disconnect(self.namespacesArrayUpdated)
        self._opcua_model.setOpcUaClient(self._opcua_client)
        self._ui.treeView.header().setSectionResizeMode(1, QHeaderView.Interactive)

    def clientError(self, error):
        qWarning(f"Client error changed {error}")

    def clientState(self, state):
        qDebug(f"Client state changed {state}")

    def updateUiState(self):
        # allow changing the backend only if it was not already created
        self._ui.opcUaPlugin.setEnabled(not self._opcua_client)
        text = "Disconnect" if self._client_connected else "Connect"
        self._ui.connectButton.setText(text)

        if self._client_connected:
            self._ui.host.setEnabled(False)
            self._ui.servers.setEnabled(False)
            self._ui.endpoints.setEnabled(False)
            self._ui.findServersButton.setEnabled(False)
            self._ui.getEndpointsButton.setEnabled(False)
            self._ui.connectButton.setEnabled(True)
        else:
            self._ui.host.setEnabled(True)
            self._ui.servers.setEnabled(self._ui.servers.count() > 0)
            self._ui.endpoints.setEnabled(self._ui.endpoints.count() > 0)
            self._ui.findServersButton.setDisabled(len(self._ui.host.text()) == 0)
            self._ui.getEndpointsButton.setEnabled(self._ui.servers.currentIndex() != -1)
            self._ui.connectButton.setEnabled(self._ui.endpoints.currentIndex() != -1)

        if not self._opcua_client:
            self._ui.servers.setEnabled(False)
            self._ui.endpoints.setEnabled(False)
            self._ui.getEndpointsButton.setEnabled(False)
            self._ui.connectButton.setEnabled(False)

    @Slot(str, str, QColor)
    def log(self, text, context, color):
        cf = self._ui.log.currentCharFormat()
        cf.setForeground(color)
        self._ui.log.setCurrentCharFormat(cf)
        self._ui.log.appendPlainText(text)
        if context:
            cf.setForeground(Qt.gray)
            self._ui.log.setCurrentCharFormat(cf)
            self._ui.log.insertPlainText(context)

    def createPkiPath(self, path):
        msg = f"Creating PKI path '{path}': "
        dir = QDir()
        ret = dir.mkpath(path)
        if ret:
            qDebug(msg + "SUCCESS.")
        else:
            qWarning(msg + "FAILED.")
        return ret

    def createPkiFolders(self):
        result = self.createPkiPath(self._pki_config.trustListDirectory())
        if not result:
            return result

        result = self.createPkiPath(self._pki_config.revocationListDirectory())
        if not result:
            return result

        result = self.createPkiPath(self._pki_config.issuerListDirectory())
        if not result:
            return result

        result = self.createPkiPath(self._pki_config.issuerRevocationListDirectory())
        if not result:
            return result

        return result

    @Slot(QOpcUaErrorState)
    def showErrorDialog(self, error_state):
        result = 0
        status_code = QOpcUa.statusToString(error_state.errorCode())
        if error_state.isClientSideError():
            msg = "The client reported: "
        else:
            msg = "The server reported: "

        step = error_state.connectionStep()
        if step == QOpcUaErrorState.ConnectionStep.CertificateValidation:
            msg += dedent(
                """
                Server certificate validation failed with error {:04X} ({}).
                Click 'Abort' to abort the connect, or 'Ignore' to continue connecting.
                """).format(error_state.errorCode(), status_code)
            dialog = CertificateDialog(self)
            result = dialog.showCertificate(msg, self._endpoint.serverCertificate(),
                                            self._pki_config.trustListDirectory())
            error_state.setIgnoreError(result == 1)
        elif step == QOpcUaErrorState.ConnectionStep.OpenSecureChannel:
            msg += "OpenSecureChannel failed with error {:04X} ({}).".format(
                error_state.errorCode(), status_code)
            QMessageBox.warning(self, "Connection Error", msg)
        elif step == QOpcUaErrorState.ConnectionStep.CreateSession:
            msg += "CreateSession failed with error {:04X} ({}).".format(
                error_state.errorCode(), status_code)
            QMessageBox.warning(self, "Connection Error", msg)
        elif step == QOpcUaErrorState.ConnectionStep.ActivateSession:
            msg += "ActivateSession failed with error {:04X} ({}).".format(
                error_state.errorCode(), status_code)
            QMessageBox.warning(self, "Connection Error", msg)

    @Slot()
    def showHelpDialog(self):
        dialog = HelpDialog(self)
        dialog.exec()
<?xml version="1.0" encoding="UTF-8"?>
<!-- Copy of the C++ example's UI file with the help action added. -->
<ui version="4.0">
 <class>MainWindow</class>
 <widget class="QMainWindow" name="MainWindow">
  <property name="geometry">
   <rect>
    <x>0</x>
    <y>0</y>
    <width>800</width>
    <height>600</height>
   </rect>
  </property>
  <property name="minimumSize">
   <size>
    <width>800</width>
    <height>0</height>
   </size>
  </property>
  <property name="windowTitle">
   <string>MainWindow</string>
  </property>
  <widget class="QWidget" name="centralwidget">
   <layout class="QGridLayout" name="gridLayout">
    <item row="1" column="0">
     <widget class="QTreeView" name="treeView">
      <property name="alternatingRowColors">
       <bool>true</bool>
      </property>
      <property name="selectionBehavior">
       <enum>QAbstractItemView::SelectItems</enum>
      </property>
     </widget>
    </item>
    <item row="0" column="0">
     <layout class="QVBoxLayout" name="verticalLayout">
      <item>
       <layout class="QGridLayout" name="gridLayout_2">
        <item row="0" column="0">
         <widget class="QLabel" name="label">
          <property name="text">
           <string>Select OPC UA Backend:</string>
          </property>
         </widget>
        </item>
        <item row="1" column="2">
         <widget class="QPushButton" name="findServersButton">
          <property name="text">
           <string>Find Servers</string>
          </property>
         </widget>
        </item>
        <item row="2" column="2">
         <widget class="QPushButton" name="getEndpointsButton">
          <property name="text">
           <string>Get Endpoints</string>
          </property>
         </widget>
        </item>
        <item row="2" column="1">
         <widget class="QComboBox" name="servers"/>
        </item>
        <item row="2" column="0">
         <widget class="QLabel" name="label_3">
          <property name="text">
           <string>Select OPC UA Server:</string>
          </property>
         </widget>
        </item>
        <item row="1" column="1">
         <widget class="QLineEdit" name="host">
          <property name="clearButtonEnabled">
           <bool>true</bool>
          </property>
         </widget>
        </item>
        <item row="0" column="1">
         <widget class="QComboBox" name="opcUaPlugin"/>
        </item>
        <item row="1" column="0">
         <widget class="QLabel" name="label_2">
          <property name="text">
           <string>Select host to discover:</string>
          </property>
         </widget>
        </item>
        <item row="3" column="0">
         <widget class="QLabel" name="label_5">
          <property name="text">
           <string>Select OPC UA Endpoint:</string>
          </property>
         </widget>
        </item>
        <item row="3" column="1">
         <widget class="QComboBox" name="endpoints">
          <property name="sizePolicy">
           <sizepolicy hsizetype="Expanding" vsizetype="Preferred">
            <horstretch>0</horstretch>
            <verstretch>0</verstretch>
           </sizepolicy>
          </property>
         </widget>
        </item>
        <item row="3" column="2">
         <widget class="QPushButton" name="connectButton">
          <property name="text">
           <string>Connect</string>
          </property>
         </widget>
        </item>
       </layout>
      </item>
     </layout>
    </item>
    <item row="2" column="0">
     <widget class="QLabel" name="label_4">
      <property name="text">
       <string>Log:</string>
      </property>
     </widget>
    </item>
    <item row="3" column="0">
     <widget class="QPlainTextEdit" name="log">
      <property name="lineWrapMode">
       <enum>QPlainTextEdit::NoWrap</enum>
      </property>
      <property name="readOnly">
       <bool>true</bool>
      </property>
     </widget>
    </item>
   </layout>
  </widget>
  <widget class="QMenuBar" name="menubar">
   <property name="geometry">
    <rect>
     <x>0</x>
     <y>0</y>
     <width>800</width>
     <height>30</height>
    </rect>
   </property>
   <widget class="QMenu" name="menuFile">
    <property name="title">
     <string>File</string>
    </property>
    <addaction name="quitAction"/>
   </widget>
   <widget class="QMenu" name="menuHelp">
    <property name="title">
     <string>Help</string>
    </property>
    <addaction name="helpAction"/>
    <addaction name="aboutAction"/>
   </widget>
   <addaction name="menuFile"/>
   <addaction name="menuHelp"/>
  </widget>
  <widget class="QStatusBar" name="statusbar"/>
  <action name="quitAction">
   <property name="text">
    <string>Quit</string>
   </property>
  </action>
  <action name="actionHelp">
   <property name="text">
    <string>Help</string>
   </property>
  </action>
  <action name="actionAbout_Qt">
   <property name="text">
    <string>About Qt</string>
   </property>
  </action>
  <action name="helpAction">
   <property name="text">
    <string>Help</string>
   </property>
  </action>
  <action name="aboutAction">
   <property name="text">
    <string>About Qt</string>
   </property>
  </action>
 </widget>
 <resources/>
 <connections/>
</ui>
# Copyright (C) 2021 The Qt Company Ltd.
# Copyright (C) 2018 Unified Automation GmbH
# SPDX-License-Identifier: LicenseRef-Qt-Commercial OR BSD-3-Clause

from treeitem import TreeItem

from PySide6.QtCore import QAbstractItemModel, QModelIndex, Qt


_COLUMN_TITLES = ["BrowseName", "Value", "NodeClass", "DataType", "NodeId",
                  "DisplayName", "Description"]


class OpcUaModel(QAbstractItemModel):
    def __init__(self, parent):
        super(OpcUaModel, self).__init__(parent)
        self._opcua_client = None
        self._root_item = None

    def setOpcUaClient(self, client):
        self.beginResetModel()
        self._opcua_client = client
        if self._opcua_client:
            self._root_item = TreeItem(client.node("ns=0;i=84"), self)
        else:
            self._root_item = None
        self.endResetModel()

    def opcUaClient(self):
        return self._opcua_client

    def data(self, index, role):
        if not index.isValid():
            return None

        item = index.internalPointer()
        if role == Qt.DisplayRole:
            return item.data(index.column())
        if role == Qt.DecorationRole and index.column() == 0:
            return item.icon(index.column())

        return None

    def headerData(self, section, orientation, role):
        if role != Qt.DisplayRole:
            return None

        if orientation == Qt.Vertical:
            return "Row {}".format(section)

        if section < len(_COLUMN_TITLES):
            return _COLUMN_TITLES[section]
        return "Column {}".format(section)

    def index(self, row, column, parent):
        if not self.hasIndex(row, column, parent):
            return QModelIndex()

        item = parent.internalPointer().child(row) if parent.isValid() else self._root_item
        return self.createIndex(row, column, item) if item else QModelIndex()

    def parent(self, index):
        if not index.isValid():
            return QModelIndex()

        child_item = index.internalPointer()
        if child_item == self._root_item:
            return QModelIndex()
        parent_item = child_item.parentItem()
        if not parent_item:
            return QModelIndex()

        return self.createIndex(parent_item.row(), 0, parent_item)

    def rowCount(self, parent):
        if not self._opcua_client or parent.column() > 0:
            return 0

        if not parent.isValid():
            return 1  # only one root item

        parent_item = parent.internalPointer()
        return parent_item.childCount() if parent_item else 0

    def columnCount(self, parent):
        if parent.isValid():
            return parent.internalPointer().columnCount()
        return self._root_item.columnCount() if self._root_item else 0
# Copyright (C) 2021 The Qt Company Ltd.
# Copyright (C) 2018 Unified Automation GmbH
# SPDX-License-Identifier: LicenseRef-Qt-Commercial OR BSD-3-Clause

from PySide6.QtGui import QPixmap
from PySide6.QtCore import Qt, QObject, qWarning
from PySide6.QtOpcUa import QOpcUa

# Columns: 0: NodeId, 1: Value, 2: NodeClass, 3: DataType, 4: BrowseName,
#          5: DisplayName, 6: Description
_numberOfDisplayColumns = 7
_object_pixmap = None
_variable_pixmap = None
_method_pixmap = None
_default_pixmap = None

_DESIRED_ATTRIBUTES = (QOpcUa.NodeAttribute.Value
                       | QOpcUa.NodeAttribute.NodeClass
                       | QOpcUa.NodeAttribute.Description
                       | QOpcUa.NodeAttribute.DataType
                       | QOpcUa.NodeAttribute.BrowseName
                       | QOpcUa.NodeAttribute.DisplayName)

_NODE_CLASSES = {QOpcUa.NodeClass.Undefined: 'Undefined',
                 QOpcUa.NodeClass.Object: 'Object',
                 QOpcUa.NodeClass.Variable: 'Variable',
                 QOpcUa.NodeClass.Method: 'Method',
                 QOpcUa.NodeClass.ObjectType: 'ObjectType',
                 QOpcUa.NodeClass.VariableType: 'VariableType',
                 QOpcUa.NodeClass.ReferenceType: 'ReferenceType',
                 QOpcUa.NodeClass.DataType: 'DataType',
                 QOpcUa.NodeClass.View: 'View'}


def create_pixmap(color):
    p = QPixmap(10, 10)
    p.fill(color)
    return p


class TreeItem(QObject):

    def __init__(self, node, model):
        super(TreeItem, self).__init__(None)

        self._opc_node = node
        self._model = model
        self._attributes_ready = False
        self._browse_started = False
        self._child_items = []
        self._child_node_ids = []
        self._parent_item = None
        self._node_browse_name = ''
        self._node_id = ''
        self._node_display_name = ''
        self._node_class = QOpcUa.NodeClass.Undefined

        self._opc_node.attributeRead.connect(self.handleAttributes)
        self._opc_node.browseFinished.connect(self.browseFinished)

        if not self._opc_node.readAttributes(_DESIRED_ATTRIBUTES):
            qWarning("Reading attributes {} failed".format(self._opc_node.nodeId))

    @staticmethod
    def create_from_browsing_data(node, model, browsingData, parent):
        result = TreeItem(node, model)
        result._parent_item = parent
        result._node_browse_name = browsingData.browseName().name()
        result._node_class = browsingData.nodeClass()
        result._node_id = browsingData.targetNodeId().nodeId()
        result._node_display_name = browsingData.displayName().text()
        return result

    def child(self, row):
        return self._child_items[row]

    def childIndex(self, child):
        return self._child_items.index(child)

    def childCount(self):
        self.startBrowsing()
        return len(self._child_items)

    def columnCount(self):
        return _numberOfDisplayColumns

    def data(self, column):
        if column == 0:
            return self._node_browse_name
        if column == 1:
            if not self._attributes_ready:
                return "Loading ..."
            attribute = self._opc_node.attribute(QOpcUa.NodeAttribute.DataType)
            value = self._opc_node.attribute(QOpcUa.NodeAttribute.Value)
            return str(value) if value else ''
        if column == 2:
            name = _NODE_CLASSES.get(self._node_class)
            return "{} ({})".format(name, self._node_class) if name else str(self._node_class)
        if column == 3:
            if not self._attributes_ready:
                return "Loading ..."
            attribute = self._opc_node.attribute(QOpcUa.NodeAttribute.DataType)
            typeId = attribute if attribute else ''
            enumEntry = QOpcUa.namespace0IdFromNodeId(typeId)
            if enumEntry == QOpcUa.NodeIds.Namespace0.Unknown:
                return typeId
            return "{} ({})".format(QOpcUa.namespace0IdName(enumEntry), typeId)
        if column == 4:
            return self._node_id
        if column == 5:
            return self._node_display_name
        if column == 6:
            if not self._attributes_ready:
                return "Loading ..."
            description = self._opc_node.attribute(QOpcUa.NodeAttribute.Description)
            return description.text() if description else ''
        return None

    def row(self):
        return self._parent_item.childIndex(self) if self._parent_item else 0

    def parentItem(self):
        return self._parent_item

    def appendChild(self, child):
        if not child:
            return
        if not self.hasChildNodeItem(child._node_id):
            self._child_items.append(child)
            self._child_node_ids.append(child._node_id)

    def icon(self, column):
        global _object_pixmap, _variable_pixmap, _method_pixmap, _default_pixmap

        if column != 0 or not self._opc_node:
            return QPixmap()
        if self._node_class == QOpcUa.NodeClass.Object:
            if not _object_pixmap:
                _object_pixmap = create_pixmap(Qt.darkGreen)
            return _object_pixmap
        if self._node_class == QOpcUa.NodeClass.Variable:
            if not _variable_pixmap:
                _variable_pixmap = create_pixmap(Qt.darkBlue)
            return _variable_pixmap
        if self._node_class == QOpcUa.NodeClass.Method:
            if not _method_pixmap:
                _method_pixmap = create_pixmap(Qt.darkRed)
            return _method_pixmap
        if not _default_pixmap:
            _default_pixmap = create_pixmap(Qt.gray)
        return _default_pixmap

    def hasChildNodeItem(self, nodeId):
        return nodeId in self._child_node_ids

    def startBrowsing(self):
        if self._browse_started:
            return

        if not self._opc_node.browseChildren():
            qWarning("Browsing node {} failed".format(self._opc_node.nodeId()))
        else:
            self._browse_started = True

    def handleAttributes(self, attr):
        if attr & QOpcUa.NodeAttribute.NodeClass:
            self._node_class = self._opc_node.attribute(QOpcUa.NodeAttribute.NodeClass)
        if attr & QOpcUa.NodeAttribute.BrowseName:
            name_attr = self._opc_node.attribute(QOpcUa.NodeAttribute.BrowseName)
            self._node_browse_name = name_attr.name()
        if attr & QOpcUa.NodeAttribute.DisplayName:
            display_name_attr = self._opc_node.attribute(QOpcUa.NodeAttribute.DisplayName)
            self._node_display_name = display_name_attr.text()

        self._attributes_ready = True

        row = self.row()
        start_index = self._model.createIndex(row, 0, self)
        end_index = self._model.createIndex(row, _numberOfDisplayColumns - 1, self)
        self._model.dataChanged.emit(start_index, end_index)

    def browseFinished(self, children, status_code):
        if status_code != QOpcUa.Good:
            qWarning("Browsing node {} finally failed: {}".format(
                self._opc_node.nodeId(), status_code))
            return
        row = self.row()
        start_index = self._model.createIndex(row, 0, self)
        for item in children:
            node_id = item.targetNodeId()
            if self.hasChildNodeItem(node_id.nodeId()):
                continue
            node = self._model.opcUaClient().node(node_id)
            if not node:
                qWarning("Failed to instantiate node: {}".format(node_id.nodeId()))
                continue

            child_item_count = len(self._child_items)
            self._model.beginInsertRows(start_index, child_item_count,
                                        child_item_count + 1)
            self.appendChild(TreeItem.create_from_browsing_data(node, self._model, item, self))
            self._model.endInsertRows()

        end_index = self._model.createIndex(row, _numberOfDisplayColumns - 1, self)
        self._model.dataChanged.emit(start_index, end_index)