Window.show_connection_dialog() - Code Metrics - Inspection of "Add support for encryption" - FreeOpcUa/opcua-client-gui - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 70d6df...ac5e60 )

by Olivier

created 2016-12-08 08:57 UTC

Window.show_connection_dialog() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

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

#! /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 uaclient.uaclient import UaClient
from uaclient.mainwindow_ui import Ui_MainWindow
from uaclient.connection_dialog import ConnectionDialog
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


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 node is None:
            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)

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

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

FreeOpcUa / opcua-client-gui

Push — master ( 70d6df...ac5e60 )

Window.show_connection_dialog() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like