diff --git a/CMakeLists.txt b/CMakeLists.txt index 42c60d6..972344b 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -26,6 +26,7 @@ add_subdirectory(src/softbus_pipeline) add_subdirectory(src/softbus_monitor) add_subdirectory(src/softbus_storage) add_subdirectory(src/softbus_opcua) +add_subdirectory(src/softbus_canopen_common) add_subdirectory(src/plugins/protocol_modbus) add_subdirectory(src/plugins/protocol_canopen) add_subdirectory(src/softbus_edge) diff --git a/cmake/SoftbusPlatform.cmake b/cmake/SoftbusPlatform.cmake index 5af3b6c..62f9f8b 100644 --- a/cmake/SoftbusPlatform.cmake +++ b/cmake/SoftbusPlatform.cmake @@ -104,3 +104,18 @@ function(softbus_apply_udev target) target_compile_definitions(${target} PRIVATE SOFTBUS_HAS_UDEV=0) endif() endfunction() + +function(softbus_apply_socketcan target) + if(WIN32) + set(SOFTBUS_HAS_SOCKETCAN OFF) + else() + include(CheckIncludeFileCXX) + check_include_file_cxx("linux/can.h" SOFTBUS_HAS_SOCKETCAN) + endif() + if(SOFTBUS_HAS_SOCKETCAN) + target_compile_definitions(${target} PRIVATE SOFTBUS_HAS_SOCKETCAN=1) + else() + target_compile_definitions(${target} PRIVATE SOFTBUS_HAS_SOCKETCAN=0) + message(WARNING "linux/can.h not found; SocketCAN will be simulated for ${target}") + endif() +endfunction() diff --git a/config/daemon/bindings_global.json b/config/daemon/bindings_global.json index bd7e8e6..8e28cda 100644 --- a/config/daemon/bindings_global.json +++ b/config/daemon/bindings_global.json @@ -5,15 +5,34 @@ "edgeId": "edge01", "physicalPointId": "edge01/phy_modbus_s1_r40001", "logicalPointId": "EdgeDevice_01/MotorTemperature", - "scale": 0.1, - "enabled": false + "enabled": true + }, + { + "bindingId": "bind_speed_01", + "edgeId": "edge01", + "physicalPointId": "edge01/phy_modbus_s1_r40002", + "logicalPointId": "EdgeDevice_01/MotorSpeed", + "enabled": true + }, + { + "bindingId": "bind_current_01", + "edgeId": "edge01", + "physicalPointId": "edge01/phy_modbus_tcp_s255_r40001", + "logicalPointId": "EdgeDevice_01/MotorCurrent", + "enabled": true + }, + { + "bindingId": "bind_voltage_01", + "edgeId": "edge01", + "physicalPointId": "edge01/phy_modbus_tcp_s255_r40002", + "logicalPointId": "EdgeDevice_01/MotorVoltage", + "enabled": true }, { "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/logical_points.json b/config/daemon/logical_points.json index 4947cb0..01f59f0 100644 --- a/config/daemon/logical_points.json +++ b/config/daemon/logical_points.json @@ -18,6 +18,39 @@ "opcuaNodeId": "ns=1;s=DemoSite/DemoSystem/DemoAsset/EdgeDevice_01/MotorTemperature", "dataTopic": "data/EdgeDevice_01/MotorTemperature" }, + { + "logicalPointId": "MotorSpeed", + "displayName": "电机转速", + "signalType": "AI", + "unit": "rpm", + "range": [0, 1000], + "access": "read", + "writable": false, + "opcuaNodeId": "ns=1;s=DemoSite/DemoSystem/DemoAsset/EdgeDevice_01/MotorSpeed", + "dataTopic": "data/EdgeDevice_01/MotorSpeed" + }, + { + "logicalPointId": "MotorCurrent", + "displayName": "电机电流", + "signalType": "AI", + "unit": "A", + "range": [0, 100], + "access": "read", + "writable": false, + "opcuaNodeId": "ns=1;s=DemoSite/DemoSystem/DemoAsset/EdgeDevice_01/MotorCurrent", + "dataTopic": "data/EdgeDevice_01/MotorCurrent" + }, + { + "logicalPointId": "MotorVoltage", + "displayName": "电机电压", + "signalType": "AI", + "unit": "V", + "range": [0, 100], + "access": "read", + "writable": false, + "opcuaNodeId": "ns=1;s=DemoSite/DemoSystem/DemoAsset/EdgeDevice_01/MotorVoltage", + "dataTopic": "data/EdgeDevice_01/MotorVoltage" + }, { "logicalPointId": "ValveOpening", "displayName": "阀门开度", diff --git a/config/daemon/point_table.json b/config/daemon/point_table.json index 3a59970..349e608 100644 --- a/config/daemon/point_table.json +++ b/config/daemon/point_table.json @@ -23,6 +23,51 @@ "opcuaNodeId": "ns=1;s=DemoSite/DemoSystem/DemoAsset/EdgeDevice_01/MotorTemperature", "dataTopic": "data/EdgeDevice_01/MotorTemperature" }, + { + "pointId": "MotorSpeed", + "name": "MotorSpeed", + "displayName": "电机转速", + "signalType": "AI", + "physicalChannel": "AI_Channel_02", + "register": 40002, + "unit": "rpm", + "scale": 1.0, + "range": [0, 1000], + "access": "read", + "writable": false, + "opcuaNodeId": "ns=1;s=DemoSite/DemoSystem/DemoAsset/EdgeDevice_01/MotorSpeed", + "dataTopic": "data/EdgeDevice_01/MotorSpeed" + }, + { + "pointId": "MotorCurrent", + "name": "MotorCurrent", + "displayName": "电机电流", + "signalType": "AI", + "physicalChannel": "AI_Channel_03", + "register": 40001, + "unit": "A", + "scale": 0.1, + "range": [0, 100], + "access": "read", + "writable": false, + "opcuaNodeId": "ns=1;s=DemoSite/DemoSystem/DemoAsset/EdgeDevice_01/MotorCurrent", + "dataTopic": "data/EdgeDevice_01/MotorCurrent" + }, + { + "pointId": "MotorVoltage", + "name": "MotorVoltage", + "displayName": "电机电压", + "signalType": "AI", + "physicalChannel": "AI_Channel_04", + "register": 40002, + "unit": "V", + "scale": 1.0, + "range": [0, 100], + "access": "read", + "writable": false, + "opcuaNodeId": "ns=1;s=DemoSite/DemoSystem/DemoAsset/EdgeDevice_01/MotorVoltage", + "dataTopic": "data/EdgeDevice_01/MotorVoltage" + }, { "pointId": "ValveOpening", "name": "ValveOpening", diff --git a/config/daemon/point_table.yaml b/config/daemon/point_table.yaml index 91b8b34..78e42cb 100644 --- a/config/daemon/point_table.yaml +++ b/config/daemon/point_table.yaml @@ -13,15 +13,47 @@ devices: physicalChannel: AI_Channel_01 register: 40001 unit: "℃" + scale: 0.1 range: [0, 200] access: read dataTopic: data/EdgeDevice_01/MotorTemperature + - pointId: MotorSpeed + displayName: 电机转速 + signalType: AI + physicalChannel: AI_Channel_02 + register: 40002 + unit: rpm + scale: 1.0 + range: [0, 1000] + access: read + dataTopic: data/EdgeDevice_01/MotorSpeed + - pointId: MotorCurrent + displayName: 电机电流 + signalType: AI + physicalChannel: AI_Channel_03 + register: 40001 + unit: A + scale: 0.1 + range: [0, 100] + access: read + dataTopic: data/EdgeDevice_01/MotorCurrent + - pointId: MotorVoltage + displayName: 电机电压 + signalType: AI + physicalChannel: AI_Channel_04 + register: 40002 + unit: V + scale: 1.0 + range: [0, 100] + access: read + dataTopic: data/EdgeDevice_01/MotorVoltage - pointId: ValveOpening displayName: 阀门开度 signalType: AO physicalChannel: AO_Channel_01 register: 40002 unit: "%" + scale: 1.0 range: [0, 100] access: read_write cmdTopic: control/cmd/EdgeDevice_01/ValveOpening/set diff --git a/config/edge/bindings_local.json b/config/edge/bindings_local.json index 1596ca0..5d175b8 100644 --- a/config/edge/bindings_local.json +++ b/config/edge/bindings_local.json @@ -5,7 +5,25 @@ "bindingId": "bind_temp_01", "physicalPointId": "edge01/phy_modbus_s1_r40001", "logicalPointId": "EdgeDevice_01/MotorTemperature", - "enabled": false + "enabled": true + }, + { + "bindingId": "bind_speed_01", + "physicalPointId": "edge01/phy_modbus_s1_r40002", + "logicalPointId": "EdgeDevice_01/MotorSpeed", + "enabled": true + }, + { + "bindingId": "bind_current_01", + "physicalPointId": "edge01/phy_modbus_tcp_s255_r40001", + "logicalPointId": "EdgeDevice_01/MotorCurrent", + "enabled": true + }, + { + "bindingId": "bind_voltage_01", + "physicalPointId": "edge01/phy_modbus_tcp_s255_r40002", + "logicalPointId": "EdgeDevice_01/MotorVoltage", + "enabled": true }, { "bindingId": "bind_valve_01", diff --git a/config/edge/physical_channels.json b/config/edge/physical_channels.json index 66cba85..cb78376 100644 --- a/config/edge/physical_channels.json +++ b/config/edge/physical_channels.json @@ -10,6 +10,22 @@ "dataBits": 8, "stopBits": 1, "pollIntervalMs": 500 + }, + { + "channelId": "eth_modbus_tcp_0", + "protocol": "modbus", + "transport": "tcp_rtu", + "host": "192.168.31.253", + "tcpPort": 502, + "pollIntervalMs": 500 + }, + { + "channelId": "can_0", + "protocol": "canopen", + "transport": "can", + "interface": "can0", + "bitrate": 500000, + "pollIntervalMs": 200 } ] } diff --git a/config/edge/physical_devices.json b/config/edge/physical_devices.json index bba002b..1f282bf 100644 --- a/config/edge/physical_devices.json +++ b/config/edge/physical_devices.json @@ -24,6 +24,65 @@ "writable": true } ] + }, + { + "physicalDeviceId": "phy_modbus_tcp_slave_255", + "channelId": "eth_modbus_tcp_0", + "protocol": "modbus", + "slaveId": 255, + "points": [ + { + "physicalPointId": "phy_modbus_tcp_s255_r40001", + "register": 40001, + "function": 3, + "scale": 0.1, + "signalType": "AI", + "writable": false, + "pollIntervalMs": 500 + }, + { + "physicalPointId": "phy_modbus_tcp_s255_r40002", + "register": 40002, + "function": 3, + "writeFunction": 6, + "scale": 1.0, + "signalType": "AO", + "writable": true, + "pollIntervalMs": 500 + } + ] + }, + { + "physicalDeviceId": "phy_canopen_node_1", + "channelId": "can_0", + "protocol": "canopen", + "nodeId": 1, + "edsPath": "samples/minimal.eds", + "points": [ + { + "physicalPointId": "phy_can_s1_temp", + "objectIndex": 25600, + "subIndex": 1, + "pdoType": "tpdo", + "cobId": 385, + "scale": 0.1, + "signalType": "AI", + "writable": false, + "reportOnChange": true, + "deadband": 0.5, + "pollIntervalMs": 500 + }, + { + "physicalPointId": "phy_can_s1_setpoint", + "objectIndex": 25601, + "subIndex": 1, + "pdoType": "sdo", + "scale": 1.0, + "signalType": "AO", + "writable": true, + "pollIntervalMs": 1000 + } + ] } ] } diff --git a/config/edge/samples/minimal.eds b/config/edge/samples/minimal.eds new file mode 100644 index 0000000..5ed7d6a --- /dev/null +++ b/config/edge/samples/minimal.eds @@ -0,0 +1,47 @@ +[DeviceInfo] +VendorName=Softbus +ProductName=DemoCANopenNode + +[1000] +ParameterName=Device type +DataType=0x0007 +AccessType=ro +DefaultValue=0x00000191 + +[1001] +ParameterName=Error register +DataType=0x0005 +AccessType=ro +DefaultValue=0x00 + +[1800] +ParameterName=TPDO communication parameter +ObjectType=0x09 + +[1800sub1] +ParameterName=COB-ID used by TPDO +DataType=0x0007 +AccessType=rw +DefaultValue=0x00000180 + +[1800sub2] +ParameterName=Transmission type +DataType=0x0005 +AccessType=rw +DefaultValue=0xFF + +[1A00] +ParameterName=TPDO mapping parameter +ObjectType=0x09 + +[1A00sub1] +ParameterName=1st mapped object +DataType=0x0007 +AccessType=rw +DefaultValue=0x64000110 + +[6400] +ParameterName=Temperature +DataType=0x0006 +AccessType=ro +DefaultValue=0x00 diff --git a/config/edge/samples/physical_channels_can.json b/config/edge/samples/physical_channels_can.json new file mode 100644 index 0000000..a2e606f --- /dev/null +++ b/config/edge/samples/physical_channels_can.json @@ -0,0 +1,12 @@ +{ + "channels": [ + { + "channelId": "can_0", + "protocol": "canopen", + "transport": "can", + "interface": "can0", + "bitrate": 500000, + "pollIntervalMs": 200 + } + ] +} diff --git a/config/edge/samples/physical_devices_can.json b/config/edge/samples/physical_devices_can.json new file mode 100644 index 0000000..39db45a --- /dev/null +++ b/config/edge/samples/physical_devices_can.json @@ -0,0 +1,36 @@ +{ + "devices": [ + { + "physicalDeviceId": "phy_canopen_node_1", + "channelId": "can_0", + "protocol": "canopen", + "nodeId": 1, + "edsPath": "samples/minimal.eds", + "points": [ + { + "physicalPointId": "phy_can_s1_temp", + "objectIndex": 25600, + "subIndex": 1, + "pdoType": "tpdo", + "cobId": 385, + "scale": 0.1, + "signalType": "AI", + "writable": false, + "reportOnChange": true, + "deadband": 0.5, + "pollIntervalMs": 500 + }, + { + "physicalPointId": "phy_can_s1_setpoint", + "objectIndex": 25601, + "subIndex": 1, + "pdoType": "sdo", + "scale": 1.0, + "signalType": "AO", + "writable": true, + "pollIntervalMs": 1000 + } + ] + } + ] +} diff --git a/docs/edge/tcp_rtu_callstack.md b/docs/edge/tcp_rtu_callstack.md new file mode 100644 index 0000000..bcd5746 --- /dev/null +++ b/docs/edge/tcp_rtu_callstack.md @@ -0,0 +1,251 @@ +# Edge Modbus TCP 点调用栈（含 RTU-over-TCP） + +本文以配置点 **`phy_modbus_tcp_s1_r40001`** 为例，说明从 `physical_devices.json` 到硬件读数、再到 uplink 上报的完整调用路径。 + +运行时物理点 ID 为 `{edgeId}/{physicalPointId}`，例如 `edge01/phy_modbus_tcp_s1_r40001`。 + +相关配置： + +- 点：`config/edge/physical_devices.json` → `phy_modbus_tcp_slave_255` +- 通道：`config/edge/physical_channels.json` → `eth_modbus_tcp_0`，`transport: "tcp_rtu"` + +--- + +## 1. 设计要点：没有独立的 TcpRtuIngress + +edge **不按传输方式** 创建 ingress，而按 **`protocol`** 创建： + +| 配置字段 | 作用 | +|----------|------| +| `protocol: "modbus"` | 走 `ModbusIngress`（串口 + TCP + RTU-over-TCP 共用） | +| `transport: "rtu"` | driver 层：`libmodbus` 串口 | +| `transport: "tcp"` | driver 层：`libmodbus` 标准 Modbus TCP（MBAP） | +| `transport: "tcp_rtu"` | driver 层：`TcpRtuSession`（RTU 帧跑在 TCP 上） | + +自定义协议分支在 **`ModbusChannelManager`** 内，由 `transport` 字段触发，上层 `ModbusIngress` 无感知。 + +``` +physical_devices.json (点) + ↓ +ModbusIngress（统一采集） + ↓ +ModbusChannelManager::openChannel / readRegisters + ├─ transport: rtu → libmodbus 串口 + ├─ transport: tcp → libmodbus 标准 TCP + └─ transport: tcp_rtu → TcpRtuSession +``` + +--- + +## 2. 配置锚点 + +```json +{ + "physicalDeviceId": "phy_modbus_tcp_slave_255", + "channelId": "eth_modbus_tcp_0", + "protocol": "modbus", + "slaveId": 255, + "points": [ + { + "physicalPointId": "phy_modbus_tcp_s1_r40001", + "register": 40001, + "function": 3, + "scale": 0.1, + "pollIntervalMs": 500 + } + ] +} +``` + +```json +{ + "channelId": "eth_modbus_tcp_0", + "protocol": "modbus", + "transport": "tcp_rtu", + "host": "192.168.31.253", + "tcpPort": 502 +} +``` + +--- + +## 3. 阶段 A：启动（一次性） + +```mermaid +flowchart TD + A["main:52 runtime.init()"] --> B["EdgeRuntime::init:35 loadDevices"] + B --> C["PhysicalRegistry::loadDevices:89 qualifyPointId"] + C --> D["rebuildPollIndex:143 写入 pointIndex_"] + D --> E["EdgeRuntime::rebuildIngress:16 createIngress"] + E --> F["DeviceFactory::createIngress:34 ModbusIngress"] + F --> G["ModbusIngress::open:24 openChannel(eth_modbus_tcp_0)"] + G --> H{"transport == tcp_rtu?"} + H -->|是| I["ModbusChannel::openChannel:39-51 TcpRtuSession::connect:31"] + G --> K["PollPlanner::reset:18 注册点周期 500ms"] +``` + +| 步骤 | 函数 | 文件:行号 | 说明 | +|------|------|-----------|------| +| 1 | `main` | `src/softbus_edge/src/main.cpp:52` | `runtime.init(config, ...)` | +| 2 | `EdgeRuntime::init` | `src/softbus_edge/src/EdgeRuntime.cpp:35` | `physicalRegistry_.loadDevices(..., edgeId)` | +| 3 | `PhysicalRegistry::loadDevices` | `src/softbus_edge/src/PhysicalRegistry.cpp:89-91` | JSON 解析，`qualifyPointId` → `edge01/phy_modbus_tcp_s1_r40001` | +| 4 | `PhysicalRegistry::rebuildPollIndex` | `PhysicalRegistry.cpp:142-145` | 组装 `ResolvedPhysicalPoint{point, device, channel}` 写入 `pointIndex_` | +| 5 | `EdgeRuntime::rebuildIngress` | `EdgeRuntime.cpp:16-19` | `createIngress` → `ingress_->open()` | +| 6 | `DeviceFactory::createIngress` | `src/softbus_edge/src/DeviceFactory.cpp:33-35` | 创建 `ModbusIngress` | +| 7 | `ModbusIngress::open` | `src/softbus_edge/src/ingress/modbus/ModbusIngress.cpp:23-24` | 对所有 `protocol==modbus` 通道调用 `openChannel` | +| 8 | `ModbusChannelManager::openChannel` | `src/softbus_edge/src/driver/modbus/ModbusChannel.cpp:39-51` | `tcp_rtu` → `TcpRtuSession::connect(host, port)` | +| 9 | `TcpRtuSession::connect` | `src/softbus_edge/src/driver/modbus/TcpRtuSession.cpp:31-62` | TCP `socket` + `connect`，保存 `socketFd_` | +| 10 | `PollPlanner::reset` | `src/softbus_edge/src/scheduler/PollPlanner.cpp:9-18` | 注册该点 `intervalMs=500` | + +启动成功后日志示例： + +``` +Opened RTU-over-TCP channel eth_modbus_tcp_0 192.168.31.253:502 +``` + +--- + +## 4. 阶段 B：运行循环（每 50ms tick，点每 500ms 到期） + +主循环：`main.cpp:76` → `EdgeRuntime::tick()` → `pollIngress()` → `ModbusIngress::poll()`。 + +```mermaid +sequenceDiagram + participant M as main.cpp:76 + participant RT as EdgeRuntime::tick:153 + participant PI as EdgeRuntime::pollIngress:141 + participant MI as ModbusIngress::poll:57 + participant PP as PollPlanner + participant MPS as ModbusPollScheduler + participant MCM as ModbusChannelManager + participant TR as TcpRtuSession + participant UL as UplinkClient + + M->>RT: tick() + RT->>PI: pollIngress() + PI->>MI: ingress_->poll() + MI->>PP: duePointIds() + MI->>MI: resolvePoint + MI->>MPS: planBatches() + MI->>MCM: readRegisters(startAddr, count) + MCM->>TR: readHoldingRegisters(255, 0, 2) + TR->>TR: transaction RTU send/recv + TR-->>MCM: values + MCM-->>MI: values + MI->>MI: buildModbusFrame RawPacket + PI->>UL: sendRawPacket (uplink 已连接时) +``` + +### 4.1 文本调用栈（带行号） + +``` +main.cpp:76 + └─ EdgeRuntime::tick() EdgeRuntime.cpp:153-156 + └─ EdgeRuntime::pollIngress() EdgeRuntime.cpp:141-150 + └─ ModbusIngress::poll() ModbusIngress.cpp:57-130 + │ + ├─ PollPlanner::duePointIds() PollPlanner.cpp:22-31 + │ → 含 "edge01/phy_modbus_tcp_s1_r40001"（500ms 周期） + │ + ├─ PhysicalRegistry::resolvePoint() PhysicalRegistry.cpp:183-190 + │ → ResolvedPhysicalPoint { + │ point.registerAddress = 40001 + │ device.slaveId = 255 + │ channel.channelId = "eth_modbus_tcp_0" + │ channel.transport = "tcp_rtu" + │ } + │ + ├─ ModbusPollScheduler::planBatches() ModbusPollScheduler.cpp:31-104 + │ → modbusAddress(40001) = 0 + │ → 常与 r40002 合并 batch{startAddr=0, count=2} + │ + ├─ ModbusChannelManager::readRegisters() ModbusChannel.cpp:121-132 + │ → channels_["eth_modbus_tcp_0"].tcpRtu 非空 + │ └─ TcpRtuSession::readHoldingRegisters(255, 0, 2) TcpRtuSession.cpp:97-137 + │ ├─ 组 PDU: FF 03 00 00 00 02 :106-113 + │ └─ TcpRtuSession::transaction(pdu) :73-94 + │ ├─ crc16 + append CRC :79-82 + │ ├─ send(socketFd_) :84 + │ └─ recv(socketFd_) :88-94 + │ + ├─ ChangeFilter::shouldReport() ModbusIngress.cpp:112-115 + ├─ buildModbusFrame(255, raw) ModbusIngress.cpp:44-54, :124 + ├─ packet.sourceId = "edge01/phy_modbus_tcp_s1_r40001" :122 + ├─ packet.protocolTag = "modbus" :123 + └─ PollPlanner::markPolled() ModbusIngress.cpp:126 + PollPlanner.cpp:34-41 + + └─ UplinkClient::sendRawPacket(packet) EdgeRuntime.cpp:147-148 + └─ UplinkClient::sendRawPacket() UplinkClient.cpp:30-32 + └─ TcpUplinkClient::sendRawPacket() (softbus_transport) +``` + +--- + +## 5. 自定义协议触发点（仅两处） + +**打开通道（一次性）：** + +```39:51:src/softbus_edge/src/driver/modbus/ModbusChannel.cpp + if (isTcpRtuTransport(channel)) { + handle.tcpRtu = std::make_unique(); + if (!handle.tcpRtu->connect(channel.host, channel.tcpPort)) { + // ... + } +``` + +**读寄存器（每次轮询）：** + +```129:131:src/softbus_edge/src/driver/modbus/ModbusChannel.cpp + if (it->second.tcpRtu) { + return it->second.tcpRtu->readHoldingRegisters( + point.device.slaveId, startAddr, count); +``` + +`ModbusIngress` 在 `ModbusIngress.cpp:89` 统一调用 `readRegisters`，不区分底层实现。 + +--- + +## 6. 本点一次读写的实际参数 + +| 项 | 值 | +|----|-----| +| `physicalPointId` | `edge01/phy_modbus_tcp_s1_r40001` | +| `channelId` | `eth_modbus_tcp_0` | +| `slaveId` | `255` → RTU 帧首字节 `0xFF` | +| `register 40001` | 内部地址 `startAddr = 0` | +| 合并读 | 常与 `r40002` 一次 `count=2` | +| 网线 payload | `FF 03 00 00 00 02` + CRC（**非** Modbus TCP MBAP） | +| 上行 RawPacket | `protocolTag="modbus"`，payload 为模拟 0x03 响应帧 | + +--- + +## 7. 为何需要 tcp_rtu + +部分设备（如 `192.168.31.253:502`）在 TCP 502 端口上收发的是 **Modbus RTU 帧**，而非标准 Modbus TCP（无 MBAP 事务头）。 + +| 方式 | 结果 | +|------|------| +| `transport: "tcp"` + libmodbus | 连接成功，但读寄存器 `Invalid data` | +| `transport: "tcp_rtu"` + TcpRtuSession | 读 slave 255 寄存器成功 | + +daemon 侧仍用 `protocol_modbus` 解析上行 `RawPacket`，**无需知道** 底层是串口还是 RTU-over-TCP。 + +--- + +## 8. 与串口点的对比 + +| 层级 | 串口 `phy_modbus_s1_r40001` | TCP `phy_modbus_tcp_s1_r40001` | +|------|----------------------------|--------------------------------| +| Ingress | 同一 `ModbusIngress::poll` | 同一 `ModbusIngress::poll` | +| openChannel | `ModbusChannel.cpp:55-76` libmodbus RTU | `ModbusChannel.cpp:39-51` TcpRtuSession | +| readRegisters | `modbus_read_registers` | `TcpRtuSession::readHoldingRegisters` | +| 上行 protocolTag | `modbus` | `modbus` | + +--- + +## 9. 相关文档 + +- [softbus_edge.md](../softbus_edge.md) — edge 总览 +- [edge_device_setup.md](../edge_device_setup.md) — 配置与启动 +- [UPLINK_WIRE_PROTOCOL.md](../UPLINK_WIRE_PROTOCOL.md) — uplink 线协议 diff --git a/docs/softbus_edge.md b/docs/softbus_edge.md index 99bf37b..7963776 100644 --- a/docs/softbus_edge.md +++ b/docs/softbus_edge.md @@ -8,7 +8,7 @@ `softbus_edge` 是边缘代理进程，运行在靠近现场设备的一侧，负责： -- **物理域采集（ingress）**：按 `physical_*.json` 轮询 Modbus RTU/TCP 等设备，产出 `RawPacket` +- **物理域采集（ingress）**：按 `physical_*.json` 轮询 Modbus / CANopen / 裸以太网设备，产出 `RawPacket` - **上行（egress/uplink）**：经 TCP 将 `RawPacket`、心跳、注册信息发送给 `softbus_daemon` - **下行（egress）**：接收 daemon 下发的 `ExecuteCommand`，写回物理设备（Modbus 写寄存器等） @@ -16,7 +16,7 @@ edge **不**加载逻辑点表、**不**操作 OPC UA、**不**运行 `protocol_ | 模式 | 说明 | |------|------| -| **正常模式** | libmodbus + `ModbusIngress`，读真实硬件 | +| **正常模式** | `CompositeIngress`（Modbus + CANopen + Eth），读真实硬件 | | **Mock 模式** | `--mock`，无硬件，模拟 Modbus 帧上行 | | **独立模式** | daemon 未启动时 edge 仍可运行，后台重连 uplink | @@ -93,8 +93,8 @@ edge 上行时**不做** binding 初筛：所有已配置物理点都会采集 → 物理点 ID 自动加前缀：edge01/phy_modbus_s1_r40001 4. [可选] BindingRegistry.loadLocal(bindings_local.json) 5. UplinkClient + CommandExecutor 初始化 -6. DeviceFactory.createIngress → ModbusIngress 或 MockModbusIngress -7. ModbusChannelManager.openChannel（逐通道 libmodbus 连接） +6. DeviceFactory.createIngress → CompositeIngress（按 protocol 创建子 ingress） +7. 各 ChannelManager.openChannel（libmodbus / SocketCAN / asio） 8. uplink.connect()；成功则 sendEdgeRegister(edgeId) → 失败仅 WARN，edge 独立运行并后台重连 ``` @@ -122,7 +122,11 @@ edge 上行时**不做** binding 初筛：所有已配置物理点都会采集 | 字段 | 说明 | |------|------| | `transport` | `rtu`（串口）或 `tcp`（Modbus TCP，需 `host` + `tcpPort`） | -| `pollIntervalMs` | 该通道下所有点的轮询周期（当前整通道统一） | +| `pollIntervalMs` | 通道默认轮询周期（点可覆盖） | +| `interface` / `bitrate` | CANopen 通道：`can0`、`500000` | +| `host` / `tcpPort` | Modbus TCP 或裸以太网 | + +CANopen 通道示例见 `config/edge/physical_channels_can.json`。 ### 3.4 物理设备与点（`physical_devices.json`） @@ -141,7 +145,10 @@ edge 上行时**不做** binding 初筛：所有已配置物理点都会采集 "function": 3, "scale": 0.1, "signalType": "AI", - "writable": false + "writable": false, + "pollIntervalMs": 1000, + "reportOnChange": true, + "deadband": 0.5 } ] } @@ -149,8 +156,22 @@ edge 上行时**不做** binding 初筛：所有已配置物理点都会采集 } ``` +**点级可选字段：** + +| 字段 | 说明 | +|------|------| +| `pollIntervalMs` | 覆盖通道默认周期 | +| `reportOnChange` | `true` 时启用变化上报 | +| `deadband` | 工程值死区（配合 `reportOnChange`） | +| `objectIndex` / `subIndex` | CANopen 对象字典索引 | +| `pdoType` | `tpdo` / `sdo` / `rpdo` | +| `cobId` | PDO COB-ID 覆盖 | +| `nodeId` / `edsPath` | 设备级 CANopen 字段 | + 运行时 `physicalPointId` 会变为 `{edgeId}/{physicalPointId}`，作为 `RawPacket.sourceId` 上报。 +CANopen 示例见 `config/edge/physical_devices_can.json` 与 `config/edge/samples/minimal.eds`。 + ### 3.5 热加载 | 信号 | 行为 | @@ -163,18 +184,38 @@ edge 上行时**不做** binding 初筛：所有已配置物理点都会采集 ## 4. 组件架构 +### 4.0 目录结构 + +``` +include/softbus_edge/ + ingress/ IIngressPort, CompositeIngress + modbus/ ModbusIngress, MockModbusIngress + can/ CanIngress + eth/ EthIngress + driver/ + modbus/ ModbusChannelManager (libmodbus) + can/ CanChannelManager (SocketCAN) + eth/ EthChannelManager (asio) + egress/ + modbus/ ModbusEgress + can/ CanEgress (SDO 写) + eth/ EthEgress + scheduler/ ModbusPollScheduler, PollPlanner, ChangeFilter + canopen/ NmtManager, SdoClient, PdoMapper + config/ runtime/ factory/ discovery/ +``` + ``` main.cpp └── 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 封装 + ├── PhysicalRegistry + ├── CompositeIngress + │ ├── ModbusIngress + ModbusPollScheduler / PollPlanner / ChangeFilter + │ ├── CanIngress + NmtManager / PdoMapper / SdoClient + │ └── EthIngress + ├── ModbusEgress / CanEgress / EthEgress + ├── CommandExecutor ← 按 protocol 分发写 + └── UplinkClient ``` ### 4.1 IIngressPort — 入站抽象 @@ -185,9 +226,13 @@ main.cpp | `close()` | 关闭采集通道 | | `poll()` | 轮询，返回 `vector` | -**ModbusIngress**：按 `pollIntervalMs` 遍历 `pollPoints()`，经 `ModbusChannelManager` 读寄存器，组装 Modbus 0x03 响应帧作为 `payload`。 +**ModbusIngress**：`PollPlanner` 调度到期点 → `ModbusPollScheduler` 合并相邻寄存器批量读 → `ChangeFilter` 变化过滤 → `RawPacket`。 -**MockModbusIngress**：无硬件，生成递增值模拟帧。 +**CanIngress**：启动时 NMT Start + 加载 EDS；TPDO 监听 + SDO 慢点轮询；`protocolTag: "canopen"`。 + +**EthIngress**：裸 TCP/UDP 收字节；`protocolTag: "raw_eth"`。 + +**MockModbusIngress**：`--mock` 模式，无硬件模拟帧。 ### 4.2 ModbusChannelManager — 硬件 I/O @@ -214,10 +259,12 @@ main.cpp ### 4.4 CommandExecutor — 下行写 -daemon 经 `bindings_global` 在命令中注入 `physicalPointId` + `edgeId`，edge 直接写 Modbus： +daemon 经 `bindings_global` 在命令中注入 `physicalPointId` + `edgeId`，edge 按协议写： ``` -ExecuteCommand.params.physicalPointId → PhysicalRegistry → ModbusEgress.write +modbus → ModbusEgress +canopen → CanEgress (SDO download) +raw → EthEgress ``` 仅当命令**缺少** `physicalPointId` 时，才用 `bindings_local.json` 做 `logicalPointId → physicalPointId` 备用解析。 @@ -310,8 +357,10 @@ while (!shouldStop) { | `softbus_registry` | BindingRegistry（仅 local 可选） | | `softbus_transport` | TcpUplinkClient、UplinkWire | | `softbus_api` | ExecuteCommand、SoftbusAck | -| **libmodbus** | Modbus RTU/TCP 硬件 I/O（vendor 或系统包） | -| asio | TcpUplinkClient 底层 | +| **libmodbus** | Modbus RTU/TCP 硬件 I/O | +| **SocketCAN** | Linux `linux/can.h`（`SOFTBUS_HAS_SOCKETCAN`） | +| **softbus_canopen_common** | EDS 解析、对象字典 | +| asio | Uplink + EthChannel | --- @@ -351,68 +400,46 @@ kill -HUP $(pidof softbus_edge) --- -## 9. 未来扩展：CANopen 与以太网设备 +## 9. 多协议与物理层优化（已实现） -当前仅 **Modbus RTU/TCP** 在 edge 侧有完整 ingress/egress。CANopen 与更多以太网协议按同一双域模型扩展。 +### 9.1 多 Ingress 并存 -### 9.1 目标架构 +`physical_channels.json` 中可混合 `protocol: modbus | canopen | raw`，`DeviceFactory` 为每种协议创建子 ingress，由 `CompositeIngress` 统一 `poll()`。 -``` -physical_channels.json ──→ ChannelDriver（按 protocol 选择） -physical_devices.json ──→ PhysicalRegistry - │ - ┌───────────┼───────────┐ - ▼ ▼ ▼ - ModbusIngress CanIngress EthIngress（规划） - │ │ │ - └───────────┴───────────┘ - ▼ - RawPacket → UplinkClient -``` +### 9.2 Modbus 采集优化 -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 全量 | +| **ModbusPollScheduler** | 同 (channel, slave, function) 下合并相邻寄存器，单次最多 125 寄存器 | +| **PollPlanner** | 每点独立 `pollIntervalMs`（缺省继承通道） | +| **ChangeFilter** | `reportOnChange` + `deadband` 物理层过滤 | -### 9.3 以太网设备（规划） +### 9.3 CANopen（edge 已实现） -| 类型 | edge 侧 | 说明 | -|------|---------|------| -| **Modbus TCP** | 已支持 | `physical_channels` 设 `transport: tcp`，`host` + `tcpPort` | -| **裸 TCP / 自定义帧** | `EthIngress`（规划） | 按 channel 配置帧格式或插件化 driver | -| **UDP 组播** | 待定 | 适用于部分工业相机/传感器 | +| 组件 | 说明 | +|------|------| +| **CanChannelManager** | SocketCAN 收发（无头文件时模拟降级） | +| **NmtManager** | 启动时 NMT Start Remote Node | +| **EdsDcfLoader** | `softbus_canopen_common` 解析 EDS | +| **PdoMapper** | TPDO 订阅表 + 帧解包 | +| **SdoClient** | SDO upload（慢点）/ download（下行写） | +| **daemon** | `protocol_canopen.so` 解析仍占位 | -以太网通道示例（Modbus TCP，已可用）： +CAN 前置：`sudo ip link set can0 up type can bitrate 500000` -```json -{ - "channelId": "eth_modbus_0", - "protocol": "modbus", - "transport": "tcp", - "host": "192.168.1.10", - "tcpPort": 502, - "pollIntervalMs": 200 -} -``` +### 9.4 裸以太网（骨架） -### 9.4 edge 侧待增强（Modbus 及通用） +`protocol: "raw"` + `transport: tcp|udp` → `EthIngress` 原样上行；`EthEgress` 支持原始字节写。 + +Modbus TCP 仍走 `protocol: modbus` + `transport: tcp`。 + +### 9.5 待增强 | 项 | 说明 | |----|------| -| **ModbusPollScheduler** | 合并相邻寄存器读，减少串口占用 | -| **按点 pollInterval** | 不同信号类型不同采集周期 | -| **变化上报 / deadband** | 物理层过滤，减少 uplink 流量 | | **DeviceDiscovery** | 总线扫描、动态写入 `physical_devices` | -| **多 ingress 并存** | 同一 edge 上 Modbus + CANopen 并行 | - -上述优化均在**物理域**完成，不改变 daemon 双域模型。 +| **daemon CANopen 解析** | `CanopenParser` 完整实现 | +| **Eth 帧解析** | 自定义帧格式插件化 | --- @@ -430,6 +457,7 @@ daemon 侧已有对应解析插件（`protocol_modbus`、`protocol_canopen`） | 文档 | 内容 | |------|------| +| [edge/tcp_rtu_callstack.md](edge/tcp_rtu_callstack.md) | Modbus TCP 点完整调用栈（含 RTU-over-TCP） | | [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 | diff --git a/src/plugins/protocol_canopen/CMakeLists.txt b/src/plugins/protocol_canopen/CMakeLists.txt index 77db819..cce1943 100644 --- a/src/plugins/protocol_canopen/CMakeLists.txt +++ b/src/plugins/protocol_canopen/CMakeLists.txt @@ -6,6 +6,9 @@ add_library(protocol_canopen SHARED softbus_export_include(protocol_canopen "${CMAKE_CURRENT_SOURCE_DIR}/include") -target_link_libraries(protocol_canopen PRIVATE softbus_core) +target_link_libraries(protocol_canopen PRIVATE + softbus_core + softbus_canopen_common +) softbus_add_plugin(protocol_canopen) softbus_apply_platform_libs(protocol_canopen) diff --git a/src/plugins/protocol_canopen/include/canopen/CanopenNodeManager.h b/src/plugins/protocol_canopen/include/canopen/CanopenNodeManager.h deleted file mode 100644 index 75a4fa6..0000000 --- a/src/plugins/protocol_canopen/include/canopen/CanopenNodeManager.h +++ /dev/null @@ -1,19 +0,0 @@ -#pragma once - -#include -#include -#include - -namespace canopen { - -class CanopenNodeManager { -public: - bool registerNode(uint8_t nodeId); - bool setState(uint8_t nodeId, const std::string& state); - std::string state(uint8_t nodeId) const; - -private: - std::map states_; -}; - -} // namespace canopen diff --git a/src/plugins/protocol_canopen/include/canopen/CanopenObjectDictionary.h b/src/plugins/protocol_canopen/include/canopen/CanopenObjectDictionary.h deleted file mode 100644 index b4b1e51..0000000 --- a/src/plugins/protocol_canopen/include/canopen/CanopenObjectDictionary.h +++ /dev/null @@ -1,25 +0,0 @@ -#pragma once - -#include - -#include -#include -#include - -namespace canopen { - -class CanopenObjectDictionary { -public: - bool loadEds(const std::string& path); - bool loadDcf(const std::string& path); - bool hasEntry(uint16_t index, uint8_t subIndex) const; - -private: - struct Entry { - std::string name; - std::string dataType; - }; - std::map entries_; -}; - -} // namespace canopen diff --git a/src/plugins/protocol_canopen/src/canopen_plugin.cpp b/src/plugins/protocol_canopen/src/canopen_plugin.cpp index 3efc15a..44c358d 100644 --- a/src/plugins/protocol_canopen/src/canopen_plugin.cpp +++ b/src/plugins/protocol_canopen/src/canopen_plugin.cpp @@ -1,9 +1,6 @@ #include #include #include -#include -#include -#include #include namespace canopen { @@ -18,20 +15,6 @@ softbus::core::ParseResult CanopenParser::parseFrame(const uint8_t*, std::size_t return {}; } -bool CanopenObjectDictionary::loadEds(const std::string&) { return false; } -bool CanopenObjectDictionary::loadDcf(const std::string&) { return false; } -bool CanopenObjectDictionary::hasEntry(uint16_t, uint8_t) const { return false; } - -bool EdsDcfLoader::load(const std::string&, CanopenObjectDictionary*) { return false; } - -bool CanopenNodeManager::registerNode(uint8_t) { return false; } -bool CanopenNodeManager::setState(uint8_t, const std::string&) { return false; } -std::string CanopenNodeManager::state(uint8_t nodeId) const -{ - const auto it = states_.find(nodeId); - return it == states_.end() ? "Unknown" : it->second; -} - bool CanopenMapper::mapToDom(uint16_t, uint8_t, softbus::core::DOMMessage*) { return false; } } // namespace canopen diff --git a/src/softbus_canopen_common/CMakeLists.txt b/src/softbus_canopen_common/CMakeLists.txt new file mode 100644 index 0000000..e6b2db9 --- /dev/null +++ b/src/softbus_canopen_common/CMakeLists.txt @@ -0,0 +1,10 @@ +softbus_package(softbus_canopen_common) + +add_library(softbus_canopen_common STATIC + src/CanopenObjectDictionary.cpp + src/EdsDcfLoader.cpp +) + +softbus_export_include(softbus_canopen_common "${CMAKE_CURRENT_SOURCE_DIR}/include") + +target_link_libraries(softbus_canopen_common PUBLIC softbus_core) diff --git a/src/softbus_canopen_common/include/canopen/CanopenObjectDictionary.h b/src/softbus_canopen_common/include/canopen/CanopenObjectDictionary.h new file mode 100644 index 0000000..52c4ba5 --- /dev/null +++ b/src/softbus_canopen_common/include/canopen/CanopenObjectDictionary.h @@ -0,0 +1,35 @@ +#pragma once + +#include + +#include +#include +#include +#include +#include + +namespace canopen { + +class CanopenObjectDictionary { +public: + bool loadEds(const std::string& path); + bool loadDcf(const std::string& path); + bool hasEntry(uint16_t index, uint8_t subIndex) const; + std::optional entry(uint16_t index, uint8_t subIndex) const; + std::vector tpdoDefinitions() const; + std::vector rpdoDefinitions() const; + + void registerEntry(const ObjectEntry& entry); + void registerTpdo(const PdoDefinition& pdo); + void registerRpdo(const PdoDefinition& pdo); + void clear(); + +private: + static uint32_t makeKey(uint16_t index, uint8_t subIndex); + + std::map entries_; + std::map tpdoByCommIndex_; + std::map rpdoByCommIndex_; +}; + +} // namespace canopen diff --git a/src/softbus_canopen_common/include/canopen/CanopenTypes.h b/src/softbus_canopen_common/include/canopen/CanopenTypes.h new file mode 100644 index 0000000..4d9af91 --- /dev/null +++ b/src/softbus_canopen_common/include/canopen/CanopenTypes.h @@ -0,0 +1,30 @@ +#pragma once + +#include +#include +#include + +namespace canopen { + +struct ObjectEntry { + uint16_t index{0}; + uint8_t subIndex{0}; + std::string name; + std::string dataType; + uint32_t defaultValue{0}; +}; + +struct PdoMappingEntry { + uint16_t index{0}; + uint8_t subIndex{0}; + uint8_t bitLength{16}; +}; + +struct PdoDefinition { + uint16_t commParamIndex{0}; + uint32_t cobId{0}; + uint8_t transmissionType{0}; + std::vector mappings; +}; + +} // namespace canopen diff --git a/src/plugins/protocol_canopen/include/canopen/EdsDcfLoader.h b/src/softbus_canopen_common/include/canopen/EdsDcfLoader.h similarity index 100% rename from src/plugins/protocol_canopen/include/canopen/EdsDcfLoader.h rename to src/softbus_canopen_common/include/canopen/EdsDcfLoader.h diff --git a/src/softbus_canopen_common/src/CanopenObjectDictionary.cpp b/src/softbus_canopen_common/src/CanopenObjectDictionary.cpp new file mode 100644 index 0000000..f54a307 --- /dev/null +++ b/src/softbus_canopen_common/src/CanopenObjectDictionary.cpp @@ -0,0 +1,78 @@ +#include +#include + +namespace canopen { + +uint32_t CanopenObjectDictionary::makeKey(uint16_t index, uint8_t subIndex) +{ + return (static_cast(index) << 8) | subIndex; +} + +bool CanopenObjectDictionary::loadEds(const std::string& path) +{ + EdsDcfLoader loader; + return loader.load(path, this); +} + +bool CanopenObjectDictionary::loadDcf(const std::string& path) +{ + return loadEds(path); +} + +bool CanopenObjectDictionary::hasEntry(uint16_t index, uint8_t subIndex) const +{ + return entries_.count(makeKey(index, subIndex)) != 0; +} + +std::optional CanopenObjectDictionary::entry(uint16_t index, uint8_t subIndex) const +{ + const auto it = entries_.find(makeKey(index, subIndex)); + if (it == entries_.end()) { + return std::nullopt; + } + return it->second; +} + +std::vector CanopenObjectDictionary::tpdoDefinitions() const +{ + std::vector out; + out.reserve(tpdoByCommIndex_.size()); + for (const auto& [_, def] : tpdoByCommIndex_) { + out.push_back(def); + } + return out; +} + +std::vector CanopenObjectDictionary::rpdoDefinitions() const +{ + std::vector out; + out.reserve(rpdoByCommIndex_.size()); + for (const auto& [_, def] : rpdoByCommIndex_) { + out.push_back(def); + } + return out; +} + +void CanopenObjectDictionary::registerEntry(const ObjectEntry& entry) +{ + entries_[makeKey(entry.index, entry.subIndex)] = entry; +} + +void CanopenObjectDictionary::registerTpdo(const PdoDefinition& pdo) +{ + tpdoByCommIndex_[pdo.commParamIndex] = pdo; +} + +void CanopenObjectDictionary::registerRpdo(const PdoDefinition& pdo) +{ + rpdoByCommIndex_[pdo.commParamIndex] = pdo; +} + +void CanopenObjectDictionary::clear() +{ + entries_.clear(); + tpdoByCommIndex_.clear(); + rpdoByCommIndex_.clear(); +} + +} // namespace canopen diff --git a/src/softbus_canopen_common/src/EdsDcfLoader.cpp b/src/softbus_canopen_common/src/EdsDcfLoader.cpp new file mode 100644 index 0000000..83e0776 --- /dev/null +++ b/src/softbus_canopen_common/src/EdsDcfLoader.cpp @@ -0,0 +1,190 @@ +#include + +#include +#include +#include + +namespace canopen { + +namespace { + +uint16_t parseHexIndex(const std::string& section) +{ + if (section.empty()) { + return 0; + } + return static_cast(std::stoul(section, nullptr, 16)); +} + +uint32_t parseHexValue(const std::string& value) +{ + std::string trimmed = value; + if (trimmed.size() >= 2 && trimmed[0] == '0' && + (trimmed[1] == 'x' || trimmed[1] == 'X')) { + return static_cast(std::stoul(trimmed, nullptr, 16)); + } + if (!trimmed.empty() && std::isxdigit(static_cast(trimmed[0]))) { + return static_cast(std::stoul(trimmed, nullptr, 16)); + } + return static_cast(std::stoul(trimmed, nullptr, 10)); +} + +bool isTpdoMappingIndex(uint16_t index) +{ + return index >= 0x1A00 && index <= 0x1BFF; +} + +bool isRpdoMappingIndex(uint16_t index) +{ + return index >= 0x1600 && index <= 0x17FF; +} + +bool isTpdoCommIndex(uint16_t index) +{ + return index >= 0x1800 && index <= 0x19FF; +} + +bool isRpdoCommIndex(uint16_t index) +{ + return index >= 0x1400 && index <= 0x15FF; +} + +} // namespace + +bool EdsDcfLoader::load(const std::string& path, CanopenObjectDictionary* dictionary) +{ + if (!dictionary) { + return false; + } + std::ifstream ifs(path); + if (!ifs) { + return false; + } + + dictionary->clear(); + + std::string line; + uint16_t currentIndex = 0; + uint8_t currentSub = 0; + ObjectEntry currentEntry{}; + PdoDefinition currentPdo{}; + bool inSection = false; + + auto flushEntry = [&]() { + if (!inSection || currentIndex == 0) { + return; + } + currentEntry.index = currentIndex; + currentEntry.subIndex = currentSub; + dictionary->registerEntry(currentEntry); + + if (isTpdoMappingIndex(currentIndex) && currentSub >= 1) { + PdoMappingEntry map; + map.index = static_cast((currentEntry.defaultValue >> 16) & 0xFFFF); + map.subIndex = static_cast(currentEntry.defaultValue & 0xFF); + map.bitLength = 16; + currentPdo.mappings.push_back(map); + } else if (isRpdoMappingIndex(currentIndex) && currentSub >= 1) { + PdoMappingEntry map; + map.index = static_cast((currentEntry.defaultValue >> 16) & 0xFFFF); + map.subIndex = static_cast(currentEntry.defaultValue & 0xFF); + map.bitLength = 16; + currentPdo.mappings.push_back(map); + } + }; + + auto flushPdo = [&](uint16_t index) { + if (isTpdoCommIndex(index)) { + currentPdo.commParamIndex = index; + dictionary->registerTpdo(currentPdo); + } else if (isRpdoCommIndex(index)) { + currentPdo.commParamIndex = index; + dictionary->registerRpdo(currentPdo); + } + currentPdo = PdoDefinition{}; + }; + + while (std::getline(ifs, line)) { + if (line.empty() || line[0] == ';') { + continue; + } + if (line[0] == '[') { + flushEntry(); + if (inSection && (isTpdoCommIndex(currentIndex) || isRpdoCommIndex(currentIndex))) { + flushPdo(currentIndex); + } else if (inSection && + (isTpdoMappingIndex(currentIndex) || isRpdoMappingIndex(currentIndex))) { + if (isTpdoMappingIndex(currentIndex)) { + currentPdo.commParamIndex = static_cast(currentIndex - 0x200); + dictionary->registerTpdo(currentPdo); + } else { + currentPdo.commParamIndex = static_cast(currentIndex - 0x200); + dictionary->registerRpdo(currentPdo); + } + currentPdo = PdoDefinition{}; + } + + const auto end = line.find(']'); + const std::string section = line.substr(1, end - 1); + const auto dot = section.find('.'); + currentIndex = parseHexIndex(dot == std::string::npos ? section : section.substr(0, dot)); + currentSub = dot == std::string::npos + ? 0 + : static_cast(std::stoul(section.substr(dot + 1), nullptr, 10)); + currentEntry = ObjectEntry{}; + inSection = true; + + if (isTpdoCommIndex(currentIndex) || isRpdoCommIndex(currentIndex) || + isTpdoMappingIndex(currentIndex) || isRpdoMappingIndex(currentIndex)) { + currentPdo = PdoDefinition{}; + currentPdo.commParamIndex = isTpdoMappingIndex(currentIndex) + ? static_cast(currentIndex - 0x200) + : isRpdoMappingIndex(currentIndex) + ? static_cast(currentIndex - 0x200) + : currentIndex; + } + continue; + } + + const auto eq = line.find('='); + if (eq == std::string::npos) { + continue; + } + const std::string key = line.substr(0, eq); + const std::string value = line.substr(eq + 1); + + if (key == "ParameterName") { + currentEntry.name = value; + } else if (key == "DataType") { + currentEntry.dataType = value; + } else if (key == "DefaultValue") { + currentEntry.defaultValue = parseHexValue(value); + if (currentSub == 1 && isTpdoCommIndex(currentIndex)) { + currentPdo.cobId = currentEntry.defaultValue & 0x7FF; + } else if (currentSub == 1 && isRpdoCommIndex(currentIndex)) { + currentPdo.cobId = currentEntry.defaultValue & 0x7FF; + } + } else if (key == "TransmissionType" && currentSub == 2) { + currentPdo.transmissionType = static_cast(parseHexValue(value)); + } + } + + flushEntry(); + if (inSection && (isTpdoCommIndex(currentIndex) || isRpdoCommIndex(currentIndex))) { + flushPdo(currentIndex); + } else if (inSection && + (isTpdoMappingIndex(currentIndex) || isRpdoMappingIndex(currentIndex))) { + if (isTpdoMappingIndex(currentIndex)) { + currentPdo.commParamIndex = static_cast(currentIndex - 0x200); + dictionary->registerTpdo(currentPdo); + } else { + currentPdo.commParamIndex = static_cast(currentIndex - 0x200); + dictionary->registerRpdo(currentPdo); + } + } + + return !dictionary->tpdoDefinitions().empty() || !dictionary->rpdoDefinitions().empty() || + dictionary->hasEntry(0x1000, 0); +} + +} // namespace canopen diff --git a/src/softbus_edge/CMakeLists.txt b/src/softbus_edge/CMakeLists.txt index 633481b..4bcd72a 100644 --- a/src/softbus_edge/CMakeLists.txt +++ b/src/softbus_edge/CMakeLists.txt @@ -3,18 +3,32 @@ 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/PhysicalRegistry.cpp src/UplinkClient.cpp src/DeviceFactory.cpp src/DeviceDiscovery.cpp src/ShadowChannelStore.cpp src/CommandExecutor.cpp + src/ingress/CompositeIngress.cpp + src/ingress/modbus/ModbusIngress.cpp + src/ingress/modbus/MockModbusIngress.cpp + src/ingress/can/CanIngress.cpp + src/ingress/eth/EthIngress.cpp + src/driver/modbus/ModbusChannel.cpp + src/driver/modbus/TcpRtuSession.cpp + src/driver/can/CanChannel.cpp + src/driver/eth/EthChannel.cpp + src/egress/modbus/ModbusEgress.cpp + src/egress/can/CanEgress.cpp + src/egress/eth/EthEgress.cpp + src/scheduler/ChangeFilter.cpp + src/scheduler/PollPlanner.cpp + src/scheduler/ModbusPollScheduler.cpp + src/canopen/NmtManager.cpp + src/canopen/SdoClient.cpp + src/canopen/PdoMapper.cpp ) softbus_export_include(softbus_edge "${CMAKE_CURRENT_SOURCE_DIR}/include") @@ -24,8 +38,9 @@ target_link_libraries(softbus_edge PRIVATE softbus_registry softbus_transport softbus_api + softbus_canopen_common asio ) -# 添加 libmodbus 库 softbus_apply_libmodbus(softbus_edge) +softbus_apply_socketcan(softbus_edge) softbus_apply_platform_libs(softbus_edge) diff --git a/src/softbus_edge/include/softbus_edge/canopen/NmtManager.h b/src/softbus_edge/include/softbus_edge/canopen/NmtManager.h new file mode 100644 index 0000000..d3062c2 --- /dev/null +++ b/src/softbus_edge/include/softbus_edge/canopen/NmtManager.h @@ -0,0 +1,23 @@ +#pragma once + +#include + +#include +#include +#include + +namespace softbus::edge { + +class NmtManager { +public: + bool startRemoteNode(CanChannelManager* channels, + const std::string& channelId, + uint8_t nodeId); + void setState(uint8_t nodeId, const std::string& state); + std::string state(uint8_t nodeId) const; + +private: + std::unordered_map states_; +}; + +} // namespace softbus::edge diff --git a/src/softbus_edge/include/softbus_edge/canopen/PdoMapper.h b/src/softbus_edge/include/softbus_edge/canopen/PdoMapper.h new file mode 100644 index 0000000..73947ee --- /dev/null +++ b/src/softbus_edge/include/softbus_edge/canopen/PdoMapper.h @@ -0,0 +1,41 @@ +#pragma once + +#include + +#include + +#include +#include +#include +#include + +namespace softbus::edge { + +struct MappedCanPoint { + std::string physicalPointId; + uint16_t objectIndex{0}; + uint8_t subIndex{0}; + int byteOffset{0}; + double scale{1.0}; +}; + +struct PdoSubscription { + std::string channelId; + uint8_t nodeId{0}; + uint32_t cobId{0}; + std::vector points; +}; + +class PdoMapper { +public: + void rebuild(const PhysicalRegistry& registry, + const std::unordered_map& dictionaries); + const std::vector& subscriptions() const { return subscriptions_; } + std::vector matchFrame(uint32_t cobId, const std::vector& data) const; + +private: + std::vector subscriptions_; + std::unordered_map cobIndex_; +}; + +} // namespace softbus::edge diff --git a/src/softbus_edge/include/softbus_edge/canopen/SdoClient.h b/src/softbus_edge/include/softbus_edge/canopen/SdoClient.h new file mode 100644 index 0000000..89b46d2 --- /dev/null +++ b/src/softbus_edge/include/softbus_edge/canopen/SdoClient.h @@ -0,0 +1,29 @@ +#pragma once + +#include +#include + +#include +#include +#include + +namespace softbus::edge { + +class SdoClient { +public: + bool upload(CanChannelManager* channels, + const ResolvedPhysicalPoint& point, + std::vector& outData); + bool download(CanChannelManager* channels, + const ResolvedPhysicalPoint& point, + const std::vector& data); + +private: + static uint32_t sdoRxCobId(uint8_t nodeId); + static uint32_t sdoTxCobId(uint8_t nodeId); + std::optional> waitResponse(CanChannelManager* channels, + const std::string& channelId, + uint32_t expectedCobId); +}; + +} // namespace 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 1cc5d7d..be032fb 100644 --- a/src/softbus_edge/include/softbus_edge/config/EdgeConfig.h +++ b/src/softbus_edge/include/softbus_edge/config/EdgeConfig.h @@ -16,6 +16,7 @@ struct EdgeConfig { std::string physicalDevicesPath{"physical_devices.json"}; std::string bindingsLocalPath{"bindings_local.json"}; std::string bindingsPatchPath; + std::string configBaseDir; 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 index 58b5470..509d69b 100644 --- a/src/softbus_edge/include/softbus_edge/config/PhysicalRegistry.h +++ b/src/softbus_edge/include/softbus_edge/config/PhysicalRegistry.h @@ -5,6 +5,7 @@ #include #include #include +#include #include namespace softbus::edge { @@ -26,6 +27,10 @@ public: const std::vector& devices() const { return devices_; } const std::vector& pollPoints() const { return pollPoints_; } + std::vector channelsByProtocol(const std::string& protocol) const; + std::vector pollPointsByProtocol(const std::string& protocol) const; + std::unordered_set protocols() const; + std::optional resolvePoint(const std::string& physicalPointId) const; std::optional findChannel(const std::string& channelId) const; diff --git a/src/softbus_edge/include/softbus_edge/config/PhysicalTypes.h b/src/softbus_edge/include/softbus_edge/config/PhysicalTypes.h index d4b8f42..702aa7a 100644 --- a/src/softbus_edge/include/softbus_edge/config/PhysicalTypes.h +++ b/src/softbus_edge/include/softbus_edge/config/PhysicalTypes.h @@ -18,6 +18,8 @@ struct PhysicalChannelDef { int dataBits{8}; int stopBits{1}; int pollIntervalMs{500}; + std::string canInterface{"can0"}; + int bitrate{500000}; }; struct PhysicalPointDef { @@ -28,6 +30,13 @@ struct PhysicalPointDef { double scale{1.0}; std::string signalType{"AI"}; bool writable{false}; + int pollIntervalMs{0}; + bool reportOnChange{false}; + double deadband{0.0}; + uint16_t objectIndex{0}; + uint8_t subIndex{0}; + std::string pdoType{"tpdo"}; + uint32_t cobId{0}; }; struct PhysicalDeviceDef { @@ -35,6 +44,8 @@ struct PhysicalDeviceDef { std::string channelId; std::string protocol{"modbus"}; int slaveId{1}; + int nodeId{0}; + std::string edsPath; std::vector points; }; diff --git a/src/softbus_edge/include/softbus_edge/driver/can/CanChannel.h b/src/softbus_edge/include/softbus_edge/driver/can/CanChannel.h new file mode 100644 index 0000000..c59e0cf --- /dev/null +++ b/src/softbus_edge/include/softbus_edge/driver/can/CanChannel.h @@ -0,0 +1,37 @@ +#pragma once + +#include + +#include +#include +#include +#include +#include + +namespace softbus::edge { + +struct CanFrame { + uint32_t canId{0}; + std::vector data; +}; + +class CanChannelManager { +public: + bool openChannel(const PhysicalChannelDef& channel); + void closeChannel(const std::string& channelId); + void closeAll(); + bool sendFrame(const std::string& channelId, const CanFrame& frame); + std::vector recvFrames(const std::string& channelId, int maxFrames = 16); + +private: + struct ChannelHandle { + PhysicalChannelDef config; +#if defined(SOFTBUS_HAS_SOCKETCAN) && SOFTBUS_HAS_SOCKETCAN + int socketFd{-1}; +#endif + }; + + std::unordered_map channels_; +}; + +} // namespace softbus::edge diff --git a/src/softbus_edge/include/softbus_edge/driver/eth/EthChannel.h b/src/softbus_edge/include/softbus_edge/driver/eth/EthChannel.h new file mode 100644 index 0000000..8c068d2 --- /dev/null +++ b/src/softbus_edge/include/softbus_edge/driver/eth/EthChannel.h @@ -0,0 +1,36 @@ +#pragma once + +#include + +#include + +#include +#include +#include +#include +#include + +namespace softbus::edge { + +class EthChannelManager { +public: + EthChannelManager(); + bool openChannel(const PhysicalChannelDef& channel); + void closeChannel(const std::string& channelId); + void closeAll(); + std::vector> recv(const std::string& channelId, int maxPackets = 8); + bool send(const std::string& channelId, const std::vector& data); + +private: + struct ChannelHandle { + PhysicalChannelDef config; + std::unique_ptr tcpSocket; + std::unique_ptr udpSocket; + asio::ip::udp::endpoint udpRemote; + }; + + asio::io_context io_; + std::unordered_map channels_; +}; + +} // namespace softbus::edge diff --git a/src/softbus_edge/include/softbus_edge/driver/ModbusChannel.h b/src/softbus_edge/include/softbus_edge/driver/modbus/ModbusChannel.h similarity index 68% rename from src/softbus_edge/include/softbus_edge/driver/ModbusChannel.h rename to src/softbus_edge/include/softbus_edge/driver/modbus/ModbusChannel.h index 95267ad..4fb3de0 100644 --- a/src/softbus_edge/include/softbus_edge/driver/ModbusChannel.h +++ b/src/softbus_edge/include/softbus_edge/driver/modbus/ModbusChannel.h @@ -1,11 +1,14 @@ #pragma once #include +#include #include +#include #include #include #include +#include namespace softbus::edge { @@ -15,11 +18,15 @@ public: void closeChannel(const std::string& channelId); void closeAll(); std::optional readRegister(const ResolvedPhysicalPoint& point); + std::optional> readRegisters(const ResolvedPhysicalPoint& point, + int startAddr, + int count); bool writeRegister(const ResolvedPhysicalPoint& point, uint16_t rawValue); private: struct ChannelHandle { PhysicalChannelDef config; + std::unique_ptr tcpRtu; #if defined(SOFTBUS_HAS_LIBMODBUS) && SOFTBUS_HAS_LIBMODBUS void* ctx{nullptr}; #endif diff --git a/src/softbus_edge/include/softbus_edge/driver/modbus/TcpRtuSession.h b/src/softbus_edge/include/softbus_edge/driver/modbus/TcpRtuSession.h new file mode 100644 index 0000000..18763f0 --- /dev/null +++ b/src/softbus_edge/include/softbus_edge/driver/modbus/TcpRtuSession.h @@ -0,0 +1,28 @@ +#pragma once + +#include +#include +#include +#include + +namespace softbus::edge { + +class TcpRtuSession { +public: + bool connect(const std::string& host, uint16_t port, int timeoutMs = 2000); + void disconnect(); + bool isConnected() const { return socketFd_ >= 0; } + + std::optional> readHoldingRegisters(int slaveId, + int startAddr, + int count); + bool writeSingleRegister(int slaveId, int address, uint16_t value); + +private: + static uint16_t crc16(const uint8_t* data, std::size_t len); + std::optional> transaction(const std::vector& pdu); + + int socketFd_{-1}; +}; + +} // 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 7f654de..b5378dd 100644 --- a/src/softbus_edge/include/softbus_edge/egress/CommandExecutor.h +++ b/src/softbus_edge/include/softbus_edge/egress/CommandExecutor.h @@ -2,7 +2,9 @@ #include #include -#include +#include +#include +#include #include #include #include @@ -13,6 +15,8 @@ class CommandExecutor { public: CommandExecutor(PhysicalRegistry* physicalRegistry, ModbusEgress* modbusEgress, + CanEgress* canEgress, + EthEgress* ethEgress, ShadowChannelStore* mockStore, softbus::registry::BindingRegistry* localBindings, UplinkClient* uplink, @@ -28,6 +32,8 @@ private: PhysicalRegistry* physicalRegistry_{nullptr}; ModbusEgress* modbusEgress_{nullptr}; + CanEgress* canEgress_{nullptr}; + EthEgress* ethEgress_{nullptr}; ShadowChannelStore* mockStore_{nullptr}; softbus::registry::BindingRegistry* localBindings_{nullptr}; UplinkClient* uplink_{nullptr}; diff --git a/src/softbus_edge/include/softbus_edge/egress/can/CanEgress.h b/src/softbus_edge/include/softbus_edge/egress/can/CanEgress.h new file mode 100644 index 0000000..6d27ab2 --- /dev/null +++ b/src/softbus_edge/include/softbus_edge/egress/can/CanEgress.h @@ -0,0 +1,24 @@ +#pragma once + +#include +#include +#include + +#include + +namespace softbus::edge { + +class CanEgress { +public: + explicit CanEgress(PhysicalRegistry* registry, CanChannelManager* channels); + + bool writeByPhysicalPointId(const std::string& physicalPointId, double engineeringValue, + int* rawOut = nullptr); + +private: + PhysicalRegistry* registry_{nullptr}; + CanChannelManager* channels_{nullptr}; + SdoClient sdo_; +}; + +} // namespace softbus::edge diff --git a/src/softbus_edge/include/softbus_edge/egress/eth/EthEgress.h b/src/softbus_edge/include/softbus_edge/egress/eth/EthEgress.h new file mode 100644 index 0000000..ead0afc --- /dev/null +++ b/src/softbus_edge/include/softbus_edge/egress/eth/EthEgress.h @@ -0,0 +1,23 @@ +#pragma once + +#include +#include + +#include +#include + +namespace softbus::edge { + +class EthEgress { +public: + explicit EthEgress(PhysicalRegistry* registry, EthChannelManager* channels); + + bool writeByPhysicalPointId(const std::string& physicalPointId, + const std::vector& data); + +private: + PhysicalRegistry* registry_{nullptr}; + EthChannelManager* channels_{nullptr}; +}; + +} // namespace softbus::edge diff --git a/src/softbus_edge/include/softbus_edge/egress/ModbusEgress.h b/src/softbus_edge/include/softbus_edge/egress/modbus/ModbusEgress.h similarity index 88% rename from src/softbus_edge/include/softbus_edge/egress/ModbusEgress.h rename to src/softbus_edge/include/softbus_edge/egress/modbus/ModbusEgress.h index b0f50f0..1c95afb 100644 --- a/src/softbus_edge/include/softbus_edge/egress/ModbusEgress.h +++ b/src/softbus_edge/include/softbus_edge/egress/modbus/ModbusEgress.h @@ -1,9 +1,8 @@ #pragma once #include -#include +#include -#include #include namespace softbus::edge { diff --git a/src/softbus_edge/include/softbus_edge/factory/DeviceFactory.h b/src/softbus_edge/include/softbus_edge/factory/DeviceFactory.h index aa25ddc..c89fc4c 100644 --- a/src/softbus_edge/include/softbus_edge/factory/DeviceFactory.h +++ b/src/softbus_edge/include/softbus_edge/factory/DeviceFactory.h @@ -2,11 +2,10 @@ #include #include -#include -#include +#include +#include +#include #include -#include -#include #include #include @@ -19,7 +18,9 @@ public: std::unique_ptr createIngress(const EdgeConfig& config, PhysicalRegistry* registry, - ModbusChannelManager* channels, + ModbusChannelManager* modbusChannels, + CanChannelManager* canChannels, + EthChannelManager* ethChannels, bool useMock, const std::string& mockObjectRef); }; diff --git a/src/softbus_edge/include/softbus_edge/ingress/CompositeIngress.h b/src/softbus_edge/include/softbus_edge/ingress/CompositeIngress.h new file mode 100644 index 0000000..4ade2ed --- /dev/null +++ b/src/softbus_edge/include/softbus_edge/ingress/CompositeIngress.h @@ -0,0 +1,22 @@ +#pragma once + +#include + +#include +#include + +namespace softbus::edge { + +class CompositeIngress : public IIngressPort { +public: + explicit CompositeIngress(std::vector> children); + + bool open() override; + void close() override; + std::vector poll() override; + +private: + std::vector> children_; +}; + +} // namespace softbus::edge diff --git a/src/softbus_edge/include/softbus_edge/ingress/can/CanIngress.h b/src/softbus_edge/include/softbus_edge/ingress/can/CanIngress.h new file mode 100644 index 0000000..429f7fd --- /dev/null +++ b/src/softbus_edge/include/softbus_edge/ingress/can/CanIngress.h @@ -0,0 +1,48 @@ +#pragma once + +#include + +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include + +namespace softbus::edge { + +class CanIngress : public IIngressPort { +public: + CanIngress(PhysicalRegistry* registry, + CanChannelManager* channels, + std::string edgeIdPrefix, + std::string configBaseDir); + + bool open() override; + void close() override; + std::vector poll() override; + +private: + double decodeValue(const MappedCanPoint& mapped, const std::vector& data) const; + std::vector buildCanopenPayload(uint32_t cobId, const std::vector& data) const; + bool loadDictionaries(); + + PhysicalRegistry* registry_{nullptr}; + CanChannelManager* channels_{nullptr}; + std::string edgeIdPrefix_; + std::string configBaseDir_; + bool open_{false}; + NmtManager nmt_; + SdoClient sdo_; + PdoMapper pdoMapper_; + PollPlanner planner_; + ChangeFilter changeFilter_; + std::unordered_map dictionaries_; +}; + +} // namespace softbus::edge diff --git a/src/softbus_edge/include/softbus_edge/ingress/eth/EthIngress.h b/src/softbus_edge/include/softbus_edge/ingress/eth/EthIngress.h new file mode 100644 index 0000000..5365bfc --- /dev/null +++ b/src/softbus_edge/include/softbus_edge/ingress/eth/EthIngress.h @@ -0,0 +1,26 @@ +#pragma once + +#include +#include +#include + +#include + +namespace softbus::edge { + +class EthIngress : public IIngressPort { +public: + EthIngress(PhysicalRegistry* registry, EthChannelManager* channels, std::string edgeIdPrefix); + + bool open() override; + void close() override; + std::vector poll() override; + +private: + PhysicalRegistry* registry_{nullptr}; + EthChannelManager* channels_{nullptr}; + std::string edgeIdPrefix_; + bool open_{false}; +}; + +} // namespace softbus::edge diff --git a/src/softbus_edge/include/softbus_edge/ingress/MockModbusIngress.h b/src/softbus_edge/include/softbus_edge/ingress/modbus/MockModbusIngress.h similarity index 100% rename from src/softbus_edge/include/softbus_edge/ingress/MockModbusIngress.h rename to src/softbus_edge/include/softbus_edge/ingress/modbus/MockModbusIngress.h diff --git a/src/softbus_edge/include/softbus_edge/ingress/ModbusIngress.h b/src/softbus_edge/include/softbus_edge/ingress/modbus/ModbusIngress.h similarity index 69% rename from src/softbus_edge/include/softbus_edge/ingress/ModbusIngress.h rename to src/softbus_edge/include/softbus_edge/ingress/modbus/ModbusIngress.h index 93763d8..410015c 100644 --- a/src/softbus_edge/include/softbus_edge/ingress/ModbusIngress.h +++ b/src/softbus_edge/include/softbus_edge/ingress/modbus/ModbusIngress.h @@ -1,10 +1,12 @@ #pragma once #include -#include +#include #include +#include +#include +#include -#include #include namespace softbus::edge { @@ -24,7 +26,9 @@ private: ModbusChannelManager* channels_{nullptr}; std::string edgeIdPrefix_; bool open_{false}; - std::chrono::steady_clock::time_point lastPoll_{}; + PollPlanner planner_; + ChangeFilter changeFilter_; + ModbusPollScheduler scheduler_; }; } // 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 index 1b5b2e4..683c2f7 100644 --- a/src/softbus_edge/include/softbus_edge/runtime/EdgeRuntime.h +++ b/src/softbus_edge/include/softbus_edge/runtime/EdgeRuntime.h @@ -2,9 +2,13 @@ #include #include -#include +#include +#include +#include +#include +#include +#include #include -#include #include #include #include @@ -31,6 +35,8 @@ public: private: void tickUplink(); void pollIngress(); + void rebuildIngress(); + EdgeConfig config_; bool useMock_{false}; std::string mockObjectRef_; @@ -38,7 +44,11 @@ private: softbus::registry::BindingRegistry localBindings_; PhysicalRegistry physicalRegistry_; ModbusChannelManager modbusChannels_; + CanChannelManager canChannels_; + EthChannelManager ethChannels_; ModbusEgress modbusEgress_{&physicalRegistry_, &modbusChannels_}; + CanEgress canEgress_{&physicalRegistry_, &canChannels_}; + EthEgress ethEgress_{&physicalRegistry_, ðChannels_}; ShadowChannelStore mockStore_; std::unique_ptr uplink_; std::unique_ptr ingress_; @@ -47,5 +57,4 @@ private: std::chrono::steady_clock::time_point lastUplinkReconnectAttempt_{}; }; - } // namespace softbus::edge diff --git a/src/softbus_edge/include/softbus_edge/scheduler/ChangeFilter.h b/src/softbus_edge/include/softbus_edge/scheduler/ChangeFilter.h new file mode 100644 index 0000000..dc58f09 --- /dev/null +++ b/src/softbus_edge/include/softbus_edge/scheduler/ChangeFilter.h @@ -0,0 +1,21 @@ +#pragma once + +#include + +#include +#include + +namespace softbus::edge { + +class ChangeFilter { +public: + bool shouldReport(const PhysicalPointDef& point, double engineeringValue) const; + void update(const std::string& physicalPointId, double engineeringValue); + void clear(); + +private: + std::unordered_map lastValues_; + std::unordered_map hasSample_; +}; + +} // namespace softbus::edge diff --git a/src/softbus_edge/include/softbus_edge/scheduler/ModbusPollScheduler.h b/src/softbus_edge/include/softbus_edge/scheduler/ModbusPollScheduler.h new file mode 100644 index 0000000..3acdd48 --- /dev/null +++ b/src/softbus_edge/include/softbus_edge/scheduler/ModbusPollScheduler.h @@ -0,0 +1,31 @@ +#pragma once + +#include + +#include +#include + +namespace softbus::edge { + +struct PollBatchPoint { + std::string physicalPointId; + int offsetInBatch{0}; +}; + +struct PollBatch { + std::string channelId; + int slaveId{0}; + int functionCode{3}; + int startAddr{0}; + int count{0}; + std::vector points; +}; + +class ModbusPollScheduler { +public: + static int modbusAddress(int registerAddress); + std::vector planBatches( + const std::vector& duePoints) const; +}; + +} // namespace softbus::edge diff --git a/src/softbus_edge/include/softbus_edge/scheduler/PollPlanner.h b/src/softbus_edge/include/softbus_edge/scheduler/PollPlanner.h new file mode 100644 index 0000000..d8664b1 --- /dev/null +++ b/src/softbus_edge/include/softbus_edge/scheduler/PollPlanner.h @@ -0,0 +1,27 @@ +#pragma once + +#include + +#include +#include +#include +#include + +namespace softbus::edge { + +class PollPlanner { +public: + void reset(const PhysicalRegistry& registry, const std::string& protocol); + std::vector duePointIds(); + void markPolled(const std::string& physicalPointId); + +private: + struct PointSchedule { + int intervalMs{500}; + std::chrono::steady_clock::time_point nextDue{}; + }; + + std::unordered_map schedules_; +}; + +} // namespace softbus::edge diff --git a/src/softbus_edge/src/CommandExecutor.cpp b/src/softbus_edge/src/CommandExecutor.cpp index 9535323..38c34fa 100644 --- a/src/softbus_edge/src/CommandExecutor.cpp +++ b/src/softbus_edge/src/CommandExecutor.cpp @@ -3,10 +3,14 @@ #include #include +#include + namespace softbus::edge { CommandExecutor::CommandExecutor(PhysicalRegistry* physicalRegistry, ModbusEgress* modbusEgress, + CanEgress* canEgress, + EthEgress* ethEgress, ShadowChannelStore* mockStore, softbus::registry::BindingRegistry* localBindings, UplinkClient* uplink, @@ -14,6 +18,8 @@ CommandExecutor::CommandExecutor(PhysicalRegistry* physicalRegistry, bool useMock) : physicalRegistry_(physicalRegistry) , modbusEgress_(modbusEgress) + , canEgress_(canEgress) + , ethEgress_(ethEgress) , mockStore_(mockStore) , localBindings_(localBindings) , uplink_(uplink) @@ -80,9 +86,9 @@ softbus::core::SoftbusAck CommandExecutor::execute(const softbus::api::ExecuteCo return ack; } - if (!modbusEgress_ || !physicalRegistry_) { + if (!physicalRegistry_) { ack.success = false; - ack.error = "modbus egress not initialized"; + ack.error = "physical registry not initialized"; return ack; } @@ -100,9 +106,24 @@ softbus::core::SoftbusAck CommandExecutor::execute(const softbus::api::ExecuteCo } int raw = 0; - if (!modbusEgress_->writeByPhysicalPointId(physicalPointId, value, &raw)) { + bool ok = false; + if (resolved->device.protocol == "canopen") { + ok = canEgress_ && canEgress_->writeByPhysicalPointId(physicalPointId, value, &raw); + } else if (resolved->device.protocol == "raw") { + const int rawValue = static_cast(std::lround(value)); + std::vector payload{ + static_cast(rawValue & 0xFF), + static_cast((rawValue >> 8) & 0xFF) + }; + ok = ethEgress_ && ethEgress_->writeByPhysicalPointId(physicalPointId, payload); + raw = rawValue; + } else { + ok = modbusEgress_ && modbusEgress_->writeByPhysicalPointId(physicalPointId, value, &raw); + } + + if (!ok) { ack.success = false; - ack.error = "modbus write failed"; + ack.error = "egress write failed"; return ack; } diff --git a/src/softbus_edge/src/DeviceFactory.cpp b/src/softbus_edge/src/DeviceFactory.cpp index 39cb5c2..bb15c82 100644 --- a/src/softbus_edge/src/DeviceFactory.cpp +++ b/src/softbus_edge/src/DeviceFactory.cpp @@ -1,5 +1,11 @@ #include +#include +#include +#include +#include +#include + #include namespace softbus::edge { @@ -10,15 +16,42 @@ std::string DeviceFactory::createRuntimeDeviceId() } std::unique_ptr DeviceFactory::createIngress(const EdgeConfig& config, - PhysicalRegistry* registry, - ModbusChannelManager* channels, - bool useMock, - const std::string& mockObjectRef) + PhysicalRegistry* registry, + ModbusChannelManager* modbusChannels, + CanChannelManager* canChannels, + EthChannelManager* ethChannels, + bool useMock, + const std::string& mockObjectRef) { if (useMock) { return std::make_unique(mockObjectRef, config.edgeId); } - return std::make_unique(registry, channels, config.edgeId); + + std::vector> children; + const auto protocols = registry->protocols(); + + if (protocols.count("modbus") != 0) { + children.push_back( + std::make_unique(registry, modbusChannels, config.edgeId)); + } + if (protocols.count("canopen") != 0) { + children.push_back(std::make_unique( + registry, canChannels, config.edgeId, config.configBaseDir)); + } + if (protocols.count("raw") != 0) { + children.push_back( + std::make_unique(registry, ethChannels, config.edgeId)); + } + + if (children.empty()) { + children.push_back( + std::make_unique(registry, modbusChannels, config.edgeId)); + } + + if (children.size() == 1) { + return std::move(children.front()); + } + return std::make_unique(std::move(children)); } } // namespace softbus::edge diff --git a/src/softbus_edge/src/EdgeConfig.cpp b/src/softbus_edge/src/EdgeConfig.cpp index e0c64fd..9d81358 100644 --- a/src/softbus_edge/src/EdgeConfig.cpp +++ b/src/softbus_edge/src/EdgeConfig.cpp @@ -47,6 +47,7 @@ bool EdgeConfig::loadFromFile(const std::string& path) bindingsPatchPath = edge.value("bindingsPatch", bindingsPatchPath); const fs::path baseDir = fs::path(path).parent_path(); + configBaseDir = fs::absolute(baseDir).string(); resolveConfigPath(baseDir, physicalChannelsPath); resolveConfigPath(baseDir, physicalDevicesPath); resolveConfigPath(baseDir, bindingsLocalPath); diff --git a/src/softbus_edge/src/EdgeRuntime.cpp b/src/softbus_edge/src/EdgeRuntime.cpp index 45edf33..a1dcd4b 100644 --- a/src/softbus_edge/src/EdgeRuntime.cpp +++ b/src/softbus_edge/src/EdgeRuntime.cpp @@ -4,6 +4,23 @@ namespace softbus::edge { +void EdgeRuntime::rebuildIngress() +{ + if (ingress_) { + ingress_->close(); + } + modbusChannels_.closeAll(); + canChannels_.closeAll(); + ethChannels_.closeAll(); + + ingress_ = factory_.createIngress( + config_, &physicalRegistry_, &modbusChannels_, &canChannels_, ðChannels_, + useMock_, mockObjectRef_); + if (!ingress_->open()) { + SB_LOG_ERROR("EdgeRuntime", "Failed to open ingress"); + } +} + bool EdgeRuntime::init(const EdgeConfig& config, bool useMock, const std::string& mockObjectRef) { config_ = config; @@ -26,8 +43,9 @@ bool EdgeRuntime::init(const EdgeConfig& config, bool useMock, const std::string } uplink_ = std::make_unique(config.daemonHost, config.daemonPort); + // 创建命令执行器 用于执行下行命令 executor_ = std::make_unique( - &physicalRegistry_, &modbusEgress_, &mockStore_, &localBindings_, + &physicalRegistry_, &modbusEgress_, &canEgress_, ðEgress_, &mockStore_, &localBindings_, uplink_.get(), config.deviceId, useMock); uplink_->setCommandHandler([this](const softbus::api::ExecuteCommand& cmd) { @@ -36,11 +54,14 @@ bool EdgeRuntime::init(const EdgeConfig& config, bool useMock, const std::string } }); - ingress_ = factory_.createIngress(config, &physicalRegistry_, &modbusChannels_, useMock, mockObjectRef); - if (!ingress_->open()) { - SB_LOG_ERROR("EdgeRuntime", "Failed to open ingress"); + rebuildIngress(); + if (!ingress_) { + SB_LOG_ERROR("EdgeRuntime", "Failed to create ingress"); return false; } + if (!useMock && physicalRegistry_.channels().empty()) { + SB_LOG_WARN("EdgeRuntime", "No physical channels configured"); + } if (!uplink_->connect()) { SB_LOG_WARN( @@ -62,11 +83,9 @@ void EdgeRuntime::reloadPhysical() if (useMock_) { return; } - ingress_->close(); - modbusChannels_.closeAll(); physicalRegistry_.reloadChannels(config_.physicalChannelsPath); physicalRegistry_.reloadDevices(config_.physicalDevicesPath, config_.edgeId); - ingress_->open(); + rebuildIngress(); SB_LOG_INFO("EdgeRuntime", "Physical config reloaded"); } @@ -85,6 +104,8 @@ void EdgeRuntime::shutdown() ingress_->close(); } modbusChannels_.closeAll(); + canChannels_.closeAll(); + ethChannels_.closeAll(); if (uplink_) { uplink_->disconnect(); } diff --git a/src/softbus_edge/src/ModbusIngress.cpp b/src/softbus_edge/src/ModbusIngress.cpp deleted file mode 100644 index b9b98ee..0000000 --- a/src/softbus_edge/src/ModbusIngress.cpp +++ /dev/null @@ -1,92 +0,0 @@ -#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 index 6275080..a824188 100644 --- a/src/softbus_edge/src/PhysicalRegistry.cpp +++ b/src/softbus_edge/src/PhysicalRegistry.cpp @@ -47,6 +47,8 @@ bool PhysicalRegistry::loadChannels(const std::string& path) def.dataBits = ch.value("dataBits", 8); def.stopBits = ch.value("stopBits", 1); def.pollIntervalMs = ch.value("pollIntervalMs", 500); + def.canInterface = ch.value("interface", ch.value("canInterface", "can0")); + def.bitrate = ch.value("bitrate", 500000); if (def.channelId.empty()) { continue; } @@ -76,6 +78,8 @@ bool PhysicalRegistry::loadDevices(const std::string& path, const std::string& e device.channelId = dev.value("channelId", ""); device.protocol = dev.value("protocol", "modbus"); device.slaveId = dev.value("slaveId", 1); + device.nodeId = dev.value("nodeId", 0); + device.edsPath = dev.value("edsPath", ""); if (!dev.contains("points")) { devices_.push_back(std::move(device)); continue; @@ -90,7 +94,20 @@ bool PhysicalRegistry::loadDevices(const std::string& path, const std::string& e 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) { + point.pollIntervalMs = pt.value("pollIntervalMs", 0); + point.reportOnChange = pt.value("reportOnChange", false); + point.deadband = pt.value("deadband", 0.0); + point.objectIndex = static_cast(pt.value("objectIndex", 0)); + point.subIndex = static_cast(pt.value("subIndex", 0)); + point.pdoType = pt.value("pdoType", "tpdo"); + point.cobId = static_cast(pt.value("cobId", 0)); + if (point.physicalPointId.empty()) { + continue; + } + if (device.protocol == "modbus" && point.registerAddress == 0) { + continue; + } + if (device.protocol == "canopen" && point.objectIndex == 0) { continue; } device.points.push_back(point); @@ -130,6 +147,38 @@ void PhysicalRegistry::rebuildPollIndex(const std::string& edgeIdPrefix) } } +std::vector PhysicalRegistry::channelsByProtocol(const std::string& protocol) const +{ + std::vector out; + for (const auto& ch : channels_) { + if (ch.protocol == protocol) { + out.push_back(ch); + } + } + return out; +} + +std::vector PhysicalRegistry::pollPointsByProtocol(const std::string& protocol) const +{ + std::vector out; + for (const auto& point : pollPoints_) { + const auto resolved = resolvePoint(point.physicalPointId); + if (resolved && resolved->device.protocol == protocol) { + out.push_back(point); + } + } + return out; +} + +std::unordered_set PhysicalRegistry::protocols() const +{ + std::unordered_set out; + for (const auto& ch : channels_) { + out.insert(ch.protocol); + } + return out; +} + std::optional PhysicalRegistry::resolvePoint( const std::string& physicalPointId) const { diff --git a/src/softbus_edge/src/canopen/NmtManager.cpp b/src/softbus_edge/src/canopen/NmtManager.cpp new file mode 100644 index 0000000..1bdb902 --- /dev/null +++ b/src/softbus_edge/src/canopen/NmtManager.cpp @@ -0,0 +1,33 @@ +#include + +namespace softbus::edge { + +bool NmtManager::startRemoteNode(CanChannelManager* channels, + const std::string& channelId, + uint8_t nodeId) +{ + if (!channels) { + return false; + } + CanFrame frame; + frame.canId = 0x000; + frame.data = {0x01, nodeId}; + const bool ok = channels->sendFrame(channelId, frame); + if (ok) { + setState(nodeId, "Operational"); + } + return ok; +} + +void NmtManager::setState(uint8_t nodeId, const std::string& state) +{ + states_[nodeId] = state; +} + +std::string NmtManager::state(uint8_t nodeId) const +{ + const auto it = states_.find(nodeId); + return it == states_.end() ? "Unknown" : it->second; +} + +} // namespace softbus::edge diff --git a/src/softbus_edge/src/canopen/PdoMapper.cpp b/src/softbus_edge/src/canopen/PdoMapper.cpp new file mode 100644 index 0000000..4273668 --- /dev/null +++ b/src/softbus_edge/src/canopen/PdoMapper.cpp @@ -0,0 +1,83 @@ +#include + +namespace softbus::edge { + +void PdoMapper::rebuild( + const PhysicalRegistry& registry, + const std::unordered_map& dictionaries) +{ + subscriptions_.clear(); + cobIndex_.clear(); + + for (const auto& device : registry.devices()) { + if (device.protocol != "canopen" || device.nodeId == 0) { + continue; + } + const auto channel = registry.findChannel(device.channelId); + if (!channel) { + continue; + } + + const auto dictIt = dictionaries.find(device.physicalDeviceId); + const canopen::CanopenObjectDictionary* dictionary = + dictIt == dictionaries.end() ? nullptr : &dictIt->second; + + PdoSubscription sub; + sub.channelId = device.channelId; + sub.nodeId = static_cast(device.nodeId); + + if (dictionary) { + for (const auto& tpdo : dictionary->tpdoDefinitions()) { + if (tpdo.cobId != 0) { + sub.cobId = tpdo.cobId + static_cast(device.nodeId); + break; + } + } + } + if (sub.cobId == 0) { + sub.cobId = 0x180u + static_cast(device.nodeId); + } + + int byteOffset = 0; + for (const auto& point : device.points) { + if (point.pdoType != "tpdo" && point.pdoType != "sdo") { + continue; + } + MappedCanPoint mapped; + mapped.physicalPointId = point.physicalPointId; + mapped.objectIndex = point.objectIndex; + mapped.subIndex = point.subIndex; + mapped.byteOffset = byteOffset; + mapped.scale = point.scale; + if (point.cobId != 0) { + sub.cobId = point.cobId; + } + sub.points.push_back(mapped); + byteOffset += 2; + } + + if (!sub.points.empty()) { + cobIndex_[sub.cobId] = subscriptions_.size(); + subscriptions_.push_back(std::move(sub)); + } + } +} + +std::vector PdoMapper::matchFrame(uint32_t cobId, + const std::vector& data) const +{ + const auto it = cobIndex_.find(cobId); + if (it == cobIndex_.end() || it->second >= subscriptions_.size()) { + return {}; + } + const auto& sub = subscriptions_[it->second]; + std::vector matched; + for (const auto& point : sub.points) { + if (point.byteOffset + 1 < static_cast(data.size())) { + matched.push_back(point); + } + } + return matched; +} + +} // namespace softbus::edge diff --git a/src/softbus_edge/src/canopen/SdoClient.cpp b/src/softbus_edge/src/canopen/SdoClient.cpp new file mode 100644 index 0000000..31324f4 --- /dev/null +++ b/src/softbus_edge/src/canopen/SdoClient.cpp @@ -0,0 +1,112 @@ +#include + +#include +#include + +namespace softbus::edge { + +uint32_t SdoClient::sdoRxCobId(uint8_t nodeId) +{ + return 0x600u + nodeId; +} + +uint32_t SdoClient::sdoTxCobId(uint8_t nodeId) +{ + return 0x580u + nodeId; +} + +std::optional> SdoClient::waitResponse(CanChannelManager* channels, + const std::string& channelId, + uint32_t expectedCobId) +{ + for (int attempt = 0; attempt < 10; ++attempt) { + const auto frames = channels->recvFrames(channelId, 8); + for (const auto& frame : frames) { + if (frame.canId == expectedCobId && frame.data.size() >= 4) { + return frame.data; + } + } + std::this_thread::sleep_for(std::chrono::milliseconds(5)); + } + return std::nullopt; +} + +bool SdoClient::upload(CanChannelManager* channels, + const ResolvedPhysicalPoint& point, + std::vector& outData) +{ + if (!channels || point.device.nodeId == 0) { + return false; + } + + const uint8_t nodeId = static_cast(point.device.nodeId); + CanFrame request; + request.canId = sdoRxCobId(nodeId); + request.data = { + 0x40, + static_cast(point.point.objectIndex & 0xFF), + static_cast((point.point.objectIndex >> 8) & 0xFF), + point.point.subIndex, + 0x00, 0x00, 0x00, 0x00 + }; + if (!channels->sendFrame(point.channel.channelId, request)) { + return false; + } + + const auto response = waitResponse(channels, point.channel.channelId, sdoTxCobId(nodeId)); + if (!response || response->size() < 4) { + return false; + } + + outData.clear(); + if ((*response)[0] == 0x43) { + outData.insert(outData.end(), response->begin() + 4, response->end()); + } else if ((*response)[0] == 0x4B && response->size() >= 6) { + outData.push_back((*response)[4]); + outData.push_back((*response)[5]); + } else if ((*response)[0] == 0x4F && response->size() >= 5) { + outData.push_back((*response)[4]); + } + return !outData.empty(); +} + +bool SdoClient::download(CanChannelManager* channels, + const ResolvedPhysicalPoint& point, + const std::vector& data) +{ + if (!channels || point.device.nodeId == 0 || data.empty()) { + return false; + } + + const uint8_t nodeId = static_cast(point.device.nodeId); + CanFrame request; + request.canId = sdoRxCobId(nodeId); + request.data.resize(8, 0x00); + request.data[1] = static_cast(point.point.objectIndex & 0xFF); + request.data[2] = static_cast((point.point.objectIndex >> 8) & 0xFF); + request.data[3] = point.point.subIndex; + + if (data.size() == 1) { + request.data[0] = 0x2F; + request.data[4] = data[0]; + } else if (data.size() == 2) { + request.data[0] = 0x2B; + request.data[4] = data[0]; + request.data[5] = data[1]; + } else { + request.data[0] = 0x23; + request.data[4] = data[0]; + request.data[5] = data.size() > 1 ? data[1] : 0x00; + request.data[6] = data.size() > 2 ? data[2] : 0x00; + request.data[7] = data.size() > 3 ? data[3] : 0x00; + } + + if (!channels->sendFrame(point.channel.channelId, request)) { + return false; + } + + const auto response = waitResponse(channels, point.channel.channelId, sdoTxCobId(nodeId)); + return response && !response->empty() && ((*response)[0] & 0xE0) == 0x60; +} + +} // namespace softbus::edge diff --git a/src/softbus_edge/src/driver/can/CanChannel.cpp b/src/softbus_edge/src/driver/can/CanChannel.cpp new file mode 100644 index 0000000..a72f43e --- /dev/null +++ b/src/softbus_edge/src/driver/can/CanChannel.cpp @@ -0,0 +1,133 @@ +#include + +#include + +#if defined(SOFTBUS_HAS_SOCKETCAN) && SOFTBUS_HAS_SOCKETCAN +#include +#include +#include +#include +#include +#include +#endif + +#include + +namespace softbus::edge { + +bool CanChannelManager::openChannel(const PhysicalChannelDef& channel) +{ + closeChannel(channel.channelId); + ChannelHandle handle; + handle.config = channel; + +#if defined(SOFTBUS_HAS_SOCKETCAN) && SOFTBUS_HAS_SOCKETCAN + const int fd = socket(PF_CAN, SOCK_RAW, CAN_RAW); + if (fd < 0) { + SB_LOG_ERROR("CanChannel", "Failed to create CAN socket for ", channel.channelId); + return false; + } + + struct ifreq ifr {}; + std::strncpy(ifr.ifr_name, channel.canInterface.c_str(), IFNAMSIZ - 1); + if (ioctl(fd, SIOCGIFINDEX, &ifr) < 0) { + SB_LOG_ERROR("CanChannel", "Failed to resolve interface ", channel.canInterface); + close(fd); + return false; + } + + struct sockaddr_can addr {}; + addr.can_family = AF_CAN; + addr.can_ifindex = ifr.ifr_ifindex; + if (bind(fd, reinterpret_cast(&addr), sizeof(addr)) < 0) { + SB_LOG_ERROR("CanChannel", "Failed to bind CAN socket on ", channel.canInterface); + close(fd); + return false; + } + + handle.socketFd = fd; + channels_[channel.channelId] = handle; + SB_LOG_INFO("CanChannel", "Opened CAN channel ", channel.channelId, " on ", channel.canInterface); + return true; +#else + channels_[channel.channelId] = handle; + SB_LOG_WARN("CanChannel", "SocketCAN unavailable; channel ", channel.channelId, " simulated"); + return true; +#endif +} + +void CanChannelManager::closeChannel(const std::string& channelId) +{ + const auto it = channels_.find(channelId); + if (it == channels_.end()) { + return; + } +#if defined(SOFTBUS_HAS_SOCKETCAN) && SOFTBUS_HAS_SOCKETCAN + if (it->second.socketFd >= 0) { + close(it->second.socketFd); + } +#endif + channels_.erase(it); +} + +void CanChannelManager::closeAll() +{ + for (const auto& [id, _] : channels_) { + (void)_; + closeChannel(id); + } +} + +bool CanChannelManager::sendFrame(const std::string& channelId, const CanFrame& frame) +{ + const auto it = channels_.find(channelId); + if (it == channels_.end()) { + return false; + } + +#if defined(SOFTBUS_HAS_SOCKETCAN) && SOFTBUS_HAS_SOCKETCAN + if (it->second.socketFd < 0) { + return false; + } + struct can_frame raw {}; + raw.can_id = frame.canId; + raw.can_dlc = static_cast<__u8>(std::min(frame.data.size(), 8)); + for (std::size_t i = 0; i < raw.can_dlc; ++i) { + raw.data[i] = frame.data[i]; + } + const ssize_t sent = write(it->second.socketFd, &raw, sizeof(raw)); + return sent == static_cast(sizeof(raw)); +#else + (void)frame; + return true; +#endif +} + +std::vector CanChannelManager::recvFrames(const std::string& channelId, int maxFrames) +{ + std::vector frames; + const auto it = channels_.find(channelId); + if (it == channels_.end()) { + return frames; + } + +#if defined(SOFTBUS_HAS_SOCKETCAN) && SOFTBUS_HAS_SOCKETCAN + if (it->second.socketFd < 0) { + return frames; + } + for (int i = 0; i < maxFrames; ++i) { + struct can_frame raw {}; + const ssize_t n = read(it->second.socketFd, &raw, sizeof(raw)); + if (n != static_cast(sizeof(raw))) { + break; + } + CanFrame frame; + frame.canId = raw.can_id; + frame.data.assign(raw.data, raw.data + raw.can_dlc); + frames.push_back(std::move(frame)); + } +#endif + return frames; +} + +} // namespace softbus::edge diff --git a/src/softbus_edge/src/driver/eth/EthChannel.cpp b/src/softbus_edge/src/driver/eth/EthChannel.cpp new file mode 100644 index 0000000..6719e72 --- /dev/null +++ b/src/softbus_edge/src/driver/eth/EthChannel.cpp @@ -0,0 +1,101 @@ +#include + +#include + +#include + +namespace softbus::edge { + +EthChannelManager::EthChannelManager() + : io_(1) +{ +} + +bool EthChannelManager::openChannel(const PhysicalChannelDef& channel) +{ + closeChannel(channel.channelId); + ChannelHandle handle; + handle.config = channel; + + try { + if (channel.transport == "udp") { + asio::ip::udp::endpoint endpoint( + asio::ip::make_address(channel.host.empty() ? "0.0.0.0" : channel.host), + channel.tcpPort); + handle.udpSocket = std::make_unique(io_); + handle.udpSocket->open(endpoint.protocol()); + if (channel.host.empty()) { + handle.udpSocket->bind(endpoint); + } else { + handle.udpRemote = endpoint; + } + } else { + handle.tcpSocket = std::make_unique(io_); + asio::ip::tcp::endpoint endpoint( + asio::ip::make_address(channel.host.empty() ? "127.0.0.1" : channel.host), + channel.tcpPort); + handle.tcpSocket->connect(endpoint); + handle.tcpSocket->non_blocking(true); + } + channels_[channel.channelId] = std::move(handle); + SB_LOG_INFO("EthChannel", "Opened raw channel ", channel.channelId); + return true; + } catch (const std::exception& ex) { + SB_LOG_WARN("EthChannel", "Failed to open channel ", channel.channelId, ": ", ex.what()); + return false; + } +} + +void EthChannelManager::closeChannel(const std::string& channelId) +{ + channels_.erase(channelId); +} + +void EthChannelManager::closeAll() +{ + channels_.clear(); +} + +std::vector> EthChannelManager::recv(const std::string& channelId, + int maxPackets) +{ + std::vector> packets; + const auto it = channels_.find(channelId); + if (it == channels_.end()) { + return packets; + } + + for (int i = 0; i < maxPackets; ++i) { + std::array buffer{}; + std::error_code ec; + std::size_t n = 0; + if (it->second.tcpSocket) { + n = it->second.tcpSocket->read_some(asio::buffer(buffer), ec); + } else if (it->second.udpSocket) { + asio::ip::udp::endpoint remote; + n = it->second.udpSocket->receive_from(asio::buffer(buffer), remote, 0, ec); + } + if (ec || n == 0) { + break; + } + packets.emplace_back(buffer.begin(), buffer.begin() + static_cast(n)); + } + return packets; +} + +bool EthChannelManager::send(const std::string& channelId, const std::vector& data) +{ + const auto it = channels_.find(channelId); + if (it == channels_.end()) { + return false; + } + std::error_code ec; + if (it->second.tcpSocket) { + asio::write(*it->second.tcpSocket, asio::buffer(data), ec); + } else if (it->second.udpSocket) { + it->second.udpSocket->send_to(asio::buffer(data), it->second.udpRemote, 0, ec); + } + return !ec; +} + +} // namespace softbus::edge diff --git a/src/softbus_edge/src/ModbusChannel.cpp b/src/softbus_edge/src/driver/modbus/ModbusChannel.cpp similarity index 58% rename from src/softbus_edge/src/ModbusChannel.cpp rename to src/softbus_edge/src/driver/modbus/ModbusChannel.cpp index 3dd8b3b..49f654e 100644 --- a/src/softbus_edge/src/ModbusChannel.cpp +++ b/src/softbus_edge/src/driver/modbus/ModbusChannel.cpp @@ -1,4 +1,4 @@ -#include +#include #include @@ -6,10 +6,19 @@ #include #endif -#include +#include namespace softbus::edge { +namespace { + +bool isTcpRtuTransport(const PhysicalChannelDef& channel) +{ + return channel.transport == "tcp_rtu" || channel.transport == "rtu_tcp"; +} + +} // namespace + int ModbusChannelManager::modbusAddress(int registerAddress) const { if (registerAddress >= 40001) { @@ -27,6 +36,21 @@ bool ModbusChannelManager::openChannel(const PhysicalChannelDef& channel) ChannelHandle handle; handle.config = channel; + if (isTcpRtuTransport(channel)) { + handle.tcpRtu = std::make_unique(); + if (!handle.tcpRtu->connect(channel.host, channel.tcpPort)) { + SB_LOG_ERROR( + "ModbusChannel", "Failed RTU-over-TCP connect ", channel.channelId, + " ", channel.host, ":", channel.tcpPort); + return false; + } + channels_[channel.channelId] = std::move(handle); + SB_LOG_INFO( + "ModbusChannel", "Opened RTU-over-TCP channel ", channel.channelId, + " ", channel.host, ":", channel.tcpPort); + return true; + } + #if defined(SOFTBUS_HAS_LIBMODBUS) && SOFTBUS_HAS_LIBMODBUS if (channel.transport == "tcp") { handle.ctx = modbus_new_tcp(channel.host.c_str(), channel.tcpPort); @@ -40,12 +64,15 @@ bool ModbusChannelManager::openChannel(const PhysicalChannelDef& channel) return false; } modbus_set_response_timeout(static_cast(handle.ctx), 0, 500000); + if (channel.transport == "tcp") { + modbus_enable_quirks(static_cast(handle.ctx), MODBUS_QUIRK_MAX_SLAVE); + } 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; + channels_[channel.channelId] = std::move(handle); SB_LOG_INFO("ModbusChannel", "Opened channel ", channel.channelId); return true; #else @@ -62,6 +89,9 @@ void ModbusChannelManager::closeChannel(const std::string& channelId) if (it == channels_.end()) { return; } + if (it->second.tcpRtu) { + it->second.tcpRtu->disconnect(); + } #if defined(SOFTBUS_HAS_LIBMODBUS) && SOFTBUS_HAS_LIBMODBUS if (it->second.ctx) { modbus_close(static_cast(it->second.ctx)); @@ -81,37 +111,69 @@ void ModbusChannelManager::closeAll() std::optional ModbusChannelManager::readRegister(const ResolvedPhysicalPoint& point) { -#if defined(SOFTBUS_HAS_LIBMODBUS) && SOFTBUS_HAS_LIBMODBUS + const auto values = readRegisters(point, modbusAddress(point.point.registerAddress), 1); + if (!values || values->empty()) { + return std::nullopt; + } + return values->front(); +} + +std::optional> ModbusChannelManager::readRegisters( + const ResolvedPhysicalPoint& point, int startAddr, int count) +{ const auto it = channels_.find(point.channel.channelId); - if (it == channels_.end() || !it->second.ctx) { + if (it == channels_.end() || count <= 0) { + return std::nullopt; + } + + if (it->second.tcpRtu) { + return it->second.tcpRtu->readHoldingRegisters( + point.device.slaveId, startAddr, count); + } + +#if defined(SOFTBUS_HAS_LIBMODBUS) && SOFTBUS_HAS_LIBMODBUS + if (!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); + std::vector values(static_cast(count)); const int rc = modbus_read_registers( - static_cast(it->second.ctx), addr, 1, &value); - if (rc != 1) { + static_cast(it->second.ctx), startAddr, count, values.data()); + if (rc != count) { return std::nullopt; } - return value; + return values; #else static uint16_t sim = 250; - sim = static_cast((sim + 1) % 500); + std::vector values(static_cast(count)); + for (int i = 0; i < count; ++i) { + sim = static_cast((sim + 1) % 500); + values[static_cast(i)] = sim; + } (void)point; - return sim; + (void)startAddr; + return values; #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) { + if (it == channels_.end()) { + return false; + } + + const int addr = modbusAddress(point.point.registerAddress); + if (it->second.tcpRtu) { + return it->second.tcpRtu->writeSingleRegister( + point.device.slaveId, addr, rawValue); + } + +#if defined(SOFTBUS_HAS_LIBMODBUS) && SOFTBUS_HAS_LIBMODBUS + if (!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; diff --git a/src/softbus_edge/src/driver/modbus/TcpRtuSession.cpp b/src/softbus_edge/src/driver/modbus/TcpRtuSession.cpp new file mode 100644 index 0000000..674c6f2 --- /dev/null +++ b/src/softbus_edge/src/driver/modbus/TcpRtuSession.cpp @@ -0,0 +1,159 @@ +#include + +#include +#include +#include +#include +#include +#include + +#include +#include + +namespace softbus::edge { + +uint16_t TcpRtuSession::crc16(const uint8_t* data, std::size_t len) +{ + uint16_t crc = 0xFFFF; + for (std::size_t i = 0; i < len; ++i) { + crc ^= data[i]; + for (int bit = 0; bit < 8; ++bit) { + if (crc & 0x0001) { + crc = static_cast((crc >> 1) ^ 0xA001); + } else { + crc = static_cast(crc >> 1); + } + } + } + return crc; +} + +bool TcpRtuSession::connect(const std::string& host, uint16_t port, int timeoutMs) +{ + disconnect(); + const int fd = socket(AF_INET, SOCK_STREAM, 0); + if (fd < 0) { + return false; + } + + sockaddr_in addr {}; + addr.sin_family = AF_INET; + addr.sin_port = htons(port); + if (inet_pton(AF_INET, host.c_str(), &addr.sin_addr) != 1) { + close(fd); + return false; + } + + const int flags = 1; + setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &flags, sizeof(flags)); + + if (::connect(fd, reinterpret_cast(&addr), sizeof(addr)) != 0) { + close(fd); + return false; + } + + timeval tv {}; + tv.tv_sec = timeoutMs / 1000; + tv.tv_usec = (timeoutMs % 1000) * 1000; + setsockopt(fd, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv)); + setsockopt(fd, SOL_SOCKET, SO_SNDTIMEO, &tv, sizeof(tv)); + + socketFd_ = fd; + return true; +} + +void TcpRtuSession::disconnect() +{ + if (socketFd_ >= 0) { + close(socketFd_); + socketFd_ = -1; + } +} + +std::optional> TcpRtuSession::transaction(const std::vector& pdu) +{ + if (socketFd_ < 0 || pdu.empty()) { + return std::nullopt; + } + + const uint16_t crc = crc16(pdu.data(), pdu.size()); + std::vector frame = pdu; + frame.push_back(static_cast(crc & 0xFF)); + frame.push_back(static_cast((crc >> 8) & 0xFF)); + + if (send(socketFd_, frame.data(), frame.size(), 0) != static_cast(frame.size())) { + return std::nullopt; + } + + uint8_t buffer[256]{}; + const ssize_t n = recv(socketFd_, buffer, sizeof(buffer), 0); + if (n < 5) { + return std::nullopt; + } + + return std::vector(buffer, buffer + n); +} + +std::optional> TcpRtuSession::readHoldingRegisters(int slaveId, + int startAddr, + int count) +{ + if (count <= 0 || count > 125) { + return std::nullopt; + } + + const auto slave = static_cast(slaveId); + std::vector pdu{ + slave, + 0x03, + static_cast((startAddr >> 8) & 0xFF), + static_cast(startAddr & 0xFF), + static_cast((count >> 8) & 0xFF), + static_cast(count & 0xFF) + }; + + const auto response = transaction(pdu); + if (!response || response->size() < 5) { + return std::nullopt; + } + if ((*response)[0] != slave || (*response)[1] != 0x03) { + return std::nullopt; + } + if ((*response)[1] & 0x80) { + return std::nullopt; + } + + const int byteCount = (*response)[2]; + if (byteCount != count * 2 || static_cast(3 + byteCount + 2) > response->size()) { + return std::nullopt; + } + + std::vector values(static_cast(count)); + for (int i = 0; i < count; ++i) { + const std::size_t off = static_cast(3 + i * 2); + values[static_cast(i)] = static_cast( + (static_cast((*response)[off]) << 8) | (*response)[off + 1]); + } + return values; +} + +bool TcpRtuSession::writeSingleRegister(int slaveId, int address, uint16_t value) +{ + const auto slave = static_cast(slaveId); + std::vector pdu{ + slave, + 0x06, + static_cast((address >> 8) & 0xFF), + static_cast(address & 0xFF), + static_cast((value >> 8) & 0xFF), + static_cast(value & 0xFF) + }; + + const auto response = transaction(pdu); + if (!response || response->size() < 8) { + return false; + } + return (*response)[0] == slave && (*response)[1] == 0x06; +} + +} // namespace softbus::edge diff --git a/src/softbus_edge/src/egress/can/CanEgress.cpp b/src/softbus_edge/src/egress/can/CanEgress.cpp new file mode 100644 index 0000000..fc3166c --- /dev/null +++ b/src/softbus_edge/src/egress/can/CanEgress.cpp @@ -0,0 +1,38 @@ +#include + +#include + +namespace softbus::edge { + +CanEgress::CanEgress(PhysicalRegistry* registry, CanChannelManager* channels) + : registry_(registry) + , channels_(channels) +{ +} + +bool CanEgress::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; + } + + std::vector data; + data.push_back(static_cast(raw & 0xFF)); + data.push_back(static_cast((raw >> 8) & 0xFF)); + return sdo_.download(channels_, *resolved, data); +} + +} // namespace softbus::edge diff --git a/src/softbus_edge/src/egress/eth/EthEgress.cpp b/src/softbus_edge/src/egress/eth/EthEgress.cpp new file mode 100644 index 0000000..85332ba --- /dev/null +++ b/src/softbus_edge/src/egress/eth/EthEgress.cpp @@ -0,0 +1,24 @@ +#include + +namespace softbus::edge { + +EthEgress::EthEgress(PhysicalRegistry* registry, EthChannelManager* channels) + : registry_(registry) + , channels_(channels) +{ +} + +bool EthEgress::writeByPhysicalPointId(const std::string& physicalPointId, + const std::vector& data) +{ + if (!registry_ || !channels_) { + return false; + } + const auto resolved = registry_->resolvePoint(physicalPointId); + if (!resolved || !resolved->point.writable) { + return false; + } + return channels_->send(resolved->channel.channelId, data); +} + +} // namespace softbus::edge diff --git a/src/softbus_edge/src/ModbusEgress.cpp b/src/softbus_edge/src/egress/modbus/ModbusEgress.cpp similarity index 94% rename from src/softbus_edge/src/ModbusEgress.cpp rename to src/softbus_edge/src/egress/modbus/ModbusEgress.cpp index 8c4e38a..1d50787 100644 --- a/src/softbus_edge/src/ModbusEgress.cpp +++ b/src/softbus_edge/src/egress/modbus/ModbusEgress.cpp @@ -1,4 +1,4 @@ -#include +#include #include diff --git a/src/softbus_edge/src/ingress/CompositeIngress.cpp b/src/softbus_edge/src/ingress/CompositeIngress.cpp new file mode 100644 index 0000000..2bd71ee --- /dev/null +++ b/src/softbus_edge/src/ingress/CompositeIngress.cpp @@ -0,0 +1,38 @@ +#include + +namespace softbus::edge { + +CompositeIngress::CompositeIngress(std::vector> children) + : children_(std::move(children)) +{ +} + +bool CompositeIngress::open() +{ + bool ok = true; + for (const auto& child : children_) { + ok = child->open() && ok; + } + return ok || children_.empty(); +} + +void CompositeIngress::close() +{ + for (const auto& child : children_) { + child->close(); + } +} + +std::vector CompositeIngress::poll() +{ + std::vector packets; + for (const auto& child : children_) { + auto batch = child->poll(); + packets.insert(packets.end(), + std::make_move_iterator(batch.begin()), + std::make_move_iterator(batch.end())); + } + return packets; +} + +} // namespace softbus::edge diff --git a/src/softbus_edge/src/ingress/can/CanIngress.cpp b/src/softbus_edge/src/ingress/can/CanIngress.cpp new file mode 100644 index 0000000..cef9e35 --- /dev/null +++ b/src/softbus_edge/src/ingress/can/CanIngress.cpp @@ -0,0 +1,165 @@ +#include + +#include +#include +#include + +#include + +namespace softbus::edge { + +CanIngress::CanIngress(PhysicalRegistry* registry, + CanChannelManager* channels, + std::string edgeIdPrefix, + std::string configBaseDir) + : registry_(registry) + , channels_(channels) + , edgeIdPrefix_(std::move(edgeIdPrefix)) + , configBaseDir_(std::move(configBaseDir)) +{ +} + +bool CanIngress::loadDictionaries() +{ + dictionaries_.clear(); + for (const auto& device : registry_->devices()) { + if (device.protocol != "canopen" || device.edsPath.empty()) { + continue; + } + const auto path = std::filesystem::path(configBaseDir_) / device.edsPath; + canopen::CanopenObjectDictionary dictionary; + if (dictionary.loadEds(path.string())) { + dictionaries_[device.physicalDeviceId] = std::move(dictionary); + } else { + SB_LOG_WARN("CanIngress", "Failed to load EDS for ", device.physicalDeviceId); + } + } + return true; +} + +bool CanIngress::open() +{ + if (!registry_ || !channels_) { + return false; + } + loadDictionaries(); + for (const auto& ch : registry_->channelsByProtocol("canopen")) { + if (!channels_->openChannel(ch)) { + return false; + } + } + for (const auto& device : registry_->devices()) { + if (device.protocol != "canopen" || device.nodeId == 0) { + continue; + } + nmt_.startRemoteNode(channels_, device.channelId, static_cast(device.nodeId)); + } + pdoMapper_.rebuild(*registry_, dictionaries_); + planner_.reset(*registry_, "canopen"); + changeFilter_.clear(); + open_ = true; + return true; +} + +void CanIngress::close() +{ + if (channels_) { + for (const auto& ch : registry_->channelsByProtocol("canopen")) { + channels_->closeChannel(ch.channelId); + } + } + open_ = false; +} + +double CanIngress::decodeValue(const MappedCanPoint& mapped, + const std::vector& data) const +{ + if (mapped.byteOffset + 1 >= static_cast(data.size())) { + return 0.0; + } + const uint16_t raw = static_cast(data[static_cast(mapped.byteOffset)]) | + (static_cast(data[static_cast(mapped.byteOffset + 1)]) << 8); + return static_cast(raw) * mapped.scale; +} + +std::vector CanIngress::buildCanopenPayload(uint32_t cobId, + const std::vector& data) const +{ + std::vector payload; + payload.push_back(static_cast(cobId & 0xFF)); + payload.push_back(static_cast((cobId >> 8) & 0xFF)); + payload.push_back(static_cast(data.size())); + payload.insert(payload.end(), data.begin(), data.end()); + return payload; +} + +std::vector CanIngress::poll() +{ + if (!open_ || !registry_ || !channels_) { + return {}; + } + + std::vector packets; + + for (const auto& sub : pdoMapper_.subscriptions()) { + const auto frames = channels_->recvFrames(sub.channelId, 16); + for (const auto& frame : frames) { + const auto matched = pdoMapper_.matchFrame(frame.canId, frame.data); + for (const auto& mapped : matched) { + const double engineering = decodeValue(mapped, frame.data); + const auto resolved = registry_->resolvePoint(mapped.physicalPointId); + if (!resolved) { + continue; + } + if (!changeFilter_.shouldReport(resolved->point, engineering)) { + continue; + } + changeFilter_.update(mapped.physicalPointId, engineering); + + softbus::core::RawPacket packet; + packet.traceId = softbus::core::DeviceIdentity::generateTraceId(); + packet.timestampNs = softbus::core::HwClock::nowNs(); + packet.runtimeDeviceId = edgeIdPrefix_; + packet.sourceId = mapped.physicalPointId; + packet.protocolTag = "canopen"; + packet.payload = buildCanopenPayload(frame.canId, frame.data); + packets.push_back(std::move(packet)); + } + } + } + + const auto dueIds = planner_.duePointIds(); + for (const auto& pointId : dueIds) { + const auto resolved = registry_->resolvePoint(pointId); + if (!resolved || resolved->point.pdoType != "sdo") { + continue; + } + std::vector data; + if (!sdo_.upload(channels_, *resolved, data)) { + continue; + } + uint16_t raw = data.size() >= 2 + ? static_cast(data[0] | (data[1] << 8)) + : (data.empty() ? 0 : data[0]); + const double engineering = static_cast(raw) * resolved->point.scale; + if (!changeFilter_.shouldReport(resolved->point, engineering)) { + planner_.markPolled(pointId); + continue; + } + changeFilter_.update(pointId, engineering); + + softbus::core::RawPacket packet; + packet.traceId = softbus::core::DeviceIdentity::generateTraceId(); + packet.timestampNs = softbus::core::HwClock::nowNs(); + packet.runtimeDeviceId = edgeIdPrefix_; + packet.sourceId = pointId; + packet.protocolTag = "canopen"; + packet.payload = data; + packets.push_back(std::move(packet)); + planner_.markPolled(pointId); + } + + return packets; +} + +} // namespace softbus::edge diff --git a/src/softbus_edge/src/ingress/eth/EthIngress.cpp b/src/softbus_edge/src/ingress/eth/EthIngress.cpp new file mode 100644 index 0000000..92c4a09 --- /dev/null +++ b/src/softbus_edge/src/ingress/eth/EthIngress.cpp @@ -0,0 +1,62 @@ +#include + +#include +#include + +namespace softbus::edge { + +EthIngress::EthIngress(PhysicalRegistry* registry, + EthChannelManager* channels, + std::string edgeIdPrefix) + : registry_(registry) + , channels_(channels) + , edgeIdPrefix_(std::move(edgeIdPrefix)) +{ +} + +bool EthIngress::open() +{ + if (!registry_ || !channels_) { + return false; + } + for (const auto& ch : registry_->channelsByProtocol("raw")) { + if (!channels_->openChannel(ch)) { + return false; + } + } + open_ = true; + return true; +} + +void EthIngress::close() +{ + if (channels_) { + channels_->closeAll(); + } + open_ = false; +} + +std::vector EthIngress::poll() +{ + if (!open_ || !registry_ || !channels_) { + return {}; + } + + std::vector packets; + for (const auto& ch : registry_->channelsByProtocol("raw")) { + const auto chunks = channels_->recv(ch.channelId, 8); + for (const auto& chunk : chunks) { + softbus::core::RawPacket packet; + packet.traceId = softbus::core::DeviceIdentity::generateTraceId(); + packet.timestampNs = softbus::core::HwClock::nowNs(); + packet.runtimeDeviceId = edgeIdPrefix_; + packet.sourceId = edgeIdPrefix_ + "/raw/" + ch.channelId; + packet.protocolTag = "raw_eth"; + packet.payload = chunk; + packets.push_back(std::move(packet)); + } + } + return packets; +} + +} // namespace softbus::edge diff --git a/src/softbus_edge/src/MockModbusIngress.cpp b/src/softbus_edge/src/ingress/modbus/MockModbusIngress.cpp similarity index 83% rename from src/softbus_edge/src/MockModbusIngress.cpp rename to src/softbus_edge/src/ingress/modbus/MockModbusIngress.cpp index 8eca06a..8326c13 100644 --- a/src/softbus_edge/src/MockModbusIngress.cpp +++ b/src/softbus_edge/src/ingress/modbus/MockModbusIngress.cpp @@ -1,4 +1,4 @@ -#include +#include #include #include @@ -24,7 +24,6 @@ void MockModbusIngress::close() std::vector MockModbusIngress::buildFrame(uint16_t registerValue) const { - // slave=1, func=0x03, byteCount=2, value hi/lo, crc lo/hi (placeholder) return { 0x01, 0x03, 0x02, static_cast((registerValue >> 8) & 0xFF), @@ -39,9 +38,8 @@ std::vector MockModbusIngress::poll() return {}; } - const auto traceId = softbus::core::DeviceIdentity::generateTraceId(); softbus::core::RawPacket packet; - packet.traceId = traceId; + packet.traceId = softbus::core::DeviceIdentity::generateTraceId(); packet.timestampNs = softbus::core::HwClock::nowNs(); packet.runtimeDeviceId = runtimeDeviceId_; packet.sourceId = objectRef_; diff --git a/src/softbus_edge/src/ingress/modbus/ModbusIngress.cpp b/src/softbus_edge/src/ingress/modbus/ModbusIngress.cpp new file mode 100644 index 0000000..5407e3a --- /dev/null +++ b/src/softbus_edge/src/ingress/modbus/ModbusIngress.cpp @@ -0,0 +1,133 @@ +#include + +#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_->channelsByProtocol("modbus")) { + if (!channels_->openChannel(ch)) { + return false; + } + } + planner_.reset(*registry_, "modbus"); + changeFilter_.clear(); + open_ = true; + return true; +} + +void ModbusIngress::close() +{ + if (channels_) { + for (const auto& ch : registry_->channelsByProtocol("modbus")) { + channels_->closeChannel(ch.channelId); + } + } + 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 dueIds = planner_.duePointIds(); + if (dueIds.empty()) { + return {}; + } + + std::vector duePoints; + duePoints.reserve(dueIds.size()); + for (const auto& pointId : dueIds) { + const auto resolved = registry_->resolvePoint(pointId); + if (resolved && resolved->device.protocol == "modbus") { + duePoints.push_back(*resolved); + } + } + + const auto batches = scheduler_.planBatches(duePoints); + std::vector packets; + + for (const auto& batch : batches) { + if (batch.points.empty()) { + continue; + } + const auto firstResolved = registry_->resolvePoint(batch.points.front().physicalPointId); + if (!firstResolved) { + continue; + } + + const auto values = channels_->readRegisters(*firstResolved, batch.startAddr, batch.count); + if (!values) { + continue; + } + + SB_LOG_DEBUG("ModbusIngress", + "Batch read channel=", batch.channelId, + " slave=", batch.slaveId, + " start=", batch.startAddr, + " count=", batch.count); + + for (const auto& batchPoint : batch.points) { + const auto resolved = registry_->resolvePoint(batchPoint.physicalPointId); + if (!resolved) { + continue; + } + if (batchPoint.offsetInBatch < 0 || + static_cast(batchPoint.offsetInBatch) >= values->size()) { + continue; + } + + const uint16_t raw = (*values)[static_cast(batchPoint.offsetInBatch)]; + const double engineering = raw * resolved->point.scale; + if (!changeFilter_.shouldReport(resolved->point, engineering)) { + planner_.markPolled(batchPoint.physicalPointId); + continue; + } + changeFilter_.update(batchPoint.physicalPointId, engineering); + + softbus::core::RawPacket packet; + packet.traceId = softbus::core::DeviceIdentity::generateTraceId(); + packet.timestampNs = softbus::core::HwClock::nowNs(); + packet.runtimeDeviceId = edgeIdPrefix_; + packet.sourceId = batchPoint.physicalPointId; + packet.protocolTag = "modbus"; + packet.payload = buildModbusFrame(resolved->device.slaveId, raw); + packets.push_back(std::move(packet)); + planner_.markPolled(batchPoint.physicalPointId); + } + } + + return packets; +} + +} // namespace softbus::edge diff --git a/src/softbus_edge/src/main.cpp b/src/softbus_edge/src/main.cpp index 3783abd..2511621 100644 --- a/src/softbus_edge/src/main.cpp +++ b/src/softbus_edge/src/main.cpp @@ -1,9 +1,4 @@ // 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 diff --git a/src/softbus_edge/src/scheduler/ChangeFilter.cpp b/src/softbus_edge/src/scheduler/ChangeFilter.cpp new file mode 100644 index 0000000..306c7b1 --- /dev/null +++ b/src/softbus_edge/src/scheduler/ChangeFilter.cpp @@ -0,0 +1,35 @@ +#include + +#include + +namespace softbus::edge { + +bool ChangeFilter::shouldReport(const PhysicalPointDef& point, double engineeringValue) const +{ + if (!point.reportOnChange) { + return true; + } + const auto it = hasSample_.find(point.physicalPointId); + if (it == hasSample_.end() || !it->second) { + return true; + } + const auto valueIt = lastValues_.find(point.physicalPointId); + if (valueIt == lastValues_.end()) { + return true; + } + return std::abs(engineeringValue - valueIt->second) > point.deadband; +} + +void ChangeFilter::update(const std::string& physicalPointId, double engineeringValue) +{ + lastValues_[physicalPointId] = engineeringValue; + hasSample_[physicalPointId] = true; +} + +void ChangeFilter::clear() +{ + lastValues_.clear(); + hasSample_.clear(); +} + +} // namespace softbus::edge diff --git a/src/softbus_edge/src/scheduler/ModbusPollScheduler.cpp b/src/softbus_edge/src/scheduler/ModbusPollScheduler.cpp new file mode 100644 index 0000000..5c55bed --- /dev/null +++ b/src/softbus_edge/src/scheduler/ModbusPollScheduler.cpp @@ -0,0 +1,103 @@ +#include + +#include +#include +#include + +namespace softbus::edge { + +namespace { + +constexpr int kMaxRegistersPerRead = 125; + +struct SortablePoint { + ResolvedPhysicalPoint resolved; + int address{0}; +}; + +} // namespace + +int ModbusPollScheduler::modbusAddress(int registerAddress) +{ + if (registerAddress >= 40001) { + return registerAddress - 40001; + } + if (registerAddress >= 30001) { + return registerAddress - 30001; + } + return registerAddress; +} + +std::vector ModbusPollScheduler::planBatches( + const std::vector& duePoints) const +{ + std::map, std::vector> grouped; + + for (const auto& resolved : duePoints) { + SortablePoint item; + item.resolved = resolved; + item.address = modbusAddress(resolved.point.registerAddress); + const auto key = std::make_tuple( + resolved.channel.channelId, resolved.device.slaveId, resolved.point.functionCode); + grouped[key].push_back(item); + } + + std::vector batches; + for (auto& [key, points] : grouped) { + std::sort(points.begin(), points.end(), + [](const SortablePoint& a, const SortablePoint& b) { + return a.address < b.address; + }); + + PollBatch current; + current.channelId = std::get<0>(key); + current.slaveId = std::get<1>(key); + current.functionCode = std::get<2>(key); + current.startAddr = -1; + current.count = 0; + + auto flush = [&]() { + if (current.count > 0) { + batches.push_back(current); + } + current = PollBatch{}; + current.channelId = std::get<0>(key); + current.slaveId = std::get<1>(key); + current.functionCode = std::get<2>(key); + current.startAddr = -1; + current.count = 0; + }; + + for (const auto& item : points) { + if (current.count == 0) { + current.startAddr = item.address; + current.count = 1; + current.points.push_back( + PollBatchPoint{item.resolved.point.physicalPointId, 0}); + continue; + } + + const int nextAddr = item.address; + const int endAddr = current.startAddr + current.count - 1; + const bool contiguous = nextAddr == endAddr + 1; + const bool withinLimit = current.count < kMaxRegistersPerRead; + + if (contiguous && withinLimit) { + current.points.push_back( + PollBatchPoint{item.resolved.point.physicalPointId, nextAddr - current.startAddr}); + ++current.count; + } else { + flush(); + current.startAddr = item.address; + current.count = 1; + current.points.push_back( + PollBatchPoint{item.resolved.point.physicalPointId, 0}); + } + } + flush(); + } + + return batches; +} + +} // namespace softbus::edge diff --git a/src/softbus_edge/src/scheduler/PollPlanner.cpp b/src/softbus_edge/src/scheduler/PollPlanner.cpp new file mode 100644 index 0000000..0e22975 --- /dev/null +++ b/src/softbus_edge/src/scheduler/PollPlanner.cpp @@ -0,0 +1,44 @@ +#include + +namespace softbus::edge { + +void PollPlanner::reset(const PhysicalRegistry& registry, const std::string& protocol) +{ + schedules_.clear(); + const auto now = std::chrono::steady_clock::now(); + for (const auto& point : registry.pollPointsByProtocol(protocol)) { + const auto resolved = registry.resolvePoint(point.physicalPointId); + if (!resolved) { + continue; + } + PointSchedule schedule; + schedule.intervalMs = point.pollIntervalMs > 0 ? point.pollIntervalMs + : resolved->channel.pollIntervalMs; + schedule.nextDue = now; + schedules_[point.physicalPointId] = schedule; + } +} + +std::vector PollPlanner::duePointIds() +{ + std::vector due; + const auto now = std::chrono::steady_clock::now(); + for (auto& [pointId, schedule] : schedules_) { + if (now >= schedule.nextDue) { + due.push_back(pointId); + } + } + return due; +} + +void PollPlanner::markPolled(const std::string& physicalPointId) +{ + const auto it = schedules_.find(physicalPointId); + if (it == schedules_.end()) { + return; + } + it->second.nextDue = std::chrono::steady_clock::now() + + std::chrono::milliseconds(it->second.intervalMs); +} + +} // namespace softbus::edge diff --git a/src/softbus_transport/src/TcpUplinkClient.cpp b/src/softbus_transport/src/TcpUplinkClient.cpp index 2e10bc1..5f33dce 100644 --- a/src/softbus_transport/src/TcpUplinkClient.cpp +++ b/src/softbus_transport/src/TcpUplinkClient.cpp @@ -29,6 +29,9 @@ bool TcpUplinkClient::connect() if (connected_) { return true; } + if (ioThread_.joinable()) { + disconnect(); + } socket_ = std::make_unique(io_); std::error_code ec; socket_->connect(asio::ip::tcp::endpoint(asio::ip::make_address(host_), port_), ec);