EventUI._subscribe() - Code Metrics - Inspection of "Allow to write value attribute of variables and pr..." - FreeOpcUa/opcua-client-gui - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( a0ec22...d4a82e )

by Olivier

created 2016-06-19 11:15 UTC

EventUI._subscribe() B

↳ Parent: EventUI

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	5
dl	0
loc	15
rs	8.5454
c	0
b	0
f	0

#! /usr/bin/env python3

import sys
from datetime import datetime
from enum import Enum

from PyQt5.QtCore import pyqtSignal, QTimer, Qt, QObject, QSettings, QModelIndex
from PyQt5.QtGui import QStandardItemModel, QStandardItem, QIcon
from PyQt5.QtWidgets import QMainWindow, QWidget, QApplication, QAbstractItemView, QMenu, QAction

from opcua import ua
from opcua.common.ua_utils import string_to_variant, variant_to_string

from freeopcuaclient.uaclient import UaClient
from freeopcuaclient.mainwindow_ui import Ui_MainWindow
from freeopcuaclient import resources


class DataChangeHandler(QObject):
    data_change_fired = pyqtSignal(object, str, str)

    def datachange_notification(self, node, val, data):
        if data.monitored_item.Value.SourceTimestamp:
            dato = data.monitored_item.Value.SourceTimestamp.isoformat()
        elif data.monitored_item.Value.ServerTimestamp:
            dato = data.monitored_item.Value.ServerTimestamp.isoformat()
        else:
            dato = datetime.now().isoformat()
        self.data_change_fired.emit(node, str(val), dato)


class EventHandler(QObject):
    event_fired = pyqtSignal(object)

    def event_notification(self, event):
        self.event_fired.emit(event)


class EventUI(object):

    def __init__(self, window, uaclient):

        self.window = window
        self.uaclient = uaclient
        self._handler = EventHandler()
        self._subscribed_nodes = []  # FIXME: not really needed
        self.model = QStandardItemModel()
        self.window.ui.evView.setModel(self.model)
        self.window.ui.actionSubscribeEvent.triggered.connect(self._subscribe)
        self.window.ui.actionUnsubscribeEvents.triggered.connect(self._unsubscribe)
        # context menu
        self.window.ui.treeView.addAction(self.window.ui.actionSubscribeEvent)
        self.window.ui.treeView.addAction(self.window.ui.actionUnsubscribeEvents)
        self._handler.event_fired.connect(self._update_event_model, type=Qt.QueuedConnection)

    def clear(self):
        self._subscribed_nodes = []
        self.model.clear()

    def _subscribe(self):
        node = self.window.get_current_node()
        if node is None:
            return
        if node in self._subscribed_nodes:
            print("allready subscribed to event for node: ", node)
            return
        self.window.ui.evDockWidget.raise_()
        try:
            self.uaclient.subscribe_events(node, self._handler)
        except Exception as ex:
            self.window.show_error(ex)
            raise
        else:
            self._subscribed_nodes.append(node)

    def _unsubscribe(self):
        node = self.window.get_current_node()
        if node is None:
            return
        self._subscribed_nodes.remove(node)
        self.uaclient.unsubscribe_events(node)

    def _update_event_model(self, event):
        self.model.appendRow([QStandardItem(str(event))])


class DataChangeUI(object):

    def __init__(self, window, uaclient):

        self.window = window
        self.uaclient = uaclient
        self._subhandler = DataChangeHandler()
        self._subscribed_nodes = []
        self.model = QStandardItemModel()
        self.window.ui.subView.setModel(self.model)
        self.window.ui.subView.horizontalHeader().setSectionResizeMode(1)

        self.window.ui.actionSubscribeDataChange.triggered.connect(self._subscribe)
        self.window.ui.actionUnsubscribeDataChange.triggered.connect(self._unsubscribe)

        # populate contextual menu
        self.window.ui.treeView.addAction(self.window.ui.actionSubscribeDataChange)
        self.window.ui.treeView.addAction(self.window.ui.actionUnsubscribeDataChange)

        # handle subscriptions
        self._subhandler.data_change_fired.connect(self._update_subscription_model, type=Qt.QueuedConnection)

    def clear(self):
        self._subscribed_nodes = []
        self.model.clear()

    def _subscribe(self):
        node = self.window.get_current_node()
        if node is None:
            return
        if node in self._subscribed_nodes:
            print("allready subscribed to node: ", node)
            return
        self.model.setHorizontalHeaderLabels(["DisplayName", "Value", "Timestamp"])
        row = [QStandardItem(node.display_name), QStandardItem("No Data yet"), QStandardItem("")]
        row[0].setData(node)
        self.model.appendRow(row)
        self._subscribed_nodes.append(node)
        self.window.ui.subDockWidget.raise_()
        try:
            self.uaclient.subscribe_datachange(node, self._subhandler)
        except Exception as ex:
            self.window.show_error(ex)
            idx = self.model.indexFromItem(row[0])
            self.model.takeRow(idx.row())
            raise

    def _unsubscribe(self):
        node = self.window.get_current_node()
        if node is None:
            return
        self.uaclient.unsubscribe_datachange(node)
        self._subscribed_nodes.remove(node)
        i = 0
        while self.model.item(i):
            item = self.model.item(i)
            if item.data() == node:
                self.model.removeRow(i)
            i += 1

    def _update_subscription_model(self, node, value, timestamp):
        i = 0
        while self.model.item(i):
            item = self.model.item(i)
            if item.data() == node:
                it = self.model.item(i, 1)
                it.setText(value)
                it_ts = self.model.item(i, 2)
                it_ts.setText(timestamp)
            i += 1


class AttrsUI(object):

    def __init__(self, window, uaclient):

        self.window = window
        self.uaclient = uaclient
        self.model = QStandardItemModel()
        self.window.ui.attrView.setModel(self.model)
        self.window.ui.attrView.doubleClicked.connect(self.edit_attr)
        self.model.itemChanged.connect(self.edit_attr_finished)
        self.window.ui.attrView.header().setSectionResizeMode(1)

        self.window.ui.treeView.activated.connect(self.show_attrs)
        self.window.ui.treeView.clicked.connect(self.show_attrs)
        self.window.ui.attrRefreshButton.clicked.connect(self.show_attrs)

        # Context menu
        self.window.ui.attrView.setContextMenuPolicy(Qt.CustomContextMenu)
        self.window.ui.attrView.customContextMenuRequested.connect(self.showContextMenu)
        copyaction = QAction("&Copy Value", self.model)
        copyaction.triggered.connect(self._copy_value)
        self._contextMenu = QMenu()
        self._contextMenu.addAction(copyaction)

    def edit_attr(self, idx):
        if idx.column() != 1:
            return
        attritem = self.model.item(idx.row(), 0)
        if attritem.text() == "Value":
            self.window.ui.attrView.edit(idx)

    def edit_attr_finished(self, item):
        try:
            var = item.data()
            val = item.text()
            var = string_to_variant(val, var.VariantType)
            self.current_node.set_value(var)
        except Exception as ex:
            self.window.show_error(ex)
            raise
        finally:
            var = self.current_node.get_data_value().Value
            item.setText(variant_to_string(var))

    def showContextMenu(self, position):
        item = self.get_current_item()
        if item:
            self._contextMenu.exec_(self.window.ui.attrView.mapToGlobal(position))

    def get_current_item(self, col_idx=0):
        idx = self.window.ui.attrView.currentIndex()
        return self.model.item(idx.row(), col_idx)

    def _copy_value(self, position):
        it = self.get_current_item(1)
        if it:
            QApplication.clipboard().setText(it.text())

    def clear(self):
        self.model.clear()

    def show_attrs(self, idx):
        if not isinstance(idx, QModelIndex):
            idx = None
        self.current_node = self.window.get_current_node(idx)
        self.model.clear()
        if self.current_node:
            self._show_attrs(self.current_node)

    def _show_attrs(self, node):
        try:
            attrs = self.uaclient.get_all_attrs(node)
        except Exception as ex:
            self.window.show_error(ex)
            raise
        self.model.setHorizontalHeaderLabels(['Attribute', 'Value', 'DataType'])
        for k, variant in attrs.items():
            string = variant_to_string(variant)
            vitem = QStandardItem(string)
            vitem.setData(variant)
            self.model.appendRow([QStandardItem(k), vitem, QStandardItem(variant.VariantType.name)])


class RefsUI(object):

    def __init__(self, window, uaclient):
        self.window = window
        self.uaclient = uaclient
        self.model = QStandardItemModel()
        self.window.ui.refView.setModel(self.model)
        self.window.ui.refView.horizontalHeader().setSectionResizeMode(1)

        self.window.ui.treeView.activated.connect(self.show_refs)
        self.window.ui.treeView.clicked.connect(self.show_refs)

    def clear(self):
        self.model.clear()

    def show_refs(self, idx):
        node = self.window.get_current_node(idx)
        self.model.clear()
        if node:
            self._show_refs(node)

    def _show_refs(self, node):
        self.model.setHorizontalHeaderLabels(['ReferenceType', 'NodeId', "BrowseName", "TypeDefinition"])
        try:
            refs = self.uaclient.get_all_refs(node)
        except Exception as ex:
            self.window.show_error(ex)
            raise
        for ref in refs:
            self.model.appendRow([QStandardItem(str(ref.ReferenceTypeId)),
                                  QStandardItem(str(ref.NodeId)),
                                  QStandardItem(str(ref.BrowseName)),
                                  QStandardItem(str(ref.TypeDefinition))])


class TreeUI(object):

    def __init__(self, window, uaclient):

        self.window = window
        self.uaclient = uaclient
        self.model = TreeViewModel(self.uaclient)
        self.model.clear()  # FIXME: do we need this?
        self.model.error.connect(self.window.show_error)
        self.window.ui.treeView.setModel(self.model)
        self.window.ui.treeView.setUniformRowHeights(True)
        self.window.ui.treeView.setSelectionBehavior(QAbstractItemView.SelectRows)
        self.window.ui.treeView.header().setSectionResizeMode(1)
        self.window.ui.actionCopyPath.triggered.connect(self._copy_path)
        self.window.ui.actionCopyNodeId.triggered.connect(self._copy_nodeid)
        # add items to context menu
        self.window.ui.treeView.addAction(self.window.ui.actionCopyPath)
        self.window.ui.treeView.addAction(self.window.ui.actionCopyNodeId)

    def clear(self):
        self.model.clear()

    def start(self):
        self.model.clear()
        self.model.add_item(*self.uaclient.get_root_node_and_desc())

    def _copy_path(self):
        path = self.get_current_path()
        path = ",".join(path)
        QApplication.clipboard().setText(path)

    def _copy_nodeid(self):
        node = self.get_current_node()
        if node:
            text = node.nodeid.to_string()
        else:
            text = ""
        QApplication.clipboard().setText(text)

    def get_current_path(self):
        idx = self.window.ui.treeView.currentIndex()
        idx = idx.sibling(idx.row(), 0)
        it = self.model.itemFromIndex(idx)
        path = []
        while it and it.data():
            node = it.data()
            name = node.get_browse_name().to_string()
            path.insert(0, name)
            it = it.parent()
        return path

    def get_current_node(self, idx=None):
        if idx is None:
            idx = self.window.ui.treeView.currentIndex()
        idx = idx.sibling(idx.row(), 0)
        it = self.model.itemFromIndex(idx)
        if not it:
            return None
        node = it.data()
        if not node:
            print("No node for item:", it, it.text())
            return None
        node.display_name = it.text()  # FIXME: hack
        return node


class Window(QMainWindow):

    def __init__(self):
        QMainWindow.__init__(self)
        self.ui = Ui_MainWindow()
        self.ui.setupUi(self)
        self.setWindowIcon(QIcon(":/network.svg"))

        # fix stuff imposible to do in qtdesigner
        # remove dock titlebar for addressbar
        w = QWidget()
        self.ui.addrDockWidget.setTitleBarWidget(w)
        # tabify some docks
        self.tabifyDockWidget(self.ui.evDockWidget, self.ui.subDockWidget)
        self.tabifyDockWidget(self.ui.subDockWidget, self.ui.refDockWidget)

        # we only show statusbar in case of errors
        self.ui.statusBar.hide()

        # load settings, seconds arg is default
        self.settings = QSettings("FreeOpcUa", "FreeOpcUaClient")
        self._address_list = self.settings.value("address_list", ["opc.tcp://localhost:4840", "opc.tcp://localhost:53530/OPCUA/SimulationServer/"])
        self._address_list_max_count = int(self.settings.value("address_list_max_count", 10))

        # init widgets
        for addr in self._address_list:
            self.ui.addrComboBox.insertItem(-1, addr)

        self.uaclient = UaClient()

        self.tree_ui = TreeUI(self, self.uaclient)
        self.refs_ui = RefsUI(self, self.uaclient)
        self.attrs_ui = AttrsUI(self, self.uaclient)
        self.datachange_ui = DataChangeUI(self, self.uaclient)
        self.event_ui = EventUI(self, self.uaclient)

        self.resize(int(self.settings.value("main_window_width", 800)), int(self.settings.value("main_window_height", 600)))
        self.restoreState(self.settings.value("main_window_state", b""))

        self.ui.connectButton.clicked.connect(self._connect)
        self.ui.disconnectButton.clicked.connect(self._disconnect)
        # self.ui.treeView.expanded.connect(self._fit)

        self.ui.actionConnect.triggered.connect(self._connect)
        self.ui.actionDisconnect.triggered.connect(self._disconnect)

    def show_error(self, msg, level=1):
        print("showing error: ", msg, level)
        self.ui.statusBar.show()
        self.ui.statusBar.setStyleSheet("QStatusBar { background-color : red; color : black; }")
        self.ui.statusBar.showMessage(str(msg))
        QTimer.singleShot(1500, self.ui.statusBar.hide)

    def get_current_node(self, idx=None):
        return self.tree_ui.get_current_node(idx)

    def get_uaclient(self):
        return self.uaclient

    def _connect(self):
        uri = self.ui.addrComboBox.currentText()
        try:
            self.uaclient.connect(uri)
        except Exception as ex:
            self.show_error(ex)
            raise

        self._update_address_list(uri)
        self.tree_ui.start()

    def _update_address_list(self, uri):
        if uri == self._address_list[0]:
            return
        if uri in self._address_list:
            self._address_list.remove(uri)
        self._address_list.insert(0, uri)
        if len(self._address_list) > self._address_list_max_count:
            self._address_list.pop(-1)

    def _disconnect(self):
        try:
            self.uaclient.disconnect()
        except Exception as ex:
            self.show_error(ex)
            raise
        finally:
            self.tree_ui.clear()
            self.refs_ui.clear()
            self.attrs_ui.clear()
            self.datachange_ui.clear()
            self.event_ui.clear()

    def closeEvent(self, event):
        self.settings.setValue("main_window_width", self.size().width())
        self.settings.setValue("main_window_height", self.size().height())
        self.settings.setValue("main_window_state", self.saveState())
        self.settings.setValue("address_list", self._address_list)
        self._disconnect()
        event.accept()


class TreeViewModel(QStandardItemModel):

    error = pyqtSignal(str)

    def __init__(self, uaclient):
        super(TreeViewModel, self).__init__()
        self.uaclient = uaclient
        self._fetched = []

    def clear(self):
        QStandardItemModel.clear(self)
        self._fetched = []
        self.setHorizontalHeaderLabels(['DisplayName', "BrowseName", 'NodeId'])

    def add_item(self, node, desc, parent=None):
        item = [QStandardItem(desc.DisplayName.to_string()), QStandardItem(desc.BrowseName.to_string()), QStandardItem(desc.NodeId.to_string())]
        if desc.NodeClass == ua.NodeClass.Object:
            if desc.TypeDefinition == ua.TwoByteNodeId(ua.ObjectIds.FolderType):
                item[0].setIcon(QIcon(":/folder.svg"))
            else:
                item[0].setIcon(QIcon(":/object.svg"))
        elif desc.NodeClass == ua.NodeClass.Variable:
            if desc.TypeDefinition == ua.TwoByteNodeId(ua.ObjectIds.PropertyType):
                item[0].setIcon(QIcon(":/property.svg"))
            else:
                item[0].setIcon(QIcon(":/variable.svg"))
        elif desc.NodeClass == ua.NodeClass.Method:
                item[0].setIcon(QIcon(":/method.svg"))

        item[0].setData(node)
        if parent:
            return parent.appendRow(item)
        else:
            return self.appendRow(item)

    def canFetchMore(self, idx):
        item = self.itemFromIndex(idx)
        if not item:
            return True
        node = item.data()
        if node not in self._fetched:
            self._fetched.append(node)
            return True
        return False

    def hasChildren(self, idx):
        item = self.itemFromIndex(idx)
        if not item:
            return True
        node = item.data()
        if node in self._fetched:
            return QStandardItemModel.hasChildren(self, idx)
        return True

    def fetchMore(self, idx):
        parent = self.itemFromIndex(idx)
        if parent:
            self._fetchMore(parent)

    def _fetchMore(self, parent):
        try:
            for node, attrs in self.uaclient.get_children(parent.data()):
                self.add_item(node, attrs, parent)
        except Exception as ex:
            self.error.emit(ex)
            raise


def main():
    app = QApplication(sys.argv)
    client = Window()
    client.show()
    sys.exit(app.exec_())


if __name__ == "__main__":
    main()


1		#! /usr/bin/env python3
2
3		import sys
4		from datetime import datetime
5		from enum import Enum
6
7		from PyQt5.QtCore import pyqtSignal, QTimer, Qt, QObject, QSettings, QModelIndex
8		from PyQt5.QtGui import QStandardItemModel, QStandardItem, QIcon
9		from PyQt5.QtWidgets import QMainWindow, QWidget, QApplication, QAbstractItemView, QMenu, QAction
10
11		from opcua import ua
12		from opcua.common.ua_utils import string_to_variant, variant_to_string
13
14		from freeopcuaclient.uaclient import UaClient
15		from freeopcuaclient.mainwindow_ui import Ui_MainWindow
16		from freeopcuaclient import resources
17
18
19		class DataChangeHandler(QObject):
20		data_change_fired = pyqtSignal(object, str, str)
21
22		def datachange_notification(self, node, val, data):
23		if data.monitored_item.Value.SourceTimestamp:
24		dato = data.monitored_item.Value.SourceTimestamp.isoformat()
25		elif data.monitored_item.Value.ServerTimestamp:
26		dato = data.monitored_item.Value.ServerTimestamp.isoformat()
27		else:
28		dato = datetime.now().isoformat()
29		self.data_change_fired.emit(node, str(val), dato)
30
31
32		class EventHandler(QObject):
33		event_fired = pyqtSignal(object)
34
35		def event_notification(self, event):
36		self.event_fired.emit(event)
37
38
39		class EventUI(object):
40
41	View Code Duplication	def __init__(self, window, uaclient):
		0 ignored issues – show Duplication introduced 2016-06-19 11:16 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
42		self.window = window
43		self.uaclient = uaclient
44		self._handler = EventHandler()
45		self._subscribed_nodes = [] # FIXME: not really needed
46		self.model = QStandardItemModel()
47		self.window.ui.evView.setModel(self.model)
48		self.window.ui.actionSubscribeEvent.triggered.connect(self._subscribe)
49		self.window.ui.actionUnsubscribeEvents.triggered.connect(self._unsubscribe)
50		# context menu
51		self.window.ui.treeView.addAction(self.window.ui.actionSubscribeEvent)
52		self.window.ui.treeView.addAction(self.window.ui.actionUnsubscribeEvents)
53		self._handler.event_fired.connect(self._update_event_model, type=Qt.QueuedConnection)
54
55		def clear(self):
56		self._subscribed_nodes = []
57		self.model.clear()
58
59		def _subscribe(self):
60		node = self.window.get_current_node()
61		if node is None:
62		return
63		if node in self._subscribed_nodes:
64		print("allready subscribed to event for node: ", node)
65		return
66		self.window.ui.evDockWidget.raise_()
67		try:
68		self.uaclient.subscribe_events(node, self._handler)
69		except Exception as ex:
70		self.window.show_error(ex)
71		raise
72		else:
73		self._subscribed_nodes.append(node)
74
75		def _unsubscribe(self):
76		node = self.window.get_current_node()
77		if node is None:
78		return
79		self._subscribed_nodes.remove(node)
80		self.uaclient.unsubscribe_events(node)
81
82		def _update_event_model(self, event):
83		self.model.appendRow([QStandardItem(str(event))])
84
85
86		class DataChangeUI(object):
87
88	View Code Duplication	def __init__(self, window, uaclient):
		0 ignored issues – show Duplication introduced 2016-06-19 11:16 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
89		self.window = window
90		self.uaclient = uaclient
91		self._subhandler = DataChangeHandler()
92		self._subscribed_nodes = []
93		self.model = QStandardItemModel()
94		self.window.ui.subView.setModel(self.model)
95		self.window.ui.subView.horizontalHeader().setSectionResizeMode(1)
96
97		self.window.ui.actionSubscribeDataChange.triggered.connect(self._subscribe)
98		self.window.ui.actionUnsubscribeDataChange.triggered.connect(self._unsubscribe)
99
100		# populate contextual menu
101		self.window.ui.treeView.addAction(self.window.ui.actionSubscribeDataChange)
102		self.window.ui.treeView.addAction(self.window.ui.actionUnsubscribeDataChange)
103
104		# handle subscriptions
105		self._subhandler.data_change_fired.connect(self._update_subscription_model, type=Qt.QueuedConnection)
106
107		def clear(self):
108		self._subscribed_nodes = []
109		self.model.clear()
110
111		def _subscribe(self):
112		node = self.window.get_current_node()
113		if node is None:
114		return
115		if node in self._subscribed_nodes:
116		print("allready subscribed to node: ", node)
117		return
118		self.model.setHorizontalHeaderLabels(["DisplayName", "Value", "Timestamp"])
119		row = [QStandardItem(node.display_name), QStandardItem("No Data yet"), QStandardItem("")]
120		row[0].setData(node)
121		self.model.appendRow(row)
122		self._subscribed_nodes.append(node)
123		self.window.ui.subDockWidget.raise_()
124		try:
125		self.uaclient.subscribe_datachange(node, self._subhandler)
126		except Exception as ex:
127		self.window.show_error(ex)
128		idx = self.model.indexFromItem(row[0])
129		self.model.takeRow(idx.row())
130		raise
131
132		def _unsubscribe(self):
133		node = self.window.get_current_node()
134		if node is None:
135		return
136		self.uaclient.unsubscribe_datachange(node)
137		self._subscribed_nodes.remove(node)
138		i = 0
139		while self.model.item(i):
140		item = self.model.item(i)
141		if item.data() == node:
142		self.model.removeRow(i)
143		i += 1
144
145		def _update_subscription_model(self, node, value, timestamp):
146		i = 0
147		while self.model.item(i):
148		item = self.model.item(i)
149		if item.data() == node:
150		it = self.model.item(i, 1)
151		it.setText(value)
152		it_ts = self.model.item(i, 2)
153		it_ts.setText(timestamp)
154		i += 1
155
156
157		class AttrsUI(object):
158
159	View Code Duplication	def __init__(self, window, uaclient):
		0 ignored issues – show Duplication introduced 2016-06-19 11:16 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
160		self.window = window
161		self.uaclient = uaclient
162		self.model = QStandardItemModel()
163		self.window.ui.attrView.setModel(self.model)
164		self.window.ui.attrView.doubleClicked.connect(self.edit_attr)
165		self.model.itemChanged.connect(self.edit_attr_finished)
166		self.window.ui.attrView.header().setSectionResizeMode(1)
167
168		self.window.ui.treeView.activated.connect(self.show_attrs)
169		self.window.ui.treeView.clicked.connect(self.show_attrs)
170		self.window.ui.attrRefreshButton.clicked.connect(self.show_attrs)
171
172		# Context menu
173		self.window.ui.attrView.setContextMenuPolicy(Qt.CustomContextMenu)
174		self.window.ui.attrView.customContextMenuRequested.connect(self.showContextMenu)
175		copyaction = QAction("&Copy Value", self.model)
176		copyaction.triggered.connect(self._copy_value)
177		self._contextMenu = QMenu()
178		self._contextMenu.addAction(copyaction)
179
180		def edit_attr(self, idx):
181		if idx.column() != 1:
182		return
183		attritem = self.model.item(idx.row(), 0)
184		if attritem.text() == "Value":
185		self.window.ui.attrView.edit(idx)
186
187		def edit_attr_finished(self, item):
188		try:
189		var = item.data()
190		val = item.text()
191		var = string_to_variant(val, var.VariantType)
192		self.current_node.set_value(var)
193		except Exception as ex:
194		self.window.show_error(ex)
195		raise
196		finally:
197		var = self.current_node.get_data_value().Value
198		item.setText(variant_to_string(var))
199
200		def showContextMenu(self, position):
201		item = self.get_current_item()
202		if item:
203		self._contextMenu.exec_(self.window.ui.attrView.mapToGlobal(position))
204
205		def get_current_item(self, col_idx=0):
206		idx = self.window.ui.attrView.currentIndex()
207		return self.model.item(idx.row(), col_idx)
208
209		def _copy_value(self, position):
210		it = self.get_current_item(1)
211		if it:
212		QApplication.clipboard().setText(it.text())
213
214		def clear(self):
215		self.model.clear()
216
217		def show_attrs(self, idx):
218		if not isinstance(idx, QModelIndex):
219		idx = None
220		self.current_node = self.window.get_current_node(idx)
221		self.model.clear()
222		if self.current_node:
223		self._show_attrs(self.current_node)
224
225		def _show_attrs(self, node):
226		try:
227		attrs = self.uaclient.get_all_attrs(node)
228		except Exception as ex:
229		self.window.show_error(ex)
230		raise
231		self.model.setHorizontalHeaderLabels(['Attribute', 'Value', 'DataType'])
232		for k, variant in attrs.items():
233		string = variant_to_string(variant)
234		vitem = QStandardItem(string)
235		vitem.setData(variant)
236		self.model.appendRow([QStandardItem(k), vitem, QStandardItem(variant.VariantType.name)])
237
238
239		class RefsUI(object):
240
241		def __init__(self, window, uaclient):
242		self.window = window
243		self.uaclient = uaclient
244		self.model = QStandardItemModel()
245		self.window.ui.refView.setModel(self.model)
246		self.window.ui.refView.horizontalHeader().setSectionResizeMode(1)
247
248		self.window.ui.treeView.activated.connect(self.show_refs)
249		self.window.ui.treeView.clicked.connect(self.show_refs)
250
251		def clear(self):
252		self.model.clear()
253
254		def show_refs(self, idx):
255		node = self.window.get_current_node(idx)
256		self.model.clear()
257		if node:
258		self._show_refs(node)
259
260		def _show_refs(self, node):
261		self.model.setHorizontalHeaderLabels(['ReferenceType', 'NodeId', "BrowseName", "TypeDefinition"])
262		try:
263		refs = self.uaclient.get_all_refs(node)
264		except Exception as ex:
265		self.window.show_error(ex)
266		raise
267		for ref in refs:
268		self.model.appendRow([QStandardItem(str(ref.ReferenceTypeId)),
269		QStandardItem(str(ref.NodeId)),
270		QStandardItem(str(ref.BrowseName)),
271		QStandardItem(str(ref.TypeDefinition))])
272
273
274		class TreeUI(object):
275
276	View Code Duplication	def __init__(self, window, uaclient):
		0 ignored issues – show Duplication introduced 2016-06-19 11:16 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
277		self.window = window
278		self.uaclient = uaclient
279		self.model = TreeViewModel(self.uaclient)
280		self.model.clear() # FIXME: do we need this?
281		self.model.error.connect(self.window.show_error)
282		self.window.ui.treeView.setModel(self.model)
283		self.window.ui.treeView.setUniformRowHeights(True)
284		self.window.ui.treeView.setSelectionBehavior(QAbstractItemView.SelectRows)
285		self.window.ui.treeView.header().setSectionResizeMode(1)
286		self.window.ui.actionCopyPath.triggered.connect(self._copy_path)
287		self.window.ui.actionCopyNodeId.triggered.connect(self._copy_nodeid)
288		# add items to context menu
289		self.window.ui.treeView.addAction(self.window.ui.actionCopyPath)
290		self.window.ui.treeView.addAction(self.window.ui.actionCopyNodeId)
291
292		def clear(self):
293		self.model.clear()
294
295		def start(self):
296		self.model.clear()
297		self.model.add_item(*self.uaclient.get_root_node_and_desc())
298
299		def _copy_path(self):
300		path = self.get_current_path()
301		path = ",".join(path)
302		QApplication.clipboard().setText(path)
303
304		def _copy_nodeid(self):
305		node = self.get_current_node()
306		if node:
307		text = node.nodeid.to_string()
308		else:
309		text = ""
310		QApplication.clipboard().setText(text)
311
312		def get_current_path(self):
313		idx = self.window.ui.treeView.currentIndex()
314		idx = idx.sibling(idx.row(), 0)
315		it = self.model.itemFromIndex(idx)
316		path = []
317		while it and it.data():
318		node = it.data()
319		name = node.get_browse_name().to_string()
320		path.insert(0, name)
321		it = it.parent()
322		return path
323
324		def get_current_node(self, idx=None):
325		if idx is None:
326		idx = self.window.ui.treeView.currentIndex()
327		idx = idx.sibling(idx.row(), 0)
328		it = self.model.itemFromIndex(idx)
329		if not it:
330		return None
331		node = it.data()
332		if not node:
333		print("No node for item:", it, it.text())
334		return None
335		node.display_name = it.text() # FIXME: hack
336		return node
337
338
339		class Window(QMainWindow):
340
341		def __init__(self):
342		QMainWindow.__init__(self)
343		self.ui = Ui_MainWindow()
344		self.ui.setupUi(self)
345		self.setWindowIcon(QIcon(":/network.svg"))
346
347		# fix stuff imposible to do in qtdesigner
348		# remove dock titlebar for addressbar
349		w = QWidget()
350		self.ui.addrDockWidget.setTitleBarWidget(w)
351		# tabify some docks
352		self.tabifyDockWidget(self.ui.evDockWidget, self.ui.subDockWidget)
353		self.tabifyDockWidget(self.ui.subDockWidget, self.ui.refDockWidget)
354
355		# we only show statusbar in case of errors
356		self.ui.statusBar.hide()
357
358		# load settings, seconds arg is default
359		self.settings = QSettings("FreeOpcUa", "FreeOpcUaClient")
360		self._address_list = self.settings.value("address_list", ["opc.tcp://localhost:4840", "opc.tcp://localhost:53530/OPCUA/SimulationServer/"])
361		self._address_list_max_count = int(self.settings.value("address_list_max_count", 10))
362
363		# init widgets
364		for addr in self._address_list:
365		self.ui.addrComboBox.insertItem(-1, addr)
366
367		self.uaclient = UaClient()
368
369		self.tree_ui = TreeUI(self, self.uaclient)
370		self.refs_ui = RefsUI(self, self.uaclient)
371		self.attrs_ui = AttrsUI(self, self.uaclient)
372		self.datachange_ui = DataChangeUI(self, self.uaclient)
373		self.event_ui = EventUI(self, self.uaclient)
374
375		self.resize(int(self.settings.value("main_window_width", 800)), int(self.settings.value("main_window_height", 600)))
376		self.restoreState(self.settings.value("main_window_state", b""))
377
378		self.ui.connectButton.clicked.connect(self._connect)
379		self.ui.disconnectButton.clicked.connect(self._disconnect)
380		# self.ui.treeView.expanded.connect(self._fit)
381
382		self.ui.actionConnect.triggered.connect(self._connect)
383		self.ui.actionDisconnect.triggered.connect(self._disconnect)
384
385		def show_error(self, msg, level=1):
386		print("showing error: ", msg, level)
387		self.ui.statusBar.show()
388		self.ui.statusBar.setStyleSheet("QStatusBar { background-color : red; color : black; }")
389		self.ui.statusBar.showMessage(str(msg))
390		QTimer.singleShot(1500, self.ui.statusBar.hide)
391
392		def get_current_node(self, idx=None):
393		return self.tree_ui.get_current_node(idx)
394
395		def get_uaclient(self):
396		return self.uaclient
397
398		def _connect(self):
399		uri = self.ui.addrComboBox.currentText()
400		try:
401		self.uaclient.connect(uri)
402		except Exception as ex:
403		self.show_error(ex)
404		raise
405
406		self._update_address_list(uri)
407		self.tree_ui.start()
408
409		def _update_address_list(self, uri):
410		if uri == self._address_list[0]:
411		return
412		if uri in self._address_list:
413		self._address_list.remove(uri)
414		self._address_list.insert(0, uri)
415		if len(self._address_list) > self._address_list_max_count:
416		self._address_list.pop(-1)
417
418		def _disconnect(self):
419		try:
420		self.uaclient.disconnect()
421		except Exception as ex:
422		self.show_error(ex)
423		raise
424		finally:
425		self.tree_ui.clear()
426		self.refs_ui.clear()
427		self.attrs_ui.clear()
428		self.datachange_ui.clear()
429		self.event_ui.clear()
430
431		def closeEvent(self, event):
432		self.settings.setValue("main_window_width", self.size().width())
433		self.settings.setValue("main_window_height", self.size().height())
434		self.settings.setValue("main_window_state", self.saveState())
435		self.settings.setValue("address_list", self._address_list)
436		self._disconnect()
437		event.accept()
438
439
440		class TreeViewModel(QStandardItemModel):
441
442		error = pyqtSignal(str)
443
444		def __init__(self, uaclient):
445		super(TreeViewModel, self).__init__()
446		self.uaclient = uaclient
447		self._fetched = []
448
449		def clear(self):
450		QStandardItemModel.clear(self)
451		self._fetched = []
452		self.setHorizontalHeaderLabels(['DisplayName', "BrowseName", 'NodeId'])
453
454		def add_item(self, node, desc, parent=None):
455		item = [QStandardItem(desc.DisplayName.to_string()), QStandardItem(desc.BrowseName.to_string()), QStandardItem(desc.NodeId.to_string())]
456		if desc.NodeClass == ua.NodeClass.Object:
457		if desc.TypeDefinition == ua.TwoByteNodeId(ua.ObjectIds.FolderType):
458		item[0].setIcon(QIcon(":/folder.svg"))
459		else:
460		item[0].setIcon(QIcon(":/object.svg"))
461		elif desc.NodeClass == ua.NodeClass.Variable:
462		if desc.TypeDefinition == ua.TwoByteNodeId(ua.ObjectIds.PropertyType):
463		item[0].setIcon(QIcon(":/property.svg"))
464		else:
465		item[0].setIcon(QIcon(":/variable.svg"))
466		elif desc.NodeClass == ua.NodeClass.Method:
467		item[0].setIcon(QIcon(":/method.svg"))
468
469		item[0].setData(node)
470		if parent:
471		return parent.appendRow(item)
472		else:
473		return self.appendRow(item)
474
475		def canFetchMore(self, idx):
476		item = self.itemFromIndex(idx)
477		if not item:
478		return True
479		node = item.data()
480		if node not in self._fetched:
481		self._fetched.append(node)
482		return True
483		return False
484
485		def hasChildren(self, idx):
486		item = self.itemFromIndex(idx)
487		if not item:
488		return True
489		node = item.data()
490		if node in self._fetched:
491		return QStandardItemModel.hasChildren(self, idx)
492		return True
493
494		def fetchMore(self, idx):
495		parent = self.itemFromIndex(idx)
496		if parent:
497		self._fetchMore(parent)
498
499		def _fetchMore(self, parent):
500		try:
501		for node, attrs in self.uaclient.get_children(parent.data()):
502		self.add_item(node, attrs, parent)
503		except Exception as ex:
504		self.error.emit(ex)
505		raise
506
507
508		def main():
509		app = QApplication(sys.argv)
510		client = Window()
511		client.show()
512		sys.exit(app.exec_())
513
514
515		if __name__ == "__main__":
516		main()
517

FreeOpcUa / opcua-client-gui

Push — master ( a0ec22...d4a82e )

EventUI._subscribe() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like