Window.save_current_node() - Code Metrics - Inspection of "require minimum version of opcua and uawidgets" - FreeOpcUa/opcua-client-gui - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( e7ced7...6f1de5 )

by Olivier

created 2018-03-31 06:26 UTC

Window.save_current_node() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
cc	3
c	1
b	0
f	0
dl	0
loc	9
rs	9.6666

#! /usr/bin/env python3

import sys

from datetime import datetime
import inspect
from enum import Enum
import logging

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

from opcua import ua
from opcua import Node

from uaclient.uaclient import UaClient
from uaclient.mainwindow_ui import Ui_MainWindow
from uaclient.connection_dialog import ConnectionDialog
from uaclient.graphwidget import GraphUI

from uawidgets import resources
from uawidgets.attrs_widget import AttrsWidget
from uawidgets.tree_widget import TreeWidget
from uawidgets.refs_widget import RefsWidget
from uawidgets.utils import trycatchslot
from uawidgets.logger import QtHandler


logger = logging.getLogger(__name__)


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 show_error(self, *args):
        self.window.show_error(*args)

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

    @trycatchslot
    def _subscribe(self, node=None):
        logger.info("Subscribing to %s", node)
        if not node:
            node = self.window.get_current_node()
            if node is None:
                return
        if node in self._subscribed_nodes:
            logger.info("allready subscribed to event for node: %s", node)
            return
        logger.info("Subscribing to events for %s", 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)

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

    @trycatchslot
    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 show_error(self, *args):
        self.window.show_error(*args)

    @trycatchslot
    def _subscribe(self, node=None):
        if not isinstance(node, Node):
            node = self.window.get_current_node()
            if node is None:
                return
        if node in self._subscribed_nodes:
            logger.warning("allready subscribed to node: %s ", 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

    @trycatchslot
    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)
        self.tabifyDockWidget(self.ui.refDockWidget, self.ui.graphDockWidget)

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

        # setup QSettings for application and get a settings object
        QCoreApplication.setOrganizationName("FreeOpcUa")
        QCoreApplication.setApplicationName("OpcUaClient")
        self.settings = QSettings()

        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.graph_ui = GraphUI(self, self.uaclient)

        self.ui.addrComboBox.currentTextChanged.connect(self._uri_changed)
        self._uri_changed(self.ui.addrComboBox.currentText())  # force update for current value at startup

        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)))
        data = self.settings.value("main_window_state", None)
        if data:
            self.restoreState(data)

        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.connectOptionButton.clicked.connect(self.show_connection_dialog)

    def _uri_changed(self, uri):
        self.uaclient.load_security_settings(uri)

    def show_connection_dialog(self):
        dia = ConnectionDialog(self, self.ui.addrComboBox.currentText())
        dia.security_mode = self.uaclient.security_mode
        dia.security_policy = self.uaclient.security_policy
        dia.certificate_path = self.uaclient.certificate_path
        dia.private_key_path = self.uaclient.private_key_path
        ret = dia.exec_()
        if ret:
            self.uaclient.security_mode = dia.security_mode
            self.uaclient.security_policy = dia.security_policy
            self.uaclient.certificate_path = dia.certificate_path
            self.uaclient.private_key_path = dia.private_key_path

    @trycatchslot
    def show_refs(self, idx):
        node = self.get_current_node(idx)
        if node:
            self.refs_ui.show_refs(node)
    
    @trycatchslot
    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):
        logger.warning("showing error: %s")
        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

    @trycatchslot
    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.set_root_node(self.uaclient.client.get_root_node())
        self.load_current_node()

    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.save_current_node()
            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.tree_ui.save_state()
        self.attrs_ui.save_state()
        self.refs_ui.save_state()
        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 save_current_node(self):
        current_node = self.tree_ui.get_current_node()
        if current_node:
            mysettings = self.settings.value("current_node", None)
            if mysettings is None:
                mysettings = {}
            uri = self.ui.addrComboBox.currentText()
            mysettings[uri] = current_node.nodeid.to_string()
            self.settings.setValue("current_node", mysettings)

    def load_current_node(self):
        mysettings = self.settings.value("current_node", None)
        if mysettings is None:
            return
        uri = self.ui.addrComboBox.currentText()
        if uri in mysettings:
            nodeid = ua.NodeId.from_string(mysettings[uri])
            node = self.uaclient.client.get_node(nodeid)
            self.tree_ui.expand_to_node(node)


def main():
    app = QApplication(sys.argv)
    client = Window()
    handler = QtHandler(client.ui.logTextEdit)
    logging.getLogger().addHandler(handler)
    logging.getLogger("uaclient").setLevel(logging.INFO)
    logging.getLogger("uawidgets").setLevel(logging.INFO)
    #logging.getLogger("opcua").setLevel(logging.INFO)  # to enable logging of ua server
   
    client.show()
    sys.exit(app.exec_())


if __name__ == "__main__":
    main()


1		#! /usr/bin/env python3
2
3		import sys
4
5		from datetime import datetime
6		import inspect
7		from enum import Enum
8		import logging
9
10		from PyQt5.QtCore import pyqtSignal, QTimer, Qt, QObject, QSettings, QModelIndex, QMimeData, QCoreApplication
11		from PyQt5.QtGui import QStandardItemModel, QStandardItem, QIcon
12		from PyQt5.QtWidgets import QMainWindow, QWidget, QApplication, QAbstractItemView, QMenu, QAction
13
14		from opcua import ua
15		from opcua import Node
16
17		from uaclient.uaclient import UaClient
18		from uaclient.mainwindow_ui import Ui_MainWindow
19		from uaclient.connection_dialog import ConnectionDialog
20		from uaclient.graphwidget import GraphUI
21
22		from uawidgets import resources
23		from uawidgets.attrs_widget import AttrsWidget
24		from uawidgets.tree_widget import TreeWidget
25		from uawidgets.refs_widget import RefsWidget
26		from uawidgets.utils import trycatchslot
27		from uawidgets.logger import QtHandler
28
29
30		logger = logging.getLogger(__name__)
31
32
33		class DataChangeHandler(QObject):
34		data_change_fired = pyqtSignal(object, str, str)
35
36		def datachange_notification(self, node, val, data):
37		if data.monitored_item.Value.SourceTimestamp:
38		dato = data.monitored_item.Value.SourceTimestamp.isoformat()
39		elif data.monitored_item.Value.ServerTimestamp:
40		dato = data.monitored_item.Value.ServerTimestamp.isoformat()
41		else:
42		dato = datetime.now().isoformat()
43		self.data_change_fired.emit(node, str(val), dato)
44
45
46		class EventHandler(QObject):
47		event_fired = pyqtSignal(object)
48
49		def event_notification(self, event):
50		self.event_fired.emit(event)
51
52
53		class EventUI(object):
54
55	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...
56		self.window = window
57		self.uaclient = uaclient
58		self._handler = EventHandler()
59		self._subscribed_nodes = [] # FIXME: not really needed
60		self.model = QStandardItemModel()
61		self.window.ui.evView.setModel(self.model)
62		self.window.ui.actionSubscribeEvent.triggered.connect(self._subscribe)
63		self.window.ui.actionUnsubscribeEvents.triggered.connect(self._unsubscribe)
64		# context menu
65		self.window.ui.treeView.addAction(self.window.ui.actionSubscribeEvent)
66		self.window.ui.treeView.addAction(self.window.ui.actionUnsubscribeEvents)
67		self._handler.event_fired.connect(self._update_event_model, type=Qt.QueuedConnection)
68
69		# accept drops
70		self.model.canDropMimeData = self.canDropMimeData
71		self.model.dropMimeData = self.dropMimeData
72
73		def canDropMimeData(self, mdata, action, row, column, parent):
74		return True
75
76		def show_error(self, *args):
77		self.window.show_error(*args)
78
79		def dropMimeData(self, mdata, action, row, column, parent):
80		node = self.uaclient.client.get_node(mdata.text())
81		self._subscribe(node)
82		return True
83
84		def clear(self):
85		self._subscribed_nodes = []
86		self.model.clear()
87
88		@trycatchslot
89		def _subscribe(self, node=None):
90		logger.info("Subscribing to %s", node)
91		if not node:
92		node = self.window.get_current_node()
93		if node is None:
94		return
95		if node in self._subscribed_nodes:
96		logger.info("allready subscribed to event for node: %s", node)
97		return
98		logger.info("Subscribing to events for %s", node)
99		self.window.ui.evDockWidget.raise_()
100		try:
101		self.uaclient.subscribe_events(node, self._handler)
102		except Exception as ex:
103		self.window.show_error(ex)
104		raise
105		else:
106		self._subscribed_nodes.append(node)
107
108		@trycatchslot
109		def _unsubscribe(self):
110		node = self.window.get_current_node()
111		if node is None:
112		return
113		self._subscribed_nodes.remove(node)
114		self.uaclient.unsubscribe_events(node)
115
116		@trycatchslot
117		def _update_event_model(self, event):
118		self.model.appendRow([QStandardItem(str(event))])
119
120
121		class DataChangeUI(object):
122
123	View Code Duplication	def __init__(self, window, uaclient):
		0 ignored issues – show Duplication introduced 2018-03-31 06:27 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
124		self.window = window
125		self.uaclient = uaclient
126		self._subhandler = DataChangeHandler()
127		self._subscribed_nodes = []
128		self.model = QStandardItemModel()
129		self.window.ui.subView.setModel(self.model)
130		self.window.ui.subView.horizontalHeader().setSectionResizeMode(1)
131
132		self.window.ui.actionSubscribeDataChange.triggered.connect(self._subscribe)
133		self.window.ui.actionUnsubscribeDataChange.triggered.connect(self._unsubscribe)
134
135		# populate contextual menu
136		self.window.ui.treeView.addAction(self.window.ui.actionSubscribeDataChange)
137		self.window.ui.treeView.addAction(self.window.ui.actionUnsubscribeDataChange)
138
139		# handle subscriptions
140		self._subhandler.data_change_fired.connect(self._update_subscription_model, type=Qt.QueuedConnection)
141
142		# accept drops
143		self.model.canDropMimeData = self.canDropMimeData
144		self.model.dropMimeData = self.dropMimeData
145
146		def canDropMimeData(self, mdata, action, row, column, parent):
147		return True
148
149		def dropMimeData(self, mdata, action, row, column, parent):
150		node = self.uaclient.client.get_node(mdata.text())
151		self._subscribe(node)
152		return True
153
154		def clear(self):
155		self._subscribed_nodes = []
156		self.model.clear()
157
158		def show_error(self, *args):
159		self.window.show_error(*args)
160
161		@trycatchslot
162		def _subscribe(self, node=None):
163		if not isinstance(node, Node):
164		node = self.window.get_current_node()
165		if node is None:
166		return
167		if node in self._subscribed_nodes:
168		logger.warning("allready subscribed to node: %s ", node)
169		return
170		self.model.setHorizontalHeaderLabels(["DisplayName", "Value", "Timestamp"])
171		text = str(node.get_display_name().Text)
172		row = [QStandardItem(text), QStandardItem("No Data yet"), QStandardItem("")]
173		row[0].setData(node)
174		self.model.appendRow(row)
175		self._subscribed_nodes.append(node)
176		self.window.ui.subDockWidget.raise_()
177		try:
178		self.uaclient.subscribe_datachange(node, self._subhandler)
179		except Exception as ex:
180		self.window.show_error(ex)
181		idx = self.model.indexFromItem(row[0])
182		self.model.takeRow(idx.row())
183		raise
184
185		@trycatchslot
186		def _unsubscribe(self):
187		node = self.window.get_current_node()
188		if node is None:
189		return
190		self.uaclient.unsubscribe_datachange(node)
191		self._subscribed_nodes.remove(node)
192		i = 0
193		while self.model.item(i):
194		item = self.model.item(i)
195		if item.data() == node:
196		self.model.removeRow(i)
197		i += 1
198
199		def _update_subscription_model(self, node, value, timestamp):
200		i = 0
201		while self.model.item(i):
202		item = self.model.item(i)
203		if item.data() == node:
204		it = self.model.item(i, 1)
205		it.setText(value)
206		it_ts = self.model.item(i, 2)
207		it_ts.setText(timestamp)
208		i += 1
209
210
211
212		class Window(QMainWindow):
213
214		def __init__(self):
215		QMainWindow.__init__(self)
216		self.ui = Ui_MainWindow()
217		self.ui.setupUi(self)
218		self.setWindowIcon(QIcon(":/network.svg"))
219
220		# fix stuff imposible to do in qtdesigner
221		# remove dock titlebar for addressbar
222		w = QWidget()
223		self.ui.addrDockWidget.setTitleBarWidget(w)
224		# tabify some docks
225		self.tabifyDockWidget(self.ui.evDockWidget, self.ui.subDockWidget)
226		self.tabifyDockWidget(self.ui.subDockWidget, self.ui.refDockWidget)
227		self.tabifyDockWidget(self.ui.refDockWidget, self.ui.graphDockWidget)
228
229		# we only show statusbar in case of errors
230		self.ui.statusBar.hide()
231
232		# setup QSettings for application and get a settings object
233		QCoreApplication.setOrganizationName("FreeOpcUa")
234		QCoreApplication.setApplicationName("OpcUaClient")
235		self.settings = QSettings()
236
237		self._address_list = self.settings.value("address_list", ["opc.tcp://localhost:4840", "opc.tcp://localhost:53530/OPCUA/SimulationServer/"])
238		self._address_list_max_count = int(self.settings.value("address_list_max_count", 10))
239
240		# init widgets
241		for addr in self._address_list:
242		self.ui.addrComboBox.insertItem(-1, addr)
243
244		self.uaclient = UaClient()
245
246		self.tree_ui = TreeWidget(self.ui.treeView)
247		self.tree_ui.error.connect(self.show_error)
248		self.refs_ui = RefsWidget(self.ui.refView)
249		self.refs_ui.error.connect(self.show_error)
250		self.attrs_ui = AttrsWidget(self.ui.attrView)
251		self.attrs_ui.error.connect(self.show_error)
252		self.datachange_ui = DataChangeUI(self, self.uaclient)
253		self.event_ui = EventUI(self, self.uaclient)
254		self.graph_ui = GraphUI(self, self.uaclient)
255
256		self.ui.addrComboBox.currentTextChanged.connect(self._uri_changed)
257		self._uri_changed(self.ui.addrComboBox.currentText()) # force update for current value at startup
258
259		self.ui.treeView.activated.connect(self.show_refs)
260		self.ui.treeView.clicked.connect(self.show_refs)
261		self.ui.actionCopyPath.triggered.connect(self.tree_ui.copy_path)
262		self.ui.actionCopyNodeId.triggered.connect(self.tree_ui.copy_nodeid)
263		# add items to context menu
264		self.ui.treeView.addAction(self.ui.actionCopyPath)
265		self.ui.treeView.addAction(self.ui.actionCopyNodeId)
266
267		self.ui.treeView.activated.connect(self.show_attrs)
268		self.ui.treeView.clicked.connect(self.show_attrs)
269		self.ui.attrRefreshButton.clicked.connect(self.show_attrs)
270
271		self.resize(int(self.settings.value("main_window_width", 800)), int(self.settings.value("main_window_height", 600)))
272		data = self.settings.value("main_window_state", None)
273		if data:
274		self.restoreState(data)
275
276		self.ui.connectButton.clicked.connect(self.connect)
277		self.ui.disconnectButton.clicked.connect(self.disconnect)
278		# self.ui.treeView.expanded.connect(self._fit)
279
280		self.ui.actionConnect.triggered.connect(self.connect)
281		self.ui.actionDisconnect.triggered.connect(self.disconnect)
282
283		self.ui.connectOptionButton.clicked.connect(self.show_connection_dialog)
284
285		def _uri_changed(self, uri):
286		self.uaclient.load_security_settings(uri)
287
288		def show_connection_dialog(self):
289		dia = ConnectionDialog(self, self.ui.addrComboBox.currentText())
290		dia.security_mode = self.uaclient.security_mode
291		dia.security_policy = self.uaclient.security_policy
292		dia.certificate_path = self.uaclient.certificate_path
293		dia.private_key_path = self.uaclient.private_key_path
294		ret = dia.exec_()
295		if ret:
296		self.uaclient.security_mode = dia.security_mode
297		self.uaclient.security_policy = dia.security_policy
298		self.uaclient.certificate_path = dia.certificate_path
299		self.uaclient.private_key_path = dia.private_key_path
300
301		@trycatchslot
302		def show_refs(self, idx):
303		node = self.get_current_node(idx)
304		if node:
305		self.refs_ui.show_refs(node)
306
307		@trycatchslot
308		def show_attrs(self, idx):
309		if not isinstance(idx, QModelIndex):
310		idx = None
311		node = self.get_current_node(idx)
312		if node:
313		self.attrs_ui.show_attrs(node)
314
315		def show_error(self, msg):
316		logger.warning("showing error: %s")
317		self.ui.statusBar.show()
318		self.ui.statusBar.setStyleSheet("QStatusBar { background-color : red; color : black; }")
319		self.ui.statusBar.showMessage(str(msg))
320		QTimer.singleShot(1500, self.ui.statusBar.hide)
321
322		def get_current_node(self, idx=None):
323		return self.tree_ui.get_current_node(idx)
324
325		def get_uaclient(self):
326		return self.uaclient
327
328		@trycatchslot
329		def connect(self):
330		uri = self.ui.addrComboBox.currentText()
331		try:
332		self.uaclient.connect(uri)
333		except Exception as ex:
334		self.show_error(ex)
335		raise
336
337		self._update_address_list(uri)
338		self.tree_ui.set_root_node(self.uaclient.client.get_root_node())
339		self.load_current_node()
340
341		def _update_address_list(self, uri):
342		if uri == self._address_list[0]:
343		return
344		if uri in self._address_list:
345		self._address_list.remove(uri)
346		self._address_list.insert(0, uri)
347		if len(self._address_list) > self._address_list_max_count:
348		self._address_list.pop(-1)
349
350		def disconnect(self):
351		try:
352		self.uaclient.disconnect()
353		except Exception as ex:
354		self.show_error(ex)
355		raise
356		finally:
357		self.save_current_node()
358		self.tree_ui.clear()
359		self.refs_ui.clear()
360		self.attrs_ui.clear()
361		self.datachange_ui.clear()
362		self.event_ui.clear()
363
364
365		def closeEvent(self, event):
366		self.tree_ui.save_state()
367		self.attrs_ui.save_state()
368		self.refs_ui.save_state()
369		self.settings.setValue("main_window_width", self.size().width())
370		self.settings.setValue("main_window_height", self.size().height())
371		self.settings.setValue("main_window_state", self.saveState())
372		self.settings.setValue("address_list", self._address_list)
373		self.disconnect()
374		event.accept()
375
376		def save_current_node(self):
377		current_node = self.tree_ui.get_current_node()
378		if current_node:
379		mysettings = self.settings.value("current_node", None)
380		if mysettings is None:
381		mysettings = {}
382		uri = self.ui.addrComboBox.currentText()
383		mysettings[uri] = current_node.nodeid.to_string()
384		self.settings.setValue("current_node", mysettings)
385
386		def load_current_node(self):
387		mysettings = self.settings.value("current_node", None)
388		if mysettings is None:
389		return
390		uri = self.ui.addrComboBox.currentText()
391		if uri in mysettings:
392		nodeid = ua.NodeId.from_string(mysettings[uri])
393		node = self.uaclient.client.get_node(nodeid)
394		self.tree_ui.expand_to_node(node)
395
396
397		def main():
398		app = QApplication(sys.argv)
399		client = Window()
400		handler = QtHandler(client.ui.logTextEdit)
401		logging.getLogger().addHandler(handler)
402		logging.getLogger("uaclient").setLevel(logging.INFO)
403		logging.getLogger("uawidgets").setLevel(logging.INFO)
404		#logging.getLogger("opcua").setLevel(logging.INFO) # to enable logging of ua server
405
406		client.show()
407		sys.exit(app.exec_())
408
409
410		if __name__ == "__main__":
411		main()
412

FreeOpcUa / opcua-client-gui

Push — master ( e7ced7...6f1de5 )

Window.save_current_node() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like