client-alarms-and-conditions

Complexity

Total Complexity

5

Size/Duplication

Total Lines	51
Duplicated Lines	0 %

Importance

Changes

0

4 Methods

Rating	Name	Duplication	Size	Complexity
A	SubHandler.datachange_notification()	0	2	1
A	OpcUaClient.__init__()	0	2	1
A	OpcUaClient.init()	0	18	1
A	SubHandler.event_notification()	0	2	1

# coding: utf-8

import asyncio

from asyncua import Client, ua


class SubHandler(object):

    def datachange_notification(self, node, val, data):
        print("Python: New data change event", node, val, data)

    def event_notification(self, event):
        print("Python new event:", event)


class OpcUaClient(object):

    def __init__(self, endpoint):
        self.client = Client(endpoint)

    async def init(self, ):
        await self.client.connect()
        objects = self.client.get_objects_node()
        idx = await self.client.get_namespace_index("http://examples.freeopcua.github.io")

        path = ['%s:ConditionObject' % idx]
        con_obj = await objects.get_child(path)
        condition = self.client.get_node(ua.NodeId(2830))

        handler = SubHandler()
        sub = await self.client.create_subscription(500, handler)
        con_handle = await sub.subscribe_events(con_obj, condition)

        path = ['%s:AlarmObject' % idx]
        alarm_obj = await objects.get_child(path)
        alarm = self.client.get_node(ua.NodeId(10637))

        alarm_handle = await sub.subscribe_events(alarm_obj, alarm)


async def start():
    client = OpcUaClient("opc.tcp://localhost:4840")
    await client.init()


if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    loop.create_task(start())
    loop.run_forever()