266 lines
11 KiB
Python
266 lines
11 KiB
Python
#!/usr/bin/env python3
|
||
from PySide6.QtCore import QObject, Signal, QThread
|
||
from opcua import Client, ua
|
||
import time
|
||
from datetime import datetime
|
||
import configparser
|
||
|
||
class OpcuaUiSignal(QObject):
|
||
value_changed = Signal(str, str, object)
|
||
opc_disconnected = Signal(str) # OPC服务断开信号，参数：断开原因
|
||
opc_reconnected = Signal() # OPC重连成功信号
|
||
opc_log = Signal(str) # OPC运行日志信号，参数：日志信息
|
||
|
||
# Opcua回调处理器
|
||
class SubscriptionHandler:
|
||
def __init__(self, opc_signal:OpcuaUiSignal):
|
||
self.node_id_to_name = {}
|
||
self.opc_signal = opc_signal
|
||
|
||
def datachange_notification(self, node, val, data):
|
||
try:
|
||
node_id = node.nodeid.to_string()
|
||
var_name = self.node_id_to_name.get(node_id)
|
||
self.opc_signal.value_changed.emit(node_id, var_name, val)
|
||
except Exception as e:
|
||
err_msg = f"opcua解析值变化事件失败: {e}"
|
||
self.opc_signal.opc_log.emit(err_msg)
|
||
|
||
class OpcuaUiClient(QThread):
|
||
def __init__(self, parent=None):
|
||
super().__init__(parent)
|
||
self.server_url = ""
|
||
self.client = None
|
||
|
||
self.connected = False
|
||
self.subscription = None
|
||
self.monitored_items = []
|
||
self.is_running = True # 线程运行标志位
|
||
self.node_id_mapping = {} # node_id 和 可读变量名的映射表
|
||
self.is_reconnect_tip_sent = False # 重连失败提示是否已发送
|
||
|
||
self.opc_signal = OpcuaUiSignal()
|
||
self.handler = SubscriptionHandler(self.opc_signal)
|
||
|
||
self.target_var_paths = []
|
||
|
||
# 参数
|
||
self.heartbeat_interval = None # 心跳检测间隔
|
||
self.reconnect_interval = None # 首次/掉线重连间隔
|
||
self.sub_interval = None # 订阅间隔 (单位:ms)
|
||
|
||
|
||
def stop_run(self):
|
||
"""停止线程+断开连接"""
|
||
self.is_running = False
|
||
self.disconnect()
|
||
self.wait()
|
||
print("opcua客户端线程已退出")
|
||
|
||
def connect(self):
|
||
"""连接到OPC服务器"""
|
||
try:
|
||
self.client.connect()
|
||
self.connected = True
|
||
msg = f"成功连接到OPCUA服务器: {self.server_url}"
|
||
print(msg)
|
||
self.opc_signal.opc_log.emit(msg)
|
||
self.is_reconnect_tip_sent = False
|
||
return True
|
||
except Exception as e:
|
||
self.connected = False
|
||
err_msg = f"连接OPCUA服务器失败: {e}"
|
||
print(err_msg)
|
||
if not self.is_reconnect_tip_sent:
|
||
self.opc_signal.opc_log.emit(err_msg)
|
||
# 标记为已发送，后续不重复在UI上显示
|
||
self.is_reconnect_tip_sent = True
|
||
return False
|
||
|
||
def disconnect(self):
|
||
"""断开连接"""
|
||
self.connected = False
|
||
try:
|
||
if self.monitored_items:
|
||
for item in self.monitored_items:
|
||
try:
|
||
self.subscription.unsubscribe(item)
|
||
except Exception:
|
||
pass
|
||
self.monitored_items.clear()
|
||
if self.subscription:
|
||
try:
|
||
self.subscription.delete()
|
||
except Exception:
|
||
pass
|
||
self.subscription = None
|
||
if self.client:
|
||
try:
|
||
self.client.disconnect()
|
||
except Exception:
|
||
pass
|
||
self.node_id_mapping.clear()
|
||
if hasattr(self, 'handler') and self.handler:
|
||
self.handler.node_id_to_name = {}
|
||
except Exception as e:
|
||
print(f"opcua断开连接异常: {e}")
|
||
|
||
def build_node_id_mapping(self):
|
||
"""根据object_name+var_name路径获取nodeid,建立映射表"""
|
||
if not self.connected:
|
||
return False
|
||
try:
|
||
# self.opc_signal.opc_log.emit("开始构建nodeid映射表...")
|
||
objects_node = self.client.get_objects_node()
|
||
self.handler.node_id_to_name = self.node_id_mapping
|
||
for var_name, path_list in self.target_var_paths:
|
||
target_node = objects_node.get_child(path_list)
|
||
node_id = target_node.nodeid.to_string()
|
||
self.node_id_mapping[node_id] = var_name
|
||
# self.opc_signal.opc_log.emit("nodeid映射表构建成功")
|
||
return True
|
||
except Exception as e:
|
||
err_msg = f"构建{var_name}映射表失败: {e}"
|
||
print(err_msg)
|
||
self.opc_signal.opc_log.emit(err_msg)
|
||
return False
|
||
|
||
def create_multi_subscription(self, interval=None):
|
||
"""订阅多个变量(基于映射表的nodeid)"""
|
||
if not self.connected:
|
||
return
|
||
if not self.node_id_mapping and not self.build_node_id_mapping():
|
||
return
|
||
try:
|
||
interval = int(interval) if interval else self.sub_interval
|
||
self.subscription = self.client.create_subscription(interval, self.handler)
|
||
self.opc_signal.opc_log.emit(f"opcua订阅创建成功(间隔：{interval}ms)")
|
||
for node_id, var_name in self.node_id_mapping.items():
|
||
var_node = self.client.get_node(node_id)
|
||
monitored_item = self.subscription.subscribe_data_change(var_node)
|
||
self.monitored_items.append(monitored_item)
|
||
# self.opc_signal.opc_log.emit(f"已订阅变量: {var_name} (nodeid: {node_id})")
|
||
print(f"已订阅变量: {var_name} (nodeid: {node_id})")
|
||
except Exception as e:
|
||
err_msg = f"创建批量订阅失败: {e}"
|
||
print(err_msg)
|
||
self.opc_signal.opc_log.emit(err_msg)
|
||
|
||
def read_opc_config(self, cfg_path = "config/opc_config.ini"):
|
||
"""读取OPC配置文件, 初始化所有参数和节点列表"""
|
||
try:
|
||
cfg = configparser.ConfigParser()
|
||
cfg.read(cfg_path, encoding="utf-8")
|
||
# 1. 读取服务器基础配置
|
||
self.server_url = cfg.get("OPC_SERVER_CONFIG", "server_url")
|
||
self.heartbeat_interval = cfg.getint("OPC_SERVER_CONFIG", "heartbeat_interval")
|
||
self.reconnect_interval = cfg.getint("OPC_SERVER_CONFIG", "reconnect_interval")
|
||
self.sub_interval = cfg.getint("OPC_SERVER_CONFIG", "sub_interval")
|
||
|
||
# 2. 读取OPC节点配置
|
||
node_section = cfg["OPC_NODE_LIST"]
|
||
for readable_name, node_path_str in node_section.items():
|
||
node_path_list = node_path_str.split(",")
|
||
self.target_var_paths.append( (readable_name, node_path_list) )
|
||
# print("target_var_paths", self.target_var_paths)
|
||
except Exception as e:
|
||
print(f"读取配置文件失败: {e}，使用默认配置启动！")
|
||
self.server_url = "opc.tcp://localhost:4840/zjsh_feed/server/"
|
||
self.heartbeat_interval = 4
|
||
self.reconnect_interval = 2
|
||
self.sub_interval = 500
|
||
self.target_var_paths = [
|
||
("upper_weight", ["2:upper", "2:upper_weight"]),
|
||
("lower_weight", ["2:lower", "2:lower_weight"])
|
||
]
|
||
# 参数合法性检验
|
||
self.heartbeat_interval = self.heartbeat_interval if isinstance(self.heartbeat_interval, int) and self.heartbeat_interval >=1 else 4
|
||
self.reconnect_interval = self.reconnect_interval if isinstance(self.reconnect_interval, int) and self.reconnect_interval >=1 else 2
|
||
self.sub_interval = self.sub_interval if isinstance(self.sub_interval, int) and self.sub_interval >=100 else 500
|
||
|
||
def write_value_by_name(self, var_readable_name, value):
|
||
"""
|
||
根据变量可读名称写入值(主要用于修改方量， 方量的类型为 Double类型)
|
||
:param var_readable_name: 变量可读名称（如"upper_weight")
|
||
:param value: 要写入的值
|
||
"""
|
||
if not self.connected:
|
||
self.opc_signal.opc_log.emit(f"{var_readable_name}写入失败: OPC服务未连接")
|
||
return
|
||
target_node_id = None
|
||
for node_id, name in self.node_id_mapping.items():
|
||
if name == var_readable_name:
|
||
target_node_id = node_id
|
||
break
|
||
if not target_node_id:
|
||
# self.opc_signal.opc_log.emit(f"写入失败：未找到变量名 {var_readable_name} 对应的nodeid")
|
||
return
|
||
try:
|
||
target_node = self.client.get_node(target_node_id)
|
||
# variant = ua.Variant(float(value), ua.VariantType.Double)
|
||
target_node.set_value(value)
|
||
# self.opc_signal.opc_log.emit(f"写入成功：{var_readable_name} = {value}")
|
||
except Exception as e:
|
||
err_msg = f"opcua写入值失败: {e}"
|
||
print(err_msg)
|
||
self.opc_signal.opc_log.emit(err_msg)
|
||
|
||
# ===== 心跳检测函数 =====
|
||
def _heartbeat_check(self):
|
||
"""心跳检测: 判断opc服务是否存活"""
|
||
try:
|
||
self.client.get_node("i=2258").get_value()
|
||
return True
|
||
except Exception as e:
|
||
err_msg = f"心跳检测失败, OPCUA服务已断开 {e}"
|
||
print(err_msg)
|
||
self.opc_signal.opc_log.emit(err_msg)
|
||
return False
|
||
|
||
# ===== 掉线重连函数 =====
|
||
def _auto_reconnect(self):
|
||
"""掉线后自动重连+重建映射+恢复订阅"""
|
||
self.opc_signal.opc_disconnected.emit("OPC服务掉线, 开始自动重连...")
|
||
try:
|
||
self.disconnect()
|
||
except Exception as e:
|
||
print(f"_auto_reconnect: 断开旧连接时出现异常: {e}")
|
||
while self.is_running:
|
||
# self.opc_signal.opc_log.emit(f"重试连接OPC服务器: {self.server_url}")
|
||
if self.connect():
|
||
self.build_node_id_mapping()
|
||
self.create_multi_subscription()
|
||
self.opc_signal.opc_reconnected.emit()
|
||
self.opc_signal.opc_log.emit("OPCUA服务器重连成功, 所有订阅已恢复正常")
|
||
print("OPCUA服务器重连成功, 所有订阅已恢复正常")
|
||
break
|
||
time.sleep(self.reconnect_interval)
|
||
|
||
def _init_connect_with_retry(self):
|
||
"""连接opc服务器"""
|
||
# self.opc_signal.opc_log.emit("OPC客户端初始化, 开始连接服务器...")
|
||
print("OPC客户端初始化, 开始连接服务器...")
|
||
while self.is_running:
|
||
if self.connect():
|
||
self.build_node_id_mapping()
|
||
self.create_multi_subscription()
|
||
break
|
||
# self.opc_signal.opc_log.emit(f"连接OPCUA服务器失败, {self.reconnect_interval}秒后重试...")
|
||
time.sleep(self.reconnect_interval)
|
||
|
||
def run(self) -> None:
|
||
"""opcua客户端线程主函数"""
|
||
self.read_opc_config() # 读取配置文件
|
||
self.client = Client(self.server_url) # 初始化opc客户端
|
||
|
||
# 连接opc服务器
|
||
self._init_connect_with_retry()
|
||
|
||
while self.is_running:
|
||
if self.connected:
|
||
if not self._heartbeat_check():
|
||
self.connected = False
|
||
self._auto_reconnect()
|
||
else:
|
||
self._auto_reconnect()
|
||
time.sleep(self.heartbeat_interval) |