Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
type_name = 'CustomizedStruct2'
created_type = await srv.dict_builder.create_data_type(type_name)
assert isinstance(created_type, StructNode)
# Test data type node
type_node = srv.srv.get_node(created_type.data_type)
assert await type_node.get_browse_name() == ua.QualifiedName(type_name, srv.idx)
assert await type_node.get_node_class() == ua.NodeClass.DataType
assert (await type_node.get_parent()).nodeid == ua.NodeId(ua.ObjectIds.Structure, 0)
assert ua.NodeId(ua.ObjectIds.HasSubtype, 0) == (await type_node.get_references(refs=ua.ObjectIds.HasSubtype))[0].ReferenceTypeId
assert await type_node.get_display_name() == ua.LocalizedText(type_name)
# Test description node
n = srv.srv.get_node(srv.dict_builder.dict_id)
desc_node = await n.get_child(f"{srv.dict_builder._idx}:{type_name}")
assert await desc_node.get_browse_name() == ua.QualifiedName(type_name, srv.idx)
assert await desc_node.get_node_class() == ua.NodeClass.Variable
assert (await desc_node.get_parent()).nodeid == srv.dict_builder.dict_id
assert ua.NodeId(ua.ObjectIds.HasComponent, 0) == (await desc_node.get_references(refs=ua.ObjectIds.HasComponent))[0].ReferenceTypeId
assert await desc_node.get_type_definition() == ua.NodeId(ua.ObjectIds.DataTypeDescriptionType, 0)
assert await desc_node.get_display_name() == ua.LocalizedText(type_name)
assert await desc_node.get_data_type() == ua.NodeId(ua.ObjectIds.String)
assert await desc_node.get_value() == type_name
assert await desc_node.get_value_rank() == -1
# Test object node
obj_node = (await type_node.get_children(refs=ua.ObjectIds.HasEncoding))[0]
assert await obj_node.get_browse_name() == ua.QualifiedName('Default Binary', 0)
assert await obj_node.get_node_class() == ua.NodeClass.Object
assert (await obj_node.get_references(refs=ua.ObjectIds.HasEncoding))[0].NodeId == type_node.nodeid
assert ua.NodeId(ua.ObjectIds.HasEncoding, 0) == (await obj_node.get_references(refs=ua.ObjectIds.HasEncoding))[0].ReferenceTypeId
assert await obj_node.get_type_definition() == ua.NodeId(ua.ObjectIds.DataTypeEncodingType, 0)
desc_attributes = ua.VariableAttributes()
desc_attributes.DisplayName = ua.LocalizedText(type_name)
desc_attributes.DataType = ua.NodeId(ua.ObjectIds.String)
desc_attributes.Value = ua.Variant(name, ua.VariantType.String)
desc_attributes.ValueRank = -1
desc_node.NodeAttributes = desc_attributes
res = await self._session_server.add_nodes([desc_node])
description_node_id = res[0].AddedNodeId
added.append(description_node_id)
# create object node which the loaded python class should link to
obj_node = ua.AddNodesItem()
obj_node.RequestedNewNodeId = ua.NodeId(0, self._idx)
obj_node.BrowseName = ua.QualifiedName('Default Binary', 0)
obj_node.NodeClass = ua.NodeClass.Object
obj_node.ParentNodeId = data_type_node_id
obj_node.ReferenceTypeId = ua.NodeId(ua.ObjectIds.HasEncoding, 0)
obj_node.TypeDefinition = ua.NodeId(ua.ObjectIds.DataTypeEncodingType, 0)
obj_attributes = ua.ObjectAttributes()
obj_attributes.DisplayName = ua.LocalizedText('Default Binary')
obj_attributes.EventNotifier = 0
obj_node.NodeAttributes = obj_attributes
res = await self._session_server.add_nodes([obj_node])
bind_obj_node_id = res[0].AddedNodeId
added.append(bind_obj_node_id)
await self._link_nodes(bind_obj_node_id, data_type_node_id, description_node_id)
self._type_dictionary.append_struct(type_name)
return StructNode(self, data_type_node_id, type_name, added)
def get_variables(self):
"""
return variables of node.
properties are child nodes with a reference of type HasComponent and a NodeClass of Variable
"""
return self.get_children(refs=ua.ObjectIds.HasComponent, nodeclassmask=ua.NodeClass.Variable)
def get_methods(self):
"""
return methods of node.
properties are child nodes with a reference of type HasComponent and a NodeClass of Method
"""
return self.get_children(refs=ua.ObjectIds.HasComponent, nodeclassmask=ua.NodeClass.Method)
def _add_reference_no_check(self, sourcedata, addref):
rdesc = ua.ReferenceDescription()
rdesc.ReferenceTypeId = addref.ReferenceTypeId
rdesc.IsForward = addref.IsForward
rdesc.NodeId = addref.TargetNodeId
if addref.TargetNodeClass == ua.NodeClass.Unspecified:
rdesc.NodeClass = self._aspace.read_attribute_value(
addref.TargetNodeId, ua.AttributeIds.NodeClass).Value.Value
else:
rdesc.NodeClass = addref.TargetNodeClass
bname = self._aspace.read_attribute_value(addref.TargetNodeId, ua.AttributeIds.BrowseName).Value.Value
if bname:
rdesc.BrowseName = bname
dname = self._aspace.read_attribute_value(addref.TargetNodeId, ua.AttributeIds.DisplayName).Value.Value
if dname:
rdesc.DisplayName = dname
return self._add_unique_reference(sourcedata, rdesc)
def _lsprint_long(pnode, depth, indent=""):
if not indent:
print("{0:30} {1:25} {2:25} {3:10} {4:30} {5:25}".format("DisplayName", "NodeId", "BrowseName", "DataType", "Timestamp", "Value"))
print("")
for node in pnode.get_children():
attrs = node.read_attributes([ua.AttributeIds.DisplayName,
ua.AttributeIds.BrowseName,
ua.AttributeIds.NodeClass,
ua.AttributeIds.WriteMask,
ua.AttributeIds.UserWriteMask,
ua.AttributeIds.DataType,
ua.AttributeIds.Value])
name, bname, nclass, mask, umask, dtype, val = [attr.Value.Value for attr in attrs]
update = attrs[-1].ServerTimestamp
if nclass == ua.NodeClass.Variable:
print("{0}{1:30} {2:25} {3:25} {4:10} {5!s:30} {6!s:25}".format(indent, name.to_string(), node.nodeid.to_string(), bname.to_string(), dtype.to_string(), update, val))
else:
print("{0}{1:30} {2:25} {3:25}".format(indent, name.to_string(), bname.to_string(), node.nodeid.to_string()))
if depth:
_lsprint_long(node, depth - 1, indent + " ")
def create_standard_address_space_Part17(server):
node = ua.AddNodesItem()
node.RequestedNewNodeId = NumericNodeId(23455, 0)
node.BrowseName = QualifiedName('AliasNameType', 0)
node.NodeClass = NodeClass.ObjectType
node.ParentNodeId = NumericNodeId(58, 0)
node.ReferenceTypeId = NumericNodeId(45, 0)
attrs = ua.ObjectTypeAttributes()
attrs.DisplayName = LocalizedText("AliasNameType")
attrs.IsAbstract = False
node.NodeAttributes = attrs
server.add_nodes([node])
refs = []
ref = ua.AddReferencesItem()
ref.IsForward = False
ref.ReferenceTypeId = NumericNodeId(45, 0)
ref.SourceNodeId = NumericNodeId(23455, 0)
ref.TargetNodeClass = NodeClass.DataType
ref.TargetNodeId = NumericNodeId(58, 0)
refs.append(ref)
server.add_references(refs)
attrs.EventNotifier = 0
node.NodeAttributes = attrs
server.add_nodes([node])
refs = []
ref = ua.AddReferencesItem()
ref.IsForward = False
ref.ReferenceTypeId = NumericNodeId(38, 0)
ref.SourceNodeId = NumericNodeId(23499, 0)
ref.TargetNodeClass = NodeClass.DataType
ref.TargetNodeId = NumericNodeId(23468, 0)
refs.append(ref)
ref = ua.AddReferencesItem()
ref.IsForward = True
ref.ReferenceTypeId = NumericNodeId(39, 0)
ref.SourceNodeId = NumericNodeId(23499, 0)
ref.TargetNodeClass = NodeClass.DataType
ref.TargetNodeId = NumericNodeId(23502, 0)
refs.append(ref)
ref = ua.AddReferencesItem()
ref.IsForward = True
ref.ReferenceTypeId = NumericNodeId(40, 0)
ref.SourceNodeId = NumericNodeId(23499, 0)
ref.TargetNodeClass = NodeClass.DataType
ref.TargetNodeId = NumericNodeId(76, 0)
refs.append(ref)
server.add_references(refs)
node = ua.AddNodesItem()
node.RequestedNewNodeId = NumericNodeId(23505, 0)
node.BrowseName = QualifiedName('Default XML', 0)
node.NodeClass = NodeClass.Object
node.ParentNodeId = NumericNodeId(23468, 0)
async def browse_nodes(node: Node):
"""
Build a nested node tree dict by recursion (filtered by OPC UA objects and variables).
"""
node_class = await node.read_node_class()
children = []
for child in await node.get_children():
if await child.read_node_class() in [ua.NodeClass.Object, ua.NodeClass.Variable]:
children.append(
await browse_nodes(child)
)
if node_class != ua.NodeClass.Variable:
var_type = None
else:
try:
var_type = (await node.read_data_type_as_variant_type()).value
except ua.UaError:
_logger.warning('Node Variable Type could not be determined for %r', node)
var_type = None
return {
'id': node.nodeid.to_string(),
'name': (await node.read_display_name()).Text,
'cls': node_class.value,
'children': children,
async def _add_node_common(self, nodetype, node):
browsename = await node.read_browse_name()
nodeid = node.nodeid
parent = await node.get_parent()
displayname = (await node.read_display_name()).Text
desc = await node.read_description()
if desc:
desc = desc.Text
node_el = Et.SubElement(self.etree.getroot(), nodetype)
node_el.attrib["NodeId"] = self._node_to_string(nodeid)
node_el.attrib["BrowseName"] = self._bname_to_string(browsename)
if parent is not None:
node_class = await node.read_node_class()
if node_class in (ua.NodeClass.Object, ua.NodeClass.Variable, ua.NodeClass.Method):
node_el.attrib["ParentNodeId"] = self._node_to_string(parent)
self._add_sub_el(node_el, 'DisplayName', displayname)
if desc not in (None, ""):
self._add_sub_el(node_el, 'Description', desc)
# FIXME: add WriteMask and UserWriteMask
return node_el