Completed
Push — master ( b499cc...5902be )
by Olivier
08:44 queued 01:39
created

_instantiate_node()   F

Complexity

Conditions 14

Size

Total Lines 83

Duplication

Lines 0
Ratio 0 %

Code Coverage

Tests 39
CRAP Score 14.9663

Importance

Changes 3
Bugs 0 Features 0
Metric Value
cc 14
c 3
b 0
f 0
dl 0
loc 83
ccs 39
cts 47
cp 0.8298
crap 14.9663
rs 2

How to fix   Long Method    Complexity   

Long Method

Small methods make your code easier to understand, in particular if combined with a good name. Besides, if your method is small, finding a good name is usually much easier.

For example, if you find yourself adding comments to a method's body, this is usually a good sign to extract the commented part to a new method, and use the comment as a starting point when coming up with a good name for this new method.

Commonly applied refactorings include:

Complexity

Complex classes like _instantiate_node() 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
"""
2
Instantiate a new node and its child nodes from a node type.
3
"""
4
5 1
import logging
6
7 1
from opcua import Node
8 1
from opcua import ua
9 1
from opcua.common import ua_utils
10 1
from opcua.common.copy_node import _rdesc_from_node, _read_and_copy_attrs
11
12 1
logger = logging.getLogger(__name__)
13
14
15 1
def instantiate(parent, node_type, nodeid=None, bname=None, dname=None, idx=0, instantiate_optional=True):
16
    """
17
    instantiate a node type under a parent node.
18
    nodeid and browse name of new node can be specified, or just namespace index
19
    If they exists children of the node type, such as components, variables and
20
    properties are also instantiated
21
    """
22 1
    rdesc = _rdesc_from_node(parent, node_type)
23 1
    rdesc.TypeDefinition = node_type.nodeid
24
25 1
    if nodeid is None:
26 1
        nodeid = ua.NodeId(namespaceidx=idx)  # will trigger automatic node generation in namespace idx
27 1
    if bname is None:
28
        bname = rdesc.BrowseName
29 1
    elif isinstance(bname, str):
30 1
        bname = ua.QualifiedName.from_string(bname)
31
32 1
    nodeids = _instantiate_node(
33
        parent.server,
34
        Node(parent.server, rdesc.NodeId),
35
        parent.nodeid,
36
        rdesc,
37
        nodeid,
38
        bname,
39
        dname=dname,
40
        instantiate_optional=instantiate_optional)
41 1
    return [Node(parent.server, nid) for nid in nodeids]
42
43
44 1
def _instantiate_node(server,
45
                      node_type,
46
                      parentid,
47
                      rdesc,
48
                      nodeid,
49
                      bname,
50
                      dname=None,
51
                      recursive=True,
52
                      instantiate_optional=True):
53
    """
54
    instantiate a node type under parent
55
    """
56 1
    addnode = ua.AddNodesItem()
57 1
    addnode.RequestedNewNodeId = nodeid
58 1
    addnode.BrowseName = bname
59 1
    addnode.ParentNodeId = parentid
60 1
    addnode.ReferenceTypeId = rdesc.ReferenceTypeId
61 1
    addnode.TypeDefinition = rdesc.TypeDefinition
62
63 1
    if rdesc.NodeClass in (ua.NodeClass.Object, ua.NodeClass.ObjectType):
64 1
        addnode.NodeClass = ua.NodeClass.Object
65 1
        _read_and_copy_attrs(node_type, ua.ObjectAttributes(), addnode)
66
67 1
    elif rdesc.NodeClass in (ua.NodeClass.Variable, ua.NodeClass.VariableType):
68 1
        addnode.NodeClass = ua.NodeClass.Variable
69 1
        _read_and_copy_attrs(node_type, ua.VariableAttributes(), addnode)
70 1
    elif rdesc.NodeClass in (ua.NodeClass.Method, ):
71 1
        addnode.NodeClass = ua.NodeClass.Method
72 1
        _read_and_copy_attrs(node_type, ua.MethodAttributes(), addnode)
73
    elif rdesc.NodeClass in (ua.NodeClass.DataType, ):
74
        addnode.NodeClass = ua.NodeClass.DataType
75
        _read_and_copy_attrs(node_type, ua.DataTypeAttributes(), addnode)
76
    else:
77
        logger.error("Instantiate: Node class not supported: %s", rdesc.NodeClass)
78
        raise RuntimeError("Instantiate: Node class not supported")
79
        return
80 1
    if dname is not None:
81 1
        addnode.NodeAttributes.DisplayName = dname
82
83 1
    res = server.add_nodes([addnode])[0]
84 1
    added_nodes = [res.AddedNodeId]
85
86 1
    if recursive:
87 1
        parents = ua_utils.get_node_supertypes(node_type, includeitself=True)
88 1
        node = Node(server, res.AddedNodeId)
89 1
        for parent in parents:
90 1
            descs = parent.get_children_descriptions(includesubtypes=False)
91 1
            for c_rdesc in descs:
92
                # skip items that already exists, prefer the 'lowest' one in object hierarchy
93 1
                if not ua_utils.is_child_present(node, c_rdesc.BrowseName):
94
95 1
                    c_node_type = Node(server, c_rdesc.NodeId)
96 1
                    refs = c_node_type.get_referenced_nodes(refs=ua.ObjectIds.HasModellingRule)
97 1
                    if not refs:
98
                        # spec says to ignore nodes without modelling rules
99 1
                        logger.info("Instantiate: Skip node without modelling rule %s as part of %s", c_rdesc.BrowseName, addnode.BrowseName)
100 1
                        continue
101
                        # exclude nodes with optional ModellingRule if requested
102 1
                    if not instantiate_optional and refs[0].nodeid == ua.NodeId(ua.ObjectIds.ModellingRule_Optional):
103
                        logger.info("Instantiate: Skip optional node %s as part of %s", c_rdesc.BrowseName, addnode.BrowseName)
104
                        continue
105
106
                    # if root node being instantiated has a String NodeId, create the children with a String NodeId
107 1
                    if res.AddedNodeId.NodeIdType is ua.NodeIdType.String:
108 1
                        inst_nodeid = res.AddedNodeId.Identifier + "." + c_rdesc.BrowseName.Name
109 1
                        nodeids = _instantiate_node(
110
                            server,
111
                            c_node_type,
112
                            res.AddedNodeId,
113
                            c_rdesc,
114
                            nodeid=ua.NodeId(identifier=inst_nodeid, namespaceidx=res.AddedNodeId.NamespaceIndex),
115
                            bname=c_rdesc.BrowseName)
116
                    else:
117 1
                        nodeids = _instantiate_node(
118
                            server,
119
                            c_node_type,
120
                            res.AddedNodeId,
121
                            c_rdesc,
122
                            nodeid=ua.NodeId(namespaceidx=res.AddedNodeId.NamespaceIndex),
123
                            bname=c_rdesc.BrowseName)
124 1
                    added_nodes.extend(nodeids)
125
126
    return added_nodes
127