Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_data_type_to_variant_type(self):
test_data = {
ua.ObjectIds.Boolean: ua.VariantType.Boolean,
ua.ObjectIds.Byte: ua.VariantType.Byte,
ua.ObjectIds.String: ua.VariantType.String,
ua.ObjectIds.Int32: ua.VariantType.Int32,
ua.ObjectIds.UInt32: ua.VariantType.UInt32,
ua.ObjectIds.NodeId: ua.VariantType.NodeId,
ua.ObjectIds.LocalizedText: ua.VariantType.LocalizedText,
ua.ObjectIds.Structure: ua.VariantType.ExtensionObject,
ua.ObjectIds.EnumValueType: ua.VariantType.ExtensionObject,
ua.ObjectIds.Enumeration: ua.VariantType.Int32, # enumeration
ua.ObjectIds.AttributeWriteMask: ua.VariantType.UInt32,
ua.ObjectIds.AxisScaleEnumeration: ua.VariantType.Int32 # enumeration
}
for dt, vdt in test_data.items():
self.assertEqual(ua_utils.data_type_to_variant_type(self.opc.get_node(ua.NodeId(dt))), vdt)
async def test_several_different_events_2(opc):
objects = opc.server.nodes.objects
o = await objects.add_object(3, 'MyObject')
etype1 = await opc.server.create_custom_event_type(
2, 'MyEvent1', ua.ObjectIds.BaseEventType,
[('PropertyNum', ua.VariantType.Float), ('PropertyString', ua.VariantType.String)]
)
evgen1 = await opc.server.get_event_generator(etype1, o)
etype2 = await opc.server.create_custom_event_type(
2, 'MyEvent2', ua.ObjectIds.BaseEventType,
[('PropertyNum2', ua.VariantType.Float), ('PropertyString', ua.VariantType.String)]
)
evgen2 = await opc.server.get_event_generator(etype2, o)
etype3 = await opc.server.create_custom_event_type(
2, 'MyEvent3', ua.ObjectIds.BaseEventType,
[('PropertyNum3', ua.VariantType.Float), ('PropertyString', ua.VariantType.String)]
)
evgen3 = await opc.server.get_event_generator(etype3, o)
myhandler = MySubHandler2()
sub = await opc.opc.create_subscription(100, myhandler)
handle = await sub.subscribe_events(o, [etype1, etype3])
def test_read_server_state(self):
statenode = self.opc.get_node(ua.NodeId(ua.ObjectIds.Server_ServerStatus_State))
state = statenode.get_value()
self.assertEqual(state, 0)
async def create_custom_event_type(self, idx, name, basetype=ua.ObjectIds.BaseEventType, properties=None) -> Coroutine:
if properties is None:
properties = []
return await self._create_custom_type(idx, name, basetype, properties, [], [])
server = Server()
server.set_endpoint("opc.tcp://0.0.0.0:4840/freeopcua/server/")
# setup our own namespace, not really necessary but should as spec
uri = "http://examples.freeopcua.github.io"
idx = server.register_namespace(uri)
# get Objects node, this is where we should put our custom stuff
objects = server.nodes.objects
# populating our address space
myobj = objects.add_object(idx, "MyObject")
# Creating a custom event: Approach 1
# The custom event object automatically will have members from its parent (BaseEventType)
etype = server.create_custom_event_type(2, 'MyFirstEvent', ua.ObjectIds.BaseEventType,
[('MyNumericProperty', ua.VariantType.Float),
('MyStringProperty', ua.VariantType.String)])
# create second event
etype2 = server.create_custom_event_type(2, 'MySecondEvent', ua.ObjectIds.BaseEventType,
[('MyOtherProperty', ua.VariantType.Float)])
# get an event generator for the myobj node which generates custom events
myevgen = server.get_event_generator(etype, myobj)
myevgen.event.Severity = 500
myevgen.event.MyStringProperty = ua.Variant("hello world")
myevgen.event.MyNumericProperty = ua.Variant(-456)
# get another event generator for the myobj node which generates different custom events
myevgen2 = server.get_event_generator(etype2, myobj)
myevgen2.event.Severity = 123
myevgen2.event.MyOtherProperty = ua.Variant(1.337)
async def create_custom_variable_type(self,
idx,
name,
basetype=ua.ObjectIds.BaseVariableType,
properties=None,
variables=None,
methods=None) -> Coroutine:
if properties is None:
properties = []
if variables is None:
variables = []
if methods is None:
methods = []
return await self._create_custom_type(idx, name, basetype, properties, variables, methods)
async def get_type_definition(self):
"""
returns type definition of the node.
"""
references = await self.get_references(refs=ua.ObjectIds.HasTypeDefinition,
direction=ua.BrowseDirection.Forward)
if len(references) == 0:
return None
return references[0].NodeId
def get_encoding_refs(self):
return self.get_referenced_nodes(ua.ObjectIds.HasEncoding, ua.BrowseDirection.Forward)
ref.SourceNodeId = NumericNodeId(11204, 0)
ref.TargetNodeClass = NodeClass.DataType
ref.TargetNodeId = NumericNodeId(11203, 0)
refs.append(ref)
server.add_references(refs)
node = ua.AddNodesItem()
node.RequestedNewNodeId = NumericNodeId(11205, 0)
node.BrowseName = QualifiedName('PercentDataBad', 0)
node.NodeClass = NodeClass.Variable
node.ParentNodeId = NumericNodeId(11203, 0)
node.ReferenceTypeId = NumericNodeId(46, 0)
node.TypeDefinition = NumericNodeId(68, 0)
attrs = ua.VariableAttributes()
attrs.DisplayName = LocalizedText("PercentDataBad")
attrs.DataType = ua.NodeId(ua.ObjectIds.Byte)
attrs.ValueRank = -1
node.NodeAttributes = attrs
server.add_nodes([node])
refs = []
ref = ua.AddReferencesItem()
ref.IsForward = True
ref.ReferenceTypeId = NumericNodeId(40, 0)
ref.SourceNodeId = NumericNodeId(11205, 0)
ref.TargetNodeClass = NodeClass.DataType
ref.TargetNodeId = NumericNodeId(68, 0)
refs.append(ref)
ref = ua.AddReferencesItem()
ref.IsForward = False
ref.ReferenceTypeId = NumericNodeId(46, 0)
ref.SourceNodeId = NumericNodeId(11205, 0)
ref.TargetNodeClass = NodeClass.DataType
def __init__(self, sourcenode=None, message=None, severity=1):
super(AuditNodeManagementEvent, self).__init__(sourcenode, message, severity)
self.EventType = ua.NodeId(ua.ObjectIds.AuditNodeManagementEventType)