EventUI._unsubscribe() - Code Metrics - Inspection of "prepare UI for connection settings. accept drops f..." - FreeOpcUa/opcua-client-gui - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 26ea6b...5668b2 )

by Olivier

created 2016-07-24 09:18 UTC

EventUI._unsubscribe() A

↳ Parent: EventUI

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	2
dl	0
loc	6
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, QMimeData
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)

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

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

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


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

    def _subscribe(self, node=None):
        print("SUB", node)
        if not node:
            node = self.window.get_current_node()
            if node is None:
                return
        if node in self._subscribed_nodes:
            print("allready subscribed to event for node: ", node)
            return
        print("Subscribing to events for ", node)
        self.window.ui.evDockWidget.raise_()
        try:
            self.uaclient.subscribe_events(node, self._handler)
        except Exception as ex:
            self.window.show_error(ex)
            raise
        else:
            self._subscribed_nodes.append(node)

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

    def _update_event_model(self, event):
        self.model.appendRow([QStandardItem(str(event))])


class DataChangeUI(object):

    def __init__(self, window, uaclient):

        self.window = window
        self.uaclient = uaclient
        self._subhandler = DataChangeHandler()
        self._subscribed_nodes = []
        self.model = QStandardItemModel()
        self.window.ui.subView.setModel(self.model)
        self.window.ui.subView.horizontalHeader().setSectionResizeMode(1)

        self.window.ui.actionSubscribeDataChange.triggered.connect(self._subscribe)
        self.window.ui.actionUnsubscribeDataChange.triggered.connect(self._unsubscribe)

        # populate contextual menu
        self.window.ui.treeView.addAction(self.window.ui.actionSubscribeDataChange)
        self.window.ui.treeView.addAction(self.window.ui.actionUnsubscribeDataChange)

        # handle subscriptions
        self._subhandler.data_change_fired.connect(self._update_subscription_model, type=Qt.QueuedConnection)
        
        # accept drops
        self.model.canDropMimeData = self.canDropMimeData
        self.model.dropMimeData = self.dropMimeData

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

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

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

    def _subscribe(self, node=None):
        if node is None:
            node = self.window.get_current_node()
            if node is None:
                return
        if node in self._subscribed_nodes:
            print("allready subscribed to node: ", node)
            return
        self.model.setHorizontalHeaderLabels(["DisplayName", "Value", "Timestamp"])
        text = str(node.get_display_name().Text)
        row = [QStandardItem(text), QStandardItem("No Data yet"), QStandardItem("")]
        row[0].setData(node)
        self.model.appendRow(row)
        self._subscribed_nodes.append(node)
        self.window.ui.subDockWidget.raise_()
        try:
            self.uaclient.subscribe_datachange(node, self._subhandler)
        except Exception as ex:
            self.window.show_error(ex)
            idx = self.model.indexFromItem(row[0])
            self.model.takeRow(idx.row())
            raise

    def _unsubscribe(self):
        node = self.window.get_current_node()
        if node is None:
            return
        self.uaclient.unsubscribe_datachange(node)
        self._subscribed_nodes.remove(node)
        i = 0
        while self.model.item(i):
            item = self.model.item(i)
            if item.data() == node:
                self.model.removeRow(i)
            i += 1

    def _update_subscription_model(self, node, value, timestamp):
        i = 0
        while self.model.item(i):
            item = self.model.item(i)
            if item.data() == node:
                it = self.model.item(i, 1)
                it.setText(value)
                it_ts = self.model.item(i, 2)
                it_ts.setText(timestamp)
            i += 1


class 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 isinstance(dv.Value.Value.Identifier, int) and 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([e.name for e in ua.int_to_AccessLevel(dv.Value.Value)])
            elif name in ("WriteMask", "UserWriteMask"):
                string = ",".join([e.name for e in ua.int_to_WriteMask(dv.Value.Value)])
            elif name in ("EventNotifier"):
                string = ",".join([e.name for e in 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 = ref.NodeId.to_string()
            if ref.TypeDefinition.Identifier in ua.ObjectIdNames:
                typedef = ua.ObjectIdNames[ref.TypeDefinition.Identifier]
            else:
                typedef = ref.TypeDefinition.to_string()
            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
        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)

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

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

    def show_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()).items():
                self.add_item(node, attrs, parent)
        except Exception as ex:
            self.error.emit(ex)
            raise

    #def flags(self, idx):
        #item = self.itemFromIndex(idx)
        #flags = QStandardItemModel.flags(self, idx)
        #if not item:
            #return flags
        #node = item.data()
        #if node and node.get_node_class() == ua.NodeClass.Variable:
            ## FIXME not efficient to query, should be stored in data()
            ##print(1, flags)
            #return flags | Qt.ItemIsDropEnabled
        #else:
            #print(2, flags)
            #return flags

    #def mimeTypes(self):
        #return ["application/vnd.text.list"]

    def mimeData(self, idxs):
        mdata = QMimeData()
        nodes = []
        for idx in idxs:
            item = self.itemFromIndex(idx)
            if item:
                node = item.data()
                if node:
                    nodes.append(node.nodeid.to_string())
        mdata.setText(", ".join(nodes))
        return mdata


def main():
    app = QApplication(sys.argv)
    client = Window()
    client.show()
    sys.exit(app.exec_())


if __name__ == "__main__":
    main()


1		#! /usr/bin/env python3
2
3		import sys
4		from datetime import datetime
5		from enum import Enum
6
7		from PyQt5.QtCore import pyqtSignal, QTimer, Qt, QObject, QSettings, QModelIndex, QMimeData
8		from PyQt5.QtGui import QStandardItemModel, QStandardItem, QIcon
9		from PyQt5.QtWidgets import QMainWindow, QWidget, QApplication, QAbstractItemView, QMenu, QAction
10
11		from opcua import ua
12		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		# accept drops
56		self.model.canDropMimeData = self.canDropMimeData
57		self.model.dropMimeData = self.dropMimeData
58
59		def canDropMimeData(self, mdata, action, row, column, parent):
60		return True
61
62		def dropMimeData(self, mdata, action, row, column, parent):
63		node = self.uaclient.client.get_node(mdata.text())
64		print("SUB 1", mdata.text(), node)
65		self._subscribe(node)
66		return True
67
68
69		def clear(self):
70		self._subscribed_nodes = []
71		self.model.clear()
72
73		def _subscribe(self, node=None):
74		print("SUB", node)
75		if not node:
76		node = self.window.get_current_node()
77		if node is None:
78		return
79		if node in self._subscribed_nodes:
80		print("allready subscribed to event for node: ", node)
81		return
82		print("Subscribing to events for ", node)
83		self.window.ui.evDockWidget.raise_()
84		try:
85		self.uaclient.subscribe_events(node, self._handler)
86		except Exception as ex:
87		self.window.show_error(ex)
88		raise
89		else:
90		self._subscribed_nodes.append(node)
91
92		def _unsubscribe(self):
93		node = self.window.get_current_node()
94		if node is None:
95		return
96		self._subscribed_nodes.remove(node)
97		self.uaclient.unsubscribe_events(node)
98
99		def _update_event_model(self, event):
100		self.model.appendRow([QStandardItem(str(event))])
101
102
103		class DataChangeUI(object):
104
105	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...
106		self.window = window
107		self.uaclient = uaclient
108		self._subhandler = DataChangeHandler()
109		self._subscribed_nodes = []
110		self.model = QStandardItemModel()
111		self.window.ui.subView.setModel(self.model)
112		self.window.ui.subView.horizontalHeader().setSectionResizeMode(1)
113
114		self.window.ui.actionSubscribeDataChange.triggered.connect(self._subscribe)
115		self.window.ui.actionUnsubscribeDataChange.triggered.connect(self._unsubscribe)
116
117		# populate contextual menu
118		self.window.ui.treeView.addAction(self.window.ui.actionSubscribeDataChange)
119		self.window.ui.treeView.addAction(self.window.ui.actionUnsubscribeDataChange)
120
121		# handle subscriptions
122		self._subhandler.data_change_fired.connect(self._update_subscription_model, type=Qt.QueuedConnection)
123
124		# accept drops
125		self.model.canDropMimeData = self.canDropMimeData
126		self.model.dropMimeData = self.dropMimeData
127
128		def canDropMimeData(self, mdata, action, row, column, parent):
129		return True
130
131		def dropMimeData(self, mdata, action, row, column, parent):
132		node = self.uaclient.client.get_node(mdata.text())
133		self._subscribe(node)
134		return True
135
136		def clear(self):
137		self._subscribed_nodes = []
138		self.model.clear()
139
140		def _subscribe(self, node=None):
141		if node is None:
142		node = self.window.get_current_node()
143		if node is None:
144		return
145		if node in self._subscribed_nodes:
146		print("allready subscribed to node: ", node)
147		return
148		self.model.setHorizontalHeaderLabels(["DisplayName", "Value", "Timestamp"])
149		text = str(node.get_display_name().Text)
150		row = [QStandardItem(text), QStandardItem("No Data yet"), QStandardItem("")]
151		row[0].setData(node)
152		self.model.appendRow(row)
153		self._subscribed_nodes.append(node)
154		self.window.ui.subDockWidget.raise_()
155		try:
156		self.uaclient.subscribe_datachange(node, self._subhandler)
157		except Exception as ex:
158		self.window.show_error(ex)
159		idx = self.model.indexFromItem(row[0])
160		self.model.takeRow(idx.row())
161		raise
162
163		def _unsubscribe(self):
164		node = self.window.get_current_node()
165		if node is None:
166		return
167		self.uaclient.unsubscribe_datachange(node)
168		self._subscribed_nodes.remove(node)
169		i = 0
170		while self.model.item(i):
171		item = self.model.item(i)
172		if item.data() == node:
173		self.model.removeRow(i)
174		i += 1
175
176		def _update_subscription_model(self, node, value, timestamp):
177		i = 0
178		while self.model.item(i):
179		item = self.model.item(i)
180		if item.data() == node:
181		it = self.model.item(i, 1)
182		it.setText(value)
183		it_ts = self.model.item(i, 2)
184		it_ts.setText(timestamp)
185		i += 1
186
187
188		class AttrsUI(object):
189
190		def __init__(self, window, uaclient):
191		self.window = window
192		self.uaclient = uaclient
193		self.model = QStandardItemModel()
194		self.window.ui.attrView.setModel(self.model)
195		self.window.ui.attrView.doubleClicked.connect(self.edit_attr)
196		self.model.itemChanged.connect(self.edit_attr_finished)
197		self.window.ui.attrView.header().setSectionResizeMode(1)
198
199		self.window.ui.treeView.activated.connect(self.show_attrs)
200		self.window.ui.treeView.clicked.connect(self.show_attrs)
201		self.window.ui.attrRefreshButton.clicked.connect(self.show_attrs)
202
203		# Context menu
204		self.window.ui.attrView.setContextMenuPolicy(Qt.CustomContextMenu)
205		self.window.ui.attrView.customContextMenuRequested.connect(self.showContextMenu)
206		copyaction = QAction("&Copy Value", self.model)
207		copyaction.triggered.connect(self._copy_value)
208		self._contextMenu = QMenu()
209		self._contextMenu.addAction(copyaction)
210
211		def _check_edit(self, item):
212		"""
213		filter only element we want to edit.
214		take either idx eller item as argument
215		"""
216		if item.column() != 1:
217		return False
218		name_item = self.model.item(item.row(), 0)
219		if name_item.text() != "Value":
220		return False
221		return True
222
223		def edit_attr(self, idx):
224		if not self._check_edit(idx):
225		return
226		attritem = self.model.item(idx.row(), 0)
227		if attritem.text() == "Value":
228		self.window.ui.attrView.edit(idx)
229
230		def edit_attr_finished(self, item):
231		if not self._check_edit(item):
232		return
233		try:
234		var = item.data()
235		val = item.text()
236		var = string_to_variant(val, var.VariantType)
237		self.current_node.set_value(var)
238		except Exception as ex:
239		self.window.show_error(ex)
240		raise
241		finally:
242		dv = self.current_node.get_data_value()
243		item.setText(variant_to_string(dv.Value))
244		name_item = self.model.item(item.row(), 0)
245		name_item.child(0, 1).setText(val_to_string(dv.ServerTimestamp))
246		name_item.child(1, 1).setText(val_to_string(dv.SourceTimestamp))
247
248		def showContextMenu(self, position):
249		item = self.get_current_item()
250		if item:
251		self._contextMenu.exec_(self.window.ui.attrView.mapToGlobal(position))
252
253		def get_current_item(self, col_idx=0):
254		idx = self.window.ui.attrView.currentIndex()
255		return self.model.item(idx.row(), col_idx)
256
257		def _copy_value(self, position):
258		it = self.get_current_item(1)
259		if it:
260		QApplication.clipboard().setText(it.text())
261
262		def clear(self):
263		self.model.clear()
264
265		def show_attrs(self, idx):
266		if not isinstance(idx, QModelIndex):
267		idx = None
268		self.current_node = self.window.get_current_node(idx)
269		self.model.clear()
270		if self.current_node:
271		self._show_attrs(self.current_node)
272		self.window.ui.attrView.expandAll()
273
274		def _show_attrs(self, node):
275		try:
276		attrs = self.uaclient.get_all_attrs(node)
277		except Exception as ex:
278		self.window.show_error(ex)
279		raise
280		self.model.setHorizontalHeaderLabels(['Attribute', 'Value', 'DataType'])
281		for name, dv in attrs:
282		if name == "DataType":
283		if isinstance(dv.Value.Value.Identifier, int) and dv.Value.Value.Identifier < 63:
284		string = ua.DataType_to_VariantType(dv.Value.Value).name
285		elif dv.Value.Value.Identifier in ua.ObjectIdNames:
286		string = ua.ObjectIdNames[dv.Value.Value.Identifier]
287		else:
288		string = dv.Value.Value.to_string()
289		elif name in ("AccessLevel", "UserAccessLevel"):
290		string = ",".join([e.name for e in ua.int_to_AccessLevel(dv.Value.Value)])
291		elif name in ("WriteMask", "UserWriteMask"):
292		string = ",".join([e.name for e in ua.int_to_WriteMask(dv.Value.Value)])
293		elif name in ("EventNotifier"):
294		string = ",".join([e.name for e in ua.int_to_EventNotifier(dv.Value.Value)])
295		else:
296		string = variant_to_string(dv.Value)
297		name_item = QStandardItem(name)
298		vitem = QStandardItem(string)
299		vitem.setData(dv.Value)
300		self.model.appendRow([name_item, vitem, QStandardItem(dv.Value.VariantType.name)])
301		if name == "Value":
302		string = val_to_string(dv.ServerTimestamp)
303		name_item.appendRow([QStandardItem("Server Timestamp"), QStandardItem(string), QStandardItem(ua.VariantType.DateTime.name)])
304		string = val_to_string(dv.SourceTimestamp)
305		name_item.appendRow([QStandardItem("Source Timestamp"), QStandardItem(string), QStandardItem(ua.VariantType.DateTime.name)])
306
307
308
309		class RefsUI(object):
310
311		def __init__(self, window, uaclient):
312		self.window = window
313		self.uaclient = uaclient
314		self.model = QStandardItemModel()
315		self.window.ui.refView.setModel(self.model)
316		self.window.ui.refView.horizontalHeader().setSectionResizeMode(1)
317
318		self.window.ui.treeView.activated.connect(self.show_refs)
319		self.window.ui.treeView.clicked.connect(self.show_refs)
320
321		def clear(self):
322		self.model.clear()
323
324		def show_refs(self, idx):
325		node = self.window.get_current_node(idx)
326		self.model.clear()
327		if node:
328		self._show_refs(node)
329
330		def _show_refs(self, node):
331		self.model.setHorizontalHeaderLabels(['ReferenceType', 'NodeId', "BrowseName", "TypeDefinition"])
332		try:
333		refs = self.uaclient.get_all_refs(node)
334		except Exception as ex:
335		self.window.show_error(ex)
336		raise
337		for ref in refs:
338		typename = ua.ObjectIdNames[ref.ReferenceTypeId.Identifier]
339		if ref.NodeId.NamespaceIndex == 0 and ref.NodeId.Identifier in ua.ObjectIdNames:
340		nodeid = ua.ObjectIdNames[ref.NodeId.Identifier]
341		else:
342		nodeid = ref.NodeId.to_string()
343		if ref.TypeDefinition.Identifier in ua.ObjectIdNames:
344		typedef = ua.ObjectIdNames[ref.TypeDefinition.Identifier]
345		else:
346		typedef = ref.TypeDefinition.to_string()
347		self.model.appendRow([QStandardItem(typename),
348		QStandardItem(nodeid),
349		QStandardItem(ref.BrowseName.to_string()),
350		QStandardItem(typedef)
351		])
352
353
354		class TreeUI(object):
355
356		def __init__(self, window, uaclient):
357		self.window = window
358		self.uaclient = uaclient
359		self.model = TreeViewModel(self.uaclient)
360		self.model.clear() # FIXME: do we need this?
361		self.model.error.connect(self.window.show_error)
362		self.window.ui.treeView.setModel(self.model)
363		#self.window.ui.treeView.setUniformRowHeights(True)
364		self.window.ui.treeView.setSelectionBehavior(QAbstractItemView.SelectRows)
365		self.window.ui.treeView.header().setSectionResizeMode(1)
366		self.window.ui.actionCopyPath.triggered.connect(self._copy_path)
367		self.window.ui.actionCopyNodeId.triggered.connect(self._copy_nodeid)
368		# add items to context menu
369		self.window.ui.treeView.addAction(self.window.ui.actionCopyPath)
370		self.window.ui.treeView.addAction(self.window.ui.actionCopyNodeId)
371
372		def clear(self):
373		self.model.clear()
374
375		def start(self):
376		self.model.clear()
377		self.model.add_item(*self.uaclient.get_root_node_and_desc())
378
379		def _copy_path(self):
380		path = self.get_current_path()
381		path = ",".join(path)
382		QApplication.clipboard().setText(path)
383
384		def _copy_nodeid(self):
385		node = self.get_current_node()
386		if node:
387		text = node.nodeid.to_string()
388		else:
389		text = ""
390		QApplication.clipboard().setText(text)
391
392		def get_current_path(self):
393		idx = self.window.ui.treeView.currentIndex()
394		idx = idx.sibling(idx.row(), 0)
395		it = self.model.itemFromIndex(idx)
396		path = []
397		while it and it.data():
398		node = it.data()
399		name = node.get_browse_name().to_string()
400		path.insert(0, name)
401		it = it.parent()
402		return path
403
404		def get_current_node(self, idx=None):
405		if idx is None:
406		idx = self.window.ui.treeView.currentIndex()
407		idx = idx.sibling(idx.row(), 0)
408		it = self.model.itemFromIndex(idx)
409		if not it:
410		return None
411		node = it.data()
412		if not node:
413		print("No node for item:", it, it.text())
414		return None
415		return node
416
417
418		class Window(QMainWindow):
419
420		def __init__(self):
421		QMainWindow.__init__(self)
422		self.ui = Ui_MainWindow()
423		self.ui.setupUi(self)
424		self.setWindowIcon(QIcon(":/network.svg"))
425
426		# fix stuff imposible to do in qtdesigner
427		# remove dock titlebar for addressbar
428		w = QWidget()
429		self.ui.addrDockWidget.setTitleBarWidget(w)
430		# tabify some docks
431		self.tabifyDockWidget(self.ui.evDockWidget, self.ui.subDockWidget)
432		self.tabifyDockWidget(self.ui.subDockWidget, self.ui.refDockWidget)
433
434		# we only show statusbar in case of errors
435		self.ui.statusBar.hide()
436
437		# load settings, seconds arg is default
438		self.settings = QSettings("FreeOpcUa", "FreeOpcUaClient")
439		self._address_list = self.settings.value("address_list", ["opc.tcp://localhost:4840", "opc.tcp://localhost:53530/OPCUA/SimulationServer/"])
440		self._address_list_max_count = int(self.settings.value("address_list_max_count", 10))
441
442		# init widgets
443		for addr in self._address_list:
444		self.ui.addrComboBox.insertItem(-1, addr)
445
446		self.uaclient = UaClient()
447
448		self.tree_ui = TreeUI(self, self.uaclient)
449		self.refs_ui = RefsUI(self, self.uaclient)
450		self.attrs_ui = AttrsUI(self, self.uaclient)
451		self.datachange_ui = DataChangeUI(self, self.uaclient)
452		self.event_ui = EventUI(self, self.uaclient)
453
454		self.resize(int(self.settings.value("main_window_width", 800)), int(self.settings.value("main_window_height", 600)))
455		self.restoreState(self.settings.value("main_window_state", b"", type="QByteArray"))
456
457		self.ui.connectButton.clicked.connect(self._connect)
458		self.ui.disconnectButton.clicked.connect(self._disconnect)
459		# self.ui.treeView.expanded.connect(self._fit)
460
461		self.ui.actionConnect.triggered.connect(self._connect)
462		self.ui.actionDisconnect.triggered.connect(self._disconnect)
463
464		self.ui.modeComboBox.addItem("None")
465		self.ui.modeComboBox.addItem("Sign")
466		self.ui.modeComboBox.addItem("SignAndEncrypt")
467
468		self.ui.policyComboBox.addItem("None")
469		self.ui.policyComboBox.addItem("Basic128RSA15")
470		self.ui.policyComboBox.addItem("Basic256")
471		self.ui.policyComboBox.addItem("Basic256SHA256")
472
473		def show_error(self, msg, level=1):
474		print("showing error: ", msg, level)
475		self.ui.statusBar.show()
476		self.ui.statusBar.setStyleSheet("QStatusBar { background-color : red; color : black; }")
477		self.ui.statusBar.showMessage(str(msg))
478		QTimer.singleShot(1500, self.ui.statusBar.hide)
479
480		def get_current_node(self, idx=None):
481		return self.tree_ui.get_current_node(idx)
482
483		def get_uaclient(self):
484		return self.uaclient
485
486		def _connect(self):
487		uri = self.ui.addrComboBox.currentText()
488		try:
489		self.uaclient.connect(uri)
490		except Exception as ex:
491		self.show_error(ex)
492		raise
493
494		self._update_address_list(uri)
495		self.tree_ui.start()
496
497		def _update_address_list(self, uri):
498		if uri == self._address_list[0]:
499		return
500		if uri in self._address_list:
501		self._address_list.remove(uri)
502		self._address_list.insert(0, uri)
503		if len(self._address_list) > self._address_list_max_count:
504		self._address_list.pop(-1)
505
506		def _disconnect(self):
507		try:
508		self.uaclient.disconnect()
509		except Exception as ex:
510		self.show_error(ex)
511		raise
512		finally:
513		self.tree_ui.clear()
514		self.refs_ui.clear()
515		self.attrs_ui.clear()
516		self.datachange_ui.clear()
517		self.event_ui.clear()
518
519		def closeEvent(self, event):
520		self.settings.setValue("main_window_width", self.size().width())
521		self.settings.setValue("main_window_height", self.size().height())
522		self.settings.setValue("main_window_state", self.saveState())
523		self.settings.setValue("address_list", self._address_list)
524		self._disconnect()
525		event.accept()
526
527
528		class TreeViewModel(QStandardItemModel):
529
530		error = pyqtSignal(str)
531
532		def __init__(self, uaclient):
533		super(TreeViewModel, self).__init__()
534		self.uaclient = uaclient
535		self._fetched = []
536
537		def clear(self):
538		QStandardItemModel.clear(self)
539		self._fetched = []
540		self.setHorizontalHeaderLabels(['DisplayName', "BrowseName", 'NodeId'])
541
542		def add_item(self, node, desc, parent=None):
543		item = [QStandardItem(desc.DisplayName.to_string()), QStandardItem(desc.BrowseName.to_string()), QStandardItem(desc.NodeId.to_string())]
544		if desc.NodeClass == ua.NodeClass.Object:
545		if desc.TypeDefinition == ua.TwoByteNodeId(ua.ObjectIds.FolderType):
546		item[0].setIcon(QIcon(":/folder.svg"))
547		else:
548		item[0].setIcon(QIcon(":/object.svg"))
549		elif desc.NodeClass == ua.NodeClass.Variable:
550		if desc.TypeDefinition == ua.TwoByteNodeId(ua.ObjectIds.PropertyType):
551		item[0].setIcon(QIcon(":/property.svg"))
552		else:
553		item[0].setIcon(QIcon(":/variable.svg"))
554		elif desc.NodeClass == ua.NodeClass.Method:
555		item[0].setIcon(QIcon(":/method.svg"))
556
557		item[0].setData(node)
558		if parent:
559		return parent.appendRow(item)
560		else:
561		return self.appendRow(item)
562
563		def canFetchMore(self, idx):
564		item = self.itemFromIndex(idx)
565		if not item:
566		return True
567		node = item.data()
568		if node not in self._fetched:
569		self._fetched.append(node)
570		return True
571		return False
572
573		def hasChildren(self, idx):
574		item = self.itemFromIndex(idx)
575		if not item:
576		return True
577		node = item.data()
578		if node in self._fetched:
579		return QStandardItemModel.hasChildren(self, idx)
580		return True
581
582		def fetchMore(self, idx):
583		parent = self.itemFromIndex(idx)
584		if parent:
585		self._fetchMore(parent)
586
587		def _fetchMore(self, parent):
588		try:
589		for node, attrs in self.uaclient.get_children(parent.data()).items():
590		self.add_item(node, attrs, parent)
591		except Exception as ex:
592		self.error.emit(ex)
593		raise
594
595		#def flags(self, idx):
596		#item = self.itemFromIndex(idx)
597		#flags = QStandardItemModel.flags(self, idx)
598		#if not item:
599		#return flags
600		#node = item.data()
601		#if node and node.get_node_class() == ua.NodeClass.Variable:
602		## FIXME not efficient to query, should be stored in data()
603		##print(1, flags)
604		#return flags \| Qt.ItemIsDropEnabled
605		#else:
606		#print(2, flags)
607		#return flags
608
609		#def mimeTypes(self):
610		#return ["application/vnd.text.list"]
611
612		def mimeData(self, idxs):
613		mdata = QMimeData()
614		nodes = []
615		for idx in idxs:
616		item = self.itemFromIndex(idx)
617		if item:
618		node = item.data()
619		if node:
620		nodes.append(node.nodeid.to_string())
621		mdata.setText(", ".join(nodes))
622		return mdata
623
624
625		def main():
626		app = QApplication(sys.argv)
627		client = Window()
628		client.show()
629		sys.exit(app.exec_())
630
631
632		if __name__ == "__main__":
633		main()
634

FreeOpcUa / opcua-client-gui

Push — master ( 26ea6b...5668b2 )

EventUI._unsubscribe() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like