Commit 986cf74d authored by Denis Štogl's avatar Denis Štogl

Updated example. Shown two ways for defining events

parent 6cfba61d
import sys
sys.path.insert(0, "..")
import logging
try:
from IPython import embed
except ImportError:
import code
def embed():
vars = globals()
vars.update(locals())
shell = code.InteractiveConsole(vars)
shell.interact()
from opcua import ua, uamethod, Server, EventGenerator
class SubHandler(object):
"""
Subscription Handler. To receive events from server for a subscription
"""
def datachange_notification(self, node, val, data):
print("Python: New data change event", node, val)
def event_notification(self, event):
print("Python: New event", event)
# method to be exposed through server
def func(parent, variant):
ret = False
if variant.Value % 2 == 0:
ret = True
return [ua.Variant(ret, ua.VariantType.Boolean)]
# method to be exposed through server
# uses a decorator to automatically convert to and from variants
@uamethod
def multiply(parent, x, y):
print("multiply method call with parameters: ", x, y)
return x * y
from opcua import ua, Server
if __name__ == "__main__":
# optional: setup logging
logging.basicConfig(level=logging.WARN)
#logger = logging.getLogger("opcua.address_space")
# logger.setLevel(logging.DEBUG)
#logger = logging.getLogger("opcua.internal_server")
# logger.setLevel(logging.DEBUG)
#logger = logging.getLogger("opcua.binary_server_asyncio")
# logger.setLevel(logging.DEBUG)
#logger = logging.getLogger("opcua.uaprocessor")
# logger.setLevel(logging.DEBUG)
logger = logging.getLogger("opcua.subscription_service")
logger.setLevel(logging.DEBUG)
# now setup our server
# setup our server
server = Server()
#server.disable_clock()
#server.set_endpoint("opc.tcp://localhost:4840/freeopcua/server/")
server.set_endpoint("opc.tcp://0.0.0.0:4840/freeopcua/server/")
server.set_server_name("FreeOpcUa Example Server")
# setup our own namespace
# setup our own namespace, not really necessary but should as spec
uri = "http://examples.freeopcua.github.io"
idx = server.register_namespace(uri)
......@@ -77,37 +19,35 @@ if __name__ == "__main__":
objects = server.get_objects_node()
# populating our address space
myfolder = objects.add_folder(idx, "myEmptyFolder")
myobj = objects.add_object(idx, "MyObject")
myvar = myobj.add_variable(idx, "MyVariable", 6.7)
myvar.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])
# creating an custom event object
# 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, 'MyEvent', ua.ObjectIds.BaseEventType, [('MyNumericProperty', ua.VariantType.Float), ('MyStringProperty', ua.VariantType.String)])
etype = server.create_custom_event_type(2, 'MyFirstEvent', ua.ObjectIds.BaseEventType, [('MyNumericProperty', ua.VariantType.Float), ('MyStringProperty', ua.VariantType.String)])
myevgen = server.get_event_generator(etype, myobj)
myevgen.event.Severity = 500
# Creating a custom event: Approach 2
base_etype = server.get_node(ua.ObjectIds.BaseEventType)
custom_etype = base_etype.add_subtype(2, 'MySecondEvent')
custom_etype.add_property(2, 'MyIntProperty', ua.Variant(None, ua.VariantType.Int32))
custom_etype.add_property(2, 'MyBoolProperty', ua.Variant(None, ua.VariantType.Boolean))
mysecondevgen = server.get_event_generator(custom_etype, myobj)
# starting!
server.start()
print("Available loggers are: ", logging.Logger.manager.loggerDict.keys())
try:
# enable following if you want to subscribe to nodes on server side
#handler = SubHandler()
#sub = server.create_subscription(500, handler)
#handle = sub.subscribe_data_change(myvar)
# trigger event, all subscribed clients wil receive it
# time.sleep is here just because we want to see events in UaExpert
import time
time.sleep(10)
print "Triggering event..."
myevgen.trigger(message="This is MyEvent with MyNumericProperty and MyStringProperty.")
embed()
myevgen.trigger(message="This is MyFirstEvent with MyNumericProperty and MyStringProperty.")
mysecondevgen.trigger(message="This is MySecondEvent with MyIntProperty and MyBoolProperty.")
time.sleep(20)
finally:
#close connection, remove subcsriptions, etc
server.stop()
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment