A Minimal OPC-UA Server#
Let’s start exploring the asyncua package by building a minimal runnable server.
Most of the hard work will be hidden behind the scene and we only need to implement the
application specific code.
The complete code will look like the code below. In the next sections we will look at the different parts of the code, so don’t be overwhelmed by the code snippet!
1import asyncio
2import logging
3
4from asyncua import Server, ua
5from asyncua.common.methods import uamethod
6
7
8@uamethod
9def func(parent, value):
10 return value * 2
11
12
13async def main():
14 _logger = logging.getLogger(__name__)
15 # setup our server
16 server = Server()
17 await server.init()
18 server.set_endpoint("opc.tcp://0.0.0.0:4840/freeopcua/server/")
19
20 # set up our own namespace, not really necessary but should as spec
21 uri = "http://examples.freeopcua.github.io"
22 idx = await server.register_namespace(uri)
23
24 # populating our address space
25 # server.nodes, contains links to very common nodes like objects and root
26 myobj = await server.nodes.objects.add_object(idx, "MyObject")
27 myvar = await myobj.add_variable(idx, "MyVariable", 6.7)
28 # Set MyVariable to be writable by clients
29 await myvar.set_writable()
30 await server.nodes.objects.add_method(
31 ua.NodeId("ServerMethod", idx),
32 ua.QualifiedName("ServerMethod", idx),
33 func,
34 [ua.VariantType.Int64],
35 [ua.VariantType.Int64],
36 )
37 _logger.info("Starting server!")
38 async with server:
39 while True:
40 await asyncio.sleep(1)
41 new_val = await myvar.get_value() + 0.1
42 _logger.info("Set value of %s to %.1f", myvar, new_val)
43 await myvar.write_value(new_val)
44
45
46if __name__ == "__main__":
47 logging.basicConfig(level=logging.DEBUG)
48 asyncio.run(main(), debug=True)
Before we even look at the code in detail, let’s try out what our server can do.
Start the server in a terminal with python server-minimal.py and open a new console.
In the new console you now can use the CLI tools (see Command Line Tools) provided by the package to explore
the server. The following session gives you an idea how the tools can be used.
$ uals --url=opc.tcp://127.0.0.1:4840 # List root node
Browsing node i=84 at opc.tcp://127.0.0.1:4840
DisplayName NodeId BrowseName Value
LocalizedText(Locale=None, Text='Objects') i=85 0:Objects
LocalizedText(Locale=None, Text='Types') i=86 0:Types
LocalizedText(Locale=None, Text='Views') i=87 0:Views
$ uals --url=opc.tcp://127.0.0.1:4840 --nodeid i=85 # List 0:Objects
Browsing node i=85 at opc.tcp://127.0.0.1:4840
DisplayName NodeId BrowseName Value
LocalizedText(Locale=None, Text='Server') i=2253 0:Server
LocalizedText(Locale=None, Text='Aliases') i=23470 0:Aliases
LocalizedText(Locale=None, Text='MyObject') ns=2;i=1 2:MyObject
LocalizedText(Locale=None, Text='ServerMethod') ns=2;s=ServerMethod 2:ServerMethod
$ # In the last two lines we can see our own MyObject and ServerMethod
$ # Lets read a value!
$ uaread --url=opc.tcp://127.0.0.1:4840 --nodeid "ns=2;i=2" # By NodeId
7.599999999999997
$ uaread --url=opc.tcp://127.0.0.1:4840 --path "0:Objects,2:MyObject,2:MyVariable" # By BrowsePath
12.199999999999996
Seems like our server is working and we can browse through the nodes, read values, … So let’s start working through the code!
Imports, Basic Setup & Configuration#
In the first few lines the relevant packages, classes and methods are imported.
While the logging module is optional (just remove all calls to the logging module),
asyncio is required to actually run our main function. From the asyncua
package we need the Server, the asyncua.ua
module and the uamethod() decorator.
Ignore the @uamethod ... part for the moment and jump straight into the
async def main() function:
async def main():
_logger = logging.getLogger(__name__)
# setup our server
server = Server()
await server.init()
server.set_endpoint("opc.tcp://0.0.0.0:4840/freeopcua/server/")
# set up our own namespace, not really necessary but should as spec
uri = "http://examples.freeopcua.github.io"
idx = await server.register_namespace(uri)
Todo
The init() and set_endpoint()
methods have no docstrings but are referenced in the next section.
Here the server is created and initialized (init()).
The endpoint is configured (set_endpoint()) and a custom namespace
is registered (register_namespace()). It’s recommended (required??)
that all custom objects, variables and methods live in a separate namespace. We store its index as idx.
We’ll need it later to add our custom objects to the namespace.
Creating Objects and Variables#
myobj = await server.nodes.objects.add_object(idx, "MyObject")
myvar = await myobj.add_variable(idx, "MyVariable", 6.7)
# Set MyVariable to be writable by clients
await myvar.set_writable()
In the next lines, the custom object “MyObject” is created and a variable is added to this object.
Note that by default all variables are read-only, so we need to be explicit and make it writable.
The add_object() / add_variable() calls
are actually just calling create_object(), respectively
create_variable() internally. You can find more information on
how nodes and variables are created in the API docs of these methods.
Adding Methods#
With the code we have written so far, we would already have a server which can be run and exposes some custom data. But to complete the example, we also add a method which is callable by clients:
@uamethod
def func(parent, value):
return value * 2
await server.nodes.objects.add_method(
ua.NodeId("ServerMethod", idx),
ua.QualifiedName("ServerMethod", idx),
func,
[ua.VariantType.Int64],
[ua.VariantType.Int64],
)
To do this, a function, decorated with the uamethod() decorator,
is created and, similar to the objects and variables, registered on the server. It would
also be possible to register a undecorated function on the server, but in this case the
conversion from and to UA Variant types would be up to us.
Starting the Server#
_logger.info("Starting server!")
async with server:
while True:
await asyncio.sleep(1)
new_val = await myvar.get_value() + 0.1
_logger.info("Set value of %s to %.1f", myvar, new_val)
await myvar.write_value(new_val)
if __name__ == "__main__":
logging.basicConfig(level=logging.DEBUG)
asyncio.run(main(), debug=True)
Using the server as a context manager with async with server: ... allows us to
hide starting and shutting down the server nicely. In order to keep the server alive
a endless loop must be present. In this example the loop is also used to periodically
update the variable in our custom object.
Now that we have a working server, let’s go on and write A Minimal OPC-UA Client!