Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
async def main():
server = Server()
await server.init()
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 = await server.register_namespace(uri)
# populating our address space
myobj = await server.nodes.objects.add_object(idx, "MyObject")
# Creating a custom event: Approach 1
# The custom event object automatically will have members from its parent (BaseEventType)
etype = await server.create_custom_event_type(
idx, 'MyFirstEvent', ua.ObjectIds.BaseEventType,
[('MyNumericProperty', ua.VariantType.Float),
('MyStringProperty', ua.VariantType.String)]
)
myevgen = await server.get_event_generator(etype, myobj)
# Creating a custom event: Approach 2
custom_etype = await server.nodes.base_event_type.add_object_type(2, 'MySecondEvent')
await custom_etype.add_property(2, 'MyIntProperty', ua.Variant(0, ua.VariantType.Int32))
await custom_etype.add_property(2, 'MyBoolProperty', ua.Variant(True, ua.VariantType.Boolean))
mysecondevgen = await server.get_event_generator(custom_etype, myobj)
async with server:
count = 0
while True:
await asyncio.sleep(1)
myevgen.event.Message = ua.LocalizedText("MyFirstEvent %d" % count)
myevgen.event.Severity = count
myevgen.event.MyNumericProperty = count
server = Server()
await server.init()
server.set_endpoint('opc.tcp://0.0.0.0:4840/UA/SampleServer')
server.set_server_name('Custom structure demo server')
# idx name will be used later for creating the xml used in data type dictionary
url = 'http://examples.freeopcua.github.io'
idx = await server.register_namespace(url)
dict_builder = DataTypeDictionaryBuilder(server, idx, url, 'MyDictionary')
await dict_builder.init()
# add one basic structure
basic_struct_name = 'BasicStructure'
basic_struct = await dict_builder.create_data_type(basic_struct_name)
basic_struct.add_field('ID', ua.VariantType.Int32)
basic_struct.add_field('Gender', ua.VariantType.Boolean)
basic_struct.add_field('Comments', ua.VariantType.String)
# add an advance structure which uses our basic structure
nested_struct_name = 'NestedStructure'
nested_struct = await dict_builder.create_data_type(nested_struct_name)
nested_struct.add_field('Name', ua.VariantType.String)
nested_struct.add_field('Surname', ua.VariantType.String)
# add a list of simple structure as field
nested_struct.add_field('StuffArray', basic_struct, is_array=True)
# this operation will write the OPC dict string to our new data type dictionary
# namely the 'MyDictionary'
await dict_builder.set_dict_byte_string()
dev.add_property(idx, "device_id", "0340").set_modelling_rule(True)
ctrl = dev.add_object(idx, "controller")
ctrl.set_modelling_rule(True)
ctrl.add_property(idx, "state", "Idle").set_modelling_rule(True)
# populating our address space
# First a folder to organise our nodes
myfolder = server.nodes.objects.add_folder(idx, "myEmptyFolder")
# instanciate one instance of our device
mydevice = server.nodes.objects.add_object(idx, "Device0001", dev)
mydevice_var = mydevice.get_child([f"{idx}:controller", "{idx}:state"]) # get proxy to our device state variable
# create directly some objects and variables
myobj = server.nodes.objects.add_object(idx, "MyObject")
myvar = myobj.add_variable(idx, "MyVariable", 6.7)
mysin = myobj.add_variable(idx, "MySin", 0, ua.VariantType.Float)
myvar.set_writable() # Set MyVariable to be writable by clients
mystringvar = myobj.add_variable(idx, "MyStringVariable", "Really nice string")
mystringvar.set_writable() # Set MyVariable to be writable by clients
mydtvar = myobj.add_variable(idx, "MyDateTimeVar", datetime.utcnow())
mydtvar.set_writable() # Set MyVariable to be writable by clients
myarrayvar = myobj.add_variable(idx, "myarrayvar", [6.7, 7.9])
myarrayvar = myobj.add_variable(idx, "myStronglytTypedVariable", ua.Variant([], ua.VariantType.UInt32))
myprop = myobj.add_property(idx, "myproperty", "I am a property")
mymethod = myobj.add_method(idx, "mymethod", func, [ua.VariantType.Int64], [ua.VariantType.Boolean])
multiply_node = myobj.add_method(idx, "multiply", multiply, [ua.VariantType.Int64, ua.VariantType.Int64], [ua.VariantType.Int64])
# import some nodes from xml
server.import_xml("custom_nodes.xml")
# creating a default event object
# The event object automatically will have members for all events properties
idx = await server.register_namespace(uri)
# get Objects node, this is where we should put our custom stuff
objects = server.nodes.objects
# populating our address space
await objects.add_folder(idx, "myEmptyFolder")
myobj = await objects.add_object(idx, "MyObject")
myvar = await myobj.add_variable(idx, "MyVariable", 6.7)
await myvar.set_writable() # Set MyVariable to be writable by clients
myarrayvar = await myobj.add_variable(idx, "myarrayvar", [6.7, 7.9])
await myobj.add_variable(
idx, "myStronglytTypedVariable", ua.Variant([], ua.VariantType.UInt32)
)
await myobj.add_property(idx, "myproperty", "I am a property")
await myobj.add_method(idx, "mymethod", func, [ua.VariantType.Int64], [ua.VariantType.Boolean])
inargx = ua.Argument()
inargx.Name = "x"
inargx.DataType = ua.NodeId(ua.ObjectIds.Int64)
inargx.ValueRank = -1
inargx.ArrayDimensions = []
inargx.Description = ua.LocalizedText("First number x")
inargy = ua.Argument()
inargy.Name = "y"
inargy.DataType = ua.NodeId(ua.ObjectIds.Int64)
inargy.ValueRank = -1
inargy.ArrayDimensions = []
inargy.Description = ua.LocalizedText("Second number y")
outarg = ua.Argument()
outarg.Name = "Result"
outarg.DataType = ua.NodeId(ua.ObjectIds.Int64)
def __init__(self, sourcenode=None, message=None, severity=1):
super(ProgressEvent, self).__init__(sourcenode, message, severity)
self.EventType = ua.NodeId(ua.ObjectIds.ProgressEventType)
self.add_property('Context', None, ua.VariantType.Variant)
self.add_property('Progress', None, ua.VariantType.UInt16)
device = await client.nodes.objects.get_child(["6:MyObjects", "6:MyDevice"])
method = await device.get_child("6:MyMethod")
result = await device.call_method(method, ua.Variant("sin"), ua.Variant(180, ua.VariantType.Double))
print("Mehtod result is: ", result)
handler = SubHandler()
sub = await client.create_subscription(500, handler)
handle = await sub.subscribe_data_change(var)
handle2 = await sub.subscribe_events(evtypes=2788)
cond = await client.nodes.root.get_child(["0:Types", "0:EventTypes", "0:BaseEventType", "0:ConditionType"])
for _ in range(5):
# refresh server condition to force generation of events
await cond.call_method("0:ConditionRefresh", ua.Variant(sub.subscription_id, ua.VariantType.UInt32))
await asyncio.sleep(1)
await sub.unsubscribe(handle)
await sub.unsubscribe(handle2)
await sub.delete()
def from_binary(uatype, data):
"""
unpack data given an uatype as a string or a python class having a ua_types member
"""
if isinstance(uatype, str) and uatype.startswith("ListOf"):
utype = uatype[6:]
if hasattr(ua.VariantType, utype):
vtype = getattr(ua.VariantType, utype)
return unpack_uatype_array(vtype, data)
size = Primitives.Int32.unpack(data)
return [from_binary(utype, data) for _ in range(size)]
elif isinstance(uatype, str) and hasattr(ua.VariantType, uatype):
vtype = getattr(ua.VariantType, uatype)
return unpack_uatype(vtype, data)
elif isinstance(uatype, str) and hasattr(Primitives, uatype):
return getattr(Primitives, uatype).unpack(data)
else:
return struct_from_binary(uatype, data)
elif args.datatype == "Guid":
return _arg_to_variant(val, array, bytes, ua.VariantType.Guid)
elif args.datatype == "ByteString":
return _arg_to_variant(val, array, bytes, ua.VariantType.ByteString)
elif args.datatype == "xml":
return _arg_to_variant(val, array, str, ua.VariantType.XmlElement)
elif args.datatype == "nodeid":
return _arg_to_variant(val, array, ua.NodeId.from_string, ua.VariantType.NodeId)
elif args.datatype == "expandednodeid":
return _arg_to_variant(val, array, ua.ExpandedNodeId.from_string, ua.VariantType.ExpandedNodeId)
elif args.datatype == "statuscode":
return _arg_to_variant(val, array, int, ua.VariantType.StatusCode)
elif args.datatype in ("qualifiedname", "browsename"):
return _arg_to_variant(val, array, ua.QualifiedName.from_string, ua.VariantType.QualifiedName)
elif args.datatype == "LocalizedText":
return _arg_to_variant(val, array, ua.LocalizedText, ua.VariantType.LocalizedText)
def _add_nodeattributes(self, item, nodedata, add_timestamps):
self._add_node_attr(item, nodedata, "AccessLevel", ua.VariantType.Byte)
self._add_node_attr(item, nodedata, "ArrayDimensions", ua.VariantType.UInt32)
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.Int32)
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.Boolean)
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.UInt32)