diff --git a/.gitignore b/.gitignore index 91ab649..13a372c 100644 --- a/.gitignore +++ b/.gitignore @@ -34,7 +34,9 @@ Thumbs.db *.rc /.qmake.cache /.qmake.stash - +/.cursor/ +/.cache/ +/.vscode/ # qtcreator generated files *.pro.user* *.qbs.user* diff --git a/bin/softbus_modbus_rtu_framer_test b/bin/softbus_modbus_rtu_framer_test new file mode 100644 index 0000000..737a653 Binary files /dev/null and b/bin/softbus_modbus_rtu_framer_test differ diff --git a/bin/softbus_modbus_rtu_pdu_codec_test b/bin/softbus_modbus_rtu_pdu_codec_test new file mode 100644 index 0000000..32a4021 Binary files /dev/null and b/bin/softbus_modbus_rtu_pdu_codec_test differ diff --git a/bin/softbus_protocol_parse_filter_test b/bin/softbus_protocol_parse_filter_test new file mode 100644 index 0000000..8f4eb79 Binary files /dev/null and b/bin/softbus_protocol_parse_filter_test differ diff --git a/config/device_profiles.json b/config/device_profiles.json new file mode 100644 index 0000000..740947c --- /dev/null +++ b/config/device_profiles.json @@ -0,0 +1,36 @@ +{ + "profiles": [ + { + "deviceId": "*", + "protocol": "modbus_rtu", + "mappings": [ + { + "pointId": "temperature", + "metadataId": "MD_TEMP_001", + "registerAddress": 40001, + "byteOffset": 0, + "byteLength": 4, + "bitOffset": -1, + "endianness": "BIG_ENDIAN", + "rawType": "INT32", + "scaleFactor": 1.0, + "offsetFactor": 0.0, + "expression": "x * 0.1" + }, + { + "pointId": "pressure", + "metadataId": "MD_PRESS_001", + "registerAddress": 40003, + "byteOffset": 4, + "byteLength": 4, + "bitOffset": -1, + "endianness": "BIG_ENDIAN", + "rawType": "INT32", + "scaleFactor": 1.0, + "offsetFactor": 0.0, + "expression": "x * 0.001" + } + ] + } + ] +} diff --git a/config/metadata_dictionary.json b/config/metadata_dictionary.json new file mode 100644 index 0000000..9f7d2ac --- /dev/null +++ b/config/metadata_dictionary.json @@ -0,0 +1,28 @@ +{ + "metadatas": [ + { + "metadataId": "MD_TEMP_001", + "name": "主管道温度", + "description": "由两个寄存器组合得到温度", + "domain": "PROCESS_PARAM", + "semanticType": "FLOAT64", + "unit": "C", + "minValue": -40.0, + "maxValue": 180.0, + "deadband": 0.1, + "isWriteable": false + }, + { + "metadataId": "MD_PRESS_001", + "name": "主管道压力", + "description": "由两个寄存器组合得到压力", + "domain": "PROCESS_PARAM", + "semanticType": "FLOAT64", + "unit": "MPa", + "minValue": 0.0, + "maxValue": 25.0, + "deadband": 0.01, + "isWriteable": false + } + ] +} diff --git a/src/core/metadata/MetadataRegistry.cpp b/src/core/metadata/MetadataRegistry.cpp new file mode 100644 index 0000000..9f4ec50 --- /dev/null +++ b/src/core/metadata/MetadataRegistry.cpp @@ -0,0 +1,133 @@ +#include "core/metadata/MetadataRegistry.h" + +#include +#include +#include +#include + +namespace softbus::core::metadata +{ + +namespace +{ + +using softbus::core::models::DataDomain; +using softbus::core::models::DataType; + +DataDomain parseDomain(const QString& s) +{ + const QString x = s.trimmed().toUpper(); + if (x == QStringLiteral("PROCESS_PARAM")) { + return DataDomain::PROCESS_PARAM; + } + if (x == QStringLiteral("CONTROL_CMD")) { + return DataDomain::CONTROL_CMD; + } + if (x == QStringLiteral("MONITOR_DATA")) { + return DataDomain::MONITOR_DATA; + } + if (x == QStringLiteral("HEALTH_DIAG")) { + return DataDomain::HEALTH_DIAG; + } + if (x == QStringLiteral("CONFIG_PARAM")) { + return DataDomain::CONFIG_PARAM; + } + return DataDomain::UNKNOWN; +} + +DataType parseDataType(const QString& s) +{ + const QString x = s.trimmed().toUpper(); + if (x == QStringLiteral("BOOL")) { + return DataType::BOOL; + } + if (x == QStringLiteral("INT32")) { + return DataType::INT32; + } + if (x == QStringLiteral("UINT32")) { + return DataType::UINT32; + } + if (x == QStringLiteral("INT64")) { + return DataType::INT64; + } + if (x == QStringLiteral("FLOAT32")) { + return DataType::FLOAT32; + } + if (x == QStringLiteral("FLOAT64")) { + return DataType::FLOAT64; + } + if (x == QStringLiteral("STRING")) { + return DataType::STRING; + } + if (x == QStringLiteral("BYTE_ARRAY")) { + return DataType::BYTE_ARRAY; + } + return DataType::FLOAT64; +} + +} // namespace + +bool MetadataRegistry::loadFromFile(const QString& filePath) +{ + m_lastError.clear(); + m_defsById.clear(); + + QFile f(filePath); + if (!f.open(QIODevice::ReadOnly)) { + m_lastError = QStringLiteral("failed to open metadata file: %1").arg(filePath); + return false; + } + + QJsonParseError pe; + const auto doc = QJsonDocument::fromJson(f.readAll(), &pe); + if (pe.error != QJsonParseError::NoError || !doc.isObject()) { + m_lastError = QStringLiteral("invalid metadata json: %1").arg(pe.errorString()); + return false; + } + + const auto root = doc.object(); + const auto arr = root.value(QStringLiteral("metadatas")).toArray(); + for (const auto& item : arr) { + if (!item.isObject()) { + continue; + } + const auto one = parseOne(item.toObject()); + if (!one.has_value() || one->metadataId.isEmpty()) { + continue; + } + m_defsById.insert(one->metadataId, *one); + } + + if (m_defsById.isEmpty()) { + m_lastError = QStringLiteral("metadata file contains no valid entries"); + return false; + } + return true; +} + +const softbus::core::models::MetadataDef* MetadataRegistry::findById(const QString& metadataId) const +{ + const auto it = m_defsById.constFind(metadataId); + if (it == m_defsById.constEnd()) { + return nullptr; + } + return &it.value(); +} + +std::optional MetadataRegistry::parseOne(const QJsonObject& o) +{ + softbus::core::models::MetadataDef def; + def.metadataId = o.value(QStringLiteral("metadataId")).toString(); + def.name = o.value(QStringLiteral("name")).toString(); + def.description = o.value(QStringLiteral("description")).toString(); + def.domain = parseDomain(o.value(QStringLiteral("domain")).toString()); + def.semanticType = parseDataType(o.value(QStringLiteral("semanticType")).toString()); + def.unit = o.value(QStringLiteral("unit")).toString(); + def.minValue = o.value(QStringLiteral("minValue")).toDouble(0.0); + def.maxValue = o.value(QStringLiteral("maxValue")).toDouble(0.0); + def.deadband = o.value(QStringLiteral("deadband")).toDouble(0.0); + def.isWriteable = o.value(QStringLiteral("isWriteable")).toBool(false); + return def; +} + +} // namespace softbus::core::metadata diff --git a/src/core/metadata/MetadataRegistry.h b/src/core/metadata/MetadataRegistry.h new file mode 100644 index 0000000..9a1a4d6 --- /dev/null +++ b/src/core/metadata/MetadataRegistry.h @@ -0,0 +1,31 @@ +#pragma once + +#include + +#include +#include +#include + +#include "core/models/MetadataDef.h" + +namespace softbus::core::metadata +{ + +class MetadataRegistry +{ +public: + bool loadFromFile(const QString& filePath); + + const softbus::core::models::MetadataDef* findById(const QString& metadataId) const; + bool empty() const { return m_defsById.isEmpty(); } + const QString& lastError() const { return m_lastError; } + +private: + static std::optional parseOne(const QJsonObject& o); + +private: + QHash m_defsById; + QString m_lastError; +}; + +} // namespace softbus::core::metadata diff --git a/src/core/metadata/ProfileRegistry.cpp b/src/core/metadata/ProfileRegistry.cpp new file mode 100644 index 0000000..f448078 --- /dev/null +++ b/src/core/metadata/ProfileRegistry.cpp @@ -0,0 +1,160 @@ +#include "core/metadata/ProfileRegistry.h" + +#include +#include +#include +#include + +namespace softbus::core::metadata +{ + +namespace +{ + +QString normalizedProtocol(const QString& protocol) +{ + return protocol.trimmed().toLower(); +} + +} // namespace + +bool ProfileRegistry::loadFromFile(const QString& filePath) +{ + m_lastError.clear(); + m_profilesByKey.clear(); + + QFile f(filePath); + if (!f.open(QIODevice::ReadOnly)) { + m_lastError = QStringLiteral("failed to open profile file: %1").arg(filePath); + return false; + } + + QJsonParseError pe; + const auto doc = QJsonDocument::fromJson(f.readAll(), &pe); + if (pe.error != QJsonParseError::NoError || !doc.isObject()) { + m_lastError = QStringLiteral("invalid profile json: %1").arg(pe.errorString()); + return false; + } + + const auto root = doc.object(); + const auto arr = root.value(QStringLiteral("profiles")).toArray(); + for (const auto& item : arr) { + if (!item.isObject()) { + continue; + } + const auto obj = item.toObject(); + + softbus::core::models::DeviceDataProfile p; + p.deviceId = obj.value(QStringLiteral("deviceId")).toString(); + p.protocol = normalizedProtocol(obj.value(QStringLiteral("protocol")).toString()); + + const auto mappings = obj.value(QStringLiteral("mappings")).toArray(); + for (const auto& m : mappings) { + if (!m.isObject()) { + continue; + } + const auto one = parseMapping(m.toObject()); + if (one.has_value()) { + p.mappings.append(*one); + } + } + if (p.deviceId.isEmpty() || p.protocol.isEmpty() || p.mappings.isEmpty()) { + continue; + } + m_profilesByKey.insert(makeKey(p.deviceId, p.protocol), std::move(p)); + } + + if (m_profilesByKey.isEmpty()) { + m_lastError = QStringLiteral("profile file contains no valid entries"); + return false; + } + return true; +} + +const softbus::core::models::DeviceDataProfile* ProfileRegistry::find(const QString& deviceId, + const QString& protocol) const +{ + const auto protocolNorm = normalizedProtocol(protocol); + const auto exactIt = m_profilesByKey.constFind(makeKey(deviceId, protocolNorm)); + if (exactIt != m_profilesByKey.constEnd()) { + return &exactIt.value(); + } + const auto wildcard = m_profilesByKey.constFind(makeKey(QStringLiteral("*"), protocolNorm)); + if (wildcard != m_profilesByKey.constEnd()) { + return &wildcard.value(); + } + return nullptr; +} + +QString ProfileRegistry::makeKey(const QString& deviceId, const QString& protocol) +{ + return deviceId + QStringLiteral("::") + protocol; +} + +std::optional ProfileRegistry::parseMapping( + const QJsonObject& o) +{ + softbus::core::models::DataPointMapping m; + m.pointId = o.value(QStringLiteral("pointId")).toString(); + m.metadataId = o.value(QStringLiteral("metadataId")).toString(); + m.registerAddress = static_cast( + o.value(QStringLiteral("registerAddress")).toInt(static_cast(m.registerAddress))); + m.byteOffset = static_cast(o.value(QStringLiteral("byteOffset")).toInt(0)); + m.byteLength = static_cast(o.value(QStringLiteral("byteLength")).toInt(0)); + m.bitOffset = static_cast(o.value(QStringLiteral("bitOffset")).toInt(-1)); + m.endianness = parseEndianness(o.value(QStringLiteral("endianness")).toString()); + m.rawType = parseDataType(o.value(QStringLiteral("rawType")).toString()); + m.scaleFactor = o.value(QStringLiteral("scaleFactor")).toDouble(1.0); + m.offsetFactor = o.value(QStringLiteral("offsetFactor")).toDouble(0.0); + m.expression = o.value(QStringLiteral("expression")).toString(); + if (m.pointId.isEmpty() || m.metadataId.isEmpty() || m.byteLength == 0) { + return std::nullopt; + } + return m; +} + +softbus::core::models::DataType ProfileRegistry::parseDataType(const QString& s) +{ + const QString x = s.trimmed().toUpper(); + using softbus::core::models::DataType; + if (x == QStringLiteral("BOOL")) { + return DataType::BOOL; + } + if (x == QStringLiteral("INT32")) { + return DataType::INT32; + } + if (x == QStringLiteral("UINT32")) { + return DataType::UINT32; + } + if (x == QStringLiteral("INT64")) { + return DataType::INT64; + } + if (x == QStringLiteral("FLOAT32")) { + return DataType::FLOAT32; + } + if (x == QStringLiteral("FLOAT64")) { + return DataType::FLOAT64; + } + if (x == QStringLiteral("STRING")) { + return DataType::STRING; + } + if (x == QStringLiteral("BYTE_ARRAY")) { + return DataType::BYTE_ARRAY; + } + return DataType::INT32; +} + +softbus::core::models::Endianness ProfileRegistry::parseEndianness(const QString& s) +{ + const QString x = s.trimmed().toUpper(); + using softbus::core::models::Endianness; + if (x == QStringLiteral("LITTLE_ENDIAN")) { + return Endianness::LittleEndian; + } + if (x == QStringLiteral("BIG_ENDIAN_BYTE_SWAP")) { + return Endianness::BigEndianByteSwap; + } + return Endianness::BigEndian; +} + +} // namespace softbus::core::metadata diff --git a/src/core/metadata/ProfileRegistry.h b/src/core/metadata/ProfileRegistry.h new file mode 100644 index 0000000..6dc0921 --- /dev/null +++ b/src/core/metadata/ProfileRegistry.h @@ -0,0 +1,35 @@ +#pragma once + +#include + +#include +#include +#include + +#include "core/models/DataMapping.h" + +namespace softbus::core::metadata +{ + +class ProfileRegistry +{ +public: + bool loadFromFile(const QString& filePath); + const softbus::core::models::DeviceDataProfile* find(const QString& deviceId, + const QString& protocol) const; + + const QString& lastError() const { return m_lastError; } + bool empty() const { return m_profilesByKey.isEmpty(); } + +private: + static QString makeKey(const QString& deviceId, const QString& protocol); + static std::optional parseMapping(const QJsonObject& o); + static softbus::core::models::DataType parseDataType(const QString& s); + static softbus::core::models::Endianness parseEndianness(const QString& s); + +private: + QHash m_profilesByKey; + QString m_lastError; +}; + +} // namespace softbus::core::metadata diff --git a/src/core/models/DOMMessage.h b/src/core/models/DOMMessage.h new file mode 100644 index 0000000..5e5a789 --- /dev/null +++ b/src/core/models/DOMMessage.h @@ -0,0 +1,39 @@ +#pragma once +// 定义DOM消息 +#include +#include + +#include + +#include "core/models/MetadataDef.h" +#include "core/models/Types.h" + +namespace softbus::core::models +{ + +using DOMValue = std::variant; + +class DOMMessage +{ +public: + DOMMessage() = default; + + QString messageId; + const MetadataDef* metaRef{nullptr}; + std::int64_t timestamp{0}; + DOMValue value{0.0}; + DataQuality quality{DataQuality::BAD}; + + DataDomain domain() const { return metaRef ? metaRef->domain : DataDomain::UNKNOWN; } + + template + T getValueAs() const + { + if (const auto* p = std::get_if(&value)) { + return *p; + } + return T{}; + } +}; + +} // namespace softbus::core::models diff --git a/src/core/models/DataMapping.h b/src/core/models/DataMapping.h new file mode 100644 index 0000000..8f96dda --- /dev/null +++ b/src/core/models/DataMapping.h @@ -0,0 +1,42 @@ +#pragma once + +#include + +#include +#include + +#include "core/models/Types.h" + +namespace softbus::core::models +{ + +enum class Endianness : std::uint8_t +{ + BigEndian = 0, + LittleEndian, + BigEndianByteSwap +}; + +struct DataPointMapping +{ + QString pointId; + QString metadataId; + std::uint32_t registerAddress{0xFFFFFFFFu}; // 可选：用于按地址匹配 + std::uint16_t byteOffset{0}; + std::uint16_t byteLength{0}; + std::int8_t bitOffset{-1}; + Endianness endianness{Endianness::BigEndian}; + DataType rawType{DataType::INT32}; + double scaleFactor{1.0}; + double offsetFactor{0.0}; + QString expression; +}; + +struct DeviceDataProfile +{ + QString deviceId; + QString protocol; + QVector mappings; +}; + +} // namespace softbus::core::models diff --git a/src/core/models/MetadataDef.h b/src/core/models/MetadataDef.h new file mode 100644 index 0000000..42ec20b --- /dev/null +++ b/src/core/models/MetadataDef.h @@ -0,0 +1,24 @@ +#pragma once +// 定义元数据定义 +#include + +#include "core/models/Types.h" + +namespace softbus::core::models +{ + +struct MetadataDef +{ + QString metadataId; + QString name; + QString description; + DataDomain domain{DataDomain::UNKNOWN}; + DataType semanticType{DataType::FLOAT64}; + QString unit; + double minValue{0.0}; + double maxValue{0.0}; + double deadband{0.0}; + bool isWriteable{false}; +}; + +} // namespace softbus::core::models diff --git a/src/core/models/Types.h b/src/core/models/Types.h new file mode 100644 index 0000000..942d67b --- /dev/null +++ b/src/core/models/Types.h @@ -0,0 +1,38 @@ +#pragma once +// 定义数据域类型、数据质量类型、数据类型 +#include + +namespace softbus::core::models +{ + +enum class DataDomain : std::uint8_t +{ + PROCESS_PARAM = 0, // 过程参数 + CONTROL_CMD = 1, // 控制指令 + MONITOR_DATA = 2, + HEALTH_DIAG = 3, + CONFIG_PARAM = 4, + UNKNOWN = 255 +}; + +enum class DataQuality : std::uint8_t +{ + GOOD = 0, + BAD = 1, + UNCERTAIN = 2, + STALE = 3 +}; + +enum class DataType : std::uint8_t +{ + BOOL = 0, + INT32, + UINT32, + INT64, + FLOAT32, + FLOAT64, + STRING, + BYTE_ARRAY +}; + +} // namespace softbus::core::models diff --git a/src/core/threading/MutexQueue.h b/src/core/threading/MutexQueue.h new file mode 100644 index 0000000..f78ec53 --- /dev/null +++ b/src/core/threading/MutexQueue.h @@ -0,0 +1,79 @@ +#pragma once + +#include +#include +#include +#include +#include + +#include + +namespace softbus::core::threading +{ + +/// 多生产者多消费者安全队列（互斥 + 条件变量）。 +template +class MutexQueue +{ +public: + explicit MutexQueue(std::size_t maxDepth = 0) + : m_maxDepth(maxDepth) + { + } + + bool tryPush(T value) + { + std::lock_guard lk(m_mutex); + if (m_maxDepth > 0 && m_queue.size() >= m_maxDepth) { + return false; + } + m_queue.push_back(std::move(value)); + m_cv.notify_all(); + return true; + } + + void push(T value) + { + std::lock_guard lk(m_mutex); + m_queue.push_back(std::move(value)); + m_cv.notify_all(); + } + + /// `running=false` 且队列空时返回 false。 + bool popWait(T& out, const std::atomic& running) + { + std::unique_lock lk(m_mutex); + m_cv.wait(lk, [&] { return !m_queue.empty() || !running.load(std::memory_order_acquire); }); + if (m_queue.empty()) { + return false; + } + out = std::move(m_queue.front()); + m_queue.pop_front(); + return true; + } + + template + bool popWaitFor(T& out, const std::atomic& running, std::chrono::duration timeout) + { + std::unique_lock lk(m_mutex); + if (!m_cv.wait_for(lk, timeout, [&] { + return !m_queue.empty() || !running.load(std::memory_order_acquire); + })) { + return false; + } + if (m_queue.empty()) { + return false; + } + out = std::move(m_queue.front()); + m_queue.pop_front(); + return true; + } + +private: + std::mutex m_mutex; + std::condition_variable m_cv; + std::deque m_queue; + std::size_t m_maxDepth; +}; + +} // namespace softbus::core::threading diff --git a/src/message_bus/pipeline/protocol/ProtocolEnvelope.h b/src/message_bus/pipeline/protocol/ProtocolEnvelope.h new file mode 100644 index 0000000..890eeeb --- /dev/null +++ b/src/message_bus/pipeline/protocol/ProtocolEnvelope.h @@ -0,0 +1,25 @@ +#pragma once + +#include + +#include "message_bus/pipeline/protocol/modbusrtu/ModbusRtuPdu.h" + +namespace softbus::message_bus::pipeline::protocol +{ + +enum class ProtocolFamily : std::uint8_t +{ + None = 0, + ModbusRtu, +}; + +/// 流水线内协议解析结果的统一外壳；后续可在此 variant 中增加 CAN 等协议。 +using ProtocolEnvelopePdu = std::variant; + +struct ProtocolEnvelope +{ + ProtocolFamily family{ProtocolFamily::None}; + ProtocolEnvelopePdu pdu{}; +}; + +} // namespace softbus::message_bus::pipeline::protocol diff --git a/src/message_bus/pipeline/protocol/modbusrtu/ModbusRtuCodec.cpp b/src/message_bus/pipeline/protocol/modbusrtu/ModbusRtuCodec.cpp new file mode 100644 index 0000000..807f520 --- /dev/null +++ b/src/message_bus/pipeline/protocol/modbusrtu/ModbusRtuCodec.cpp @@ -0,0 +1,428 @@ +#include "message_bus/pipeline/protocol/modbusrtu/ModbusRtuCodec.h" + +#include + +#include "message_bus/pipeline/PipelineContext.h" +#include "message_bus/pipeline/stages/1_framers/ModbusRtuFraming.h" + +namespace softbus::message_bus::pipeline::protocol +{ + +namespace +{ + +using softbus::message_bus::pipeline::BusDirection; +using softbus::message_bus::pipeline::stages::framers::modbusRtuCrc16; +using softbus::message_bus::pipeline::stages::framers::modbusRtuCrcOk; + +void appendCrcLe(QByteArray& adu) +{ + const auto crc = modbusRtuCrc16(reinterpret_cast(adu.constData()), + static_cast(adu.size())); + adu.append(static_cast(crc & 0xFF)); + adu.append(static_cast((crc >> 8) & 0xFF)); +} + +} // namespace + +// 功能 ：解码Modbus RTU帧 如果解码失败 则返回false 如果解码成功 则返回true +bool decodeModbusRtu(BusDirection dir, + const std::uint8_t* p, + std::size_t n, + ModbusRtuPdu& out, + QString* errMsg) +{ + auto fail = [&](const char* msg) { + if (errMsg) { + *errMsg = QString::fromUtf8(msg); + } + return false; + }; + + if (!p || n < 4) { + return fail("adu_too_short"); + } + + if (p[1] & 0x80) { + if (n < 5) { + return fail("exception_incomplete"); + } + if (!modbusRtuCrcOk(p, 5)) { + return fail("exception_crc"); + } + out = ModbusExceptionPdu{p[0], p[1], p[2]}; + return true; + } + + const std::uint8_t unit = p[0]; + const std::uint8_t fc = p[1]; + auto parseReadReq = [&]() -> bool { + if (n != 8 || !modbusRtuCrcOk(p, 8)) { + return false; + } + const std::uint16_t addr = static_cast((p[2] << 8) | p[3]); + const std::uint16_t qty = static_cast((p[4] << 8) | p[5]); + switch (fc) { + case 0x01: + out = ReadCoilsRequest{unit, addr, qty}; + return true; + case 0x02: + out = ReadDiscreteInputsRequest{unit, addr, qty}; + return true; + case 0x03: + out = ReadHoldingRegistersRequest{unit, addr, qty}; + return true; + case 0x04: + out = ReadInputRegistersRequest{unit, addr, qty}; + return true; + default: + return false; + } + }; + + auto parseWriteSingleReqOrResp = [&]() -> bool { + if (n != 8 || !modbusRtuCrcOk(p, 8)) { + return false; + } + const std::uint16_t addr = static_cast((p[2] << 8) | p[3]); + const std::uint16_t val16 = static_cast((p[4] << 8) | p[5]); + if (fc == 0x05) { + const std::uint8_t coil = (val16 == 0xFF00) ? 1 : 0; + if (dir == BusDirection::Upstream) { + out = WriteSingleCoilResponse{unit, addr, coil}; + } else { + out = WriteSingleCoilRequest{unit, addr, coil}; + } + return true; + } + if (fc == 0x06) { + if (dir == BusDirection::Upstream) { + out = WriteSingleRegisterResponse{unit, addr, val16}; + } else { + out = WriteSingleRegisterRequest{unit, addr, val16}; + } + return true; + } + return false; + }; + + auto parseWriteMultiReq = [&]() -> bool { + if (n < 9) { + return false; + } + const std::uint16_t start = static_cast((p[2] << 8) | p[3]); + const std::uint16_t qty = static_cast((p[4] << 8) | p[5]); + const std::uint8_t byteCount = p[6]; + if (n != static_cast(9) + byteCount) { + return false; + } + if (!modbusRtuCrcOk(p, n)) { + return false; + } + if (fc == 0x0F) { + QVector vals; + vals.reserve(byteCount); + for (std::uint8_t i = 0; i < byteCount; ++i) { + vals.append(p[7 + i]); + } + out = WriteMultipleCoilsRequest{unit, start, qty, std::move(vals)}; + return true; + } + if (fc == 0x10) { + QVector vals; + vals.reserve(byteCount / 2); + for (std::uint8_t i = 0; i + 1 < byteCount; i += 2) { + vals.append(static_cast((p[7 + i] << 8) | p[7 + i + 1])); + } + out = WriteMultipleRegistersRequest{unit, start, qty, std::move(vals)}; + return true; + } + return false; + }; + + auto parseWriteMultiResp = [&]() -> bool { + if (n != 8 || !modbusRtuCrcOk(p, 8)) { + return false; + } + const std::uint16_t start = static_cast((p[2] << 8) | p[3]); + const std::uint16_t qty = static_cast((p[4] << 8) | p[5]); + if (fc == 0x0F) { + out = WriteMultipleCoilsResponse{unit, start, qty, {}}; + return true; + } + if (fc == 0x10) { + out = WriteMultipleRegistersResponse{unit, start, qty, {}}; + return true; + } + return false; + }; + + // 功能 ：解析Modbus的读取响应帧 + // function code range 01 02 03 04 + auto parseReadResp = [&]() -> bool { + if (n < 5) { + return false; + } + const std::uint8_t bc = p[2]; + const std::size_t total = static_cast(5) + bc; + if (n < total || !modbusRtuCrcOk(p, total)) { + return false; + } + if (fc == 0x01) { + QVector coils; + coils.reserve(bc); + for (std::uint8_t i = 0; i < bc; ++i) { + coils.append(p[3 + i]); + } + out = ReadCoilsResponse{unit, std::move(coils)}; + return true; + } + if (fc == 0x02) { + QVector inputs; + inputs.reserve(bc); + for (std::uint8_t i = 0; i < bc; ++i) { + inputs.append(p[3 + i]); + } + out = ReadDiscreteInputsResponse{unit, std::move(inputs)}; + return true; + } + if (fc == 0x03 || fc == 0x04) { + QVector regs; + regs.reserve(bc / 2); + for (std::uint8_t i = 0; i + 1 < bc; i += 2) { + regs.append(static_cast((p[3 + i] << 8) | p[3 + i + 1])); + } + if (fc == 0x03) { + out = ReadHoldingRegistersResponse{unit, std::move(regs)}; + } else { + out = ReadInputRegistersResponse{unit, std::move(regs)}; + } + return true; + } + return false; + }; + + // if (dir == BusDirection::Downstream) { + // // 功能 ：解析Modbus的下行请求帧 + // // function code range 01 02 03 04 05 06 0F 10 + // if (parseReadReq() || parseWriteSingleReqOrResp() || parseWriteMultiReq()) { + // return true; + // } + // out = ModbusRtuUnsupportedPdu{unit, fc}; + // return true; + // } + + // if (dir == BusDirection::Upstream) { + // // 功能 ：解析Modbus的上行响应帧 + // // function code range 01 02 03 04 05 06 0F 10 + // if (parseReadResp() || parseWriteSingleReqOrResp() || parseWriteMultiResp()) { + // return true; + // } + // // 功能 ：解析Modbus的下行请求帧 + // if (parseReadReq() || parseWriteMultiReq()) { + // return true; + // } + // out = ModbusRtuUnsupportedPdu{unit, fc}; + // return true; + // } + + if (parseReadReq() || parseWriteSingleReqOrResp() || parseWriteMultiReq() || parseReadResp() + || parseWriteMultiResp()) { + return true; + } + out = ModbusRtuUnsupportedPdu{unit, fc}; + return true; +} + +bool encodeModbusRtu(const ModbusRtuPdu& pdu, QByteArray& out, QString* errMsg) +{ + auto fail = [&](const char* msg) { + if (errMsg) { + *errMsg = QString::fromUtf8(msg); + } + return false; + }; + + out.clear(); + + if (const auto* ex = std::get_if(&pdu)) { + out.append(static_cast(ex->unitId)); + out.append(static_cast(ex->function)); + out.append(static_cast(ex->exceptionCode)); + appendCrcLe(out); + return true; + } + auto encodeReadReq = [&](std::uint8_t unit, std::uint8_t fc, std::uint16_t start, std::uint16_t qty) { + out.append(static_cast(unit)); + out.append(static_cast(fc)); + out.append(static_cast((start >> 8) & 0xFF)); + out.append(static_cast(start & 0xFF)); + out.append(static_cast((qty >> 8) & 0xFF)); + out.append(static_cast(qty & 0xFF)); + appendCrcLe(out); + return true; + }; + auto encodeReadRegResp = [&](std::uint8_t unit, std::uint8_t fc, const QVector& regs) { + if (regs.size() > 127) { + return false; + } + out.append(static_cast(unit)); + out.append(static_cast(fc)); + out.append(static_cast(regs.size() * 2)); + for (std::uint16_t v : regs) { + out.append(static_cast((v >> 8) & 0xFF)); + out.append(static_cast(v & 0xFF)); + } + appendCrcLe(out); + return true; + }; + auto encodeReadBitsResp = [&](std::uint8_t unit, std::uint8_t fc, const QVector& bytes) { + if (bytes.size() > 255) { + return false; + } + out.append(static_cast(unit)); + out.append(static_cast(fc)); + out.append(static_cast(bytes.size())); + for (std::uint8_t b : bytes) { + out.append(static_cast(b)); + } + appendCrcLe(out); + return true; + }; + if (const auto* req = std::get_if(&pdu)) { + return encodeReadReq(req->unitId, 0x01, req->startAddress, req->quantity); + } + if (const auto* req = std::get_if(&pdu)) { + return encodeReadReq(req->unitId, 0x02, req->startAddress, req->quantity); + } + if (const auto* req = std::get_if(&pdu)) { + return encodeReadReq(req->unitId, 0x03, req->startAddress, req->quantity); + } + if (const auto* req = std::get_if(&pdu)) { + return encodeReadReq(req->unitId, 0x04, req->startAddress, req->quantity); + } + if (const auto* resp = std::get_if(&pdu)) { + if (!encodeReadRegResp(resp->unitId, 0x03, resp->registers)) { + return fail("too_many_registers"); + } + return true; + } + if (const auto* resp = std::get_if(&pdu)) { + if (!encodeReadRegResp(resp->unitId, 0x04, resp->inputRegisters)) { + return fail("too_many_registers"); + } + return true; + } + if (const auto* resp = std::get_if(&pdu)) { + if (!encodeReadBitsResp(resp->unitId, 0x01, resp->coils)) { + return fail("too_many_bytes"); + } + return true; + } + if (const auto* resp = std::get_if(&pdu)) { + if (!encodeReadBitsResp(resp->unitId, 0x02, resp->discreteInputs)) { + return fail("too_many_bytes"); + } + return true; + } + if (const auto* req = std::get_if(&pdu)) { + out.append(static_cast(req->unitId)); + out.append(static_cast(0x05)); + out.append(static_cast((req->address >> 8) & 0xFF)); + out.append(static_cast(req->address & 0xFF)); + const std::uint16_t coilWord = req->value ? 0xFF00 : 0x0000; + out.append(static_cast((coilWord >> 8) & 0xFF)); + out.append(static_cast(coilWord & 0xFF)); + appendCrcLe(out); + return true; + } + if (const auto* resp = std::get_if(&pdu)) { + out.append(static_cast(resp->unitId)); + out.append(static_cast(0x05)); + out.append(static_cast((resp->address >> 8) & 0xFF)); + out.append(static_cast(resp->address & 0xFF)); + const std::uint16_t coilWord = resp->value ? 0xFF00 : 0x0000; + out.append(static_cast((coilWord >> 8) & 0xFF)); + out.append(static_cast(coilWord & 0xFF)); + appendCrcLe(out); + return true; + } + if (const auto* req = std::get_if(&pdu)) { + out.append(static_cast(req->unitId)); + out.append(static_cast(0x06)); + out.append(static_cast((req->address >> 8) & 0xFF)); + out.append(static_cast(req->address & 0xFF)); + out.append(static_cast((req->value >> 8) & 0xFF)); + out.append(static_cast(req->value & 0xFF)); + appendCrcLe(out); + return true; + } + if (const auto* resp = std::get_if(&pdu)) { + out.append(static_cast(resp->unitId)); + out.append(static_cast(0x06)); + out.append(static_cast((resp->address >> 8) & 0xFF)); + out.append(static_cast(resp->address & 0xFF)); + out.append(static_cast((resp->value >> 8) & 0xFF)); + out.append(static_cast(resp->value & 0xFF)); + appendCrcLe(out); + return true; + } + if (const auto* req = std::get_if(&pdu)) { + if (req->values.size() > 255) { + return fail("too_many_bytes"); + } + out.append(static_cast(req->unitId)); + out.append(static_cast(0x0F)); + out.append(static_cast((req->startAddress >> 8) & 0xFF)); + out.append(static_cast(req->startAddress & 0xFF)); + out.append(static_cast((req->quantity >> 8) & 0xFF)); + out.append(static_cast(req->quantity & 0xFF)); + out.append(static_cast(req->values.size())); + for (std::uint8_t b : req->values) { + out.append(static_cast(b)); + } + appendCrcLe(out); + return true; + } + if (const auto* resp = std::get_if(&pdu)) { + out.append(static_cast(resp->unitId)); + out.append(static_cast(0x0F)); + out.append(static_cast((resp->startAddress >> 8) & 0xFF)); + out.append(static_cast(resp->startAddress & 0xFF)); + out.append(static_cast((resp->quantity >> 8) & 0xFF)); + out.append(static_cast(resp->quantity & 0xFF)); + appendCrcLe(out); + return true; + } + if (const auto* req = std::get_if(&pdu)) { + if (req->values.size() > 127) { + return fail("too_many_registers"); + } + out.append(static_cast(req->unitId)); + out.append(static_cast(0x10)); + out.append(static_cast((req->startAddress >> 8) & 0xFF)); + out.append(static_cast(req->startAddress & 0xFF)); + out.append(static_cast((req->quantity >> 8) & 0xFF)); + out.append(static_cast(req->quantity & 0xFF)); + out.append(static_cast(req->values.size() * 2)); + for (std::uint16_t v : req->values) { + out.append(static_cast((v >> 8) & 0xFF)); + out.append(static_cast(v & 0xFF)); + } + appendCrcLe(out); + return true; + } + if (const auto* resp = std::get_if(&pdu)) { + out.append(static_cast(resp->unitId)); + out.append(static_cast(0x10)); + out.append(static_cast((resp->startAddress >> 8) & 0xFF)); + out.append(static_cast(resp->startAddress & 0xFF)); + out.append(static_cast((resp->quantity >> 8) & 0xFF)); + out.append(static_cast(resp->quantity & 0xFF)); + appendCrcLe(out); + return true; + } + return fail("unsupported_encode"); +} + +} // namespace softbus::message_bus::pipeline::protocol diff --git a/src/message_bus/pipeline/protocol/modbusrtu/ModbusRtuCodec.h b/src/message_bus/pipeline/protocol/modbusrtu/ModbusRtuCodec.h new file mode 100644 index 0000000..977db10 --- /dev/null +++ b/src/message_bus/pipeline/protocol/modbusrtu/ModbusRtuCodec.h @@ -0,0 +1,27 @@ +#pragma once + +#include +#include + +#include +#include + +#include "message_bus/pipeline/protocol/modbusrtu/ModbusRtuPdu.h" + +namespace softbus::message_bus::pipeline +{ +enum class BusDirection; +} + +namespace softbus::message_bus::pipeline::protocol +{ + +bool decodeModbusRtu(softbus::message_bus::pipeline::BusDirection dir, + const std::uint8_t* p, + std::size_t n, + ModbusRtuPdu& out, + QString* errMsg = nullptr); + +bool encodeModbusRtu(const ModbusRtuPdu& pdu, QByteArray& out, QString* errMsg = nullptr); + +} // namespace softbus::message_bus::pipeline::protocol diff --git a/src/message_bus/pipeline/protocol/modbusrtu/ModbusRtuPdu.h b/src/message_bus/pipeline/protocol/modbusrtu/ModbusRtuPdu.h new file mode 100644 index 0000000..444882a --- /dev/null +++ b/src/message_bus/pipeline/protocol/modbusrtu/ModbusRtuPdu.h @@ -0,0 +1,304 @@ +#pragma once +// 定义ModbusRTU PDU 类型 包括 +// 异常响应、 +// 读保持寄存器请求、 +// 读保持寄存器响应、不支持的PDU +#include + +#include +#include +#include +#include +#include +#include // 用于定义variant类型 variant类型可以存储多种类型的数据 + +namespace softbus::message_bus::pipeline::protocol +{ + +struct ModbusExceptionPdu +{ + std::uint8_t unitId{0}; + std::uint8_t function{0}; + std::uint8_t exceptionCode{0}; +}; +// 01 功能码 读线圈 +struct ReadCoilsRequest +{ + std::uint8_t unitId{0}; + std::uint16_t startAddress{0}; + std::uint16_t quantity{0}; +}; +// 01 功能码 读线圈响应 +struct ReadCoilsResponse +{ + std::uint8_t unitId{0}; + QVector coils; +}; + +// 02 功能码 读离散输入 +struct ReadDiscreteInputsRequest +{ + std::uint8_t unitId{0}; + std::uint16_t startAddress{0}; + std::uint16_t quantity{0}; +}; +// 02 功能码 读离散输入响应 +struct ReadDiscreteInputsResponse +{ + std::uint8_t unitId{0}; + QVector discreteInputs; +}; + +// 03 功能码 读保持寄存器 +struct ReadHoldingRegistersRequest +{ + std::uint8_t unitId{0}; + std::uint16_t startAddress{0}; + std::uint16_t quantity{0}; +}; + +// 03 功能码 读保持寄存器响应 +struct ReadHoldingRegistersResponse +{ + std::uint8_t unitId{0}; + QVector registers; +}; + +// 04 功能码 读输入寄存器 +struct ReadInputRegistersRequest +{ + std::uint8_t unitId{0}; + std::uint16_t startAddress{0}; + std::uint16_t quantity{0}; +}; +// 04 功能码 读输入寄存器响应 +struct ReadInputRegistersResponse +{ + std::uint8_t unitId{0}; + QVector inputRegisters; +}; + +// 05 功能码 写单个线圈 +struct WriteSingleCoilRequest +{ + std::uint8_t unitId{0}; + std::uint16_t address{0}; + std::uint8_t value{0}; +}; +// 05 功能码 写单个线圈响应 +struct WriteSingleCoilResponse +{ + std::uint8_t unitId{0}; + std::uint16_t address{0}; + std::uint8_t value{0}; +}; +// 06 功能码 写单个寄存器 +struct WriteSingleRegisterRequest +{ + std::uint8_t unitId{0}; + std::uint16_t address{0}; + std::uint16_t value{0}; +}; +// 06 功能码 写单个寄存器响应 +struct WriteSingleRegisterResponse +{ + std::uint8_t unitId{0}; + std::uint16_t address{0}; + std::uint16_t value{0}; +}; +// 15 功能码 写多个线圈 +struct WriteMultipleCoilsRequest +{ + std::uint8_t unitId{0}; + std::uint16_t startAddress{0}; + std::uint16_t quantity{0}; + QVector values; +}; +// 15 功能码 写多个线圈响应 +struct WriteMultipleCoilsResponse +{ + std::uint8_t unitId{0}; + std::uint16_t startAddress{0}; + std::uint16_t quantity{0}; + QVector values; +}; + +// 16 功能码 写多个寄存器 +struct WriteMultipleRegistersRequest +{ + std::uint8_t unitId{0}; + std::uint16_t startAddress{0}; + std::uint16_t quantity{0}; + QVector values; +}; +// 16 功能码 写多个寄存器响应 +struct WriteMultipleRegistersResponse +{ + std::uint8_t unitId{0}; + std::uint16_t startAddress{0}; + std::uint16_t quantity{0}; + QVector values; +}; + +struct ModbusRtuUnsupportedPdu +{ + std::uint8_t unitId{0}; + std::uint8_t functionCode{0}; +}; + +using ModbusRtuPdu = std::variant; +// + +inline QString toText(const ModbusRtuPdu& pdu) +{ + return std::visit([](const auto& value) -> QString { + using T = std::decay_t; + + auto bytesToHex = [](const QVector& bytes) -> QString { + QStringList parts; + parts.reserve(bytes.size()); + for (const std::uint8_t b : bytes) { + parts.push_back(QStringLiteral("0x%1").arg(b, 2, 16, QChar('0'))); + } + return QStringLiteral("[%1]").arg(parts.join(QStringLiteral(", "))); + }; + auto wordsToHex = [](const QVector& words) -> QString { + QStringList parts; + parts.reserve(words.size()); + for (const std::uint16_t w : words) { + parts.push_back(QStringLiteral("0x%1").arg(w, 4, 16, QChar('0'))); + } + return QStringLiteral("[%1]").arg(parts.join(QStringLiteral(", "))); + }; + + if constexpr (std::is_same_v) { + return QStringLiteral("ModbusExceptionPdu: unitId=%1, function=0x%2, exceptionCode=0x%3") + .arg(value.unitId) + .arg(value.function, 2, 16, QChar('0')) + .arg(value.exceptionCode, 2, 16, QChar('0')); + } + if constexpr (std::is_same_v) { + const ReadCoilsRequest& request = value; + return QStringLiteral("ReadCoilsRequest: unitId=%1, startAddress=%2, quantity=%3") + .arg(request.unitId) + .arg(request.startAddress) + .arg(request.quantity); + } + if constexpr (std::is_same_v) { + return QStringLiteral("ReadCoilsResponse: unitId=%1, coils=%2") + .arg(value.unitId) + .arg(bytesToHex(value.coils)); + } + if constexpr (std::is_same_v) { + return QStringLiteral("ReadDiscreteInputsRequest: unitId=%1, startAddress=%2, quantity=%3") + .arg(value.unitId) + .arg(value.startAddress) + .arg(value.quantity); + } + if constexpr (std::is_same_v) { + return QStringLiteral("ReadDiscreteInputsResponse: unitId=%1, discreteInputs=%2") + .arg(value.unitId) + .arg(bytesToHex(value.discreteInputs)); + } + if constexpr (std::is_same_v) { + return QStringLiteral("ReadHoldingRegistersRequest: unitId=%1, startAddress=%2, quantity=%3") + .arg(value.unitId) + .arg(value.startAddress) + .arg(value.quantity); + } + if constexpr (std::is_same_v) { + return QStringLiteral("ReadHoldingRegistersResponse: unitId=%1, registers=%2") + .arg(value.unitId) + .arg(wordsToHex(value.registers)); + } + if constexpr (std::is_same_v) { + return QStringLiteral("ReadInputRegistersRequest: unitId=%1, startAddress=%2, quantity=%3") + .arg(value.unitId) + .arg(value.startAddress) + .arg(value.quantity); + } + if constexpr (std::is_same_v) { + return QStringLiteral("ReadInputRegistersResponse: unitId=%1, inputRegisters=%2") + .arg(value.unitId) + .arg(wordsToHex(value.inputRegisters)); + } + if constexpr (std::is_same_v) { + return QStringLiteral("WriteSingleCoilRequest: unitId=%1, address=%2, value=0x%3") + .arg(value.unitId) + .arg(value.address) + .arg(value.value, 2, 16, QChar('0')); + } + if constexpr (std::is_same_v) { + return QStringLiteral("WriteSingleCoilResponse: unitId=%1, address=%2, value=0x%3") + .arg(value.unitId) + .arg(value.address) + .arg(value.value, 2, 16, QChar('0')); + } + if constexpr (std::is_same_v) { + return QStringLiteral("WriteSingleRegisterRequest: unitId=%1, address=%2, value=0x%3") + .arg(value.unitId) + .arg(value.address) + .arg(value.value, 4, 16, QChar('0')); + } + if constexpr (std::is_same_v) { + return QStringLiteral("WriteSingleRegisterResponse: unitId=%1, address=%2, value=0x%3") + .arg(value.unitId) + .arg(value.address) + .arg(value.value, 4, 16, QChar('0')); + } + if constexpr (std::is_same_v) { + return QStringLiteral("WriteMultipleCoilsRequest: unitId=%1, startAddress=%2, quantity=%3, values=%4") + .arg(value.unitId) + .arg(value.startAddress) + .arg(value.quantity) + .arg(bytesToHex(value.values)); + } + if constexpr (std::is_same_v) { + return QStringLiteral("WriteMultipleCoilsResponse: unitId=%1, startAddress=%2, quantity=%3, values=%4") + .arg(value.unitId) + .arg(value.startAddress) + .arg(value.quantity) + .arg(bytesToHex(value.values)); + } + if constexpr (std::is_same_v) { + return QStringLiteral("WriteMultipleRegistersRequest: unitId=%1, startAddress=%2, quantity=%3, values=%4") + .arg(value.unitId) + .arg(value.startAddress) + .arg(value.quantity) + .arg(wordsToHex(value.values)); + } + if constexpr (std::is_same_v) { + return QStringLiteral("WriteMultipleRegistersResponse: unitId=%1, startAddress=%2, quantity=%3, values=%4") + .arg(value.unitId) + .arg(value.startAddress) + .arg(value.quantity) + .arg(wordsToHex(value.values)); + } + if constexpr (std::is_same_v) { + return QStringLiteral("ModbusRtuUnsupportedPdu: unitId=%1, functionCode=0x%2") + .arg(value.unitId) + .arg(value.functionCode, 2, 16, QChar('0')); + } + return QStringLiteral("Unknown"); + }, pdu); +} + +} // namespace softbus::message_bus::pipeline::protocol + diff --git a/src/message_bus/pipeline/protocol/modbusrtu/ModbusRtuProtocolPlugin.cpp b/src/message_bus/pipeline/protocol/modbusrtu/ModbusRtuProtocolPlugin.cpp new file mode 100644 index 0000000..2e4ef0c --- /dev/null +++ b/src/message_bus/pipeline/protocol/modbusrtu/ModbusRtuProtocolPlugin.cpp @@ -0,0 +1,244 @@ +#include "message_bus/pipeline/protocol/modbusrtu/ModbusRtuProtocolPlugin.h" + +#include "core/plugin_system/IProtocolPlugin.h" +#include "message_bus/pipeline/PipelineContext.h" +#include "message_bus/pipeline/protocol/ProtocolEnvelope.h" +#include "message_bus/pipeline/protocol/modbusrtu/ModbusRtuCodec.h" +#include "message_bus/pipeline/stages/1_framers/ModbusRtuFraming.h" + +#include +#include + +#include +#include +#include + +namespace softbus::core::plugin_system +{ + +namespace +{ + +using softbus::message_bus::pipeline::protocol::ModbusExceptionPdu; +using softbus::message_bus::pipeline::protocol::ModbusRtuPdu; +using softbus::message_bus::pipeline::protocol::ModbusRtuUnsupportedPdu; +using softbus::message_bus::pipeline::protocol::ProtocolFamily; +using softbus::message_bus::pipeline::protocol::ReadCoilsRequest; +using softbus::message_bus::pipeline::protocol::ReadCoilsResponse; +using softbus::message_bus::pipeline::protocol::ReadDiscreteInputsRequest; +using softbus::message_bus::pipeline::protocol::ReadDiscreteInputsResponse; +using softbus::message_bus::pipeline::protocol::ReadHoldingRegistersRequest; +using softbus::message_bus::pipeline::protocol::ReadHoldingRegistersResponse; +using softbus::message_bus::pipeline::protocol::ReadInputRegistersRequest; +using softbus::message_bus::pipeline::protocol::ReadInputRegistersResponse; +using softbus::message_bus::pipeline::protocol::WriteSingleCoilRequest; +using softbus::message_bus::pipeline::protocol::WriteSingleCoilResponse; +using softbus::message_bus::pipeline::protocol::WriteSingleRegisterRequest; +using softbus::message_bus::pipeline::protocol::WriteSingleRegisterResponse; +using softbus::message_bus::pipeline::protocol::WriteMultipleCoilsRequest; +using softbus::message_bus::pipeline::protocol::WriteMultipleCoilsResponse; +using softbus::message_bus::pipeline::protocol::WriteMultipleRegistersRequest; +using softbus::message_bus::pipeline::protocol::WriteMultipleRegistersResponse; +using softbus::message_bus::pipeline::protocol::decodeModbusRtu; + +template +struct overloaded : Ts... +{ + using Ts::operator()...; +}; +template +overloaded(Ts...) -> overloaded; + +void syncLegacyJson(const ModbusRtuPdu& pdu, QJsonObject& o) +{ + o = QJsonObject(); + std::visit( + overloaded{ + [&](const ModbusExceptionPdu& e) { + o.insert(QStringLiteral("unitId"), e.unitId); + o.insert(QStringLiteral("function"), e.function); + o.insert(QStringLiteral("exceptionCode"), e.exceptionCode); + o.insert(QStringLiteral("exception"), true); + }, + [&](const ReadHoldingRegistersResponse& r) { + QJsonArray regs; + for (std::uint16_t v : r.registers) { + regs.append(static_cast(v)); + } + o.insert(QStringLiteral("unitId"), r.unitId); + o.insert(QStringLiteral("function"), 3); + o.insert(QStringLiteral("byteCount"), r.registers.size() * 2); + o.insert(QStringLiteral("registers"), regs); + }, + [&](const ReadHoldingRegistersRequest& r) { + o.insert(QStringLiteral("unitId"), r.unitId); + o.insert(QStringLiteral("function"), 3); + o.insert(QStringLiteral("startAddress"), r.startAddress); + o.insert(QStringLiteral("quantity"), r.quantity); + }, + [&](const ReadCoilsRequest& r) { + o.insert(QStringLiteral("unitId"), r.unitId); + o.insert(QStringLiteral("function"), 1); + o.insert(QStringLiteral("startAddress"), r.startAddress); + o.insert(QStringLiteral("quantity"), r.quantity); + }, + [&](const ReadDiscreteInputsRequest& r) { + o.insert(QStringLiteral("unitId"), r.unitId); + o.insert(QStringLiteral("function"), 2); + o.insert(QStringLiteral("startAddress"), r.startAddress); + o.insert(QStringLiteral("quantity"), r.quantity); + }, + [&](const ReadInputRegistersRequest& r) { + o.insert(QStringLiteral("unitId"), r.unitId); + o.insert(QStringLiteral("function"), 4); + o.insert(QStringLiteral("startAddress"), r.startAddress); + o.insert(QStringLiteral("quantity"), r.quantity); + }, + [&](const ReadCoilsResponse& r) { + QJsonArray bits; + for (std::uint8_t b : r.coils) { + bits.append(static_cast(b)); + } + o.insert(QStringLiteral("unitId"), r.unitId); + o.insert(QStringLiteral("function"), 1); + o.insert(QStringLiteral("byteCount"), r.coils.size()); + o.insert(QStringLiteral("coils"), bits); + }, + [&](const ReadDiscreteInputsResponse& r) { + QJsonArray bits; + for (std::uint8_t b : r.discreteInputs) { + bits.append(static_cast(b)); + } + o.insert(QStringLiteral("unitId"), r.unitId); + o.insert(QStringLiteral("function"), 2); + o.insert(QStringLiteral("byteCount"), r.discreteInputs.size()); + o.insert(QStringLiteral("discreteInputs"), bits); + }, + [&](const ReadInputRegistersResponse& r) { + QJsonArray regs; + for (std::uint16_t v : r.inputRegisters) { + regs.append(static_cast(v)); + } + o.insert(QStringLiteral("unitId"), r.unitId); + o.insert(QStringLiteral("function"), 4); + o.insert(QStringLiteral("byteCount"), r.inputRegisters.size() * 2); + o.insert(QStringLiteral("inputRegisters"), regs); + }, + [&](const WriteSingleCoilRequest& r) { + o.insert(QStringLiteral("unitId"), r.unitId); + o.insert(QStringLiteral("function"), 5); + o.insert(QStringLiteral("address"), r.address); + o.insert(QStringLiteral("value"), r.value); + }, + [&](const WriteSingleCoilResponse& r) { + o.insert(QStringLiteral("unitId"), r.unitId); + o.insert(QStringLiteral("function"), 5); + o.insert(QStringLiteral("address"), r.address); + o.insert(QStringLiteral("value"), r.value); + }, + [&](const WriteSingleRegisterRequest& r) { + o.insert(QStringLiteral("unitId"), r.unitId); + o.insert(QStringLiteral("function"), 6); + o.insert(QStringLiteral("address"), r.address); + o.insert(QStringLiteral("value"), r.value); + }, + [&](const WriteSingleRegisterResponse& r) { + o.insert(QStringLiteral("unitId"), r.unitId); + o.insert(QStringLiteral("function"), 6); + o.insert(QStringLiteral("address"), r.address); + o.insert(QStringLiteral("value"), r.value); + }, + [&](const WriteMultipleCoilsRequest& r) { + QJsonArray vals; + for (std::uint8_t b : r.values) { + vals.append(static_cast(b)); + } + o.insert(QStringLiteral("unitId"), r.unitId); + o.insert(QStringLiteral("function"), 15); + o.insert(QStringLiteral("startAddress"), r.startAddress); + o.insert(QStringLiteral("quantity"), r.quantity); + o.insert(QStringLiteral("values"), vals); + }, + [&](const WriteMultipleCoilsResponse& r) { + o.insert(QStringLiteral("unitId"), r.unitId); + o.insert(QStringLiteral("function"), 15); + o.insert(QStringLiteral("startAddress"), r.startAddress); + o.insert(QStringLiteral("quantity"), r.quantity); + }, + [&](const WriteMultipleRegistersRequest& r) { + QJsonArray vals; + for (std::uint16_t v : r.values) { + vals.append(static_cast(v)); + } + o.insert(QStringLiteral("unitId"), r.unitId); + o.insert(QStringLiteral("function"), 16); + o.insert(QStringLiteral("startAddress"), r.startAddress); + o.insert(QStringLiteral("quantity"), r.quantity); + o.insert(QStringLiteral("values"), vals); + }, + [&](const WriteMultipleRegistersResponse& r) { + o.insert(QStringLiteral("unitId"), r.unitId); + o.insert(QStringLiteral("function"), 16); + o.insert(QStringLiteral("startAddress"), r.startAddress); + o.insert(QStringLiteral("quantity"), r.quantity); + }, + [&](const ModbusRtuUnsupportedPdu&) {}, + [&](const auto&) {}}, + pdu); +} + +class ModbusRtuSession final : public IProtocolSession +{ +public: + bool feed(softbus::message_bus::pipeline::PipelineContext& ctx) override + { + using softbus::message_bus::pipeline::FrameKind; + if (ctx.frameKind != FrameKind::CompleteFrame || !ctx.payload.valid()) { + return false; + } + const std::size_t n = ctx.payloadSize ? ctx.payloadSize : ctx.payload.length; + const std::uint8_t* p = ctx.payload.bytes(); + if (!p || n < 4) { + return false; + } + + ModbusRtuPdu pdu; + if (!decodeModbusRtu(ctx.direction, p, n, pdu, nullptr)) { + ctx.protocol.family = ProtocolFamily::None; + ctx.protocol.pdu = {}; + ctx.parsed = QJsonObject(); + return false; + } + ctx.protocol.family = ProtocolFamily::ModbusRtu; + ctx.protocol.pdu = pdu; + syncLegacyJson(pdu, ctx.parsed); + return true; + } + + void reset() override {} +}; + +} // namespace + +class ModbusRtuProtocolPlugin final : public IProtocolPlugin +{ +public: + QString pluginId() const override { return QStringLiteral("modbus_rtu"); } + + bool supports(const QString& protocolHint) const override + { + return softbus::message_bus::pipeline::stages::framers::protocolHintIsModbusRtu(protocolHint); + } + + std::unique_ptr createSession() override + { + return std::make_unique(); + } +}; + +std::shared_ptr makeModbusRtuProtocolPlugin() +{ + return std::make_shared(); +} + +} // namespace softbus::core::plugin_system diff --git a/src/message_bus/pipeline/protocol/modbusrtu/ModbusRtuProtocolPlugin.h b/src/message_bus/pipeline/protocol/modbusrtu/ModbusRtuProtocolPlugin.h new file mode 100644 index 0000000..d408b98 --- /dev/null +++ b/src/message_bus/pipeline/protocol/modbusrtu/ModbusRtuProtocolPlugin.h @@ -0,0 +1,12 @@ +#pragma once + +#include + +#include "core/plugin_system/IProtocolPlugin.h" + +namespace softbus::core::plugin_system +{ + +std::shared_ptr makeModbusRtuProtocolPlugin(); + +} // namespace softbus::core::plugin_system diff --git a/src/message_bus/pipeline/stages/1_framers/ModbusRtuFramer.cpp b/src/message_bus/pipeline/stages/1_framers/ModbusRtuFramer.cpp new file mode 100644 index 0000000..b1b00e9 --- /dev/null +++ b/src/message_bus/pipeline/stages/1_framers/ModbusRtuFramer.cpp @@ -0,0 +1,201 @@ +#include "message_bus/pipeline/stages/1_framers/ModbusRtuFramer.h" + +#include +#include +#include +#include + +#include "message_bus/pipeline/stages/1_framers/ModbusRtuFraming.h" +#include "utils/logs/logging.h" + +namespace softbus::message_bus::pipeline::stages::framers +{ + +ModbusRtuFramer::ModbusRtuFramer(std::shared_ptr pool, + ModbusRtuFramerConfig cfg, + QString exclusiveEndpoint) + : m_pool(std::move(pool)) + , m_cfg(cfg) + , m_exclusiveEndpoint(std::move(exclusiveEndpoint)) +{ +} + +// 功能 ：追加数据块 将数据块追加到会话中 通过时间戳记录最后接收时间 时间间隔超过最大帧间隔时间 则清空会话 如果超过最大ADU长度 则记录溢出 并记录溢出数量 +void ModbusRtuFramer::appendChunk(Session& session, const std::uint8_t* p, std::size_t n) +{ + if (!p || n == 0) { + return; + } + session.lastRx = std::chrono::steady_clock::now(); + if (session.buf.size() < m_cfg.maxAdu) { + session.buf.resize(m_cfg.maxAdu); + } + if (session.len + n > m_cfg.maxAdu) { + ++m_cnt.overflow; + session.len = 0; + LOG_WARNING() << "ModbusRtuFramer: buffer_overflow maxAdu=" << m_cfg.maxAdu; + return; + } + // `p/n` 来自 chunkIn.payload.bytes()，也就是串口读回的原始字节块（当前函数会再次拷贝进 session.buf） + const std::size_t beforeLen = session.len; + std::ostringstream oss; + oss << std::hex << std::setfill('0'); + const std::size_t show = std::min(n, static_cast(16)); + for (std::size_t i = 0; i < show; ++i) { + oss << std::setw(2) << static_cast(p[i]); + if (i + 1 < show) { + oss << ' '; + } + } + if (n > show) { + oss << " ..."; + } + const QString hexPreview = QString::fromStdString(oss.str()); + // LOG_INFO() << "ModbusRtuFramer: appendChunk n=" << n << " session.len(before)=" << beforeLen + // << " p=" << static_cast(p) << " bytes(16)=" << hexPreview; + std::memcpy(session.buf.data() + session.len, p, n); + session.len += n; + m_cnt.bytesIn += n; +} +// 功能 ：尝试消费缓冲区 如果时间间隔超过最大帧间隔时间 则清空会话 如果超过最大ADU长度 则记录溢出 并记录溢出数量 +void ModbusRtuFramer::tryConsumeBuffer(Session& session, + const QString& endpoint, + QString traceId, + int deviceId, + const QString& stableKey, + const QString& protocolHint, + QDateTime ts, + BusDirection dir, + std::vector& out, + QString* framerErr) +{ + (void)framerErr; + + if (m_cfg.interFrameTimeoutMs > 0 && session.len > 0) { + const auto now = std::chrono::steady_clock::now(); + const auto ms = std::chrono::duration_cast(now - session.lastRx).count(); + if (ms >= m_cfg.interFrameTimeoutMs) { + session.len = 0; + } + } + + while (session.len > 0) { + const std::uint8_t* buf = session.buf.data(); + auto pred = + predictModbusRtuAduLength(buf, session.len, m_cfg.maxAdu); + if (!pred) { + break; + } + if (session.len < *pred) { + break; + } + + // std::ostringstream oss; + // oss << std::hex << std::setfill('0'); + // for (std::size_t i = 0; i < session.len; ++i) { + // oss << std::setw(2) << static_cast(buf[i]); + // if (i + 1 < session.len) { + // oss << ' '; + // } + // } + // const QString hexPreview = QString::fromStdString(oss.str()); + // LOG_INFO() << "ModbusRtuFramer: tryConsumeBuffer buf(16)=" << hexPreview << " session.len=" << session.len << " pred=" << *pred ; + + if (modbusRtuCrcOk(buf, *pred)) { + auto blk = m_pool->allocate(*pred); + if (!blk || !blk->data || blk->capacity < *pred) { + if (framerErr) { + *framerErr = QStringLiteral("pool_exhausted"); + } + LOG_WARNING() << "ModbusRtuFramer: pool_exhausted endpoint=" << endpoint; + return; + } + std::memcpy(blk->data, buf, *pred); + blk->size = *pred; + + softbus::message_bus::pipeline::PipelineContext ctx; + ctx.deviceId = deviceId; + ctx.stableKey = stableKey; + ctx.endpoint = endpoint; + ctx.timestamp = ts; + ctx.protocolHint = protocolHint; + ctx.traceId = traceId; + ctx.direction = dir; + ctx.frameKind = softbus::message_bus::pipeline::FrameKind::CompleteFrame; + ctx.payloadSize = *pred; + ctx.payload.block = std::move(blk); + ctx.payload.offset = 0; + ctx.payload.length = *pred; + + out.push_back(std::move(ctx)); + ++m_cnt.framesEmitted; + + const std::size_t rest = session.len - *pred; + if (rest > 0) { + std::memmove(session.buf.data(), buf + *pred, rest); + } + session.len = rest; + // LOG_INFO() << "ModbusRtuFramer success to consume buffer rest=" << rest; + continue; + } + + ++m_cnt.crcFail; + ++m_cnt.resyncBytes; + // LOG_WARNING() << "ModbusRtuFramer: crc_mismatch resync_byte endpoint=" << endpoint + // << "traceId=" << traceId << "buf_len=" << session.len; + + if (session.len <= 1) { + session.len = 0; + break; + } + std::memmove(session.buf.data(), buf + 1, session.len - 1); + --session.len; + } +} + +void ModbusRtuFramer::feed(softbus::message_bus::pipeline::PipelineContext& chunkIn, + std::vector& completeFramesOut) +{ + if (!m_pool || !chunkIn.payload.valid()) { + return; + } + if (!protocolHintIsModbusRtu(chunkIn.protocolHint)) { + return; + } + + const std::uint8_t* p = chunkIn.payload.bytes(); + const std::size_t n = chunkIn.payloadSize ? chunkIn.payloadSize : chunkIn.payload.length; + if (!p || n == 0) { + return; + } + + Session* session = nullptr; + if (!m_exclusiveEndpoint.isEmpty()) { + session = &m_exclusiveSession; + } else { + session = &m_sessions[chunkIn.endpoint]; + } + + appendChunk(*session, p, n); + + QString err; + tryConsumeBuffer(*session, + chunkIn.endpoint, + chunkIn.traceId, + chunkIn.deviceId, + chunkIn.stableKey, + chunkIn.protocolHint, + chunkIn.timestamp, + chunkIn.direction, + completeFramesOut, + &err); + if (!err.isEmpty()) { + chunkIn.framerError = err; + } + + chunkIn.payload = {}; + chunkIn.payloadSize = 0; + chunkIn.frameKind = softbus::message_bus::pipeline::FrameKind::StreamChunk; +} + +} // namespace softbus::message_bus::pipeline::stages::framers diff --git a/src/message_bus/pipeline/stages/1_framers/ModbusRtuFramer.h b/src/message_bus/pipeline/stages/1_framers/ModbusRtuFramer.h new file mode 100644 index 0000000..8532f3d --- /dev/null +++ b/src/message_bus/pipeline/stages/1_framers/ModbusRtuFramer.h @@ -0,0 +1,79 @@ +#pragma once + +#include +#include +#include +#include +#include +#include + +#include +#include + +#include "core/memory/MemoryPool.h" +#include "message_bus/pipeline/PipelineContext.h" +#include "message_bus/pipeline/stages/1_framers/IFramer.h" + +namespace softbus::message_bus::pipeline::stages::framers +{ + +struct ModbusRtuFramerConfig +{ + std::size_t maxAdu{256}; + int interFrameTimeoutMs{0}; +}; + +struct ModbusRtuFramerCounters +{ + std::atomic bytesIn{0}; // 输入字节数 + std::atomic framesEmitted{0}; // 输出帧数 + std::atomic resyncBytes{0}; // 重同步字节数 + std::atomic crcFail{0}; // CRC校验失败数 + std::atomic overflow{0}; // 溢出数 +}; + +/// 单 worker、多 endpoint：使用内部 QHash（单线程访问 Session，无锁）。 +/// 并行成帧：构造时传入 `exclusiveEndpoint`，仅重组该串口字节流（每 endpoint 一实例）。 +class ModbusRtuFramer final : public IFramer +{ +public: + ModbusRtuFramer(std::shared_ptr pool, + ModbusRtuFramerConfig cfg = {}, + QString exclusiveEndpoint = {}); + + void feed(softbus::message_bus::pipeline::PipelineContext& chunkIn, + std::vector& completeFramesOut) override; + + const ModbusRtuFramerCounters& counters() const { return m_cnt; } + ModbusRtuFramerCounters& counters() { return m_cnt; } + void setConfig(ModbusRtuFramerConfig cfg) { m_cfg = cfg; } + +private: + struct Session + { + std::vector buf; + std::size_t len{0}; + std::chrono::steady_clock::time_point lastRx{std::chrono::steady_clock::now()}; + }; + + void appendChunk(Session& session, const std::uint8_t* p, std::size_t n); + void tryConsumeBuffer(Session& session, + const QString& endpoint, + QString traceId, + int deviceId, + const QString& stableKey, + const QString& protocolHint, + QDateTime ts, + BusDirection dir, + std::vector& out, + QString* framerErr); + + std::shared_ptr m_pool; + ModbusRtuFramerConfig m_cfg; + ModbusRtuFramerCounters m_cnt; + QString m_exclusiveEndpoint; + QHash m_sessions; + Session m_exclusiveSession; +}; + +} // namespace softbus::message_bus::pipeline::stages::framers diff --git a/src/message_bus/pipeline/stages/1_framers/ModbusRtuFraming.cpp b/src/message_bus/pipeline/stages/1_framers/ModbusRtuFraming.cpp new file mode 100644 index 0000000..65518cd --- /dev/null +++ b/src/message_bus/pipeline/stages/1_framers/ModbusRtuFraming.cpp @@ -0,0 +1,125 @@ +#include "message_bus/pipeline/stages/1_framers/ModbusRtuFraming.h" + +#include + +namespace softbus::message_bus::pipeline::stages::framers +{ + +std::uint16_t modbusRtuCrc16(const std::uint8_t* data, std::size_t len) +{ + std::uint16_t crc = 0xFFFF; + for (std::size_t i = 0; i < len; ++i) { + crc ^= data[i]; + for (int k = 0; k < 8; ++k) { + if (crc & 1) { + crc = static_cast((crc >> 1) ^ 0xA001); + } else { + crc = static_cast(crc >> 1); + } + } + } + return crc; +} + +// 功能 ：校验CRC 如果CRC校验不正确 则返回false 如果CRC校验正确 则返回true +// 参数 ：adu 数据缓冲区 aduLen 数据长度 +// 返回值 ：如果CRC校验不正确 则返回false 如果CRC校验正确 则返回true +// 逻辑：如果数据缓冲区为空 则返回false 如果数据长度小于4 则返回false 如果数据长度大于等于4 则返回true +// 如果数据长度大于等于4 则返回true +// 如果数据长度大于等于4 则返回true +bool modbusRtuCrcOk(const std::uint8_t* adu, std::size_t aduLen) +{ + if (!adu || aduLen < 4) { + return false; + } + const std::size_t body = aduLen - 2; + const std::uint16_t got = + static_cast(adu[body] | (static_cast(adu[body + 1]) << 8)); + const std::uint16_t want = modbusRtuCrc16(adu, body); + return got == want; +} + +static bool fcIsReadRange(std::uint8_t fc) +{ + return fc == 0x01 || fc == 0x02 || fc == 0x03 || fc == 0x04; +} + +// 功能 ：预测Modbus RTU帧长度 如果长度不足 则返回nullopt 如果长度足够 则返回长度 +// 参数 ：buf 数据缓冲区 len 数据长度 maxAdu 最大ADU长度 +// 返回值 ：如果长度不足 则返回nullopt 如果长度足够 则返回长度 +// 逻辑： unit 是设备地址 fc 是功能码 如果fc是异常码 则返回5 如果fc是读指令 则返回8 如果fc是写指令 则返回8 如果fc是其他指令 则返回nullopt +// 如果fc是读指令 则检查buf长度是否大于等于8 并且CRC校验是否正确 如果正确 则返回8 +// 如果fc是写指令 则检查buf长度是否大于等于8 并且CRC校验是否正确 如果正确 则返回8 + +std::optional predictModbusRtuAduLength(const std::uint8_t* buf, + std::size_t len, + std::size_t maxAdu) +{ + if (len < 2 || !buf) { + return std::nullopt; + } + const std::uint8_t unit = buf[0]; + (void)unit; + const std::uint8_t fc = buf[1]; + + if (fc & 0x80) { + return len < 5 ? std::nullopt : std::optional(5); + } + + if (fcIsReadRange(fc)) { + if (len >= 8 && modbusRtuCrcOk(buf, 8)) { + return std::optional(8); + } + if (len < 3) { + return std::nullopt; + } + const std::size_t byteCount = buf[2]; + const std::size_t total = 5 + byteCount; + if (total > maxAdu) { + return std::nullopt; + } + if (len < total) { + return std::nullopt; + } + if (modbusRtuCrcOk(buf, total) ) { + return total; + } + return std::optional(8); + } + + if (fc == 0x05 || fc == 0x06) { + return std::optional(8); + } + + if (fc == 0x16) { + return std::optional(10); + } + + if (fc == 0x0F || fc == 0x10) { + if (len >= 8 && modbusRtuCrcOk(buf, 8)) { + return std::optional(8); + } + if (len >= 7) { + const std::uint8_t bc = buf[6]; + const std::size_t req = 9 + static_cast(bc); + if (bc <= 242 && req <= maxAdu && req >= 9) { + return req; + } + } + return len >= 2 ? std::optional(8) : std::nullopt; + } + + if (len < 6) { + return std::nullopt; + } + return std::optional(8); +} + +bool protocolHintIsModbusRtu(const QString& hint) +{ + const QString h = hint.trimmed().toLower(); + return h == QStringLiteral("modbus_rtu") || h == QStringLiteral("modbus-rtu") + || h == QStringLiteral("modbusrtu"); +} + +} // namespace softbus::message_bus::pipeline::stages::framers diff --git a/src/message_bus/pipeline/stages/1_framers/ModbusRtuFraming.h b/src/message_bus/pipeline/stages/1_framers/ModbusRtuFraming.h new file mode 100644 index 0000000..71c4609 --- /dev/null +++ b/src/message_bus/pipeline/stages/1_framers/ModbusRtuFraming.h @@ -0,0 +1,24 @@ +#pragma once +// 功能 ：定义Modbus RTU帧解析器 包含CRC校验、帧预测、协议提示判断 +#include +#include +#include + +#include + +namespace softbus::message_bus::pipeline::stages::framers +{ + +std::uint16_t modbusRtuCrc16(const std::uint8_t* data, std::size_t len); + +/// 校验 ADU（含尾端 CRC LE）；`aduLen` 为整帧长度（unit..crc_hi）。 +bool modbusRtuCrcOk(const std::uint8_t* adu, std::size_t aduLen); + +/// 根据已缓冲前缀预测整帧 ADU 长度（含 2 字节 CRC）；未知或不足返回 nullopt。 +std::optional predictModbusRtuAduLength(const std::uint8_t* buf, + std::size_t len, + std::size_t maxAdu); + +bool protocolHintIsModbusRtu(const QString& hint); + +} // namespace softbus::message_bus::pipeline::stages::framers diff --git a/src/message_bus/pipeline/stages/1_framers/PassthroughFramer.cpp b/src/message_bus/pipeline/stages/1_framers/PassthroughFramer.cpp new file mode 100644 index 0000000..329acfd --- /dev/null +++ b/src/message_bus/pipeline/stages/1_framers/PassthroughFramer.cpp @@ -0,0 +1,54 @@ +#include "message_bus/pipeline/stages/1_framers/PassthroughFramer.h" + +#include + +namespace softbus::message_bus::pipeline::stages::framers +{ + +PassthroughFramer::PassthroughFramer(std::shared_ptr pool) + : m_pool(std::move(pool)) +{ +} + +void PassthroughFramer::feed(softbus::message_bus::pipeline::PipelineContext& chunkIn, + std::vector& completeFramesOut) +{ + if (!m_pool || !chunkIn.payload.valid()) { + return; + } + const std::size_t n = chunkIn.payloadSize ? chunkIn.payloadSize : chunkIn.payload.length; + if (n == 0) { + return; + } + auto blk = m_pool->allocate(n); + if (!blk || !blk->data || blk->capacity < n) { + chunkIn.framerError = QStringLiteral("pool_exhausted"); + return; + } + const std::uint8_t* src = chunkIn.payload.bytes(); + if (!src) { + return; + } + std::memcpy(blk->data, src, n); + blk->size = n; + + softbus::message_bus::pipeline::PipelineContext out; + out.deviceId = chunkIn.deviceId; + out.stableKey = chunkIn.stableKey; + out.endpoint = chunkIn.endpoint; + out.timestamp = chunkIn.timestamp; + out.protocolHint = chunkIn.protocolHint; + out.traceId = chunkIn.traceId; + out.direction = chunkIn.direction; + out.frameKind = softbus::message_bus::pipeline::FrameKind::CompleteFrame; + out.payloadSize = n; + out.payload.block = std::move(blk); + out.payload.offset = 0; + out.payload.length = n; + + completeFramesOut.push_back(std::move(out)); + chunkIn.payload = {}; + chunkIn.payloadSize = 0; +} + +} // namespace softbus::message_bus::pipeline::stages::framers diff --git a/src/message_bus/pipeline/stages/1_framers/PassthroughFramer.h b/src/message_bus/pipeline/stages/1_framers/PassthroughFramer.h new file mode 100644 index 0000000..a7b3dee --- /dev/null +++ b/src/message_bus/pipeline/stages/1_framers/PassthroughFramer.h @@ -0,0 +1,25 @@ +#pragma once + +#include +#include + +#include "core/memory/MemoryPool.h" +#include "message_bus/pipeline/stages/1_framers/IFramer.h" + +namespace softbus::message_bus::pipeline::stages::framers +{ + +/// 将单个 StreamChunk 拷贝为 CompleteFrame（整帧生命周期由新池块持有）。 +class PassthroughFramer final : public IFramer +{ +public: + explicit PassthroughFramer(std::shared_ptr pool); + + void feed(softbus::message_bus::pipeline::PipelineContext& chunkIn, + std::vector& completeFramesOut) override; + +private: + std::shared_ptr m_pool; +}; + +} // namespace softbus::message_bus::pipeline::stages::framers diff --git a/src/message_bus/pipeline/stages/3_filters/ProtocolParseFilter.cpp b/src/message_bus/pipeline/stages/3_filters/ProtocolParseFilter.cpp new file mode 100644 index 0000000..2e41d36 --- /dev/null +++ b/src/message_bus/pipeline/stages/3_filters/ProtocolParseFilter.cpp @@ -0,0 +1,565 @@ +#include "message_bus/pipeline/stages/3_filters/ProtocolParseFilter.h" + +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include + +#include + +#include "message_bus/pipeline/protocol/modbusrtu/ModbusRtuPdu.h" +#include "message_bus/pipeline/protocol/ProtocolEnvelope.h" +#include "utils/logs/logging.h" + +namespace softbus::message_bus::pipeline::stages::filters +{ + +namespace +{ + +using softbus::core::models::DOMMessage; +using softbus::core::models::DataPointMapping; +using softbus::core::models::DataQuality; +using softbus::core::models::DataType; +using softbus::core::models::Endianness; +using softbus::message_bus::pipeline::protocol::ModbusRtuPdu; +using softbus::message_bus::pipeline::protocol::ProtocolFamily; +using softbus::message_bus::pipeline::protocol::ReadHoldingRegistersResponse; + +QByteArray registersToBytes(const QVector& regs) +{ + QByteArray out; + out.reserve(regs.size() * 2); + for (std::uint16_t v : regs) { + out.append(static_cast((v >> 8) & 0xFF)); + out.append(static_cast(v & 0xFF)); + } + return out; +} + +QJsonArray registersToQJsonArray(const QVector& regs) +{ + QJsonArray a; + for (std::uint16_t v : regs) { + a.append(static_cast(v)); + } + return a; +} + +std::uint64_t readUnsignedRaw(const QByteArray& bytes, const DataPointMapping& mapping, bool* ok) +{ + *ok = false; + if (mapping.byteLength == 0 || mapping.byteLength > 8) { + return 0; + } + const int offset = static_cast(mapping.byteOffset); + const int len = static_cast(mapping.byteLength); + if (offset < 0 || offset + len > bytes.size()) { + return 0; + } + + std::array buf{}; + for (int i = 0; i < len; ++i) { + buf[static_cast(i)] = static_cast(bytes[offset + i]); + } + + if (mapping.endianness == Endianness::LittleEndian) { + std::reverse(buf.begin(), buf.begin() + len); + } else if (mapping.endianness == Endianness::BigEndianByteSwap && (len % 2 == 0)) { + for (int i = 0; i + 1 < len; i += 2) { + std::swap(buf[static_cast(i)], buf[static_cast(i + 1)]); + } + } + + std::uint64_t v = 0; + for (int i = 0; i < len; ++i) { + v = (v << 8) | buf[static_cast(i)]; + } + *ok = true; + return v; +} + +double decodeRawToDouble(std::uint64_t raw, const DataPointMapping& mapping) +{ + switch (mapping.rawType) { + case DataType::BOOL: + return raw ? 1.0 : 0.0; + case DataType::INT32: + return static_cast(static_cast(raw & 0xFFFFFFFFu)); + case DataType::UINT32: + return static_cast(static_cast(raw & 0xFFFFFFFFu)); + case DataType::INT64: + return static_cast(static_cast(raw)); + case DataType::FLOAT32: { + const std::uint32_t u = static_cast(raw & 0xFFFFFFFFu); + float f = 0.0f; + std::memcpy(&f, &u, sizeof(float)); + return static_cast(f); + } + case DataType::FLOAT64: { + double d = 0.0; + std::memcpy(&d, &raw, sizeof(double)); + return d; + } + case DataType::STRING: + case DataType::BYTE_ARRAY: + default: + return static_cast(raw); + } +} + +class ExprParser +{ +public: + ExprParser(const QString& expr, const QJsonArray& regs, double x) + : m_expr(expr) + , m_regs(regs) + , m_x(x) + { + } + + bool eval(double* out) + { + m_pos = 0; + if (!out) { + return false; + } + const auto v = parseBitOr(); + skipSpaces(); + if (!v.has_value() || m_pos != m_expr.size()) { + return false; + } + *out = *v; + return true; + } + +private: + std::optional parseBitOr() + { + auto lhs = parseBitAnd(); + if (!lhs.has_value()) { + return std::nullopt; + } + while (true) { + skipSpaces(); + if (!match('|')) { + break; + } + auto rhs = parseBitAnd(); + if (!rhs.has_value()) { + return std::nullopt; + } + const auto lv = static_cast(*lhs); + const auto rv = static_cast(*rhs); + lhs = static_cast(lv | rv); + } + return lhs; + } + + std::optional parseBitAnd() + { + auto lhs = parseShift(); + if (!lhs.has_value()) { + return std::nullopt; + } + while (true) { + skipSpaces(); + if (!match('&')) { + break; + } + auto rhs = parseShift(); + if (!rhs.has_value()) { + return std::nullopt; + } + const auto lv = static_cast(*lhs); + const auto rv = static_cast(*rhs); + lhs = static_cast(lv & rv); + } + return lhs; + } + + std::optional parseShift() + { + auto lhs = parseAddSub(); + if (!lhs.has_value()) { + return std::nullopt; + } + while (true) { + skipSpaces(); + if (match(QStringLiteral("<<"))) { + auto rhs = parseAddSub(); + if (!rhs.has_value()) { + return std::nullopt; + } + lhs = static_cast(static_cast(*lhs) << static_cast(*rhs)); + continue; + } + if (match(QStringLiteral(">>"))) { + auto rhs = parseAddSub(); + if (!rhs.has_value()) { + return std::nullopt; + } + lhs = static_cast(static_cast(*lhs) >> static_cast(*rhs)); + continue; + } + break; + } + return lhs; + } + + std::optional parseAddSub() + { + auto lhs = parseMulDiv(); + if (!lhs.has_value()) { + return std::nullopt; + } + while (true) { + skipSpaces(); + if (match('+')) { + auto rhs = parseMulDiv(); + if (!rhs.has_value()) { + return std::nullopt; + } + lhs = *lhs + *rhs; + continue; + } + if (match('-')) { + auto rhs = parseMulDiv(); + if (!rhs.has_value()) { + return std::nullopt; + } + lhs = *lhs - *rhs; + continue; + } + break; + } + return lhs; + } + + std::optional parseMulDiv() + { + auto lhs = parseUnary(); + if (!lhs.has_value()) { + return std::nullopt; + } + while (true) { + skipSpaces(); + if (match('*')) { + auto rhs = parseUnary(); + if (!rhs.has_value()) { + return std::nullopt; + } + lhs = *lhs * *rhs; + continue; + } + if (match('/')) { + auto rhs = parseUnary(); + if (!rhs.has_value() || *rhs == 0.0) { + return std::nullopt; + } + lhs = *lhs / *rhs; + continue; + } + break; + } + return lhs; + } + + std::optional parseUnary() + { + skipSpaces(); + if (match('-')) { + auto v = parseUnary(); + return v.has_value() ? std::optional(-*v) : std::nullopt; + } + if (match('+')) { + return parseUnary(); + } + return parsePrimary(); + } + + std::optional parsePrimary() + { + skipSpaces(); + if (match('(')) { + auto v = parseBitOr(); + skipSpaces(); + if (!v.has_value() || !match(')')) { + return std::nullopt; + } + return v; + } + + if (peekIsAlpha()) { + return parseVariable(); + } + return parseNumber(); + } + + std::optional parseVariable() + { + const auto name = parseIdentifier(); + if (name == QStringLiteral("x")) { + return m_x; + } + if (name.startsWith(QStringLiteral("r"))) { + bool ok = false; + const int idx = name.mid(1).toInt(&ok); + if (ok && idx >= 0 && idx < m_regs.size()) { + return static_cast(m_regs[idx].toInt(0)); + } + } + return std::nullopt; + } + + std::optional parseNumber() + { + skipSpaces(); + const int start = m_pos; + bool dot = false; + while (m_pos < m_expr.size()) { + const QChar c = m_expr[m_pos]; + if (c.isDigit()) { + ++m_pos; + continue; + } + if (c == QChar('.')) { + if (dot) { + break; + } + dot = true; + ++m_pos; + continue; + } + break; + } + if (m_pos <= start) { + return std::nullopt; + } + bool ok = false; + const double v = m_expr.mid(start, m_pos - start).toDouble(&ok); + if (!ok) { + return std::nullopt; + } + return v; + } + + QString parseIdentifier() + { + skipSpaces(); + const int start = m_pos; + while (m_pos < m_expr.size()) { + const QChar c = m_expr[m_pos]; + if (c.isLetterOrNumber() || c == QChar('_')) { + ++m_pos; + continue; + } + break; + } + return m_expr.mid(start, m_pos - start); + } + + bool match(QChar c) + { + if (m_pos < m_expr.size() && m_expr[m_pos] == c) { + ++m_pos; + return true; + } + return false; + } + + bool match(const QString& token) + { + if (m_expr.mid(m_pos, token.size()) == token) { + m_pos += token.size(); + return true; + } + return false; + } + + bool peekIsAlpha() const + { + return m_pos < m_expr.size() && (m_expr[m_pos].isLetter() || m_expr[m_pos] == QChar('_')); + } + + void skipSpaces() + { + while (m_pos < m_expr.size() && m_expr[m_pos].isSpace()) { + ++m_pos; + } + } + +private: + QString m_expr; + QJsonArray m_regs; + double m_x{0.0}; + int m_pos{0}; +}; + +softbus::core::models::DOMValue castToSemantic(double value, DataType t) +{ + using softbus::core::models::DOMValue; + switch (t) { + case DataType::BOOL: + return DOMValue{value != 0.0}; + case DataType::INT32: + return DOMValue{static_cast(value)}; + case DataType::UINT32: + return DOMValue{static_cast(value)}; + case DataType::INT64: + return DOMValue{static_cast(value)}; + case DataType::FLOAT32: + return DOMValue{static_cast(value)}; + case DataType::FLOAT64: + default: + return DOMValue{value}; + } +} + +DataQuality qualityByRange(double value, const softbus::core::models::MetadataDef& meta) +{ + if (!std::isfinite(value)) { + return DataQuality::BAD; + } + if (meta.maxValue > meta.minValue && (value < meta.minValue || value > meta.maxValue)) { + return DataQuality::UNCERTAIN; + } + return DataQuality::GOOD; +} + +QString normalizedProtocol(const QString& protocolHint) +{ + return protocolHint.trimmed().toLower(); +} + +} // namespace + +ProtocolParseFilter::ProtocolParseFilter() = default; + +bool ProtocolParseFilter::apply(softbus::message_bus::pipeline::PipelineContext& ctx) +{ + if (ctx.direction == softbus::message_bus::pipeline::BusDirection::Downstream) { + return true; + } + if (!ensureLoaded()) { + return true; // 保持兼容：配置缺失不阻断流水线 + } + if (normalizedProtocol(ctx.protocolHint).contains(QStringLiteral("modbus"))) { + return parseModbus(ctx); + } + return true; +} + +bool ProtocolParseFilter::ensureLoaded() +{ + std::call_once(m_loadOnce, [this]() { + QString configBase = qEnvironmentVariable("SOFTBUS_CONFIG_DIR"); + if (!configBase.isEmpty() && !configBase.endsWith(QChar('/'))) { + configBase.append(QChar('/')); + } + const QString homeCfg = configBase.isEmpty() + ? (QDir::homePath() + QStringLiteral("/softbus/config/")) + : configBase; + const QString metaHome = homeCfg + QStringLiteral("metadata_dictionary.json"); + const QString profHome = homeCfg + QStringLiteral("device_profiles.json"); + const QString metaLocal = QStringLiteral("config/metadata_dictionary.json"); + const QString profLocal = QStringLiteral("config/device_profiles.json"); + + const QString metaPath = QFileInfo::exists(metaHome) ? metaHome : metaLocal; + const QString profPath = QFileInfo::exists(profHome) ? profHome : profLocal; + + const bool metaOk = m_metadataRegistry.loadFromFile(metaPath); + const bool profOk = m_profileRegistry.loadFromFile(profPath); + if (!metaOk || !profOk) { + LOG_WARNING() << "ProtocolParseFilter: config load failed metadata=" + << m_metadataRegistry.lastError() << " profile=" + << m_profileRegistry.lastError(); + m_loaded = false; + return; + } + m_loaded = true; + }); + return m_loaded; +} + +bool ProtocolParseFilter::parseModbus(softbus::message_bus::pipeline::PipelineContext& ctx) +{ + QVector regsVec; + QJsonArray regsForExpr; + + if (ctx.protocol.family == ProtocolFamily::ModbusRtu) { + if (const auto* m = std::get_if(&ctx.protocol.pdu)) { + if (const auto* resp = std::get_if(&*m)) { + regsVec = resp->registers; + regsForExpr = registersToQJsonArray(regsVec); + } + } + } + if (regsVec.isEmpty()) { + const QJsonArray regsLegacy = ctx.parsed.value(QStringLiteral("registers")).toArray(); + if (regsLegacy.isEmpty()) { + return true; + } + regsForExpr = regsLegacy; + for (const auto& reg : regsLegacy) { + regsVec.append(static_cast(reg.toInt(0) & 0xFFFF)); + } + } + + const auto* profile = m_profileRegistry.find(ctx.stableKey, ctx.protocolHint); + if (!profile) { + profile = m_profileRegistry.find(QString::number(ctx.deviceId), ctx.protocolHint); + } + if (!profile) { + return true; + } + + const QByteArray bytes = registersToBytes(regsVec); + ctx.domMessages.clear(); + ctx.domMessages.reserve(profile->mappings.size()); + + for (const auto& mapping : profile->mappings) { + const auto* meta = m_metadataRegistry.findById(mapping.metadataId); + if (!meta) { + continue; + } + + bool ok = false; + const std::uint64_t raw = readUnsignedRaw(bytes, mapping, &ok); + if (!ok) { + continue; + } + + double v = decodeRawToDouble(raw, mapping); + v = v * mapping.scaleFactor + mapping.offsetFactor; + if (!mapping.expression.trimmed().isEmpty()) { + ExprParser parser(mapping.expression, regsForExpr, v); + double exprV = 0.0; + if (parser.eval(&exprV)) { + v = exprV; + } + } + + DOMMessage msg; + msg.messageId = QStringLiteral("%1.%2") + .arg(ctx.stableKey.isEmpty() ? QString::number(ctx.deviceId) : ctx.stableKey, + mapping.pointId); + msg.metaRef = meta; + msg.timestamp = ctx.timestamp.toMSecsSinceEpoch(); + msg.value = castToSemantic(v, meta->semanticType); + msg.quality = qualityByRange(v, *meta); + ctx.domMessages.append(std::move(msg)); + } + return true; +} + +} // namespace softbus::message_bus::pipeline::stages::filters diff --git a/src/message_bus/pipeline/stages/3_filters/ProtocolParseFilter.h b/src/message_bus/pipeline/stages/3_filters/ProtocolParseFilter.h new file mode 100644 index 0000000..7b15a89 --- /dev/null +++ b/src/message_bus/pipeline/stages/3_filters/ProtocolParseFilter.h @@ -0,0 +1,30 @@ +#pragma once + +#include + +#include "core/metadata/MetadataRegistry.h" +#include "core/metadata/ProfileRegistry.h" +#include "message_bus/pipeline/stages/3_filters/IFilterStage.h" + +namespace softbus::message_bus::pipeline::stages::filters +{ + +class ProtocolParseFilter final : public IFilterStage +{ +public: + ProtocolParseFilter(); + + bool apply(softbus::message_bus::pipeline::PipelineContext& ctx) override; + +private: + bool ensureLoaded(); + bool parseModbus(softbus::message_bus::pipeline::PipelineContext& ctx); + +private: + softbus::core::metadata::MetadataRegistry m_metadataRegistry; + softbus::core::metadata::ProfileRegistry m_profileRegistry; + std::once_flag m_loadOnce; + bool m_loaded{false}; +}; + +} // namespace softbus::message_bus::pipeline::stages::filters diff --git a/tests/modbus_rtu_framer_test.cpp b/tests/modbus_rtu_framer_test.cpp new file mode 100644 index 0000000..619a901 --- /dev/null +++ b/tests/modbus_rtu_framer_test.cpp @@ -0,0 +1,144 @@ +#include "message_bus/pipeline/stages/1_framers/ModbusRtuFramer.h" +#include "message_bus/pipeline/stages/1_framers/ModbusRtuFraming.h" + +#include + +#include +#include + +using softbus::core::memory::MemoryPool; +using softbus::message_bus::pipeline::FrameKind; +using softbus::message_bus::pipeline::PipelineContext; +using softbus::message_bus::pipeline::stages::framers::ModbusRtuFramer; +using softbus::message_bus::pipeline::stages::framers::ModbusRtuFramerConfig; +using softbus::message_bus::pipeline::stages::framers::modbusRtuCrc16; + +namespace +{ + +void appendCrcLe(std::vector& v) +{ + const std::uint16_t c = modbusRtuCrc16(v.data(), v.size()); + v.push_back(static_cast(c & 0xFF)); + v.push_back(static_cast(c >> 8)); +} + +MemoryPool::BlockPtr makeBlock(MemoryPool& pool, const std::vector& bytes) +{ + auto blk = pool.allocate(bytes.size()); + if (!blk || !blk->data) { + QTest::qFail("pool.allocate failed", __FILE__, __LINE__); + return {}; + } + std::memcpy(blk->data, bytes.data(), bytes.size()); + blk->size = bytes.size(); + return blk; +} + +} // namespace + +class ModbusRtuFramerTest : public QObject +{ + Q_OBJECT + +private slots: + void half_then_rest_emits_one_frame(); + void sticky_two_frames_in_one_chunk(); + void crc_error_triggers_resync_and_recovery(); +}; + +void ModbusRtuFramerTest::half_then_rest_emits_one_frame() +{ + auto pool = std::make_shared(512, 64); + ModbusRtuFramer framer(pool, ModbusRtuFramerConfig{}, QStringLiteral("/ttyUSB0")); + std::vector a{0x01, 0x03, 0x04, 0x12, 0x34, 0x56, 0x78}; + appendCrcLe(a); + + std::vector part1(a.begin(), a.begin() + 4); + std::vector part2(a.begin() + 4, a.end()); + + PipelineContext c1; + c1.endpoint = QStringLiteral("/ttyUSB0"); + c1.protocolHint = QStringLiteral("modbus_rtu"); + c1.frameKind = FrameKind::StreamChunk; + c1.payload.block = makeBlock(*pool, part1); + c1.payload.offset = 0; + c1.payload.length = part1.size(); + c1.payloadSize = part1.size(); + + std::vector out; + framer.feed(c1, out); + QCOMPARE(int(out.size()), 0); + + PipelineContext c2; + c2.endpoint = QStringLiteral("/ttyUSB0"); + c2.protocolHint = QStringLiteral("modbus_rtu"); + c2.frameKind = FrameKind::StreamChunk; + c2.payload.block = makeBlock(*pool, part2); + c2.payload.offset = 0; + c2.payload.length = part2.size(); + c2.payloadSize = part2.size(); + framer.feed(c2, out); + QCOMPARE(int(out.size()), 1); + QCOMPARE(out[0].frameKind, FrameKind::CompleteFrame); + QCOMPARE(out[0].payloadSize, a.size()); +} + +void ModbusRtuFramerTest::sticky_two_frames_in_one_chunk() +{ + auto pool = std::make_shared(512, 64); + ModbusRtuFramer framer(pool, ModbusRtuFramerConfig{}, QStringLiteral("/ttyUSB0")); + std::vector f; + f.insert(f.end(), {0x01, 0x03, 0x02, 0x00, 0x01}); + appendCrcLe(f); + std::vector g; + g.insert(g.end(), {0x02, 0x03, 0x04, 0xAA, 0xBB, 0xCC, 0xDD}); + appendCrcLe(g); + std::vector both; + both.insert(both.end(), f.begin(), f.end()); + both.insert(both.end(), g.begin(), g.end()); + + PipelineContext c; + c.endpoint = QStringLiteral("/ttyUSB0"); + c.protocolHint = QStringLiteral("modbus_rtu"); + c.frameKind = FrameKind::StreamChunk; + c.payload.block = makeBlock(*pool, both); + c.payload.offset = 0; + c.payload.length = both.size(); + c.payloadSize = both.size(); + + std::vector out; + framer.feed(c, out); + QCOMPARE(int(out.size()), 2); +} + +void ModbusRtuFramerTest::crc_error_triggers_resync_and_recovery() +{ + auto pool = std::make_shared(512, 64); + ModbusRtuFramer framer(pool, ModbusRtuFramerConfig{}, QStringLiteral("/ttyUSB0")); + std::vector bad{0x01, 0x03, 0x02, 0x00, 0x00, 0xDE, 0xAD}; + std::vector good; + good.insert(good.end(), {0x01, 0x03, 0x02, 0x12, 0x34}); + appendCrcLe(good); + + std::vector stream; + stream.insert(stream.end(), bad.begin(), bad.end()); + stream.insert(stream.end(), good.begin(), good.end()); + + PipelineContext c; + c.endpoint = QStringLiteral("/ttyUSB0"); + c.protocolHint = QStringLiteral("modbus_rtu"); + c.frameKind = FrameKind::StreamChunk; + c.payload.block = makeBlock(*pool, stream); + c.payload.offset = 0; + c.payload.length = stream.size(); + c.payloadSize = stream.size(); + + std::vector out; + framer.feed(c, out); + QVERIFY(!out.empty()); + QCOMPARE(out[0].payloadSize, good.size()); +} + +QTEST_MAIN(ModbusRtuFramerTest) +#include "modbus_rtu_framer_test.moc" diff --git a/tests/modbus_rtu_pdu_codec_test.cpp b/tests/modbus_rtu_pdu_codec_test.cpp new file mode 100644 index 0000000..fe7b0bb --- /dev/null +++ b/tests/modbus_rtu_pdu_codec_test.cpp @@ -0,0 +1,206 @@ +#include "message_bus/pipeline/protocol/modbusrtu/ModbusRtuCodec.h" + +#include + +#include + +#include "message_bus/pipeline/PipelineContext.h" + +using softbus::message_bus::pipeline::BusDirection; +using softbus::message_bus::pipeline::protocol::ModbusExceptionPdu; +using softbus::message_bus::pipeline::protocol::ModbusRtuPdu; +using softbus::message_bus::pipeline::protocol::ReadCoilsRequest; +using softbus::message_bus::pipeline::protocol::ReadCoilsResponse; +using softbus::message_bus::pipeline::protocol::ReadDiscreteInputsRequest; +using softbus::message_bus::pipeline::protocol::ReadDiscreteInputsResponse; +using softbus::message_bus::pipeline::protocol::ReadHoldingRegistersRequest; +using softbus::message_bus::pipeline::protocol::ReadHoldingRegistersResponse; +using softbus::message_bus::pipeline::protocol::decodeModbusRtu; +using softbus::message_bus::pipeline::protocol::encodeModbusRtu; + +class ModbusRtuPduCodecTest : public QObject +{ + Q_OBJECT + +private slots: + void roundTrip_readCoilsRequest(); + void roundTrip_readCoilsResponse(); + void roundTrip_readDiscreteInputsRequest(); + void roundTrip_readDiscreteInputsResponse(); + void roundTrip_readHoldingRegistersRequest(); + void roundTrip_readHoldingRegistersResponse(); + void roundTrip_exception(); + void unknownDirection_heuristic_request_vs_response(); +}; + +void ModbusRtuPduCodecTest::roundTrip_readHoldingRegistersRequest() +{ + const ReadHoldingRegistersRequest req{0x01, 40001, 4}; + QByteArray adu; + QVERIFY(encodeModbusRtu(ModbusRtuPdu{req}, adu, nullptr)); + QCOMPARE(adu.size(), 8); + + ModbusRtuPdu out; + QVERIFY(decodeModbusRtu(BusDirection::Downstream, + reinterpret_cast(adu.constData()), + static_cast(adu.size()), + out, + nullptr)); + const auto* back = std::get_if(&out); + QVERIFY(back); + QCOMPARE(static_cast(back->unitId), 1); + QCOMPARE(static_cast(back->startAddress), 40001); + QCOMPARE(static_cast(back->quantity), 4); +} + +void ModbusRtuPduCodecTest::roundTrip_readCoilsRequest() +{ + const ReadCoilsRequest req{0x01, 0x0010, 8}; + QByteArray adu; + QVERIFY(encodeModbusRtu(ModbusRtuPdu{req}, adu, nullptr)); + QCOMPARE(adu.size(), 8); + + ModbusRtuPdu out; + QVERIFY(decodeModbusRtu(BusDirection::Downstream, + reinterpret_cast(adu.constData()), + static_cast(adu.size()), + out, + nullptr)); + const auto* back = std::get_if(&out); + QVERIFY(back); + QCOMPARE(static_cast(back->startAddress), 0x10); + QCOMPARE(static_cast(back->quantity), 8); +} + +void ModbusRtuPduCodecTest::roundTrip_readCoilsResponse() +{ + ReadCoilsResponse resp; + resp.unitId = 0x01; + resp.coils = QVector{0x55, 0x01}; + QByteArray adu; + QVERIFY(encodeModbusRtu(ModbusRtuPdu{resp}, adu, nullptr)); + + ModbusRtuPdu out; + QVERIFY(decodeModbusRtu(BusDirection::Upstream, + reinterpret_cast(adu.constData()), + static_cast(adu.size()), + out, + nullptr)); + const auto* back = std::get_if(&out); + QVERIFY(back); + QCOMPARE(back->coils.size(), 2); + QCOMPARE(static_cast(back->coils[0]), 0x55); +} + +void ModbusRtuPduCodecTest::roundTrip_readDiscreteInputsRequest() +{ + const ReadDiscreteInputsRequest req{0x01, 0x0020, 16}; + QByteArray adu; + QVERIFY(encodeModbusRtu(ModbusRtuPdu{req}, adu, nullptr)); + + ModbusRtuPdu out; + QVERIFY(decodeModbusRtu(BusDirection::Downstream, + reinterpret_cast(adu.constData()), + static_cast(adu.size()), + out, + nullptr)); + const auto* back = std::get_if(&out); + QVERIFY(back); + QCOMPARE(static_cast(back->startAddress), 0x20); + QCOMPARE(static_cast(back->quantity), 16); +} + +void ModbusRtuPduCodecTest::roundTrip_readDiscreteInputsResponse() +{ + ReadDiscreteInputsResponse resp; + resp.unitId = 0x01; + resp.discreteInputs = QVector{0x0F}; + QByteArray adu; + QVERIFY(encodeModbusRtu(ModbusRtuPdu{resp}, adu, nullptr)); + + ModbusRtuPdu out; + QVERIFY(decodeModbusRtu(BusDirection::Upstream, + reinterpret_cast(adu.constData()), + static_cast(adu.size()), + out, + nullptr)); + const auto* back = std::get_if(&out); + QVERIFY(back); + QCOMPARE(back->discreteInputs.size(), 1); + QCOMPARE(static_cast(back->discreteInputs[0]), 0x0F); +} + +void ModbusRtuPduCodecTest::roundTrip_readHoldingRegistersResponse() +{ + ReadHoldingRegistersResponse resp; + resp.unitId = 0x02; + resp.registers = QVector{0x00FF, 0x1388}; + + QByteArray adu; + QVERIFY(encodeModbusRtu(ModbusRtuPdu{resp}, adu, nullptr)); + + ModbusRtuPdu out; + QVERIFY(decodeModbusRtu(BusDirection::Upstream, + reinterpret_cast(adu.constData()), + static_cast(adu.size()), + out, + nullptr)); + const auto* back = std::get_if(&out); + QVERIFY(back); + QCOMPARE(static_cast(back->unitId), 2); + QCOMPARE(back->registers.size(), 2); + QCOMPARE(static_cast(back->registers[0]), 255); + QCOMPARE(static_cast(back->registers[1]), 5000); +} + +void ModbusRtuPduCodecTest::roundTrip_exception() +{ + const ModbusExceptionPdu ex{0x05, static_cast(0x83), 0x02}; + QByteArray adu; + QVERIFY(encodeModbusRtu(ModbusRtuPdu{ex}, adu, nullptr)); + QCOMPARE(adu.size(), 5); + + ModbusRtuPdu out; + QVERIFY(decodeModbusRtu(BusDirection::Upstream, + reinterpret_cast(adu.constData()), + static_cast(adu.size()), + out, + nullptr)); + const auto* back = std::get_if(&out); + QVERIFY(back); + QCOMPARE(static_cast(back->unitId), 5); + QCOMPARE(static_cast(back->function), 0x83); + QCOMPARE(static_cast(back->exceptionCode), 2); +} + +void ModbusRtuPduCodecTest::unknownDirection_heuristic_request_vs_response() +{ + ReadHoldingRegistersRequest req{0x01, 0, 1}; + QByteArray aduReq; + QVERIFY(encodeModbusRtu(ModbusRtuPdu{req}, aduReq, nullptr)); + + ModbusRtuPdu asReq; + QVERIFY(decodeModbusRtu(BusDirection::Unknown, + reinterpret_cast(aduReq.constData()), + static_cast(aduReq.size()), + asReq, + nullptr)); + QVERIFY(std::holds_alternative(asReq)); + + ReadHoldingRegistersResponse resp; + resp.unitId = 1; + resp.registers = {0x00AA}; + QByteArray aduResp; + QVERIFY(encodeModbusRtu(ModbusRtuPdu{resp}, aduResp, nullptr)); + + ModbusRtuPdu asResp; + QVERIFY(decodeModbusRtu(BusDirection::Unknown, + reinterpret_cast(aduResp.constData()), + static_cast(aduResp.size()), + asResp, + nullptr)); + QVERIFY(std::holds_alternative(asResp)); +} + +QTEST_MAIN(ModbusRtuPduCodecTest) +#include "modbus_rtu_pdu_codec_test.moc" diff --git a/tests/protocol_parse_filter_test.cpp b/tests/protocol_parse_filter_test.cpp new file mode 100644 index 0000000..26f0124 --- /dev/null +++ b/tests/protocol_parse_filter_test.cpp @@ -0,0 +1,90 @@ +#include "message_bus/pipeline/stages/3_filters/ProtocolParseFilter.h" + +#include + +#include +#include +#include + +#include "message_bus/pipeline/protocol/modbusrtu/ModbusRtuPdu.h" +#include "message_bus/pipeline/protocol/ProtocolEnvelope.h" + +using softbus::message_bus::pipeline::BusDirection; +using softbus::message_bus::pipeline::PipelineContext; +using softbus::message_bus::pipeline::protocol::ModbusRtuPdu; +using softbus::message_bus::pipeline::protocol::ProtocolFamily; +using softbus::message_bus::pipeline::protocol::ReadHoldingRegistersResponse; +using softbus::message_bus::pipeline::stages::filters::ProtocolParseFilter; + +class ProtocolParseFilterTest : public QObject +{ + Q_OBJECT + +private slots: + void initTestCase(); + void parses_multiple_dom_messages(); + +private: + QString m_cfgDir; +}; + +void ProtocolParseFilterTest::initTestCase() +{ + m_cfgDir = QDir::tempPath() + QStringLiteral("/softbus_test_cfg"); + QDir().mkpath(m_cfgDir); + + QFile mf(m_cfgDir + QStringLiteral("/metadata_dictionary.json")); + QVERIFY(mf.open(QIODevice::WriteOnly | QIODevice::Truncate)); + mf.write(R"({ + "metadatas":[ + {"metadataId":"MD_TEMP_001","name":"Temp","description":"t","domain":"PROCESS_PARAM","semanticType":"FLOAT64","unit":"C","minValue":-40,"maxValue":180,"deadband":0.1,"isWriteable":false}, + {"metadataId":"MD_PRESS_001","name":"Press","description":"p","domain":"PROCESS_PARAM","semanticType":"FLOAT64","unit":"MPa","minValue":0,"maxValue":25,"deadband":0.01,"isWriteable":false} + ] + })"); + mf.close(); + + QFile pf(m_cfgDir + QStringLiteral("/device_profiles.json")); + QVERIFY(pf.open(QIODevice::WriteOnly | QIODevice::Truncate)); + pf.write(R"({ + "profiles":[ + {"deviceId":"devA","protocol":"modbus_rtu","mappings":[ + {"pointId":"temperature","metadataId":"MD_TEMP_001","byteOffset":0,"byteLength":4,"rawType":"INT32","endianness":"BIG_ENDIAN","scaleFactor":1.0,"offsetFactor":0.0,"expression":"x*0.1"}, + {"pointId":"pressure","metadataId":"MD_PRESS_001","byteOffset":4,"byteLength":4,"rawType":"INT32","endianness":"BIG_ENDIAN","scaleFactor":1.0,"offsetFactor":0.0,"expression":"x*0.001"} + ]} + ] + })"); + pf.close(); + + qputenv("SOFTBUS_CONFIG_DIR", m_cfgDir.toUtf8()); +} + +void ProtocolParseFilterTest::parses_multiple_dom_messages() +{ + ProtocolParseFilter filter; + PipelineContext ctx; + ctx.direction = BusDirection::Upstream; + ctx.protocolHint = QStringLiteral("modbus_rtu"); + ctx.stableKey = QStringLiteral("devA"); + ctx.deviceId = 1; + ctx.timestamp = QDateTime::currentDateTimeUtc(); + + ReadHoldingRegistersResponse resp; + resp.unitId = 1; + resp.registers = QVector{0, 255, 0, 5000}; + ctx.protocol.family = ProtocolFamily::ModbusRtu; + ctx.protocol.pdu = ModbusRtuPdu{std::move(resp)}; + + QVERIFY(filter.apply(ctx)); + QCOMPARE(ctx.domMessages.size(), 2); + + QCOMPARE(ctx.domMessages[0].messageId, QStringLiteral("devA.temperature")); + QCOMPARE(ctx.domMessages[0].quality, softbus::core::models::DataQuality::GOOD); + QCOMPARE(ctx.domMessages[0].domain(), softbus::core::models::DataDomain::PROCESS_PARAM); + QCOMPARE(ctx.domMessages[0].getValueAs(), 25.5); + + QCOMPARE(ctx.domMessages[1].messageId, QStringLiteral("devA.pressure")); + QCOMPARE(ctx.domMessages[1].getValueAs(), 5.0); +} + +QTEST_MAIN(ProtocolParseFilterTest) +#include "protocol_parse_filter_test.moc"