opcua.NodeData.__init__() - Code Metrics - Inspection of "fix test failed in python2 (windows)" - FreeOpcUa/python-opcua - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#86)

unknown

created 2015-12-21 19:57 UTC

opcua.NodeData.init() A

↳ Parent: opcua.NodeData

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	5
CRAP Score	1

Metric	Value
dl	0
loc	5
ccs	5
cts	5
cp	1
rs	9.4286
cc	1
crap	1

# -*- coding: utf-8 -*-

from threading import RLock
import logging
import pickle
from datetime import datetime

from opcua import ua
from opcua.users import User


class AttributeValue(object):

    def __init__(self, value):
        self.value = value
        self.value_callback = None
        self.datachange_callbacks = {}

    def __str__(self):
        return "AttributeValue({})".format(self.value)
    __repr__ = __str__


class NodeData(object):

    def __init__(self, nodeid):
        self.nodeid = nodeid
        self.attributes = {}
        self.references = []
        self.call = None

    def __str__(self):
        return "NodeData(id:{}, attrs:{}, refs:{})".format(self.nodeid, self.attributes, self.references)
    __repr__ = __str__


class AttributeService(object):

    def __init__(self, aspace):
        self.logger = logging.getLogger(__name__)
        self._aspace = aspace

    def read(self, params):
        self.logger.debug("read %s", params)
        res = []
        for readvalue in params.NodesToRead:
            res.append(self._aspace.get_attribute_value(readvalue.NodeId, readvalue.AttributeId))
        return res

    def write(self, params, user=User.Admin):
        self.logger.debug("write %s as user %s", params, user)
        res = []
        for writevalue in params.NodesToWrite:
            if user != User.Admin:
                if writevalue.AttributeId != ua.AttributeIds.Value:
                    res.append(ua.StatusCode(ua.StatusCodes.BadUserAccessDenied))
                    continue
                al = self._aspace.get_attribute_value(writevalue.NodeId, ua.AttributeIds.AccessLevel)
                ual = self._aspace.get_attribute_value(writevalue.NodeId, ua.AttributeIds.UserAccessLevel)
                if not al.Value.Value & ua.AccessLevelMask.CurrentWrite or not ual.Value.Value & ua.AccessLevelMask.CurrentWrite:
                    res.append(ua.StatusCode(ua.StatusCodes.BadUserAccessDenied))
                    continue
            res.append(self._aspace.set_attribute_value(writevalue.NodeId, writevalue.AttributeId, writevalue.Value))
        return res


class ViewService(object):

    def __init__(self, aspace):
        self.logger = logging.getLogger(__name__)
        self._aspace = aspace

    def browse(self, params):
        self.logger.debug("browse %s", params)
        res = []
        for desc in params.NodesToBrowse:
            res.append(self._browse(desc))
        return res

    def _browse(self, desc):
        res = ua.BrowseResult()
        if desc.NodeId not in self._aspace:
            res.StatusCode = ua.StatusCode(ua.StatusCodes.BadNodeIdInvalid)
            return res
        node = self._aspace[desc.NodeId]
        for ref in node.references:
            if not self._is_suitable_ref(desc, ref):
                continue
            res.References.append(ref)
        return res

    def _is_suitable_ref(self, desc, ref):
        if not self._suitable_direction(desc.BrowseDirection, ref.IsForward):
            self.logger.debug("%s is not suitable due to direction", ref)
            return False
        if not self._suitable_reftype(desc.ReferenceTypeId, ref.ReferenceTypeId, desc.IncludeSubtypes):
            self.logger.debug("%s is not suitable due to type", ref)
            return False
        if desc.NodeClassMask and ((desc.NodeClassMask & ref.NodeClass) == 0):
            self.logger.debug("%s is not suitable due to class", ref)
            return False
        self.logger.debug("%s is a suitable ref for desc %s", ref, desc)
        return True

    def _suitable_reftype(self, ref1, ref2, subtypes):
        """
        """
        if ref1.Identifier == ref2.Identifier:
            return True
        if not subtypes and ref2.Identifier == ua.ObjectIds.HasSubtype:
            return False
        oktypes = self._get_sub_ref(ref1)
        return ref2 in oktypes

    def _get_sub_ref(self, ref):
        res = []
        nodedata = self._aspace[ref]
        for ref in nodedata.references:
            if ref.ReferenceTypeId.Identifier == ua.ObjectIds.HasSubtype:
                res.append(ref.NodeId)
                res += self._get_sub_ref(ref.NodeId)
        return res

    def _suitable_direction(self, desc, isforward):
        if desc == ua.BrowseDirection.Both:
            return True
        if desc == ua.BrowseDirection.Forward and isforward:
            return True
        return False

    def translate_browsepaths_to_nodeids(self, browsepaths):
        self.logger.debug("translate browsepath: %s", browsepaths)
        results = []
        for path in browsepaths:
            results.append(self._translate_browsepath_to_nodeid(path))
        return results

    def _translate_browsepath_to_nodeid(self, path):
        self.logger.debug("looking at path: %s", path)
        res = ua.BrowsePathResult()
        if path.StartingNode not in self._aspace:
            res.StatusCode = ua.StatusCode(ua.StatusCodes.BadNodeIdInvalid)
            return res
        current = path.StartingNode
        for el in path.RelativePath.Elements:
            nodeid = self._find_element_in_node(el, current)
            if not nodeid:
                res.StatusCode = ua.StatusCode(ua.StatusCodes.BadNoMatch)
                return res
            current = nodeid
        target = ua.BrowsePathTarget()
        target.TargetId = current
        target.RemainingPathIndex = 4294967295
        res.Targets = [target]
        return res

    def _find_element_in_node(self, el, nodeid):
        nodedata = self._aspace[nodeid]
        for ref in nodedata.references:
            # FIXME: here we should check other arguments!!
            if ref.BrowseName == el.TargetName:
                return ref.NodeId
        self.logger.info("element %s was not found in node %s", el, nodeid)
        return None


class NodeManagementService(object):

    def __init__(self, aspace):
        self.logger = logging.getLogger(__name__)
        self._aspace = aspace

    def add_nodes(self, addnodeitems, user=User.Admin):
        results = []
        for item in addnodeitems:
            results.append(self._add_node(item, user))
        return results

    def _add_node(self, item, user):
        result = ua.AddNodesResult()

        if item.RequestedNewNodeId in self._aspace:
            self.logger.warning("AddNodeItem: node already exists")
            result.StatusCode = ua.StatusCode(ua.StatusCodes.BadNodeIdExists)
            return result
        nodedata = NodeData(item.RequestedNewNodeId)
        # add common attrs
        nodedata.attributes[ua.AttributeIds.NodeId] = AttributeValue(ua.DataValue(ua.Variant(item.RequestedNewNodeId, ua.VariantType.NodeId)))
        nodedata.attributes[ua.AttributeIds.BrowseName] = AttributeValue(ua.DataValue(ua.Variant(item.BrowseName, ua.VariantType.QualifiedName)))
        nodedata.attributes[ua.AttributeIds.NodeClass] = AttributeValue(ua.DataValue(ua.Variant(item.NodeClass, ua.VariantType.Int32)))
        # add requested attrs
        self._add_nodeattributes(item.NodeAttributes, nodedata)

        # add parent
        if item.ParentNodeId == ua.NodeId():
            #self.logger.warning("add_node: creating node %s without parent", item.RequestedNewNodeId)
            pass
        elif item.ParentNodeId not in self._aspace:
            #self.logger.warning("add_node: while adding node %s, requested parent node %s does not exists", item.RequestedNewNodeId, item.ParentNodeId)
            result.StatusCode = ua.StatusCode(ua.StatusCodes.BadParentNodeIdInvalid)
            return result
        else:
            if not user == User.Admin:
                result.StatusCode = ua.StatusCode(ua.StatusCodes.BadUserAccessDenied)
                return result

            desc = ua.ReferenceDescription()
            desc.ReferenceTypeId = item.ReferenceTypeId
            desc.NodeId = item.RequestedNewNodeId
            desc.NodeClass = item.NodeClass
            desc.BrowseName = item.BrowseName
            desc.DisplayName = ua.LocalizedText(item.BrowseName.Name)
            desc.TypeDefinition = item.TypeDefinition
            desc.IsForward = True
            self._aspace[item.ParentNodeId].references.append(desc)

        # now add our node to db
        self._aspace[item.RequestedNewNodeId] = nodedata

        # add type definition
        if item.TypeDefinition != ua.NodeId():
            addref = ua.AddReferencesItem()
            addref.SourceNodeId = item.RequestedNewNodeId
            addref.IsForward = True
            addref.ReferenceTypeId = ua.NodeId(ua.ObjectIds.HasTypeDefinition)
            addref.TargetNodeId = item.TypeDefinition
            addref.TargetNodeClass = ua.NodeClass.DataType
            self._add_reference(addref, user)

        result.StatusCode = ua.StatusCode()
        result.AddedNodeId = item.RequestedNewNodeId

        return result

    def add_references(self, refs, user=User.Admin):
        result = []
        for ref in refs:
            result.append(self._add_reference(ref, user))
        return result

    def _add_reference(self, addref, user):
        if addref.SourceNodeId not in self._aspace:
            return ua.StatusCode(ua.StatusCodes.BadSourceNodeIdInvalid)
        if addref.TargetNodeId not in self._aspace:
            return ua.StatusCode(ua.StatusCodes.BadTargetNodeIdInvalid)
        if not user == User.Admin:
            return ua.StatusCode(ua.StatusCodes.BadUserAccessDenied)
        rdesc = ua.ReferenceDescription()
        rdesc.ReferenceTypeId = addref.ReferenceTypeId
        rdesc.IsForward = addref.IsForward
        rdesc.NodeId = addref.TargetNodeId
        rdesc.NodeClass = addref.TargetNodeClass
        bname = self._aspace.get_attribute_value(addref.TargetNodeId, ua.AttributeIds.BrowseName).Value.Value
        if bname:
            rdesc.BrowseName = bname
        dname = self._aspace.get_attribute_value(addref.TargetNodeId, ua.AttributeIds.DisplayName).Value.Value
        if dname:
            rdesc.DisplayName = dname
        self._aspace[addref.SourceNodeId].references.append(rdesc)
        return ua.StatusCode()

    def _add_node_attr(self, item, nodedata, name, vtype=None):
        if item.SpecifiedAttributes & getattr(ua.NodeAttributesMask, name):
            dv = ua.DataValue(ua.Variant(getattr(item, name), vtype))
            dv.ServerTimestamp = datetime.now()
            dv.SourceTimestamp = datetime.now()
            nodedata.attributes[getattr(ua.AttributeIds, name)] = AttributeValue(dv)

    def _add_nodeattributes(self, item, nodedata):
        self._add_node_attr(item, nodedata, "AccessLevel", ua.VariantType.Byte)
        self._add_node_attr(item, nodedata, "ArrayDimensions", ua.VariantType.Int32)
        self._add_node_attr(item, nodedata, "BrowseName", ua.VariantType.QualifiedName)
        self._add_node_attr(item, nodedata, "ContainsNoLoops", ua.VariantType.Boolean)
        self._add_node_attr(item, nodedata, "DataType", ua.VariantType.NodeId)
        self._add_node_attr(item, nodedata, "Description", ua.VariantType.LocalizedText)
        self._add_node_attr(item, nodedata, "DisplayName", ua.VariantType.LocalizedText)
        self._add_node_attr(item, nodedata, "EventNotifier", ua.VariantType.Byte)
        self._add_node_attr(item, nodedata, "Executable", ua.VariantType.Boolean)
        self._add_node_attr(item, nodedata, "Historizing", ua.VariantType.Boolean)
        self._add_node_attr(item, nodedata, "InverseName", ua.VariantType.LocalizedText)
        self._add_node_attr(item, nodedata, "IsAbstract", ua.VariantType.Boolean)
        self._add_node_attr(item, nodedata, "MinimumSamplingInterval", ua.VariantType.Double)
        self._add_node_attr(item, nodedata, "NodeClass", ua.VariantType.UInt32)
        self._add_node_attr(item, nodedata, "NodeId", ua.VariantType.NodeId)
        self._add_node_attr(item, nodedata, "Symmetric", ua.VariantType.Boolean)
        self._add_node_attr(item, nodedata, "UserAccessLevel", ua.VariantType.Byte)
        self._add_node_attr(item, nodedata, "UserExecutable", ua.VariantType.Byte)
        self._add_node_attr(item, nodedata, "UserWriteMask", ua.VariantType.Byte)
        self._add_node_attr(item, nodedata, "ValueRank", ua.VariantType.Int32)
        self._add_node_attr(item, nodedata, "WriteMask", ua.VariantType.Byte)
        self._add_node_attr(item, nodedata, "UserWriteMask", ua.VariantType.Byte)
        self._add_node_attr(item, nodedata, "Value")


class MethodService(object):

    def __init__(self, aspace):
        self.logger = logging.getLogger(__name__)
        self._aspace = aspace

    def call(self, methods):
        results = []
        for method in methods:
            results.append(self._call(method))
        return results

    def _call(self, method):
        res = ua.CallMethodResult()
        if method.ObjectId not in self._aspace or method.MethodId not in self._aspace:
            res.StatusCode = ua.StatusCode(ua.StatusCodes.BadNodeIdInvalid)
        else:
            node = self._aspace[method.MethodId]
            if node.call is None:
                res.StatusCode = ua.StatusCode(ua.StatusCodes.BadNothingToDo)
            else:
                try:
                    res.OutputArguments = node.call(method.ObjectId, *method.InputArguments)
                    for _ in method.InputArguments:
                        res.InputArgumentResults.append(ua.StatusCode())
                except Exception:
                    self.logger.exception("Error executing method call %s, an exception was raised: ", method)
                    res.StatusCode = ua.StatusCode(ua.StatusCodes.BadUnexpectedError)
        return res


class AddressSpace(object):

    """
    The address space object stores all the nodes og the OPC-UA server
    and helper methods.
    The methods are threadsafe
    """

    def __init__(self):
        self.logger = logging.getLogger(__name__)
        self._nodes = {}
        self._lock = RLock()  # FIXME: should use multiple reader, one writter pattern
        self._datachange_callback_counter = 200
        self._handle_to_attribute_map = {}

    def __getitem__(self, nodeid):
        with self._lock:
            return self._nodes.__getitem__(nodeid)

    def __setitem__(self, nodeid, value):
        with self._lock:
            return self._nodes.__setitem__(nodeid, value)

    def __contains__(self, nodeid):
        with self._lock:
            return self._nodes.__contains__(nodeid)

    def dump(self, path):
        """
        dump address space as binary to file
        """
        with open(path, 'wb') as f:
            pickle.dump(self._nodes, f)

    def load(self, path):
        """
        load address space from file, overwritting everything current address space
        """
        with open(path, 'rb') as f:
            self._nodes = pickle.load(f)

    def get_attribute_value(self, nodeid, attr):
        with self._lock:
            #self.logger.debug("get attr val: %s %s", nodeid, attr)
            if nodeid not in self._nodes:
                dv = ua.DataValue()
                dv.StatusCode = ua.StatusCode(ua.StatusCodes.BadNodeIdUnknown)
                return dv
            node = self._nodes[nodeid]
            if attr not in node.attributes:
                dv = ua.DataValue()
                dv.StatusCode = ua.StatusCode(ua.StatusCodes.BadAttributeIdInvalid)
                return dv
            attval = node.attributes[attr]
            if attval.value_callback:
                return attval.value_callback()
            return attval.value

    def set_attribute_value(self, nodeid, attr, value):
        with self._lock:
            self.logger.debug("set attr val: %s %s %s", nodeid, attr, value)
            if nodeid not in self._nodes:
                return ua.StatusCode(ua.StatusCodes.BadNodeIdUnknown)
            node = self._nodes[nodeid]
            if attr not in node.attributes:
                return ua.StatusCode(ua.StatusCodes.BadAttributeIdInvalid)
            if not value.SourceTimestamp:
                value.SourceTimestamp = datetime.now()
            if not value.ServerTimestamp:
                value.ServerTimestamp = datetime.now()

            attval = node.attributes[attr]
            old = attval.value
            attval.value = value
            if old.Value != value.Value:  # only send call callback when a value change has happend
                for k, v in attval.datachange_callbacks.items():
                    try:
                        v(k, value)
                    except Exception as ex:
                        self.logger.exception("Error calling datachange callback %s, %s, %s", k, v, ex)
            return ua.StatusCode()

    def add_datachange_callback(self, nodeid, attr, callback):
        with self._lock:
            self.logger.debug("set attr callback: %s %s %s", nodeid, attr, callback)
            if nodeid not in self._nodes:
                return ua.StatusCode(ua.StatusCodes.BadNodeIdUnknown), 0
            node = self._nodes[nodeid]
            if attr not in node.attributes:
                return ua.StatusCode(ua.StatusCodes.BadAttributeIdInvalid), 0
            attval = node.attributes[attr]
            self._datachange_callback_counter += 1
            handle = self._datachange_callback_counter
            attval.datachange_callbacks[handle] = callback
            self._handle_to_attribute_map[handle] = (nodeid, attr)
            return ua.StatusCode(), handle

    def delete_datachange_callback(self, handle):
        with self._lock:
            nodeid, attr = self._handle_to_attribute_map.pop(handle)
            self._nodes[nodeid].attributes[attr].datachange_callbacks.pop(handle)

    def add_method_callback(self, methodid, callback):
        with self._lock:
            node = self._nodes[methodid]
            node.call = callback


1		# -- coding: utf-8 --
2
3	1	from threading import RLock
4	1	import logging
5	1	import pickle
6	1	from datetime import datetime
7
8	1	from opcua import ua
9	1	from opcua.users import User
10
11
12	1	class AttributeValue(object):
13
14	1	def __init__(self, value):
15	1	self.value = value
16	1	self.value_callback = None
17	1	self.datachange_callbacks = {}
18
19	1	def __str__(self):
20		return "AttributeValue({})".format(self.value)
21	1	__repr__ = __str__
22
23
24	1	class NodeData(object):
25
26	1	def __init__(self, nodeid):
27	1	self.nodeid = nodeid
28	1	self.attributes = {}
29	1	self.references = []
30	1	self.call = None
31
32	1	def __str__(self):
33		return "NodeData(id:{}, attrs:{}, refs:{})".format(self.nodeid, self.attributes, self.references)
34	1	__repr__ = __str__
35
36
37	1	class AttributeService(object):
38
39	1	def __init__(self, aspace):
40	1	self.logger = logging.getLogger(__name__)
41	1	self._aspace = aspace
42
43	1	def read(self, params):
44	1	self.logger.debug("read %s", params)
45	1	res = []
46	1	for readvalue in params.NodesToRead:
47	1	res.append(self._aspace.get_attribute_value(readvalue.NodeId, readvalue.AttributeId))
48	1	return res
49
50	1	def write(self, params, user=User.Admin):
51	1	self.logger.debug("write %s as user %s", params, user)
52	1	res = []
53	1	for writevalue in params.NodesToWrite:
54	1	if user != User.Admin:
55	1	if writevalue.AttributeId != ua.AttributeIds.Value:
56	1	res.append(ua.StatusCode(ua.StatusCodes.BadUserAccessDenied))
57	1	continue
58	1	al = self._aspace.get_attribute_value(writevalue.NodeId, ua.AttributeIds.AccessLevel)
59	1	ual = self._aspace.get_attribute_value(writevalue.NodeId, ua.AttributeIds.UserAccessLevel)
60	1	if not al.Value.Value & ua.AccessLevelMask.CurrentWrite or not ual.Value.Value & ua.AccessLevelMask.CurrentWrite:
61	1	res.append(ua.StatusCode(ua.StatusCodes.BadUserAccessDenied))
62	1	continue
63	1	res.append(self._aspace.set_attribute_value(writevalue.NodeId, writevalue.AttributeId, writevalue.Value))
64	1	return res
65
66
67	1	class ViewService(object):
68
69	1	def __init__(self, aspace):
70	1	self.logger = logging.getLogger(__name__)
71	1	self._aspace = aspace
72
73	1	def browse(self, params):
74	1	self.logger.debug("browse %s", params)
75	1	res = []
76	1	for desc in params.NodesToBrowse:
77	1	res.append(self._browse(desc))
78	1	return res
79
80	1	def _browse(self, desc):
81	1	res = ua.BrowseResult()
82	1	if desc.NodeId not in self._aspace:
83		res.StatusCode = ua.StatusCode(ua.StatusCodes.BadNodeIdInvalid)
84		return res
85	1	node = self._aspace[desc.NodeId]
86	1	for ref in node.references:
87	1	if not self._is_suitable_ref(desc, ref):
88	1	continue
89	1	res.References.append(ref)
90	1	return res
91
92	1	def _is_suitable_ref(self, desc, ref):
93	1	if not self._suitable_direction(desc.BrowseDirection, ref.IsForward):
94		self.logger.debug("%s is not suitable due to direction", ref)
95		return False
96	1	if not self._suitable_reftype(desc.ReferenceTypeId, ref.ReferenceTypeId, desc.IncludeSubtypes):
97	1	self.logger.debug("%s is not suitable due to type", ref)
98	1	return False
99	1	if desc.NodeClassMask and ((desc.NodeClassMask & ref.NodeClass) == 0):
100	1	self.logger.debug("%s is not suitable due to class", ref)
101	1	return False
102	1	self.logger.debug("%s is a suitable ref for desc %s", ref, desc)
103	1	return True
104
105	1	def _suitable_reftype(self, ref1, ref2, subtypes):
106		"""
107		"""
108	1	if ref1.Identifier == ref2.Identifier:
109	1	return True
110	1	if not subtypes and ref2.Identifier == ua.ObjectIds.HasSubtype:
111		return False
112	1	oktypes = self._get_sub_ref(ref1)
113	1	return ref2 in oktypes
114
115	1	def _get_sub_ref(self, ref):
116	1	res = []
117	1	nodedata = self._aspace[ref]
118	1	for ref in nodedata.references:
119	1	if ref.ReferenceTypeId.Identifier == ua.ObjectIds.HasSubtype:
120	1	res.append(ref.NodeId)
121	1	res += self._get_sub_ref(ref.NodeId)
122	1	return res
123
124	1	def _suitable_direction(self, desc, isforward):
125	1	if desc == ua.BrowseDirection.Both:
126		return True
127	1	if desc == ua.BrowseDirection.Forward and isforward:
128	1	return True
129		return False
130
131	1	def translate_browsepaths_to_nodeids(self, browsepaths):
132	1	self.logger.debug("translate browsepath: %s", browsepaths)
133	1	results = []
134	1	for path in browsepaths:
135	1	results.append(self._translate_browsepath_to_nodeid(path))
136	1	return results
137
138	1	def _translate_browsepath_to_nodeid(self, path):
139	1	self.logger.debug("looking at path: %s", path)
140	1	res = ua.BrowsePathResult()
141	1	if path.StartingNode not in self._aspace:
142		res.StatusCode = ua.StatusCode(ua.StatusCodes.BadNodeIdInvalid)
143		return res
144	1	current = path.StartingNode
145	1	for el in path.RelativePath.Elements:
146	1	nodeid = self._find_element_in_node(el, current)
147	1	if not nodeid:
148	1	res.StatusCode = ua.StatusCode(ua.StatusCodes.BadNoMatch)
149	1	return res
150	1	current = nodeid
151	1	target = ua.BrowsePathTarget()
152	1	target.TargetId = current
153	1	target.RemainingPathIndex = 4294967295
154	1	res.Targets = [target]
155	1	return res
156
157	1	def _find_element_in_node(self, el, nodeid):
158	1	nodedata = self._aspace[nodeid]
159	1	for ref in nodedata.references:
160		# FIXME: here we should check other arguments!!
161	1	if ref.BrowseName == el.TargetName:
162	1	return ref.NodeId
163	1	self.logger.info("element %s was not found in node %s", el, nodeid)
164	1	return None
165
166
167	1	class NodeManagementService(object):
168
169	1	def __init__(self, aspace):
170	1	self.logger = logging.getLogger(__name__)
171	1	self._aspace = aspace
172
173	1	def add_nodes(self, addnodeitems, user=User.Admin):
174	1	results = []
175	1	for item in addnodeitems:
176	1	results.append(self._add_node(item, user))
177	1	return results
178
179	1	def _add_node(self, item, user):
180	1	result = ua.AddNodesResult()
181
182	1	if item.RequestedNewNodeId in self._aspace:
183	1	self.logger.warning("AddNodeItem: node already exists")
184	1	result.StatusCode = ua.StatusCode(ua.StatusCodes.BadNodeIdExists)
185	1	return result
186	1	nodedata = NodeData(item.RequestedNewNodeId)
187		# add common attrs
188	1	nodedata.attributes[ua.AttributeIds.NodeId] = AttributeValue(ua.DataValue(ua.Variant(item.RequestedNewNodeId, ua.VariantType.NodeId)))
189	1	nodedata.attributes[ua.AttributeIds.BrowseName] = AttributeValue(ua.DataValue(ua.Variant(item.BrowseName, ua.VariantType.QualifiedName)))
190	1	nodedata.attributes[ua.AttributeIds.NodeClass] = AttributeValue(ua.DataValue(ua.Variant(item.NodeClass, ua.VariantType.Int32)))
191		# add requested attrs
192	1	self._add_nodeattributes(item.NodeAttributes, nodedata)
193
194		# add parent
195	1	if item.ParentNodeId == ua.NodeId():
196		#self.logger.warning("add_node: creating node %s without parent", item.RequestedNewNodeId)
197	1	pass
198	1	elif item.ParentNodeId not in self._aspace:
199		#self.logger.warning("add_node: while adding node %s, requested parent node %s does not exists", item.RequestedNewNodeId, item.ParentNodeId)
200		result.StatusCode = ua.StatusCode(ua.StatusCodes.BadParentNodeIdInvalid)
201		return result
202		else:
203	1	if not user == User.Admin:
204	1	result.StatusCode = ua.StatusCode(ua.StatusCodes.BadUserAccessDenied)
205	1	return result
206
207	1	desc = ua.ReferenceDescription()
208	1	desc.ReferenceTypeId = item.ReferenceTypeId
209	1	desc.NodeId = item.RequestedNewNodeId
210	1	desc.NodeClass = item.NodeClass
211	1	desc.BrowseName = item.BrowseName
212	1	desc.DisplayName = ua.LocalizedText(item.BrowseName.Name)
213	1	desc.TypeDefinition = item.TypeDefinition
214	1	desc.IsForward = True
215	1	self._aspace[item.ParentNodeId].references.append(desc)
216
217		# now add our node to db
218	1	self._aspace[item.RequestedNewNodeId] = nodedata
219
220		# add type definition
221	1	if item.TypeDefinition != ua.NodeId():
222	1	addref = ua.AddReferencesItem()
223	1	addref.SourceNodeId = item.RequestedNewNodeId
224	1	addref.IsForward = True
225	1	addref.ReferenceTypeId = ua.NodeId(ua.ObjectIds.HasTypeDefinition)
226	1	addref.TargetNodeId = item.TypeDefinition
227	1	addref.TargetNodeClass = ua.NodeClass.DataType
228	1	self._add_reference(addref, user)
229
230	1	result.StatusCode = ua.StatusCode()
231	1	result.AddedNodeId = item.RequestedNewNodeId
232
233	1	return result
234
235	1	def add_references(self, refs, user=User.Admin):
236	1	result = []
237	1	for ref in refs:
238	1	result.append(self._add_reference(ref, user))
239	1	return result
240
241	1	def _add_reference(self, addref, user):
242	1	if addref.SourceNodeId not in self._aspace:
243		return ua.StatusCode(ua.StatusCodes.BadSourceNodeIdInvalid)
244	1	if addref.TargetNodeId not in self._aspace:
245	1	return ua.StatusCode(ua.StatusCodes.BadTargetNodeIdInvalid)
246	1	if not user == User.Admin:
247		return ua.StatusCode(ua.StatusCodes.BadUserAccessDenied)
248	1	rdesc = ua.ReferenceDescription()
249	1	rdesc.ReferenceTypeId = addref.ReferenceTypeId
250	1	rdesc.IsForward = addref.IsForward
251	1	rdesc.NodeId = addref.TargetNodeId
252	1	rdesc.NodeClass = addref.TargetNodeClass
253	1	bname = self._aspace.get_attribute_value(addref.TargetNodeId, ua.AttributeIds.BrowseName).Value.Value
254	1	if bname:
255	1	rdesc.BrowseName = bname
256	1	dname = self._aspace.get_attribute_value(addref.TargetNodeId, ua.AttributeIds.DisplayName).Value.Value
257	1	if dname:
258	1	rdesc.DisplayName = dname
259	1	self._aspace[addref.SourceNodeId].references.append(rdesc)
260	1	return ua.StatusCode()
261
262	1	def _add_node_attr(self, item, nodedata, name, vtype=None):
263	1	if item.SpecifiedAttributes & getattr(ua.NodeAttributesMask, name):
264	1	dv = ua.DataValue(ua.Variant(getattr(item, name), vtype))
265	1	dv.ServerTimestamp = datetime.now()
266	1	dv.SourceTimestamp = datetime.now()
267	1	nodedata.attributes[getattr(ua.AttributeIds, name)] = AttributeValue(dv)
268
269	1	def _add_nodeattributes(self, item, nodedata):
270	1	self._add_node_attr(item, nodedata, "AccessLevel", ua.VariantType.Byte)
271	1	self._add_node_attr(item, nodedata, "ArrayDimensions", ua.VariantType.Int32)
272	1	self._add_node_attr(item, nodedata, "BrowseName", ua.VariantType.QualifiedName)
273	1	self._add_node_attr(item, nodedata, "ContainsNoLoops", ua.VariantType.Boolean)
274	1	self._add_node_attr(item, nodedata, "DataType", ua.VariantType.NodeId)
275	1	self._add_node_attr(item, nodedata, "Description", ua.VariantType.LocalizedText)
276	1	self._add_node_attr(item, nodedata, "DisplayName", ua.VariantType.LocalizedText)
277	1	self._add_node_attr(item, nodedata, "EventNotifier", ua.VariantType.Byte)
278	1	self._add_node_attr(item, nodedata, "Executable", ua.VariantType.Boolean)
279	1	self._add_node_attr(item, nodedata, "Historizing", ua.VariantType.Boolean)
280	1	self._add_node_attr(item, nodedata, "InverseName", ua.VariantType.LocalizedText)
281	1	self._add_node_attr(item, nodedata, "IsAbstract", ua.VariantType.Boolean)
282	1	self._add_node_attr(item, nodedata, "MinimumSamplingInterval", ua.VariantType.Double)
283	1	self._add_node_attr(item, nodedata, "NodeClass", ua.VariantType.UInt32)
284	1	self._add_node_attr(item, nodedata, "NodeId", ua.VariantType.NodeId)
285	1	self._add_node_attr(item, nodedata, "Symmetric", ua.VariantType.Boolean)
286	1	self._add_node_attr(item, nodedata, "UserAccessLevel", ua.VariantType.Byte)
287	1	self._add_node_attr(item, nodedata, "UserExecutable", ua.VariantType.Byte)
288	1	self._add_node_attr(item, nodedata, "UserWriteMask", ua.VariantType.Byte)
289	1	self._add_node_attr(item, nodedata, "ValueRank", ua.VariantType.Int32)
290	1	self._add_node_attr(item, nodedata, "WriteMask", ua.VariantType.Byte)
291	1	self._add_node_attr(item, nodedata, "UserWriteMask", ua.VariantType.Byte)
292	1	self._add_node_attr(item, nodedata, "Value")
293
294
295	1	class MethodService(object):
296
297	1	def __init__(self, aspace):
298	1	self.logger = logging.getLogger(__name__)
299	1	self._aspace = aspace
300
301	1	def call(self, methods):
302	1	results = []
303	1	for method in methods:
304	1	results.append(self._call(method))
305	1	return results
306
307	1	def _call(self, method):
308	1	res = ua.CallMethodResult()
309	1	if method.ObjectId not in self._aspace or method.MethodId not in self._aspace:
310	1	res.StatusCode = ua.StatusCode(ua.StatusCodes.BadNodeIdInvalid)
311		else:
312	1	node = self._aspace[method.MethodId]
313	1	if node.call is None:
314		res.StatusCode = ua.StatusCode(ua.StatusCodes.BadNothingToDo)
315		else:
316	1	try:
317	1	res.OutputArguments = node.call(method.ObjectId, *method.InputArguments)
318	1	for _ in method.InputArguments:
319	1	res.InputArgumentResults.append(ua.StatusCode())
320	1	except Exception:
321	1	self.logger.exception("Error executing method call %s, an exception was raised: ", method)
322	1	res.StatusCode = ua.StatusCode(ua.StatusCodes.BadUnexpectedError)
323	1	return res
324
325
326	1	class AddressSpace(object):
327
328		"""
329		The address space object stores all the nodes og the OPC-UA server
330		and helper methods.
331		The methods are threadsafe
332		"""
333
334	1	def __init__(self):
335	1	self.logger = logging.getLogger(__name__)
336	1	self._nodes = {}
337	1	self._lock = RLock() # FIXME: should use multiple reader, one writter pattern
338	1	self._datachange_callback_counter = 200
339	1	self._handle_to_attribute_map = {}
340
341	1	def __getitem__(self, nodeid):
342	1	with self._lock:
343	1	return self._nodes.__getitem__(nodeid)
344
345	1	def __setitem__(self, nodeid, value):
346	1	with self._lock:
347	1	return self._nodes.__setitem__(nodeid, value)
348
349	1	def __contains__(self, nodeid):
350	1	with self._lock:
351	1	return self._nodes.__contains__(nodeid)
352
353	1	def dump(self, path):
354		"""
355		dump address space as binary to file
356		"""
357		with open(path, 'wb') as f:
358		pickle.dump(self._nodes, f)
359
360	1	def load(self, path):
361		"""
362		load address space from file, overwritting everything current address space
363		"""
364		with open(path, 'rb') as f:
365		self._nodes = pickle.load(f)
366
367	1	def get_attribute_value(self, nodeid, attr):
368	1	with self._lock:
369		#self.logger.debug("get attr val: %s %s", nodeid, attr)
370	1	if nodeid not in self._nodes:
371	1	dv = ua.DataValue()
372	1	dv.StatusCode = ua.StatusCode(ua.StatusCodes.BadNodeIdUnknown)
373	1	return dv
374	1	node = self._nodes[nodeid]
375	1	if attr not in node.attributes:
376		dv = ua.DataValue()
377		dv.StatusCode = ua.StatusCode(ua.StatusCodes.BadAttributeIdInvalid)
378		return dv
379	1	attval = node.attributes[attr]
380	1	if attval.value_callback:
381		return attval.value_callback()
382	1	return attval.value
383
384	1	def set_attribute_value(self, nodeid, attr, value):
385	1	with self._lock:
386	1	self.logger.debug("set attr val: %s %s %s", nodeid, attr, value)
387	1	if nodeid not in self._nodes:
388	1	return ua.StatusCode(ua.StatusCodes.BadNodeIdUnknown)
389	1	node = self._nodes[nodeid]
390	1	if attr not in node.attributes:
391	1	return ua.StatusCode(ua.StatusCodes.BadAttributeIdInvalid)
392	1	if not value.SourceTimestamp:
393	1	value.SourceTimestamp = datetime.now()
394	1	if not value.ServerTimestamp:
395	1	value.ServerTimestamp = datetime.now()
396
397	1	attval = node.attributes[attr]
398	1	old = attval.value
399	1	attval.value = value
400	1	if old.Value != value.Value: # only send call callback when a value change has happend
401	1	for k, v in attval.datachange_callbacks.items():
402	1	try:
403	1	v(k, value)
404		except Exception as ex:
405		self.logger.exception("Error calling datachange callback %s, %s, %s", k, v, ex)
406	1	return ua.StatusCode()
407
408	1	def add_datachange_callback(self, nodeid, attr, callback):
409	1	with self._lock:
410	1	self.logger.debug("set attr callback: %s %s %s", nodeid, attr, callback)
411	1	if nodeid not in self._nodes:
412		return ua.StatusCode(ua.StatusCodes.BadNodeIdUnknown), 0
413	1	node = self._nodes[nodeid]
414	1	if attr not in node.attributes:
415	1	return ua.StatusCode(ua.StatusCodes.BadAttributeIdInvalid), 0
416	1	attval = node.attributes[attr]
417	1	self._datachange_callback_counter += 1
418	1	handle = self._datachange_callback_counter
419	1	attval.datachange_callbacks[handle] = callback
420	1	self._handle_to_attribute_map[handle] = (nodeid, attr)
421	1	return ua.StatusCode(), handle
422
423	1	def delete_datachange_callback(self, handle):
424	1	with self._lock:
425	1	nodeid, attr = self._handle_to_attribute_map.pop(handle)
426	1	self._nodes[nodeid].attributes[attr].datachange_callbacks.pop(handle)
427
428	1	def add_method_callback(self, methodid, callback):
429	1	with self._lock:
430	1	node = self._nodes[methodid]
431		node.call = callback
432

FreeOpcUa / python-opcua

Pull Request — master (#86)

opcua.NodeData.__init__() A

Complexity

Size

Duplication

Code Coverage

Duplication Side-by-Side

Filter issues like

opcua.NodeData.init() A