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

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