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.
# 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)