RefsUI.show_refs() - Code Metrics - Inspection of "more restructure" - FreeOpcUa/opcua-client-gui - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( d1d067...128174 )

by Olivier

created 2016-07-30 07:19 UTC

RefsUI.show_refs() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	2
Bugs	0	Features	0

Metric	Value
cc	2
c	2
b	0
f	0
dl	0
loc	5
rs	9.4285

#! /usr/bin/env python3

import sys
from datetime import datetime
from enum import Enum

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

from opcua import ua

from uaclient.uaclient import UaClient
from uaclient.mainwindow_ui import Ui_MainWindow
from uaclient import resources
from uawidgets.attrs_widget import AttrsWidget
from uawidgets.tree_widget import TreeWidget
from uawidgets.refs_widget import RefsWidget


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)

        # accept drops
        self.model.canDropMimeData = self.canDropMimeData
        self.model.dropMimeData = self.dropMimeData

    def canDropMimeData(self, mdata, action, row, column, parent):
        return True

    def dropMimeData(self, mdata, action, row, column, parent):
        node = self.uaclient.client.get_node(mdata.text())
        print("SUB 1", mdata.text(), node)
        self._subscribe(node)
        return True


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

    def _subscribe(self, node=None):
        print("SUB", node)
        if not node:
            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
        print("Subscribing to events for ", node)
        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)
        
        # accept drops
        self.model.canDropMimeData = self.canDropMimeData
        self.model.dropMimeData = self.dropMimeData

    def canDropMimeData(self, mdata, action, row, column, parent):
        return True

    def dropMimeData(self, mdata, action, row, column, parent):
        node = self.uaclient.client.get_node(mdata.text())
        self._subscribe(node)
        return True

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

    def _subscribe(self, node=None):
        if node is None:
            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"])
        text = str(node.get_display_name().Text)
        row = [QStandardItem(text), 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 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 = TreeWidget(self.ui.treeView)
        self.tree_ui.error.connect(self.show_error)
        self.refs_ui = RefsWidget(self.ui.refView)
        self.refs_ui.error.connect(self.show_error)
        self.attrs_ui = AttrsWidget(self.ui.attrView)
        self.attrs_ui.error.connect(self.show_error)
        self.datachange_ui = DataChangeUI(self, self.uaclient)
        self.event_ui = EventUI(self, self.uaclient)

        self.ui.treeView.activated.connect(self.show_refs)
        self.ui.treeView.clicked.connect(self.show_refs)
        self.ui.actionCopyPath.triggered.connect(self.tree_ui.copy_path)
        self.ui.actionCopyNodeId.triggered.connect(self.tree_ui.copy_nodeid)
        # add items to context menu
        self.ui.treeView.addAction(self.ui.actionCopyPath)
        self.ui.treeView.addAction(self.ui.actionCopyNodeId)


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


        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"", type="QByteArray"))

        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)

        self.ui.modeComboBox.addItem("None")
        self.ui.modeComboBox.addItem("Sign")
        self.ui.modeComboBox.addItem("SignAndEncrypt")

        self.ui.policyComboBox.addItem("None")
        self.ui.policyComboBox.addItem("Basic128RSA15")
        self.ui.policyComboBox.addItem("Basic256")
        self.ui.policyComboBox.addItem("Basic256SHA256")

    def show_refs(self, idx):
        node = self.get_current_node(idx)
        if node:
            self.refs_ui.show_refs(node)
    
    def show_attrs(self, idx):
        if not isinstance(idx, QModelIndex):
            idx = None
        node = self.get_current_node(idx)
        if node:
            self.attrs_ui.show_attrs(node)

    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(self.uaclient.client)

    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()


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, QMimeData
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
13		from uaclient.uaclient import UaClient
14		from uaclient.mainwindow_ui import Ui_MainWindow
15		from uaclient import resources
16		from uawidgets.attrs_widget import AttrsWidget
17		from uawidgets.tree_widget import TreeWidget
18		from uawidgets.refs_widget import RefsWidget
19
20
21		class DataChangeHandler(QObject):
22		data_change_fired = pyqtSignal(object, str, str)
23
24		def datachange_notification(self, node, val, data):
25		if data.monitored_item.Value.SourceTimestamp:
26		dato = data.monitored_item.Value.SourceTimestamp.isoformat()
27		elif data.monitored_item.Value.ServerTimestamp:
28		dato = data.monitored_item.Value.ServerTimestamp.isoformat()
29		else:
30		dato = datetime.now().isoformat()
31		self.data_change_fired.emit(node, str(val), dato)
32
33
34		class EventHandler(QObject):
35		event_fired = pyqtSignal(object)
36
37		def event_notification(self, event):
38		self.event_fired.emit(event)
39
40
41		class EventUI(object):
42
43	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...
44		self.window = window
45		self.uaclient = uaclient
46		self._handler = EventHandler()
47		self._subscribed_nodes = [] # FIXME: not really needed
48		self.model = QStandardItemModel()
49		self.window.ui.evView.setModel(self.model)
50		self.window.ui.actionSubscribeEvent.triggered.connect(self._subscribe)
51		self.window.ui.actionUnsubscribeEvents.triggered.connect(self._unsubscribe)
52		# context menu
53		self.window.ui.treeView.addAction(self.window.ui.actionSubscribeEvent)
54		self.window.ui.treeView.addAction(self.window.ui.actionUnsubscribeEvents)
55		self._handler.event_fired.connect(self._update_event_model, type=Qt.QueuedConnection)
56
57		# accept drops
58		self.model.canDropMimeData = self.canDropMimeData
59		self.model.dropMimeData = self.dropMimeData
60
61		def canDropMimeData(self, mdata, action, row, column, parent):
62		return True
63
64		def dropMimeData(self, mdata, action, row, column, parent):
65		node = self.uaclient.client.get_node(mdata.text())
66		print("SUB 1", mdata.text(), node)
67		self._subscribe(node)
68		return True
69
70
71		def clear(self):
72		self._subscribed_nodes = []
73		self.model.clear()
74
75		def _subscribe(self, node=None):
76		print("SUB", node)
77		if not node:
78		node = self.window.get_current_node()
79		if node is None:
80		return
81		if node in self._subscribed_nodes:
82		print("allready subscribed to event for node: ", node)
83		return
84		print("Subscribing to events for ", node)
85		self.window.ui.evDockWidget.raise_()
86		try:
87		self.uaclient.subscribe_events(node, self._handler)
88		except Exception as ex:
89		self.window.show_error(ex)
90		raise
91		else:
92		self._subscribed_nodes.append(node)
93
94		def _unsubscribe(self):
95		node = self.window.get_current_node()
96		if node is None:
97		return
98		self._subscribed_nodes.remove(node)
99		self.uaclient.unsubscribe_events(node)
100
101		def _update_event_model(self, event):
102		self.model.appendRow([QStandardItem(str(event))])
103
104
105		class DataChangeUI(object):
106
107	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...
108		self.window = window
109		self.uaclient = uaclient
110		self._subhandler = DataChangeHandler()
111		self._subscribed_nodes = []
112		self.model = QStandardItemModel()
113		self.window.ui.subView.setModel(self.model)
114		self.window.ui.subView.horizontalHeader().setSectionResizeMode(1)
115
116		self.window.ui.actionSubscribeDataChange.triggered.connect(self._subscribe)
117		self.window.ui.actionUnsubscribeDataChange.triggered.connect(self._unsubscribe)
118
119		# populate contextual menu
120		self.window.ui.treeView.addAction(self.window.ui.actionSubscribeDataChange)
121		self.window.ui.treeView.addAction(self.window.ui.actionUnsubscribeDataChange)
122
123		# handle subscriptions
124		self._subhandler.data_change_fired.connect(self._update_subscription_model, type=Qt.QueuedConnection)
125
126		# accept drops
127		self.model.canDropMimeData = self.canDropMimeData
128		self.model.dropMimeData = self.dropMimeData
129
130		def canDropMimeData(self, mdata, action, row, column, parent):
131		return True
132
133		def dropMimeData(self, mdata, action, row, column, parent):
134		node = self.uaclient.client.get_node(mdata.text())
135		self._subscribe(node)
136		return True
137
138		def clear(self):
139		self._subscribed_nodes = []
140		self.model.clear()
141
142		def _subscribe(self, node=None):
143		if node is None:
144		node = self.window.get_current_node()
145		if node is None:
146		return
147		if node in self._subscribed_nodes:
148		print("allready subscribed to node: ", node)
149		return
150		self.model.setHorizontalHeaderLabels(["DisplayName", "Value", "Timestamp"])
151		text = str(node.get_display_name().Text)
152		row = [QStandardItem(text), QStandardItem("No Data yet"), QStandardItem("")]
153		row[0].setData(node)
154		self.model.appendRow(row)
155		self._subscribed_nodes.append(node)
156		self.window.ui.subDockWidget.raise_()
157		try:
158		self.uaclient.subscribe_datachange(node, self._subhandler)
159		except Exception as ex:
160		self.window.show_error(ex)
161		idx = self.model.indexFromItem(row[0])
162		self.model.takeRow(idx.row())
163		raise
164
165		def _unsubscribe(self):
166		node = self.window.get_current_node()
167		if node is None:
168		return
169		self.uaclient.unsubscribe_datachange(node)
170		self._subscribed_nodes.remove(node)
171		i = 0
172		while self.model.item(i):
173		item = self.model.item(i)
174		if item.data() == node:
175		self.model.removeRow(i)
176		i += 1
177
178		def _update_subscription_model(self, node, value, timestamp):
179		i = 0
180		while self.model.item(i):
181		item = self.model.item(i)
182		if item.data() == node:
183		it = self.model.item(i, 1)
184		it.setText(value)
185		it_ts = self.model.item(i, 2)
186		it_ts.setText(timestamp)
187		i += 1
188
189
190		class Window(QMainWindow):
191
192		def __init__(self):
193		QMainWindow.__init__(self)
194		self.ui = Ui_MainWindow()
195		self.ui.setupUi(self)
196		self.setWindowIcon(QIcon(":/network.svg"))
197
198		# fix stuff imposible to do in qtdesigner
199		# remove dock titlebar for addressbar
200		w = QWidget()
201		self.ui.addrDockWidget.setTitleBarWidget(w)
202		# tabify some docks
203		self.tabifyDockWidget(self.ui.evDockWidget, self.ui.subDockWidget)
204		self.tabifyDockWidget(self.ui.subDockWidget, self.ui.refDockWidget)
205
206		# we only show statusbar in case of errors
207		self.ui.statusBar.hide()
208
209		# load settings, seconds arg is default
210		self.settings = QSettings("FreeOpcUa", "FreeOpcUaClient")
211		self._address_list = self.settings.value("address_list", ["opc.tcp://localhost:4840", "opc.tcp://localhost:53530/OPCUA/SimulationServer/"])
212		self._address_list_max_count = int(self.settings.value("address_list_max_count", 10))
213
214		# init widgets
215		for addr in self._address_list:
216		self.ui.addrComboBox.insertItem(-1, addr)
217
218		self.uaclient = UaClient()
219
220		self.tree_ui = TreeWidget(self.ui.treeView)
221		self.tree_ui.error.connect(self.show_error)
222		self.refs_ui = RefsWidget(self.ui.refView)
223		self.refs_ui.error.connect(self.show_error)
224		self.attrs_ui = AttrsWidget(self.ui.attrView)
225		self.attrs_ui.error.connect(self.show_error)
226		self.datachange_ui = DataChangeUI(self, self.uaclient)
227		self.event_ui = EventUI(self, self.uaclient)
228
229		self.ui.treeView.activated.connect(self.show_refs)
230		self.ui.treeView.clicked.connect(self.show_refs)
231		self.ui.actionCopyPath.triggered.connect(self.tree_ui.copy_path)
232		self.ui.actionCopyNodeId.triggered.connect(self.tree_ui.copy_nodeid)
233		# add items to context menu
234		self.ui.treeView.addAction(self.ui.actionCopyPath)
235		self.ui.treeView.addAction(self.ui.actionCopyNodeId)
236
237
238		self.ui.treeView.activated.connect(self.show_attrs)
239		self.ui.treeView.clicked.connect(self.show_attrs)
240		self.ui.attrRefreshButton.clicked.connect(self.show_attrs)
241
242
243		self.resize(int(self.settings.value("main_window_width", 800)), int(self.settings.value("main_window_height", 600)))
244		self.restoreState(self.settings.value("main_window_state", b"", type="QByteArray"))
245
246		self.ui.connectButton.clicked.connect(self._connect)
247		self.ui.disconnectButton.clicked.connect(self._disconnect)
248		# self.ui.treeView.expanded.connect(self._fit)
249
250		self.ui.actionConnect.triggered.connect(self._connect)
251		self.ui.actionDisconnect.triggered.connect(self._disconnect)
252
253		self.ui.modeComboBox.addItem("None")
254		self.ui.modeComboBox.addItem("Sign")
255		self.ui.modeComboBox.addItem("SignAndEncrypt")
256
257		self.ui.policyComboBox.addItem("None")
258		self.ui.policyComboBox.addItem("Basic128RSA15")
259		self.ui.policyComboBox.addItem("Basic256")
260		self.ui.policyComboBox.addItem("Basic256SHA256")
261
262		def show_refs(self, idx):
263		node = self.get_current_node(idx)
264		if node:
265		self.refs_ui.show_refs(node)
266
267		def show_attrs(self, idx):
268		if not isinstance(idx, QModelIndex):
269		idx = None
270		node = self.get_current_node(idx)
271		if node:
272		self.attrs_ui.show_attrs(node)
273
274		def show_error(self, msg, level=1):
275		print("showing error: ", msg, level)
276		self.ui.statusBar.show()
277		self.ui.statusBar.setStyleSheet("QStatusBar { background-color : red; color : black; }")
278		self.ui.statusBar.showMessage(str(msg))
279		QTimer.singleShot(1500, self.ui.statusBar.hide)
280
281		def get_current_node(self, idx=None):
282		return self.tree_ui.get_current_node(idx)
283
284		def get_uaclient(self):
285		return self.uaclient
286
287		def _connect(self):
288		uri = self.ui.addrComboBox.currentText()
289		try:
290		self.uaclient.connect(uri)
291		except Exception as ex:
292		self.show_error(ex)
293		raise
294
295		self._update_address_list(uri)
296		self.tree_ui.start(self.uaclient.client)
297
298		def _update_address_list(self, uri):
299		if uri == self._address_list[0]:
300		return
301		if uri in self._address_list:
302		self._address_list.remove(uri)
303		self._address_list.insert(0, uri)
304		if len(self._address_list) > self._address_list_max_count:
305		self._address_list.pop(-1)
306
307		def _disconnect(self):
308		try:
309		self.uaclient.disconnect()
310		except Exception as ex:
311		self.show_error(ex)
312		raise
313		finally:
314		self.tree_ui.clear()
315		self.refs_ui.clear()
316		self.attrs_ui.clear()
317		self.datachange_ui.clear()
318		self.event_ui.clear()
319
320		def closeEvent(self, event):
321		self.settings.setValue("main_window_width", self.size().width())
322		self.settings.setValue("main_window_height", self.size().height())
323		self.settings.setValue("main_window_state", self.saveState())
324		self.settings.setValue("address_list", self._address_list)
325		self._disconnect()
326		event.accept()
327
328
329		def main():
330		app = QApplication(sys.argv)
331		client = Window()
332		client.show()
333		sys.exit(app.exec_())
334
335
336		if __name__ == "__main__":
337		main()
338

FreeOpcUa / opcua-client-gui

Push — master ( d1d067...128174 )

RefsUI.show_refs() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like