ModelManager._attr_written() - Code Metrics - Inspection of "split uamodeler class. Add tests" - FreeOpcUa/opcua-modeler - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( c71436...5d68fd )

by Olivier

created 2017-01-02 07:58 UTC

ModelManager._attr_written() A

↳ Parent: ModelManager

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
c	1
b	0
f	0
dl	0
loc	6
rs	9.4285
cc	3

import logging

from PyQt5.QtCore import pyqtSignal, QObject, QSettings

from opcua import ua
from opcua import copy_node
from opcua.common.instantiate import instantiate
from opcua.common.ua_utils import get_node_children

from uamodeler.server_manager import ServerManager


logger = logging.getLogger(__name__)


class ModelManager(QObject):

    """
    Manage our model. loads xml, start and close, add nodes
    No dialogs at that level, only api
    """

    error = pyqtSignal(Exception)

    def __init__(self, modeler):
        QObject.__init__(self, modeler)
        self.modeler = modeler
        self.server_mgr = ServerManager(self.modeler.ui.actionUseOpenUa)
        self.new_nodes = []  # the added nodes we will save
        self.current_path = None
        self.modified = False
        self.modeler.attrs_ui.attr_written.connect(self._attr_written)

    def delete_node(self, node):
        if node:
            nodes = get_node_children(node)
            for n in nodes:
                n.delete()
                if n in self.new_nodes:
                    self.new_nodes.remove(n)
            self.modeler.tree_ui.remove_current_item()

    def paste_node(self, node):
        parent = self.modeler.tree_ui.get_current_node()
        try:
            added_nodes = copy_node(parent, node)
        except Exception as ex:
            self.show_error(ex)
            raise
        self.new_nodes.extend(added_nodes)
        self.modeler.tree_ui.reload_current()
        self.modeler.show_refs()
        self.modified = True

    def close_model(self, force=False):
        if not force and self.modified:
            raise RuntimeError("Model is modified, use force to close it")
        self.modeler.disable_actions()
        self.modeler.tree_ui.clear()
        self.modeler.refs_ui.clear()
        self.modeler.attrs_ui.clear()
        self.modeler.idx_ui.clear()
        self.modeler.nodesets_ui.clear()
        self.current_path = None
        self.modified = False
        self.modeler.update_title("")
        self.server_mgr.stop_server()

    def new_model(self):
        if self.modified:
            raise RuntimeError("Model is modified, cannot create new model")
        del(self.new_nodes[:])  # empty list while keeping reference

        endpoint = "opc.tcp://0.0.0.0:48400/freeopcua/uamodeler/"
        logger.info("Starting server on %s", endpoint)
        self.server_mgr.start_server(endpoint)

        self.modeler.tree_ui.set_root_node(self.server_mgr.nodes.root)
        self.modeler.idx_ui.set_node(self.server_mgr.get_node(ua.ObjectIds.Server_NamespaceArray))
        self.modeler.nodesets_ui.set_server_mgr(self.server_mgr)
        self.modified = False
        self.modeler.enable_model_actions()
        self.current_path = None
        self.modeler.update_title("No Name")
        return True

    def import_xml(self, path):
        try:
            new_nodes = self.server_mgr.import_xml(path)
            self.new_nodes.extend([self.server_mgr.get_node(node) for node in new_nodes])
            self.modified = True
        except Exception as ex:
            self.show_error(ex)
            raise
        # we maybe should only reload the imported nodes
        self.modeler.tree_ui.reload()
        self.modeler.idx_ui.reload()
        return path

    def open_model(self, path):
        self.new_model()
        try:
            path = self.import_xml(path)
        except:
            self.close_model()
            raise
        self.modified = False
        self.current_path = path
        self.modeler.update_title(self.current_path)

    def save_model(self, path=None):
        if path is not None:
            self.current_path = path
            self.modeler.update_title(self.current_path)
        logger.info("Saving to %s", self.current_path)
        logger.info("Exporting  %s nodes: %s", len(self.new_nodes), self.new_nodes)
        logger.info("and namespaces: %s ", self.server_mgr.get_namespace_array()[1:])
        uris = self.server_mgr.get_namespace_array()[1:]
        try:
            self.server_mgr.export_xml(self.new_nodes, uris, self.current_path)
        except Exception as ex:
            self.show_error(ex)
            raise
        self.modified = False
        logger.info("%s saved", self.current_path)

    def _after_add(self, new_nodes):
        if isinstance(new_nodes, (list, tuple)):
            self.new_nodes.extend(new_nodes)
        else:
            self.new_nodes.append(new_nodes)
        self.modeler.tree_ui.reload_current()
        self.modeler.show_refs()
        self.modified = True

    def add_method(self, *args):
        logger.info("Creating method type with args: %s", args)
        parent = self.modeler.tree_ui.get_current_node()
        new_nodes = []
        new_node = parent.add_method(*args)
        new_nodes.append(new_node)
        new_nodes.extend(new_node.get_children())
        self._after_add(new_nodes)

    def add_object_type(self, *args):
        logger.info("Creating object type with args: %s", args)
        parent = self.modeler.tree_ui.get_current_node()
        new_node = parent.add_object_type(*args)
        self._after_add(new_node)

    def add_folder(self, *args):
        parent = self.modeler.tree_ui.get_current_node()
        logger.info("Creating folder with args: %s", args)
        new_node = parent.add_folder(*args)
        self._after_add(new_node)
        return new_node

    def add_object(self, *args):
        parent = self.modeler.tree_ui.get_current_node()
        logger.info("Creating object with args: %s", args)
        nodeid, bname, otype = args
        new_nodes = instantiate(parent, otype, bname=bname, nodeid=nodeid)
        self._after_add(new_nodes)
        return new_nodes

    def add_data_type(self, *args):
        parent = self.modeler.tree_ui.get_current_node()
        logger.info("Creating data type with args: %s", args)
        new_node = parent.add_data_type(*args)
        self._after_add(new_node)
        return new_node

    def add_variable(self, *args):
        parent = self.modeler.tree_ui.get_current_node()
        logger.info("Creating variable with args: %s", args)
        new_node = parent.add_variable(*args)
        self._after_add(new_node)
        return new_node

    def add_property(self, *args):
        parent = self.modeler.tree_ui.get_current_node()
        logger.info("Creating property with args: %s", args)
        self.settings.setValue("last_datatype", args[4])
        new_node = parent.add_property(*args)
        self._after_add(new_node)
        return new_node

    def add_variable_type(self, *args):
        parent = self.modeler.tree_ui.get_current_node()
        logger.info("Creating variable type with args: %s", args)
        nodeid, bname, datatype = args
        new_node = parent.add_variable_type(nodeid, bname, datatype.nodeid)
        self._after_add(new_node)
        return new_node

    def _attr_written(self, attr, dv):
        self.modeler_mgr.modified = True
        if attr == ua.AttributeIds.BrowseName:
            self.modeler.tree_ui.update_browse_name_current_item(dv.Value.Value)
        elif attr == ua.AttributeIds.DisplayName:
            self.modeler.tree_ui.update_display_name_current_item(dv.Value.Value)




1			import logging
2
3			from PyQt5.QtCore import pyqtSignal, QObject, QSettings
4
5			from opcua import ua
6			from opcua import copy_node
7			from opcua.common.instantiate import instantiate
8			from opcua.common.ua_utils import get_node_children
9
10			from uamodeler.server_manager import ServerManager
11
12
13			logger = logging.getLogger(__name__)
14
15
16			class ModelManager(QObject):
17
18			"""
19			Manage our model. loads xml, start and close, add nodes
20			No dialogs at that level, only api
21			"""
22
23			error = pyqtSignal(Exception)
24
25			def __init__(self, modeler):
26			QObject.__init__(self, modeler)
27			self.modeler = modeler
28			self.server_mgr = ServerManager(self.modeler.ui.actionUseOpenUa)
29			self.new_nodes = [] # the added nodes we will save
30			self.current_path = None
31			self.modified = False
32			self.modeler.attrs_ui.attr_written.connect(self._attr_written)
33
34			def delete_node(self, node):
35			if node:
36			nodes = get_node_children(node)
37			for n in nodes:
38			n.delete()
39			if n in self.new_nodes:
40			self.new_nodes.remove(n)
41			self.modeler.tree_ui.remove_current_item()
42
43			def paste_node(self, node):
44			parent = self.modeler.tree_ui.get_current_node()
45			try:
46			added_nodes = copy_node(parent, node)
47			except Exception as ex:
48			self.show_error(ex)
49			raise
50			self.new_nodes.extend(added_nodes)
51			self.modeler.tree_ui.reload_current()
52			self.modeler.show_refs()
53			self.modified = True
54
55			def close_model(self, force=False):
56			if not force and self.modified:
57			raise RuntimeError("Model is modified, use force to close it")
58			self.modeler.disable_actions()
59			self.modeler.tree_ui.clear()
60			self.modeler.refs_ui.clear()
61			self.modeler.attrs_ui.clear()
62			self.modeler.idx_ui.clear()
63			self.modeler.nodesets_ui.clear()
64			self.current_path = None
65			self.modified = False
66			self.modeler.update_title("")
67			self.server_mgr.stop_server()
68
69			def new_model(self):
70			if self.modified:
71			raise RuntimeError("Model is modified, cannot create new model")
72			del(self.new_nodes[:]) # empty list while keeping reference
73
74			endpoint = "opc.tcp://0.0.0.0:48400/freeopcua/uamodeler/"
75			logger.info("Starting server on %s", endpoint)
76			self.server_mgr.start_server(endpoint)
77
78			self.modeler.tree_ui.set_root_node(self.server_mgr.nodes.root)
79			self.modeler.idx_ui.set_node(self.server_mgr.get_node(ua.ObjectIds.Server_NamespaceArray))
80			self.modeler.nodesets_ui.set_server_mgr(self.server_mgr)
81			self.modified = False
82			self.modeler.enable_model_actions()
83			self.current_path = None
84			self.modeler.update_title("No Name")
85			return True
86
87			def import_xml(self, path):
88			try:
89			new_nodes = self.server_mgr.import_xml(path)
90			self.new_nodes.extend([self.server_mgr.get_node(node) for node in new_nodes])
91			self.modified = True
92			except Exception as ex:
93			self.show_error(ex)
94			raise
95			# we maybe should only reload the imported nodes
96			self.modeler.tree_ui.reload()
97			self.modeler.idx_ui.reload()
98			return path
99
100			def open_model(self, path):
101			self.new_model()
102			try:
103			path = self.import_xml(path)
104			except:
105			self.close_model()
106			raise
107			self.modified = False
108			self.current_path = path
109			self.modeler.update_title(self.current_path)
110
111			def save_model(self, path=None):
112			if path is not None:
113			self.current_path = path
114			self.modeler.update_title(self.current_path)
115			logger.info("Saving to %s", self.current_path)
116			logger.info("Exporting %s nodes: %s", len(self.new_nodes), self.new_nodes)
117			logger.info("and namespaces: %s ", self.server_mgr.get_namespace_array()[1:])
118			uris = self.server_mgr.get_namespace_array()[1:]
119			try:
120			self.server_mgr.export_xml(self.new_nodes, uris, self.current_path)
121			except Exception as ex:
122			self.show_error(ex)
123			raise
124			self.modified = False
125			logger.info("%s saved", self.current_path)
126
127			def _after_add(self, new_nodes):
128			if isinstance(new_nodes, (list, tuple)):
129			self.new_nodes.extend(new_nodes)
130			else:
131			self.new_nodes.append(new_nodes)
132			self.modeler.tree_ui.reload_current()
133			self.modeler.show_refs()
134			self.modified = True
135
136			def add_method(self, *args):
137			logger.info("Creating method type with args: %s", args)
138			parent = self.modeler.tree_ui.get_current_node()
139			new_nodes = []
140			new_node = parent.add_method(*args)
141			new_nodes.append(new_node)
142			new_nodes.extend(new_node.get_children())
143			self._after_add(new_nodes)
144
145			def add_object_type(self, *args):
146			logger.info("Creating object type with args: %s", args)
147			parent = self.modeler.tree_ui.get_current_node()
148			new_node = parent.add_object_type(*args)
149			self._after_add(new_node)
150
151			def add_folder(self, *args):
152			parent = self.modeler.tree_ui.get_current_node()
153			logger.info("Creating folder with args: %s", args)
154			new_node = parent.add_folder(*args)
155			self._after_add(new_node)
156			return new_node
157
158			def add_object(self, *args):
159			parent = self.modeler.tree_ui.get_current_node()
160			logger.info("Creating object with args: %s", args)
161			nodeid, bname, otype = args
162			new_nodes = instantiate(parent, otype, bname=bname, nodeid=nodeid)
163			self._after_add(new_nodes)
164			return new_nodes
165
166			def add_data_type(self, *args):
167			parent = self.modeler.tree_ui.get_current_node()
168			logger.info("Creating data type with args: %s", args)
169			new_node = parent.add_data_type(*args)
170			self._after_add(new_node)
171			return new_node
172
173			def add_variable(self, *args):
174			parent = self.modeler.tree_ui.get_current_node()
175			logger.info("Creating variable with args: %s", args)
176			new_node = parent.add_variable(*args)
177			self._after_add(new_node)
178			return new_node
179
180			def add_property(self, *args):
181			parent = self.modeler.tree_ui.get_current_node()
182			logger.info("Creating property with args: %s", args)
183			self.settings.setValue("last_datatype", args[4])
184			new_node = parent.add_property(*args)
185			self._after_add(new_node)
186			return new_node
187
188			def add_variable_type(self, *args):
189			parent = self.modeler.tree_ui.get_current_node()
190			logger.info("Creating variable type with args: %s", args)
191			nodeid, bname, datatype = args
192			new_node = parent.add_variable_type(nodeid, bname, datatype.nodeid)
193			self._after_add(new_node)
194			return new_node
195
196			def _attr_written(self, attr, dv):
197			self.modeler_mgr.modified = True
198			if attr == ua.AttributeIds.BrowseName:
199			self.modeler.tree_ui.update_browse_name_current_item(dv.Value.Value)
200			elif attr == ua.AttributeIds.DisplayName:
201			self.modeler.tree_ui.update_display_name_current_item(dv.Value.Value)
202
203
204

FreeOpcUa / opcua-modeler

Push — master ( c71436...5d68fd )

ModelManager._attr_written() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like