97 lines
3.2 KiB
Python
97 lines
3.2 KiB
Python
|
import threading
|
||
|
|
import time
|
||
|
|
from opcua import Client, ua
|
||
|
|
from opcua.ua.uaerrors import UaError
|
||
|
|
import json
|
||
|
|
|
||
|
|
|
||
|
|
class OPCClient:
|
||
|
|
def __init__(self, url='opc.tcp://127.0.0.1:4840'):
|
||
|
|
self.url = url
|
||
|
|
self.client = None
|
||
|
|
self.is_connected = False
|
||
|
|
self.reconnect_interval = 5 # 重连间隔(秒)
|
||
|
|
self.is_running = False
|
||
|
|
self.namespace_idx = None
|
||
|
|
self.data_node = None
|
||
|
|
|
||
|
|
def connect(self):
|
||
|
|
"""连接到OPC UA服务端"""
|
||
|
|
try:
|
||
|
|
if self.client:
|
||
|
|
self.client.disconnect()
|
||
|
|
|
||
|
|
self.client = Client(self.url)
|
||
|
|
self.client.set_security_string("None")
|
||
|
|
self.client.connect()
|
||
|
|
|
||
|
|
# 获取命名空间索引
|
||
|
|
self.namespace_idx = self.client.get_namespace_index("urn:python-opcua")
|
||
|
|
|
||
|
|
# 获取数据节点
|
||
|
|
self.data_node = self.client.get_node(f"ns={self.namespace_idx};i=2001") # 默认数据节点
|
||
|
|
|
||
|
|
self.is_connected = True
|
||
|
|
print(f"已连接到 OPC UA 服务端 {self.url}")
|
||
|
|
return True
|
||
|
|
except UaError as e:
|
||
|
|
print(f"OPC UA 连接错误: {e}")
|
||
|
|
self.is_connected = False
|
||
|
|
return False
|
||
|
|
except Exception as e:
|
||
|
|
print(f"连接到 OPC UA 服务端失败: {e}")
|
||
|
|
self.is_connected = False
|
||
|
|
return False
|
||
|
|
|
||
|
|
def start(self):
|
||
|
|
"""启动客户端并维持连接"""
|
||
|
|
self.is_running = True
|
||
|
|
reconnect_thread = threading.Thread(target=self._reconnect_worker, daemon=True)
|
||
|
|
reconnect_thread.start()
|
||
|
|
|
||
|
|
def _reconnect_worker(self):
|
||
|
|
"""重连工作线程"""
|
||
|
|
while self.is_running:
|
||
|
|
if not self.is_connected:
|
||
|
|
print(f"尝试重新连接到 {self.url}...")
|
||
|
|
self.connect()
|
||
|
|
time.sleep(self.reconnect_interval)
|
||
|
|
|
||
|
|
def send_data(self, data):
|
||
|
|
"""通过OPC UA发送数据到服务端"""
|
||
|
|
if not self.is_connected or not self.client:
|
||
|
|
print("OPC UA 客户端未连接到服务端")
|
||
|
|
return False
|
||
|
|
|
||
|
|
try:
|
||
|
|
# 将数据转换为JSON字符串
|
||
|
|
json_data = json.dumps(data, ensure_ascii=False)
|
||
|
|
|
||
|
|
# 写入数据到OPC UA节点
|
||
|
|
if self.data_node:
|
||
|
|
variant = ua.Variant(json_data, ua.VariantType.String)
|
||
|
|
self.data_node.set_value(variant)
|
||
|
|
print(f"已通过 OPC UA 发送数据: {json_data}")
|
||
|
|
return True
|
||
|
|
else:
|
||
|
|
print("数据节点未初始化")
|
||
|
|
return False
|
||
|
|
except UaError as e:
|
||
|
|
print(f"OPC UA 发送数据时出错: {e}")
|
||
|
|
self.is_connected = False
|
||
|
|
return False
|
||
|
|
except Exception as e:
|
||
|
|
print(f"发送数据时出错: {e}")
|
||
|
|
self.is_connected = False
|
||
|
|
return False
|
||
|
|
|
||
|
|
def stop(self):
|
||
|
|
"""停止客户端"""
|
||
|
|
self.is_running = False
|
||
|
|
self.is_connected = False
|
||
|
|
if self.client:
|
||
|
|
try:
|
||
|
|
self.client.disconnect()
|
||
|
|
except Exception as e:
|
||
|
|
print(f"断开 OPC UA 客户端时出错: {e}")
|
||
|
|
print("OPC UA 客户端已停止")
|