_instantiate_node() - Code Metrics - Inspection of "support multiple depths of inheritance for instant..." - FreeOpcUa/python-opcua - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#271)

unknown

created 2016-08-08 06:58 UTC

_instantiate_node() D

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	3
Bugs	0	Features	0

Metric	Value
cc	10
c	3
b	0
f	0
dl	0
loc	51
rs	4.2352

How to fix Long Method Complexity

"""
Instantiate a new node and its child nodes from a node type.
"""


from opcua import Node
from opcua import ua
from opcua.common import ua_utils 


def instantiate(parent, node_type, nodeid=None, bname=None, idx=0):
    """
    instantiate a node type under a parent node.
    nodeid and browse name of new node can be specified, or just namespace index
    If they exists children of the node type, such as components, variables and 
    properties are also instantiated
    """

    results = node_type.get_attributes([ua.AttributeIds.NodeClass, ua.AttributeIds.BrowseName, ua.AttributeIds.DisplayName])
    nclass, qname, dname = [res.Value.Value for res in results]

    rdesc = ua.ReferenceDescription()
    rdesc.NodeId = node_type.nodeid
    rdesc.BrowseName = qname
    rdesc.DisplayName = dname
    rdesc.NodeClass = nclass
    if parent.get_type_definition() == ua.NodeId(ua.ObjectIds.FolderType):
        rdesc.ReferenceTypeId = ua.NodeId(ua.ObjectIds.Organizes)
    else:
        rdesc.ReferenceTypeId = ua.NodeId(ua.ObjectIds.HasComponent)
    rdesc.TypeDefinition = node_type.nodeid
    if nodeid is None:
        nodeid = ua.NodeId(namespaceidx=idx)  # will trigger automatic node generation in namespace idx
    if bname is None:
        bname = rdesc.BrowseName
    elif isinstance(bname, str):
        bname = ua.QualifiedName.from_string(bname)    

    return _instantiate_node(parent.server, parent.nodeid, rdesc, nodeid, bname)


def _instantiate_node(server, parentid, rdesc, nodeid, bname, recursive=True):
    """
    instantiate a node type under parent
    """
    
    addnode = ua.AddNodesItem()
    addnode.RequestedNewNodeId = nodeid
    addnode.BrowseName = bname
    addnode.ParentNodeId = parentid
    addnode.ReferenceTypeId = rdesc.ReferenceTypeId
    addnode.TypeDefinition = rdesc.TypeDefinition

    node_type = Node(server, rdesc.NodeId)
    
    refs = node_type.get_referenced_nodes(refs=ua.ObjectIds.HasModellingRule)
    # skip optional elements
    if not(len(refs) == 1 and refs[0].nodeid == ua.NodeId(ua.ObjectIds.ModellingRule_Optional) ):
        
        if rdesc.NodeClass in (ua.NodeClass.Object, ua.NodeClass.ObjectType):
            addnode.NodeClass = ua.NodeClass.Object
            _read_and_copy_attrs(node_type, ua.ObjectAttributes(), addnode)

        elif rdesc.NodeClass in (ua.NodeClass.Variable, ua.NodeClass.VariableType):
            addnode.NodeClass = ua.NodeClass.Variable
            _read_and_copy_attrs(node_type, ua.VariableAttributes(), addnode)            
        elif rdesc.NodeClass in (ua.NodeClass.Method,):
            addnode.NodeClass = ua.NodeClass.Method
            _read_and_copy_attrs(node_type, ua.MethodAttributes(), addnode)
        else:
            print("Instantiate: Node class not supported: ", rdesc.NodeClass)
            return
    
        res = server.add_nodes([addnode])[0]
        
        if recursive:
            parents = ua_utils.get_node_supertypes(node_type, includeitself = True)
            node = Node(server, res.AddedNodeId)


                
            for parent in parents:
                descs = parent.get_children_descriptions(includesubtypes=False)
                for c_rdesc in descs:
                    # skip items that already exists, prefer the 'lowest' one in object hierarchy
                    if not ua_utils.is_child_present(node, c_rdesc.BrowseName):                    
                        _instantiate_node(server, res.AddedNodeId, c_rdesc, nodeid=ua.NodeId(namespaceidx=res.AddedNodeId.NamespaceIndex), bname=c_rdesc.BrowseName)
                    
        return Node(server, res.AddedNodeId)
    
    else:
        return None


def _read_and_copy_attrs(node_type, struct, addnode):
    names = [name for name in struct.__dict__.keys() if not name.startswith("_") and name not in ("BodyLength", "TypeId", "SpecifiedAttributes", "Encoding", "IsAbstract", "EventNotifier")]
    attrs = [getattr(ua.AttributeIds, name) for name in names]            
    for name in names:
        results = node_type.get_attributes(attrs)
    for idx, name in enumerate(names):
        if results[idx].StatusCode.is_good():
            if name == "Value":
                setattr(struct, name, results[idx].Value)
            else:
                setattr(struct, name, results[idx].Value.Value)
        else:
            print("Instantiate: while copying attributes from node type %s, attribute %s, statuscode is %s" % (node_type, name, results[idx].StatusCode))            
    addnode.NodeAttributes = struct


1			"""
2			Instantiate a new node and its child nodes from a node type.
3			"""
4
5
6			from opcua import Node
7			from opcua import ua
8			from opcua.common import ua_utils
9
10
11			def instantiate(parent, node_type, nodeid=None, bname=None, idx=0):
12			"""
13			instantiate a node type under a parent node.
14			nodeid and browse name of new node can be specified, or just namespace index
15			If they exists children of the node type, such as components, variables and
16			properties are also instantiated
17			"""
18
19			results = node_type.get_attributes([ua.AttributeIds.NodeClass, ua.AttributeIds.BrowseName, ua.AttributeIds.DisplayName])
20			nclass, qname, dname = [res.Value.Value for res in results]
21
22			rdesc = ua.ReferenceDescription()
23			rdesc.NodeId = node_type.nodeid
24			rdesc.BrowseName = qname
25			rdesc.DisplayName = dname
26			rdesc.NodeClass = nclass
27			if parent.get_type_definition() == ua.NodeId(ua.ObjectIds.FolderType):
28			rdesc.ReferenceTypeId = ua.NodeId(ua.ObjectIds.Organizes)
29			else:
30			rdesc.ReferenceTypeId = ua.NodeId(ua.ObjectIds.HasComponent)
31			rdesc.TypeDefinition = node_type.nodeid
32			if nodeid is None:
33			nodeid = ua.NodeId(namespaceidx=idx) # will trigger automatic node generation in namespace idx
34			if bname is None:
35			bname = rdesc.BrowseName
36			elif isinstance(bname, str):
37			bname = ua.QualifiedName.from_string(bname)
38
39			return _instantiate_node(parent.server, parent.nodeid, rdesc, nodeid, bname)
40
41
42			def _instantiate_node(server, parentid, rdesc, nodeid, bname, recursive=True):
43			"""
44			instantiate a node type under parent
45			"""
46
47			addnode = ua.AddNodesItem()
48			addnode.RequestedNewNodeId = nodeid
49			addnode.BrowseName = bname
50			addnode.ParentNodeId = parentid
51			addnode.ReferenceTypeId = rdesc.ReferenceTypeId
52			addnode.TypeDefinition = rdesc.TypeDefinition
53
54			node_type = Node(server, rdesc.NodeId)
55
56			refs = node_type.get_referenced_nodes(refs=ua.ObjectIds.HasModellingRule)
57			# skip optional elements
58			if not(len(refs) == 1 and refs[0].nodeid == ua.NodeId(ua.ObjectIds.ModellingRule_Optional) ):
59
60			if rdesc.NodeClass in (ua.NodeClass.Object, ua.NodeClass.ObjectType):
61			addnode.NodeClass = ua.NodeClass.Object
62			_read_and_copy_attrs(node_type, ua.ObjectAttributes(), addnode)
63
64			elif rdesc.NodeClass in (ua.NodeClass.Variable, ua.NodeClass.VariableType):
65			addnode.NodeClass = ua.NodeClass.Variable
66			_read_and_copy_attrs(node_type, ua.VariableAttributes(), addnode)
67			elif rdesc.NodeClass in (ua.NodeClass.Method,):
68			addnode.NodeClass = ua.NodeClass.Method
69			_read_and_copy_attrs(node_type, ua.MethodAttributes(), addnode)
70			else:
71			print("Instantiate: Node class not supported: ", rdesc.NodeClass)
72			return
73
74			res = server.add_nodes([addnode])[0]
75
76			if recursive:
77			parents = ua_utils.get_node_supertypes(node_type, includeitself = True)
78			node = Node(server, res.AddedNodeId)
79
80
81
82			for parent in parents:
83			descs = parent.get_children_descriptions(includesubtypes=False)
84			for c_rdesc in descs:
85			# skip items that already exists, prefer the 'lowest' one in object hierarchy
86			if not ua_utils.is_child_present(node, c_rdesc.BrowseName):
87			_instantiate_node(server, res.AddedNodeId, c_rdesc, nodeid=ua.NodeId(namespaceidx=res.AddedNodeId.NamespaceIndex), bname=c_rdesc.BrowseName)
88
89			return Node(server, res.AddedNodeId)
90
91			else:
92			return None
93
94
95			def _read_and_copy_attrs(node_type, struct, addnode):
96			names = [name for name in struct.__dict__.keys() if not name.startswith("_") and name not in ("BodyLength", "TypeId", "SpecifiedAttributes", "Encoding", "IsAbstract", "EventNotifier")]
97			attrs = [getattr(ua.AttributeIds, name) for name in names]
98			for name in names:
99			results = node_type.get_attributes(attrs)
100			for idx, name in enumerate(names):
101			if results[idx].StatusCode.is_good():
102			if name == "Value":
103			setattr(struct, name, results[idx].Value)
104			else:
105			setattr(struct, name, results[idx].Value.Value)
106			else:
107			print("Instantiate: while copying attributes from node type %s, attribute %s, statuscode is %s" % (node_type, name, results[idx].StatusCode))
108			addnode.NodeAttributes = struct
109

FreeOpcUa / python-opcua

Pull Request — master (#271)

_instantiate_node() D

Complexity

Size

Duplication

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like