zjsh_weight_system_ui/opcua_client.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
'''
# @Time    : 2025/11/17 18:51
# @Author  : reenrr
# @File    : opcua_client_test1111.py
'''
from PySide6.QtCore import QTimer, Slot, Signal, QObject, QThread, Qt
from opcua import Client, ua
from datetime import datetime

# -----------
# 参数配置
# -----------
OPC_SERVER_URL = "opc.tcp://localhost:4841/zjsh_feed/server/"

RECONNECT_INTERVAL = 2000  # 重连间隔（毫秒）
DATA_READ_INTERVAL = 2000  # 数据读取间隔（毫秒，与服务端更新频率一致）

class OpcuaClient(QObject):
    # 客户端的信号接口
    weight_updated = Signal(dict)               # 重量更新信号（传递 {"hopper_up_weight": 值, "hopper_down_weight": 值}）
    connection_status_changed = Signal(bool)    # 连接状态更新信号（True=连接，False=断开）
    upper_weight_error = Signal(bool)           # 上料斗重量异常信号（True=异常，False=正常）
    down_weight_error = Signal(bool)            # 下料斗重量异常信号（True=异常，False=正常）

    def __init__(self, parent=None):
        super().__init__(parent)

        self.is_running = False  # 系统运行状态标记
        self.latest_weight_data = {"hopper_up_weight": 0, "hopper_down_weight": 0}  # 缓存最新重量数据

        # ---------------
        # OPC UA 客户端核心配置
        # ---------------
        self.opc_client = None  # OPC UA 客户端实例（延迟初始化）
        self.opc_server_url = OPC_SERVER_URL
        self.upper_weight_node = None  # 上料斗重量节点对象
        self.lower_weight_node = None  # 下料斗重量节点对象

        self.is_opc_connected = False  # OPC UA 连接状态标记
        self.has_connected_once = False  # 至少连接成功一次标记
        self.reconnect_count = 0  # 重连次数计数器

        # 重连定时器（连接断开时触发）
        self.reconnect_timer = QTimer(self)
        self.reconnect_timer.setInterval(RECONNECT_INTERVAL)
        self.reconnect_timer.timeout.connect(self.reconnect_to_server)

        # 数据读取定时器（连接成功后触发，定时读取重量数据）
        self.data_read_timer = QTimer(self)
        self.data_read_timer.setInterval(DATA_READ_INTERVAL)
        self.data_read_timer.timeout.connect(self._read_weight_data)

        # 初始化 OPC UA 客户端
        self._init_opc_client()

    # ---------------------
    # OPC UA 初始化相关函数
    # ---------------------
    def _init_opc_client(self):
        """初始化 OPC UA 客户端实例"""
        try:
            self.opc_client = Client(self.opc_server_url)
            print("OPC UA 客户端初始化成功")
        except Exception as e:
            print(f"OPC UA 客户端初始化失败：{e}")
            self.opc_client = None

    def start_connect(self):
        """启动连接（供外部调用，主动发起连接请求）"""
        self._connect_to_server()

    # ---------------------
    # 连接管理相关函数
    # ---------------------
    def reconnect_to_server(self):
        """重连执行函数（与原 TCP 客户端一致）"""
        if not self.is_opc_connected:
            self.reconnect_count += 1
            # 日志优化：每10次重连提示一次，避免日志刷屏
            if self.reconnect_count % 10 == 0:
                print(f"已连续重连{self.reconnect_count}次，仍未连接服务端，请检查服务端状态...")
            else:
                print(f"第{self.reconnect_count}次重连请求：{self.opc_server_url}")
            self._connect_to_server()

    def _connect_to_server(self):
        """主动发起 OPC UA 连接（核心连接逻辑）"""
        if self.is_opc_connected or not self.opc_client:
            return

        try:
            # 1. 连接到 OPC UA 服务器
            self.opc_client.connect()
            print(f"OPC UA 连接成功：{self.opc_server_url}")

            # 2. 获取重量节点对象（仅连接成功时获取一次）
            objects = self.opc_client.get_objects_node()  # 根节点
            self.upper_weight_node = objects.get_child(["2:upper", "2:upper_weight"])

            objects = self.opc_client.get_objects_node()  # 根节点
            self.lower_weight_node = objects.get_child(["2:lower", "2:lower_weight"])

            # 3. 验证节点是否有效
            self._verify_nodes()

            # 4. 更新连接状态
            self.is_opc_connected = True
            self.has_connected_once = True
            self.reconnect_count = 0
            self.is_running = True

            # 5. 发送信号（连接成功+错误状态重置）
            self.connection_status_changed.emit(True)
            self.upper_weight_error.emit(False)
            self.down_weight_error.emit(False)

            # 6. 启动数据读取定时器，停止重连定时器
            self.data_read_timer.start()
            self.reconnect_timer.stop()

            # 7. 立即读取一次初始数据
            self._read_weight_data()

        except Exception as e:
            print(f"OPC UA 连接失败：{e}")
            self._handle_connect_failed()

    def _reconnect_to_server(self):
        """重连执行函数"""
        if self.is_opc_connected:
            return

        self.reconnect_count += 1
        # 日志优化：每10次重连提示一次，避免日志刷屏
        if self.reconnect_count % 10 == 0:
            print(f"已连续重连{self.reconnect_count}次，仍未连接服务端，请检查服务端状态...")
        else:
            print(f"第{self.reconnect_count}次重连请求：{self.opc_server_url}")

        self._connect_to_server()

    def _disconnect_from_server(self):
        """主动断开 OPC UA 连接"""
        if self.opc_client and self.is_opc_connected:
            try:
                self.opc_client.disconnect()
                print(f"OPC UA 连接断开：{self.opc_server_url}")
            except Exception as e:
                print(f"OPC UA 断开连接失败：{e}")

        # 更新状态
        self.is_opc_connected = False
        self.is_running = False
        self.upper_weight_node = None
        self.lower_weight_node = None

        # 停止数据读取定时器
        self.data_read_timer.stop()

    def _handle_connect_failed(self):
        """处理连接失败逻辑"""
        self.is_opc_connected = False
        self.is_running = False

        # 发送连接断开信号
        self.connection_status_changed.emit(False)
        # 发送重量清零信号
        self.weight_updated.emit({"hopper_up_weight": "", "hopper_down_weight": ""})
        # 标记重量异常
        self.upper_weight_error.emit(True)
        self.down_weight_error.emit(True)

        # 启动重连定时器（首次连接失败或重连失败时）
        if not self.reconnect_timer.isActive():
            print(f"将在{RECONNECT_INTERVAL / 1000}秒后启动重连...")
            self.reconnect_timer.start()

    def _verify_nodes(self):
        """验证重量节点是否有效（避免节点ID错误）"""
        try:
            # 读取节点的基本信息，验证节点存在
            upper_node_id = self.upper_weight_node.nodeid
            lower_node_id = self.lower_weight_node.nodeid
            print(f"节点验证成功：上料斗节点={upper_node_id}, 下料斗节点={lower_node_id}")
        except Exception as e:
            raise RuntimeError(f"节点验证失败（节点ID可能错误）：{e}")

    # ---------------------
    # 数据读取与处理相关函数
    # ---------------------
    @Slot()
    def _read_weight_data(self):
        """定时读取重量数据（核心业务逻辑）"""
        if not self.is_opc_connected or not self.upper_weight_node or not self.lower_weight_node:
            return

        try:
            # 读取 OPC UA 节点的重量值
            upper_weight = self.upper_weight_node.get_value()
            lower_weight = self.lower_weight_node.get_value()

            # 数据有效性校验（过滤异常值）
            upper_weight = round(float(upper_weight), 2) if self._is_valid_weight(upper_weight) else "error"
            lower_weight = round(float(lower_weight), 2) if self._is_valid_weight(lower_weight) else "error"

            # 构建重量数据字典（与原 TCP 客户端格式一致）
            weight_data = {
                "hopper_up_weight": upper_weight,
                "hopper_down_weight": lower_weight
            }
            self.latest_weight_data = weight_data

            # 错误状态判断（完全复用原 TCP 客户端逻辑）
            if upper_weight == "error" and lower_weight == "error":
                self.weight_updated.emit({"hopper_up_weight": "", "hopper_down_weight": ""})
                self.upper_weight_error.emit(True)
                self.down_weight_error.emit(True)
            elif upper_weight == "error":
                self.weight_updated.emit({"hopper_up_weight": "", "hopper_down_weight": lower_weight})
                self.upper_weight_error.emit(True)
                self.down_weight_error.emit(False)
            elif lower_weight == "error":
                self.weight_updated.emit({"hopper_up_weight": upper_weight, "hopper_down_weight": ""})
                self.upper_weight_error.emit(False)
                self.down_weight_error.emit(True)
            else:
                self.weight_updated.emit(weight_data)
                self.upper_weight_error.emit(False)
                self.down_weight_error.emit(False)

            print(f"OPC UA 数据读取成功：上料斗={upper_weight} kg, 下料斗={lower_weight} kg")

        except Exception as e:
            print(f"OPC UA 数据读取失败：{e}")
            # 读取失败时标记为异常
            self.weight_updated.emit({"hopper_up_weight": "", "hopper_down_weight": ""})
            self.upper_weight_error.emit(True)
            self.down_weight_error.emit(True)
            # 读取失败可能是连接异常，触发重连
            self._disconnect_from_server()
            if not self.reconnect_timer.isActive():
                self.reconnect_timer.start()

    def _is_valid_weight(self, weight_value):
        """验证重量值是否有效"""
        try:
            weight = float(weight_value)
            # 结合业务场景：重量值应大于0（模拟传感器有效范围）
            return weight >= 0
        except (ValueError, TypeError):
            return False

    # ---------------------
    # 外部接口函数（与原 TCP 客户端一致）
    # ---------------------
    def stop(self):
        """停止 OPC UA 客户端（断开连接、停止定时器）"""
        self.reconnect_timer.stop()
        self.data_read_timer.stop()
        self._disconnect_from_server()
        self.is_running = False
        print("OPC UA 客户端已停止")

    def get_current_time(self):
        """获取格式化的当前时间（保持原接口）"""
        return datetime.now().strftime('%Y-%m-%d %H:%M:%S')