Completed
Push — master ( 482a3c...c19b0f )
by Olivier
04:21
created

_instantiate_node()   F

Complexity

Conditions 15

Size

Total Lines 62

Duplication

Lines 0
Ratio 0 %

Code Coverage

Tests 40
CRAP Score 15.4989

Importance

Changes 2
Bugs 0 Features 0
Metric Value
cc 15
c 2
b 0
f 0
dl 0
loc 62
ccs 40
cts 46
cp 0.8696
crap 15.4989
rs 2.9275

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