ros_opcua_impl_python_opcua: server.py Source File

Go to the documentation of this file.
00001 import os.path
00002 try:
00003     from IPython import embed
00004 except ImportError:
00005     import code
00006 
00007     def embed():
00008         vars = globals()
00009         vars.update(locals())
00010         shell = code.InteractiveConsole(vars)
00011         shell.interact()
00012 
00013 from opcua import ua, uamethod, Server
00014 
00015 
00016 @uamethod
00017 def say_hello_xml(parent, happy):
00018     print("Calling say_hello_xml")
00019     if happy:
00020         result = "I'm happy"
00021     else:
00022         result = "I'm not happy"
00023     print(result)
00024     return result
00025 
00026 
00027 @uamethod
00028 def say_hello(parent, happy):
00029     if happy:
00030         result = "I'm happy"
00031     else:
00032         result = "I'm not happy"
00033     print(result)
00034     return result
00035 
00036 
00037 @uamethod
00038 def say_hello_array(parent, happy):
00039     if happy:
00040         result = "I'm happy"
00041     else:
00042         result = "I'm not happy"
00043     print(result)
00044     return [result, "Actually I am"]
00045 
00046 
00047 class HelloServer:
00048     def __init__(self, endpoint, name, model_filepath):
00049         self.server = Server()
00050 
00051         #  This need to be imported at the start or else it will overwrite the data
00052         self.server.import_xml(model_filepath)
00053 
00054         self.server.set_endpoint(endpoint)
00055         self.server.set_server_name(name)
00056 
00057         objects = self.server.get_objects_node()
00058 
00059         freeopcua_namespace = self.server.get_namespace_index("urn:freeopcua:python:server")
00060         hellower = objects.get_child("0:Hellower")
00061         hellower_say_hello = hellower.get_child("0:SayHello")
00062 
00063         self.server.link_method(hellower_say_hello, say_hello_xml)
00064 
00065         hellower.add_method(
00066             freeopcua_namespace, "SayHello2", say_hello, [ua.VariantType.Boolean], [ua.VariantType.String])
00067 
00068         hellower.add_method(
00069             freeopcua_namespace, "SayHelloArray", say_hello_array, [ua.VariantType.Boolean], [ua.VariantType.String])
00070 
00071     def __enter__(self):
00072         self.server.start()
00073         return self.server
00074 
00075     def __exit__(self, exc_type, exc_val, exc_tb):
00076         self.server.stop()
00077 
00078 
00079 if __name__ == '__main__':
00080     script_dir = os.path.dirname(__file__)
00081     with HelloServer(
00082             "opc.tcp://0.0.0.0:40840/freeopcua/server/",
00083             "FreeOpcUa Example Server",
00084             os.path.join(script_dir, "test_saying.xml")) as server:
00085         embed()