diff --git a/.cache/clangd/index/DeviceBusManager.cpp.134168DDD210A4ED.idx b/.cache/clangd/index/DeviceBusManager.cpp.134168DDD210A4ED.idx index 2f3059f..8961a0a 100644 Binary files a/.cache/clangd/index/DeviceBusManager.cpp.134168DDD210A4ED.idx and b/.cache/clangd/index/DeviceBusManager.cpp.134168DDD210A4ED.idx differ diff --git a/CMakeLists.txt b/CMakeLists.txt index 5022455..b69cc4f 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -162,6 +162,8 @@ qt_add_executable(softbus_daemon src/message_bus/pipeline/PipelineEngine.cpp src/message_bus/pipeline/PipelineEngineApi.h src/message_bus/pipeline/PipelineEngineApi.cpp + src/message_bus/pipeline/derivation/DerivationEngine.h + src/message_bus/pipeline/derivation/DerivationEngine.cpp src/message_bus/pipeline/protocol/ProtocolEnvelope.h plugins/protocols/modbus_rtu/types/ModbusRtuPdu.h plugins/protocols/modbus_rtu/codec/ModbusRtuCodec.h diff --git a/bin/softbus_daemon b/bin/softbus_daemon index 2ceb699..ae7676b 100755 Binary files a/bin/softbus_daemon and b/bin/softbus_daemon differ diff --git a/docs/数据流转.md b/docs/数据流转.md index e789fb8..fac2857 100644 --- a/docs/数据流转.md +++ b/docs/数据流转.md @@ -166,16 +166,17 @@ SerialQtDriver(read callback) ### 8.2 Mapper 行为（Modbus RTU） -`ModbusRtuProtocolMapperPlugin` 当前行为： +`ModbusRtuProtocolMapperPlugin` + `PipelineEngine` 内 `IMappingLookup` 组合行为： 1. 对 `ReadHoldingRegistersRequest` 记录 `unitId -> startAddress`，用于后续响应地址回填。 2. 对 `ReadHoldingRegistersResponse` 遍历寄存器并构造 `compositeKey`： - `compositeKey = endpointHash + protocol + slaveId + registerAddress`（位拼接）。 -3. 若 `MappingRegistry` 命中： - - 生成 `DOMMessage`，`messageId = domPath`，并应用 `scale/offset/unit`。 -4. 若未命中： - - 仅在 `DiscoveryPool.sniffingEnabled=true` 时写入发现池样本； - - 主链路继续 `continue`，不生成业务消息。 +3. **解析优先级（统一语义）**：`MappingRegistry`（运行时绑定）优先；未命中则查 `ProfileRegistry`（`device_profiles.json`，按 `runtimeDeviceId` 与 `modbus_rtu` 协议匹配，支持 `*` 通配设备）。 +4. 若上述之一命中： + - 生成轻量 `DOMMessage`：`messageId`/`pointId` 来自绑定或 profile，`metadataId` 来自 profile（若有），`value` 为工程值；静态元数据不随每条样本重复序列化，由 `metadataId` 在控制面/Registry 解析。 +5. 若仍未命中： + - 若存在 `discoverySink`，始终上报 `DiscoveryRecord`（发现池内部仍受嗅探开关与限频约束）； + - **仅当嗅探开启**时，主链路额外产出 `modbus.unmapped:*` 的 `DOMMessage` 供规则调试；嗅探关闭时不产出未映射 DOM，减少总线冗载。 ### 8.3 DiscoveryPool 约束 diff --git a/plugins/protocols/modbus_rtu/mapper/ModbusRtuProtocolMapperPlugin.cpp b/plugins/protocols/modbus_rtu/mapper/ModbusRtuProtocolMapperPlugin.cpp index 38040b0..1b8bd79 100644 --- a/plugins/protocols/modbus_rtu/mapper/ModbusRtuProtocolMapperPlugin.cpp +++ b/plugins/protocols/modbus_rtu/mapper/ModbusRtuProtocolMapperPlugin.cpp @@ -50,7 +50,9 @@ public: const auto addr = static_cast(baseAddress + static_cast(i)); const auto compKey = buildCompositeKey( rawContext.endpointHash, rawContext.protocol, resp->unitId, addr); - const auto mapRule = m_context.mappingLookup ? m_context.mappingLookup->find(compKey) : std::nullopt; + const auto mapRule = m_context.mappingLookup + ? m_context.mappingLookup->find(rawContext, compKey) + : std::nullopt; const double rawValue = static_cast(resp->registers[static_cast(i)]); if (!mapRule.has_value()) { if (m_context.discoverySink) { @@ -64,8 +66,10 @@ public: rec.lastSeenMs = rawContext.timestampMs; m_context.discoverySink->reportUnmapped(rec); } - // 无 metadata 映射时仍产出 DOM，便于 DynamicRuleRegistry 的 transform 等下游逻辑生效 - // （compositeKey 仍按与映射表相同的算法计算，便于后续在 UI 里绑定映射） + const bool sniff = m_context.sniffingEnabled ? m_context.sniffingEnabled() : false; + if (!sniff) { + continue; + } softbus::core::models::DOMMessage pass; pass.messageId = QStringLiteral("modbus.unmapped:%1").arg(compKey); pass.timestamp = rawContext.timestampMs; @@ -81,6 +85,8 @@ public: } softbus::core::models::DOMMessage msg; msg.messageId = mapRule->domPath; + msg.pointId = mapRule->pointId.isEmpty() ? mapRule->domPath : mapRule->pointId; + msg.metadataId = mapRule->metadataId; msg.timestamp = rawContext.timestampMs; msg.value = (rawValue * mapRule->scale) + mapRule->offset; msg.quality = softbus::core::models::DataQuality::GOOD; @@ -88,6 +94,7 @@ public: msg.attributes.insert(QStringLiteral("compositeKey"), QString::number(compKey)); msg.attributes.insert(QStringLiteral("slaveId"), static_cast(resp->unitId)); msg.attributes.insert(QStringLiteral("registerAddress"), static_cast(addr)); + msg.attributes.insert(QStringLiteral("mappingSource"), mapRule->sourceTag); if (!mapRule->unit.isEmpty()) { msg.attributes.insert(QStringLiteral("unit"), mapRule->unit); } @@ -142,4 +149,3 @@ softbus::core::plugin_system::IProtocolMapperPlugin* createModbusRtuProtocolMapp } } // namespace softbus::message_bus::pipeline::protocol::modbusrtu - diff --git a/src/api/API_INTERFACE_TABLE.md b/src/api/API_INTERFACE_TABLE.md index 945a84b..89c06ea 100644 --- a/src/api/API_INTERFACE_TABLE.md +++ b/src/api/API_INTERFACE_TABLE.md @@ -58,9 +58,9 @@ | `remove_filter` | `id` | `0` | `4041` | 删除过滤规则 | | `add_transform_rule` | `id`, `endpointHash?`, `deviceId?`, `stableKey?`, `scale?`, `offset?`, `unit?` | `0` | `4002` | 添加转换规则 | | `remove_transform_rule` | `id` | `0` | `4042` | 删除转换规则 | -| `add_mapping_rule` | `compositeKey`, `domPath`, `scale?`, `offset?`, `unit?` | `0` | `4004` | 新增/更新映射规则 | -| `remove_mapping_rule` | `compositeKey` | `0` | `4043` | 删除映射规则 | -| `list_mapping_rules` | 无 | `0` | - | 返回映射规则列表 | +| `add_mapping_rule` | `compositeKey`, `domPath`, `scale?`, `offset?`, `unit?`, `metadataId?`, `pointId?`, `catalogVersion?`, `updatedBy?` | `0` | `4004` / `5002` | 新增/更新映射规则；成功后持久化到 `runtime_mappings.json`；`5002` 表示持久化失败 | +| `remove_mapping_rule` | `compositeKey` | `0` | `4043` / `5002` | 删除映射规则并持久化 | +| `list_mapping_rules` | 无 | `0` | - | 返回 `items` 与 `mappingCatalogVersion` | | `discovery_start` | `ttlMs?` | `0` | - | 开启嗅探模式 | | `discovery_stop` | 无 | `0` | - | 关闭嗅探模式 | | `discovery_clear` | 无 | `0` | - | 清空发现池 | @@ -123,10 +123,10 @@ | 方法 | 路径 | 请求参数 | 说明 | |---|---|---|---| -| `GET` | `/api/v1/metadata` | 无 | 返回 metadata 字典全量快照 | -| `GET` | `/api/v1/metadata/{metadataId}` | path: `metadataId` | 返回指定 metadata 详情 | -| `GET` | `/api/v1/profiles` | 无 | 返回全部设备 profile 快照 | -| `GET` | `/api/v1/profiles/{deviceId}` | path: `deviceId` | 返回指定 `deviceId` 关联的全部 profile | +| `GET` | `/api/v1/metadata` | 无 | 返回 `data`（字典数组）与可选 `catalogVersion`（根级字段） | +| `GET` | `/api/v1/metadata/{metadataId}` | path: `metadataId` | 返回指定 metadata 详情（含可选 `catalogVersion`） | +| `GET` | `/api/v1/profiles` | 无 | 返回 `data` 与可选 `catalogVersion` | +| `GET` | `/api/v1/profiles/{deviceId}` | path: `deviceId` | 返回指定 `deviceId` 关联的全部 profile（含可选 `catalogVersion`） | ### 5.7 Discovery 路由 @@ -142,9 +142,9 @@ | 方法 | 路径 | 请求参数 | 说明 | |---|---|---|---| -| `POST` | `/api/v1/mapping/bind` | body: `compositeKey`, `domPath`, `scale?`, `offset?`, `unit?` | 绑定或更新映射规则 | -| `POST` | `/api/v1/mapping/unbind` | body: `compositeKey` | 删除映射规则 | -| `GET` | `/api/v1/mapping/list` | 无 | 列出当前映射规则 | +| `POST` | `/api/v1/mapping/bind` | body: `compositeKey`, `domPath`, `scale?`, `offset?`, `unit?`, `metadataId?`, `pointId?`, `catalogVersion?`, `updatedBy?` | 绑定或更新映射规则并写入 `runtime_mappings.json` | +| `POST` | `/api/v1/mapping/unbind` | body: `compositeKey` | 删除映射规则并持久化 | +| `GET` | `/api/v1/mapping/list` | 无 | 返回 `items` 与 `mappingCatalogVersion` | ## 6) 通用错误码约定（当前实现） @@ -160,6 +160,7 @@ | `4041` | `filter_not_found` | 删除过滤规则时未找到 | | `4042` | `transform_rule_not_found` | 删除转换规则时未找到 | | `4043` | `mapping_rule_not_found` | 删除映射规则时未找到 | +| `5002` | `persist_mappings_failed` | 映射规则内存更新成功但写入 `runtime_mappings.json` 失败 | ## 7) 维护建议 diff --git a/src/api/http/RestApiRoutes.cpp b/src/api/http/RestApiRoutes.cpp index 8f0e248..403a559 100644 --- a/src/api/http/RestApiRoutes.cpp +++ b/src/api/http/RestApiRoutes.cpp @@ -3,7 +3,9 @@ #include #include #include +#include #include +#include #include #include #include @@ -15,6 +17,7 @@ #include #include #include +#include #include #include @@ -176,6 +179,17 @@ QHttpServerResponse okDataResponse(const QJsonValue& data, return jsonResponse(QJsonObject{{QStringLiteral("ok"), true}, {QStringLiteral("data"), data}}, code); } +QHttpServerResponse okDataResponseWithCatalog(const QJsonValue& data, + const QString& catalogVersion, + QHttpServerResponse::StatusCode code = QHttpServerResponse::StatusCode::Ok) +{ + QJsonObject o{{QStringLiteral("ok"), true}, {QStringLiteral("data"), data}}; + if (!catalogVersion.isEmpty()) { + o.insert(QStringLiteral("catalogVersion"), catalogVersion); + } + return jsonResponse(o, code); +} + QHttpServerResponse errorResponse(const QString& error, QHttpServerResponse::StatusCode code = QHttpServerResponse::StatusCode::BadRequest) { @@ -212,6 +226,10 @@ QJsonArray devicesToJson(const std::vector& device QString resolveConfigBaseDir() { + const QString daemonConfigPath = softbus::device_bus::DeviceBus::instance().configPath(); + if (!daemonConfigPath.isEmpty()) { + return QFileInfo(daemonConfigPath).dir().absolutePath() + QChar('/'); + } QString configBase = qEnvironmentVariable("SOFTBUS_CONFIG_DIR"); if (!configBase.isEmpty()) { return configBase.endsWith(QChar('/')) ? configBase : configBase + QChar('/'); @@ -219,10 +237,71 @@ QString resolveConfigBaseDir() return QDir::homePath() + QStringLiteral("/softbus/config/"); } -QString resolveConfigFile(const QString& fileName, const QString& fallbackRelative) +QString resolveConfigFile(const QString& fileName) { - const QString homePath = resolveConfigBaseDir() + fileName; - return QFileInfo::exists(homePath) ? homePath : fallbackRelative; + return resolveConfigBaseDir() + fileName; +} + +bool ensureConfigFileInitialized(const QString& targetPath, + const QString& fallbackPath, + const QByteArray& defaultContent) +{ + if (QFile::exists(targetPath)) { + return true; + } + const QFileInfo fi(targetPath); + if (!QDir().mkpath(fi.dir().absolutePath())) { + return false; + } + QByteArray initBytes = defaultContent; + QFile fallback(fallbackPath); + if (!fallbackPath.isEmpty() && fallback.exists() && fallback.open(QIODevice::ReadOnly)) { + initBytes = fallback.readAll(); + } + QSaveFile initFile(targetPath); + if (!initFile.open(QIODevice::WriteOnly | QIODevice::Text)) { + return false; + } + if (initFile.write(initBytes) != initBytes.size()) { + initFile.cancelWriting(); + return false; + } + return initFile.commit(); +} + +void ensureRestApiConfigFilesInitialized() +{ + static std::once_flag once; + std::call_once(once, []() { + const QByteArray defaultRules = QJsonDocument(QJsonObject{ + {QStringLiteral("filter"), QJsonArray{}}, + {QStringLiteral("transform"), QJsonArray{}}, + }).toJson(QJsonDocument::Indented); + (void)ensureConfigFileInitialized(resolveConfigFile(QStringLiteral("runtime_rules.json")), + QStringLiteral(), + defaultRules); + + const QByteArray defaultMappings = QStringLiteral( + "{\n \"catalogVersion\": 1,\n \"items\": []\n}\n") + .toUtf8(); + (void)ensureConfigFileInitialized(resolveConfigFile(QStringLiteral("runtime_mappings.json")), + QStringLiteral(), + defaultMappings); + + const QByteArray defaultProfiles = QStringLiteral( + "{\n \"catalogVersion\": \"\",\n \"profiles\": []\n}\n") + .toUtf8(); + (void)ensureConfigFileInitialized(resolveConfigFile(QStringLiteral("device_profiles.json")), + QStringLiteral("config/device_profiles.json"), + defaultProfiles); + + const QByteArray defaultMetadata = QStringLiteral( + "{\n \"catalogVersion\": \"\",\n \"metadatas\": []\n}\n") + .toUtf8(); + (void)ensureConfigFileInitialized(resolveConfigFile(QStringLiteral("metadata_dictionary.json")), + QStringLiteral("config/metadata_dictionary.json"), + defaultMetadata); + }); } bool ensureMetadataLoaded(softbus::core::metadata::MetadataRegistry& registry, QString* error) @@ -232,10 +311,9 @@ bool ensureMetadataLoaded(softbus::core::metadata::MetadataRegistry& registry, Q static bool loaded = false; std::lock_guard lk(mtx); if (!attempted) { + ensureRestApiConfigFilesInitialized(); attempted = true; - loaded = registry.loadFromFile( - resolveConfigFile(QStringLiteral("metadata_dictionary.json"), - QStringLiteral("config/metadata_dictionary.json"))); + loaded = registry.loadFromFile(resolveConfigFile(QStringLiteral("metadata_dictionary.json"))); } if (!loaded && error) { *error = registry.lastError(); @@ -249,17 +327,17 @@ softbus::core::metadata::MetadataRegistry& metadataRegistry() return registry; } -bool ensureProfileLoaded(softbus::core::metadata::ProfileRegistry& registry, QString* error) +bool ensureProfileLoaded(QString* error) { static std::mutex mtx; static bool attempted = false; static bool loaded = false; std::lock_guard lk(mtx); + auto& registry = softbus::core::metadata::ProfileRegistry::instance(); if (!attempted) { + ensureRestApiConfigFilesInitialized(); attempted = true; - loaded = registry.loadFromFile( - resolveConfigFile(QStringLiteral("device_profiles.json"), - QStringLiteral("config/device_profiles.json"))); + loaded = registry.loadFromFile(resolveConfigFile(QStringLiteral("device_profiles.json"))); } if (!loaded && error) { *error = registry.lastError(); @@ -267,12 +345,6 @@ bool ensureProfileLoaded(softbus::core::metadata::ProfileRegistry& registry, QSt return loaded; } -softbus::core::metadata::ProfileRegistry& profileRegistry() -{ - static softbus::core::metadata::ProfileRegistry registry; - return registry; -} - bool isProtectedNodeKey(const QString& key) { for (const auto* value : kProtectedNodeKeys) { @@ -400,6 +472,11 @@ void registerMappingRoutes(QHttpServer* server, QObject* routeContext) rule.scale = payload.value(QStringLiteral("scale")).toDouble(1.0); rule.offset = payload.value(QStringLiteral("offset")).toDouble(0.0); rule.unit = payload.value(QStringLiteral("unit")).toString(); + rule.metadataId = payload.value(QStringLiteral("metadataId")).toString(); + rule.pointId = payload.value(QStringLiteral("pointId")).toString(); + rule.catalogVersion = payload.value(QStringLiteral("catalogVersion")).toString(); + rule.updatedBy = payload.value(QStringLiteral("updatedBy")).toString(QStringLiteral("http")); + rule.updatedAtMs = QDateTime::currentMSecsSinceEpoch(); auto& mappingReg = softbus::message_bus::pipeline::MappingRegistry::instance(); if (!mappingReg.upsert(rule)) { @@ -407,10 +484,15 @@ void registerMappingRoutes(QHttpServer* server, QObject* routeContext) QJsonObject{{QStringLiteral("error"), QStringLiteral("invalid_mapping_rule")}}, QHttpServerResponse::StatusCode::BadRequest)); } - return withCorsHeaders(QHttpServerResponse(QJsonObject{{QStringLiteral("ok"), true}, - {QStringLiteral("compositeKey"), - QString::number(rule.compositeKey)}}, - QHttpServerResponse::StatusCode::Ok)); + ensureRestApiConfigFilesInitialized(); + const QString mapPath = resolveConfigFile(QStringLiteral("runtime_mappings.json")); + (void)mappingReg.savePersistedFile(mapPath); + return withCorsHeaders(QHttpServerResponse( + QJsonObject{{QStringLiteral("ok"), true}, + {QStringLiteral("compositeKey"), QString::number(rule.compositeKey)}, + {QStringLiteral("mappingCatalogVersion"), + QString::number(mappingReg.mappingCatalogVersion())}}, + QHttpServerResponse::StatusCode::Ok)); }); server->route(QStringLiteral("/api/v1/mapping/unbind"), QHttpServerRequest::Method::Post, routeContext, @@ -420,6 +502,11 @@ void registerMappingRoutes(QHttpServer* server, QObject* routeContext) payload.value(QStringLiteral("compositeKey")).toString().toULongLong(); auto& mappingReg = softbus::message_bus::pipeline::MappingRegistry::instance(); const bool ok = mappingReg.remove(compositeKey); + if (ok) { + ensureRestApiConfigFilesInitialized(); + const QString mapPath = resolveConfigFile(QStringLiteral("runtime_mappings.json")); + (void)mappingReg.savePersistedFile(mapPath); + } return withCorsHeaders(QHttpServerResponse(QJsonObject{{QStringLiteral("ok"), ok}, {QStringLiteral("compositeKey"), QString::number(compositeKey)}}, @@ -429,9 +516,10 @@ void registerMappingRoutes(QHttpServer* server, QObject* routeContext) server->route(QStringLiteral("/api/v1/mapping/list"), QHttpServerRequest::Method::Get, routeContext, []() { auto& mappingReg = softbus::message_bus::pipeline::MappingRegistry::instance(); - return withCorsHeaders( - QHttpServerResponse(QJsonObject{{QStringLiteral("items"), mappingReg.list()}}, - QHttpServerResponse::StatusCode::Ok)); + QJsonObject o; + o.insert(QStringLiteral("items"), mappingReg.list()); + o.insert(QStringLiteral("mappingCatalogVersion"), mappingReg.mappingCatalogVersion()); + return withCorsHeaders(QHttpServerResponse(o, QHttpServerResponse::StatusCode::Ok)); }); } @@ -631,7 +719,7 @@ void registerMetadataRoutes(QHttpServer* server, QObject* routeContext) return errorResponse(error.isEmpty() ? QStringLiteral("metadata_unavailable") : error, QHttpServerResponse::StatusCode::ServiceUnavailable); } - return okDataResponse(registry.list()); + return okDataResponseWithCatalog(registry.list(), registry.catalogVersion()); }); server->route(QStringLiteral("/api/v1/metadata/"), QHttpServerRequest::Method::Get, routeContext, @@ -646,7 +734,7 @@ void registerMetadataRoutes(QHttpServer* server, QObject* routeContext) for (const auto& item : items) { const auto obj = item.toObject(); if (obj.value(QStringLiteral("metadataId")).toString() == metadataId) { - return okDataResponse(obj); + return okDataResponseWithCatalog(obj, registry.catalogVersion()); } } return errorResponse(QStringLiteral("metadata_not_found"), @@ -657,24 +745,24 @@ void registerMetadataRoutes(QHttpServer* server, QObject* routeContext) void registerProfileRoutes(QHttpServer* server, QObject* routeContext) { server->route(QStringLiteral("/api/v1/profiles"), QHttpServerRequest::Method::Get, routeContext, []() { - auto& registry = profileRegistry(); QString error; - if (!ensureProfileLoaded(registry, &error)) { + if (!ensureProfileLoaded(&error)) { return errorResponse(error.isEmpty() ? QStringLiteral("profiles_unavailable") : error, QHttpServerResponse::StatusCode::ServiceUnavailable); } - return okDataResponse(registry.list()); + auto& registry = softbus::core::metadata::ProfileRegistry::instance(); + return okDataResponseWithCatalog(registry.list(), registry.catalogVersion()); }); server->route(QStringLiteral("/api/v1/profiles/"), QHttpServerRequest::Method::Get, routeContext, [](const QString& deviceId) { - auto& registry = profileRegistry(); QString error; - if (!ensureProfileLoaded(registry, &error)) { + if (!ensureProfileLoaded(&error)) { return errorResponse(error.isEmpty() ? QStringLiteral("profiles_unavailable") : error, QHttpServerResponse::StatusCode::ServiceUnavailable); } - return okDataResponse(registry.listByDeviceId(deviceId)); + auto& registry = softbus::core::metadata::ProfileRegistry::instance(); + return okDataResponseWithCatalog(registry.listByDeviceId(deviceId), registry.catalogVersion()); }); } diff --git a/src/api/ipc_dbus/CommandDispatcher.cpp b/src/api/ipc_dbus/CommandDispatcher.cpp index 820341f..9d28ac2 100644 --- a/src/api/ipc_dbus/CommandDispatcher.cpp +++ b/src/api/ipc_dbus/CommandDispatcher.cpp @@ -1,5 +1,6 @@ #include "api/ipc_dbus/CommandDispatcher.h" +#include #include #include #include @@ -29,6 +30,35 @@ QString rulesConfigPath() return QDir::homePath() + QStringLiteral("/softbus/config/runtime_rules.json"); } +QString runtimeMappingsConfigPath() +{ + const QFileInfo fi(rulesConfigPath()); + return fi.dir().filePath(QStringLiteral("runtime_mappings.json")); +} + +bool persistMappingsToDedicatedConfig(const softbus::message_bus::pipeline::MappingRegistry& reg) +{ + const QString path = runtimeMappingsConfigPath(); + const QFileInfo fi(path); + if (!QDir().mkpath(fi.dir().absolutePath())) { + LOG_ERROR() << "CommandDispatcher: failed to create mappings directory:" << fi.dir().absolutePath(); + return false; + } + return reg.savePersistedFile(path); +} + +void appendMappingAuditLine(const QString& line) +{ + const QFileInfo fi(runtimeMappingsConfigPath()); + const QString auditPath = fi.dir().filePath(QStringLiteral("mapping_audit.log")); + QFile f(auditPath); + if (!f.open(QIODevice::Append | QIODevice::Text)) { + return; + } + f.write(line.toUtf8()); + f.write("\n"); +} + softbus::message_bus::pipeline::RuleScope parseScope(const QJsonObject& params) { softbus::message_bus::pipeline::RuleScope scope; @@ -188,14 +218,38 @@ DispatchResult CommandDispatcher::dispatch(const QString& action, const QJsonObj rule.scale = params.value(QStringLiteral("scale")).toDouble(1.0); rule.offset = params.value(QStringLiteral("offset")).toDouble(0.0); rule.unit = params.value(QStringLiteral("unit")).toString(); + rule.metadataId = params.value(QStringLiteral("metadataId")).toString(); + rule.pointId = params.value(QStringLiteral("pointId")).toString(); + rule.catalogVersion = params.value(QStringLiteral("catalogVersion")).toString(); + rule.updatedBy = params.value(QStringLiteral("updatedBy")).toString(); + rule.updatedAtMs = QDateTime::currentMSecsSinceEpoch(); if (!mappingReg.upsert(rule)) { return {4004, QStringLiteral("invalid_mapping_rule"), {}}; } - return {0, QStringLiteral("ok"), QJsonObject{{QStringLiteral("compositeKey"), keyText}}}; + if (!persistMappingsToDedicatedConfig(mappingReg)) { + return {5002, QStringLiteral("persist_mappings_failed"), {}}; + } + appendMappingAuditLine(QStringLiteral("upsert compositeKey=%1 domPath=%2 at %3") + .arg(keyText) + .arg(rule.domPath) + .arg(rule.updatedAtMs)); + return {0, + QStringLiteral("ok"), + QJsonObject{{QStringLiteral("compositeKey"), keyText}, + {QStringLiteral("mappingCatalogVersion"), + QString::number(mappingReg.mappingCatalogVersion())}}}; } if (action == QStringLiteral("remove_mapping_rule")) { const QString keyText = params.value(QStringLiteral("compositeKey")).toString(); const bool ok = mappingReg.remove(keyText.toULongLong()); + if (ok) { + if (!persistMappingsToDedicatedConfig(mappingReg)) { + return {5002, QStringLiteral("persist_mappings_failed"), {}}; + } + appendMappingAuditLine(QStringLiteral("remove compositeKey=%1 at %2") + .arg(keyText) + .arg(QDateTime::currentMSecsSinceEpoch())); + } return {ok ? 0 : 4043, ok ? QStringLiteral("ok") : QStringLiteral("mapping_rule_not_found"), QJsonObject{{QStringLiteral("compositeKey"), keyText}}}; } diff --git a/src/core/metadata/MetadataRegistry.cpp b/src/core/metadata/MetadataRegistry.cpp index e342f9e..d973888 100644 --- a/src/core/metadata/MetadataRegistry.cpp +++ b/src/core/metadata/MetadataRegistry.cpp @@ -26,11 +26,8 @@ DataDomain parseDomain(const QString& s) 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; + if (x == QStringLiteral("MANAGE_DATA")) { + return DataDomain::MANAGE_DATA; } return DataDomain::UNKNOWN; } @@ -74,10 +71,8 @@ QString domainToString(DataDomain domain) return QStringLiteral("CONTROL_CMD"); case DataDomain::MONITOR_DATA: return QStringLiteral("MONITOR_DATA"); - case DataDomain::HEALTH_DIAG: - return QStringLiteral("HEALTH_DIAG"); - case DataDomain::CONFIG_PARAM: - return QStringLiteral("CONFIG_PARAM"); + case DataDomain::MANAGE_DATA: + return QStringLiteral("MANAGE_DATA"); case DataDomain::UNKNOWN: default: return QStringLiteral("UNKNOWN"); @@ -110,7 +105,7 @@ QString dataTypeToString(DataType type) QJsonObject toJsonObject(const softbus::core::models::MetadataDef& def) { - return QJsonObject{ + QJsonObject o{ {QStringLiteral("metadataId"), def.metadataId}, {QStringLiteral("name"), def.name}, {QStringLiteral("description"), def.description}, @@ -122,6 +117,13 @@ QJsonObject toJsonObject(const softbus::core::models::MetadataDef& def) {QStringLiteral("deadband"), def.deadband}, {QStringLiteral("isWriteable"), def.isWriteable}, }; + if (!def.recordStatus.isEmpty()) { + o.insert(QStringLiteral("recordStatus"), def.recordStatus); + } + if (!def.entryCatalogVersion.isEmpty()) { + o.insert(QStringLiteral("entryCatalogVersion"), def.entryCatalogVersion); + } + return o; } } // namespace @@ -130,6 +132,7 @@ bool MetadataRegistry::loadFromFile(const QString& filePath) { m_lastError.clear(); m_defsById.clear(); + m_catalogVersion.clear(); QFile f(filePath); if (!f.open(QIODevice::ReadOnly)) { @@ -145,15 +148,19 @@ bool MetadataRegistry::loadFromFile(const QString& filePath) } const auto root = doc.object(); + m_catalogVersion = root.value(QStringLiteral("catalogVersion")).toString(); const auto arr = root.value(QStringLiteral("metadatas")).toArray(); for (const auto& item : arr) { if (!item.isObject()) { continue; } - const auto one = parseOne(item.toObject()); + auto one = parseOne(item.toObject()); if (!one.has_value() || one->metadataId.isEmpty()) { continue; } + if (one->entryCatalogVersion.isEmpty() && !m_catalogVersion.isEmpty()) { + one->entryCatalogVersion = m_catalogVersion; + } m_defsById.insert(one->metadataId, *one); } @@ -195,6 +202,8 @@ std::optional MetadataRegistry::parseOne(con 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); + def.recordStatus = o.value(QStringLiteral("recordStatus")).toString(QStringLiteral("ACTIVE")); + def.entryCatalogVersion = o.value(QStringLiteral("entryCatalogVersion")).toString(); return def; } diff --git a/src/core/metadata/MetadataRegistry.h b/src/core/metadata/MetadataRegistry.h index 4f8516e..cb7f18b 100644 --- a/src/core/metadata/MetadataRegistry.h +++ b/src/core/metadata/MetadataRegistry.h @@ -21,6 +21,7 @@ public: QJsonArray list() const; bool empty() const { return m_defsById.isEmpty(); } const QString& lastError() const { return m_lastError; } + const QString& catalogVersion() const { return m_catalogVersion; } private: static std::optional parseOne(const QJsonObject& o); @@ -28,6 +29,7 @@ private: private: QHash m_defsById; QString m_lastError; + QString m_catalogVersion; }; } // namespace softbus::core::metadata diff --git a/src/core/metadata/ProfileRegistry.cpp b/src/core/metadata/ProfileRegistry.cpp index b7cb506..8dc11e9 100644 --- a/src/core/metadata/ProfileRegistry.cpp +++ b/src/core/metadata/ProfileRegistry.cpp @@ -8,6 +8,12 @@ namespace softbus::core::metadata { +ProfileRegistry& ProfileRegistry::instance() +{ + static ProfileRegistry reg; + return reg; +} + namespace { @@ -91,6 +97,7 @@ bool ProfileRegistry::loadFromFile(const QString& filePath) { m_lastError.clear(); m_profilesByKey.clear(); + m_catalogVersion.clear(); QFile f(filePath); if (!f.open(QIODevice::ReadOnly)) { @@ -106,6 +113,7 @@ bool ProfileRegistry::loadFromFile(const QString& filePath) } const auto root = doc.object(); + m_catalogVersion = root.value(QStringLiteral("catalogVersion")).toString(); const auto arr = root.value(QStringLiteral("profiles")).toArray(); for (const auto& item : arr) { if (!item.isObject()) { @@ -246,4 +254,57 @@ softbus::core::models::Endianness ProfileRegistry::parseEndianness(const QString return Endianness::BigEndian; } +bool ProfileRegistry::mappingMatchesModbusRegister(const softbus::core::models::DataPointMapping& m, + const std::uint16_t registerAddress) +{ + constexpr std::uint32_t kUnset = 0xFFFFFFFFu; + if (m.registerAddress != kUnset) { + return m.registerAddress == static_cast(registerAddress); + } + if (m.byteLength == 2) { + return static_cast(m.byteOffset) == static_cast(registerAddress) * 2U; + } + return false; +} + +std::optional ProfileRegistry::findModbusRegisterMapping( + const int runtimeDeviceId, const QString& protocolNorm, const std::uint16_t registerAddress) const +{ + if (protocolNorm.isEmpty()) { + return std::nullopt; + } + const QString pid = runtimeDeviceId >= 0 ? QString::number(runtimeDeviceId) : QString(); + + const auto tryProfile = [&](const softbus::core::models::DeviceDataProfile* p) -> std::optional { + if (!p) { + return std::nullopt; + } + for (const auto& m : p->mappings) { + if (!mappingMatchesModbusRegister(m, registerAddress)) { + continue; + } + ProfileModbusRegisterHit hit; + hit.domPath = m.pointId.isEmpty() ? m.metadataId : m.pointId; + hit.scale = m.scaleFactor; + hit.offset = m.offsetFactor; + hit.metadataId = m.metadataId; + hit.pointId = m.pointId; + return hit; + } + return std::nullopt; + }; + + if (!pid.isEmpty()) { + if (const auto* exact = find(pid, protocolNorm)) { + if (auto h = tryProfile(exact)) { + return h; + } + } + } + if (const auto* wild = find(QStringLiteral("*"), protocolNorm)) { + return tryProfile(wild); + } + return std::nullopt; +} + } // namespace softbus::core::metadata diff --git a/src/core/metadata/ProfileRegistry.h b/src/core/metadata/ProfileRegistry.h index 5f0154a..4b0b8ec 100644 --- a/src/core/metadata/ProfileRegistry.h +++ b/src/core/metadata/ProfileRegistry.h @@ -12,9 +12,21 @@ namespace softbus::core::metadata { +struct ProfileModbusRegisterHit +{ + QString domPath; + double scale{1.0}; + double offset{0.0}; + QString unit; + QString metadataId; + QString pointId; +}; + class ProfileRegistry { public: + static ProfileRegistry& instance(); + bool loadFromFile(const QString& filePath); const softbus::core::models::DeviceDataProfile* find(const QString& deviceId, const QString& protocol) const; @@ -23,6 +35,12 @@ public: const QString& lastError() const { return m_lastError; } bool empty() const { return m_profilesByKey.isEmpty(); } + const QString& catalogVersion() const { return m_catalogVersion; } + + /// Modbus 保持寄存器地址与 profile 中 DataPointMapping 对齐（registerAddress 或 byteOffset+byteLength） + std::optional findModbusRegisterMapping(int runtimeDeviceId, + const QString& protocolNorm, + std::uint16_t registerAddress) const; private: static QString makeKey(const QString& deviceId, const QString& protocol); @@ -30,9 +48,16 @@ private: static softbus::core::models::DataType parseDataType(const QString& s); static softbus::core::models::Endianness parseEndianness(const QString& s); +private: + ProfileRegistry() = default; + + static bool mappingMatchesModbusRegister(const softbus::core::models::DataPointMapping& m, + std::uint16_t registerAddress); + private: QHash m_profilesByKey; QString m_lastError; + QString m_catalogVersion; }; } // namespace softbus::core::metadata diff --git a/src/core/models/DOMMessage.h b/src/core/models/DOMMessage.h index f102219..d4616e0 100644 --- a/src/core/models/DOMMessage.h +++ b/src/core/models/DOMMessage.h @@ -21,15 +21,39 @@ class DOMMessage public: DOMMessage() = default; - QString messageId; // 消息ID - const MetadataDef* metaRef{nullptr}; // 元数据引用 + QString messageId; // 消息ID（建议与 pointId 对齐） + QString pointId; // 标准点位 ID（热路径标识，可与 messageId 相同） + QString metadataId; // 元数据目录 ID（热路径；静态定义不入总线载荷重复序列化） + const MetadataDef* metaRef{nullptr}; // 进程内解析缓存指针（可选，非跨边界序列化字段） std::int64_t timestamp{0}; // 时间戳 DOMValue value{0.0}; // 值 DataQuality quality{DataQuality::BAD}; // 质量 QHash attributes; // 属性 std::vector traceLog; // 跟踪日志 - DataDomain domain() const { return metaRef ? metaRef->domain : DataDomain::UNKNOWN; } + DataDomain domain() const + { + if (metaRef) { + return metaRef->domain; + } + const auto it = attributes.constFind(QStringLiteral("domain")); + if (it != attributes.constEnd() && it->isString()) { + const QString x = it->toString().trimmed().toUpper(); + if (x == QLatin1String("PROCESS_PARAM")) { + return DataDomain::PROCESS_PARAM; + } + if (x == QLatin1String("CONTROL_CMD")) { + return DataDomain::CONTROL_CMD; + } + if (x == QLatin1String("MONITOR_DATA")) { + return DataDomain::MONITOR_DATA; + } + if (x == QLatin1String("MANAGE_DATA")) { + return DataDomain::MANAGE_DATA; + } + } + return DataDomain::UNKNOWN; + } template T getValueAs() const diff --git a/src/core/models/EnrichedMessage.h b/src/core/models/EnrichedMessage.h index a0f1e1b..b2e30c0 100644 --- a/src/core/models/EnrichedMessage.h +++ b/src/core/models/EnrichedMessage.h @@ -14,6 +14,9 @@ struct EnrichedMessage QString sourceMessageId; QString logicalName; QString unit; + QString metadataId; + QString pointId; + QString catalogVersion; DOMValue engineeringValue{0.0}; DataQuality quality{DataQuality::BAD}; bool alarm{false}; diff --git a/src/core/models/MetadataDef.h b/src/core/models/MetadataDef.h index 075a8fb..b32ec39 100644 --- a/src/core/models/MetadataDef.h +++ b/src/core/models/MetadataDef.h @@ -19,6 +19,8 @@ struct MetadataDef double maxValue{0.0}; // 最大值 double deadband{0.0}; // 死区 bool isWriteable{false}; // 是否可写 + QString recordStatus{QStringLiteral("ACTIVE")}; // 治理：ACTIVE / DEPRECATED 作用：控制数据是否有效 + QString entryCatalogVersion; // 写入该条时的目录版本（可选） }; } // namespace softbus::core::models diff --git a/src/core/models/Types.h b/src/core/models/Types.h index 942d67b..d3abaa4 100644 --- a/src/core/models/Types.h +++ b/src/core/models/Types.h @@ -2,6 +2,8 @@ // 定义数据域类型、数据质量类型、数据类型 #include +#include + namespace softbus::core::models { @@ -9,9 +11,8 @@ enum class DataDomain : std::uint8_t { PROCESS_PARAM = 0, // 过程参数 CONTROL_CMD = 1, // 控制指令 - MONITOR_DATA = 2, - HEALTH_DIAG = 3, - CONFIG_PARAM = 4, + MONITOR_DATA = 2, // 监控数据 + MANAGE_DATA = 3, // 管理数据 UNKNOWN = 255 }; @@ -35,4 +36,21 @@ enum class DataType : std::uint8_t BYTE_ARRAY }; +inline QString dataDomainToString(DataDomain domain) +{ + switch (domain) { + case DataDomain::PROCESS_PARAM: + return QStringLiteral("PROCESS_PARAM"); + case DataDomain::CONTROL_CMD: + return QStringLiteral("CONTROL_CMD"); + case DataDomain::MONITOR_DATA: + return QStringLiteral("MONITOR_DATA"); + case DataDomain::MANAGE_DATA: + return QStringLiteral("MANAGE_DATA"); + case DataDomain::UNKNOWN: + default: + return QStringLiteral("UNKNOWN"); + } +} + } // namespace softbus::core::models diff --git a/src/core/plugin_system/IMapperDependencies.h b/src/core/plugin_system/IMapperDependencies.h index 0fc16b0..b0c91ee 100644 --- a/src/core/plugin_system/IMapperDependencies.h +++ b/src/core/plugin_system/IMapperDependencies.h @@ -1,6 +1,7 @@ #pragma once #include +#include #include #include @@ -17,6 +18,9 @@ struct MappingLookupResult double scale{1.0}; double offset{0.0}; QString unit; + QString metadataId; + QString pointId; + QString sourceTag; // "mapping" | "profile" }; struct DiscoveryRecord @@ -34,7 +38,8 @@ class IMappingLookup { public: virtual ~IMappingLookup() = default; - virtual std::optional find(std::uint64_t compositeKey) const = 0; + virtual std::optional find(const softbus::core::models::RawBusMessage& raw, + std::uint64_t compositeKey) const = 0; }; class IDiscoverySink @@ -48,6 +53,7 @@ struct MapperSessionContext { IMappingLookup* mappingLookup{nullptr}; IDiscoverySink* discoverySink{nullptr}; + std::function sniffingEnabled; QJsonObject params; }; diff --git a/src/device_bus/manager/DeviceBusManager.cpp b/src/device_bus/manager/DeviceBusManager.cpp index b12bd21..6ead10e 100644 --- a/src/device_bus/manager/DeviceBusManager.cpp +++ b/src/device_bus/manager/DeviceBusManager.cpp @@ -16,6 +16,8 @@ #include "message_bus/egress/EgressRouter.h" #include "message_bus/ingress/DriverIngressAdapter.h" #include "message_bus/pipeline/PipelineEngine.h" +#include "core/metadata/ProfileRegistry.h" +#include "message_bus/pipeline/mapping/MappingRegistry.h" #include "message_bus/pipeline/rules/DynamicRuleRegistry.h" #include "core/plugin_system/PluginManager.h" #include "device_bus/manager/SerialDeviceManager.h" @@ -44,39 +46,118 @@ QString rulesConfigPath() return QDir::homePath() + QStringLiteral("/softbus/config/runtime_rules.json"); } +QString runtimeMappingsConfigPath() +{ + const QFileInfo fi(rulesConfigPath()); + return fi.dir().filePath(QStringLiteral("runtime_mappings.json")); +} + +bool ensureConfigFileInitialized(const QString& targetPath, + const QString& fallbackPath, + const QByteArray& defaultContent, + const char* fileLabel) +{ + if (QFile::exists(targetPath)) { + return true; + } + const QFileInfo fi(targetPath); + if (!QDir().mkpath(fi.dir().absolutePath())) { + LOG_WARNING() << "DeviceBusManager: failed to create config directory:" << fi.dir().absolutePath(); + return false; + } + + QByteArray initBytes = defaultContent; + QFile fallback(fallbackPath); + if (!fallbackPath.isEmpty() && fallback.exists() && fallback.open(QIODevice::ReadOnly)) { + initBytes = fallback.readAll(); + } + + QSaveFile initFile(targetPath); + if (!initFile.open(QIODevice::WriteOnly | QIODevice::Text)) { + LOG_WARNING() << "DeviceBusManager: failed to create" << fileLabel << "file:" << targetPath; + return false; + } + if (initFile.write(initBytes) != initBytes.size()) { + LOG_WARNING() << "DeviceBusManager: failed to initialize" << fileLabel << "file:" << targetPath; + initFile.cancelWriting(); + return false; + } + if (!initFile.commit()) { + LOG_WARNING() << "DeviceBusManager: failed to commit" << fileLabel << "file:" << targetPath; + return false; + } + LOG_INFO() << "DeviceBusManager: initialized" << fileLabel << "file:" << targetPath; + return true; +} + +void ensureCoreConfigFilesInitialized() +{ + const QFileInfo rulesFi(rulesConfigPath()); + const QString baseDir = rulesFi.dir().absolutePath(); + + const QByteArray defaultRules = QJsonDocument(QJsonObject{ + {QStringLiteral("filter"), QJsonArray{}}, + {QStringLiteral("transform"), QJsonArray{}}, + }).toJson(QJsonDocument::Indented); + (void)ensureConfigFileInitialized(rulesConfigPath(), + QStringLiteral(), + defaultRules, + "runtime_rules"); + + const QByteArray defaultMappings = QStringLiteral( + "{\n \"catalogVersion\": 1,\n \"items\": []\n}\n") + .toUtf8(); + (void)ensureConfigFileInitialized(runtimeMappingsConfigPath(), + QStringLiteral(), + defaultMappings, + "runtime_mappings"); + + const QString profilesPath = QDir(baseDir).filePath(QStringLiteral("device_profiles.json")); + const QByteArray defaultProfiles = QStringLiteral( + "{\n \"catalogVersion\": \"\",\n \"profiles\": []\n}\n") + .toUtf8(); + (void)ensureConfigFileInitialized(profilesPath, + QStringLiteral("config/device_profiles.json"), + defaultProfiles, + "device_profiles"); + + const QString metadataPath = QDir(baseDir).filePath(QStringLiteral("metadata_dictionary.json")); + const QByteArray defaultMetadata = QStringLiteral( + "{\n \"catalogVersion\": \"\",\n \"metadatas\": []\n}\n") + .toUtf8(); + (void)ensureConfigFileInitialized(metadataPath, + QStringLiteral("config/metadata_dictionary.json"), + defaultMetadata, + "metadata_dictionary"); +} + +void loadRuntimeMappingsFromDedicatedConfig() +{ + const QString path = runtimeMappingsConfigPath(); + if (!softbus::message_bus::pipeline::MappingRegistry::instance().loadPersistedFile(path)) { + LOG_WARNING() << "DeviceBusManager: failed to load runtime_mappings from" << path; + } else { + LOG_INFO() << "DeviceBusManager: loaded runtime_mappings from" << path; + } +} + +void loadProfileCatalogSingleton() +{ + const QString dir = QFileInfo(rulesConfigPath()).dir().absolutePath(); + const QString profilePath = QDir(dir).filePath(QStringLiteral("device_profiles.json")); + if (!softbus::core::metadata::ProfileRegistry::instance().loadFromFile(profilePath)) { + LOG_WARNING() << "DeviceBusManager: profile catalog load failed path=" << profilePath + << "err=" << softbus::core::metadata::ProfileRegistry::instance().lastError(); + } else { + LOG_INFO() << "DeviceBusManager: loaded device_profiles from" << profilePath; + } +} + void loadDynamicRulesFromDedicatedConfig() { auto& reg = softbus::message_bus::pipeline::DynamicRuleRegistry::instance(); const QString path = rulesConfigPath(); QFile file(path); - if (!file.exists()) { - const QFileInfo fi(path); - if (!QDir().mkpath(fi.dir().absolutePath())) { - LOG_WARNING() << "DeviceBusManager: failed to create rules config directory:" << fi.dir().absolutePath(); - return; - } - QSaveFile initFile(path); - if (!initFile.open(QIODevice::WriteOnly | QIODevice::Text)) { - LOG_WARNING() << "DeviceBusManager: failed to create rules config file:" << path; - return; - } - const QJsonObject emptyRules{ - {QStringLiteral("filter"), QJsonArray{}}, - {QStringLiteral("transform"), QJsonArray{}}, - }; - const QByteArray initBytes = QJsonDocument(emptyRules).toJson(QJsonDocument::Indented); - if (initFile.write(initBytes) != initBytes.size()) { - LOG_WARNING() << "DeviceBusManager: failed to initialize rules config file:" << path; - initFile.cancelWriting(); - return; - } - if (!initFile.commit()) { - LOG_WARNING() << "DeviceBusManager: failed to commit initialized rules config file:" << path; - return; - } - LOG_INFO() << "DeviceBusManager: created empty rules config file:" << path; - return; - } if (!file.open(QIODevice::ReadOnly | QIODevice::Text)) { LOG_WARNING() << "DeviceBusManager: failed to open rules config:" << path; return; @@ -207,7 +288,10 @@ bool DeviceBusManager::initialize(const QJsonObject& rootConfig) m_pipelineEngine->setMemoryPool(m_memoryPool); m_pipelineEngine->loadRuntimeConfigFromJson(rootConfig.value(QStringLiteral("pipeline")).toObject()); m_pipelineEngine->setPluginManager(m_pluginManager); + ensureCoreConfigFilesInitialized(); loadDynamicRulesFromDedicatedConfig(); + loadRuntimeMappingsFromDedicatedConfig(); + loadProfileCatalogSingleton(); m_pipelineEngine->start(); } if (!m_ingressAdapter) { diff --git a/src/message_bus/pipeline/PipelineEngine.cpp b/src/message_bus/pipeline/PipelineEngine.cpp index bc385ca..6821fab 100644 --- a/src/message_bus/pipeline/PipelineEngine.cpp +++ b/src/message_bus/pipeline/PipelineEngine.cpp @@ -7,7 +7,11 @@ #include #include "core/models/MessageRoutingUtils.h" +#include "device_bus/DeviceBus.h" +#include "device_bus/tree/DeviceTreeModel.h" +#include "core/metadata/ProfileRegistry.h" #include "message_bus/pipeline/PipelineEngineApi.h" +#include "message_bus/pipeline/derivation/DerivationEngine.h" #include "message_bus/pipeline/discovery/DiscoveryPool.h" #include "message_bus/pipeline/mapping/MappingRegistry.h" #include "message_bus/pipeline/stages/3_filters/ProtocolParseFilter.h" @@ -37,24 +41,113 @@ struct EndpointFramingState namespace { +QString protocolNormForProfile(softbus::core::models::ProtocolType p) +{ + using softbus::core::models::ProtocolType; + switch (p) { + case ProtocolType::MODBUS_RTU: + return QStringLiteral("modbus_rtu"); + case ProtocolType::MODBUS_ASCII: + return QStringLiteral("modbus_ascii"); + case ProtocolType::MODBUS_TCP: + return QStringLiteral("modbus_tcp"); + default: + return {}; + } +} + class MappingLookupAdapter final : public softbus::core::plugin_system::IMappingLookup { public: - std::optional find(std::uint64_t compositeKey) const override + std::optional find( + const softbus::core::models::RawBusMessage& raw, + std::uint64_t compositeKey) const override { auto rule = softbus::message_bus::pipeline::MappingRegistry::instance().find(compositeKey); - if (!rule.has_value()) { + if (rule.has_value()) { + softbus::core::plugin_system::MappingLookupResult out; + out.domPath = rule->domPath; + out.scale = rule->scale; + out.offset = rule->offset; + out.unit = rule->unit; + out.metadataId = rule->metadataId; + out.pointId = rule->pointId.isEmpty() ? rule->domPath : rule->pointId; + out.sourceTag = QStringLiteral("mapping"); + return out; + } + + const QString proto = protocolNormForProfile(raw.protocol); + if (proto.isEmpty() || raw.protocol != softbus::core::models::ProtocolType::MODBUS_RTU) { + return std::nullopt; + } + const std::uint16_t regAddr = static_cast(compositeKey & 0xFFFFu); + const auto hit = softbus::core::metadata::ProfileRegistry::instance().findModbusRegisterMapping( + raw.deviceId, proto, regAddr); + if (!hit.has_value()) { return std::nullopt; } softbus::core::plugin_system::MappingLookupResult out; - out.domPath = rule->domPath; - out.scale = rule->scale; - out.offset = rule->offset; - out.unit = rule->unit; + out.domPath = hit->domPath; + out.scale = hit->scale; + out.offset = hit->offset; + out.unit = hit->unit; + out.metadataId = hit->metadataId; + out.pointId = hit->pointId.isEmpty() ? hit->domPath : hit->pointId; + out.sourceTag = QStringLiteral("profile"); return out; } }; +void attachPesmRuntimeContext(const softbus::core::models::RawBusMessage& raw, + std::vector& msgs) +{ + if (msgs.empty()) { + return; + } + auto tree = softbus::device_bus::DeviceBus::instance().deviceTreeModel(); + std::optional devInfo; + if (raw.deviceId >= 0) { + devInfo = softbus::device_bus::DeviceBus::instance().findDeviceById(raw.deviceId); + } + const DeviceTreeNode* node = nullptr; + if (tree && devInfo.has_value()) { + const int idx = tree->findIndexByEndpoint(devInfo->endpoint); + if (idx >= 0 && idx < tree->nodes().size()) { + node = &tree->nodes().at(idx); + } + } + + for (auto& msg : msgs) { + msg.attributes.insert(QStringLiteral("endpointHash"), static_cast(raw.endpointHash)); + if (raw.deviceId >= 0) { + msg.attributes.insert(QStringLiteral("runtimeDeviceId"), raw.deviceId); + } + if (devInfo.has_value()) { + msg.attributes.insert(QStringLiteral("deviceEndpoint"), devInfo->endpoint); + if (!devInfo->protocol.isEmpty()) { + msg.attributes.insert(QStringLiteral("deviceProtocol"), devInfo->protocol); + } + } + if (node) { + msg.attributes.insert(QStringLiteral("deviceNodeId"), node->id); + const QJsonValue av = node->params.value(QStringLiteral("assetId")); + if (av.isString() && !av.toString().isEmpty()) { + msg.attributes.insert(QStringLiteral("assetId"), av.toString()); + } + const QJsonValue roleV = node->params.value(QStringLiteral("treeRole")); + if (roleV.isString() && !roleV.toString().isEmpty()) { + msg.attributes.insert(QStringLiteral("treeRole"), roleV.toString()); + } + } + if (!msg.pointId.isEmpty()) { + msg.attributes.insert(QStringLiteral("pointId"), msg.pointId); + } + if (!msg.metadataId.isEmpty()) { + msg.attributes.insert(QStringLiteral("metadataId"), msg.metadataId); + } + } +} + class DiscoverySinkAdapter final : public softbus::core::plugin_system::IDiscoverySink { public: @@ -584,6 +677,18 @@ bool PipelineEngine::runStages234(const softbus::core::models::RawBusMessage& ct << "quality=" << static_cast(enriched.quality) << "alarm=" << enriched.alarm << "traceCount=" << static_cast(enriched.traceLog.size()); + + std::vector derived; + softbus::message_bus::pipeline::derivation::DerivationEngine::instance().deriveFromEnriched(enriched, derived); + for (const auto& d : derived) { + const QString dv = std::visit( + [](const auto& x) { return QString::number(static_cast(x), 'g', 12); }, + d.engineeringValue); + LOG_INFO() << "PipelineEngine: derived message" + << "sourceMessageId=" << d.sourceMessageId + << "engineeringValue=" << dv + << "traceCount=" << static_cast(d.traceLog.size()); + } } return true; } @@ -652,6 +757,9 @@ std::vector PipelineEngine::stage2Parser(cons softbus::core::plugin_system::MapperSessionContext context; context.mappingLookup = m_mappingLookup.get(); context.discoverySink = m_discoverySink.get(); + context.sniffingEnabled = []() { + return softbus::message_bus::pipeline::DiscoveryPool::instance().isSniffingEnabled(); + }; auto s = mapperPlugin->createMapperSession(context); m_mapperSessions.insert( mapperKey, @@ -663,7 +771,9 @@ std::vector PipelineEngine::stage2Parser(cons } session = it.value(); } - return session->map(ctx, envelope); + auto messages = session->map(ctx, envelope); + attachPesmRuntimeContext(ctx, messages); + return messages; } bool PipelineEngine::stage3Filter(softbus::core::models::DOMMessage& message) diff --git a/src/message_bus/pipeline/derivation/DerivationEngine.cpp b/src/message_bus/pipeline/derivation/DerivationEngine.cpp new file mode 100644 index 0000000..5103467 --- /dev/null +++ b/src/message_bus/pipeline/derivation/DerivationEngine.cpp @@ -0,0 +1,44 @@ +#include "message_bus/pipeline/derivation/DerivationEngine.h" + +#include + +namespace softbus::message_bus::pipeline::derivation +{ + +DerivationEngine& DerivationEngine::instance() +{ + static DerivationEngine eng; + return eng; +} + +void DerivationEngine::deriveFromEnriched(const softbus::core::models::EnrichedMessage& in, + std::vector& outAppend) const +{ + if (!m_enabled) { + return; + } + if (!in.attributes.value(QStringLiteral("sb.deriveMirror")).toBool()) { + return; + } + const QString srcPoint = in.attributes.value(QStringLiteral("pointId")).toString(); + if (srcPoint.isEmpty()) { + return; + } + // MVP：若存在派生后缀配置则复制一份工程值到派生点（示例约定：pointId + ".mirror"） + const QString mirrorId = srcPoint + QStringLiteral(".mirror"); + softbus::core::models::EnrichedMessage derived; + derived.sourceMessageId = mirrorId; + derived.logicalName = in.logicalName + QStringLiteral(" (derived)"); + derived.unit = in.unit; + derived.engineeringValue = in.engineeringValue; + derived.quality = in.quality; + derived.attributes = in.attributes; + derived.traceLog = in.traceLog; + derived.attributes.insert(QStringLiteral("pointId"), mirrorId); + derived.attributes.insert(QStringLiteral("derivedFrom"), QJsonArray{srcPoint}); + derived.attributes.insert(QStringLiteral("derivationRuleId"), QStringLiteral("builtin.mirror")); + derived.attributes.insert(QStringLiteral("derivationVersion"), QStringLiteral("1")); + outAppend.push_back(std::move(derived)); +} + +} // namespace softbus::message_bus::pipeline::derivation diff --git a/src/message_bus/pipeline/derivation/DerivationEngine.h b/src/message_bus/pipeline/derivation/DerivationEngine.h new file mode 100644 index 0000000..edcf9e6 --- /dev/null +++ b/src/message_bus/pipeline/derivation/DerivationEngine.h @@ -0,0 +1,30 @@ +#pragma once + +#include + +#include "core/models/EnrichedMessage.h" + +namespace softbus::message_bus::pipeline::derivation +{ + +/// 最小推导层：基于已富化的标准点生成可追溯派生点（占位 + 单规则 MVP） +class DerivationEngine +{ +public: + static DerivationEngine& instance(); + + /// 根据当前配置，从输入生成零条或多条派生富化消息（不改变输入） + void deriveFromEnriched(const softbus::core::models::EnrichedMessage& in, + std::vector& outAppend) const; + + void setEnabled(bool on) { m_enabled = on; } + bool enabled() const { return m_enabled; } + +private: + DerivationEngine() = default; + +private: + bool m_enabled{true}; +}; + +} // namespace softbus::message_bus::pipeline::derivation diff --git a/src/message_bus/pipeline/mapping/MappingRegistry.cpp b/src/message_bus/pipeline/mapping/MappingRegistry.cpp index 15dfc1a..e0b69e8 100644 --- a/src/message_bus/pipeline/mapping/MappingRegistry.cpp +++ b/src/message_bus/pipeline/mapping/MappingRegistry.cpp @@ -1,5 +1,13 @@ #include "message_bus/pipeline/mapping/MappingRegistry.h" +#include +#include +#include +#include +#include +#include +#include + namespace softbus::message_bus::pipeline { @@ -15,14 +23,23 @@ bool MappingRegistry::upsert(const MappingRule& rule) return false; } QWriteLocker lk(&m_lock); - m_rules[rule.compositeKey] = rule; + MappingRule r = rule; + if (r.updatedAtMs <= 0) { + r.updatedAtMs = QDateTime::currentMSecsSinceEpoch(); + } + m_rules[rule.compositeKey] = std::move(r); + ++m_catalogVersion; return true; } bool MappingRegistry::remove(std::uint64_t compositeKey) { QWriteLocker lk(&m_lock); - return m_rules.erase(compositeKey) > 0; + const bool ok = m_rules.erase(compositeKey) > 0; + if (ok) { + ++m_catalogVersion; + } + return ok; } std::optional MappingRegistry::find(std::uint64_t compositeKey) const @@ -58,6 +75,21 @@ QJsonArray MappingRegistry::list() const o.insert(QStringLiteral("scale"), rule.scale); o.insert(QStringLiteral("offset"), rule.offset); o.insert(QStringLiteral("unit"), rule.unit); + if (!rule.metadataId.isEmpty()) { + o.insert(QStringLiteral("metadataId"), rule.metadataId); + } + if (!rule.pointId.isEmpty()) { + o.insert(QStringLiteral("pointId"), rule.pointId); + } + if (!rule.catalogVersion.isEmpty()) { + o.insert(QStringLiteral("catalogVersion"), rule.catalogVersion); + } + if (rule.updatedAtMs > 0) { + o.insert(QStringLiteral("updatedAtMs"), rule.updatedAtMs); + } + if (!rule.updatedBy.isEmpty()) { + o.insert(QStringLiteral("updatedBy"), rule.updatedBy); + } arr.push_back(o); } return arr; @@ -66,7 +98,110 @@ QJsonArray MappingRegistry::list() const QJsonObject MappingRegistry::status() const { QReadLocker lk(&m_lock); - return QJsonObject{{QStringLiteral("mappingRuleCount"), static_cast(m_rules.size())}}; + QJsonObject o; + o.insert(QStringLiteral("mappingRuleCount"), static_cast(m_rules.size())); + o.insert(QStringLiteral("mappingCatalogVersion"), m_catalogVersion); + return o; +} + +bool MappingRegistry::loadPersistedFile(const QString& path) +{ + QFile f(path); + if (!f.exists()) { + return true; + } + if (!f.open(QIODevice::ReadOnly | QIODevice::Text)) { + return false; + } + QJsonParseError pe; + const QJsonDocument doc = QJsonDocument::fromJson(f.readAll(), &pe); + if (pe.error != QJsonParseError::NoError || !doc.isObject()) { + return false; + } + const QJsonObject root = doc.object(); + const qint64 ver = root.value(QStringLiteral("catalogVersion")).toVariant().toLongLong(); + const QJsonArray items = root.value(QStringLiteral("items")).toArray(); + + QWriteLocker lk(&m_lock); + m_rules.clear(); + for (const auto& v : items) { + if (!v.isObject()) { + continue; + } + const QJsonObject o = v.toObject(); + MappingRule rule; + rule.compositeKey = o.value(QStringLiteral("compositeKey")).toString().toULongLong(); + rule.domPath = o.value(QStringLiteral("domPath")).toString(); + rule.scale = o.value(QStringLiteral("scale")).toDouble(1.0); + rule.offset = o.value(QStringLiteral("offset")).toDouble(0.0); + rule.unit = o.value(QStringLiteral("unit")).toString(); + rule.metadataId = o.value(QStringLiteral("metadataId")).toString(); + rule.pointId = o.value(QStringLiteral("pointId")).toString(); + rule.catalogVersion = o.value(QStringLiteral("catalogVersion")).toString(); + rule.updatedAtMs = o.value(QStringLiteral("updatedAtMs")).toVariant().toLongLong(); + rule.updatedBy = o.value(QStringLiteral("updatedBy")).toString(); + if (rule.compositeKey != 0 && !rule.domPath.isEmpty()) { + m_rules[rule.compositeKey] = rule; + } + } + if (ver > 0) { + m_catalogVersion = ver; + } + return true; +} + +bool MappingRegistry::savePersistedFile(const QString& path) const +{ + QJsonArray items; + { + QReadLocker lk(&m_lock); + for (const auto& kv : m_rules) { + const auto& rule = kv.second; + QJsonObject o; + o.insert(QStringLiteral("compositeKey"), QString::number(rule.compositeKey)); + o.insert(QStringLiteral("domPath"), rule.domPath); + o.insert(QStringLiteral("scale"), rule.scale); + o.insert(QStringLiteral("offset"), rule.offset); + o.insert(QStringLiteral("unit"), rule.unit); + if (!rule.metadataId.isEmpty()) { + o.insert(QStringLiteral("metadataId"), rule.metadataId); + } + if (!rule.pointId.isEmpty()) { + o.insert(QStringLiteral("pointId"), rule.pointId); + } + if (!rule.catalogVersion.isEmpty()) { + o.insert(QStringLiteral("catalogVersion"), rule.catalogVersion); + } + if (rule.updatedAtMs > 0) { + o.insert(QStringLiteral("updatedAtMs"), rule.updatedAtMs); + } + if (!rule.updatedBy.isEmpty()) { + o.insert(QStringLiteral("updatedBy"), rule.updatedBy); + } + items.push_back(o); + } + } + + qint64 verSnap = 1; + { + QReadLocker lk(&m_lock); + verSnap = m_catalogVersion; + } + + QJsonObject root; + root.insert(QStringLiteral("catalogVersion"), verSnap); + root.insert(QStringLiteral("items"), items); + + QSaveFile out(path); + if (!out.open(QIODevice::WriteOnly | QIODevice::Text)) { + return false; + } + const QByteArray bytes = QJsonDocument(root).toJson(QJsonDocument::Indented); + if (out.write(bytes) != bytes.size()) { + out.cancelWriting(); + return false; + } + return out.commit(); } } // namespace softbus::message_bus::pipeline diff --git a/src/message_bus/pipeline/mapping/MappingRegistry.h b/src/message_bus/pipeline/mapping/MappingRegistry.h index bcc41ab..d56d164 100644 --- a/src/message_bus/pipeline/mapping/MappingRegistry.h +++ b/src/message_bus/pipeline/mapping/MappingRegistry.h @@ -19,6 +19,11 @@ struct MappingRule double scale{1.0}; double offset{0.0}; QString unit; + QString metadataId; + QString pointId; + QString catalogVersion; + qint64 updatedAtMs{0}; + QString updatedBy; }; class MappingRegistry @@ -34,12 +39,17 @@ public: QJsonArray list() const; QJsonObject status() const; + bool loadPersistedFile(const QString& path); + bool savePersistedFile(const QString& path) const; + qint64 mappingCatalogVersion() const { return m_catalogVersion; } + private: MappingRegistry() = default; private: mutable QReadWriteLock m_lock; std::unordered_map m_rules; + qint64 m_catalogVersion{1}; }; } // namespace softbus::message_bus::pipeline diff --git a/src/message_bus/pipeline/stages/3_filters/ProtocolParseFilter.cpp b/src/message_bus/pipeline/stages/3_filters/ProtocolParseFilter.cpp index 1884fdf..fef92eb 100644 --- a/src/message_bus/pipeline/stages/3_filters/ProtocolParseFilter.cpp +++ b/src/message_bus/pipeline/stages/3_filters/ProtocolParseFilter.cpp @@ -13,14 +13,18 @@ bool ProtocolParseFilter::process(softbus::core::models::DOMMessage& message) if (!ensureLoaded()) { return true; } - // Message-stage filter: only normalize quality by metadata range. - if (!message.metaRef) { - return true; + const softbus::core::models::MetadataDef* meta = message.metaRef; + if (!meta && !message.metadataId.isEmpty()) { + meta = m_metadataRegistry.findById(message.metadataId); } - const auto* meta = m_metadataRegistry.findById(message.metaRef->metadataId); if (!meta) { return true; } + if (message.metadataId.isEmpty()) { + message.metadataId = meta->metadataId; + } + message.attributes.insert(QStringLiteral("domain"), + softbus::core::models::dataDomainToString(meta->domain)); const double v = std::visit( [](const auto& x) { return static_cast(x); }, message.value); @@ -36,12 +40,27 @@ ProtocolParseFilter::enrich(const softbus::core::models::DOMMessage& message) { softbus::core::models::EnrichedMessage out; out.sourceMessageId = message.messageId; - out.logicalName = message.metaRef ? message.metaRef->name : message.messageId; - out.unit = message.metaRef ? message.metaRef->unit : QString(); + out.metadataId = message.metadataId; + out.pointId = message.pointId.isEmpty() ? message.messageId : message.pointId; + + const softbus::core::models::MetadataDef* meta = message.metaRef; + if (!meta && !message.metadataId.isEmpty()) { + meta = m_metadataRegistry.findById(message.metadataId); + } + out.logicalName = meta ? meta->name : message.messageId; + out.unit = meta ? meta->unit : QString(); + if (!meta && message.attributes.contains(QStringLiteral("unit"))) { + out.unit = message.attributes.value(QStringLiteral("unit")).toString(); + } out.engineeringValue = message.value; out.quality = message.quality; out.attributes = message.attributes; out.traceLog = message.traceLog; + if (meta && !meta->entryCatalogVersion.isEmpty()) { + out.catalogVersion = meta->entryCatalogVersion; + } else if (!m_metadataRegistry.catalogVersion().isEmpty()) { + out.catalogVersion = m_metadataRegistry.catalogVersion(); + } return out; } @@ -64,4 +83,3 @@ bool ProtocolParseFilter::ensureLoaded() const } } // namespace softbus::message_bus::pipeline::stages::filters -