AttrsUI._show_attrs() - Code Metrics - Inspection of "more robuts display of DataType" - FreeOpcUa/opcua-client-gui - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 66639b...e27484 )

by Olivier

created 2016-06-25 08:06 UTC

AttrsUI._show_attrs() F

↳ Parent: AttrsUI

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	6
Bugs	0	Features	0

Metric	Value
cc	10
c	6
b	0
f	0
dl	0
loc	32
rs	3.1304

How to fix Complexity

#! /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:
            if name == "DataType":
                if dv.Value.Value.Identifier < 63:
                    string = ua.DataType_to_VariantType(dv.Value.Value).name
                elif dv.Value.Value.Identifier in ua.ObjectIdNames:
                    string = ua.ObjectIdNames[dv.Value.Value.Identifier]
                else:
                    string = dv.Value.Value.to_string()
            elif name in ("AccessLevel", "UserAccessLevel"):
                string = ",".join(ua.int_to_AccessLevel(dv.Value.Value))
            elif name in ("WriteMask", "UserWriteMask"):
                string = ",".join(ua.int_to_WriteMask(dv.Value.Value))
            elif name in ("EventNotifier"):
                string = ",".join(ua.int_to_EventNotifier(dv.Value.Value))
            else:
                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:
            typename = ua.ObjectIdNames[ref.ReferenceTypeId.Identifier]
            if ref.NodeId.NamespaceIndex == 0 and ref.NodeId.Identifier in ua.ObjectIdNames:
                nodeid = ua.ObjectIdNames[ref.NodeId.Identifier]
            else:
                nodeid = str(ref.NodeId)
            typedef = ua.ObjectIdNames[ref.TypeDefinition.Identifier]
            self.model.appendRow([QStandardItem(typename),
                                  QStandardItem(nodeid),
                                  QStandardItem(ref.BrowseName.to_string()),
                                  QStandardItem(typedef)
                                  ])


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

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

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

    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		def __init__(self, window, uaclient):
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		if name == "DataType":
252		if dv.Value.Value.Identifier < 63:
253		string = ua.DataType_to_VariantType(dv.Value.Value).name
254		elif dv.Value.Value.Identifier in ua.ObjectIdNames:
255		string = ua.ObjectIdNames[dv.Value.Value.Identifier]
256		else:
257		string = dv.Value.Value.to_string()
258		elif name in ("AccessLevel", "UserAccessLevel"):
259		string = ",".join(ua.int_to_AccessLevel(dv.Value.Value))
260		elif name in ("WriteMask", "UserWriteMask"):
261		string = ",".join(ua.int_to_WriteMask(dv.Value.Value))
262		elif name in ("EventNotifier"):
263		string = ",".join(ua.int_to_EventNotifier(dv.Value.Value))
264		else:
265		string = variant_to_string(dv.Value)
266		name_item = QStandardItem(name)
267		vitem = QStandardItem(string)
268		vitem.setData(dv.Value)
269		self.model.appendRow([name_item, vitem, QStandardItem(dv.Value.VariantType.name)])
270		if name == "Value":
271		string = val_to_string(dv.ServerTimestamp)
272		name_item.appendRow([QStandardItem("Server Timestamp"), QStandardItem(string), QStandardItem(ua.VariantType.DateTime.name)])
273		string = val_to_string(dv.SourceTimestamp)
274		name_item.appendRow([QStandardItem("Source Timestamp"), QStandardItem(string), QStandardItem(ua.VariantType.DateTime.name)])
275
276
277
278		class RefsUI(object):
279
280		def __init__(self, window, uaclient):
281		self.window = window
282		self.uaclient = uaclient
283		self.model = QStandardItemModel()
284		self.window.ui.refView.setModel(self.model)
285		self.window.ui.refView.horizontalHeader().setSectionResizeMode(1)
286
287		self.window.ui.treeView.activated.connect(self.show_refs)
288		self.window.ui.treeView.clicked.connect(self.show_refs)
289
290		def clear(self):
291		self.model.clear()
292
293		def show_refs(self, idx):
294		node = self.window.get_current_node(idx)
295		self.model.clear()
296		if node:
297		self._show_refs(node)
298
299		def _show_refs(self, node):
300		self.model.setHorizontalHeaderLabels(['ReferenceType', 'NodeId', "BrowseName", "TypeDefinition"])
301		try:
302		refs = self.uaclient.get_all_refs(node)
303		except Exception as ex:
304		self.window.show_error(ex)
305		raise
306		for ref in refs:
307		typename = ua.ObjectIdNames[ref.ReferenceTypeId.Identifier]
308		if ref.NodeId.NamespaceIndex == 0 and ref.NodeId.Identifier in ua.ObjectIdNames:
309		nodeid = ua.ObjectIdNames[ref.NodeId.Identifier]
310		else:
311		nodeid = str(ref.NodeId)
312		typedef = ua.ObjectIdNames[ref.TypeDefinition.Identifier]
313		self.model.appendRow([QStandardItem(typename),
314		QStandardItem(nodeid),
315		QStandardItem(ref.BrowseName.to_string()),
316		QStandardItem(typedef)
317		])
318
319
320		class TreeUI(object):
321
322		def __init__(self, window, uaclient):
323		self.window = window
324		self.uaclient = uaclient
325		self.model = TreeViewModel(self.uaclient)
326		self.model.clear() # FIXME: do we need this?
327		self.model.error.connect(self.window.show_error)
328		self.window.ui.treeView.setModel(self.model)
329		#self.window.ui.treeView.setUniformRowHeights(True)
330		self.window.ui.treeView.setSelectionBehavior(QAbstractItemView.SelectRows)
331		self.window.ui.treeView.header().setSectionResizeMode(1)
332		self.window.ui.actionCopyPath.triggered.connect(self._copy_path)
333		self.window.ui.actionCopyNodeId.triggered.connect(self._copy_nodeid)
334		# add items to context menu
335		self.window.ui.treeView.addAction(self.window.ui.actionCopyPath)
336		self.window.ui.treeView.addAction(self.window.ui.actionCopyNodeId)
337
338		def clear(self):
339		self.model.clear()
340
341		def start(self):
342		self.model.clear()
343		self.model.add_item(*self.uaclient.get_root_node_and_desc())
344
345		def _copy_path(self):
346		path = self.get_current_path()
347		path = ",".join(path)
348		QApplication.clipboard().setText(path)
349
350		def _copy_nodeid(self):
351		node = self.get_current_node()
352		if node:
353		text = node.nodeid.to_string()
354		else:
355		text = ""
356		QApplication.clipboard().setText(text)
357
358		def get_current_path(self):
359		idx = self.window.ui.treeView.currentIndex()
360		idx = idx.sibling(idx.row(), 0)
361		it = self.model.itemFromIndex(idx)
362		path = []
363		while it and it.data():
364		node = it.data()
365		name = node.get_browse_name().to_string()
366		path.insert(0, name)
367		it = it.parent()
368		return path
369
370		def get_current_node(self, idx=None):
371		if idx is None:
372		idx = self.window.ui.treeView.currentIndex()
373		idx = idx.sibling(idx.row(), 0)
374		it = self.model.itemFromIndex(idx)
375		if not it:
376		return None
377		node = it.data()
378		if not node:
379		print("No node for item:", it, it.text())
380		return None
381		node.display_name = it.text() # FIXME: hack
382		return node
383
384
385		class Window(QMainWindow):
386
387		def __init__(self):
388		QMainWindow.__init__(self)
389		self.ui = Ui_MainWindow()
390		self.ui.setupUi(self)
391		self.setWindowIcon(QIcon(":/network.svg"))
392
393		# fix stuff imposible to do in qtdesigner
394		# remove dock titlebar for addressbar
395		w = QWidget()
396		self.ui.addrDockWidget.setTitleBarWidget(w)
397		# tabify some docks
398		self.tabifyDockWidget(self.ui.evDockWidget, self.ui.subDockWidget)
399		self.tabifyDockWidget(self.ui.subDockWidget, self.ui.refDockWidget)
400
401		# we only show statusbar in case of errors
402		self.ui.statusBar.hide()
403
404		# load settings, seconds arg is default
405		self.settings = QSettings("FreeOpcUa", "FreeOpcUaClient")
406		self._address_list = self.settings.value("address_list", ["opc.tcp://localhost:4840", "opc.tcp://localhost:53530/OPCUA/SimulationServer/"])
407		self._address_list_max_count = int(self.settings.value("address_list_max_count", 10))
408
409		# init widgets
410		for addr in self._address_list:
411		self.ui.addrComboBox.insertItem(-1, addr)
412
413		self.uaclient = UaClient()
414
415		self.tree_ui = TreeUI(self, self.uaclient)
416		self.refs_ui = RefsUI(self, self.uaclient)
417		self.attrs_ui = AttrsUI(self, self.uaclient)
418		self.datachange_ui = DataChangeUI(self, self.uaclient)
419		self.event_ui = EventUI(self, self.uaclient)
420
421		self.resize(int(self.settings.value("main_window_width", 800)), int(self.settings.value("main_window_height", 600)))
422		self.restoreState(self.settings.value("main_window_state", b"", type="QByteArray"))
423
424		self.ui.connectButton.clicked.connect(self._connect)
425		self.ui.disconnectButton.clicked.connect(self._disconnect)
426		# self.ui.treeView.expanded.connect(self._fit)
427
428		self.ui.actionConnect.triggered.connect(self._connect)
429		self.ui.actionDisconnect.triggered.connect(self._disconnect)
430
431		def show_error(self, msg, level=1):
432		print("showing error: ", msg, level)
433		self.ui.statusBar.show()
434		self.ui.statusBar.setStyleSheet("QStatusBar { background-color : red; color : black; }")
435		self.ui.statusBar.showMessage(str(msg))
436		QTimer.singleShot(1500, self.ui.statusBar.hide)
437
438		def get_current_node(self, idx=None):
439		return self.tree_ui.get_current_node(idx)
440
441		def get_uaclient(self):
442		return self.uaclient
443
444		def _connect(self):
445		uri = self.ui.addrComboBox.currentText()
446		try:
447		self.uaclient.connect(uri)
448		except Exception as ex:
449		self.show_error(ex)
450		raise
451
452		self._update_address_list(uri)
453		self.tree_ui.start()
454
455		def _update_address_list(self, uri):
456		if uri == self._address_list[0]:
457		return
458		if uri in self._address_list:
459		self._address_list.remove(uri)
460		self._address_list.insert(0, uri)
461		if len(self._address_list) > self._address_list_max_count:
462		self._address_list.pop(-1)
463
464		def _disconnect(self):
465		try:
466		self.uaclient.disconnect()
467		except Exception as ex:
468		self.show_error(ex)
469		raise
470		finally:
471		self.tree_ui.clear()
472		self.refs_ui.clear()
473		self.attrs_ui.clear()
474		self.datachange_ui.clear()
475		self.event_ui.clear()
476
477		def closeEvent(self, event):
478		self.settings.setValue("main_window_width", self.size().width())
479		self.settings.setValue("main_window_height", self.size().height())
480		self.settings.setValue("main_window_state", self.saveState())
481		self.settings.setValue("address_list", self._address_list)
482		self._disconnect()
483		event.accept()
484
485
486		class TreeViewModel(QStandardItemModel):
487
488		error = pyqtSignal(str)
489
490		def __init__(self, uaclient):
491		super(TreeViewModel, self).__init__()
492		self.uaclient = uaclient
493		self._fetched = []
494
495		def clear(self):
496		QStandardItemModel.clear(self)
497		self._fetched = []
498		self.setHorizontalHeaderLabels(['DisplayName', "BrowseName", 'NodeId'])
499
500		def add_item(self, node, desc, parent=None):
501		item = [QStandardItem(desc.DisplayName.to_string()), QStandardItem(desc.BrowseName.to_string()), QStandardItem(desc.NodeId.to_string())]
502		if desc.NodeClass == ua.NodeClass.Object:
503		if desc.TypeDefinition == ua.TwoByteNodeId(ua.ObjectIds.FolderType):
504		item[0].setIcon(QIcon(":/folder.svg"))
505		else:
506		item[0].setIcon(QIcon(":/object.svg"))
507		elif desc.NodeClass == ua.NodeClass.Variable:
508		if desc.TypeDefinition == ua.TwoByteNodeId(ua.ObjectIds.PropertyType):
509		item[0].setIcon(QIcon(":/property.svg"))
510		else:
511		item[0].setIcon(QIcon(":/variable.svg"))
512		elif desc.NodeClass == ua.NodeClass.Method:
513		item[0].setIcon(QIcon(":/method.svg"))
514
515		item[0].setData(node)
516		if parent:
517		return parent.appendRow(item)
518		else:
519		return self.appendRow(item)
520
521		def canFetchMore(self, idx):
522		item = self.itemFromIndex(idx)
523		if not item:
524		return True
525		node = item.data()
526		if node not in self._fetched:
527		self._fetched.append(node)
528		return True
529		return False
530
531		def hasChildren(self, idx):
532		item = self.itemFromIndex(idx)
533		if not item:
534		return True
535		node = item.data()
536		if node in self._fetched:
537		return QStandardItemModel.hasChildren(self, idx)
538		return True
539
540		def fetchMore(self, idx):
541		parent = self.itemFromIndex(idx)
542		if parent:
543		self._fetchMore(parent)
544
545		def _fetchMore(self, parent):
546		try:
547		for node, attrs in self.uaclient.get_children(parent.data()):
548		self.add_item(node, attrs, parent)
549		except Exception as ex:
550		self.error.emit(ex)
551		raise
552
553
554		def main():
555		app = QApplication(sys.argv)
556		client = Window()
557		client.show()
558		sys.exit(app.exec_())
559
560
561		if __name__ == "__main__":
562		main()
563

FreeOpcUa / opcua-client-gui

Push — master ( 66639b...e27484 )

AttrsUI._show_attrs() F

Complexity

Size

Duplication

Importance

How to fix Complexity

Complexity

Duplication Side-by-Side

Filter issues like