suyu_two/app/src/infrastructure/plc_communication_service.cpp

#include "plc_communication_service.h"

#include "plc_register_repository.h"

#include <QModbusDataUnit>
#include <QModbusDevice>
#include <QModbusReply>
#include <QModbusRtuSerialMaster>
#include <QSerialPort>
#include <QVariant>

#include <algorithm>
#include <utility>

namespace {

constexpr int kMaximumReadCount = 120;

std::string toUtf8(const QString &value)
{
    const QByteArray bytes = value.toUtf8();
    return std::string(bytes.constData(), static_cast<std::size_t>(bytes.size()));
}

QModbusDataUnit::RegisterType registerType(RegisterArea area)
{
    return area == RegisterArea::M
        ? QModbusDataUnit::Coils : QModbusDataUnit::HoldingRegisters;
}

QString modbusErrorText(QModbusDevice::Error error)
{
    switch (error)
    {
    case QModbusDevice::ReadError:
        return QStringLiteral("读取 PLC 数据失败");
    case QModbusDevice::WriteError:
        return QStringLiteral("写入 PLC 数据失败");
    case QModbusDevice::ConnectionError:
        return QStringLiteral("PLC 串口连接失败");
    case QModbusDevice::ConfigurationError:
        return QStringLiteral("PLC 串口参数配置错误");
    case QModbusDevice::TimeoutError:
        return QStringLiteral("PLC 通信超时，请检查站号、串口参数和 RS-485 接线");
    case QModbusDevice::ProtocolError:
        return QStringLiteral("PLC 返回了无效或异常的 Modbus 响应");
    case QModbusDevice::ReplyAbortedError:
        return QStringLiteral("PLC 通信请求已取消");
    case QModbusDevice::UnknownError:
        return QStringLiteral("PLC 通信发生未知错误");
    case QModbusDevice::NoError:
    default:
        return QStringLiteral("PLC 通信失败");
    }
}

} // namespace

PlcCommunicationService::PlcCommunicationService(
    PlcRegisterRepository &repository,
    QObject *parent)
    : QObject(parent),
      repository_(repository),
      master_(std::make_unique<QModbusRtuSerialMaster>())
{
    repository_.setWriteHandlers(
        [this](const RegisterAddress &address, bool value)
        {
            return sendBitWrite(address, value);
        },
        [this](const RegisterAddress &address, std::int16_t value)
        {
            return sendWordWrite(address, value);
        });
    connect(&poll_timer_, &QTimer::timeout, this, &PlcCommunicationService::pollNextBlock);
    connect(master_.get(), &QModbusClient::stateChanged,
            this,
            [this](QModbusDevice::State device_state)
            {
                if (device_state == QModbusDevice::ConnectedState)
                {
                    setState(PlcConnectionState::Connected);
                    poll_timer_.start(configuration_.pollIntervalMs);
                    pollNextBlock();
                }
                else if (device_state == QModbusDevice::ConnectingState)
                {
                    setState(PlcConnectionState::Connecting);
                }
                else if (device_state == QModbusDevice::UnconnectedState
                         && state_ != PlcConnectionState::Faulted)
                {
                    setState(PlcConnectionState::Disconnected);
                }
            });
    connect(master_.get(), &QModbusClient::errorOccurred,
            this,
            [this](QModbusDevice::Error error)
            {
                if (error != QModbusDevice::NoError)
                {
                    setError(modbusErrorText(error));
                }
            });
}

PlcCommunicationService::~PlcCommunicationService() = default;

PlcCommunicationResult PlcCommunicationService::connectDevice(
    const PlcSerialConfiguration &configuration)
{
    if (QString::fromStdString(configuration.portName).trimmed().isEmpty()
        || configuration.serverAddress < 1 || configuration.serverAddress > 247)
    {
        return {false, "必须填写串口端口，并将 PLC 站号设置为 1~247"};
    }
    if (master_->state() != QModbusDevice::UnconnectedState)
    {
        return {false, "PLC 连接已经启动，请先断开当前连接"};
    }
    configuration_ = configuration;
    last_error_.clear();
    master_->setConnectionParameter(
        QModbusDevice::SerialPortNameParameter,
        QString::fromStdString(configuration.portName));
    master_->setConnectionParameter(
        QModbusDevice::SerialBaudRateParameter, configuration.baudRate);
    master_->setConnectionParameter(
        QModbusDevice::SerialDataBitsParameter,
        static_cast<QSerialPort::DataBits>(configuration.dataBits));
    master_->setConnectionParameter(
        QModbusDevice::SerialParityParameter,
        static_cast<QSerialPort::Parity>(configuration.parity));
    master_->setConnectionParameter(
        QModbusDevice::SerialStopBitsParameter,
        static_cast<QSerialPort::StopBits>(configuration.stopBits));
    master_->setTimeout(configuration.responseTimeoutMs);
    master_->setNumberOfRetries(configuration.retries);
    repository_.invalidate();
    initial_read_completed_ = false;
    emit initialReadCompletedChanged(false);
    if (initial_read_changed_callback_)
    {
        initial_read_changed_callback_(false);
    }
    rebuildPollBlocks();
    setState(PlcConnectionState::Connecting);
    if (!master_->connectDevice())
    {
        setError(modbusErrorText(master_->error()));
        return {false, last_error_};
    }
    return {true, {}};
}

void PlcCommunicationService::disconnectDevice()
{
    poll_timer_.stop();
    master_->disconnectDevice();
    repository_.invalidate();
    initial_read_completed_ = false;
    emit initialReadCompletedChanged(false);
    if (initial_read_changed_callback_)
    {
        initial_read_changed_callback_(false);
    }
    setState(PlcConnectionState::Disconnected);
}

void PlcCommunicationService::setPollAddresses(
    const std::vector<RegisterAddress> &addresses)
{
    poll_addresses_ = addresses;
    rebuildPollBlocks();
}

PlcConnectionState PlcCommunicationService::state() const
{
    return state_;
}

bool PlcCommunicationService::initialReadCompleted() const
{
    return initial_read_completed_;
}

const std::string &PlcCommunicationService::lastError() const
{
    return last_error_;
}

const PlcSerialConfiguration &PlcCommunicationService::configuration() const
{
    return configuration_;
}

void PlcCommunicationService::setCallbacks(
    std::function<void()> state_changed,
    std::function<void(bool)> initial_read_changed,
    std::function<void()> cache_updated,
    std::function<void(const std::string &)> error_reported)
{
    state_changed_callback_ = std::move(state_changed);
    initial_read_changed_callback_ = std::move(initial_read_changed);
    cache_updated_callback_ = std::move(cache_updated);
    error_reported_callback_ = std::move(error_reported);
}

void PlcCommunicationService::rebuildPollBlocks()
{
    std::vector<RegisterAddress> addresses = poll_addresses_;
    if (addresses.empty())
    {
        addresses = {
            RegisterAddress{RegisterArea::M, 0},
            RegisterAddress{RegisterArea::D, 0}};
    }
    std::sort(
        addresses.begin(), addresses.end(),
        [](const RegisterAddress &left, const RegisterAddress &right)
        {
            if (left.area() != right.area())
            {
                return left.area() == RegisterArea::M;
            }
            return left.index() < right.index();
        });
    addresses.erase(std::unique(addresses.begin(), addresses.end()), addresses.end());
    poll_blocks_.clear();
    for (const RegisterAddress &address : addresses)
    {
        if (!address.isValid())
        {
            continue;
        }
        if (poll_blocks_.empty()
            || poll_blocks_.back().area != address.area()
            || address.index() > poll_blocks_.back().startAddress
                + poll_blocks_.back().count
            || poll_blocks_.back().count >= kMaximumReadCount)
        {
            poll_blocks_.push_back({address.area(), address.index(), 1});
        }
        else
        {
            poll_blocks_.back().count = address.index()
                - poll_blocks_.back().startAddress + 1;
        }
    }
    next_poll_block_ = 0;
    initial_blocks_read_.assign(poll_blocks_.size(), false);
    initial_read_completed_ = false;
}

void PlcCommunicationService::pollNextBlock()
{
    if (state_ != PlcConnectionState::Connected
        || pending_reply_ != nullptr || poll_blocks_.empty())
    {
        return;
    }
    const std::size_t block_index = next_poll_block_;
    const PollBlock block = poll_blocks_.at(block_index);
    next_poll_block_ = (next_poll_block_ + 1U) % poll_blocks_.size();
    QModbusDataUnit request(
        registerType(block.area), block.startAddress, static_cast<quint16>(block.count));
    QModbusReply *reply = master_->sendReadRequest(request, configuration_.serverAddress);
    if (reply == nullptr)
    {
        setError(modbusErrorText(master_->error()));
        return;
    }
    pending_reply_ = reply;
    connect(reply, &QModbusReply::finished,
            this,
            [this, reply, block, block_index]
            {
                handleReadFinished(reply, block);
                if (state_ == PlcConnectionState::Connected
                    && reply->error() == QModbusDevice::NoError
                    && block_index < initial_blocks_read_.size())
                {
                    initial_blocks_read_[block_index] = true;
                    const bool completed = std::all_of(
                        initial_blocks_read_.cbegin(),
                        initial_blocks_read_.cend(),
                        [](bool read) { return read; });
                    if (completed && !initial_read_completed_)
                    {
                        initial_read_completed_ = true;
                        emit initialReadCompletedChanged(true);
                        if (initial_read_changed_callback_)
                        {
                            initial_read_changed_callback_(true);
                        }
                    }
                }
                if (pending_reply_ == reply)
                {
                    pending_reply_ = nullptr;
                }
                reply->deleteLater();
            });
}

void PlcCommunicationService::handleReadFinished(QModbusReply *reply, PollBlock block)
{
    if (state_ != PlcConnectionState::Connected)
    {
        return;
    }
    if (reply->error() != QModbusDevice::NoError)
    {
        setError(modbusErrorText(reply->error()));
        return;
    }
    const QModbusDataUnit result = reply->result();
    for (uint index = 0; index < result.valueCount(); ++index)
    {
        const int address = block.startAddress + static_cast<int>(index);
        if (block.area == RegisterArea::M)
        {
            repository_.updateBit(address, result.value(index) != 0U);
        }
        else
        {
            repository_.updateWord(address, static_cast<std::int16_t>(result.value(index)));
        }
    }
    last_error_.clear();
    emit cacheUpdated();
    if (cache_updated_callback_)
    {
        cache_updated_callback_();
    }
}

RegisterWriteResult PlcCommunicationService::sendBitWrite(
    const RegisterAddress &address, bool value)
{
    if (state_ != PlcConnectionState::Connected)
    {
        return {false, RegisterError::Unavailable};
    }
    QModbusDataUnit unit(QModbusDataUnit::Coils, address.index(), 1);
    unit.setValue(0, value ? 1U : 0U);
    QModbusReply *reply = master_->sendWriteRequest(unit, configuration_.serverAddress);
    if (reply == nullptr)
    {
        setError(modbusErrorText(master_->error()));
        return {false, RegisterError::WriteRejected};
    }
    connect(reply, &QModbusReply::finished,
            this,
            [this, reply]
            {
                if (reply->error() != QModbusDevice::NoError)
                {
                    setError(modbusErrorText(reply->error()));
                }
                reply->deleteLater();
            });
    return {true, RegisterError::None};
}

RegisterWriteResult PlcCommunicationService::sendWordWrite(
    const RegisterAddress &address, std::int16_t value)
{
    if (state_ != PlcConnectionState::Connected)
    {
        return {false, RegisterError::Unavailable};
    }
    QModbusDataUnit unit(QModbusDataUnit::HoldingRegisters, address.index(), 1);
    unit.setValue(0, static_cast<quint16>(value));
    QModbusReply *reply = master_->sendWriteRequest(unit, configuration_.serverAddress);
    if (reply == nullptr)
    {
        setError(modbusErrorText(master_->error()));
        return {false, RegisterError::WriteRejected};
    }
    connect(reply, &QModbusReply::finished,
            this,
            [this, reply]
            {
                if (reply->error() != QModbusDevice::NoError)
                {
                    setError(modbusErrorText(reply->error()));
                }
                reply->deleteLater();
            });
    return {true, RegisterError::None};
}

void PlcCommunicationService::setState(PlcConnectionState state)
{
    if (state_ == state)
    {
        return;
    }
    state_ = state;
    emit stateChanged();
    if (state_changed_callback_)
    {
        state_changed_callback_();
    }
}

void PlcCommunicationService::setError(const QString &message)
{
    last_error_ = toUtf8(message);
    poll_timer_.stop();
    setState(PlcConnectionState::Faulted);
    emit communicationError(message);
    if (error_reported_callback_)
    {
        error_reported_callback_(last_error_);
    }
}