phase 2 tcp

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)

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()

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
     }
   ]

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": "阀门开度",

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",

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

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",

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
     }
   ]
 }

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
+        }
+      ]
     }
   ]
 }

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

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
+    }
+  ]
+}

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
+        }
+      ]
+    }
+  ]
+}

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<TcpRtuSession>();
+        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 线协议

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<RawPacket>` |
 
-**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 |

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)

src/plugins/protocol_canopen/include/canopen/CanopenNodeManager.h

@@ -1,19 +0,0 @@
-#pragma once
-
-#include <cstdint>
-#include <map>
-#include <string>
-
-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<uint8_t, std::string> states_;
-};
-
-} // namespace canopen

src/plugins/protocol_canopen/include/canopen/CanopenObjectDictionary.h

@@ -1,25 +0,0 @@
-#pragma once
-
-#include <softbus_core/models/DOMMessage.h>
-
-#include <cstdint>
-#include <map>
-#include <string>
-
-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<uint32_t, Entry> entries_;
-};
-
-} // namespace canopen

src/plugins/protocol_canopen/src/canopen_plugin.cpp

@@ -1,9 +1,6 @@
 #include <canopen/CanopenFramer.h>
 #include <canopen/CanopenParser.h>
 #include <softbus_core/plugin/IParser.h>
-#include <canopen/CanopenObjectDictionary.h>
-#include <canopen/EdsDcfLoader.h>
-#include <canopen/CanopenNodeManager.h>
 #include <canopen/CanopenMapper.h>
 
 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

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)

src/softbus_canopen_common/include/canopen/CanopenObjectDictionary.h

@@ -0,0 +1,35 @@
+#pragma once
+
+#include <canopen/CanopenTypes.h>
+
+#include <cstdint>
+#include <map>
+#include <optional>
+#include <string>
+#include <vector>
+
+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<ObjectEntry> entry(uint16_t index, uint8_t subIndex) const;
+    std::vector<PdoDefinition> tpdoDefinitions() const;
+    std::vector<PdoDefinition> 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<uint32_t, ObjectEntry> entries_;
+    std::map<uint16_t, PdoDefinition> tpdoByCommIndex_;
+    std::map<uint16_t, PdoDefinition> rpdoByCommIndex_;
+};
+
+} // namespace canopen

src/softbus_canopen_common/include/canopen/CanopenTypes.h

@@ -0,0 +1,30 @@
+#pragma once
+
+#include <cstdint>
+#include <string>
+#include <vector>
+
+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<PdoMappingEntry> mappings;
+};
+
+} // namespace canopen

src/softbus_canopen_common/src/CanopenObjectDictionary.cpp

@@ -0,0 +1,78 @@
+#include <canopen/CanopenObjectDictionary.h>
+#include <canopen/EdsDcfLoader.h>
+
+namespace canopen {
+
+uint32_t CanopenObjectDictionary::makeKey(uint16_t index, uint8_t subIndex)
+{
+    return (static_cast<uint32_t>(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<ObjectEntry> 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<PdoDefinition> CanopenObjectDictionary::tpdoDefinitions() const
+{
+    std::vector<PdoDefinition> out;
+    out.reserve(tpdoByCommIndex_.size());
+    for (const auto& [_, def] : tpdoByCommIndex_) {
+        out.push_back(def);
+    }
+    return out;
+}
+
+std::vector<PdoDefinition> CanopenObjectDictionary::rpdoDefinitions() const
+{
+    std::vector<PdoDefinition> 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

src/softbus_canopen_common/src/EdsDcfLoader.cpp

@@ -0,0 +1,190 @@
+#include <canopen/EdsDcfLoader.h>
+
+#include <cctype>
+#include <fstream>
+#include <sstream>
+
+namespace canopen {
+
+namespace {
+
+uint16_t parseHexIndex(const std::string& section)
+{
+    if (section.empty()) {
+        return 0;
+    }
+    return static_cast<uint16_t>(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<uint32_t>(std::stoul(trimmed, nullptr, 16));
+    }
+    if (!trimmed.empty() && std::isxdigit(static_cast<unsigned char>(trimmed[0]))) {
+        return static_cast<uint32_t>(std::stoul(trimmed, nullptr, 16));
+    }
+    return static_cast<uint32_t>(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<uint16_t>((currentEntry.defaultValue >> 16) & 0xFFFF);
+            map.subIndex = static_cast<uint8_t>(currentEntry.defaultValue & 0xFF);
+            map.bitLength = 16;
+            currentPdo.mappings.push_back(map);
+        } else if (isRpdoMappingIndex(currentIndex) && currentSub >= 1) {
+            PdoMappingEntry map;
+            map.index = static_cast<uint16_t>((currentEntry.defaultValue >> 16) & 0xFFFF);
+            map.subIndex = static_cast<uint8_t>(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<uint16_t>(currentIndex - 0x200);
+                    dictionary->registerTpdo(currentPdo);
+                } else {
+                    currentPdo.commParamIndex = static_cast<uint16_t>(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<uint8_t>(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<uint16_t>(currentIndex - 0x200)
+                    : isRpdoMappingIndex(currentIndex)
+                        ? static_cast<uint16_t>(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<uint8_t>(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<uint16_t>(currentIndex - 0x200);
+            dictionary->registerTpdo(currentPdo);
+        } else {
+            currentPdo.commParamIndex = static_cast<uint16_t>(currentIndex - 0x200);
+            dictionary->registerRpdo(currentPdo);
+        }
+    }
+
+    return !dictionary->tpdoDefinitions().empty() || !dictionary->rpdoDefinitions().empty() ||
+           dictionary->hasEntry(0x1000, 0);
+}
+
+} // namespace canopen

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)

src/softbus_edge/include/softbus_edge/canopen/NmtManager.h

@@ -0,0 +1,23 @@
+#pragma once
+
+#include <softbus_edge/driver/can/CanChannel.h>
+
+#include <cstdint>
+#include <string>
+#include <unordered_map>
+
+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<uint8_t, std::string> states_;
+};
+
+} // namespace softbus::edge

src/softbus_edge/include/softbus_edge/canopen/PdoMapper.h

@@ -0,0 +1,41 @@
+#pragma once
+
+#include <canopen/CanopenObjectDictionary.h>
+
+#include <softbus_edge/config/PhysicalRegistry.h>
+
+#include <cstdint>
+#include <string>
+#include <unordered_map>
+#include <vector>
+
+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<MappedCanPoint> points;
+};
+
+class PdoMapper {
+public:
+    void rebuild(const PhysicalRegistry& registry,
+                 const std::unordered_map<std::string, canopen::CanopenObjectDictionary>& dictionaries);
+    const std::vector<PdoSubscription>& subscriptions() const { return subscriptions_; }
+    std::vector<MappedCanPoint> matchFrame(uint32_t cobId, const std::vector<uint8_t>& data) const;
+
+private:
+    std::vector<PdoSubscription> subscriptions_;
+    std::unordered_map<uint32_t, std::size_t> cobIndex_;
+};
+
+} // namespace softbus::edge

src/softbus_edge/include/softbus_edge/canopen/SdoClient.h

@@ -0,0 +1,29 @@
+#pragma once
+
+#include <softbus_edge/config/PhysicalRegistry.h>
+#include <softbus_edge/driver/can/CanChannel.h>
+
+#include <cstdint>
+#include <optional>
+#include <vector>
+
+namespace softbus::edge {
+
+class SdoClient {
+public:
+    bool upload(CanChannelManager* channels,
+                const ResolvedPhysicalPoint& point,
+                std::vector<uint8_t>& outData);
+    bool download(CanChannelManager* channels,
+                  const ResolvedPhysicalPoint& point,
+                  const std::vector<uint8_t>& data);
+
+private:
+    static uint32_t sdoRxCobId(uint8_t nodeId);
+    static uint32_t sdoTxCobId(uint8_t nodeId);
+    std::optional<std::vector<uint8_t>> waitResponse(CanChannelManager* channels,
+                                                     const std::string& channelId,
+                                                     uint32_t expectedCobId);
+};
+
+} // namespace softbus::edge

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);
 };

src/softbus_edge/include/softbus_edge/config/PhysicalRegistry.h

@@ -5,6 +5,7 @@
 #include <optional>
 #include <string>
 #include <unordered_map>
+#include <unordered_set>
 #include <vector>
 
 namespace softbus::edge {
@@ -26,6 +27,10 @@ public:
     const std::vector<PhysicalDeviceDef>& devices() const { return devices_; }
     const std::vector<PhysicalPointDef>& pollPoints() const { return pollPoints_; }
 
+    std::vector<PhysicalChannelDef> channelsByProtocol(const std::string& protocol) const;
+    std::vector<PhysicalPointDef> pollPointsByProtocol(const std::string& protocol) const;
+    std::unordered_set<std::string> protocols() const;
+
     std::optional<ResolvedPhysicalPoint> resolvePoint(const std::string& physicalPointId) const;
     std::optional<PhysicalChannelDef> findChannel(const std::string& channelId) const;
 

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<PhysicalPointDef> points;
 };
 

src/softbus_edge/include/softbus_edge/driver/can/CanChannel.h

@@ -0,0 +1,37 @@
+#pragma once
+
+#include <softbus_edge/config/PhysicalTypes.h>
+
+#include <cstdint>
+#include <optional>
+#include <string>
+#include <unordered_map>
+#include <vector>
+
+namespace softbus::edge {
+
+struct CanFrame {
+    uint32_t canId{0};
+    std::vector<uint8_t> 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<CanFrame> 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<std::string, ChannelHandle> channels_;
+};
+
+} // namespace softbus::edge

src/softbus_edge/include/softbus_edge/driver/eth/EthChannel.h

@@ -0,0 +1,36 @@
+#pragma once
+
+#include <softbus_edge/config/PhysicalTypes.h>
+
+#include <asio.hpp>
+
+#include <memory>
+#include <optional>
+#include <string>
+#include <unordered_map>
+#include <vector>
+
+namespace softbus::edge {
+
+class EthChannelManager {
+public:
+    EthChannelManager();
+    bool openChannel(const PhysicalChannelDef& channel);
+    void closeChannel(const std::string& channelId);
+    void closeAll();
+    std::vector<std::vector<uint8_t>> recv(const std::string& channelId, int maxPackets = 8);
+    bool send(const std::string& channelId, const std::vector<uint8_t>& data);
+
+private:
+    struct ChannelHandle {
+        PhysicalChannelDef config;
+        std::unique_ptr<asio::ip::tcp::socket> tcpSocket;
+        std::unique_ptr<asio::ip::udp::socket> udpSocket;
+        asio::ip::udp::endpoint udpRemote;
+    };
+
+    asio::io_context io_;
+    std::unordered_map<std::string, ChannelHandle> channels_;
+};
+
+} // namespace softbus::edge

src/softbus_edge/include/softbus_edge/driver/modbus/ModbusChannel.h

@@ -1,11 +1,14 @@
 #pragma once
 
 #include <softbus_edge/config/PhysicalRegistry.h>
+#include <softbus_edge/driver/modbus/TcpRtuSession.h>
 
 #include <cstdint>
+#include <memory>
 #include <optional>
 #include <string>
 #include <unordered_map>
+#include <vector>
 
 namespace softbus::edge {
 
@@ -15,11 +18,15 @@ public:
     void closeChannel(const std::string& channelId);
     void closeAll();
     std::optional<uint16_t> readRegister(const ResolvedPhysicalPoint& point);
+    std::optional<std::vector<uint16_t>> readRegisters(const ResolvedPhysicalPoint& point,
+                                                         int startAddr,
+                                                         int count);
     bool writeRegister(const ResolvedPhysicalPoint& point, uint16_t rawValue);
 
 private:
     struct ChannelHandle {
         PhysicalChannelDef config;
+        std::unique_ptr<TcpRtuSession> tcpRtu;
 #if defined(SOFTBUS_HAS_LIBMODBUS) && SOFTBUS_HAS_LIBMODBUS
         void* ctx{nullptr};
 #endif

src/softbus_edge/include/softbus_edge/driver/modbus/TcpRtuSession.h

@@ -0,0 +1,28 @@
+#pragma once
+
+#include <cstdint>
+#include <optional>
+#include <string>
+#include <vector>
+
+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<std::vector<uint16_t>> 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<std::vector<uint8_t>> transaction(const std::vector<uint8_t>& pdu);
+
+    int socketFd_{-1};
+};
+
+} // namespace softbus::edge

src/softbus_edge/include/softbus_edge/egress/CommandExecutor.h

@@ -2,7 +2,9 @@
 
 #include <softbus_api/ExecuteCommand.h>
 #include <softbus_edge/config/PhysicalRegistry.h>
-#include <softbus_edge/egress/ModbusEgress.h>
+#include <softbus_edge/egress/can/CanEgress.h>
+#include <softbus_edge/egress/eth/EthEgress.h>
+#include <softbus_edge/egress/modbus/ModbusEgress.h>
 #include <softbus_edge/egress/ShadowChannelStore.h>
 #include <softbus_edge/egress/UplinkClient.h>
 #include <softbus_registry/BindingRegistry.h>
@@ -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};

src/softbus_edge/include/softbus_edge/egress/can/CanEgress.h

@@ -0,0 +1,24 @@
+#pragma once
+
+#include <softbus_edge/canopen/SdoClient.h>
+#include <softbus_edge/config/PhysicalRegistry.h>
+#include <softbus_edge/driver/can/CanChannel.h>
+
+#include <string>
+
+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

src/softbus_edge/include/softbus_edge/egress/eth/EthEgress.h

@@ -0,0 +1,23 @@
+#pragma once
+
+#include <softbus_edge/config/PhysicalRegistry.h>
+#include <softbus_edge/driver/eth/EthChannel.h>
+
+#include <string>
+#include <vector>
+
+namespace softbus::edge {
+
+class EthEgress {
+public:
+    explicit EthEgress(PhysicalRegistry* registry, EthChannelManager* channels);
+
+    bool writeByPhysicalPointId(const std::string& physicalPointId,
+                                const std::vector<uint8_t>& data);
+
+private:
+    PhysicalRegistry* registry_{nullptr};
+    EthChannelManager* channels_{nullptr};
+};
+
+} // namespace softbus::edge

src/softbus_edge/include/softbus_edge/egress/modbus/ModbusEgress.h

@@ -1,9 +1,8 @@
 #pragma once
 
 #include <softbus_edge/config/PhysicalRegistry.h>
-#include <softbus_edge/driver/ModbusChannel.h>
+#include <softbus_edge/driver/modbus/ModbusChannel.h>
 
-#include <optional>
 #include <string>
 
 namespace softbus::edge {

src/softbus_edge/include/softbus_edge/factory/DeviceFactory.h

@@ -2,11 +2,10 @@
 
 #include <softbus_edge/config/EdgeConfig.h>
 #include <softbus_edge/config/PhysicalRegistry.h>
-#include <softbus_edge/driver/ModbusChannel.h>
-#include <softbus_edge/egress/ModbusEgress.h>
+#include <softbus_edge/driver/can/CanChannel.h>
+#include <softbus_edge/driver/eth/EthChannel.h>
+#include <softbus_edge/driver/modbus/ModbusChannel.h>
 #include <softbus_edge/ingress/IIngressPort.h>
-#include <softbus_edge/ingress/MockModbusIngress.h>
-#include <softbus_edge/ingress/ModbusIngress.h>
 
 #include <memory>
 #include <string>
@@ -19,7 +18,9 @@ public:
 
     std::unique_ptr<IIngressPort> createIngress(const EdgeConfig& config,
                                                 PhysicalRegistry* registry,
-                                                ModbusChannelManager* channels,
+                                                ModbusChannelManager* modbusChannels,
+                                                CanChannelManager* canChannels,
+                                                EthChannelManager* ethChannels,
                                                 bool useMock,
                                                 const std::string& mockObjectRef);
 };

src/softbus_edge/include/softbus_edge/ingress/CompositeIngress.h

@@ -0,0 +1,22 @@
+#pragma once
+
+#include <softbus_edge/ingress/IIngressPort.h>
+
+#include <memory>
+#include <vector>
+
+namespace softbus::edge {
+
+class CompositeIngress : public IIngressPort {
+public:
+    explicit CompositeIngress(std::vector<std::unique_ptr<IIngressPort>> children);
+
+    bool open() override;
+    void close() override;
+    std::vector<softbus::core::RawPacket> poll() override;
+
+private:
+    std::vector<std::unique_ptr<IIngressPort>> children_;
+};
+
+} // namespace softbus::edge

src/softbus_edge/include/softbus_edge/ingress/can/CanIngress.h

@@ -0,0 +1,48 @@
+#pragma once
+
+#include <canopen/CanopenObjectDictionary.h>
+
+#include <softbus_edge/canopen/NmtManager.h>
+#include <softbus_edge/canopen/PdoMapper.h>
+#include <softbus_edge/canopen/SdoClient.h>
+#include <softbus_edge/config/PhysicalRegistry.h>
+#include <softbus_edge/driver/can/CanChannel.h>
+#include <softbus_edge/ingress/IIngressPort.h>
+#include <softbus_edge/scheduler/ChangeFilter.h>
+#include <softbus_edge/scheduler/PollPlanner.h>
+
+#include <string>
+#include <unordered_map>
+
+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<softbus::core::RawPacket> poll() override;
+
+private:
+    double decodeValue(const MappedCanPoint& mapped, const std::vector<uint8_t>& data) const;
+    std::vector<uint8_t> buildCanopenPayload(uint32_t cobId, const std::vector<uint8_t>& 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<std::string, canopen::CanopenObjectDictionary> dictionaries_;
+};
+
+} // namespace softbus::edge

src/softbus_edge/include/softbus_edge/ingress/eth/EthIngress.h

@@ -0,0 +1,26 @@
+#pragma once
+
+#include <softbus_edge/config/PhysicalRegistry.h>
+#include <softbus_edge/driver/eth/EthChannel.h>
+#include <softbus_edge/ingress/IIngressPort.h>
+
+#include <string>
+
+namespace softbus::edge {
+
+class EthIngress : public IIngressPort {
+public:
+    EthIngress(PhysicalRegistry* registry, EthChannelManager* channels, std::string edgeIdPrefix);
+
+    bool open() override;
+    void close() override;
+    std::vector<softbus::core::RawPacket> poll() override;
+
+private:
+    PhysicalRegistry* registry_{nullptr};
+    EthChannelManager* channels_{nullptr};
+    std::string edgeIdPrefix_;
+    bool open_{false};
+};
+
+} // namespace softbus::edge

src/softbus_edge/include/softbus_edge/ingress/modbus/ModbusIngress.h

@@ -1,10 +1,12 @@
 #pragma once
 
 #include <softbus_edge/config/PhysicalRegistry.h>
-#include <softbus_edge/driver/ModbusChannel.h>
+#include <softbus_edge/driver/modbus/ModbusChannel.h>
 #include <softbus_edge/ingress/IIngressPort.h>
+#include <softbus_edge/scheduler/ChangeFilter.h>
+#include <softbus_edge/scheduler/ModbusPollScheduler.h>
+#include <softbus_edge/scheduler/PollPlanner.h>
 
-#include <chrono>
 #include <string>
 
 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

src/softbus_edge/include/softbus_edge/runtime/EdgeRuntime.h

@@ -2,9 +2,13 @@
 
 #include <softbus_edge/config/EdgeConfig.h>
 #include <softbus_edge/config/PhysicalRegistry.h>
-#include <softbus_edge/driver/ModbusChannel.h>
+#include <softbus_edge/driver/can/CanChannel.h>
+#include <softbus_edge/driver/eth/EthChannel.h>
+#include <softbus_edge/driver/modbus/ModbusChannel.h>
+#include <softbus_edge/egress/can/CanEgress.h>
+#include <softbus_edge/egress/eth/EthEgress.h>
+#include <softbus_edge/egress/modbus/ModbusEgress.h>
 #include <softbus_edge/egress/CommandExecutor.h>
-#include <softbus_edge/egress/ModbusEgress.h>
 #include <softbus_edge/egress/ShadowChannelStore.h>
 #include <softbus_edge/egress/UplinkClient.h>
 #include <softbus_edge/factory/DeviceFactory.h>
@@ -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_, &ethChannels_};
     ShadowChannelStore mockStore_;
     std::unique_ptr<UplinkClient> uplink_;
     std::unique_ptr<IIngressPort> ingress_;
@@ -47,5 +57,4 @@ private:
     std::chrono::steady_clock::time_point lastUplinkReconnectAttempt_{};
 };
 
-
 } // namespace softbus::edge

src/softbus_edge/include/softbus_edge/scheduler/ChangeFilter.h

@@ -0,0 +1,21 @@
+#pragma once
+
+#include <softbus_edge/config/PhysicalTypes.h>
+
+#include <string>
+#include <unordered_map>
+
+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<std::string, double> lastValues_;
+    std::unordered_map<std::string, bool> hasSample_;
+};
+
+} // namespace softbus::edge

src/softbus_edge/include/softbus_edge/scheduler/ModbusPollScheduler.h

@@ -0,0 +1,31 @@
+#pragma once
+
+#include <softbus_edge/config/PhysicalRegistry.h>
+
+#include <string>
+#include <vector>
+
+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<PollBatchPoint> points;
+};
+
+class ModbusPollScheduler {
+public:
+    static int modbusAddress(int registerAddress);
+    std::vector<PollBatch> planBatches(
+        const std::vector<ResolvedPhysicalPoint>& duePoints) const;
+};
+
+} // namespace softbus::edge

src/softbus_edge/include/softbus_edge/scheduler/PollPlanner.h

@@ -0,0 +1,27 @@
+#pragma once
+
+#include <softbus_edge/config/PhysicalRegistry.h>
+
+#include <chrono>
+#include <string>
+#include <unordered_map>
+#include <vector>
+
+namespace softbus::edge {
+
+class PollPlanner {
+public:
+    void reset(const PhysicalRegistry& registry, const std::string& protocol);
+    std::vector<std::string> duePointIds();
+    void markPolled(const std::string& physicalPointId);
+
+private:
+    struct PointSchedule {
+        int intervalMs{500};
+        std::chrono::steady_clock::time_point nextDue{};
+    };
+
+    std::unordered_map<std::string, PointSchedule> schedules_;
+};
+
+} // namespace softbus::edge

src/softbus_edge/src/CommandExecutor.cpp

@@ -3,10 +3,14 @@
 #include <softbus_core/identity/DeviceIdentity.h>
 #include <softbus_core/logging/Logger.h>
 
+#include <cmath>
+
 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<int>(std::lround(value));
+        std::vector<uint8_t> payload{
+            static_cast<uint8_t>(rawValue & 0xFF),
+            static_cast<uint8_t>((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;
     }
 

src/softbus_edge/src/DeviceFactory.cpp

@@ -1,5 +1,11 @@
 #include <softbus_edge/factory/DeviceFactory.h>
 
+#include <softbus_edge/ingress/CompositeIngress.h>
+#include <softbus_edge/ingress/can/CanIngress.h>
+#include <softbus_edge/ingress/eth/EthIngress.h>
+#include <softbus_edge/ingress/modbus/MockModbusIngress.h>
+#include <softbus_edge/ingress/modbus/ModbusIngress.h>
+
 #include <softbus_core/identity/DeviceIdentity.h>
 
 namespace softbus::edge {
@@ -11,14 +17,41 @@ std::string DeviceFactory::createRuntimeDeviceId()
 
 std::unique_ptr<IIngressPort> DeviceFactory::createIngress(const EdgeConfig& config,
                                                            PhysicalRegistry* registry,
-                                                         ModbusChannelManager* channels,
+                                                           ModbusChannelManager* modbusChannels,
+                                                           CanChannelManager* canChannels,
+                                                           EthChannelManager* ethChannels,
                                                            bool useMock,
                                                            const std::string& mockObjectRef)
 {
     if (useMock) {
         return std::make_unique<MockModbusIngress>(mockObjectRef, config.edgeId);
     }
-    return std::make_unique<ModbusIngress>(registry, channels, config.edgeId);
+
+    std::vector<std::unique_ptr<IIngressPort>> children;
+    const auto protocols = registry->protocols();
+
+    if (protocols.count("modbus") != 0) {
+        children.push_back(
+            std::make_unique<ModbusIngress>(registry, modbusChannels, config.edgeId));
+    }
+    if (protocols.count("canopen") != 0) {
+        children.push_back(std::make_unique<CanIngress>(
+            registry, canChannels, config.edgeId, config.configBaseDir));
+    }
+    if (protocols.count("raw") != 0) {
+        children.push_back(
+            std::make_unique<EthIngress>(registry, ethChannels, config.edgeId));
+    }
+
+    if (children.empty()) {
+        children.push_back(
+            std::make_unique<ModbusIngress>(registry, modbusChannels, config.edgeId));
+    }
+
+    if (children.size() == 1) {
+        return std::move(children.front());
+    }
+    return std::make_unique<CompositeIngress>(std::move(children));
 }
 
 } // namespace softbus::edge

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);

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_, &ethChannels_,
+        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<UplinkClient>(config.daemonHost, config.daemonPort);
+    // 创建命令执行器 用于执行下行命令
     executor_ = std::make_unique<CommandExecutor>(
-        &physicalRegistry_, &modbusEgress_, &mockStore_, &localBindings_,
+        &physicalRegistry_, &modbusEgress_, &canEgress_, &ethEgress_, &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();
     }

src/softbus_edge/src/ModbusIngress.cpp

@@ -1,92 +0,0 @@
-#include <softbus_edge/ingress/ModbusIngress.h>
-
-#include <softbus_core/identity/DeviceIdentity.h>
-#include <softbus_core/time/HwClock.h>
-
-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<uint8_t> ModbusIngress::buildModbusFrame(int slaveId, uint16_t registerValue) const
-{
-    return {
-        static_cast<uint8_t>(slaveId),
-        0x03,
-        0x02,
-        static_cast<uint8_t>((registerValue >> 8) & 0xFF),
-        static_cast<uint8_t>(registerValue & 0xFF),
-        0x00,
-        0x00
-    };
-}
-
-std::vector<softbus::core::RawPacket> 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<softbus::core::RawPacket> 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

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<uint16_t>(pt.value("objectIndex", 0));
+            point.subIndex = static_cast<uint8_t>(pt.value("subIndex", 0));
+            point.pdoType = pt.value("pdoType", "tpdo");
+            point.cobId = static_cast<uint32_t>(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<PhysicalChannelDef> PhysicalRegistry::channelsByProtocol(const std::string& protocol) const
+{
+    std::vector<PhysicalChannelDef> out;
+    for (const auto& ch : channels_) {
+        if (ch.protocol == protocol) {
+            out.push_back(ch);
+        }
+    }
+    return out;
+}
+
+std::vector<PhysicalPointDef> PhysicalRegistry::pollPointsByProtocol(const std::string& protocol) const
+{
+    std::vector<PhysicalPointDef> 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<std::string> PhysicalRegistry::protocols() const
+{
+    std::unordered_set<std::string> out;
+    for (const auto& ch : channels_) {
+        out.insert(ch.protocol);
+    }
+    return out;
+}
+
 std::optional<ResolvedPhysicalPoint> PhysicalRegistry::resolvePoint(
     const std::string& physicalPointId) const
 {

src/softbus_edge/src/canopen/NmtManager.cpp

@@ -0,0 +1,33 @@
+#include <softbus_edge/canopen/NmtManager.h>
+
+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

src/softbus_edge/src/canopen/PdoMapper.cpp

@@ -0,0 +1,83 @@
+#include <softbus_edge/canopen/PdoMapper.h>
+
+namespace softbus::edge {
+
+void PdoMapper::rebuild(
+    const PhysicalRegistry& registry,
+    const std::unordered_map<std::string, canopen::CanopenObjectDictionary>& 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<uint8_t>(device.nodeId);
+
+        if (dictionary) {
+            for (const auto& tpdo : dictionary->tpdoDefinitions()) {
+                if (tpdo.cobId != 0) {
+                    sub.cobId = tpdo.cobId + static_cast<uint32_t>(device.nodeId);
+                    break;
+                }
+            }
+        }
+        if (sub.cobId == 0) {
+            sub.cobId = 0x180u + static_cast<uint32_t>(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<MappedCanPoint> PdoMapper::matchFrame(uint32_t cobId,
+                                                  const std::vector<uint8_t>& 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<MappedCanPoint> matched;
+    for (const auto& point : sub.points) {
+        if (point.byteOffset + 1 < static_cast<int>(data.size())) {
+            matched.push_back(point);
+        }
+    }
+    return matched;
+}
+
+} // namespace softbus::edge

src/softbus_edge/src/canopen/SdoClient.cpp

@@ -0,0 +1,112 @@
+#include <softbus_edge/canopen/SdoClient.h>
+
+#include <chrono>
+#include <thread>
+
+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<std::vector<uint8_t>> 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<uint8_t>& outData)
+{
+    if (!channels || point.device.nodeId == 0) {
+        return false;
+    }
+
+    const uint8_t nodeId = static_cast<uint8_t>(point.device.nodeId);
+    CanFrame request;
+    request.canId = sdoRxCobId(nodeId);
+    request.data = {
+        0x40,
+        static_cast<uint8_t>(point.point.objectIndex & 0xFF),
+        static_cast<uint8_t>((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<uint8_t>& data)
+{
+    if (!channels || point.device.nodeId == 0 || data.empty()) {
+        return false;
+    }
+
+    const uint8_t nodeId = static_cast<uint8_t>(point.device.nodeId);
+    CanFrame request;
+    request.canId = sdoRxCobId(nodeId);
+    request.data.resize(8, 0x00);
+    request.data[1] = static_cast<uint8_t>(point.point.objectIndex & 0xFF);
+    request.data[2] = static_cast<uint8_t>((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

src/softbus_edge/src/driver/can/CanChannel.cpp

@@ -0,0 +1,133 @@
+#include <softbus_edge/driver/can/CanChannel.h>
+
+#include <softbus_core/logging/Logger.h>
+
+#if defined(SOFTBUS_HAS_SOCKETCAN) && SOFTBUS_HAS_SOCKETCAN
+#include <linux/can.h>
+#include <linux/can/raw.h>
+#include <net/if.h>
+#include <sys/ioctl.h>
+#include <sys/socket.h>
+#include <unistd.h>
+#endif
+
+#include <cstring>
+
+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<struct sockaddr*>(&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<std::size_t>(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<ssize_t>(sizeof(raw));
+#else
+    (void)frame;
+    return true;
+#endif
+}
+
+std::vector<CanFrame> CanChannelManager::recvFrames(const std::string& channelId, int maxFrames)
+{
+    std::vector<CanFrame> 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<ssize_t>(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

src/softbus_edge/src/driver/eth/EthChannel.cpp

@@ -0,0 +1,101 @@
+#include <softbus_edge/driver/eth/EthChannel.h>
+
+#include <softbus_core/logging/Logger.h>
+
+#include <array>
+
+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<asio::ip::udp::socket>(io_);
+            handle.udpSocket->open(endpoint.protocol());
+            if (channel.host.empty()) {
+                handle.udpSocket->bind(endpoint);
+            } else {
+                handle.udpRemote = endpoint;
+            }
+        } else {
+            handle.tcpSocket = std::make_unique<asio::ip::tcp::socket>(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<std::vector<uint8_t>> EthChannelManager::recv(const std::string& channelId,
+                                                          int maxPackets)
+{
+    std::vector<std::vector<uint8_t>> packets;
+    const auto it = channels_.find(channelId);
+    if (it == channels_.end()) {
+        return packets;
+    }
+
+    for (int i = 0; i < maxPackets; ++i) {
+        std::array<uint8_t, 1500> 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<std::ptrdiff_t>(n));
+    }
+    return packets;
+}
+
+bool EthChannelManager::send(const std::string& channelId, const std::vector<uint8_t>& 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

src/softbus_edge/src/driver/modbus/ModbusChannel.cpp

@@ -1,4 +1,4 @@
-#include <softbus_edge/driver/ModbusChannel.h>
+#include <softbus_edge/driver/modbus/ModbusChannel.h>
 
 #include <softbus_core/logging/Logger.h>
 
@@ -6,10 +6,19 @@
 #include <modbus.h>
 #endif
 
-#include <cmath>
+#include <vector>
 
 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<TcpRtuSession>();
+        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<modbus_t*>(handle.ctx), 0, 500000);
+    if (channel.transport == "tcp") {
+        modbus_enable_quirks(static_cast<modbus_t*>(handle.ctx), MODBUS_QUIRK_MAX_SLAVE);
+    }
     if (modbus_connect(static_cast<modbus_t*>(handle.ctx)) != 0) {
         SB_LOG_ERROR("ModbusChannel", "Failed to connect channel ", channel.channelId);
         modbus_free(static_cast<modbus_t*>(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<modbus_t*>(it->second.ctx));
@@ -81,37 +111,69 @@ void ModbusChannelManager::closeAll()
 
 std::optional<uint16_t> 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<std::vector<uint16_t>> 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<modbus_t*>(it->second.ctx), point.device.slaveId);
-    uint16_t value = 0;
-    const int addr = modbusAddress(point.point.registerAddress);
+    std::vector<uint16_t> values(static_cast<std::size_t>(count));
     const int rc = modbus_read_registers(
-        static_cast<modbus_t*>(it->second.ctx), addr, 1, &value);
-    if (rc != 1) {
+        static_cast<modbus_t*>(it->second.ctx), startAddr, count, values.data());
+    if (rc != count) {
         return std::nullopt;
     }
-    return value;
+    return values;
 #else
     static uint16_t sim = 250;
+    std::vector<uint16_t> values(static_cast<std::size_t>(count));
+    for (int i = 0; i < count; ++i) {
         sim = static_cast<uint16_t>((sim + 1) % 500);
+        values[static_cast<std::size_t>(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<modbus_t*>(it->second.ctx), point.device.slaveId);
-    const int addr = modbusAddress(point.point.registerAddress);
     return modbus_write_register(static_cast<modbus_t*>(it->second.ctx), addr, rawValue) == 1;
 #else
     (void)point;

src/softbus_edge/src/driver/modbus/TcpRtuSession.cpp

@@ -0,0 +1,159 @@
+#include <softbus_edge/driver/modbus/TcpRtuSession.h>
+
+#include <arpa/inet.h>
+#include <netinet/in.h>
+#include <netinet/tcp.h>
+#include <sys/select.h>
+#include <sys/socket.h>
+#include <unistd.h>
+
+#include <cstring>
+#include <vector>
+
+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<uint16_t>((crc >> 1) ^ 0xA001);
+            } else {
+                crc = static_cast<uint16_t>(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<sockaddr*>(&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<std::vector<uint8_t>> TcpRtuSession::transaction(const std::vector<uint8_t>& pdu)
+{
+    if (socketFd_ < 0 || pdu.empty()) {
+        return std::nullopt;
+    }
+
+    const uint16_t crc = crc16(pdu.data(), pdu.size());
+    std::vector<uint8_t> frame = pdu;
+    frame.push_back(static_cast<uint8_t>(crc & 0xFF));
+    frame.push_back(static_cast<uint8_t>((crc >> 8) & 0xFF));
+
+    if (send(socketFd_, frame.data(), frame.size(), 0) != static_cast<ssize_t>(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<uint8_t>(buffer, buffer + n);
+}
+
+std::optional<std::vector<uint16_t>> TcpRtuSession::readHoldingRegisters(int slaveId,
+                                                                         int startAddr,
+                                                                         int count)
+{
+    if (count <= 0 || count > 125) {
+        return std::nullopt;
+    }
+
+    const auto slave = static_cast<uint8_t>(slaveId);
+    std::vector<uint8_t> pdu{
+        slave,
+        0x03,
+        static_cast<uint8_t>((startAddr >> 8) & 0xFF),
+        static_cast<uint8_t>(startAddr & 0xFF),
+        static_cast<uint8_t>((count >> 8) & 0xFF),
+        static_cast<uint8_t>(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<std::size_t>(3 + byteCount + 2) > response->size()) {
+        return std::nullopt;
+    }
+
+    std::vector<uint16_t> values(static_cast<std::size_t>(count));
+    for (int i = 0; i < count; ++i) {
+        const std::size_t off = static_cast<std::size_t>(3 + i * 2);
+        values[static_cast<std::size_t>(i)] = static_cast<uint16_t>(
+            (static_cast<uint16_t>((*response)[off]) << 8) | (*response)[off + 1]);
+    }
+    return values;
+}
+
+bool TcpRtuSession::writeSingleRegister(int slaveId, int address, uint16_t value)
+{
+    const auto slave = static_cast<uint8_t>(slaveId);
+    std::vector<uint8_t> pdu{
+        slave,
+        0x06,
+        static_cast<uint8_t>((address >> 8) & 0xFF),
+        static_cast<uint8_t>(address & 0xFF),
+        static_cast<uint8_t>((value >> 8) & 0xFF),
+        static_cast<uint8_t>(value & 0xFF)
+    };
+
+    const auto response = transaction(pdu);
+    if (!response || response->size() < 8) {
+        return false;
+    }
+    return (*response)[0] == slave && (*response)[1] == 0x06;
+}
+
+} // namespace softbus::edge

src/softbus_edge/src/egress/can/CanEgress.cpp

@@ -0,0 +1,38 @@
+#include <softbus_edge/egress/can/CanEgress.h>
+
+#include <cmath>
+
+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<int>(std::lround(engineeringValue))
+        : static_cast<int>(std::lround(engineeringValue / resolved->point.scale));
+    if (rawOut) {
+        *rawOut = raw;
+    }
+
+    std::vector<uint8_t> data;
+    data.push_back(static_cast<uint8_t>(raw & 0xFF));
+    data.push_back(static_cast<uint8_t>((raw >> 8) & 0xFF));
+    return sdo_.download(channels_, *resolved, data);
+}
+
+} // namespace softbus::edge

src/softbus_edge/src/egress/eth/EthEgress.cpp

@@ -0,0 +1,24 @@
+#include <softbus_edge/egress/eth/EthEgress.h>
+
+namespace softbus::edge {
+
+EthEgress::EthEgress(PhysicalRegistry* registry, EthChannelManager* channels)
+    : registry_(registry)
+    , channels_(channels)
+{
+}
+
+bool EthEgress::writeByPhysicalPointId(const std::string& physicalPointId,
+                                       const std::vector<uint8_t>& 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

src/softbus_edge/src/egress/modbus/ModbusEgress.cpp

@@ -1,4 +1,4 @@
-#include <softbus_edge/egress/ModbusEgress.h>
+#include <softbus_edge/egress/modbus/ModbusEgress.h>
 
 #include <cmath>
 

src/softbus_edge/src/ingress/CompositeIngress.cpp

@@ -0,0 +1,38 @@
+#include <softbus_edge/ingress/CompositeIngress.h>
+
+namespace softbus::edge {
+
+CompositeIngress::CompositeIngress(std::vector<std::unique_ptr<IIngressPort>> 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<softbus::core::RawPacket> CompositeIngress::poll()
+{
+    std::vector<softbus::core::RawPacket> 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

src/softbus_edge/src/ingress/can/CanIngress.cpp

@@ -0,0 +1,165 @@
+#include <softbus_edge/ingress/can/CanIngress.h>
+
+#include <softbus_core/identity/DeviceIdentity.h>
+#include <softbus_core/logging/Logger.h>
+#include <softbus_core/time/HwClock.h>
+
+#include <filesystem>
+
+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<uint8_t>(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<uint8_t>& data) const
+{
+    if (mapped.byteOffset + 1 >= static_cast<int>(data.size())) {
+        return 0.0;
+    }
+    const uint16_t raw = static_cast<uint16_t>(data[static_cast<std::size_t>(mapped.byteOffset)]) |
+                         (static_cast<uint16_t>(data[static_cast<std::size_t>(mapped.byteOffset + 1)]) << 8);
+    return static_cast<double>(raw) * mapped.scale;
+}
+
+std::vector<uint8_t> CanIngress::buildCanopenPayload(uint32_t cobId,
+                                                     const std::vector<uint8_t>& data) const
+{
+    std::vector<uint8_t> payload;
+    payload.push_back(static_cast<uint8_t>(cobId & 0xFF));
+    payload.push_back(static_cast<uint8_t>((cobId >> 8) & 0xFF));
+    payload.push_back(static_cast<uint8_t>(data.size()));
+    payload.insert(payload.end(), data.begin(), data.end());
+    return payload;
+}
+
+std::vector<softbus::core::RawPacket> CanIngress::poll()
+{
+    if (!open_ || !registry_ || !channels_) {
+        return {};
+    }
+
+    std::vector<softbus::core::RawPacket> 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<uint8_t> data;
+        if (!sdo_.upload(channels_, *resolved, data)) {
+            continue;
+        }
+        uint16_t raw = data.size() >= 2
+            ? static_cast<uint16_t>(data[0] | (data[1] << 8))
+            : (data.empty() ? 0 : data[0]);
+        const double engineering = static_cast<double>(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

src/softbus_edge/src/ingress/eth/EthIngress.cpp

@@ -0,0 +1,62 @@
+#include <softbus_edge/ingress/eth/EthIngress.h>
+
+#include <softbus_core/identity/DeviceIdentity.h>
+#include <softbus_core/time/HwClock.h>
+
+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<softbus::core::RawPacket> EthIngress::poll()
+{
+    if (!open_ || !registry_ || !channels_) {
+        return {};
+    }
+
+    std::vector<softbus::core::RawPacket> 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

src/softbus_edge/src/ingress/modbus/MockModbusIngress.cpp

@@ -1,4 +1,4 @@
-#include <softbus_edge/ingress/MockModbusIngress.h>
+#include <softbus_edge/ingress/modbus/MockModbusIngress.h>
 
 #include <softbus_core/identity/DeviceIdentity.h>
 #include <softbus_core/time/HwClock.h>
@@ -24,7 +24,6 @@ void MockModbusIngress::close()
 
 std::vector<uint8_t> 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<uint8_t>((registerValue >> 8) & 0xFF),
@@ -39,9 +38,8 @@ std::vector<softbus::core::RawPacket> 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_;

src/softbus_edge/src/ingress/modbus/ModbusIngress.cpp

@@ -0,0 +1,133 @@
+#include <softbus_edge/ingress/modbus/ModbusIngress.h>
+
+#include <softbus_core/identity/DeviceIdentity.h>
+#include <softbus_core/logging/Logger.h>
+#include <softbus_core/time/HwClock.h>
+
+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<uint8_t> ModbusIngress::buildModbusFrame(int slaveId, uint16_t registerValue) const
+{
+    return {
+        static_cast<uint8_t>(slaveId),
+        0x03,
+        0x02,
+        static_cast<uint8_t>((registerValue >> 8) & 0xFF),
+        static_cast<uint8_t>(registerValue & 0xFF),
+        0x00,
+        0x00
+    };
+}
+
+std::vector<softbus::core::RawPacket> ModbusIngress::poll()
+{
+    if (!open_ || !registry_ || !channels_) {
+        return {};
+    }
+
+    const auto dueIds = planner_.duePointIds();
+    if (dueIds.empty()) {
+        return {};
+    }
+
+    std::vector<ResolvedPhysicalPoint> 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<softbus::core::RawPacket> 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<std::size_t>(batchPoint.offsetInBatch) >= values->size()) {
+                continue;
+            }
+
+            const uint16_t raw = (*values)[static_cast<std::size_t>(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

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 <softbus_edge/config/EdgeConfig.h>
 #include <softbus_edge/runtime/EdgeRuntime.h>
 

src/softbus_edge/src/scheduler/ChangeFilter.cpp

@@ -0,0 +1,35 @@
+#include <softbus_edge/scheduler/ChangeFilter.h>
+
+#include <cmath>
+
+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

src/softbus_edge/src/scheduler/ModbusPollScheduler.cpp

@@ -0,0 +1,103 @@
+#include <softbus_edge/scheduler/ModbusPollScheduler.h>
+
+#include <algorithm>
+#include <map>
+#include <tuple>
+
+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<PollBatch> ModbusPollScheduler::planBatches(
+    const std::vector<ResolvedPhysicalPoint>& duePoints) const
+{
+    std::map<std::tuple<std::string, int, int>, std::vector<SortablePoint>> 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<PollBatch> 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

src/softbus_edge/src/scheduler/PollPlanner.cpp

@@ -0,0 +1,44 @@
+#include <softbus_edge/scheduler/PollPlanner.h>
+
+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<std::string> PollPlanner::duePointIds()
+{
+    std::vector<std::string> 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

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<asio::ip::tcp::socket>(io_);
     std::error_code ec;
     socket_->connect(asio::ip::tcp::endpoint(asio::ip::make_address(host_), port_), ec);