How to Efficiently Monitor OPC UA Server Variable Nodes with Python

1
import asyncio
2
import random
3

4
from asyncua import Server
5

6
ENDPOINT = 'opc.tcp://localhost:4840'
7
NAMESPACE = 'http://examples.freeopcua.github.io'
8

9

10
async def main() -> None:
11
    # Start a server.
12
    server = Server()
13
    await server.init()
14
    server.set_endpoint(ENDPOINT)
15
    idx = await server.register_namespace(NAMESPACE)
16
    await server.start()
17
    print(f'Server started: {server}')
18

19
    # Create a node.
20
    myobj = await server.get_objects_node().add_object(idx, 'MyObject')
21
    myvar = await myobj.add_variable(idx, 'MyVariable', 1)
22
    await myvar.set_writable()
23

24
    # Write a new value every second.
25
    while True:
26
        await myvar.write_value(random.randint(1, 100))
27
        await asyncio.sleep(1)
28

29

30
if __name__ == '__main__':
31
    asyncio.run(main())

1
import asyncio
2

3
from asyncua import Client, Node
4
from asyncua.common.subscription import DataChangeNotif, SubHandler
5

6
ENDPOINT = 'opc.tcp://localhost:4840'
7
NAMESPACE = 'http://examples.freeopcua.github.io'
8

9

10
class MyHandler(SubHandler):
11
    def __init__(self):
12
        self._queue = asyncio.Queue()
13

14
    def datachange_notification(self, node: Node, value, data: DataChangeNotif) -> None:
15
        self._queue.put_nowait([node, value, data])
16
        print(f'Data change notification was received and queued.')
17

18
    async def process(self) -> None:
19
        try:
20
            while True:
21
                # Get data in a queue.
22
                [node, value, data] = self._queue.get_nowait()
23
                path = await node.get_path(as_string=True)
24

25
                # *** Write your processing code ***
26

27
                print(f'New value {value} of "{path}" was processed.')
28

29
        except asyncio.QueueEmpty:
30
            pass
31

32

33
async def main() -> None:
34
    async with Client(url=ENDPOINT) as client:
35
        # Get a variable node.
36
        idx = await client.get_namespace_index(NAMESPACE)
37
        node = await client.get_objects_node().get_child([f'{idx}:MyObject', f'{idx}:MyVariable'])
38

39
        # Subscribe data change.
40
        handler = MyHandler()
41
        subscription = await client.create_subscription(period=0, handler=handler)
42
        await subscription.subscribe_data_change(node)
43

44
        # Process data change every 100ms
45
        while True:
46
            await handler.process()
47
            await asyncio.sleep(0.1)
48

49

50
if __name__ == '__main__':
51
    asyncio.run(main())