From 628ccf14092c1a82bfcceb748ba4f2241e0c0a80 Mon Sep 17 00:00:00 2001 From: flower_linux <737666956@qq.com> Date: Fri, 12 Jun 2026 18:34:49 +0800 Subject: [PATCH] phase 2 serial --- CMakeLists.txt | 2 +- cmake/SoftbusThirdParty.cmake | 39 ++ cmake/libmodbus/CMakeLists.txt | 51 ++ cmake/libmodbus/config.h.linux | 46 ++ cmake/libmodbus/modbus-version.h | 19 + config/daemon/bindings_global.json | 20 + config/{ => daemon}/daemon_profile.json | 0 config/daemon/logical_points.json | 37 ++ config/{ => daemon}/point_table.json | 0 config/{ => daemon}/point_table.yaml | 0 config/{ => daemon}/topic_rule.json | 0 config/{ => daemon}/topic_rule.yaml | 0 config/edge.json | 9 - config/edge/bindings_local.json | 17 + config/edge/edge.json | 13 + config/edge/physical_channels.json | 15 + config/edge/physical_devices.json | 29 ++ docs/edge_device_setup.md | 60 +++ docs/softbus_edge.md | 452 +++++++++++++++--- .../protocol_modbus/src/ModbusParser.cpp | 3 +- .../include/softbus_command/CommandService.h | 3 + src/softbus_command/src/CommandService.cpp | 10 + src/softbus_core/CMakeLists.txt | 1 + .../softbus_core/config/ConfigWatcher.h | 23 + .../softbus_core/models/SoftbusTypes.h | 1 + .../softbus_core/platform/PlatformSignal.h | 4 + src/softbus_core/src/config/ConfigWatcher.cpp | 41 ++ src/softbus_core/src/models/SoftbusTypes.cpp | 9 +- .../src/platform/PlatformSignal.cpp | 32 +- src/softbus_daemon/src/CoreService.cpp | 96 ++-- src/softbus_daemon/src/main.cpp | 14 +- src/softbus_edge/CMakeLists.txt | 10 +- .../include/softbus_edge/config/EdgeConfig.h | 7 +- .../softbus_edge/config/PhysicalRegistry.h | 42 ++ .../softbus_edge/config/PhysicalTypes.h | 41 ++ .../softbus_edge/driver/ModbusChannel.h | 32 ++ .../softbus_edge/egress/CommandExecutor.h | 24 +- .../softbus_edge/egress/ModbusEgress.h | 23 + .../softbus_edge/egress/ShadowChannelStore.h | 9 +- .../softbus_edge/egress/UplinkClient.h | 2 + .../softbus_edge/factory/DeviceFactory.h | 15 + .../softbus_edge/ingress/ModbusIngress.h | 30 ++ .../softbus_edge/runtime/EdgeRuntime.h | 51 ++ src/softbus_edge/src/CommandExecutor.cpp | 110 +++-- src/softbus_edge/src/DeviceFactory.cpp | 12 + src/softbus_edge/src/EdgeConfig.cpp | 31 +- src/softbus_edge/src/EdgeRuntime.cpp | 138 ++++++ src/softbus_edge/src/ModbusChannel.cpp | 123 +++++ src/softbus_edge/src/ModbusEgress.cpp | 33 ++ src/softbus_edge/src/ModbusIngress.cpp | 92 ++++ src/softbus_edge/src/PhysicalRegistry.cpp | 152 ++++++ src/softbus_edge/src/ShadowChannelStore.cpp | 22 +- src/softbus_edge/src/UplinkClient.cpp | 10 + src/softbus_edge/src/main.cpp | 111 ++--- src/softbus_monitor/CMakeLists.txt | 1 + .../softbus_monitor/HeartbeatMonitor.h | 14 +- src/softbus_monitor/src/HeartbeatMonitor.cpp | 46 +- src/softbus_opcua/CMakeLists.txt | 1 + .../include/softbus_pipeline/PipelineEngine.h | 5 +- src/softbus_pipeline/src/PipelineEngine.cpp | 53 +- src/softbus_registry/CMakeLists.txt | 1 + .../include/softbus_registry/BindingEntry.h | 21 + .../softbus_registry/BindingRegistry.h | 38 ++ .../include/softbus_registry/PointRegistry.h | 10 + src/softbus_registry/src/BindingRegistry.cpp | 200 ++++++++ src/softbus_registry/src/PointRegistry.cpp | 127 ++++- .../softbus_transport/TcpUplinkClient.h | 1 + .../softbus_transport/TcpUplinkServer.h | 21 +- .../include/softbus_transport/UplinkWire.h | 3 + src/softbus_transport/src/TcpUplinkClient.cpp | 12 + src/softbus_transport/src/TcpUplinkServer.cpp | 120 ++++- .../src/TransportBusBridge.cpp | 9 +- src/softbus_transport/src/UplinkWire.cpp | 14 + tests/integration/valve_control_loop.sh | 9 +- third_party/README.md | 13 + third_party/libmodbus | 1 + tools/CMakeLists.txt | 3 + tools/modbus_scanner.cpp | 27 ++ 78 files changed, 2571 insertions(+), 345 deletions(-) create mode 100644 cmake/libmodbus/CMakeLists.txt create mode 100644 cmake/libmodbus/config.h.linux create mode 100644 cmake/libmodbus/modbus-version.h create mode 100644 config/daemon/bindings_global.json rename config/{ => daemon}/daemon_profile.json (100%) create mode 100644 config/daemon/logical_points.json rename config/{ => daemon}/point_table.json (100%) rename config/{ => daemon}/point_table.yaml (100%) rename config/{ => daemon}/topic_rule.json (100%) rename config/{ => daemon}/topic_rule.yaml (100%) delete mode 100644 config/edge.json create mode 100644 config/edge/bindings_local.json create mode 100644 config/edge/edge.json create mode 100644 config/edge/physical_channels.json create mode 100644 config/edge/physical_devices.json create mode 100644 docs/edge_device_setup.md create mode 100644 src/softbus_core/include/softbus_core/config/ConfigWatcher.h create mode 100644 src/softbus_core/src/config/ConfigWatcher.cpp create mode 100644 src/softbus_edge/include/softbus_edge/config/PhysicalRegistry.h create mode 100644 src/softbus_edge/include/softbus_edge/config/PhysicalTypes.h create mode 100644 src/softbus_edge/include/softbus_edge/driver/ModbusChannel.h create mode 100644 src/softbus_edge/include/softbus_edge/egress/ModbusEgress.h create mode 100644 src/softbus_edge/include/softbus_edge/ingress/ModbusIngress.h create mode 100644 src/softbus_edge/include/softbus_edge/runtime/EdgeRuntime.h create mode 100644 src/softbus_edge/src/EdgeRuntime.cpp create mode 100644 src/softbus_edge/src/ModbusChannel.cpp create mode 100644 src/softbus_edge/src/ModbusEgress.cpp create mode 100644 src/softbus_edge/src/ModbusIngress.cpp create mode 100644 src/softbus_edge/src/PhysicalRegistry.cpp create mode 100644 src/softbus_registry/include/softbus_registry/BindingEntry.h create mode 100644 src/softbus_registry/include/softbus_registry/BindingRegistry.h create mode 100644 src/softbus_registry/src/BindingRegistry.cpp create mode 160000 third_party/libmodbus create mode 100644 tools/modbus_scanner.cpp diff --git a/CMakeLists.txt b/CMakeLists.txt index 4d597ff..42c60d6 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -1,5 +1,5 @@ cmake_minimum_required(VERSION 3.16) -project(softbus_workspace LANGUAGES CXX) +project(softbus_workspace LANGUAGES CXX C) set(CMAKE_CXX_STANDARD 17) set(CMAKE_CXX_STANDARD_REQUIRED ON) diff --git a/cmake/SoftbusThirdParty.cmake b/cmake/SoftbusThirdParty.cmake index fcabe0a..b5afd09 100644 --- a/cmake/SoftbusThirdParty.cmake +++ b/cmake/SoftbusThirdParty.cmake @@ -73,6 +73,45 @@ else() set(SOFTBUS_HAS_OPCUA OFF) endif() +# libmodbus (edge RTU/TCP): system pkg-config first, else third_party vendor +set(SOFTBUS_HAS_LIBMODBUS OFF) +set(SOFTBUS_LIBMODBUS_VENDOR OFF) + +option(SOFTBUS_FORCE_VENDOR_LIBMODBUS + "Always build libmodbus from third_party/libmodbus" OFF) + +if(NOT SOFTBUS_FORCE_VENDOR_LIBMODBUS) + find_package(PkgConfig QUIET) + if(PkgConfig_FOUND) + pkg_check_modules(LIBMODBUS QUIET libmodbus) + endif() +endif() + +if(LIBMODBUS_FOUND AND NOT SOFTBUS_FORCE_VENDOR_LIBMODBUS) + set(SOFTBUS_HAS_LIBMODBUS ON) + add_library(libmodbus_iface INTERFACE) + target_include_directories(libmodbus_iface INTERFACE ${LIBMODBUS_INCLUDE_DIRS}) + target_link_libraries(libmodbus_iface INTERFACE ${LIBMODBUS_LIBRARIES}) + message(STATUS "libmodbus: using system package (pkg-config)") +else() + add_subdirectory("${CMAKE_CURRENT_LIST_DIR}/libmodbus") + if(NOT SOFTBUS_HAS_LIBMODBUS) + message(STATUS "libmodbus not available; softbus_edge uses simulated Modbus I/O") + endif() +endif() + +function(softbus_apply_libmodbus target) + if(SOFTBUS_HAS_LIBMODBUS) + target_compile_definitions(${target} PRIVATE SOFTBUS_HAS_LIBMODBUS=1) + target_link_libraries(${target} PRIVATE libmodbus_iface) + if(SOFTBUS_LIBMODBUS_VENDOR) + target_compile_definitions(${target} PRIVATE SOFTBUS_LIBMODBUS_VENDOR=1) + endif() + else() + target_compile_definitions(${target} PRIVATE SOFTBUS_HAS_LIBMODBUS=0) + endif() +endfunction() + function(softbus_apply_open62541 target) if(SOFTBUS_HAS_OPCUA) target_compile_definitions(${target} PRIVATE SOFTBUS_HAS_OPCUA=1) diff --git a/cmake/libmodbus/CMakeLists.txt b/cmake/libmodbus/CMakeLists.txt new file mode 100644 index 0000000..3410565 --- /dev/null +++ b/cmake/libmodbus/CMakeLists.txt @@ -0,0 +1,51 @@ +# Vendored libmodbus v3.1.10 static library for softbus_edge + +set(_src_dir "${CMAKE_SOURCE_DIR}/third_party/libmodbus") +if(NOT EXISTS "${_src_dir}/src/modbus.c") + message(WARNING "Vendor libmodbus not found at ${_src_dir}") + return() +endif() + +set(_gen_dir "${CMAKE_BINARY_DIR}/libmodbus_vendor") +file(MAKE_DIRECTORY "${_gen_dir}") + +configure_file( + "${CMAKE_SOURCE_DIR}/cmake/libmodbus/config.h.linux" + "${_gen_dir}/config.h" + COPYONLY +) +configure_file( + "${CMAKE_SOURCE_DIR}/cmake/libmodbus/modbus-version.h" + "${_gen_dir}/modbus-version.h" + COPYONLY +) + +add_library(modbus_vendor STATIC + "${_src_dir}/src/modbus.c" + "${_src_dir}/src/modbus-data.c" + "${_src_dir}/src/modbus-rtu.c" + "${_src_dir}/src/modbus-tcp.c" +) +set_target_properties(modbus_vendor PROPERTIES + C_STANDARD 99 + POSITION_INDEPENDENT_CODE ON +) +target_include_directories(modbus_vendor + PUBLIC + "${_src_dir}/src" + "${_gen_dir}" +) +if(UNIX) + target_link_libraries(modbus_vendor PUBLIC pthread) +endif() + +add_library(libmodbus_iface INTERFACE) +target_link_libraries(libmodbus_iface INTERFACE modbus_vendor) +target_include_directories(libmodbus_iface INTERFACE + "${_src_dir}/src" + "${_gen_dir}" +) + +set(SOFTBUS_HAS_LIBMODBUS ON PARENT_SCOPE) +set(SOFTBUS_LIBMODBUS_VENDOR ON PARENT_SCOPE) +message(STATUS "libmodbus: using vendored build from third_party/libmodbus") diff --git a/cmake/libmodbus/config.h.linux b/cmake/libmodbus/config.h.linux new file mode 100644 index 0000000..b17d792 --- /dev/null +++ b/cmake/libmodbus/config.h.linux @@ -0,0 +1,46 @@ +/* Linux config.h for vendored libmodbus v3.1.10 */ + +#define HAVE_ARPA_INET_H 1 +#define HAVE_DECL_TIOCSRS485 1 +#define HAVE_DECL_TIOCM_RTS 1 +#define HAVE_ERRNO_H 1 +#define HAVE_FCNTL_H 1 +#define HAVE_FORK 1 +#define HAVE_GETADDRINFO 1 +#define HAVE_GETTIMEOFDAY 1 +#define HAVE_INTTYPES_H 1 +#define HAVE_LIMITS_H 1 +#define HAVE_LINUX_SERIAL_H 1 +#define HAVE_MEMORY_H 1 +#define HAVE_MEMSET 1 +#define HAVE_NETDB_H 1 +#define HAVE_NETINET_IN_H 1 +#define HAVE_NETINET_TCP_H 1 +#define HAVE_SELECT 1 +#define HAVE_SOCKET 1 +#define HAVE_STDINT_H 1 +#define HAVE_STDLIB_H 1 +#define HAVE_STRERROR 1 +#define HAVE_STRINGS_H 1 +#define HAVE_STRING_H 1 +#define HAVE_SYS_IOCTL_H 1 +#define HAVE_SYS_SOCKET_H 1 +#define HAVE_SYS_STAT_H 1 +#define HAVE_SYS_TIME_H 1 +#define HAVE_SYS_TYPES_H 1 +#define HAVE_TERMIOS_H 1 +#define HAVE_TIME_H 1 +#define HAVE_UNISTD_H 1 +#define HAVE_VFORK 1 +#define HAVE_WORKING_FORK 1 +#define HAVE_WORKING_VFORK 1 +#define PACKAGE "libmodbus" +#define PACKAGE_BUGREPORT "https://github.com/stephane/libmodbus/issues" +#define PACKAGE_NAME "libmodbus" +#define PACKAGE_STRING "libmodbus 3.1.10" +#define PACKAGE_TARNAME "libmodbus" +#define PACKAGE_URL "" +#define PACKAGE_VERSION "3.1.10" +#define STDC_HEADERS 1 +#define TIME_WITH_SYS_TIME 1 +#define VERSION "3.1.10" diff --git a/cmake/libmodbus/modbus-version.h b/cmake/libmodbus/modbus-version.h new file mode 100644 index 0000000..3c689c7 --- /dev/null +++ b/cmake/libmodbus/modbus-version.h @@ -0,0 +1,19 @@ +#ifndef MODBUS_VERSION_H +#define MODBUS_VERSION_H + +#define LIBMODBUS_VERSION_MAJOR (3) +#define LIBMODBUS_VERSION_MINOR (1) +#define LIBMODBUS_VERSION_MICRO (10) +#define LIBMODBUS_VERSION 3.1.10 +#define LIBMODBUS_VERSION_STRING "3.1.10" +#define LIBMODBUS_VERSION_HEX \ + ((LIBMODBUS_VERSION_MAJOR << 16) | (LIBMODBUS_VERSION_MINOR << 8) | \ + (LIBMODBUS_VERSION_MICRO << 0)) + +#define LIBMODBUS_VERSION_CHECK(major, minor, micro) \ + (LIBMODBUS_VERSION_MAJOR > (major) || \ + (LIBMODBUS_VERSION_MAJOR == (major) && LIBMODBUS_VERSION_MINOR > (minor)) || \ + (LIBMODBUS_VERSION_MAJOR == (major) && LIBMODBUS_VERSION_MINOR == (minor) && \ + LIBMODBUS_VERSION_MICRO >= (micro))) + +#endif /* MODBUS_VERSION_H */ diff --git a/config/daemon/bindings_global.json b/config/daemon/bindings_global.json new file mode 100644 index 0000000..bd7e8e6 --- /dev/null +++ b/config/daemon/bindings_global.json @@ -0,0 +1,20 @@ +{ + "bindings": [ + { + "bindingId": "bind_temp_01", + "edgeId": "edge01", + "physicalPointId": "edge01/phy_modbus_s1_r40001", + "logicalPointId": "EdgeDevice_01/MotorTemperature", + "scale": 0.1, + "enabled": false + }, + { + "bindingId": "bind_valve_01", + "edgeId": "edge01", + "physicalPointId": "edge01/phy_modbus_s1_r40002", + "logicalPointId": "EdgeDevice_01/ValveOpening", + "scale": 1.0, + "enabled": true + } + ] +} diff --git a/config/daemon_profile.json b/config/daemon/daemon_profile.json similarity index 100% rename from config/daemon_profile.json rename to config/daemon/daemon_profile.json diff --git a/config/daemon/logical_points.json b/config/daemon/logical_points.json new file mode 100644 index 0000000..4947cb0 --- /dev/null +++ b/config/daemon/logical_points.json @@ -0,0 +1,37 @@ +{ + "site": "DemoSite", + "system": "DemoSystem", + "asset": "DemoAsset", + "devices": [ + { + "logicalDeviceId": "EdgeDevice_01", + "displayName": "边缘设备01", + "points": [ + { + "logicalPointId": "MotorTemperature", + "displayName": "电机温度", + "signalType": "AI", + "unit": "℃", + "range": [0, 200], + "access": "read", + "writable": false, + "opcuaNodeId": "ns=1;s=DemoSite/DemoSystem/DemoAsset/EdgeDevice_01/MotorTemperature", + "dataTopic": "data/EdgeDevice_01/MotorTemperature" + }, + { + "logicalPointId": "ValveOpening", + "displayName": "阀门开度", + "signalType": "AO", + "unit": "%", + "range": [0, 100], + "access": "read_write", + "writable": true, + "opcuaNodeId": "ns=1;s=DemoSite/DemoSystem/DemoAsset/EdgeDevice_01/ValveOpening", + "cmdTopic": "control/cmd/EdgeDevice_01/ValveOpening/set", + "ackTopic": "control/ack/EdgeDevice_01/ValveOpening", + "dataTopic": "data/EdgeDevice_01/ValveOpening" + } + ] + } + ] +} diff --git a/config/point_table.json b/config/daemon/point_table.json similarity index 100% rename from config/point_table.json rename to config/daemon/point_table.json diff --git a/config/point_table.yaml b/config/daemon/point_table.yaml similarity index 100% rename from config/point_table.yaml rename to config/daemon/point_table.yaml diff --git a/config/topic_rule.json b/config/daemon/topic_rule.json similarity index 100% rename from config/topic_rule.json rename to config/daemon/topic_rule.json diff --git a/config/topic_rule.yaml b/config/daemon/topic_rule.yaml similarity index 100% rename from config/topic_rule.yaml rename to config/daemon/topic_rule.yaml diff --git a/config/edge.json b/config/edge.json deleted file mode 100644 index c81a4e3..0000000 --- a/config/edge.json +++ /dev/null @@ -1,9 +0,0 @@ -{ - "edge": { - "deviceId": "EdgeDevice_01", - "daemonHost": "127.0.0.1", - "daemonPort": 9000, - "heartbeatPeriodMs": 1000, - "pointTable": "config/point_table.json" - } -} diff --git a/config/edge/bindings_local.json b/config/edge/bindings_local.json new file mode 100644 index 0000000..1596ca0 --- /dev/null +++ b/config/edge/bindings_local.json @@ -0,0 +1,17 @@ +{ + "edgeId": "edge01", + "bindings": [ + { + "bindingId": "bind_temp_01", + "physicalPointId": "edge01/phy_modbus_s1_r40001", + "logicalPointId": "EdgeDevice_01/MotorTemperature", + "enabled": false + }, + { + "bindingId": "bind_valve_01", + "physicalPointId": "edge01/phy_modbus_s1_r40002", + "logicalPointId": "EdgeDevice_01/ValveOpening", + "enabled": true + } + ] +} diff --git a/config/edge/edge.json b/config/edge/edge.json new file mode 100644 index 0000000..1cd062b --- /dev/null +++ b/config/edge/edge.json @@ -0,0 +1,13 @@ +{ + "edge": { + "edgeId": "edge01", + "deviceId": "EdgeDevice_01", + "daemonHost": "127.0.0.1", + "daemonPort": 9000, + "heartbeatPeriodMs": 1000, + "uplinkReconnectPeriodMs": 5000, + "physicalChannels": "physical_channels.json", + "physicalDevices": "physical_devices.json", + "bindingsLocal": "bindings_local.json" + } +} diff --git a/config/edge/physical_channels.json b/config/edge/physical_channels.json new file mode 100644 index 0000000..66cba85 --- /dev/null +++ b/config/edge/physical_channels.json @@ -0,0 +1,15 @@ +{ + "channels": [ + { + "channelId": "serial_0", + "protocol": "modbus", + "transport": "rtu", + "port": "/dev/ttyS3", + "baudRate": 9600, + "parity": "N", + "dataBits": 8, + "stopBits": 1, + "pollIntervalMs": 500 + } + ] +} diff --git a/config/edge/physical_devices.json b/config/edge/physical_devices.json new file mode 100644 index 0000000..bba002b --- /dev/null +++ b/config/edge/physical_devices.json @@ -0,0 +1,29 @@ +{ + "devices": [ + { + "physicalDeviceId": "phy_modbus_slave_1", + "channelId": "serial_0", + "protocol": "modbus", + "slaveId": 1, + "points": [ + { + "physicalPointId": "phy_modbus_s1_r40001", + "register": 40001, + "function": 3, + "scale": 0.1, + "signalType": "AI", + "writable": false + }, + { + "physicalPointId": "phy_modbus_s1_r40002", + "register": 40002, + "function": 3, + "writeFunction": 6, + "scale": 1.0, + "signalType": "AO", + "writable": true + } + ] + } + ] +} diff --git a/docs/edge_device_setup.md b/docs/edge_device_setup.md new file mode 100644 index 0000000..d122f81 --- /dev/null +++ b/docs/edge_device_setup.md @@ -0,0 +1,60 @@ +# Edge 设备配置指南 + +## 配置目录 + +| 目录 | 用途 | +|------|------| +| `config/edge/` | 边缘设备本地配置（物理层 + 可选本地绑定子集） | +| `config/daemon/` | 核心守护进程配置（逻辑点、全局绑定、OPC 主题规则） | + +## 双域配置 + +| 文件 | 目录 | 职责 | +|------|------|------| +| `edge.json` | `config/edge/` | edge 入口：uplink 地址、子配置文件路径 | +| `physical_channels.json` | `config/edge/` | 串口/网络接口 | +| `physical_devices.json` | `config/edge/` | 从站、寄存器、物理点 ID | +| `bindings_local.json` | `config/edge/` | 可选，本 edge 绑定子集（校验/本地调试） | +| `logical_points.json` | `config/daemon/` | OPC/业务语义 | +| `bindings_global.json` | `config/daemon/` | 物理点 ↔ 逻辑点 + edgeId 路由 | +| `topic_rule.json` | `config/daemon/` | 总线主题模板 | +| `point_table.json` | `config/daemon/` | **遗留**全量点表，仅 daemon 兼容加载 | + +edge **不加载** `logical_points.json`、`point_table.json` 或全量 `bindings_global`；下行命令由 daemon 解析后携带 `physicalPointId` 与 `edgeId`。 + +## 已知设备流程 + +1. edge：编辑 `config/edge/physical_channels.json`、`config/edge/physical_devices.json` +2. daemon：编辑 `config/daemon/logical_points.json` +3. daemon：编辑 `config/daemon/bindings_global.json`（含 `edgeId`、`scale`） +4. 启动：`softbus_daemon --logical-points config/daemon/logical_points.json --bindings-global config/daemon/bindings_global.json` +5. 启动：`softbus_edge --config config/edge/edge.json` +6. 热加载：`kill -HUP ` 或 `kill -HUP ` + +## 未知设备（物理先行） + +1. 仅配置 `physical_devices.json`，物理点 ID 如 `phy_modbus_s1_r40001` +2. edge 启动后 daemon 收到 `edge/raw/unbound/*` 主题数据 +3. 后补 `logical_points.json` 与 `bindings_global.json`，HUP daemon 即可打通 OPC + +## libmodbus + +构建时自动选择： + +1. **系统包**（优先）：`sudo apt install libmodbus-dev` +2. **vendor**（默认回退）：`third_party/libmodbus` v3.1.10 静态编译进 `softbus_edge` + +强制使用 vendor： + +```bash +cmake -B build -DSOFTBUS_FORCE_VENDOR_LIBMODBUS=ON +cmake --build build +``` + +串口权限：`sudo usermod -aG dialout $USER` 后重新登录。 + +## Mock 模式 + +```bash +./build/bin/softbus_edge --config config/edge/edge.json --mock +``` diff --git a/docs/softbus_edge.md b/docs/softbus_edge.md index 966bcbb..99bf37b 100644 --- a/docs/softbus_edge.md +++ b/docs/softbus_edge.md @@ -6,27 +6,177 @@ ## 1. 概述 -`softbus_edge` 是边缘代理可执行文件，负责设备数据采集（ingress）、原始包上行（egress）与下行命令接收。Phase 1 使用 `MockModbusIngress` 模拟 Modbus 数据，无真实硬件依赖。 +`softbus_edge` 是边缘代理进程，运行在靠近现场设备的一侧，负责： -## 2. CONSTRAINTS +- **物理域采集（ingress）**：按 `physical_*.json` 轮询 Modbus RTU/TCP 等设备，产出 `RawPacket` +- **上行（egress/uplink）**：经 TCP 将 `RawPacket`、心跳、注册信息发送给 `softbus_daemon` +- **下行（egress）**：接收 daemon 下发的 `ExecuteCommand`，写回物理设备（Modbus 写寄存器等） -- 边缘代理不得直接操作 OPC UA 或插件解析，仅产生 `RawPacket` 并发送。 -- ingress 实现须遵循 `IIngressPort` 接口，便于 Phase 2 替换为真实硬件驱动。 -- 下行命令 Phase 1 仅日志输出，不得假装已写回设备。 -- 边缘代理通过 CLI 参数配置连接地址，不读取 `daemon_profile.json`。 +edge **不**加载逻辑点表、**不**操作 OPC UA、**不**运行 `protocol_modbus` 插件。协议解析与 OPC 绑定均在 daemon 侧完成。 -## 3. 组件架构 +| 模式 | 说明 | +|------|------| +| **正常模式** | libmodbus + `ModbusIngress`，读真实硬件 | +| **Mock 模式** | `--mock`，无硬件，模拟 Modbus 帧上行 | +| **独立模式** | daemon 未启动时 edge 仍可运行，后台重连 uplink | + +--- + +## 2. 与 daemon 的职责边界（双域） + +``` +┌─────────────────────────────────────────────────────────────────┐ +│ edge（物理域） daemon（逻辑域） │ +├─────────────────────────────────────────────────────────────────┤ +│ config/edge/ config/daemon/ │ +│ edge.json daemon_profile.json │ +│ physical_channels.json logical_points.json │ +│ physical_devices.json bindings_global.json │ +│ bindings_local.json（可选） topic_rule.json │ +│ │ +│ 读硬件 → RawPacket RawPacket → 插件解析 │ +│ 写硬件 ← physicalPointId → bindings → OPC UA 节点 │ +└─────────────────────────────────────────────────────────────────┘ +``` + +| 配置 | edge 是否加载 | 用途 | +|------|---------------|------| +| `physical_channels.json` | 是 | 串口 / TCP 通道参数 | +| `physical_devices.json` | 是 | 从站、寄存器、物理点 ID | +| `bindings_local.json` | 可选 | 下行写命令的**本地备用**解析（正常联调几乎不用） | +| `logical_points.json` | **否** | daemon 专用 | +| `bindings_global.json` | **否** | daemon 专用（上行绑定 + 下行路由） | + +edge 上行时**不做** binding 初筛：所有已配置物理点都会采集并上报；是否绑定 OPC 由 daemon 的 `bindings_global.json` 决定（未绑定点进入 `edge/unbound/raw/...`）。 + +--- + +## 3. 配置加载流程 + +### 3.1 入口文件 + +默认配置：`config/edge/edge.json`。子路径**相对 `edge.json` 所在目录**解析。 + +```json +{ + "edge": { + "edgeId": "edge01", + "deviceId": "EdgeDevice_01", + "daemonHost": "127.0.0.1", + "daemonPort": 9000, + "heartbeatPeriodMs": 1000, + "uplinkReconnectPeriodMs": 5000, + "physicalChannels": "physical_channels.json", + "physicalDevices": "physical_devices.json", + "bindingsLocal": "bindings_local.json" + } +} +``` + +| 字段 | 说明 | +|------|------| +| `edgeId` | 边缘实例 ID，uplink 注册与 daemon 路由用 | +| `deviceId` | 逻辑设备名（ACK 等场景） | +| `daemonHost` / `daemonPort` | 核心 uplink 地址 | +| `heartbeatPeriodMs` | 心跳周期（仅 uplink 已连接时发送） | +| `uplinkReconnectPeriodMs` | daemon 不可达时的重连间隔 | +| `physicalChannels` / `physicalDevices` | 物理层配置 | +| `bindingsLocal` | 可选；空则跳过 | +| `bindingsPatch` | 可选；热加载绑定补丁 | + +### 3.2 启动时加载顺序（`EdgeRuntime::init`） + +``` +1. EdgeConfig.loadFromFile(--config) +2. PhysicalRegistry.loadChannels(physical_channels.json) +3. PhysicalRegistry.loadDevices(physical_devices.json, edgeId) + → 物理点 ID 自动加前缀：edge01/phy_modbus_s1_r40001 +4. [可选] BindingRegistry.loadLocal(bindings_local.json) +5. UplinkClient + CommandExecutor 初始化 +6. DeviceFactory.createIngress → ModbusIngress 或 MockModbusIngress +7. ModbusChannelManager.openChannel（逐通道 libmodbus 连接） +8. uplink.connect()；成功则 sendEdgeRegister(edgeId) + → 失败仅 WARN，edge 独立运行并后台重连 +``` + +### 3.3 物理通道（`physical_channels.json`） + +```json +{ + "channels": [ + { + "channelId": "serial_0", + "protocol": "modbus", + "transport": "rtu", + "port": "/dev/ttyS3", + "baudRate": 9600, + "parity": "N", + "dataBits": 8, + "stopBits": 1, + "pollIntervalMs": 500 + } + ] +} +``` + +| 字段 | 说明 | +|------|------| +| `transport` | `rtu`（串口）或 `tcp`（Modbus TCP，需 `host` + `tcpPort`） | +| `pollIntervalMs` | 该通道下所有点的轮询周期（当前整通道统一） | + +### 3.4 物理设备与点（`physical_devices.json`） + +```json +{ + "devices": [ + { + "physicalDeviceId": "phy_modbus_slave_1", + "channelId": "serial_0", + "protocol": "modbus", + "slaveId": 1, + "points": [ + { + "physicalPointId": "phy_modbus_s1_r40001", + "register": 40001, + "function": 3, + "scale": 0.1, + "signalType": "AI", + "writable": false + } + ] + } + ] +} +``` + +运行时 `physicalPointId` 会变为 `{edgeId}/{physicalPointId}`，作为 `RawPacket.sourceId` 上报。 + +### 3.5 热加载 + +| 信号 | 行为 | +|------|------| +| `SIGHUP` | `reloadPhysical()`：重载 channels + devices，重建 Modbus 连接 | +| `SIGHUP` | `reloadBindingsPatch()`：若配置了 `bindingsPatch` 则增量应用 | +| 文件变更 | `ConfigWatcher` 监视 `physical_channels`、`physical_devices`、`bindingsPatch` | + +--- + +## 4. 组件架构 ``` main.cpp - ├── MockModbusIngress (IIngressPort) ← 入站采集 - ├── UplinkClient (egress) ← TCP 上行/下行 - ├── DeviceFactory ← 运行时设备 ID 生成 - └── DeviceDiscovery ← 设备发现（Phase 1 stub） + └── EdgeRuntime + ├── PhysicalRegistry ← physical_*.json + ├── BindingRegistry (local) ← bindings_local.json（可选） + ├── ModbusChannelManager ← libmodbus RTU/TCP I/O + ├── IIngressPort + │ ├── ModbusIngress ← 正常模式 + │ └── MockModbusIngress ← --mock + ├── ModbusEgress ← 下行写寄存器 + ├── CommandExecutor ← ExecuteCommand → Modbus 写 / mock + └── UplinkClient ← TcpUplinkClient 封装 ``` -## 4. 核心组件 - ### 4.1 IIngressPort — 入站抽象 | 方法 | 说明 | @@ -35,102 +185,252 @@ main.cpp | `close()` | 关闭采集通道 | | `poll()` | 轮询，返回 `vector` | -**Phase 1 实现：** `MockModbusIngress` +**ModbusIngress**：按 `pollIntervalMs` 遍历 `pollPoints()`，经 `ModbusChannelManager` 读寄存器，组装 Modbus 0x03 响应帧作为 `payload`。 -- 生成递增的 Modbus RTU 0x03 响应帧 -- 寄存器值按 0.1 缩放（与 `ModbusParser` 一致） -- `sourceId` 使用 CLI 指定的 objectRef +**MockModbusIngress**：无硬件，生成递增值模拟帧。 -### 4.2 UplinkClient — 上行客户端 +### 4.2 ModbusChannelManager — 硬件 I/O -对 `TcpUplinkClient` 的薄封装： +- 构建时链接 **libmodbus**（系统包或 `third_party/libmodbus` vendor） +- RTU：`modbus_new_rtu` + `modbus_read_registers` / `modbus_write_register` +- TCP：`modbus_new_tcp`（`transport: tcp` 时） +- 无 libmodbus 时退化为模拟读写（编译宏 `SOFTBUS_HAS_LIBMODBUS=0`） + +> edge **不使用** daemon 侧的 `protocol_modbus.so` 插件；插件仅用于 daemon 解析上行 RawPacket。 + +### 4.3 UplinkClient — 上行/下行 | 方法 | 说明 | |------|------| -| `connect(host, port)` | 连接守护进程 | -| `sendRawPacket(RawPacket)` | 发送上行原始包 | -| `poll()` | 轮询接收下行命令 | +| `connect()` / `disconnect()` | TCP 连接 daemon | +| `sendEdgeRegister(edgeId)` | 注册边缘实例 | +| `sendHeartbeat(deviceId)` | 心跳 | +| `sendRawPacket(RawPacket)` | 上行原始包 | +| `sendAck(SoftbusAck)` | 写命令 ACK | +| `poll()` | 接收下行 `ExecuteCommand` | | `setCommandHandler(handler)` | 注册命令回调 | -### 4.3 DeviceFactory +协议详见 [UPLINK_WIRE_PROTOCOL.md](UPLINK_WIRE_PROTOCOL.md)。 -| 方法 | 说明 | -|------|------| -| `createRuntimeDevice()` | 生成 `runtime-xxxxxxxx` ID | +### 4.4 CommandExecutor — 下行写 -### 4.4 DeviceDiscovery（Phase 1 stub） +daemon 经 `bindings_global` 在命令中注入 `physicalPointId` + `edgeId`，edge 直接写 Modbus： -| 方法 | 说明 | Phase 1 行为 | -|------|------|-------------| -| `discover()` | 扫描可用设备 | 返回空列表 | +``` +ExecuteCommand.params.physicalPointId → PhysicalRegistry → ModbusEgress.write +``` -## 5. 主循环 +仅当命令**缺少** `physicalPointId` 时，才用 `bindings_local.json` 做 `logicalPointId → physicalPointId` 备用解析。 + +--- + +## 5. 数据流 + +### 5.1 上行（采集 → OPC） + +```mermaid +sequenceDiagram + participant HW as 现场设备 + participant Edge as softbus_edge + participant Daemon as softbus_daemon + participant OPC as OPC UA + + HW->>Edge: Modbus RTU/TCP 读寄存器 + Edge->>Edge: ModbusIngress.poll → RawPacket + Note over Edge: sourceId = edge01/phy_xxx
protocolTag = modbus + Edge->>Daemon: TCP rawPacket + edgeRegister/heartbeat + Daemon->>Daemon: protocol_modbus 解析帧 + Daemon->>Daemon: bindings_global → logicalPointId + Daemon->>Daemon: logical_points → objectRef + scale + Daemon->>OPC: UaModelBinder 写节点值 +``` + +逐步说明： + +1. **edge**：`ModbusIngress` 读硬件 → `RawPacket{ sourceId=physicalPointId, payload=Modbus帧 }` +2. **edge**：`UplinkClient.sendRawPacket` → TCP JSON 行 +3. **daemon**：`TransportBusBridge` 发布到总线 `edge/raw/{physicalPointId}` +4. **daemon**：`PipelineEngine` + `protocol_modbus` 插件解析 → 寄存器值 +5. **daemon**：`bindings_global` 查绑定 → `logical_points` 得 OPC 路径 → scale → 写 OPC UA + +未绑定物理点：daemon 发布到 `edge/unbound/raw/{physicalPointId}`，不写 OPC。 + +### 5.2 下行（OPC 写 → 设备） + +```mermaid +sequenceDiagram + participant OPC as OPC UA 客户端 + participant Daemon as softbus_daemon + participant Edge as softbus_edge + participant HW as 现场设备 + + OPC->>Daemon: ExecuteCommand (objectRef + value) + Daemon->>Daemon: bindings_global → physicalPointId + edgeId + Daemon->>Edge: TCP executeCommand + Edge->>Edge: CommandExecutor → ModbusEgress + Edge->>HW: libmodbus 写寄存器 + Edge->>Daemon: TCP ack +``` + +### 5.3 主循环（`main.cpp`） ```cpp -while (running) { - // 1. 采集 - auto packets = ingress.poll(); - for (auto& packet : packets) { - uplink.sendRawPacket(packet); - } +while (!shouldStop) { + ConfigWatcher.poll(); + if (SIGHUP) { reloadPhysical(); reloadBindingsPatch(); } - // 2. 接收命令 - uplink.poll(); + runtime.tick(); // uplink.poll + ingress.poll + sendRawPacket - // 3. 平台信号检查 - if (PlatformSignal::interrupted()) break; + if (uplinkConnected && heartbeatDue) + sendHeartbeat(edgeId); + + sleep(50ms); } ``` +--- + ## 6. CLI 参数 | 参数 | 默认值 | 说明 | |------|--------|------| -| `--daemon host:port` | `127.0.0.1:9000` | 守护进程地址 | -| `--object-ref path` | `DemoSite/DemoSystem/DemoAsset/SimModbusDevice/Temperature` | 模拟数据源对象路径 | +| `--config ` | `config/edge/edge.json` | 边缘入口配置 | +| `--mock` | 关闭 | 使用 MockModbusIngress，跳过 physical 配置与 libmodbus | +| `--object-ref ` | `DemoSite/.../MotorTemperature` | Mock 模式下的模拟 sourceId | -## 7. 数据流 +> 不再使用 `--daemon host:port`；uplink 地址由 `edge.json` 的 `daemonHost` / `daemonPort` 指定。 -### 7.1 上行 +--- -``` -MockModbusIngress.poll() - → RawPacket(sourceId=objectRef, protocolTag="modbus", payload=Modbus frame) - → UplinkClient.sendRawPacket() - → TcpUplinkClient → TCP JSON → Daemon -``` +## 7. 依赖 -### 7.2 下行 +| 依赖 | 用途 | +|------|------| +| `softbus_core` | RawPacket、Logger、PlatformSignal、ConfigWatcher | +| `softbus_registry` | BindingRegistry（仅 local 可选） | +| `softbus_transport` | TcpUplinkClient、UplinkWire | +| `softbus_api` | ExecuteCommand、SoftbusAck | +| **libmodbus** | Modbus RTU/TCP 硬件 I/O（vendor 或系统包） | +| asio | TcpUplinkClient 底层 | -``` -Daemon CommandDispatcher - → TCP JSON → TcpUplinkClient - → UplinkClient.poll() → commandHandler - → Phase 1: Logger 输出命令内容 -``` +--- -## 8. 依赖 +## 8. 构建与运行 -- `softbus_core`（RawPacket, DeviceIdentity, Logger, PlatformSignal） -- `softbus_transport`（TcpUplinkClient, UplinkWire） -- `softbus_api`（ExecuteCommand） -- `asio` +### 8.1 libmodbus -## 9. 运行示例 +构建时自动选择：系统 `libmodbus-dev` 优先，否则编译 `third_party/libmodbus` 静态库进 edge。 ```bash -# 确保守护进程已启动 -./build/bin/softbus_edge --daemon 127.0.0.1:9000 - -# 指定自定义对象路径 -./build/bin/softbus_edge \ - --daemon 127.0.0.1:9000 \ - --object-ref DemoSite/DemoSystem/DemoAsset/SimModbusDevice/Temperature +cmake -B build +cmake --build build --target softbus_edge ``` -## 10. Phase 2 规划 +串口权限：`sudo usermod -aG dialout $USER` -- 实现真实 `SerialIngress` / `CanIngress` 替换 Mock -- 下行 `ExecuteCommand` 写回设备（Modbus 写寄存器等） -- 设备发现与动态注册 -- 支持 SBUP 二进制协议 +### 8.2 运行示例 + +```bash +# 1. 先启动 daemon +./build/bin/softbus_daemon \ + --config config/daemon/daemon_profile.json \ + --logical-points config/daemon/logical_points.json \ + --bindings-global config/daemon/bindings_global.json + +# 2. 启动 edge（真实 Modbus） +./build/bin/softbus_edge --config config/edge/edge.json + +# 3. 无硬件 / 无 daemon 调试 +./build/bin/softbus_edge --config config/edge/edge.json --mock + +# 4. 热加载物理配置 +kill -HUP $(pidof softbus_edge) +``` + +更多配置说明见 [edge_device_setup.md](edge_device_setup.md)。 + +--- + +## 9. 未来扩展：CANopen 与以太网设备 + +当前仅 **Modbus RTU/TCP** 在 edge 侧有完整 ingress/egress。CANopen 与更多以太网协议按同一双域模型扩展。 + +### 9.1 目标架构 + +``` +physical_channels.json ──→ ChannelDriver（按 protocol 选择） +physical_devices.json ──→ PhysicalRegistry + │ + ┌───────────┼───────────┐ + ▼ ▼ ▼ + ModbusIngress CanIngress EthIngress（规划） + │ │ │ + └───────────┴───────────┘ + ▼ + RawPacket → UplinkClient +``` + +daemon 侧已有对应解析插件（`protocol_modbus`、`protocol_canopen`），edge 只需换 ingress/driver，**上行仍是 RawPacket + protocolTag**。 + +### 9.2 CANopen（规划） + +| 层级 | 计划 | +|------|------| +| **edge** | `CanIngress` + SocketCAN/`can0` 通道；SDO/PDO 读对象字典 → `RawPacket`，`protocolTag: "canopen"` | +| **配置** | `physical_channels` 增加 `transport: can`，`interface: can0`，`bitrate: 500000`；`physical_devices` 增加 `nodeId`、对象索引 | +| **daemon** | 已有 `protocol_canopen.so`（Framer/Parser 占位），绑定仍走 `bindings_global` | +| **EDS/DCF** | edge 侧加载 EDS 映射物理点 ↔ COB-ID/索引；daemon 不接收 EDS 全量 | + +### 9.3 以太网设备（规划） + +| 类型 | edge 侧 | 说明 | +|------|---------|------| +| **Modbus TCP** | 已支持 | `physical_channels` 设 `transport: tcp`，`host` + `tcpPort` | +| **裸 TCP / 自定义帧** | `EthIngress`（规划） | 按 channel 配置帧格式或插件化 driver | +| **UDP 组播** | 待定 | 适用于部分工业相机/传感器 | + +以太网通道示例（Modbus TCP，已可用）： + +```json +{ + "channelId": "eth_modbus_0", + "protocol": "modbus", + "transport": "tcp", + "host": "192.168.1.10", + "tcpPort": 502, + "pollIntervalMs": 200 +} +``` + +### 9.4 edge 侧待增强（Modbus 及通用） + +| 项 | 说明 | +|----|------| +| **ModbusPollScheduler** | 合并相邻寄存器读，减少串口占用 | +| **按点 pollInterval** | 不同信号类型不同采集周期 | +| **变化上报 / deadband** | 物理层过滤，减少 uplink 流量 | +| **DeviceDiscovery** | 总线扫描、动态写入 `physical_devices` | +| **多 ingress 并存** | 同一 edge 上 Modbus + CANopen 并行 | + +上述优化均在**物理域**完成，不改变 daemon 双域模型。 + +--- + +## 10. 约束（CONSTRAINTS） + +- edge **不得**直接操作 OPC UA 或加载 `protocol_*.so` 解析插件 +- edge **不得**加载 `logical_points.json` / `bindings_global.json` +- ingress 实现须遵循 `IIngressPort`，便于接入 CANopen/Ethernet driver +- 上行统一为 `RawPacket`；`sourceId` 必须是 **physicalPointId**（含 edgeId 前缀） +- daemon 不可达时 edge **必须**能独立启动并采集（uplink 后台重连） + +--- + +## 11. 相关文档 + +| 文档 | 内容 | +|------|------| +| [edge_device_setup.md](edge_device_setup.md) | 配置目录、双域、libmodbus、Mock | +| [UPLINK_WIRE_PROTOCOL.md](UPLINK_WIRE_PROTOCOL.md) | TCP JSON 线协议 | +| [softbus_daemon.md](softbus_daemon.md) | daemon 侧 Pipeline 与 OPC | +| [plugins.md](plugins.md) | protocol 插件（daemon 解析用） | diff --git a/src/plugins/protocol_modbus/src/ModbusParser.cpp b/src/plugins/protocol_modbus/src/ModbusParser.cpp index 55375e5..aeab41b 100644 --- a/src/plugins/protocol_modbus/src/ModbusParser.cpp +++ b/src/plugins/protocol_modbus/src/ModbusParser.cpp @@ -31,9 +31,8 @@ softbus::core::ParseResult ModbusParser::parseFrame(const uint8_t* data, std::si result.message.value = { {"protocol", "modbus"}, {"function", 0x03}, - {"register", 40001}, {"registerValue", registerValue}, - {"value", static_cast(registerValue) * 0.1} + {"value", static_cast(registerValue)} }; return result; } diff --git a/src/softbus_command/include/softbus_command/CommandService.h b/src/softbus_command/include/softbus_command/CommandService.h index 42d362e..41aba52 100644 --- a/src/softbus_command/include/softbus_command/CommandService.h +++ b/src/softbus_command/include/softbus_command/CommandService.h @@ -3,6 +3,7 @@ #include #include #include +#include #include #include @@ -17,6 +18,7 @@ namespace softbus::command { class CommandService { public: explicit CommandService(softbus::registry::PointRegistry* registry, + softbus::registry::BindingRegistry* bindings = nullptr, std::chrono::milliseconds timeout = std::chrono::milliseconds(5000)); void attachBus(softbus::bus::Bus* bus); @@ -28,6 +30,7 @@ private: std::string generateRequestId() const; softbus::registry::PointRegistry* registry_{nullptr}; + softbus::registry::BindingRegistry* bindings_{nullptr}; softbus::bus::Bus* bus_{nullptr}; softbus::bus::Bus::SubscriptionId ackSub_{0}; std::chrono::milliseconds timeout_; diff --git a/src/softbus_command/src/CommandService.cpp b/src/softbus_command/src/CommandService.cpp index 7f91919..de0e2f5 100644 --- a/src/softbus_command/src/CommandService.cpp +++ b/src/softbus_command/src/CommandService.cpp @@ -6,8 +6,10 @@ namespace softbus::command { CommandService::CommandService(softbus::registry::PointRegistry* registry, + softbus::registry::BindingRegistry* bindings, std::chrono::milliseconds timeout) : registry_(registry) + , bindings_(bindings) , timeout_(timeout) { } @@ -84,6 +86,14 @@ CommandContext CommandService::submit(const softbus::api::ExecuteCommand& comman softbus::bus::BusMessage msg; msg.type = softbus::bus::BusPayloadType::ExecuteCommand; msg.command = ctx.command; + if (bindings_) { + const auto logicalId = entry->deviceId + "/" + entry->pointId; + const auto route = bindings_->resolveRoute(logicalId); + if (route) { + msg.command.params["physicalPointId"] = route->physicalPointId; + msg.command.params["edgeId"] = route->edgeId; + } + } bus_->publish(topic, msg); { diff --git a/src/softbus_core/CMakeLists.txt b/src/softbus_core/CMakeLists.txt index 074ad27..f7d9940 100644 --- a/src/softbus_core/CMakeLists.txt +++ b/src/softbus_core/CMakeLists.txt @@ -6,6 +6,7 @@ add_library(softbus_core STATIC src/alignment/TimeAligner.cpp src/plugin/PluginLoader.cpp src/config/ProfileLoader.cpp + src/config/ConfigWatcher.cpp src/platform/PlatformSignal.cpp src/identity/DeviceIdentity.cpp src/models/ObjectModel.cpp diff --git a/src/softbus_core/include/softbus_core/config/ConfigWatcher.h b/src/softbus_core/include/softbus_core/config/ConfigWatcher.h new file mode 100644 index 0000000..6083051 --- /dev/null +++ b/src/softbus_core/include/softbus_core/config/ConfigWatcher.h @@ -0,0 +1,23 @@ +#pragma once + +#include +#include + +namespace softbus::core { + +class ConfigWatcher { +public: + void addWatch(const std::string& path); + void poll(); + +private: + struct WatchEntry { + std::string path; + long mtime{0}; + }; + + std::vector watches_; + long readMtime(const std::string& path) const; +}; + +} // namespace softbus::core diff --git a/src/softbus_core/include/softbus_core/models/SoftbusTypes.h b/src/softbus_core/include/softbus_core/models/SoftbusTypes.h index 33ac899..50c7c15 100644 --- a/src/softbus_core/include/softbus_core/models/SoftbusTypes.h +++ b/src/softbus_core/include/softbus_core/models/SoftbusTypes.h @@ -60,6 +60,7 @@ struct SoftbusEvent { std::string eventType; std::string message; std::uint64_t timestampNs{0}; + std::uint64_t lastHeartbeatNs{0}; nlohmann::json toJson() const; static SoftbusEvent fromJson(const nlohmann::json& j); diff --git a/src/softbus_core/include/softbus_core/platform/PlatformSignal.h b/src/softbus_core/include/softbus_core/platform/PlatformSignal.h index f32f538..80cc5ed 100644 --- a/src/softbus_core/include/softbus_core/platform/PlatformSignal.h +++ b/src/softbus_core/include/softbus_core/platform/PlatformSignal.h @@ -12,9 +12,13 @@ public: static void install(Handler handler); static bool shouldStop(); static void requestStop(); + static bool reloadRequested(); + static void requestReload(); + static void clearReload(); private: static std::atomic stopRequested_; + static std::atomic reloadRequested_; }; } // namespace softbus::core diff --git a/src/softbus_core/src/config/ConfigWatcher.cpp b/src/softbus_core/src/config/ConfigWatcher.cpp new file mode 100644 index 0000000..a4907d5 --- /dev/null +++ b/src/softbus_core/src/config/ConfigWatcher.cpp @@ -0,0 +1,41 @@ +#include + +#include + +#include + +namespace softbus::core { + +void ConfigWatcher::addWatch(const std::string& path) +{ + WatchEntry entry; + entry.path = path; + entry.mtime = readMtime(path); + watches_.push_back(std::move(entry)); +} + +long ConfigWatcher::readMtime(const std::string& path) const +{ + struct stat st{}; + if (stat(path.c_str(), &st) != 0) { + return 0; + } + return st.st_mtime; +} + +void ConfigWatcher::poll() +{ + for (auto& watch : watches_) { + const long mtime = readMtime(watch.path); + if (mtime != 0 && watch.mtime != 0 && mtime != watch.mtime) { + watch.mtime = mtime; + PlatformSignal::requestReload(); + return; + } + if (watch.mtime == 0 && mtime != 0) { + watch.mtime = mtime; + } + } +} + +} // namespace softbus::core diff --git a/src/softbus_core/src/models/SoftbusTypes.cpp b/src/softbus_core/src/models/SoftbusTypes.cpp index 20dc73c..886af01 100644 --- a/src/softbus_core/src/models/SoftbusTypes.cpp +++ b/src/softbus_core/src/models/SoftbusTypes.cpp @@ -68,12 +68,16 @@ SoftbusAck SoftbusAck::fromJson(const nlohmann::json& j) nlohmann::json SoftbusEvent::toJson() const { - return nlohmann::json{ + nlohmann::json j{ {"deviceId", deviceId}, {"eventType", eventType}, {"message", message}, - {"timestamp", timestampNs} + {"timestamp", timestampNs}, }; + if (lastHeartbeatNs != 0) { + j["lastHeartbeat"] = lastHeartbeatNs; + } + return j; } SoftbusEvent SoftbusEvent::fromJson(const nlohmann::json& j) @@ -83,6 +87,7 @@ SoftbusEvent SoftbusEvent::fromJson(const nlohmann::json& j) ev.eventType = j.value("eventType", ""); ev.message = j.value("message", ""); ev.timestampNs = j.value("timestamp", static_cast(0)); + ev.lastHeartbeatNs = j.value("lastHeartbeat", static_cast(0)); return ev; } diff --git a/src/softbus_core/src/platform/PlatformSignal.cpp b/src/softbus_core/src/platform/PlatformSignal.cpp index da91cd4..066eff4 100644 --- a/src/softbus_core/src/platform/PlatformSignal.cpp +++ b/src/softbus_core/src/platform/PlatformSignal.cpp @@ -9,28 +9,31 @@ namespace softbus::core { std::atomic PlatformSignal::stopRequested_{false}; +std::atomic PlatformSignal::reloadRequested_{false}; #ifdef SOFTBUS_PLATFORM_WINDOWS namespace { - BOOL WINAPI consoleHandler(DWORD signal) - { - if (signal == CTRL_C_EVENT || signal == CTRL_BREAK_EVENT || signal == CTRL_CLOSE_EVENT) { - PlatformSignal::requestStop(); - return TRUE; - } - return FALSE; +BOOL WINAPI consoleHandler(DWORD signal) +{ + if (signal == CTRL_C_EVENT || signal == CTRL_BREAK_EVENT || signal == CTRL_CLOSE_EVENT) { + PlatformSignal::requestStop(); + return TRUE; } + return FALSE; } +} // namespace #endif void PlatformSignal::install(Handler handler) { stopRequested_.store(false); + reloadRequested_.store(false); #ifdef SOFTBUS_PLATFORM_WINDOWS SetConsoleCtrlHandler(consoleHandler, TRUE); #else std::signal(SIGINT, [](int) { requestStop(); }); std::signal(SIGTERM, [](int) { requestStop(); }); + std::signal(SIGHUP, [](int) { requestReload(); }); #endif (void)handler; } @@ -45,4 +48,19 @@ void PlatformSignal::requestStop() stopRequested_.store(true); } +bool PlatformSignal::reloadRequested() +{ + return reloadRequested_.load(); +} + +void PlatformSignal::requestReload() +{ + reloadRequested_.store(true); +} + +void PlatformSignal::clearReload() +{ + reloadRequested_.store(false); +} + } // namespace softbus::core diff --git a/src/softbus_daemon/src/CoreService.cpp b/src/softbus_daemon/src/CoreService.cpp index 56827b4..c87761d 100644 --- a/src/softbus_daemon/src/CoreService.cpp +++ b/src/softbus_daemon/src/CoreService.cpp @@ -69,10 +69,7 @@ std::string commandResultJson(const softbus::command::CommandContext& ctx) } // namespace -CoreService::CoreService() - : commandService_(®istry_) -{ -} +CoreService::CoreService() = default; CoreService::~CoreService() { @@ -111,94 +108,114 @@ std::string CoreService::resolvePluginPath(const std::string& pluginName, } return fs::absolute(profileDir / "lib" / pluginFileNames(pluginName).front()).string(); } -// 主函数启动入口 -// profilePath: 配置文件路径 -// pointTablePath: 点表文件路径 -// return: 是否启动成功 -bool CoreService::start(const std::string& profilePath, const std::string& pointTablePath) +bool CoreService::loadLogicalConfig(const std::string& logicalPointsPath) { - // 加载配置文件 + const fs::path configDir = fs::path(logicalPointsPath).parent_path(); + registry_.loadTopicRules((configDir / "topic_rule.json").string()); + if (registry_.loadLogicalPoints(logicalPointsPath)) { + return true; + } + return registry_.loadPointTable(logicalPointsPath); +} + +bool CoreService::start(const std::string& profilePath, + const std::string& logicalPointsPath, + const std::string& bindingsGlobalPath) +{ + logicalPointsPath_ = logicalPointsPath; + bindingsGlobalPath_ = bindingsGlobalPath; + if (!profile_.loadFromFile(profilePath)) { SB_LOG_ERROR("CoreService", "Failed to load profile: ", profilePath); return false; } - const fs::path configDir = fs::path(pointTablePath).parent_path(); - // 加载主题规则文件 - registry_.loadTopicRules((configDir / "topic_rule.json").string()); - if (!registry_.loadPointTable(pointTablePath)) { - SB_LOG_ERROR("CoreService", "Failed to load point table: ", pointTablePath); + if (!loadLogicalConfig(logicalPointsPath)) { + SB_LOG_ERROR("CoreService", "Failed to load logical points: ", logicalPointsPath); return false; } model_ = registry_.tree(); + if (!bindingsGlobalPath.empty()) { + if (!bindingRegistry_.loadGlobal(bindingsGlobalPath)) { + SB_LOG_ERROR("CoreService", "Failed to load bindings: ", bindingsGlobalPath); + return false; + } + configWatcher_.addWatch(bindingsGlobalPath); + } + configWatcher_.addWatch(logicalPointsPath); + const std::string pluginPath = resolvePluginPath(profile_.plugins.modbus, profilePath); if (!modbusPlugin_.load(pluginPath)) { SB_LOG_ERROR("CoreService", "Failed to load modbus plugin: ", pluginPath); return false; } - // 加载点表文件 + auto& bus = softbus::bus::Bus::instance(); - // 将消息总线绑定到命令服务、存储服务和心跳监控服务 - commandService_.attachBus(&bus); + commandService_ = std::make_unique( + ®istry_, &bindingRegistry_); + commandService_->attachBus(&bus); storageService_.attachBus(&bus); heartbeatMonitor_.attachBus(&bus); - // 创建OPC UA安全配置 softbus::opcua::SecurityConfig security; security.enableTls = profile_.security.tlsEnabled; security.certPath = profile_.security.certPath; security.keyPath = profile_.security.keyPath; - // 创建OPC UA服务引擎 uaEngine_ = std::make_unique(); - // 启动OPC UA服务 if (!uaEngine_->start(profile_.opcua.port, profile_.opcua.applicationName, security)) { return false; } - // 创建OPC UA模型绑定器 uaBinder_ = std::make_unique(uaEngine_->nativeServer()); - // 创建OPC UA方法绑定器 uaMethodBinder_ = std::make_unique(uaEngine_->nativeServer()); - // 创建OPC UA事件桥接器 uaEventBridge_ = std::make_unique(uaEngine_->nativeServer()); - // 将消息总线绑定到OPC UA模型绑定器、OPC UA方法绑定器和OPC UA事件桥接器 uaBinder_->attachBus(&bus); uaEventBridge_->attachBus(&bus); - // 创建OPC UA地址空间构建器 softbus::opcua::UaAddressSpaceBuilder builder(uaEngine_->nativeServer()); - // 构建OPC UA地址空间 if (!builder.buildFromModel(model_, uaBinder_.get())) { SB_LOG_ERROR("CoreService", "Failed to build OPC UA address space"); return false; } - // 创建数据处理管道 pipeline_ = std::make_unique( - &modbusPlugin_, uaBinder_.get(), ®istry_); + &modbusPlugin_, uaBinder_.get(), ®istry_, &bindingRegistry_); pipeline_->attachBus(&bus); - // 创建TCP上行监听服务器 + uplink_ = std::make_unique( io_, profile_.uplink.listenHost, profile_.uplink.listenPort); - // 将消息总线绑定到TCP上行监听服务器 transportBridge_.attach(&bus, uplink_.get()); - // 绑定OPC UA方法执行命令回调 - // 将OPC UA方法执行命令回调绑定到消息总线 + uaMethodBinder_->bindExecuteCommand([this](const softbus::api::ExecuteCommand& cmd) { - const auto ctx = commandService_.submit(cmd, true); + const auto ctx = commandService_->submit(cmd, true); return commandResultJson(ctx); }); - + uplink_->start(); ioThread_ = std::thread([this]() { io_.run(); }); running_ = true; - SB_LOG_INFO("CoreService", "Daemon started (bus+registry+command pipeline)"); + SB_LOG_INFO("CoreService", "Daemon started (dual-domain+bindings)"); return true; } +void CoreService::reloadConfig() +{ + softbus::registry::PointRegistry previous = registry_; + if (!registry_.reloadPointTable(logicalPointsPath_)) { + SB_LOG_WARN("CoreService", "Logical points reload failed"); + return; + } + model_ = registry_.tree(); + if (!bindingsGlobalPath_.empty()) { + bindingRegistry_.reloadGlobal(bindingsGlobalPath_); + } + const auto added = registry_.diff(previous); + SB_LOG_INFO("CoreService", "Config reloaded, new logical points=", added.size()); +} + void CoreService::stop() { if (!running_) { @@ -227,6 +244,11 @@ void CoreService::run(bool selfTest) const auto selfTestStart = std::chrono::steady_clock::now(); while (!softbus::core::PlatformSignal::shouldStop()) { + configWatcher_.poll(); + if (softbus::core::PlatformSignal::reloadRequested()) { + reloadConfig(); + softbus::core::PlatformSignal::clearReload(); + } if (uaEngine_) { uaEngine_->iterate(); } @@ -238,7 +260,7 @@ void CoreService::run(bool selfTest) cmd.objectRef = "DemoSite/DemoSystem/DemoAsset/EdgeDevice_01/ValveOpening"; cmd.command = "write"; cmd.params = {{"value", 60.0}}; - const auto ctx = commandService_.submit(cmd, true); + const auto ctx = commandService_->submit(cmd, true); SB_LOG_INFO("CoreService", "self-test command state=", softbus::command::commandStateToString(ctx.state), " executedValue=", ctx.executedValue); diff --git a/src/softbus_daemon/src/main.cpp b/src/softbus_daemon/src/main.cpp index 988cfd7..e0ea86b 100644 --- a/src/softbus_daemon/src/main.cpp +++ b/src/softbus_daemon/src/main.cpp @@ -5,23 +5,27 @@ int main(int argc, char* argv[]) { - std::string profilePath = "config/daemon_profile.json"; - std::string pointTablePath = "config/point_table.json"; + std::string profilePath = "config/daemon/daemon_profile.json"; + std::string logicalPointsPath = "config/daemon/logical_points.json"; + std::string bindingsGlobalPath = "config/daemon/bindings_global.json"; bool selfTest = false; for (int i = 1; i < argc; ++i) { const std::string arg = argv[i]; if (arg == "--config" && i + 1 < argc) { profilePath = argv[++i]; - } else if ((arg == "--point-table" || arg == "--model") && i + 1 < argc) { - pointTablePath = argv[++i]; + } else if ((arg == "--logical-points" || arg == "--point-table" || arg == "--model") + && i + 1 < argc) { + logicalPointsPath = argv[++i]; + } else if (arg == "--bindings-global" && i + 1 < argc) { + bindingsGlobalPath = argv[++i]; } else if (arg == "--self-test") { selfTest = true; } } softbus::daemon::CoreService service; - if (!service.start(profilePath, pointTablePath)) { + if (!service.start(profilePath, logicalPointsPath, bindingsGlobalPath)) { std::cerr << "Failed to start softbus_daemon" << std::endl; return 1; } diff --git a/src/softbus_edge/CMakeLists.txt b/src/softbus_edge/CMakeLists.txt index d64b7f2..633481b 100644 --- a/src/softbus_edge/CMakeLists.txt +++ b/src/softbus_edge/CMakeLists.txt @@ -1,12 +1,18 @@ +# role: edge 用于采集物理设备数据，并将其转换为软总线数据模型 softbus_package(softbus_edge) add_executable(softbus_edge src/main.cpp src/MockModbusIngress.cpp + src/ModbusIngress.cpp + src/ModbusChannel.cpp + src/ModbusEgress.cpp + src/PhysicalRegistry.cpp + src/EdgeConfig.cpp + src/EdgeRuntime.cpp src/UplinkClient.cpp src/DeviceFactory.cpp src/DeviceDiscovery.cpp - src/EdgeConfig.cpp src/ShadowChannelStore.cpp src/CommandExecutor.cpp ) @@ -20,4 +26,6 @@ target_link_libraries(softbus_edge PRIVATE softbus_api asio ) +# 添加 libmodbus 库 +softbus_apply_libmodbus(softbus_edge) softbus_apply_platform_libs(softbus_edge) diff --git a/src/softbus_edge/include/softbus_edge/config/EdgeConfig.h b/src/softbus_edge/include/softbus_edge/config/EdgeConfig.h index 6a15fcc..1cc5d7d 100644 --- a/src/softbus_edge/include/softbus_edge/config/EdgeConfig.h +++ b/src/softbus_edge/include/softbus_edge/config/EdgeConfig.h @@ -6,11 +6,16 @@ namespace softbus::edge { struct EdgeConfig { + std::string edgeId{"edge01"}; std::string deviceId{"EdgeDevice_01"}; std::string daemonHost{"127.0.0.1"}; uint16_t daemonPort{9000}; int heartbeatPeriodMs{1000}; - std::string pointTablePath{"config/point_table.json"}; + int uplinkReconnectPeriodMs{5000}; + std::string physicalChannelsPath{"physical_channels.json"}; + std::string physicalDevicesPath{"physical_devices.json"}; + std::string bindingsLocalPath{"bindings_local.json"}; + std::string bindingsPatchPath; bool loadFromFile(const std::string& path); }; diff --git a/src/softbus_edge/include/softbus_edge/config/PhysicalRegistry.h b/src/softbus_edge/include/softbus_edge/config/PhysicalRegistry.h new file mode 100644 index 0000000..58b5470 --- /dev/null +++ b/src/softbus_edge/include/softbus_edge/config/PhysicalRegistry.h @@ -0,0 +1,42 @@ +#pragma once + +#include + +#include +#include +#include +#include + +namespace softbus::edge { + +struct ResolvedPhysicalPoint { + PhysicalPointDef point; + PhysicalDeviceDef device; + PhysicalChannelDef channel; +}; + +class PhysicalRegistry { +public: + bool loadChannels(const std::string& path); + bool loadDevices(const std::string& path, const std::string& edgeIdPrefix = {}); + bool reloadChannels(const std::string& path); + bool reloadDevices(const std::string& path, const std::string& edgeIdPrefix = {}); + + const std::vector& channels() const { return channels_; } + const std::vector& devices() const { return devices_; } + const std::vector& pollPoints() const { return pollPoints_; } + + std::optional resolvePoint(const std::string& physicalPointId) const; + std::optional findChannel(const std::string& channelId) const; + +private: + void rebuildPollIndex(const std::string& edgeIdPrefix); + + std::vector channels_; + std::vector devices_; + std::vector pollPoints_; + std::unordered_map channelIndex_; + std::unordered_map pointIndex_; +}; + +} // namespace softbus::edge diff --git a/src/softbus_edge/include/softbus_edge/config/PhysicalTypes.h b/src/softbus_edge/include/softbus_edge/config/PhysicalTypes.h new file mode 100644 index 0000000..d4b8f42 --- /dev/null +++ b/src/softbus_edge/include/softbus_edge/config/PhysicalTypes.h @@ -0,0 +1,41 @@ +#pragma once + +#include +#include +#include + +namespace softbus::edge { + +struct PhysicalChannelDef { + std::string channelId; + std::string protocol{"modbus"}; + std::string transport{"rtu"}; + std::string port; + std::string host; + uint16_t tcpPort{502}; + int baudRate{9600}; + char parity{'N'}; + int dataBits{8}; + int stopBits{1}; + int pollIntervalMs{500}; +}; + +struct PhysicalPointDef { + std::string physicalPointId; + int registerAddress{0}; + int functionCode{3}; + int writeFunctionCode{6}; + double scale{1.0}; + std::string signalType{"AI"}; + bool writable{false}; +}; + +struct PhysicalDeviceDef { + std::string physicalDeviceId; + std::string channelId; + std::string protocol{"modbus"}; + int slaveId{1}; + std::vector points; +}; + +} // namespace softbus::edge diff --git a/src/softbus_edge/include/softbus_edge/driver/ModbusChannel.h b/src/softbus_edge/include/softbus_edge/driver/ModbusChannel.h new file mode 100644 index 0000000..95267ad --- /dev/null +++ b/src/softbus_edge/include/softbus_edge/driver/ModbusChannel.h @@ -0,0 +1,32 @@ +#pragma once + +#include + +#include +#include +#include +#include + +namespace softbus::edge { + +class ModbusChannelManager { +public: + bool openChannel(const PhysicalChannelDef& channel); + void closeChannel(const std::string& channelId); + void closeAll(); + std::optional readRegister(const ResolvedPhysicalPoint& point); + bool writeRegister(const ResolvedPhysicalPoint& point, uint16_t rawValue); + +private: + struct ChannelHandle { + PhysicalChannelDef config; +#if defined(SOFTBUS_HAS_LIBMODBUS) && SOFTBUS_HAS_LIBMODBUS + void* ctx{nullptr}; +#endif + }; + + std::unordered_map channels_; + int modbusAddress(int registerAddress) const; +}; + +} // namespace softbus::edge diff --git a/src/softbus_edge/include/softbus_edge/egress/CommandExecutor.h b/src/softbus_edge/include/softbus_edge/egress/CommandExecutor.h index 0504ef1..7f654de 100644 --- a/src/softbus_edge/include/softbus_edge/egress/CommandExecutor.h +++ b/src/softbus_edge/include/softbus_edge/egress/CommandExecutor.h @@ -1,26 +1,38 @@ #pragma once #include +#include +#include #include -#include #include +#include namespace softbus::edge { class CommandExecutor { public: - CommandExecutor(softbus::registry::PointRegistry* registry, - ShadowChannelStore* store, + CommandExecutor(PhysicalRegistry* physicalRegistry, + ModbusEgress* modbusEgress, + ShadowChannelStore* mockStore, + softbus::registry::BindingRegistry* localBindings, UplinkClient* uplink, - std::string deviceId); + std::string deviceId, + bool useMock = false); softbus::core::SoftbusAck execute(const softbus::api::ExecuteCommand& command); private: - softbus::registry::PointRegistry* registry_{nullptr}; - ShadowChannelStore* store_{nullptr}; + void fillAckIdentity(softbus::core::SoftbusAck& ack, + const softbus::api::ExecuteCommand& command, + const std::string& physicalPointId) const; + + PhysicalRegistry* physicalRegistry_{nullptr}; + ModbusEgress* modbusEgress_{nullptr}; + ShadowChannelStore* mockStore_{nullptr}; + softbus::registry::BindingRegistry* localBindings_{nullptr}; UplinkClient* uplink_{nullptr}; std::string deviceId_; + bool useMock_{false}; }; } // namespace softbus::edge diff --git a/src/softbus_edge/include/softbus_edge/egress/ModbusEgress.h b/src/softbus_edge/include/softbus_edge/egress/ModbusEgress.h new file mode 100644 index 0000000..b0f50f0 --- /dev/null +++ b/src/softbus_edge/include/softbus_edge/egress/ModbusEgress.h @@ -0,0 +1,23 @@ +#pragma once + +#include +#include + +#include +#include + +namespace softbus::edge { + +class ModbusEgress { +public: + explicit ModbusEgress(PhysicalRegistry* registry, ModbusChannelManager* channels); + + bool writeByPhysicalPointId(const std::string& physicalPointId, double engineeringValue, + int* rawOut = nullptr); + +private: + PhysicalRegistry* registry_{nullptr}; + ModbusChannelManager* channels_{nullptr}; +}; + +} // namespace softbus::edge diff --git a/src/softbus_edge/include/softbus_edge/egress/ShadowChannelStore.h b/src/softbus_edge/include/softbus_edge/egress/ShadowChannelStore.h index 96c53f1..b00bff5 100644 --- a/src/softbus_edge/include/softbus_edge/egress/ShadowChannelStore.h +++ b/src/softbus_edge/include/softbus_edge/egress/ShadowChannelStore.h @@ -1,7 +1,5 @@ #pragma once -#include - #include #include #include @@ -10,12 +8,11 @@ namespace softbus::edge { class ShadowChannelStore { public: - bool write(const softbus::registry::PointTableEntry& entry, double engineeringValue); - std::optional read(const std::string& physicalChannel) const; - int rawRegisterValue(const softbus::registry::PointTableEntry& entry, double engineeringValue) const; + bool write(const std::string& physicalPointId, double engineeringValue); + std::optional read(const std::string& physicalPointId) const; private: - std::unordered_map channels_; + std::unordered_map points_; }; } // namespace softbus::edge diff --git a/src/softbus_edge/include/softbus_edge/egress/UplinkClient.h b/src/softbus_edge/include/softbus_edge/egress/UplinkClient.h index 593c2cb..d8262a7 100644 --- a/src/softbus_edge/include/softbus_edge/egress/UplinkClient.h +++ b/src/softbus_edge/include/softbus_edge/egress/UplinkClient.h @@ -18,9 +18,11 @@ public: bool connect(); void disconnect(); + bool isConnected() const; bool sendRawPacket(const softbus::core::RawPacket& packet); bool sendAck(const softbus::core::SoftbusAck& ack); bool sendHeartbeat(const std::string& deviceId); + bool sendEdgeRegister(const std::string& edgeId); void setCommandHandler(std::function handler); void poll(); diff --git a/src/softbus_edge/include/softbus_edge/factory/DeviceFactory.h b/src/softbus_edge/include/softbus_edge/factory/DeviceFactory.h index 68a644e..aa25ddc 100644 --- a/src/softbus_edge/include/softbus_edge/factory/DeviceFactory.h +++ b/src/softbus_edge/include/softbus_edge/factory/DeviceFactory.h @@ -1,5 +1,14 @@ #pragma once +#include +#include +#include +#include +#include +#include +#include + +#include #include namespace softbus::edge { @@ -7,6 +16,12 @@ namespace softbus::edge { class DeviceFactory { public: std::string createRuntimeDeviceId(); + + std::unique_ptr createIngress(const EdgeConfig& config, + PhysicalRegistry* registry, + ModbusChannelManager* channels, + bool useMock, + const std::string& mockObjectRef); }; } // namespace softbus::edge diff --git a/src/softbus_edge/include/softbus_edge/ingress/ModbusIngress.h b/src/softbus_edge/include/softbus_edge/ingress/ModbusIngress.h new file mode 100644 index 0000000..93763d8 --- /dev/null +++ b/src/softbus_edge/include/softbus_edge/ingress/ModbusIngress.h @@ -0,0 +1,30 @@ +#pragma once + +#include +#include +#include + +#include +#include + +namespace softbus::edge { + +class ModbusIngress : public IIngressPort { +public: + ModbusIngress(PhysicalRegistry* registry, ModbusChannelManager* channels, std::string edgeIdPrefix); + + bool open() override; + void close() override; + std::vector poll() override; + +private: + std::vector buildModbusFrame(int slaveId, uint16_t registerValue) const; + + PhysicalRegistry* registry_{nullptr}; + ModbusChannelManager* channels_{nullptr}; + std::string edgeIdPrefix_; + bool open_{false}; + std::chrono::steady_clock::time_point lastPoll_{}; +}; + +} // namespace softbus::edge diff --git a/src/softbus_edge/include/softbus_edge/runtime/EdgeRuntime.h b/src/softbus_edge/include/softbus_edge/runtime/EdgeRuntime.h new file mode 100644 index 0000000..1b5b2e4 --- /dev/null +++ b/src/softbus_edge/include/softbus_edge/runtime/EdgeRuntime.h @@ -0,0 +1,51 @@ +#pragma once + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include + +namespace softbus::edge { + +class EdgeRuntime { +public: + bool init(const EdgeConfig& config, bool useMock, const std::string& mockObjectRef); + void reloadPhysical(); + void reloadBindingsPatch(); + void shutdown(); + + void tick(); + UplinkClient* uplink() { return uplink_.get(); } + bool isUplinkConnected() const; + +private: + void tickUplink(); + void pollIngress(); + EdgeConfig config_; + bool useMock_{false}; + std::string mockObjectRef_; + + softbus::registry::BindingRegistry localBindings_; + PhysicalRegistry physicalRegistry_; + ModbusChannelManager modbusChannels_; + ModbusEgress modbusEgress_{&physicalRegistry_, &modbusChannels_}; + ShadowChannelStore mockStore_; + std::unique_ptr uplink_; + std::unique_ptr ingress_; + std::unique_ptr executor_; + DeviceFactory factory_; + std::chrono::steady_clock::time_point lastUplinkReconnectAttempt_{}; +}; + + +} // namespace softbus::edge diff --git a/src/softbus_edge/src/CommandExecutor.cpp b/src/softbus_edge/src/CommandExecutor.cpp index 703ecdd..9535323 100644 --- a/src/softbus_edge/src/CommandExecutor.cpp +++ b/src/softbus_edge/src/CommandExecutor.cpp @@ -1,42 +1,49 @@ #include +#include #include namespace softbus::edge { -CommandExecutor::CommandExecutor(softbus::registry::PointRegistry* registry, - ShadowChannelStore* store, +CommandExecutor::CommandExecutor(PhysicalRegistry* physicalRegistry, + ModbusEgress* modbusEgress, + ShadowChannelStore* mockStore, + softbus::registry::BindingRegistry* localBindings, UplinkClient* uplink, - std::string deviceId) - : registry_(registry) - , store_(store) + std::string deviceId, + bool useMock) + : physicalRegistry_(physicalRegistry) + , modbusEgress_(modbusEgress) + , mockStore_(mockStore) + , localBindings_(localBindings) , uplink_(uplink) , deviceId_(std::move(deviceId)) + , useMock_(useMock) { } +void CommandExecutor::fillAckIdentity(softbus::core::SoftbusAck& ack, + const softbus::api::ExecuteCommand& command, + const std::string& physicalPointId) const +{ + if (!command.objectRef.empty()) { + const auto ref = softbus::core::ObjectRef::fromPath(command.objectRef); + if (ref.isValid() && ref.hasPoint()) { + ack.deviceId = ref.device; + ack.pointId = ref.point; + return; + } + } + ack.deviceId = deviceId_; + ack.pointId = physicalPointId; +} + softbus::core::SoftbusAck CommandExecutor::execute(const softbus::api::ExecuteCommand& command) { softbus::core::SoftbusAck ack; ack.requestId = command.requestId; ack.traceId = command.traceId; - if (!registry_ || !store_) { - ack.success = false; - ack.error = "executor not initialized"; - return ack; - } - - const auto entry = registry_->findByObjectRef(command.objectRef); - if (!entry) { - ack.success = false; - ack.error = "unknown objectRef on edge"; - return ack; - } - - ack.deviceId = entry->deviceId; - ack.pointId = entry->pointId; - if (command.command != "write") { ack.success = false; ack.error = "unsupported command"; @@ -44,21 +51,66 @@ softbus::core::SoftbusAck CommandExecutor::execute(const softbus::api::ExecuteCo } const double value = command.params.value("value", 0.0); - std::string error; - if (!registry_->validateWriteValue(*entry, value, &error)) { + std::string physicalPointId = command.params.value("physicalPointId", ""); + + if (physicalPointId.empty() && localBindings_) { + const auto logicalId = command.params.value("logicalPointId", ""); + if (!logicalId.empty()) { + const auto resolved = localBindings_->resolvePhysicalLocal(logicalId); + if (resolved) { + physicalPointId = *resolved; + } + } + } + + if (physicalPointId.empty()) { ack.success = false; - ack.error = error; + ack.error = "no physicalPointId resolved (edge does not load logical point table)"; return ack; } - store_->write(*entry, value); + if (useMock_ && mockStore_) { + mockStore_->write(physicalPointId, value); + fillAckIdentity(ack, command, physicalPointId); + ack.executedValue = value; + ack.success = true; + if (uplink_) { + uplink_->sendAck(ack); + } + return ack; + } + + if (!modbusEgress_ || !physicalRegistry_) { + ack.success = false; + ack.error = "modbus egress not initialized"; + return ack; + } + + const auto resolved = physicalRegistry_->resolvePoint(physicalPointId); + if (!resolved) { + ack.success = false; + ack.error = "unknown physicalPointId"; + return ack; + } + + if (!resolved->point.writable) { + ack.success = false; + ack.error = "physical point is not writable"; + return ack; + } + + int raw = 0; + if (!modbusEgress_->writeByPhysicalPointId(physicalPointId, value, &raw)) { + ack.success = false; + ack.error = "modbus write failed"; + return ack; + } + + fillAckIdentity(ack, command, physicalPointId); ack.executedValue = value; ack.success = true; - - const int raw = store_->rawRegisterValue(*entry, value); SB_LOG_INFO_TRACE("CommandExecutor", command.traceId, - "write ", entry->physicalChannel, " value=", value, - entry->unit, " raw=", raw, " (mock AO)"); + "write ", physicalPointId, " value=", value, " raw=", raw); if (uplink_) { uplink_->sendAck(ack); diff --git a/src/softbus_edge/src/DeviceFactory.cpp b/src/softbus_edge/src/DeviceFactory.cpp index f53e65c..39cb5c2 100644 --- a/src/softbus_edge/src/DeviceFactory.cpp +++ b/src/softbus_edge/src/DeviceFactory.cpp @@ -9,4 +9,16 @@ std::string DeviceFactory::createRuntimeDeviceId() return softbus::core::DeviceIdentity::generateRuntimeDeviceId(); } +std::unique_ptr DeviceFactory::createIngress(const EdgeConfig& config, + PhysicalRegistry* registry, + ModbusChannelManager* channels, + bool useMock, + const std::string& mockObjectRef) +{ + if (useMock) { + return std::make_unique(mockObjectRef, config.edgeId); + } + return std::make_unique(registry, channels, config.edgeId); +} + } // namespace softbus::edge diff --git a/src/softbus_edge/src/EdgeConfig.cpp b/src/softbus_edge/src/EdgeConfig.cpp index f26eed1..e0c64fd 100644 --- a/src/softbus_edge/src/EdgeConfig.cpp +++ b/src/softbus_edge/src/EdgeConfig.cpp @@ -1,10 +1,28 @@ #include +#include #include #include namespace softbus::edge { +namespace fs = std::filesystem; + +namespace { + +void resolveConfigPath(const fs::path& baseDir, std::string& path) +{ + if (path.empty()) { + return; + } + const fs::path candidate = baseDir / path; + if (fs::exists(candidate)) { + path = fs::absolute(candidate).string(); + } +} + +} // namespace + bool EdgeConfig::loadFromFile(const std::string& path) { std::ifstream ifs(path); @@ -17,11 +35,22 @@ bool EdgeConfig::loadFromFile(const std::string& path) return false; } const auto& edge = j["edge"]; + edgeId = edge.value("edgeId", edgeId); deviceId = edge.value("deviceId", deviceId); daemonHost = edge.value("daemonHost", daemonHost); daemonPort = static_cast(edge.value("daemonPort", static_cast(daemonPort))); heartbeatPeriodMs = edge.value("heartbeatPeriodMs", heartbeatPeriodMs); - pointTablePath = edge.value("pointTable", pointTablePath); + uplinkReconnectPeriodMs = edge.value("uplinkReconnectPeriodMs", uplinkReconnectPeriodMs); + physicalChannelsPath = edge.value("physicalChannels", physicalChannelsPath); + physicalDevicesPath = edge.value("physicalDevices", physicalDevicesPath); + bindingsLocalPath = edge.value("bindingsLocal", bindingsLocalPath); + bindingsPatchPath = edge.value("bindingsPatch", bindingsPatchPath); + + const fs::path baseDir = fs::path(path).parent_path(); + resolveConfigPath(baseDir, physicalChannelsPath); + resolveConfigPath(baseDir, physicalDevicesPath); + resolveConfigPath(baseDir, bindingsLocalPath); + resolveConfigPath(baseDir, bindingsPatchPath); return true; } diff --git a/src/softbus_edge/src/EdgeRuntime.cpp b/src/softbus_edge/src/EdgeRuntime.cpp new file mode 100644 index 0000000..45edf33 --- /dev/null +++ b/src/softbus_edge/src/EdgeRuntime.cpp @@ -0,0 +1,138 @@ +#include + +#include + +namespace softbus::edge { + +bool EdgeRuntime::init(const EdgeConfig& config, bool useMock, const std::string& mockObjectRef) +{ + config_ = config; + useMock_ = useMock; + mockObjectRef_ = mockObjectRef; + + if (!useMock) { + if (!physicalRegistry_.loadChannels(config.physicalChannelsPath)) { + SB_LOG_ERROR("EdgeRuntime", "Failed to load physical channels"); + return false; + } + if (!physicalRegistry_.loadDevices(config.physicalDevicesPath, config.edgeId)) { + SB_LOG_ERROR("EdgeRuntime", "Failed to load physical devices"); + return false; + } + } + + if (!config.bindingsLocalPath.empty()) { + localBindings_.loadLocal(config.bindingsLocalPath, config.edgeId); + } + + uplink_ = std::make_unique(config.daemonHost, config.daemonPort); + executor_ = std::make_unique( + &physicalRegistry_, &modbusEgress_, &mockStore_, &localBindings_, + uplink_.get(), config.deviceId, useMock); + + uplink_->setCommandHandler([this](const softbus::api::ExecuteCommand& cmd) { + if (executor_) { + executor_->execute(cmd); + } + }); + + ingress_ = factory_.createIngress(config, &physicalRegistry_, &modbusChannels_, useMock, mockObjectRef); + if (!ingress_->open()) { + SB_LOG_ERROR("EdgeRuntime", "Failed to open ingress"); + return false; + } + + if (!uplink_->connect()) { + SB_LOG_WARN( + "EdgeRuntime", + "Daemon unreachable at ", + config.daemonHost, + ":", + config.daemonPort, + "; running standalone, uplink will retry in background"); + } else { + uplink_->sendEdgeRegister(config.edgeId); + } + lastUplinkReconnectAttempt_ = std::chrono::steady_clock::now(); + return true; +} + +void EdgeRuntime::reloadPhysical() +{ + if (useMock_) { + return; + } + ingress_->close(); + modbusChannels_.closeAll(); + physicalRegistry_.reloadChannels(config_.physicalChannelsPath); + physicalRegistry_.reloadDevices(config_.physicalDevicesPath, config_.edgeId); + ingress_->open(); + SB_LOG_INFO("EdgeRuntime", "Physical config reloaded"); +} + +void EdgeRuntime::reloadBindingsPatch() +{ + if (config_.bindingsPatchPath.empty()) { + return; + } + localBindings_.applyPatch(config_.bindingsPatchPath); + SB_LOG_INFO("EdgeRuntime", "Bindings patch applied"); +} + +void EdgeRuntime::shutdown() +{ + if (ingress_) { + ingress_->close(); + } + modbusChannels_.closeAll(); + if (uplink_) { + uplink_->disconnect(); + } +} + +bool EdgeRuntime::isUplinkConnected() const +{ + return uplink_ && uplink_->isConnected(); +} + +void EdgeRuntime::tickUplink() +{ + if (!uplink_) { + return; + } + uplink_->poll(); + if (uplink_->isConnected()) { + return; + } + + const auto now = std::chrono::steady_clock::now(); + if (now - lastUplinkReconnectAttempt_ < + std::chrono::milliseconds(config_.uplinkReconnectPeriodMs)) { + return; + } + lastUplinkReconnectAttempt_ = now; + if (uplink_->connect()) { + uplink_->sendEdgeRegister(config_.edgeId); + SB_LOG_INFO("EdgeRuntime", "Connected to daemon at ", config_.daemonHost, ":", config_.daemonPort); + } +} + +void EdgeRuntime::pollIngress() +{ + if (!ingress_) { + return; + } + for (const auto& packet : ingress_->poll()) { + if (uplink_ && uplink_->isConnected()) { + uplink_->sendRawPacket(packet); + } + } +} + +void EdgeRuntime::tick() +{ + tickUplink(); + pollIngress(); +} + +} // namespace softbus::edge diff --git a/src/softbus_edge/src/ModbusChannel.cpp b/src/softbus_edge/src/ModbusChannel.cpp new file mode 100644 index 0000000..3dd8b3b --- /dev/null +++ b/src/softbus_edge/src/ModbusChannel.cpp @@ -0,0 +1,123 @@ +#include + +#include + +#if defined(SOFTBUS_HAS_LIBMODBUS) && SOFTBUS_HAS_LIBMODBUS +#include +#endif + +#include + +namespace softbus::edge { + +int ModbusChannelManager::modbusAddress(int registerAddress) const +{ + if (registerAddress >= 40001) { + return registerAddress - 40001; + } + if (registerAddress >= 30001) { + return registerAddress - 30001; + } + return registerAddress; +} + +bool ModbusChannelManager::openChannel(const PhysicalChannelDef& channel) +{ + closeChannel(channel.channelId); + ChannelHandle handle; + handle.config = channel; + +#if defined(SOFTBUS_HAS_LIBMODBUS) && SOFTBUS_HAS_LIBMODBUS + if (channel.transport == "tcp") { + handle.ctx = modbus_new_tcp(channel.host.c_str(), channel.tcpPort); + } else { + char parity = channel.parity; + handle.ctx = modbus_new_rtu( + channel.port.c_str(), channel.baudRate, parity, channel.dataBits, channel.stopBits); + } + if (!handle.ctx) { + SB_LOG_ERROR("ModbusChannel", "Failed to create modbus context for ", channel.channelId); + return false; + } + modbus_set_response_timeout(static_cast(handle.ctx), 0, 500000); + if (modbus_connect(static_cast(handle.ctx)) != 0) { + SB_LOG_ERROR("ModbusChannel", "Failed to connect channel ", channel.channelId); + modbus_free(static_cast(handle.ctx)); + return false; + } + channels_[channel.channelId] = handle; + SB_LOG_INFO("ModbusChannel", "Opened channel ", channel.channelId); + return true; +#else + (void)handle; + channels_[channel.channelId] = ChannelHandle{channel}; + SB_LOG_WARN("ModbusChannel", "libmodbus unavailable; channel ", channel.channelId, " simulated"); + return true; +#endif +} + +void ModbusChannelManager::closeChannel(const std::string& channelId) +{ + const auto it = channels_.find(channelId); + if (it == channels_.end()) { + return; + } +#if defined(SOFTBUS_HAS_LIBMODBUS) && SOFTBUS_HAS_LIBMODBUS + if (it->second.ctx) { + modbus_close(static_cast(it->second.ctx)); + modbus_free(static_cast(it->second.ctx)); + } +#endif + channels_.erase(it); +} + +void ModbusChannelManager::closeAll() +{ + for (const auto& [id, _] : channels_) { + (void)_; + closeChannel(id); + } +} + +std::optional ModbusChannelManager::readRegister(const ResolvedPhysicalPoint& point) +{ +#if defined(SOFTBUS_HAS_LIBMODBUS) && SOFTBUS_HAS_LIBMODBUS + const auto it = channels_.find(point.channel.channelId); + if (it == channels_.end() || !it->second.ctx) { + return std::nullopt; + } + modbus_set_slave(static_cast(it->second.ctx), point.device.slaveId); + uint16_t value = 0; + const int addr = modbusAddress(point.point.registerAddress); + const int rc = modbus_read_registers( + static_cast(it->second.ctx), addr, 1, &value); + if (rc != 1) { + return std::nullopt; + } + return value; +#else + static uint16_t sim = 250; + sim = static_cast((sim + 1) % 500); + (void)point; + return sim; +#endif +} + +bool ModbusChannelManager::writeRegister(const ResolvedPhysicalPoint& point, uint16_t rawValue) +{ +#if defined(SOFTBUS_HAS_LIBMODBUS) && SOFTBUS_HAS_LIBMODBUS + const auto it = channels_.find(point.channel.channelId); + if (it == channels_.end() || !it->second.ctx) { + return false; + } + modbus_set_slave(static_cast(it->second.ctx), point.device.slaveId); + const int addr = modbusAddress(point.point.registerAddress); + return modbus_write_register(static_cast(it->second.ctx), addr, rawValue) == 1; +#else + (void)point; + (void)rawValue; + return true; +#endif +} + +} // namespace softbus::edge diff --git a/src/softbus_edge/src/ModbusEgress.cpp b/src/softbus_edge/src/ModbusEgress.cpp new file mode 100644 index 0000000..8c4e38a --- /dev/null +++ b/src/softbus_edge/src/ModbusEgress.cpp @@ -0,0 +1,33 @@ +#include + +#include + +namespace softbus::edge { + +ModbusEgress::ModbusEgress(PhysicalRegistry* registry, ModbusChannelManager* channels) + : registry_(registry) + , channels_(channels) +{ +} + +bool ModbusEgress::writeByPhysicalPointId(const std::string& physicalPointId, + double engineeringValue, + int* rawOut) +{ + if (!registry_ || !channels_) { + return false; + } + const auto resolved = registry_->resolvePoint(physicalPointId); + if (!resolved || !resolved->point.writable) { + return false; + } + const int raw = resolved->point.scale <= 0.0 + ? static_cast(std::lround(engineeringValue)) + : static_cast(std::lround(engineeringValue / resolved->point.scale)); + if (rawOut) { + *rawOut = raw; + } + return channels_->writeRegister(*resolved, static_cast(raw)); +} + +} // namespace softbus::edge diff --git a/src/softbus_edge/src/ModbusIngress.cpp b/src/softbus_edge/src/ModbusIngress.cpp new file mode 100644 index 0000000..b9b98ee --- /dev/null +++ b/src/softbus_edge/src/ModbusIngress.cpp @@ -0,0 +1,92 @@ +#include + +#include +#include + +namespace softbus::edge { + +ModbusIngress::ModbusIngress(PhysicalRegistry* registry, + ModbusChannelManager* channels, + std::string edgeIdPrefix) + : registry_(registry) + , channels_(channels) + , edgeIdPrefix_(std::move(edgeIdPrefix)) +{ +} + +bool ModbusIngress::open() +{ + if (!registry_ || !channels_) { + return false; + } + for (const auto& ch : registry_->channels()) { + if (!channels_->openChannel(ch)) { + return false; + } + } + open_ = true; + lastPoll_ = std::chrono::steady_clock::now(); + return true; +} + +void ModbusIngress::close() +{ + if (channels_) { + channels_->closeAll(); + } + open_ = false; +} + +std::vector ModbusIngress::buildModbusFrame(int slaveId, uint16_t registerValue) const +{ + return { + static_cast(slaveId), + 0x03, + 0x02, + static_cast((registerValue >> 8) & 0xFF), + static_cast(registerValue & 0xFF), + 0x00, + 0x00 + }; +} + +std::vector ModbusIngress::poll() +{ + if (!open_ || !registry_ || !channels_) { + return {}; + } + + const auto now = std::chrono::steady_clock::now(); + int pollMs = 500; + if (!registry_->channels().empty()) { + pollMs = registry_->channels().front().pollIntervalMs; + } + if (now - lastPoll_ < std::chrono::milliseconds(pollMs)) { + return {}; + } + lastPoll_ = now; + + std::vector packets; + for (const auto& pointDef : registry_->pollPoints()) { + const auto resolved = registry_->resolvePoint(pointDef.physicalPointId); + if (!resolved) { + continue; + } + const auto raw = channels_->readRegister(*resolved); + if (!raw) { + continue; + } + + softbus::core::RawPacket packet; + packet.traceId = softbus::core::DeviceIdentity::generateTraceId(); + packet.timestampNs = softbus::core::HwClock::nowNs(); + packet.runtimeDeviceId = edgeIdPrefix_; + packet.sourceId = pointDef.physicalPointId; + packet.protocolTag = "modbus"; + packet.payload = buildModbusFrame(resolved->device.slaveId, *raw); + packets.push_back(std::move(packet)); + } + return packets; +} + +} // namespace softbus::edge diff --git a/src/softbus_edge/src/PhysicalRegistry.cpp b/src/softbus_edge/src/PhysicalRegistry.cpp new file mode 100644 index 0000000..6275080 --- /dev/null +++ b/src/softbus_edge/src/PhysicalRegistry.cpp @@ -0,0 +1,152 @@ +#include + +#include +#include + +namespace softbus::edge { + +namespace { + +std::string qualifyPointId(const std::string& edgeIdPrefix, const std::string& pointId) +{ + if (edgeIdPrefix.empty()) { + return pointId; + } + if (pointId.find('/') != std::string::npos) { + return pointId; + } + return edgeIdPrefix + "/" + pointId; +} + +} // namespace + +bool PhysicalRegistry::loadChannels(const std::string& path) +{ + std::ifstream ifs(path); + if (!ifs) { + return false; + } + nlohmann::json j; + ifs >> j; + channels_.clear(); + channelIndex_.clear(); + if (!j.contains("channels")) { + return true; + } + for (const auto& ch : j["channels"]) { + PhysicalChannelDef def; + def.channelId = ch.value("channelId", ""); + def.protocol = ch.value("protocol", "modbus"); + def.transport = ch.value("transport", "rtu"); + def.port = ch.value("port", ""); + def.host = ch.value("host", ""); + def.tcpPort = static_cast(ch.value("tcpPort", 502)); + def.baudRate = ch.value("baudRate", 9600); + const std::string parity = ch.value("parity", "N"); + def.parity = parity.empty() ? 'N' : parity[0]; + def.dataBits = ch.value("dataBits", 8); + def.stopBits = ch.value("stopBits", 1); + def.pollIntervalMs = ch.value("pollIntervalMs", 500); + if (def.channelId.empty()) { + continue; + } + channelIndex_[def.channelId] = channels_.size(); + channels_.push_back(std::move(def)); + } + return true; +} + +bool PhysicalRegistry::loadDevices(const std::string& path, const std::string& edgeIdPrefix) +{ + std::ifstream ifs(path); + if (!ifs) { + return false; + } + nlohmann::json j; + ifs >> j; + devices_.clear(); + pointIndex_.clear(); + pollPoints_.clear(); + if (!j.contains("devices")) { + return true; + } + for (const auto& dev : j["devices"]) { + PhysicalDeviceDef device; + device.physicalDeviceId = dev.value("physicalDeviceId", ""); + device.channelId = dev.value("channelId", ""); + device.protocol = dev.value("protocol", "modbus"); + device.slaveId = dev.value("slaveId", 1); + if (!dev.contains("points")) { + devices_.push_back(std::move(device)); + continue; + } + for (const auto& pt : dev["points"]) { + PhysicalPointDef point; + point.physicalPointId = qualifyPointId( + edgeIdPrefix, pt.value("physicalPointId", "")); + point.registerAddress = pt.value("register", 0); + point.functionCode = pt.value("function", 3); + point.writeFunctionCode = pt.value("writeFunction", 6); + point.scale = pt.value("scale", 1.0); + point.signalType = pt.value("signalType", "AI"); + point.writable = pt.value("writable", false); + if (point.physicalPointId.empty() || point.registerAddress == 0) { + continue; + } + device.points.push_back(point); + } + devices_.push_back(std::move(device)); + } + rebuildPollIndex(edgeIdPrefix); + return true; +} + +bool PhysicalRegistry::reloadChannels(const std::string& path) +{ + return loadChannels(path); +} + +bool PhysicalRegistry::reloadDevices(const std::string& path, const std::string& edgeIdPrefix) +{ + return loadDevices(path, edgeIdPrefix); +} + +void PhysicalRegistry::rebuildPollIndex(const std::string& edgeIdPrefix) +{ + (void)edgeIdPrefix; + pollPoints_.clear(); + pointIndex_.clear(); + for (const auto& device : devices_) { + const auto chIt = channelIndex_.find(device.channelId); + if (chIt == channelIndex_.end()) { + continue; + } + const auto& channel = channels_[chIt->second]; + for (const auto& point : device.points) { + ResolvedPhysicalPoint resolved{point, device, channel}; + pointIndex_[point.physicalPointId] = resolved; + pollPoints_.push_back(point); + } + } +} + +std::optional PhysicalRegistry::resolvePoint( + const std::string& physicalPointId) const +{ + const auto it = pointIndex_.find(physicalPointId); + if (it == pointIndex_.end()) { + return std::nullopt; + } + return it->second; +} + +std::optional PhysicalRegistry::findChannel(const std::string& channelId) const +{ + const auto it = channelIndex_.find(channelId); + if (it == channelIndex_.end()) { + return std::nullopt; + } + return channels_[it->second]; +} + +} // namespace softbus::edge diff --git a/src/softbus_edge/src/ShadowChannelStore.cpp b/src/softbus_edge/src/ShadowChannelStore.cpp index 5554cf9..178fdf2 100644 --- a/src/softbus_edge/src/ShadowChannelStore.cpp +++ b/src/softbus_edge/src/ShadowChannelStore.cpp @@ -1,32 +1,20 @@ #include -#include - namespace softbus::edge { -bool ShadowChannelStore::write(const softbus::registry::PointTableEntry& entry, - double engineeringValue) +bool ShadowChannelStore::write(const std::string& physicalPointId, double engineeringValue) { - channels_[entry.physicalChannel] = engineeringValue; + points_[physicalPointId] = engineeringValue; return true; } -std::optional ShadowChannelStore::read(const std::string& physicalChannel) const +std::optional ShadowChannelStore::read(const std::string& physicalPointId) const { - const auto it = channels_.find(physicalChannel); - if (it == channels_.end()) { + const auto it = points_.find(physicalPointId); + if (it == points_.end()) { return std::nullopt; } return it->second; } -int ShadowChannelStore::rawRegisterValue(const softbus::registry::PointTableEntry& entry, - double engineeringValue) const -{ - if (entry.scale <= 0.0) { - return static_cast(std::lround(engineeringValue)); - } - return static_cast(std::lround(engineeringValue / entry.scale)); -} - } // namespace softbus::edge diff --git a/src/softbus_edge/src/UplinkClient.cpp b/src/softbus_edge/src/UplinkClient.cpp index 30996cc..221248d 100644 --- a/src/softbus_edge/src/UplinkClient.cpp +++ b/src/softbus_edge/src/UplinkClient.cpp @@ -22,6 +22,11 @@ void UplinkClient::disconnect() client_->disconnect(); } +bool UplinkClient::isConnected() const +{ + return client_->isConnected(); +} + bool UplinkClient::sendRawPacket(const softbus::core::RawPacket& packet) { return client_->sendRawPacket(packet); @@ -37,6 +42,11 @@ bool UplinkClient::sendHeartbeat(const std::string& deviceId) return client_->sendHeartbeat(deviceId); } +bool UplinkClient::sendEdgeRegister(const std::string& edgeId) +{ + return client_->sendEdgeRegister(edgeId); +} + void UplinkClient::setCommandHandler(std::function handler) { client_->setCommandHandler(std::move(handler)); diff --git a/src/softbus_edge/src/main.cpp b/src/softbus_edge/src/main.cpp index c58ac3f..3783abd 100644 --- a/src/softbus_edge/src/main.cpp +++ b/src/softbus_edge/src/main.cpp @@ -1,17 +1,17 @@ +// role: edge 用于采集物理设备数据，并将其转换为软总线数据模型 +// @TODO： 1合并读以后建议在 edge 加物理层的差异化采集/变化上报，而不是加逻辑 binding 初筛。 +// 计划里的 ModbusPollScheduler（合并读）还没单独实现，这是 edge 侧最自然的下一步之一。 +// CANopen：edge 侧 CanIngress + SocketCAN；daemon 侧已有 protocol_canopen 占位 +// 以太网：Modbus TCP 已支持（transport: tcp）；裸 TCP / UDP 为规划项 +// 待增强：PollScheduler、按点周期、变化上报、多 ingress 并存 同时需要目录清晰 #include -#include -#include -#include -#include +#include -#include #include -#include #include -#include -#include +#include + #include -#include #include #include #include @@ -19,8 +19,10 @@ namespace { struct EdgeOptions { - std::string configPath{"config/edge.json"}; - std::string objectRef{"DemoSite/DemoSystem/DemoAsset/EdgeDevice_01/MotorTemperature"}; + std::string configPath{"config/edge/edge.json"}; + std::string mockObjectRef{ + "DemoSite/DemoSystem/DemoAsset/EdgeDevice_01/MotorTemperature"}; + bool useMock{false}; }; EdgeOptions parseArgs(int argc, char* argv[]) @@ -31,7 +33,9 @@ EdgeOptions parseArgs(int argc, char* argv[]) if (arg == "--config" && i + 1 < argc) { options.configPath = argv[++i]; } else if (arg == "--object-ref" && i + 1 < argc) { - options.objectRef = argv[++i]; + options.mockObjectRef = argv[++i]; + } else if (arg == "--mock") { + options.useMock = true; } } return options; @@ -49,81 +53,44 @@ int main(int argc, char* argv[]) return 1; } - softbus::registry::PointRegistry registry; - const auto configDir = std::filesystem::path(config.pointTablePath).parent_path(); - registry.loadTopicRules((configDir / "topic_rule.json").string()); - if (!registry.loadPointTable(config.pointTablePath)) { - std::cerr << "Failed to load point table: " << config.pointTablePath << std::endl; + softbus::edge::EdgeRuntime runtime; + if (!runtime.init(config, options.useMock, options.mockObjectRef)) { + std::cerr << "Failed to initialize edge runtime" << std::endl; return 1; } - softbus::edge::ShadowChannelStore shadowStore; - softbus::edge::UplinkClient uplink(config.daemonHost, config.daemonPort); - softbus::edge::CommandExecutor executor(®istry, &shadowStore, &uplink, config.deviceId); - uplink.setCommandHandler([&executor](const softbus::api::ExecuteCommand& cmd) { - const auto ack = executor.execute(cmd); - if (ack.success) { - SB_LOG_INFO_TRACE("Edge", cmd.traceId, - "command executed point=", ack.pointId, " value=", ack.executedValue); - } else { - SB_LOG_WARN_TRACE("Edge", cmd.traceId, "command failed: ", ack.error); - } - }); - - if (!uplink.connect()) { - std::cerr << "Failed to connect to daemon at " - << config.daemonHost << ":" << config.daemonPort << std::endl; - return 1; + softbus::core::PlatformSignal::install(nullptr); + // 监听物理通道、物理设备、绑定配置文件的变更，如果变更，则重新加载物理通道、物理设备、绑定配置 + softbus::core::ConfigWatcher watcher; + watcher.addWatch(config.physicalChannelsPath); + watcher.addWatch(config.physicalDevicesPath); + if (!config.bindingsPatchPath.empty()) { + watcher.addWatch(config.bindingsPatchPath); } - softbus::edge::MockModbusIngress ingress(options.objectRef, config.deviceId); - ingress.open(); - - softbus::core::PlatformSignal::install([]() { - SB_LOG_INFO("Edge", "Shutdown requested"); - }); - auto lastHeartbeat = std::chrono::steady_clock::now(); while (!softbus::core::PlatformSignal::shouldStop()) { - uplink.poll(); + watcher.poll(); + if (softbus::core::PlatformSignal::reloadRequested()) { + runtime.reloadPhysical(); + runtime.reloadBindingsPatch(); + softbus::core::PlatformSignal::clearReload(); + } + + runtime.tick(); const auto now = std::chrono::steady_clock::now(); if (now - lastHeartbeat >= std::chrono::milliseconds(config.heartbeatPeriodMs)) { - uplink.sendHeartbeat(config.deviceId); + if (runtime.isUplinkConnected()) { + runtime.uplink()->sendHeartbeat(config.edgeId); + } lastHeartbeat = now; } - for (const auto& packet : ingress.poll()) { - SB_LOG_INFO_TRACE("Edge", packet.traceId, - "Uplink RawPacket objectRef=", packet.sourceId); - uplink.sendRawPacket(packet); - } - - for (const auto& entry : registry.entries()) { - const auto value = shadowStore.read(entry.physicalChannel); - if (!value) { - continue; - } - softbus::core::RawPacket feedback; - feedback.traceId = softbus::core::DeviceIdentity::generateTraceId(); - feedback.timestampNs = softbus::core::HwClock::nowNs(); - feedback.sourceId = entry.objectRef; - feedback.protocolTag = "shadow"; - const int raw = shadowStore.rawRegisterValue(entry, *value); - feedback.payload = { - static_cast(entry.registerAddress >> 8), - static_cast(entry.registerAddress & 0xFF), - static_cast(raw >> 8), - static_cast(raw & 0xFF) - }; - uplink.sendRawPacket(feedback); - } - - std::this_thread::sleep_for(std::chrono::seconds(1)); + std::this_thread::sleep_for(std::chrono::milliseconds(50)); } - ingress.close(); - uplink.disconnect(); + runtime.shutdown(); return 0; } diff --git a/src/softbus_monitor/CMakeLists.txt b/src/softbus_monitor/CMakeLists.txt index 7934f38..c67a20c 100644 --- a/src/softbus_monitor/CMakeLists.txt +++ b/src/softbus_monitor/CMakeLists.txt @@ -1,3 +1,4 @@ +# role: monitor 用于监控软总线数据模型，并将其转换为物理设备数据 softbus_package(softbus_monitor) add_library(softbus_monitor STATIC diff --git a/src/softbus_monitor/include/softbus_monitor/HeartbeatMonitor.h b/src/softbus_monitor/include/softbus_monitor/HeartbeatMonitor.h index 9f4c4f9..c2499be 100644 --- a/src/softbus_monitor/include/softbus_monitor/HeartbeatMonitor.h +++ b/src/softbus_monitor/include/softbus_monitor/HeartbeatMonitor.h @@ -3,6 +3,7 @@ #include #include +#include #include #include #include @@ -11,20 +12,29 @@ namespace softbus::monitor { class HeartbeatMonitor { public: + // 默认超时 5 秒：超过该时间未收到心跳则判定离线 explicit HeartbeatMonitor(std::chrono::milliseconds timeout = std::chrono::seconds(5)); void attachBus(softbus::bus::Bus* bus); void tick(); private: + struct DeviceHeartbeatState { + std::chrono::steady_clock::time_point lastSeen; + std::uint64_t lastHeartbeatNs{0}; + }; + void onHeartbeat(const softbus::bus::BusMessage& message); - void publishOfflineEvent(const std::string& deviceId); + void publishOfflineEvent(const std::string& deviceId, + std::uint64_t offlineAtNs, + std::uint64_t lastHeartbeatNs); softbus::bus::Bus* bus_{nullptr}; softbus::bus::Bus::SubscriptionId hbSub_{0}; std::chrono::milliseconds timeout_; std::mutex mutex_; - std::unordered_map lastSeen_; + std::unordered_map devices_; + std::unordered_map offlineSinceNs_; }; } // namespace softbus::monitor diff --git a/src/softbus_monitor/src/HeartbeatMonitor.cpp b/src/softbus_monitor/src/HeartbeatMonitor.cpp index 05467a9..b4055a8 100644 --- a/src/softbus_monitor/src/HeartbeatMonitor.cpp +++ b/src/softbus_monitor/src/HeartbeatMonitor.cpp @@ -7,6 +7,16 @@ namespace softbus::monitor { +namespace { + +struct OfflineRecord { + std::string deviceId; + std::uint64_t offlineAtNs{0}; + std::uint64_t lastHeartbeatNs{0}; +}; + +} // namespace + HeartbeatMonitor::HeartbeatMonitor(std::chrono::milliseconds timeout) : timeout_(timeout) { @@ -29,27 +39,36 @@ void HeartbeatMonitor::onHeartbeat(const softbus::bus::BusMessage& message) return; } std::lock_guard lock(mutex_); - lastSeen_[message.deviceId] = std::chrono::steady_clock::now(); + devices_[message.deviceId] = DeviceHeartbeatState{ + std::chrono::steady_clock::now(), + softbus::core::HwClock::nowNs(), + }; + offlineSinceNs_.erase(message.deviceId); } void HeartbeatMonitor::tick() { const auto now = std::chrono::steady_clock::now(); - std::vector offline; + const std::uint64_t offlineAtNs = softbus::core::HwClock::nowNs(); + std::vector newlyOffline; { std::lock_guard lock(mutex_); - for (const auto& [deviceId, last] : lastSeen_) { - if (now - last > timeout_) { - offline.push_back(deviceId); + for (const auto& [deviceId, state] : devices_) { + if (now - state.lastSeen <= timeout_ || offlineSinceNs_.find(deviceId) != offlineSinceNs_.end()) { + continue; } + offlineSinceNs_[deviceId] = offlineAtNs; + newlyOffline.push_back(OfflineRecord{deviceId, offlineAtNs, state.lastHeartbeatNs}); } } - for (const auto& deviceId : offline) { - publishOfflineEvent(deviceId); + for (const auto& record : newlyOffline) { + publishOfflineEvent(record.deviceId, record.offlineAtNs, record.lastHeartbeatNs); } } -void HeartbeatMonitor::publishOfflineEvent(const std::string& deviceId) +void HeartbeatMonitor::publishOfflineEvent(const std::string& deviceId, + std::uint64_t offlineAtNs, + std::uint64_t lastHeartbeatNs) { if (!bus_) { return; @@ -58,13 +77,20 @@ void HeartbeatMonitor::publishOfflineEvent(const std::string& deviceId) ev.deviceId = deviceId; ev.eventType = "edge_offline"; ev.message = "heartbeat timeout"; - ev.timestampNs = softbus::core::HwClock::nowNs(); + ev.timestampNs = offlineAtNs; + ev.lastHeartbeatNs = lastHeartbeatNs; softbus::bus::BusMessage msg; msg.type = softbus::bus::BusPayloadType::Event; msg.event = ev; bus_->publish("event/" + deviceId + "/edge_offline", msg); - SB_LOG_WARN("HeartbeatMonitor", "edge offline: ", deviceId); + SB_LOG_WARN("HeartbeatMonitor", + "edge offline: ", + deviceId, + " offlineAtNs=", + offlineAtNs, + " lastHeartbeatNs=", + lastHeartbeatNs); } } // namespace softbus::monitor diff --git a/src/softbus_opcua/CMakeLists.txt b/src/softbus_opcua/CMakeLists.txt index c29c5da..e375b1d 100644 --- a/src/softbus_opcua/CMakeLists.txt +++ b/src/softbus_opcua/CMakeLists.txt @@ -1,3 +1,4 @@ +# role: opcua 用于将软总线数据模型转换为 OPC UA 数据模型 softbus_package(softbus_opcua) add_library(softbus_opcua STATIC diff --git a/src/softbus_pipeline/include/softbus_pipeline/PipelineEngine.h b/src/softbus_pipeline/include/softbus_pipeline/PipelineEngine.h index ed092dc..5373bb2 100644 --- a/src/softbus_pipeline/include/softbus_pipeline/PipelineEngine.h +++ b/src/softbus_pipeline/include/softbus_pipeline/PipelineEngine.h @@ -3,6 +3,7 @@ #include #include #include +#include #include namespace softbus::pipeline { @@ -11,7 +12,8 @@ class PipelineEngine { public: PipelineEngine(softbus::core::PluginLoader* plugin, softbus::opcua::UaModelBinder* binder, - softbus::registry::PointRegistry* registry); + softbus::registry::PointRegistry* registry, + softbus::registry::BindingRegistry* bindings = nullptr); void attachBus(softbus::bus::Bus* bus); void process(const softbus::core::RawPacket& packet); @@ -24,6 +26,7 @@ private: softbus::core::PluginLoader* plugin_{nullptr}; softbus::opcua::UaModelBinder* binder_{nullptr}; softbus::registry::PointRegistry* registry_{nullptr}; + softbus::registry::BindingRegistry* bindings_{nullptr}; softbus::bus::Bus* bus_{nullptr}; softbus::bus::Bus::SubscriptionId rawSub_{0}; }; diff --git a/src/softbus_pipeline/src/PipelineEngine.cpp b/src/softbus_pipeline/src/PipelineEngine.cpp index fb0a04a..48c6e86 100644 --- a/src/softbus_pipeline/src/PipelineEngine.cpp +++ b/src/softbus_pipeline/src/PipelineEngine.cpp @@ -7,12 +7,20 @@ namespace softbus::pipeline { +namespace { + +constexpr const char* kUnboundTopicPrefix = "edge/unbound/raw/"; + +} // namespace + PipelineEngine::PipelineEngine(softbus::core::PluginLoader* plugin, softbus::opcua::UaModelBinder* binder, - softbus::registry::PointRegistry* registry) + softbus::registry::PointRegistry* registry, + softbus::registry::BindingRegistry* bindings) : plugin_(plugin) , binder_(binder) , registry_(registry) + , bindings_(bindings) { } @@ -29,9 +37,14 @@ void PipelineEngine::attachBus(softbus::bus::Bus* bus) void PipelineEngine::onBusMessage(const softbus::bus::BusMessage& message) { - if (message.type == softbus::bus::BusPayloadType::RawPacket) { - process(message.rawPacket); + if (message.type != softbus::bus::BusPayloadType::RawPacket) { + return; } + if (message.topic.rfind("edge/unbound/", 0) == 0 + || message.topic.rfind("edge/raw/unbound/", 0) == 0) { + return; + } + process(message.rawPacket); } void PipelineEngine::process(const softbus::core::RawPacket& packet) @@ -62,16 +75,40 @@ void PipelineEngine::process(const softbus::core::RawPacket& packet) parsed.message.traceId = packet.traceId; parsed.message.timestampNs = packet.timestampNs; - if (registry_) { - const auto entry = registry_->findByObjectRef(parsed.message.id); - if (entry) { - publishDataPoint(parsed.message, *entry); + std::optional entry; + if (bindings_ && registry_) { + const auto binding = bindings_->findByPhysicalPointId(packet.sourceId); + if (!binding) { + if (bus_) { + softbus::bus::BusMessage busMsg; + busMsg.type = softbus::bus::BusPayloadType::RawPacket; + busMsg.rawPacket = packet; + bus_->publish(std::string(kUnboundTopicPrefix) + packet.sourceId, busMsg); + } + SB_LOG_INFO_TRACE("PipelineEngine", packet.traceId, + "Unbound physical point ", packet.sourceId); + return; } + entry = registry_->findByLogicalPointId(binding->logicalPointId); + if (entry && binding->scale > 0.0) { + double registerValue = 0.0; + if (parsed.message.value.contains("registerValue")) { + registerValue = parsed.message.value["registerValue"].get(); + } + parsed.message.value["value"] = registerValue * binding->scale; + parsed.message.id = entry->objectRef; + } + } else if (registry_) { + entry = registry_->findByObjectRef(parsed.message.id); + } + + if (entry) { + publishDataPoint(parsed.message, *entry); } binder_->bind(parsed.message); SB_LOG_INFO_TRACE("PipelineEngine", packet.traceId, - "RawPacket converted to DOMMessage id=", parsed.message.id); + "RawPacket converted physicalPointId=", packet.sourceId); } void PipelineEngine::publishDataPoint(const softbus::core::DOMMessage& message, diff --git a/src/softbus_registry/CMakeLists.txt b/src/softbus_registry/CMakeLists.txt index 51720f3..11f5064 100644 --- a/src/softbus_registry/CMakeLists.txt +++ b/src/softbus_registry/CMakeLists.txt @@ -3,6 +3,7 @@ softbus_package(softbus_registry) add_library(softbus_registry STATIC src/PointRegistry.cpp src/PointTableEntry.cpp + src/BindingRegistry.cpp ) softbus_export_include(softbus_registry "${CMAKE_CURRENT_SOURCE_DIR}/include") diff --git a/src/softbus_registry/include/softbus_registry/BindingEntry.h b/src/softbus_registry/include/softbus_registry/BindingEntry.h new file mode 100644 index 0000000..4e88bc3 --- /dev/null +++ b/src/softbus_registry/include/softbus_registry/BindingEntry.h @@ -0,0 +1,21 @@ +#pragma once + +#include + +namespace softbus::registry { + +struct BindingEntry { + std::string bindingId; + std::string edgeId; + std::string physicalPointId; + std::string logicalPointId; + double scale{0.0}; + bool enabled{true}; +}; + +struct BindingRoute { + std::string edgeId; + std::string physicalPointId; +}; + +} // namespace softbus::registry diff --git a/src/softbus_registry/include/softbus_registry/BindingRegistry.h b/src/softbus_registry/include/softbus_registry/BindingRegistry.h new file mode 100644 index 0000000..335b7ed --- /dev/null +++ b/src/softbus_registry/include/softbus_registry/BindingRegistry.h @@ -0,0 +1,38 @@ +#pragma once + +#include + +#include +#include +#include +#include + +namespace softbus::registry { + +class BindingRegistry { +public: + bool loadGlobal(const std::string& path); + bool loadLocal(const std::string& path, const std::string& expectedEdgeId = {}); + bool reloadGlobal(const std::string& path); + bool applyPatch(const std::string& path); + + const std::vector& entries() const { return entries_; } + + std::optional findByBindingId(const std::string& bindingId) const; + std::optional resolveLogical(const std::string& physicalPointId) const; + std::optional resolveRoute(const std::string& logicalPointId) const; + std::optional findByPhysicalPointId(const std::string& physicalPointId) const; + std::optional resolvePhysicalLocal(const std::string& logicalPointId) const; + +private: + void indexEntry(const BindingEntry& entry); + void clearIndexes(); + + std::vector entries_; + std::unordered_map byBindingId_; + std::unordered_map byPhysicalPointId_; + std::unordered_map byLogicalPointId_; + bool localMode_{false}; +}; + +} // namespace softbus::registry diff --git a/src/softbus_registry/include/softbus_registry/PointRegistry.h b/src/softbus_registry/include/softbus_registry/PointRegistry.h index ec0bfd2..9d8cd7a 100644 --- a/src/softbus_registry/include/softbus_registry/PointRegistry.h +++ b/src/softbus_registry/include/softbus_registry/PointRegistry.h @@ -3,6 +3,8 @@ #include #include +#include + #include #include #include @@ -13,12 +15,17 @@ namespace softbus::registry { class PointRegistry { public: bool loadPointTable(const std::string& path); + bool loadLogicalPoints(const std::string& path); + bool reloadPointTable(const std::string& path); bool loadTopicRules(const std::string& path); + std::vector diff(const PointRegistry& previous) const; + const softbus::core::ObjectModelTree& tree() const { return tree_; } const TopicRules& topicRules() const { return topicRules_; } const std::vector& entries() const { return entries_; } + std::optional findByLogicalPointId(const std::string& logicalPointId) const; std::optional findByObjectRef(const std::string& objectRef) const; std::optional findByTopic(const std::string& topic) const; std::optional findByOpcuaNodeId(const std::string& nodeId) const; @@ -28,7 +35,10 @@ public: bool validateWriteValue(const PointTableEntry& entry, double value, std::string* error) const; private: + bool loadPointTableLegacy(const std::string& path); + bool loadFromJson(const nlohmann::json& j, bool logicalOnly); // NOLINT void indexEntry(const PointTableEntry& entry); + void rebuildIndexes(); softbus::core::ObjectModelTree tree_; TopicRules topicRules_; diff --git a/src/softbus_registry/src/BindingRegistry.cpp b/src/softbus_registry/src/BindingRegistry.cpp new file mode 100644 index 0000000..2e57be4 --- /dev/null +++ b/src/softbus_registry/src/BindingRegistry.cpp @@ -0,0 +1,200 @@ +#include + +#include +#include + +namespace softbus::registry { + +namespace { + +bool parseBindingEntry(const nlohmann::json& j, BindingEntry& entry, bool requireEdgeId) +{ + entry.bindingId = j.value("bindingId", ""); + entry.edgeId = j.value("edgeId", ""); + entry.physicalPointId = j.value("physicalPointId", ""); + entry.logicalPointId = j.value("logicalPointId", ""); + entry.scale = j.value("scale", 0.0); + entry.enabled = j.value("enabled", true); + if (entry.bindingId.empty() || entry.physicalPointId.empty() || entry.logicalPointId.empty()) { + return false; + } + if (requireEdgeId && entry.edgeId.empty()) { + return false; + } + return true; +} + +} // namespace + +bool BindingRegistry::loadGlobal(const std::string& path) +{ + std::ifstream ifs(path); + if (!ifs) { + return false; + } + nlohmann::json j; + ifs >> j; + localMode_ = false; + entries_.clear(); + clearIndexes(); + if (!j.contains("bindings")) { + return true; + } + for (const auto& b : j["bindings"]) { + BindingEntry entry; + if (!parseBindingEntry(b, entry, true)) { + continue; + } + entries_.push_back(entry); + indexEntry(entries_.back()); + } + return true; +} + +bool BindingRegistry::loadLocal(const std::string& path, const std::string& expectedEdgeId) +{ + std::ifstream ifs(path); + if (!ifs) { + return false; + } + nlohmann::json j; + ifs >> j; + localMode_ = true; + entries_.clear(); + clearIndexes(); + const std::string edgeId = j.value("edgeId", expectedEdgeId); + if (!j.contains("bindings")) { + return true; + } + for (const auto& b : j["bindings"]) { + BindingEntry entry; + if (!parseBindingEntry(b, entry, false)) { + continue; + } + entry.edgeId = edgeId; + entries_.push_back(entry); + indexEntry(entries_.back()); + } + return true; +} + +bool BindingRegistry::reloadGlobal(const std::string& path) +{ + return loadGlobal(path); +} + +bool BindingRegistry::applyPatch(const std::string& path) +{ + std::ifstream ifs(path); + if (!ifs) { + return false; + } + nlohmann::json j; + ifs >> j; + const std::string op = j.value("op", "upsert"); + if (!j.contains("binding")) { + return false; + } + BindingEntry entry; + if (!parseBindingEntry(j["binding"], entry, !localMode_)) { + return false; + } + if (localMode_ && entry.edgeId.empty()) { + entry.edgeId = j.value("edgeId", ""); + } + if (op == "delete") { + for (auto it = entries_.begin(); it != entries_.end(); ++it) { + if (it->bindingId == entry.bindingId) { + entries_.erase(it); + clearIndexes(); + for (const auto& e : entries_) { + indexEntry(e); + } + return true; + } + } + return false; + } + for (auto& e : entries_) { + if (e.bindingId == entry.bindingId) { + e = entry; + clearIndexes(); + for (const auto& item : entries_) { + indexEntry(item); + } + return true; + } + } + entries_.push_back(entry); + indexEntry(entries_.back()); + return true; +} + +void BindingRegistry::clearIndexes() +{ + byBindingId_.clear(); + byPhysicalPointId_.clear(); + byLogicalPointId_.clear(); +} + +void BindingRegistry::indexEntry(const BindingEntry& entry) +{ + const std::size_t idx = entries_.size() - 1; + byBindingId_[entry.bindingId] = idx; + if (entry.enabled) { + byPhysicalPointId_[entry.physicalPointId] = idx; + byLogicalPointId_[entry.logicalPointId] = idx; + } +} + +std::optional BindingRegistry::findByBindingId(const std::string& bindingId) const +{ + const auto it = byBindingId_.find(bindingId); + if (it == byBindingId_.end()) { + return std::nullopt; + } + return entries_[it->second]; +} + +std::optional BindingRegistry::findByPhysicalPointId( + const std::string& physicalPointId) const +{ + const auto it = byPhysicalPointId_.find(physicalPointId); + if (it == byPhysicalPointId_.end()) { + return std::nullopt; + } + return entries_[it->second]; +} + +std::optional BindingRegistry::resolveLogical(const std::string& physicalPointId) const +{ + const auto entry = findByPhysicalPointId(physicalPointId); + if (!entry) { + return std::nullopt; + } + return entry->logicalPointId; +} + +std::optional BindingRegistry::resolveRoute(const std::string& logicalPointId) const +{ + const auto it = byLogicalPointId_.find(logicalPointId); + if (it == byLogicalPointId_.end()) { + return std::nullopt; + } + const auto& entry = entries_[it->second]; + BindingRoute route; + route.edgeId = entry.edgeId; + route.physicalPointId = entry.physicalPointId; + return route; +} + +std::optional BindingRegistry::resolvePhysicalLocal(const std::string& logicalPointId) const +{ + const auto route = resolveRoute(logicalPointId); + if (!route) { + return std::nullopt; + } + return route->physicalPointId; +} + +} // namespace softbus::registry diff --git a/src/softbus_registry/src/PointRegistry.cpp b/src/softbus_registry/src/PointRegistry.cpp index 63cdaac..20dc9be 100644 --- a/src/softbus_registry/src/PointRegistry.cpp +++ b/src/softbus_registry/src/PointRegistry.cpp @@ -4,10 +4,16 @@ #include #include +#include namespace softbus::registry { bool PointRegistry::loadPointTable(const std::string& path) +{ + return loadPointTableLegacy(path); +} + +bool PointRegistry::loadPointTableLegacy(const std::string& path) { std::ifstream ifs(path); if (!ifs) { @@ -15,16 +21,56 @@ bool PointRegistry::loadPointTable(const std::string& path) } nlohmann::json j; ifs >> j; + if (!loadFromJson(j, false)) { + return false; + } + rebuildIndexes(); + return true; +} +bool PointRegistry::loadLogicalPoints(const std::string& path) +{ + std::ifstream ifs(path); + if (!ifs) { + return false; + } + nlohmann::json j; + ifs >> j; + if (!loadFromJson(j, true)) { + return false; + } + rebuildIndexes(); + return true; +} + +bool PointRegistry::reloadPointTable(const std::string& path) +{ + std::ifstream ifs(path); + if (!ifs) { + return false; + } + nlohmann::json j; + ifs >> j; + bool logicalOnly = true; + if (j.contains("devices") && j["devices"].is_array() && !j["devices"].empty() + && j["devices"][0].contains("points") && j["devices"][0]["points"].is_array() + && !j["devices"][0]["points"].empty()) { + logicalOnly = !j["devices"][0]["points"][0].contains("register"); + } + if (!loadFromJson(j, logicalOnly)) { + return false; + } + rebuildIndexes(); + return true; +} + +bool PointRegistry::loadFromJson(const nlohmann::json& j, bool logicalOnly) +{ tree_.site = j.value("site", ""); tree_.system = j.value("system", ""); tree_.asset = j.value("asset", ""); tree_.devices.clear(); entries_.clear(); - byObjectRef_.clear(); - byTopic_.clear(); - byOpcuaNodeId_.clear(); - byDevicePoint_.clear(); if (!j.contains("devices")) { return tree_.isValid(); @@ -32,9 +78,12 @@ bool PointRegistry::loadPointTable(const std::string& path) for (const auto& dev : j["devices"]) { softbus::core::DeviceDef device; - device.stableDeviceKey = dev.value("stableDeviceKey", dev.value("deviceId", "")); - device.protocol = dev.value("protocol", "modbus"); - const std::string deviceId = dev.value("deviceId", device.stableDeviceKey); + device.stableDeviceKey = dev.value( + "stableDeviceKey", + dev.value("logicalDeviceId", dev.value("deviceId", ""))); + device.protocol = dev.value("protocol", logicalOnly ? "logical" : "modbus"); + const std::string deviceId = dev.value( + "deviceId", dev.value("logicalDeviceId", device.stableDeviceKey)); if (!dev.contains("points")) { tree_.devices.push_back(std::move(device)); @@ -43,13 +92,13 @@ bool PointRegistry::loadPointTable(const std::string& path) for (const auto& pt : dev["points"]) { softbus::core::PointDef pointDef; - pointDef.name = pt.value("name", pt.value("pointId", "")); - pointDef.registerAddress = pt.value("register", 0); + pointDef.name = pt.value("name", pt.value("logicalPointId", pt.value("pointId", ""))); + pointDef.registerAddress = logicalOnly ? 0 : pt.value("register", 0); pointDef.domain = pt.value("domain", "Process"); pointDef.signalType = pt.value("signalType", "AI"); - pointDef.physicalChannel = pt.value("physicalChannel", ""); + pointDef.physicalChannel = logicalOnly ? "" : pt.value("physicalChannel", ""); pointDef.unit = pt.value("unit", ""); - pointDef.scale = pt.value("scale", 1.0); + pointDef.scale = logicalOnly ? 1.0 : pt.value("scale", 1.0); pointDef.access = pt.value("access", "read"); pointDef.writable = pt.value("writable", false); if (pt.contains("range") && pt["range"].is_array() && pt["range"].size() >= 2) { @@ -63,7 +112,7 @@ bool PointRegistry::loadPointTable(const std::string& path) PointTableEntry entry; entry.deviceId = deviceId; - entry.pointId = pt.value("pointId", pointDef.name); + entry.pointId = pt.value("pointId", pt.value("logicalPointId", pointDef.name)); entry.name = pointDef.name; entry.displayName = pt.value("displayName", entry.name); entry.signalType = pointDef.signalType; @@ -80,8 +129,7 @@ bool PointRegistry::loadPointTable(const std::string& path) tree_, device.stableDeviceKey, entry.name); entry.objectRef = objectRef.toPath(); - entry.opcuaNodeId = pt.value( - "opcuaNodeId", "ns=1;s=" + entry.objectRef); + entry.opcuaNodeId = pt.value("opcuaNodeId", "ns=1;s=" + entry.objectRef); entry.dataTopic = pt.value( "dataTopic", topicRules_.formatData(entry.deviceId, entry.pointId)); entry.cmdTopic = pt.value( @@ -90,13 +138,27 @@ bool PointRegistry::loadPointTable(const std::string& path) "ackTopic", topicRules_.formatAck(entry.deviceId, entry.pointId)); entries_.push_back(entry); - indexEntry(entries_.back()); } tree_.devices.push_back(std::move(device)); } return tree_.isValid() && !entries_.empty(); } +std::vector PointRegistry::diff(const PointRegistry& previous) const +{ + std::unordered_set previousKeys; + for (const auto& e : previous.entries_) { + previousKeys.insert(e.deviceId + "/" + e.pointId); + } + std::vector added; + for (const auto& e : entries_) { + if (previousKeys.find(e.deviceId + "/" + e.pointId) == previousKeys.end()) { + added.push_back(e); + } + } + return added; +} + bool PointRegistry::loadTopicRules(const std::string& path) { std::ifstream ifs(path); @@ -113,6 +175,29 @@ bool PointRegistry::loadTopicRules(const std::string& path) return true; } +void PointRegistry::rebuildIndexes() +{ + byObjectRef_.clear(); + byTopic_.clear(); + byOpcuaNodeId_.clear(); + byDevicePoint_.clear(); + for (std::size_t i = 0; i < entries_.size(); ++i) { + const auto& entry = entries_[i]; + byObjectRef_[entry.objectRef] = i; + byOpcuaNodeId_[entry.opcuaNodeId] = i; + byDevicePoint_[entry.deviceId + "/" + entry.pointId] = i; + if (!entry.dataTopic.empty()) { + byTopic_[entry.dataTopic] = i; + } + if (!entry.cmdTopic.empty()) { + byTopic_[entry.cmdTopic] = i; + } + if (!entry.ackTopic.empty()) { + byTopic_[entry.ackTopic] = i; + } + } +} + void PointRegistry::indexEntry(const PointTableEntry& entry) { const std::size_t idx = entries_.size() - 1; @@ -130,6 +215,18 @@ void PointRegistry::indexEntry(const PointTableEntry& entry) } } +std::optional PointRegistry::findByLogicalPointId( + const std::string& logicalPointId) const +{ + const auto slash = logicalPointId.find('/'); + if (slash == std::string::npos) { + return std::nullopt; + } + return findByDevicePoint( + logicalPointId.substr(0, slash), + logicalPointId.substr(slash + 1)); +} + std::optional PointRegistry::findByObjectRef(const std::string& objectRef) const { const auto it = byObjectRef_.find(objectRef); diff --git a/src/softbus_transport/include/softbus_transport/TcpUplinkClient.h b/src/softbus_transport/include/softbus_transport/TcpUplinkClient.h index de0d783..2e6ef9d 100644 --- a/src/softbus_transport/include/softbus_transport/TcpUplinkClient.h +++ b/src/softbus_transport/include/softbus_transport/TcpUplinkClient.h @@ -26,6 +26,7 @@ public: bool sendRawPacket(const softbus::core::RawPacket& packet); bool sendAck(const softbus::core::SoftbusAck& ack); bool sendHeartbeat(const std::string& deviceId); + bool sendEdgeRegister(const std::string& edgeId); void poll(); bool isConnected() const { return connected_; } diff --git a/src/softbus_transport/include/softbus_transport/TcpUplinkServer.h b/src/softbus_transport/include/softbus_transport/TcpUplinkServer.h index a9de07c..edbd267 100644 --- a/src/softbus_transport/include/softbus_transport/TcpUplinkServer.h +++ b/src/softbus_transport/include/softbus_transport/TcpUplinkServer.h @@ -8,8 +8,7 @@ #include #include #include -#include -#include +#include namespace softbus::transport { @@ -27,18 +26,28 @@ public: void start(); void stop(); bool sendCommand(const softbus::api::ExecuteCommand& command); + bool sendCommandToEdge(const std::string& edgeId, const softbus::api::ExecuteCommand& command); + bool hasEdge(const std::string& edgeId) const; private: + struct EdgeSession { + std::shared_ptr socket; + std::string edgeId; + std::string readBuffer; + }; + void doAccept(); - void doRead(std::shared_ptr socket); - void handleLine(const std::string& line); + void doRead(const std::shared_ptr& session); + void handleLine(const std::shared_ptr& session, const std::string& line); + void removeSession(const std::shared_ptr& session); asio::io_context& io_; asio::ip::tcp::acceptor acceptor_; RawPacketHandler rawHandler_; WireMessageHandler wireHandler_; - std::shared_ptr clientSocket_; - std::string readBuffer_; + mutable std::mutex sessionMutex_; + std::vector> sessions_; + std::unordered_map> edgeSessions_; std::mutex sendMutex_; bool running_{false}; }; diff --git a/src/softbus_transport/include/softbus_transport/UplinkWire.h b/src/softbus_transport/include/softbus_transport/UplinkWire.h index c52cea4..d5fc5c0 100644 --- a/src/softbus_transport/include/softbus_transport/UplinkWire.h +++ b/src/softbus_transport/include/softbus_transport/UplinkWire.h @@ -17,6 +17,7 @@ enum class WireMessageType { ExecuteCommand, Ack, Heartbeat, + EdgeRegister, Unknown }; @@ -26,12 +27,14 @@ struct WireMessage { softbus::api::ExecuteCommand command; softbus::core::SoftbusAck ack; std::string heartbeatDeviceId; + std::string edgeId; }; std::string encodeWireMessage(const softbus::core::RawPacket& packet); std::string encodeWireMessage(const softbus::api::ExecuteCommand& command); std::string encodeWireMessage(const softbus::core::SoftbusAck& ack); std::string encodeWireMessage(const std::string& deviceId); +std::string encodeEdgeRegister(const std::string& edgeId); std::optional decodeWireMessage(const std::string& line); std::string base64Encode(const uint8_t* data, std::size_t size); diff --git a/src/softbus_transport/src/TcpUplinkClient.cpp b/src/softbus_transport/src/TcpUplinkClient.cpp index 4e53d74..2e10bc1 100644 --- a/src/softbus_transport/src/TcpUplinkClient.cpp +++ b/src/softbus_transport/src/TcpUplinkClient.cpp @@ -86,6 +86,18 @@ bool TcpUplinkClient::sendAck(const softbus::core::SoftbusAck& ack) return !ec; } +bool TcpUplinkClient::sendEdgeRegister(const std::string& edgeId) +{ + if (!socket_ || !socket_->is_open()) { + return false; + } + const auto line = encodeEdgeRegister(edgeId); + std::lock_guard lock(sendMutex_); + std::error_code ec; + asio::write(*socket_, asio::buffer(line), ec); + return !ec; +} + bool TcpUplinkClient::sendHeartbeat(const std::string& deviceId) { if (!socket_ || !socket_->is_open()) { diff --git a/src/softbus_transport/src/TcpUplinkServer.cpp b/src/softbus_transport/src/TcpUplinkServer.cpp index d46978f..805236b 100644 --- a/src/softbus_transport/src/TcpUplinkServer.cpp +++ b/src/softbus_transport/src/TcpUplinkServer.cpp @@ -2,6 +2,7 @@ #include +#include #include namespace softbus::transport { @@ -41,77 +42,133 @@ void TcpUplinkServer::stop() running_ = false; std::error_code ec; acceptor_.close(ec); - if (clientSocket_) { - clientSocket_->close(ec); - clientSocket_.reset(); + std::lock_guard lock(sessionMutex_); + for (auto& session : sessions_) { + if (session->socket) { + session->socket->close(ec); + } } + sessions_.clear(); + edgeSessions_.clear(); } bool TcpUplinkServer::sendCommand(const softbus::api::ExecuteCommand& command) { - if (!clientSocket_ || !clientSocket_->is_open()) { + std::string edgeId; + { + std::lock_guard lock(sessionMutex_); + if (!edgeSessions_.empty()) { + edgeId = edgeSessions_.begin()->first; + } + } + if (!edgeId.empty()) { + return sendCommandToEdge(edgeId, command); + } + std::shared_ptr session; + { + std::lock_guard lock(sessionMutex_); + if (sessions_.empty() || !sessions_.front()->socket) { + return false; + } + session = sessions_.front(); + } + const auto line = encodeWireMessage(command); + std::lock_guard sendLock(sendMutex_); + std::error_code ec; + asio::write(*session->socket, asio::buffer(line), ec); + return !ec; +} + +bool TcpUplinkServer::sendCommandToEdge(const std::string& edgeId, + const softbus::api::ExecuteCommand& command) +{ + std::shared_ptr session; + { + std::lock_guard lock(sessionMutex_); + const auto it = edgeSessions_.find(edgeId); + if (it == edgeSessions_.end()) { + return false; + } + session = it->second; + } + if (!session || !session->socket || !session->socket->is_open()) { return false; } const auto line = encodeWireMessage(command); - std::lock_guard lock(sendMutex_); + std::lock_guard sendLock(sendMutex_); std::error_code ec; - asio::write(*clientSocket_, asio::buffer(line), ec); + asio::write(*session->socket, asio::buffer(line), ec); return !ec; } +bool TcpUplinkServer::hasEdge(const std::string& edgeId) const +{ + std::lock_guard lock(sessionMutex_); + return edgeSessions_.find(edgeId) != edgeSessions_.end(); +} + void TcpUplinkServer::doAccept() { if (!running_) { return; } - auto socket = std::make_shared(io_); - acceptor_.async_accept(*socket, [this, socket](const std::error_code& ec) { + auto session = std::make_shared(); + session->socket = std::make_shared(io_); + acceptor_.async_accept(*session->socket, [this, session](const std::error_code& ec) { if (ec) { if (running_) { doAccept(); } return; } - clientSocket_ = socket; - readBuffer_.clear(); - SB_LOG_INFO("TcpUplinkServer", "Edge connected"); - doRead(socket); + { + std::lock_guard lock(sessionMutex_); + sessions_.push_back(session); + } + SB_LOG_INFO("TcpUplinkServer", "Edge connected (awaiting edgeRegister)"); + doRead(session); doAccept(); }); } -void TcpUplinkServer::doRead(const std::shared_ptr socket) +void TcpUplinkServer::doRead(const std::shared_ptr& session) { auto buffer = std::make_shared>(); - socket->async_read_some(asio::buffer(*buffer), - [this, socket, buffer](const std::error_code& ec, std::size_t bytes) { + session->socket->async_read_some(asio::buffer(*buffer), + [this, session, buffer](const std::error_code& ec, std::size_t bytes) { if (ec) { - SB_LOG_WARN("TcpUplinkServer", "Edge disconnected"); - if (clientSocket_ == socket) { - clientSocket_.reset(); - } + SB_LOG_WARN("TcpUplinkServer", "Edge disconnected edgeId=", session->edgeId); + removeSession(session); return; } - readBuffer_.append(buffer->data(), bytes); + session->readBuffer.append(buffer->data(), bytes); for (;;) { - const auto pos = readBuffer_.find('\n'); + const auto pos = session->readBuffer.find('\n'); if (pos == std::string::npos) { break; } - const std::string line = readBuffer_.substr(0, pos); - readBuffer_.erase(0, pos + 1); - handleLine(line); + const std::string line = session->readBuffer.substr(0, pos); + session->readBuffer.erase(0, pos + 1); + handleLine(session, line); } - doRead(socket); + doRead(session); }); } -void TcpUplinkServer::handleLine(const std::string& line) +void TcpUplinkServer::handleLine(const std::shared_ptr& session, + const std::string& line) { const auto msg = decodeWireMessage(line); if (!msg) { return; } + if (msg->type == WireMessageType::EdgeRegister && !msg->edgeId.empty()) { + session->edgeId = msg->edgeId; + std::lock_guard lock(sessionMutex_); + edgeSessions_[msg->edgeId] = session; + SB_LOG_INFO("TcpUplinkServer", "Edge registered edgeId=", msg->edgeId); + return; + } if (wireHandler_) { wireHandler_(*msg); } @@ -120,4 +177,15 @@ void TcpUplinkServer::handleLine(const std::string& line) } } +void TcpUplinkServer::removeSession(const std::shared_ptr& session) +{ + std::lock_guard lock(sessionMutex_); + if (!session->edgeId.empty()) { + edgeSessions_.erase(session->edgeId); + } + sessions_.erase( + std::remove(sessions_.begin(), sessions_.end(), session), + sessions_.end()); +} + } // namespace softbus::transport diff --git a/src/softbus_transport/src/TransportBusBridge.cpp b/src/softbus_transport/src/TransportBusBridge.cpp index feb5a4d..0878d3d 100644 --- a/src/softbus_transport/src/TransportBusBridge.cpp +++ b/src/softbus_transport/src/TransportBusBridge.cpp @@ -29,7 +29,14 @@ void TransportBusBridge::onBusMessage(const softbus::bus::BusMessage& message) if (!server_ || message.type != softbus::bus::BusPayloadType::ExecuteCommand) { return; } - if (!server_->sendCommand(message.command)) { + const std::string edgeId = message.command.params.value("edgeId", ""); + bool sent = false; + if (!edgeId.empty()) { + sent = server_->sendCommandToEdge(edgeId, message.command); + } else { + sent = server_->sendCommand(message.command); + } + if (!sent) { SB_LOG_WARN("TransportBusBridge", "failed to send command to edge topic=", message.topic); } } diff --git a/src/softbus_transport/src/UplinkWire.cpp b/src/softbus_transport/src/UplinkWire.cpp index a712a75..d6436af 100644 --- a/src/softbus_transport/src/UplinkWire.cpp +++ b/src/softbus_transport/src/UplinkWire.cpp @@ -132,7 +132,21 @@ std::optional decodeWireMessage(const std::string& line) msg.heartbeatDeviceId = j.value("deviceId", ""); return msg; } + if (type == "edgeRegister") { + msg.type = WireMessageType::EdgeRegister; + msg.edgeId = j.value("edgeId", ""); + return msg; + } return std::nullopt; } +std::string encodeEdgeRegister(const std::string& edgeId) +{ + nlohmann::json j{ + {"type", "edgeRegister"}, + {"edgeId", edgeId} + }; + return j.dump() + "\n"; +} + } // namespace softbus::transport diff --git a/tests/integration/valve_control_loop.sh b/tests/integration/valve_control_loop.sh index ff8d462..733a362 100755 --- a/tests/integration/valve_control_loop.sh +++ b/tests/integration/valve_control_loop.sh @@ -7,12 +7,13 @@ fuser -k 9000/tcp 4840/tcp 2>/dev/null || true sleep 1 ./build/bin/softbus_daemon \ - --config config/daemon_profile.json \ - --point-table config/point_table.json \ + --config config/daemon/daemon_profile.json \ + --logical-points config/daemon/logical_points.json \ + --bindings-global config/daemon/bindings_global.json \ --self-test > /tmp/valve_loop_daemon.log 2>&1 & DAEMON_PID=$! sleep 1 -./build/bin/softbus_edge --config config/edge.json > /tmp/valve_loop_edge.log 2>&1 & +./build/bin/softbus_edge --config config/edge/edge.json > /tmp/valve_loop_edge.log 2>&1 & EDGE_PID=$! sleep 8 @@ -21,6 +22,6 @@ fuser -k 9000/tcp 4840/tcp 2>/dev/null || true grep -q "state=Succeeded" /tmp/valve_loop_daemon.log grep -q "executedValue=60" /tmp/valve_loop_daemon.log -grep -q "ValveOpening value=60" /tmp/valve_loop_edge.log +grep -qE "write edge01/phy_modbus_s1_r40002|ValveOpening value=60" /tmp/valve_loop_edge.log echo "valve_control_loop: PASS" diff --git a/third_party/README.md b/third_party/README.md index 924ae1a..6caa332 100644 --- a/third_party/README.md +++ b/third_party/README.md @@ -5,6 +5,7 @@ | 目录 | 版本 | 许可证 | 用途 | |------|------|--------|------| | `open62541/` | v1.4.6（commit `50ae40d3`） | MPL-2.0 | OPC UA 协议栈 | +| `libmodbus/` | v3.1.10 | LGPL-2.1+ | edge Modbus RTU/TCP | ## open62541 @@ -17,3 +18,15 @@ 1. 用新 tag 替换 `third_party/open62541/` 目录内容（保留 `LICENSE` 等上游文件） 2. 更新本表版本号与 commit 3. 全量编译并跑 OPC UA 联调（守护进程 + 边缘 + 客户端读写） + +## libmodbus + +- 上游： +- 由 `cmake/libmodbus/CMakeLists.txt` 编译为静态库 `modbus_vendor`，供 `softbus_edge` 链接 +- 若系统已安装 `libmodbus-dev`（pkg-config 可见），优先使用系统库；可用 `-DSOFTBUS_FORCE_VENDOR_LIBMODBUS=ON` 强制 vendor + +### 升级步骤 + +1. 替换 `third_party/libmodbus/` 为新 tag +2. 更新本表版本号 +3. 重新编译 `softbus_edge` 并验证串口读写 diff --git a/third_party/libmodbus b/third_party/libmodbus new file mode 160000 index 0000000..2cbafa3 --- /dev/null +++ b/third_party/libmodbus @@ -0,0 +1 @@ +Subproject commit 2cbafa3113e276c3697d297f68e88d112b53174d diff --git a/tools/CMakeLists.txt b/tools/CMakeLists.txt index 8203880..813c68e 100644 --- a/tools/CMakeLists.txt +++ b/tools/CMakeLists.txt @@ -9,3 +9,6 @@ target_link_libraries(topic_watcher PRIVATE softbus_bus softbus_core ) softbus_apply_platform_libs(topic_watcher) + +add_executable(modbus_scanner modbus_scanner.cpp) +softbus_apply_platform_libs(modbus_scanner) diff --git a/tools/modbus_scanner.cpp b/tools/modbus_scanner.cpp new file mode 100644 index 0000000..f5a9303 --- /dev/null +++ b/tools/modbus_scanner.cpp @@ -0,0 +1,27 @@ +#include +#include + +int main(int argc, char* argv[]) +{ + std::string port = "/dev/ttyUSB0"; + int baud = 9600; + int slaveStart = 1; + int slaveEnd = 10; + for (int i = 1; i < argc; ++i) { + const std::string arg = argv[i]; + if (arg == "--port" && i + 1 < argc) { + port = argv[++i]; + } else if (arg == "--baud" && i + 1 < argc) { + baud = std::stoi(argv[++i]); + } else if (arg == "--slave-start" && i + 1 < argc) { + slaveStart = std::stoi(argv[++i]); + } else if (arg == "--slave-end" && i + 1 < argc) { + slaveEnd = std::stoi(argv[++i]); + } + } + std::cout << "modbus_scanner: draft output for physical_devices.json\n"; + std::cout << " port=" << port << " baud=" << baud + << " slaves=" << slaveStart << "-" << slaveEnd << "\n"; + std::cout << " Run with libmodbus integration to probe registers (Phase 2 tool).\n"; + return 0; +}