AttrsUI._check_edit() - Code Metrics - Inspection of "display timestamps for Value attributes" - FreeOpcUa/opcua-client-gui - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 018d99...d16076 )

by Olivier

created 2016-06-20 05:47 UTC

AttrsUI._check_edit() A

↳ Parent: AttrsUI

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	3
dl	0
loc	11
rs	9.4285
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, val_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 _check_edit(self, item):
        """
        filter only element we want to edit.
        take either idx eller item as argument
        """
        if item.column() != 1:
            return False
        name_item = self.model.item(item.row(), 0)
        if name_item.text() != "Value":
            return False
        return True

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

    def edit_attr_finished(self, item):
        if not self._check_edit(item):
            return
        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:
            dv = self.current_node.get_data_value()
            item.setText(variant_to_string(dv.Value))
            name_item = self.model.item(item.row(), 0)
            name_item.child(0, 1).setText(val_to_string(dv.ServerTimestamp))
            name_item.child(1, 1).setText(val_to_string(dv.SourceTimestamp))

    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)
        self.window.ui.attrView.expandAll()

    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 name, dv in attrs:
            string = variant_to_string(dv.Value)
            name_item = QStandardItem(name)
            vitem = QStandardItem(string)
            vitem.setData(dv.Value)
            self.model.appendRow([name_item, vitem, QStandardItem(dv.Value.VariantType.name)])
            if name == "Value":
                string = val_to_string(dv.ServerTimestamp)
                name_item.appendRow([QStandardItem("Server Timestamp"), QStandardItem(string), QStandardItem(ua.VariantType.DateTime.name)])
                string = val_to_string(dv.SourceTimestamp)
                name_item.appendRow([QStandardItem("Source Timestamp"), QStandardItem(string), QStandardItem(ua.VariantType.DateTime.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, val_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 _check_edit(self, item):
181		"""
182		filter only element we want to edit.
183		take either idx eller item as argument
184		"""
185		if item.column() != 1:
186		return False
187		name_item = self.model.item(item.row(), 0)
188		if name_item.text() != "Value":
189		return False
190		return True
191
192		def edit_attr(self, idx):
193		if not self._check_edit(idx):
194		return
195		attritem = self.model.item(idx.row(), 0)
196		if attritem.text() == "Value":
197		self.window.ui.attrView.edit(idx)
198
199		def edit_attr_finished(self, item):
200		if not self._check_edit(item):
201		return
202		try:
203		var = item.data()
204		val = item.text()
205		var = string_to_variant(val, var.VariantType)
206		self.current_node.set_value(var)
207		except Exception as ex:
208		self.window.show_error(ex)
209		raise
210		finally:
211		dv = self.current_node.get_data_value()
212		item.setText(variant_to_string(dv.Value))
213		name_item = self.model.item(item.row(), 0)
214		name_item.child(0, 1).setText(val_to_string(dv.ServerTimestamp))
215		name_item.child(1, 1).setText(val_to_string(dv.SourceTimestamp))
216
217		def showContextMenu(self, position):
218		item = self.get_current_item()
219		if item:
220		self._contextMenu.exec_(self.window.ui.attrView.mapToGlobal(position))
221
222		def get_current_item(self, col_idx=0):
223		idx = self.window.ui.attrView.currentIndex()
224		return self.model.item(idx.row(), col_idx)
225
226		def _copy_value(self, position):
227		it = self.get_current_item(1)
228		if it:
229		QApplication.clipboard().setText(it.text())
230
231		def clear(self):
232		self.model.clear()
233
234		def show_attrs(self, idx):
235		if not isinstance(idx, QModelIndex):
236		idx = None
237		self.current_node = self.window.get_current_node(idx)
238		self.model.clear()
239		if self.current_node:
240		self._show_attrs(self.current_node)
241		self.window.ui.attrView.expandAll()
242
243		def _show_attrs(self, node):
244		try:
245		attrs = self.uaclient.get_all_attrs(node)
246		except Exception as ex:
247		self.window.show_error(ex)
248		raise
249		self.model.setHorizontalHeaderLabels(['Attribute', 'Value', 'DataType'])
250		for name, dv in attrs:
251		string = variant_to_string(dv.Value)
252		name_item = QStandardItem(name)
253		vitem = QStandardItem(string)
254		vitem.setData(dv.Value)
255		self.model.appendRow([name_item, vitem, QStandardItem(dv.Value.VariantType.name)])
256		if name == "Value":
257		string = val_to_string(dv.ServerTimestamp)
258		name_item.appendRow([QStandardItem("Server Timestamp"), QStandardItem(string), QStandardItem(ua.VariantType.DateTime.name)])
259		string = val_to_string(dv.SourceTimestamp)
260		name_item.appendRow([QStandardItem("Source Timestamp"), QStandardItem(string), QStandardItem(ua.VariantType.DateTime.name)])
261
262
263
264		class RefsUI(object):
265
266		def __init__(self, window, uaclient):
267		self.window = window
268		self.uaclient = uaclient
269		self.model = QStandardItemModel()
270		self.window.ui.refView.setModel(self.model)
271		self.window.ui.refView.horizontalHeader().setSectionResizeMode(1)
272
273		self.window.ui.treeView.activated.connect(self.show_refs)
274		self.window.ui.treeView.clicked.connect(self.show_refs)
275
276	View Code Duplication	def clear(self):
		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.model.clear()
278
279		def show_refs(self, idx):
280		node = self.window.get_current_node(idx)
281		self.model.clear()
282		if node:
283		self._show_refs(node)
284
285		def _show_refs(self, node):
286		self.model.setHorizontalHeaderLabels(['ReferenceType', 'NodeId', "BrowseName", "TypeDefinition"])
287		try:
288		refs = self.uaclient.get_all_refs(node)
289		except Exception as ex:
290		self.window.show_error(ex)
291		raise
292		for ref in refs:
293		self.model.appendRow([QStandardItem(str(ref.ReferenceTypeId)),
294		QStandardItem(str(ref.NodeId)),
295		QStandardItem(str(ref.BrowseName)),
296		QStandardItem(str(ref.TypeDefinition))])
297
298
299		class TreeUI(object):
300
301		def __init__(self, window, uaclient):
302		self.window = window
303		self.uaclient = uaclient
304		self.model = TreeViewModel(self.uaclient)
305		self.model.clear() # FIXME: do we need this?
306		self.model.error.connect(self.window.show_error)
307		self.window.ui.treeView.setModel(self.model)
308		self.window.ui.treeView.setUniformRowHeights(True)
309		self.window.ui.treeView.setSelectionBehavior(QAbstractItemView.SelectRows)
310		self.window.ui.treeView.header().setSectionResizeMode(1)
311		self.window.ui.actionCopyPath.triggered.connect(self._copy_path)
312		self.window.ui.actionCopyNodeId.triggered.connect(self._copy_nodeid)
313		# add items to context menu
314		self.window.ui.treeView.addAction(self.window.ui.actionCopyPath)
315		self.window.ui.treeView.addAction(self.window.ui.actionCopyNodeId)
316
317		def clear(self):
318		self.model.clear()
319
320		def start(self):
321		self.model.clear()
322		self.model.add_item(*self.uaclient.get_root_node_and_desc())
323
324		def _copy_path(self):
325		path = self.get_current_path()
326		path = ",".join(path)
327		QApplication.clipboard().setText(path)
328
329		def _copy_nodeid(self):
330		node = self.get_current_node()
331		if node:
332		text = node.nodeid.to_string()
333		else:
334		text = ""
335		QApplication.clipboard().setText(text)
336
337		def get_current_path(self):
338		idx = self.window.ui.treeView.currentIndex()
339		idx = idx.sibling(idx.row(), 0)
340		it = self.model.itemFromIndex(idx)
341		path = []
342		while it and it.data():
343		node = it.data()
344		name = node.get_browse_name().to_string()
345		path.insert(0, name)
346		it = it.parent()
347		return path
348
349		def get_current_node(self, idx=None):
350		if idx is None:
351		idx = self.window.ui.treeView.currentIndex()
352		idx = idx.sibling(idx.row(), 0)
353		it = self.model.itemFromIndex(idx)
354		if not it:
355		return None
356		node = it.data()
357		if not node:
358		print("No node for item:", it, it.text())
359		return None
360		node.display_name = it.text() # FIXME: hack
361		return node
362
363
364		class Window(QMainWindow):
365
366		def __init__(self):
367		QMainWindow.__init__(self)
368		self.ui = Ui_MainWindow()
369		self.ui.setupUi(self)
370		self.setWindowIcon(QIcon(":/network.svg"))
371
372		# fix stuff imposible to do in qtdesigner
373		# remove dock titlebar for addressbar
374		w = QWidget()
375		self.ui.addrDockWidget.setTitleBarWidget(w)
376		# tabify some docks
377		self.tabifyDockWidget(self.ui.evDockWidget, self.ui.subDockWidget)
378		self.tabifyDockWidget(self.ui.subDockWidget, self.ui.refDockWidget)
379
380		# we only show statusbar in case of errors
381		self.ui.statusBar.hide()
382
383		# load settings, seconds arg is default
384		self.settings = QSettings("FreeOpcUa", "FreeOpcUaClient")
385		self._address_list = self.settings.value("address_list", ["opc.tcp://localhost:4840", "opc.tcp://localhost:53530/OPCUA/SimulationServer/"])
386		self._address_list_max_count = int(self.settings.value("address_list_max_count", 10))
387
388		# init widgets
389		for addr in self._address_list:
390		self.ui.addrComboBox.insertItem(-1, addr)
391
392		self.uaclient = UaClient()
393
394		self.tree_ui = TreeUI(self, self.uaclient)
395		self.refs_ui = RefsUI(self, self.uaclient)
396		self.attrs_ui = AttrsUI(self, self.uaclient)
397		self.datachange_ui = DataChangeUI(self, self.uaclient)
398		self.event_ui = EventUI(self, self.uaclient)
399
400		self.resize(int(self.settings.value("main_window_width", 800)), int(self.settings.value("main_window_height", 600)))
401		self.restoreState(self.settings.value("main_window_state", b""))
402
403		self.ui.connectButton.clicked.connect(self._connect)
404		self.ui.disconnectButton.clicked.connect(self._disconnect)
405		# self.ui.treeView.expanded.connect(self._fit)
406
407		self.ui.actionConnect.triggered.connect(self._connect)
408		self.ui.actionDisconnect.triggered.connect(self._disconnect)
409
410		def show_error(self, msg, level=1):
411		print("showing error: ", msg, level)
412		self.ui.statusBar.show()
413		self.ui.statusBar.setStyleSheet("QStatusBar { background-color : red; color : black; }")
414		self.ui.statusBar.showMessage(str(msg))
415		QTimer.singleShot(1500, self.ui.statusBar.hide)
416
417		def get_current_node(self, idx=None):
418		return self.tree_ui.get_current_node(idx)
419
420		def get_uaclient(self):
421		return self.uaclient
422
423		def _connect(self):
424		uri = self.ui.addrComboBox.currentText()
425		try:
426		self.uaclient.connect(uri)
427		except Exception as ex:
428		self.show_error(ex)
429		raise
430
431		self._update_address_list(uri)
432		self.tree_ui.start()
433
434		def _update_address_list(self, uri):
435		if uri == self._address_list[0]:
436		return
437		if uri in self._address_list:
438		self._address_list.remove(uri)
439		self._address_list.insert(0, uri)
440		if len(self._address_list) > self._address_list_max_count:
441		self._address_list.pop(-1)
442
443		def _disconnect(self):
444		try:
445		self.uaclient.disconnect()
446		except Exception as ex:
447		self.show_error(ex)
448		raise
449		finally:
450		self.tree_ui.clear()
451		self.refs_ui.clear()
452		self.attrs_ui.clear()
453		self.datachange_ui.clear()
454		self.event_ui.clear()
455
456		def closeEvent(self, event):
457		self.settings.setValue("main_window_width", self.size().width())
458		self.settings.setValue("main_window_height", self.size().height())
459		self.settings.setValue("main_window_state", self.saveState())
460		self.settings.setValue("address_list", self._address_list)
461		self._disconnect()
462		event.accept()
463
464
465		class TreeViewModel(QStandardItemModel):
466
467		error = pyqtSignal(str)
468
469		def __init__(self, uaclient):
470		super(TreeViewModel, self).__init__()
471		self.uaclient = uaclient
472		self._fetched = []
473
474		def clear(self):
475		QStandardItemModel.clear(self)
476		self._fetched = []
477		self.setHorizontalHeaderLabels(['DisplayName', "BrowseName", 'NodeId'])
478
479		def add_item(self, node, desc, parent=None):
480		item = [QStandardItem(desc.DisplayName.to_string()), QStandardItem(desc.BrowseName.to_string()), QStandardItem(desc.NodeId.to_string())]
481		if desc.NodeClass == ua.NodeClass.Object:
482		if desc.TypeDefinition == ua.TwoByteNodeId(ua.ObjectIds.FolderType):
483		item[0].setIcon(QIcon(":/folder.svg"))
484		else:
485		item[0].setIcon(QIcon(":/object.svg"))
486		elif desc.NodeClass == ua.NodeClass.Variable:
487		if desc.TypeDefinition == ua.TwoByteNodeId(ua.ObjectIds.PropertyType):
488		item[0].setIcon(QIcon(":/property.svg"))
489		else:
490		item[0].setIcon(QIcon(":/variable.svg"))
491		elif desc.NodeClass == ua.NodeClass.Method:
492		item[0].setIcon(QIcon(":/method.svg"))
493
494		item[0].setData(node)
495		if parent:
496		return parent.appendRow(item)
497		else:
498		return self.appendRow(item)
499
500		def canFetchMore(self, idx):
501		item = self.itemFromIndex(idx)
502		if not item:
503		return True
504		node = item.data()
505		if node not in self._fetched:
506		self._fetched.append(node)
507		return True
508		return False
509
510		def hasChildren(self, idx):
511		item = self.itemFromIndex(idx)
512		if not item:
513		return True
514		node = item.data()
515		if node in self._fetched:
516		return QStandardItemModel.hasChildren(self, idx)
517		return True
518
519		def fetchMore(self, idx):
520		parent = self.itemFromIndex(idx)
521		if parent:
522		self._fetchMore(parent)
523
524		def _fetchMore(self, parent):
525		try:
526		for node, attrs in self.uaclient.get_children(parent.data()):
527		self.add_item(node, attrs, parent)
528		except Exception as ex:
529		self.error.emit(ex)
530		raise
531
532
533		def main():
534		app = QApplication(sys.argv)
535		client = Window()
536		client.show()
537		sys.exit(app.exec_())
538
539
540		if __name__ == "__main__":
541		main()
542

FreeOpcUa / opcua-client-gui

Push — master ( 018d99...d16076 )

AttrsUI._check_edit() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like