参考链接：Question about Python OPCUA session timeout and client.
import asyncio, json from asyncua import Client, ua, Node from asyncua.common.events import Event from datetime import datetime #################################################################################### # Globals: #################################################################################### # OPC UA Client server_url = "opc.tcp://192.168.10.1:4840" nodes_to_subscribe = [ #node-id "ns=4;i=1000", "ns=0;i=2267", "ns=0;i=2259", ] events_to_subscribe = [ #(eventtype-node-id, event-node-id) ] #################################################################################### # OpcUaClient: #################################################################################### class SubscriptionHandler: """ The SubscriptionHandler is used to handle the data that is received for the subscription. """ def datachange_notification(self, node: Node, val, data): """ Callback for asyncua Subscription. This method will be called when the Client received a data change message from the Server. """ print(node, val, data) def event_notification(self, event: Event): """ called for every event notification from server """ print(event) def status_change_notification(self, status): """ called for every status change notification from server """ print(status) async def opcua_client(): """ -handles connect/disconnect/reconnect/subscribe/unsubscribe -connection-monitoring with cyclic read of the service-level """ client = Client(url=server_url) handler = SubscriptionHandler() subscription = None case = 0 subscription_handle_list = [] while 1: if case == 1: #connect print("connecting...") try: await client.connect() await client.load_type_definitions() print("connected!") case = 2 except: print("connection error!") case = 1 await asyncio.sleep(5) elif case == 2: #subscribe all nodes and events print("subscribing nodes and events...") try: subscription = await client.create_subscription(50, handler) subscription_handle_list = [] if nodes_to_subscribe: for node in nodes_to_subscribe: handle = await subscription.subscribe_data_change(client.get_node(node)) subscription_handle_list.append(handle) if events_to_subscribe: for event in events_to_subscribe: handle = await subscription.subscribe_events(event[0], event[1]) subscription_handle_list.append(handle) print("subscribed!") case = 3 except: print("subscription error") case = 4 await asyncio.sleep(0) elif case == 3: #running => read cyclic the service level if it fails disconnect and unsubscribe => wait 5s => connect try: service_level = await client.get_node("ns=0;i=2267").get_value() if service_level >= 200: case = 3 else: case = 4 await asyncio.sleep(5) except: case = 4 elif case == 4: #disconnect clean = unsubscribe, delete subscription then disconnect print("unsubscribing...") try: if subscription_handle_list: for handle in subscription_handle_list: await subscription.unsubscribe(handle) await subscription.delete() print("unsubscribed!") except: print("unsubscribing error!") subscription = None subscription_handle_list = [] await asyncio.sleep(0) print("disconnecting...") try: await client.disconnect() except: print("disconnection error!") case = 0 else: #wait case = 1 await asyncio.sleep(5) #################################################################################### # Run: #################################################################################### if __name__ == "__main__": asyncio.ensure_future(opcua_client()) asyncio.get_event_loop().run_forever()
