Total Complexity | 107 |
Total Lines | 390 |
Duplicated Lines | 18.72 % |
Coverage | 82.51% |
Changes | 4 | ||
Bugs | 1 | Features | 0 |
Duplicate code is one of the most pungent code smells. A rule that is often used is to re-structure code once it is duplicated in three or more places.
Common duplication problems, and corresponding solutions are:
Complex classes like XmlImporter often do a lot of different things. To break such a class down, we need to identify a cohesive component within that class. A common approach to find such a component is to look for fields/methods that share the same prefixes, or suffixes.
Once you have determined the fields that belong together, you can apply the Extract Class refactoring. If the component makes sense as a sub-class, Extract Subclass is also a candidate, and is often faster.
1 | """ |
||
14 | 1 | class XmlImporter(object): |
|
15 | |||
16 | 1 | def __init__(self, server): |
|
17 | 1 | self.logger = logging.getLogger(__name__) |
|
18 | 1 | self.parser = None |
|
19 | 1 | self.server = server |
|
20 | 1 | self.namespaces = {} |
|
21 | 1 | self.aliases = {} |
|
22 | |||
23 | 1 | def _map_namespaces(self, namespaces_uris): |
|
24 | """ |
||
25 | creates a mapping between the namespaces in the xml file and in the server. |
||
26 | if not present the namespace is registered. |
||
27 | """ |
||
28 | 1 | namespaces = {} |
|
29 | 1 | for ns_index, ns_uri in enumerate(namespaces_uris): |
|
30 | 1 | ns_server_index = self.server.register_namespace(ns_uri) |
|
31 | 1 | namespaces[ns_index + 1] = ns_server_index |
|
32 | 1 | return namespaces |
|
33 | |||
34 | 1 | def _map_aliases(self, aliases): |
|
35 | """ |
||
36 | maps the import aliases to the correct namespaces |
||
37 | """ |
||
38 | 1 | aliases_mapped = {} |
|
39 | 1 | for alias, node_id in aliases.items(): |
|
40 | 1 | aliases_mapped[alias] = self._migrate_ns(self.to_nodeid(node_id)) |
|
41 | 1 | return aliases_mapped |
|
42 | |||
43 | 1 | def import_xml(self, xmlpath): |
|
44 | """ |
||
45 | import xml and return added nodes |
||
46 | """ |
||
47 | 1 | self.logger.info("Importing XML file %s", xmlpath) |
|
48 | 1 | self.parser = xmlparser.XMLParser(xmlpath) |
|
49 | |||
50 | 1 | dnodes = self.parser.get_node_datas() |
|
51 | 1 | dnodes = self.make_objects(dnodes) |
|
52 | |||
53 | 1 | self.namespaces = self._map_namespaces(self.parser.get_used_namespaces()) |
|
54 | 1 | self.aliases = self._map_aliases(self.parser.get_aliases()) |
|
55 | |||
56 | 1 | nodes_parsed = self._sort_nodes_by_parentid(dnodes) |
|
57 | |||
58 | 1 | nodes = [] |
|
59 | 1 | for nodedata in nodes_parsed: # self.parser: |
|
60 | 1 | if nodedata.nodetype == 'UAObject': |
|
61 | 1 | node = self.add_object(nodedata) |
|
62 | 1 | elif nodedata.nodetype == 'UAObjectType': |
|
63 | 1 | node = self.add_object_type(nodedata) |
|
64 | 1 | elif nodedata.nodetype == 'UAVariable': |
|
65 | 1 | node = self.add_variable(nodedata) |
|
66 | 1 | elif nodedata.nodetype == 'UAVariableType': |
|
67 | node = self.add_variable_type(nodedata) |
||
68 | 1 | elif nodedata.nodetype == 'UAReferenceType': |
|
69 | node = self.add_reference_type(nodedata) |
||
70 | 1 | elif nodedata.nodetype == 'UADataType': |
|
71 | 1 | node = self.add_datatype(nodedata) |
|
72 | 1 | elif nodedata.nodetype == 'UAMethod': |
|
73 | 1 | node = self.add_method(nodedata) |
|
74 | else: |
||
75 | self.logger.warning("Not implemented node type: %s ", nodedata.nodetype) |
||
76 | continue |
||
77 | 1 | nodes.append(node) |
|
78 | 1 | return nodes |
|
79 | |||
80 | 1 | def _add_node(self, node): |
|
81 | 1 | if isinstance(self.server, opcua.server.server.Server): |
|
82 | 1 | return self.server.iserver.isession.add_nodes([node]) |
|
83 | else: |
||
84 | 1 | return self.server.uaclient.add_nodes([node]) |
|
85 | |||
86 | 1 | def _add_references(self, refs): |
|
87 | 1 | if isinstance(self.server, opcua.server.server.Server): |
|
88 | 1 | return self.server.iserver.isession.add_references(refs) |
|
89 | else: |
||
90 | 1 | return self.server.uaclient.add_references(refs) |
|
91 | |||
92 | 1 | def make_objects(self, node_datas): |
|
93 | 1 | new_nodes = [] |
|
94 | 1 | for ndata in node_datas: |
|
95 | 1 | ndata.nodeid = ua.NodeId.from_string(ndata.nodeid) |
|
96 | 1 | ndata.browsename = ua.QualifiedName.from_string(ndata.browsename) |
|
97 | 1 | if ndata.parent: |
|
98 | 1 | ndata.parent = ua.NodeId.from_string(ndata.parent) |
|
99 | 1 | if ndata.parentlink: |
|
100 | 1 | ndata.parentlink = self.to_nodeid(ndata.parentlink) |
|
101 | 1 | if ndata.typedef: |
|
102 | 1 | ndata.typedef = self.to_nodeid(ndata.typedef) |
|
103 | 1 | new_nodes.append(ndata) |
|
104 | 1 | return new_nodes |
|
105 | |||
106 | 1 | def _migrate_ns(self, nodeid): |
|
107 | """ |
||
108 | Check if the index of nodeid or browsename given in the xml model file |
||
109 | must be converted to a already existing namespace id based on the files |
||
110 | namespace uri |
||
111 | |||
112 | :returns: NodeId (str) |
||
113 | """ |
||
114 | 1 | if nodeid.NamespaceIndex in self.namespaces: |
|
115 | 1 | nodeid = copy(nodeid) |
|
116 | 1 | nodeid.NamespaceIndex = self.namespaces[nodeid.NamespaceIndex] |
|
117 | 1 | return nodeid |
|
118 | |||
119 | 1 | def _get_node(self, obj): |
|
120 | 1 | node = ua.AddNodesItem() |
|
121 | 1 | node.RequestedNewNodeId = self._migrate_ns(obj.nodeid) |
|
122 | 1 | node.BrowseName = self._migrate_ns(obj.browsename) |
|
123 | 1 | self.logger.info("Importing xml node (%s, %s) as (%s %s)", obj.browsename, obj.nodeid, node.BrowseName, node.RequestedNewNodeId) |
|
124 | 1 | node.NodeClass = getattr(ua.NodeClass, obj.nodetype[2:]) |
|
125 | 1 | if obj.parent: |
|
126 | 1 | node.ParentNodeId = self._migrate_ns(obj.parent) |
|
127 | 1 | if obj.parentlink: |
|
128 | 1 | node.ReferenceTypeId = self._migrate_ns(obj.parentlink) |
|
129 | 1 | if obj.typedef: |
|
130 | 1 | node.TypeDefinition = self._migrate_ns(obj.typedef) |
|
131 | 1 | return node |
|
132 | |||
133 | 1 | def to_nodeid(self, nodeid): |
|
134 | 1 | if isinstance(nodeid, ua.NodeId): |
|
135 | return nodeid |
||
136 | 1 | elif not nodeid: |
|
137 | return ua.NodeId(ua.ObjectIds.String) |
||
138 | 1 | elif "=" in nodeid: |
|
139 | 1 | return ua.NodeId.from_string(nodeid) |
|
140 | 1 | elif hasattr(ua.ObjectIds, nodeid): |
|
141 | 1 | return ua.NodeId(getattr(ua.ObjectIds, nodeid)) |
|
142 | else: |
||
143 | 1 | if nodeid in self.aliases: |
|
144 | 1 | return self.aliases[nodeid] |
|
145 | else: |
||
146 | return ua.NodeId(getattr(ua.ObjectIds, nodeid)) |
||
147 | |||
148 | 1 | def add_object(self, obj): |
|
149 | 1 | View Code Duplication | node = self._get_node(obj) |
|
|||
150 | 1 | attrs = ua.ObjectAttributes() |
|
151 | 1 | if obj.desc: |
|
152 | 1 | attrs.Description = ua.LocalizedText(obj.desc) |
|
153 | 1 | attrs.DisplayName = ua.LocalizedText(obj.displayname) |
|
154 | 1 | attrs.EventNotifier = obj.eventnotifier |
|
155 | 1 | node.NodeAttributes = attrs |
|
156 | 1 | res = self._add_node(node) |
|
157 | 1 | self._add_refs(obj) |
|
158 | 1 | res[0].StatusCode.check() |
|
159 | 1 | return res[0].AddedNodeId |
|
160 | |||
161 | 1 | def add_object_type(self, obj): |
|
162 | 1 | View Code Duplication | node = self._get_node(obj) |
163 | 1 | attrs = ua.ObjectTypeAttributes() |
|
164 | 1 | if obj.desc: |
|
165 | attrs.Description = ua.LocalizedText(obj.desc) |
||
166 | 1 | attrs.DisplayName = ua.LocalizedText(obj.displayname) |
|
167 | 1 | attrs.IsAbstract = obj.abstract |
|
168 | 1 | node.NodeAttributes = attrs |
|
169 | 1 | res = self._add_node(node) |
|
170 | 1 | self._add_refs(obj) |
|
171 | 1 | res[0].StatusCode.check() |
|
172 | 1 | return res[0].AddedNodeId |
|
173 | |||
174 | 1 | def add_variable(self, obj): |
|
175 | 1 | node = self._get_node(obj) |
|
176 | 1 | attrs = ua.VariableAttributes() |
|
177 | 1 | if obj.desc: |
|
178 | 1 | attrs.Description = ua.LocalizedText(obj.desc) |
|
179 | 1 | attrs.DisplayName = ua.LocalizedText(obj.displayname) |
|
180 | 1 | attrs.DataType = self.to_nodeid(obj.datatype) |
|
181 | 1 | if obj.value is not None: |
|
182 | 1 | attrs.Value = self._add_variable_value(obj,) |
|
183 | 1 | if obj.rank: |
|
184 | 1 | attrs.ValueRank = obj.rank |
|
185 | 1 | if obj.accesslevel: |
|
186 | attrs.AccessLevel = obj.accesslevel |
||
187 | 1 | if obj.useraccesslevel: |
|
188 | attrs.UserAccessLevel = obj.useraccesslevel |
||
189 | 1 | if obj.minsample: |
|
190 | attrs.MinimumSamplingInterval = obj.minsample |
||
191 | 1 | if obj.dimensions: |
|
192 | 1 | attrs.ArrayDimensions = obj.dimensions |
|
193 | 1 | node.NodeAttributes = attrs |
|
194 | 1 | res = self._add_node(node) |
|
195 | 1 | self._add_refs(obj) |
|
196 | 1 | res[0].StatusCode.check() |
|
197 | 1 | return res[0].AddedNodeId |
|
198 | |||
199 | 1 | def _make_ext_obj(self, obj): |
|
200 | 1 | ext = getattr(ua, obj.objname)() |
|
201 | 1 | for name, val in obj.body: |
|
202 | 1 | if isinstance(val, str): |
|
203 | raise Exception("Error val should a dict", name, val) |
||
204 | else: |
||
205 | 1 | for attname, v in val: |
|
206 | 1 | self._set_attr(ext, attname, v) |
|
207 | 1 | return ext |
|
208 | |||
209 | 1 | def _set_attr(self, obj, attname, val): |
|
210 | # tow possible values: |
||
211 | # either we get value directly |
||
212 | # or a dict if it s an object or a list |
||
213 | 1 | if isinstance(val, str): |
|
214 | 1 | pval = xmlparser.ua_type_to_python(val, obj.ua_types[attname]) |
|
215 | 1 | setattr(obj, attname, pval) |
|
216 | else: |
||
217 | # so we have either an object or a list... |
||
218 | 1 | obj2 = getattr(obj, attname) |
|
219 | 1 | if isinstance(obj2, ua.NodeId): # NodeId representation does not follow common rules!! |
|
220 | 1 | for attname2, v2 in val: |
|
221 | 1 | if attname2 == "Identifier": |
|
222 | 1 | obj2 = ua.NodeId.from_string(v2) |
|
223 | 1 | setattr(obj, attname, obj2) |
|
224 | 1 | break |
|
225 | 1 | elif not isinstance(obj2, ua.NodeId) and not hasattr(obj2, "ua_types"): |
|
226 | # we probably have a list |
||
227 | 1 | my_list = [] |
|
228 | 1 | for vtype, v2 in val: |
|
229 | 1 | my_list.append(xmlparser.ua_type_to_python(v2, vtype)) |
|
230 | 1 | setattr(obj, attname, my_list) |
|
231 | else: |
||
232 | 1 | for attname2, v2 in val: |
|
233 | 1 | self._set_attr(obj2, attname2, v2) |
|
234 | 1 | setattr(obj, attname, obj2) |
|
235 | |||
236 | 1 | def _add_variable_value(self, obj): |
|
237 | """ |
||
238 | Returns the value for a Variable based on the objects value type. |
||
239 | """ |
||
240 | 1 | self.logger.debug("Setting value with type %s and value %s", obj.valuetype, obj.value) |
|
241 | 1 | if obj.valuetype == 'ListOfExtensionObject': |
|
242 | 1 | values = [] |
|
243 | 1 | for ext in obj.value: |
|
244 | 1 | extobj = self._make_ext_obj(ext) |
|
245 | 1 | values.append(extobj) |
|
246 | 1 | return ua.Variant(values, ua.VariantType.ExtensionObject) |
|
247 | 1 | elif obj.valuetype == 'ListOfGuid': |
|
248 | 1 | return ua.Variant([ |
|
249 | uuid.UUID(guid) for guid in obj.value |
||
250 | ], getattr(ua.VariantType, obj.valuetype[6:])) |
||
251 | 1 | elif obj.valuetype.startswith("ListOf"): |
|
252 | 1 | vtype = obj.valuetype[6:] |
|
253 | 1 | if hasattr(ua.ua_binary.Primitives, vtype): |
|
254 | 1 | return ua.Variant(obj.value, getattr(ua.VariantType, vtype)) |
|
255 | else: |
||
256 | 1 | return ua.Variant([getattr(ua, vtype)(v) for v in obj.value]) |
|
257 | 1 | elif obj.valuetype == 'ExtensionObject': |
|
258 | 1 | extobj = self._make_ext_obj(obj.value) |
|
259 | 1 | return ua.Variant(extobj, getattr(ua.VariantType, obj.valuetype)) |
|
260 | 1 | elif obj.valuetype == 'Guid': |
|
261 | 1 | return ua.Variant(uuid.UUID(obj.value), getattr(ua.VariantType, obj.valuetype)) |
|
262 | 1 | elif obj.valuetype == 'LocalizedText': |
|
263 | 1 | ltext = ua.LocalizedText() |
|
264 | 1 | for name, val in obj.value: |
|
265 | 1 | if name == "Text": |
|
266 | 1 | ltext.Text = val.encode("utf-8") |
|
267 | else: |
||
268 | self.logger.warning("While parsing localizedText value, unkown element: %s with val: %s", name, val) |
||
269 | 1 | return ua.Variant(ltext, ua.VariantType.LocalizedText) |
|
270 | 1 | elif obj.valuetype == 'NodeId': |
|
271 | 1 | return ua.Variant(ua.NodeId.from_string(obj.value)) |
|
272 | else: |
||
273 | 1 | return ua.Variant(obj.value, getattr(ua.VariantType, obj.valuetype)) |
|
274 | |||
275 | 1 | def add_variable_type(self, obj): |
|
276 | node = self._get_node(obj) |
||
277 | attrs = ua.VariableTypeAttributes() |
||
278 | if obj.desc: |
||
279 | attrs.Description = ua.LocalizedText(obj.desc) |
||
280 | attrs.DisplayName = ua.LocalizedText(obj.displayname) |
||
281 | attrs.DataType = self.to_nodeid(obj.datatype) |
||
282 | if obj.value and len(obj.value) == 1: |
||
283 | attrs.Value = obj.value[0] |
||
284 | if obj.rank: |
||
285 | attrs.ValueRank = obj.rank |
||
286 | if obj.abstract: |
||
287 | attrs.IsAbstract = obj.abstract |
||
288 | if obj.dimensions: |
||
289 | attrs.ArrayDimensions = obj.dimensions |
||
290 | node.NodeAttributes = attrs |
||
291 | res = self._add_node(node) |
||
292 | self._add_refs(obj) |
||
293 | res[0].StatusCode.check() |
||
294 | return res[0].AddedNodeId |
||
295 | |||
296 | 1 | def add_method(self, obj): |
|
297 | 1 | View Code Duplication | node = self._get_node(obj) |
298 | 1 | attrs = ua.MethodAttributes() |
|
299 | 1 | if obj.desc: |
|
300 | 1 | attrs.Description = ua.LocalizedText(obj.desc) |
|
301 | 1 | attrs.DisplayName = ua.LocalizedText(obj.displayname) |
|
302 | 1 | if obj.accesslevel: |
|
303 | attrs.AccessLevel = obj.accesslevel |
||
304 | 1 | if obj.useraccesslevel: |
|
305 | attrs.UserAccessLevel = obj.useraccesslevel |
||
306 | 1 | if obj.minsample: |
|
307 | attrs.MinimumSamplingInterval = obj.minsample |
||
308 | 1 | if obj.dimensions: |
|
309 | attrs.ArrayDimensions = obj.dimensions |
||
310 | 1 | node.NodeAttributes = attrs |
|
311 | 1 | res = self._add_node(node) |
|
312 | 1 | self._add_refs(obj) |
|
313 | 1 | res[0].StatusCode.check() |
|
314 | 1 | return res[0].AddedNodeId |
|
315 | |||
316 | 1 | def add_reference_type(self, obj): |
|
317 | View Code Duplication | node = self._get_node(obj) |
|
318 | attrs = ua.ReferenceTypeAttributes() |
||
319 | if obj.desc: |
||
320 | attrs.Description = ua.LocalizedText(obj.desc) |
||
321 | attrs.DisplayName = ua.LocalizedText(obj.displayname) |
||
322 | if obj. inversename: |
||
323 | attrs.InverseName = ua.LocalizedText(obj.inversename) |
||
324 | if obj.abstract: |
||
325 | attrs.IsAbstract = obj.abstract |
||
326 | if obj.symmetric: |
||
327 | attrs.Symmetric = obj.symmetric |
||
328 | node.NodeAttributes = attrs |
||
329 | res = self._add_node(node) |
||
330 | self._add_refs(obj) |
||
331 | res[0].StatusCode.check() |
||
332 | return res[0].AddedNodeId |
||
333 | |||
334 | 1 | def add_datatype(self, obj): |
|
335 | 1 | View Code Duplication | node = self._get_node(obj) |
336 | 1 | attrs = ua.DataTypeAttributes() |
|
337 | 1 | if obj.desc: |
|
338 | 1 | attrs.Description = ua.LocalizedText(obj.desc) |
|
339 | 1 | attrs.DisplayName = ua.LocalizedText(obj.displayname) |
|
340 | 1 | if obj.abstract: |
|
341 | attrs.IsAbstract = obj.abstract |
||
342 | 1 | node.NodeAttributes = attrs |
|
343 | 1 | res = self._add_node(node) |
|
344 | 1 | self._add_refs(obj) |
|
345 | 1 | res[0].StatusCode.check() |
|
346 | 1 | return res[0].AddedNodeId |
|
347 | |||
348 | 1 | def _add_refs(self, obj): |
|
349 | 1 | if not obj.refs: |
|
350 | 1 | return |
|
351 | 1 | refs = [] |
|
352 | 1 | for data in obj.refs: |
|
353 | 1 | ref = ua.AddReferencesItem() |
|
354 | 1 | ref.IsForward = True |
|
355 | 1 | ref.ReferenceTypeId = self.to_nodeid(data.reftype) |
|
356 | 1 | ref.SourceNodeId = self._migrate_ns(obj.nodeid) |
|
357 | 1 | ref.TargetNodeClass = ua.NodeClass.DataType |
|
358 | 1 | ref.TargetNodeId = self._migrate_ns(self.to_nodeid(data.target)) |
|
359 | 1 | refs.append(ref) |
|
360 | 1 | self._add_references(refs) |
|
361 | |||
362 | 1 | def _sort_nodes_by_parentid(self, ndatas): |
|
363 | """ |
||
364 | Sort the list of nodes according their parent node in order to respect |
||
365 | the dependency between nodes. |
||
366 | |||
367 | :param nodes: list of NodeDataObjects |
||
368 | :returns: list of sorted nodes |
||
369 | """ |
||
370 | 1 | _ndatas = list(ndatas) |
|
371 | # list of node ids that are already sorted / inserted |
||
372 | 1 | sorted_nodes_ids = [] |
|
373 | # list of sorted nodes (i.e. XML Elements) |
||
374 | 1 | sorted_ndatas = [] |
|
375 | 1 | all_node_ids = [data.nodeid for data in ndatas] |
|
376 | # list of namespace indexes that are relevant for this import |
||
377 | # we can only respect ordering nodes for namespaces indexes that |
||
378 | # are defined in the xml file itself. Thus we assume that all other |
||
379 | # references namespaces are already known to the server and should |
||
380 | # not create any dependency problems (like "NodeNotFound") |
||
381 | 1 | while len(_ndatas) > 0: |
|
382 | 1 | pop_nodes = [] |
|
383 | 1 | for ndata in _ndatas: |
|
384 | # Insert nodes that |
||
385 | # (1) have no parent / parent_ns is None (e.g. namespace 0) |
||
386 | # (2) ns is not in list of relevant namespaces |
||
387 | 1 | if ndata.nodeid.NamespaceIndex not in self.namespaces or \ |
|
388 | ndata.parent is None or \ |
||
389 | ndata.parent not in all_node_ids: |
||
390 | 1 | sorted_ndatas.append(ndata) |
|
391 | 1 | sorted_nodes_ids.append(ndata.nodeid) |
|
392 | 1 | pop_nodes.append(ndata) |
|
393 | else: |
||
394 | # Check if the nodes parent is already in the list of |
||
395 | # inserted nodes |
||
396 | 1 | if ndata.parent in sorted_nodes_ids: |
|
397 | 1 | sorted_ndatas.append(ndata) |
|
398 | 1 | sorted_nodes_ids.append(ndata.nodeid) |
|
399 | 1 | pop_nodes.append(ndata) |
|
400 | # Remove inserted nodes from the list |
||
401 | 1 | for ndata in pop_nodes: |
|
402 | 1 | _ndatas.pop(_ndatas.index(ndata)) |
|
403 | return sorted_ndatas |
||
404 |