EJM_Display/DataCenter/OpcUaManager.cpp

#include "OpcUaManager.h"
#include <QOpcUaProvider>
#include <QMap>
#include <QDebug>
#include <QDateTime>
#include <GlobalDefinitions/Variable.h>
OpcUaManager* OpcUaManager::m_instance = nullptr;
QMutex OpcUaManager::m_mutex;

OpcUaManager::OpcUaManager(QObject *parent)
    : QObject(parent)
    , mReconnectTimer(new QTimer(this))
{
    connect(mReconnectTimer, &QTimer::timeout, this, &OpcUaManager::tryReconnect);
    mReconnectTimer->setInterval(RECONNECT_INTERVAL);

}

OpcUaManager* OpcUaManager::instance(QObject *parent)
{
    if (!m_instance) {
        QMutexLocker locker(&m_mutex);
        if (!m_instance) {
            m_instance = new OpcUaManager(parent);
        }
    }
    return m_instance;
}

OpcUaManager::~OpcUaManager()
{
    disconnectFromServer();
    clearNodeCache();
    if (mClient) {
        delete mClient;
        mClient = nullptr;
    }
}

QOpcUaClient* OpcUaManager::getClient() const
{
    return mClient;
}

bool OpcUaManager::connectToServer(const QString &url, const QString &username, const QString &password)
{
    mServerUrl = QUrl(url);
    mUserName = username;
    mPassword = password;

    if (mClient) {
        disconnectFromServer();
        delete mClient;
        mClient = nullptr;
    }

    createClient();
    if (!mClient) {
        emit errorOccurred("无法创建OPC UA客户端");
        onClientStateChanged(QOpcUaClient::Disconnected);
        return false;
    }

    mClient->requestEndpoints(mServerUrl);
    connect(mClient, &QOpcUaClient::readNodeAttributesFinished,this, &OpcUaManager::onBatchReadFinished);
    return true;
}

void OpcUaManager::disconnectFromServer()
{
    if (mClient) {
        mClient->disconnectFromEndpoint();
    }

    clearNodeCache();
    mCurrentNodeId.clear();
    mPendingWriteValue = QVariant();
}

bool OpcUaManager::readNodeValue(const QString &nodeId)
{
    if (!mClient || mClient->state() != QOpcUaClient::Connected) {
        qWarning()<<("客户端未连接");
        return false;
    }

    QOpcUaNode* node = getCachedNode(nodeId);
    if (!node) {
        QString error = QString("无法获取节点: %1").arg(nodeId);
        qWarning()<<(error);
        return false;
    }

    mCurrentNodeId = nodeId;

    disconnect(node, &QOpcUaNode::attributeRead, this, &OpcUaManager::onAttributeRead);
    connect(node,
            static_cast<void (QOpcUaNode::*)(QOpcUa::NodeAttributes)>(&QOpcUaNode::attributeRead),
            this,
            &OpcUaManager::onAttributeRead);

    node->readAttributes(
        QOpcUa::NodeAttribute::Value |
        QOpcUa::NodeAttribute::AccessLevel
    );
    return true;
}
void OpcUaManager::onAttributeRead(QOpcUa::NodeAttributes attrs)
{
    QOpcUaNode* node = qobject_cast<QOpcUaNode*>(sender());
    if (!node) return;

    QString nodeId = node->nodeId();
    if (attrs.testFlag(QOpcUa::NodeAttribute::Value)) {
        gOPC_NodeValue[nodeId] = node->attribute(QOpcUa::NodeAttribute::Value);

    } else {
        gOPC_NodeValue[nodeId] = "";
    }
}
bool OpcUaManager::readNodesValue(const QList<QString>& nodeIds)
{
    // 前置校验：客户端连接状态 + 节点列表非空 + 避免并发读取
    if (!mClient || mClient->state() != QOpcUaClient::Connected) {
        qWarning()<<("批量读取失败：OPC客户端未连接");
        return false;
    }
    if (nodeIds.isEmpty()) {
        qWarning()<<("批量读取失败：节点列表为空");
        return false;
    }


    // 1. 构建批量读取请求（正确构造QOpcUaReadItem）
    QVector<QOpcUaReadItem> readItems;
    for (const QString& nodeId : nodeIds) {
        QOpcUaReadItem item;
        item.setNodeId(nodeId);
        readItems.append(item);
    }

    // 发送批量读取请求
    if(!mClient->readNodeAttributes(readItems)){
        qDebug()<<("批量读取请求发送失败");
        return false;
    }

    return true;
}

// 批量读取结果处理槽函数（参数需与信号匹配）
void OpcUaManager::onBatchReadFinished(const QVector<QOpcUaReadResult>& results)
{
    // 遍历结果（使用QOpcUaReadResult的方法获取数据）
    for (const auto& result : results) {
        if (result.statusCode() == QOpcUa::Good) {
            gOPC_NodeValue[result.nodeId()] = result.value();
            emit readCompleted(result.value(), result.nodeId());
        } else {
            gOPC_NodeValue[result.nodeId()] = "";
        }
    }
}

//bool OpcUaManager::writeNodeValue(const QString &nodeId, const QVariant &value, int maxRetry)
//{
//    if (!mClient || mClient->state() != QOpcUaClient::Connected) {
//        emit errorOccurred("客户端未连接");
//        return false;
//    }

//    QOpcUaNode* node = getCachedNode(nodeId);
//    if (!node) {
//        emit errorOccurred(QString("无法获取节点: %1").arg(nodeId));
//        return false;
//    }

//    WriteRequest request;
//    request.nodeId = nodeId;
//    request.value = value;
//    request.retryCount = maxRetry;
//    mWriteRequests[node] = request;

//    disconnect(node, &QOpcUaNode::attributeRead, this, &OpcUaManager::onPreWriteValueTypeRead);
//    connect(node,
//            static_cast<void (QOpcUaNode::*)(QOpcUa::NodeAttributes)>(&QOpcUaNode::attributeRead),
//            this,
//            &OpcUaManager::onPreWriteValueTypeRead);

//    node->readAttributes(QOpcUa::NodeAttribute::Value);
//    return writeNodeValue(nodeId,value,maxRetry,100);
//}

bool OpcUaManager::writeNodeValue(const QString &nodeId,
                                  const QVariant &value,
                                  int maxRetry /*= 3*/,
                                  int retryIntervalMs /*= 200*/)
{
    QString mNodeID = nodeId;
    if(!mNodeID.contains("ns=6;s=::AsGlobalPV:"))
        mNodeID = "ns=6;s=::AsGlobalPV:"+mNodeID;
    if (!mClient || mClient->state() != QOpcUaClient::Connected) {
        qWarning().noquote() << "[OPC] 客户端未连接，写入" << mNodeID << "失败";
        return false;
    }

    QOpcUaNode *node = mClient->node(mNodeID);
    if (!node) {
        qWarning().noquote() << "[OPC] 无法写入节点:" << mNodeID;
        return false;
    }

    for (int attempt = 1; attempt <= maxRetry; ++attempt) {
        /* ---------- 1. 先读当前值 ---------- */
        QVariant cur = gOPC_NodeValue[mNodeID];
        if (!cur.isValid()) {
            //qWarning().noquote() << "读到无效值:" << mNodeID<<gOPC_NodeValue[mNodeID];
        }

        /* ---------- 2. 类型转换 ---------- */
        QVariant v = value;
        QOpcUa::Types typeHint = QOpcUa::Types::Undefined;
        QString tn = cur.typeName();
        if (tn == "int")        { v.convert(QMetaType::Int);    typeHint = QOpcUa::Types::Int32;  }
        else if (tn == "float") { v.convert(QMetaType::Float);  typeHint = QOpcUa::Types::Float;  }
        else if (tn == "double"){ v.convert(QMetaType::Double); typeHint = QOpcUa::Types::Double; }
        else if (tn == "ushort"){ v.convert(QMetaType::UShort); typeHint = QOpcUa::Types::UInt16; }

        /* ---------- 3. 同步写 ---------- */
        struct WriteWait : QObject {
            bool ok = false;
            QEventLoop loop;
        } waiter;

        connect(node,
                static_cast<void(QOpcUaNode::*)(QOpcUa::NodeAttribute, QOpcUa::UaStatusCode)>(
                    &QOpcUaNode::attributeWritten),
                &waiter, [&waiter](QOpcUa::NodeAttribute, QOpcUa::UaStatusCode code) mutable {
                    waiter.ok = (code == QOpcUa::UaStatusCode::Good);
                    waiter.loop.quit();
                }, Qt::QueuedConnection);

        node->writeAttribute(QOpcUa::NodeAttribute::Value, v, typeHint);
        QTimer::singleShot(2000, &waiter.loop, &QEventLoop::quit);
        waiter.loop.exec();

        if (waiter.ok) {
            return true;
        }

        /* ---------- 4. 失败日志 & 重试 ---------- */
//        qWarning().noquote() << "[OPC] 第" << attempt << "/" << maxRetry
//                             << "次写入失败:" << mNodeID << "  等待重试 ...";
        if (attempt < maxRetry)
            QThread::msleep(retryIntervalMs);
    }

    qCritical().noquote() << "[OPC] 最终写入失败:" << mNodeID;
    return false;
}
QOpcUaClient::ClientState OpcUaManager::connectionState() const
{
    if (mClient) {
        return mClient->state();
    }
    return QOpcUaClient::Disconnected;
}

void OpcUaManager::createClient()
{
    QOpcUaProvider provider;
    QMap<QString, QVariant> clientParams;

    // 设置认证信息
    if (!mUserName.isEmpty() && !mPassword.isEmpty()) {
        clientParams["username"] = mUserName;
        clientParams["password"] = mPassword;
    }

    // 禁用安全策略（根据实际需求调整）
    clientParams["securityMode"] = "None";
    clientParams["securityPolicy"] = "None";

    // 性能参数设置
    clientParams["requestTimeout"] = 5000;
    clientParams["maxArrayLength"] = 10000;

    // 创建客户端（使用open62541后端）
    mClient = provider.createClient("open62541", clientParams);
    if (!mClient) {
        return;
    }

    mClient->setParent(this);

    // 连接客户端信号
    connect(mClient,
            static_cast<void (QOpcUaClient::*)(QVector<QOpcUaEndpointDescription>,
                                              QOpcUa::UaStatusCode,
                                              QUrl)>(&QOpcUaClient::endpointsRequestFinished),
            this,
            &OpcUaManager::onEndpointsRequestFinished);

    connect(mClient,
            static_cast<void (QOpcUaClient::*)(QOpcUaClient::ClientState)>(&QOpcUaClient::stateChanged),
            this,
            &OpcUaManager::onClientStateChanged);
}

void OpcUaManager::tryReconnect()
{
    if (mClient && mClient->state() == QOpcUaClient::Disconnected) {
        qint64 now = QDateTime::currentMSecsSinceEpoch();

        // 限制重连频率
        if (now - mLastReconnectAttempt > 1000) {
            mLastReconnectAttempt = now;
            qDebug() << "正在执行第" << mReconnectCount << "次重连...";
            connectToServer(mServerUrl.toString(), mUserName, mPassword);
            mReconnectCount++;

            // 重连次数过多时增加间隔时间
            if (mReconnectCount > 10) {
                mReconnectTimer->setInterval(RECONNECT_INTERVAL * 2);
            }
        }
    }
}

void OpcUaManager::onEndpointsRequestFinished(QVector<QOpcUaEndpointDescription> endpoints,
                                             QOpcUa::UaStatusCode statusCode,
                                             QUrl requestedUrl)
{
    Q_UNUSED(requestedUrl);
    if (statusCode != QOpcUa::UaStatusCode::Good || endpoints.isEmpty()) {
        QString error = QString("获取端点失败，状态码: %1").arg(static_cast<int>(statusCode));
        emit errorOccurred(error);
        mReconnectTimer->start(RECONNECT_INTERVAL);
        return;
    }

    mClient->connectToEndpoint(endpoints.first());
}

void OpcUaManager::onClientStateChanged(QOpcUaClient::ClientState state)
{
    emit stateChanged(state);

    if (state == QOpcUaClient::Disconnected) {
        qDebug() << "OPC UA 连接断开，" << RECONNECT_INTERVAL << "ms 后自动重连...";
        mReconnectTimer->start(RECONNECT_INTERVAL);
    } else if (state == QOpcUaClient::Connected) {
        mReconnectTimer->stop();
        mReconnectCount = 0; // 重置计数器
        mReconnectTimer->setInterval(RECONNECT_INTERVAL); // 重置间隔
        emit reconnected();
    }
}

QOpcUaNode* OpcUaManager::getCachedNode(const QString& nodeId)
{
    if (mNodeCache.contains(nodeId)) {
        return mNodeCache[nodeId];
    }

    if (!mClient) return nullptr;

    QOpcUaNode* node = mClient->node(nodeId);
    if (node) {
        node->setParent(this);
        mNodeCache[nodeId] = node;
    }
    return node;
}

void OpcUaManager::clearNodeCache()
{
    qDeleteAll(mNodeCache.values());
    mNodeCache.clear();
    mWriteRequests.clear();
}