Passed
Pull Request — master (#583)
by Olivier
05:50
created

_instantiate_node()   F

Complexity

Conditions 14

Size

Total Lines 81

Duplication

Lines 0
Ratio 0 %

Code Coverage

Tests 35
CRAP Score 16.1502

Importance

Changes 4
Bugs 0 Features 0
Metric Value
cc 14
c 4
b 0
f 0
dl 0
loc 81
ccs 35
cts 45
cp 0.7778
crap 16.1502
rs 2.0338

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
                    if not instantiate_optional:
97
                        # exclude nodes with optional ModellingRule if requested
98
                        refs = c_node_type.get_referenced_nodes(refs=ua.ObjectIds.HasModellingRule)
99
                        if len(refs) > 0 and refs[0].nodeid == ua.NodeId(ua.ObjectIds.ModellingRule_Optional):
100
                            logger.info("Instantiate: Skip optional node %s as part of %s", c_rdesc.BrowseName,
101
                                        addnode.BrowseName)
102
                            continue
103
104
                    # if root node being instantiated has a String NodeId, create the children with a String NodeId
105 1
                    if res.AddedNodeId.NodeIdType is ua.NodeIdType.String:
106 1
                        inst_nodeid = res.AddedNodeId.Identifier + "." + c_rdesc.BrowseName.Name
107 1
                        nodeids = _instantiate_node(
108
                            server,
109
                            c_node_type,
110
                            res.AddedNodeId,
111
                            c_rdesc,
112
                            nodeid=ua.NodeId(identifier=inst_nodeid, namespaceidx=res.AddedNodeId.NamespaceIndex),
113
                            bname=c_rdesc.BrowseName)
114
                    else:
115 1
                        nodeids = _instantiate_node(
116
                            server,
117
                            c_node_type,
118
                            res.AddedNodeId,
119
                            c_rdesc,
120
                            nodeid=ua.NodeId(namespaceidx=res.AddedNodeId.NamespaceIndex),
121
                            bname=c_rdesc.BrowseName)
122 1
                    added_nodes.extend(nodeids)
123
124
    return added_nodes
125