diff --git a/.claude/settings.local.json b/.claude/settings.local.json index f1bfdf4a..816d380a 100644 --- a/.claude/settings.local.json +++ b/.claude/settings.local.json @@ -16,7 +16,16 @@ "Bash(mvn compile:*)", "Bash(mvn test:*)", "Bash(mysql -hlocalhost -uroot -pfastbee fastbee -e \"SELECT id, serial_number, platform_id, device_type, enable FROM platform_device_mapping LIMIT 5;\")", - "Bash(mvn test-compile:*)" + "Bash(mvn test-compile:*)", + "Bash(mvn clean:*)", + "Bash(ls:*)", + "mcp__ide__getDiagnostics", + "Bash(mvn:*)", + "Bash(npx eslint:*)", + "Bash(git diff:*)", + "Bash(source \"$HOME/.nvm/nvm.sh\")", + "Bash(nvm use:*)", + "Bash(npm run:*)" ] } } diff --git a/bin/send_modbus_and_register.py b/bin/send_modbus_and_register.py index 9f211ec8..6e397d95 100644 --- a/bin/send_modbus_and_register.py +++ b/bin/send_modbus_and_register.py @@ -8,12 +8,11 @@ TARGET_PORT = 8888 # 请根据实际情况修改 # 元数组 modbus 指令 MODBUS_DATA_HEX = ( -"7e80D21F6C448E8D7e", - "01031c41b6dcec418b0000402d2148000000004017e0b8400c19efc2ad8a8acb2b", - "02031c41babecc419d8000402be3560000000040137c30bfb2f4d0c2ae9bc1b582", - "03031c41bd1c1941810000402bb38600000000bf4501bdc027dfcac2ae8853eaef", - "04031c41babecc419d8000402be3560000000040137c30bfb2f4d0c2ae9bc1D581", - "05031c41bd1c1941810000402bb38600000000bf4501bdc027dfcac2ae88538AEC" + "7e80DD33D43DFC5B7e", + "010320413d0d7541bacb2dbf42019a3f6bdb33c2b19c73c0d39f304100a253c3f9f2187f3a", + "020320414056d541b8ad60c02d4acd3fa708cdc2adfe5ac1bcf64c413630c4c3f9a81af11d", + "030320413fc48241b919bcc0006ecd3ff874cdc2ae6c0ac18c11e041877c67c3f9b431a049", + "040320413f56c341b0362ebf48b333403e3200c2addb5cc0daec4341cf60bdc3f9a40fae0b" ) # 深部位移计 diff --git a/docs/knowledge-base/06-platform-integration-design.md b/docs/knowledge-base/06-platform-integration-design.md index e957971c..78dec70d 100644 --- a/docs/knowledge-base/06-platform-integration-design.md +++ b/docs/knowledge-base/06-platform-integration-design.md @@ -26,7 +26,7 @@ FastBee 需要将设备上报数据推送到第三方监测平台，当前已落 --- -## 2. 三种执行模式 +## 2. 四种执行模式 ### 2.1 流程编排 @@ -59,7 +59,25 @@ FastBee 需要将设备上报数据推送到第三方监测平台，当前已落 实时推送不经过 `PlatformFlowExecutor`，因此不会使用流程步骤的 `param_mapping`、`result_key`、`result_save`。 -### 2.3 Token 自动刷新 +### 2.3 定时推送 + +在实时推送基础上，新增按时间间隔推送最新数据的模式。通过 `platform_device_mapping.push_mode` 字段控制： + +| push_mode | 模式 | 行为 | +|---|---|---| +| `1`（默认） | 实时推送 | 设备上报后立即推送，沿用原有链路 | +| `2` | 定时推送 | 仅由调度器按 `push_interval`（分钟）推送，`DeviceReportTrigger` 跳过实时推送 | + +调度器 `PlatformPushScheduledJob` 使用 Spring `@Scheduled(fixedDelay = 300_000)` 每 5 分钟检查一次： + +1. 查询所有 `enable='1'`、`push_mode=2`、`push_interval > 0` 的绑定。 +2. 逐条判断 `now - last_push_time >= push_interval * 60 * 1000`；从未推送过的立即触发。 +3. 从 Redis 缓存读取最新数据，按直连/网关分别调用 `DcpPlatformAdapter` 推送。 +4. 推送成功后更新 `last_push_time`；失败时保留不变，下次继续尝试。 + +定时推送不经过 `PlatformFlowExecutor`，推送数据来自 Redis 缓存而非实时 payload，因此依赖设备数据及时更新到 Redis。 + +### 2.4 Token 自动刷新 流程执行器和 DCP 实时推送适配器各自调用 `ensureToken`： @@ -125,13 +143,15 @@ DDL 以 `sql/platform-integration.sql` 为准。 | `ts` | 从流程上下文 payload 取时间戳 | | `onboard_config` | 入网参数，直接取接口 `request_schema` 或步骤覆盖配置中的 `value`，不再读取子映射表 `onboard_config` | | `subDeviceSensorIds` | 收集子设备 sensorId 列表 | +| `cascadedSubDeviceSensorIds` | 串联子设备入网：从当前 monitor 到末端 monitor 收集 sensorId，并按末端到当前倒序返回 | +| `cascadedSubDeviceDepths` | 串联子设备入网：从配置的顺序数组中截取当前 monitor 到末端 monitor，并按末端到当前倒序返回 | | `referenceSubDeviceSensorIdPairs` | 收集 `[基准点sensorId, 监测点sensorId]` 配对列表，可能展开为多次请求 | `onboard_config` 是设备入网专用的可配置入参。接口管理页或流程步骤覆盖页直接维护 `value`： ```json {"name":"sensorDepths","type":"array","source":"onboard_config","value":[1.7,3.4,5.1,6.8]} -{"name":"holeDepth","type":"number","source":"onboard_config","value":[1.7,3.4,5.1,6.8]} +{"name":"holeDepth","type":"number","source":"onboard_config","value":6.8} ``` 网关多 monitor 入网时，`PlatformDeviceMappingServiceImpl` 按 `PlatformDeviceSubMappingMapper.selectByParentId` 返回顺序筛选 monitor，并把当前下标写入 `FlowContext.currentMonitorIndex`。流程执行器解析 `onboard_config` 时： @@ -199,6 +219,7 @@ DDL 以 `sql/platform-integration.sql` 为准。 |---|---| | `platform_id` | 所属平台 | | `device_id` / `serial_number` | FastBee 设备 ID 与序列号 | +| `device_name` | 推送给平台的设备名称，默认取平台设备名称，可手动修改 | | `platform_device_id` | 平台侧 deviceId，直连设备入网后回填 | | `platform_point_id` | 平台侧 pointId，直连设备入网后回填 | | `sensor_type` | 传感器类型，来自 `platform_sensor_type` 字典；当前包含 `inclino`、`deep_inclino`、`stress`、`hyeto`、`static_level` | @@ -208,6 +229,9 @@ DDL 以 `sql/platform-integration.sql` 为准。 | `extra_mapping` | 数组字段、静力水准、上报间隔等扩展配置 | | `net_status` | `0=未入网`、`1=已入网` | | `enable` | 是否启用 | +| `push_mode` | 推送模式：`1=实时`、`2=定时`，默认 `1` | +| `push_interval` | 定时推送间隔（分钟），`0` 表示不自动推送 | +| `last_push_time` | 最近推送时间，由定时调度器更新 | ### 3.6 `platform_device_sub_mapping` 子设备映射表 @@ -219,6 +243,7 @@ DDL 以 `sql/platform-integration.sql` 为准。 |---|---| | `parent_mapping_id` | 主绑定 ID | | `sub_serial_number` | 子设备序列号 | +| `sub_device_name` | 子设备名称 | | `device_role` | `reference=基准点`、`monitor=监测点` | | `platform_device_id` | 平台分配给该监测点的 deviceId | | `platform_point_id` | 平台分配给该监测点的 pointId | @@ -284,8 +309,10 @@ fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/ │ └── dcp/ │ ├── DcpPlatformAdapter │ └── StaticLevelBufferService -└── trigger/ - └── DeviceReportTrigger +├── trigger/ +│ └── DeviceReportTrigger +└── job/ + └── PlatformPushScheduledJob ``` ```text @@ -354,6 +381,11 @@ execute(flow, context) `DeviceReportTrigger.onDeviceReport` 使用异步线程池 `FastBeeConstant.TASK.PUSH_TO_THIRD_PLATFORM_ASYNC`。 +遍历匹配的 `PlatformDeviceMapping` 时，先检查 `push_mode`： + +- `push_mode=1` 或未配置：执行实时推送（原有行为）。 +- `push_mode=2`：跳过实时推送，仅由 `PlatformPushScheduledJob` 定时推送。 + 匹配逻辑在 `PlatformDeviceMappingMapper.xml`： ```sql @@ -371,7 +403,53 @@ WHERE s.sub_serial_number = #{serialNumber} AND m.enable = '1' - `platform.code=dcp` 且 `device_type=1`：先在子表中按上报序列号找 `PlatformDeviceSubMapping`，找到后调用 `pushGatewayData`。 - 未知平台 code 只记录 warning，不推送。 -### 5.4 DCP 普通推送 +### 5.4 定时推送调度器 + +`PlatformPushScheduledJob` 位于 `com.fastbee.iot.platform.job`，核心逻辑： + +```java +@Scheduled(fixedDelay = 300_000) +public void execute() { + // 1. 查询所有启用的定时推送配置（pushMode=2） + List mappings = mappingService.list( + Wrappers.lambdaQuery() + .eq(PlatformDeviceMapping::getEnable, "1") + .eq(PlatformDeviceMapping::getPushMode, 2) + .gt(PlatformDeviceMapping::getPushInterval, 0) + ); + // 2. 遍历检查是否到达推送时间 + for (PlatformDeviceMapping mapping : mappings) { + if (isDue(mapping, now)) { + pushLatestData(mapping); + } + } +} +``` + +`isDue` 判断：`push_interval` 转换为毫秒后，若 `nowMs - lastPushTime.getTime() >= intervalMs` 则触发。`lastPushTime` 为空时立即触发。 + +`pushLatestData` 从 Redis 缓存读取设备最新数据，按 `device_type` 分发： + +- 直连设备（`device_type=0`）：调用 `DcpPlatformAdapter.pushData`。 +- 网关设备（`device_type=1`）：遍历所有子设备，逐个从 Redis 读取并调用 `DcpPlatformAdapter.pushGatewayData`。 + +推送完成后调用 `IPlatformDeviceMappingService.updateLastPushTime` 更新 `last_push_time`。 + +数据流： + +``` +设备上报数据 → Redis 缓存更新 + ↓ +┌──────────────────────────────────────┐ +│ PlatformPushScheduledJob（每5分钟） │ +│ push_mode=2 且达到时间间隔 │ +│ → 从 Redis 读最新数据 │ +│ → 推送到平台 │ +│ → 更新 last_push_time │ +└──────────────────────────────────────┘ +``` + +### 5.5 DCP 普通推送 DCP 实时推送必须在设备绑定中配置 `push_interface_id`： @@ -392,7 +470,7 @@ DCP 实时推送必须在设备绑定中配置 `push_interface_id`： 时间字段会统一转为带时区的 ISO-8601 毫秒格式，支持输入毫秒时间戳、秒时间戳、ISO 字符串、本地时间字符串。 -### 5.5 网关与数组字段推送 +### 5.6 网关与数组字段推送 网关推送入口：`DcpPlatformAdapter.pushGatewayData`。 @@ -440,7 +518,65 @@ DCP 实时推送必须在设备绑定中配置 `push_interface_id`： Redis TTL 为 `reportIntervalSec * 2`，读取时超过 `reportIntervalSec * 1.5` 视为过期。 -### 5.6 日志与重试 +### 5.7 深部位移整孔批量推送 + +深部位移等整孔场景可在推送接口上开启 `is_batch_push=1`，由 `DcpPlatformAdapter.pushGatewayData` 进入批量推送分支。该能力仍属于网关数组字段推送的一种特化形态，不单独走流程引擎。 + +核心行为： + +1. 查询当前绑定下所有 `device_role=monitor` 的子设备，顺序以 `platform_device_sub_mapping` 返回顺序为准（深部位移场景为浅到深）。 +2. 当前上报子设备使用本次 payload，其余子设备从 Redis 读取缓存；读取 Redis 时使用子设备真实 `product_id + sub_serial_number`，避免误用网关主设备产品 ID。 +3. 每个 monitor 组装为一个请求体对象，`deviceId` 使用该 monitor 入网后回填的 `platform_device_id`，`xTotalBias`、`yTotalBias` 等主体字段使用该 monitor 自身数据。 +4. 每个对象中的 `sensorData` 不是单点，而是从该 monitor 到末端 monitor 的后缀数组；深部位移中表现为“当前深度 + 更深点位”，顺序保持从浅到深。 +5. 最浅节点对象包含全孔从浅到深的点位数据，最深节点对象只包含自己。 +6. 最终一次 HTTP 请求推送整个数组。 + +批量推送不再从“数据推送接口”的 `request_schema` 读取入网参数 `sensorDepths`。`sensorDepths`、`holeDepth` 等仍属于深部位移入网/绑定接口配置；数据推送接口只需配置实际推送字段，例如 `deviceId`、`sensorData`、累计偏移量等。 + +示例请求体结构（4 个串联点，从浅到深为 A/B/C/D）： + +```json +[ + { + "deviceId": "A平台deviceId", + "xTotalBias": "A累计X偏移", + "yTotalBias": "A累计Y偏移", + "sensorData": ["A点数据", "B点数据", "C点数据", "D点数据"], + "ct": 1 + }, + { + "deviceId": "B平台deviceId", + "xTotalBias": "B累计X偏移", + "yTotalBias": "B累计Y偏移", + "sensorData": ["B点数据", "C点数据", "D点数据"], + "ct": 1 + }, + { + "deviceId": "C平台deviceId", + "xTotalBias": "C累计X偏移", + "yTotalBias": "C累计Y偏移", + "sensorData": ["C点数据", "D点数据"], + "ct": 1 + }, + { + "deviceId": "D平台deviceId", + "xTotalBias": "D累计X偏移", + "yTotalBias": "D累计Y偏移", + "sensorData": ["D点数据"], + "ct": 1 + } +] +``` + +配置与排查要点： + +- 在平台接口管理中，为需要整孔推送的“数据推送接口”开启批量推送，即 `platform_interface.is_batch_push=1`。 +- 推送接口 `request_schema` 必须包含实际数据推送字段；不要把入网接口的 `sensorDepths` 当作批量推送前置条件。 +- 子设备必须已入网，monitor 子表行必须有 `platform_device_id`。 +- Redis 中需要有各 monitor 子设备的最新物模型缓存；缺失数据的点位不会进入对应后缀 `sensorData`，若某 monitor 自身无数据则该 monitor 对象会被跳过。 +- 推送日志可在 `platform_push_log` 查看；成功日志会出现 `[DCP] 批量推送完成`。 + +### 5.8 日志与重试 - 流程步骤日志：`log_type=2`。 - 实时数据推送日志：`log_type=1`。 @@ -565,6 +701,7 @@ SET @dcp_platform_id = (SELECT id FROM platform_instance WHERE code = 'dcp' LIMI ### 7.3 DCP 字段约定 - 入网接口中的 `moduleId`、`virtualId` 使用 `source=uuid`，每次请求生成新 UUID。 +- 深部位移绑定接口的 `sensorIds` 使用 `source=cascadedSubDeviceSensorIds`，`sensorDepths` 使用 `source=cascadedSubDeviceDepths`。 - 静力水准绑定接口的 `sensorIds` 使用 `source=referenceSubDeviceSensorIdPairs`。 - 静力水准推送接口定义了数组字段 `sensorData`，其 `itemSchema` 包含 `epoch`、`height`、`temperature`、`sensorId`。 - 轴力计和雨量计 DCP 文档要求累计值递增；当前代码未在适配器中实现“查询上次成功值并跳过非递增值”的通用校验，若目标平台严格校验，需要在推送前补充该逻辑或依赖平台返回失败日志。 @@ -591,22 +728,25 @@ DCP 深部位移绑定接口： {"name":"name","type":"string","source":"device","field":"serial_number","required":true}, {"name":"moduleId","type":"string","source":"uuid","required":true}, {"name":"virtualId","type":"string","source":"uuid","required":true}, - {"name":"sensorDepths","type":"array","source":"onboard_config","value":[1.7,3.4,5.1,6.8],"required":true}, - {"name":"holeDepth","type":"number","source":"onboard_config","value":[1.7,3.4,5.1,6.8],"required":true}, - {"name":"sensorIds","type":"array","source":"subDeviceSensorIds","required":true}, + {"name":"sensorIds","type":"array","source":"cascadedSubDeviceSensorIds","required":true}, + {"name":"sensorDepths","type":"array","source":"cascadedSubDeviceDepths","value":[1.7,3.4,5.1,6.8],"required":true}, + {"name":"holeDepth","type":"number","source":"fixed","value":6.8,"required":true}, {"name":"axisDirection","type":"string","source":"onboard_config","value":"X","required":false}, - {"name":"holeElevation","type":"number","source":"onboard_config","value":0,"required":false}, + {"name":"holeElevation","type":"number","source":"fixed","value":1900,"required":false}, {"name":"zipDuring","type":"number","source":"onboard_config","value":0,"required":false}, {"name":"frequency","type":"number","source":"onboard_config","value":2147483647,"required":false}, {"name":"remark","type":"string","source":"onboard_config","value":"","required":false} ] ``` -网关多子设备绑定深部位移计时，`sensorDepths` 与 `holeDepth` 这类数组值按待入网 monitor 子设备顺序逐个拆分： +网关多子设备绑定深部位移计时，子映射表中的 monitor 顺序表示从浅到深。`sensorDepths.value` 也按浅到深配置完整孔深数组，例如 `[1.7,3.4,5.1,6.8]`。流程执行器会按当前 monitor 截取“当前节点 + 更深节点”，再倒序成 DCP 要求的“最深到当前”： -- 第 1 个 monitor：`sensorDepths` 取 `1.7`，因字段类型为 `array`，最终请求为 `"sensorDepths":[1.7]`；`holeDepth` 为 `1.7`。 -- 第 2 个 monitor：`sensorDepths` 为 `[3.4]`，`holeDepth` 为 `3.4`。 -- 依此类推。 +- 第 1 个 monitor（1.7m）：`sensorDepths=[6.8,5.1,3.4,1.7]`，`sensorIds=[6.8m sensorId,5.1m sensorId,3.4m sensorId,1.7m sensorId]`。 +- 第 2 个 monitor（3.4m）：`sensorDepths=[6.8,5.1,3.4]`，`sensorIds=[6.8m sensorId,5.1m sensorId,3.4m sensorId]`。 +- 第 3 个 monitor（5.1m）：`sensorDepths=[6.8,5.1]`，`sensorIds=[6.8m sensorId,5.1m sensorId]`。 +- 第 4 个 monitor（6.8m）：`sensorDepths=[6.8]`，`sensorIds=[6.8m sensorId]`。 + +`sensorIds` 从每个子映射的 `onboard_config.sensorId` 读取；未配置时回退为子设备序列号。`holeDepth` 是整孔深度，配置为固定数字即可，不再按 monitor 拆分。 深度类字段按工程习惯使用正数，表示距孔口向下的距离；若需要平台侧高程，按 `holeElevation - sensorDepth` 计算，不通过负深度表达方向。 @@ -636,3 +776,7 @@ DCP 深部位移绑定接口： - `PlatformInstanceController.getInfo` 只清空 `access_token`，注释提到密码脱敏，但当前代码未改写 `auth_config` 内密码。 - `platform_device_mapping.platform_device_id` 对网关设备不是主要数据来源；网关 monitor 的平台 deviceId 保存在 `platform_device_sub_mapping.platform_device_id`。 - 静力水准 monitor 先到时会缓存 monitor 数据，但当前主推送分支是在 monitor 到达时读取基准点缓存；若要实现“基准点后到自动补推所有等待 monitor”，需要在基准点上报分支补充扫描 monitor 缓存并推送。 +- 定时推送依赖 Redis 缓存数据，若设备数据未及时写入 Redis，定时推送可能推送过期数据或无数据可推。 +- 定时推送网关设备时会遍历所有子设备，大量子设备场景需关注性能和外部接口调用频率。 +- 定时推送当前使用 `@Scheduled(fixedDelay)`，不支持 Cron 表达式；如需灵活调度需引入 Quartz。 +- 生产环境建议定时推送间隔不低于 5 分钟，避免频繁调用外部接口。 diff --git a/docs/logo.png b/docs/logo.png new file mode 100644 index 00000000..6640463a Binary files /dev/null and b/docs/logo.png differ diff --git a/docs/platform/DCP接口文档(水位计).pdf b/docs/platform/DCP接口文档(水位计).pdf new file mode 100644 index 00000000..cf3c1c9f Binary files /dev/null and b/docs/platform/DCP接口文档(水位计).pdf differ diff --git a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/push/ThingsModelPushJob.java b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/push/ThingsModelPushJob.java index 6d14a807..6ea4245a 100644 --- a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/push/ThingsModelPushJob.java +++ b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/push/ThingsModelPushJob.java @@ -27,7 +27,7 @@ public class ThingsModelPushJob { private ThingsModelPushHandler pushHandler; /** - * 每分钟执行一次，检查并触发到期的推送任务 + * 每5分钟执行一次，检查并触发到期的推送任务 */ @Scheduled(fixedDelay = 300_000) public void execute() { diff --git a/fastbee-open-api/src/main/java/com/fastbee/controller/platform/PlatformPushLogController.java b/fastbee-open-api/src/main/java/com/fastbee/controller/platform/PlatformPushLogController.java index 8c24e08c..ae2b0b75 100644 --- a/fastbee-open-api/src/main/java/com/fastbee/controller/platform/PlatformPushLogController.java +++ b/fastbee-open-api/src/main/java/com/fastbee/controller/platform/PlatformPushLogController.java @@ -53,8 +53,8 @@ public class PlatformPushLogController extends BaseController { @PreAuthorize("@ss.hasPermi('platform:pushLog:remove')") @DeleteMapping("/clean") - @ApiOperation("批量删除推送日志") - public AjaxResult clean(@PathVariable Long[] ids) { + @ApiOperation("清空所有推送日志") + public AjaxResult clean() { pushLogService.cleanPushLog(); return success(); } diff --git a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/adapter/dcp/DcpPlatformAdapter.java b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/adapter/dcp/DcpPlatformAdapter.java index 50c3f2b7..7ad9b109 100644 --- a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/adapter/dcp/DcpPlatformAdapter.java +++ b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/adapter/dcp/DcpPlatformAdapter.java @@ -6,9 +6,13 @@ import com.alibaba.fastjson2.JSONObject; import com.baomidou.mybatisplus.core.toolkit.Wrappers; import com.dtflys.forest.http.ForestRequest; import com.dtflys.forest.http.ForestRequestType; +import com.fastbee.common.core.redis.RedisKeyBuilder; import com.fastbee.common.utils.spring.SpringUtils; import com.fastbee.http.model.HttpClientConfig; import com.fastbee.http.service.HttpClientFactory; +import com.fastbee.iot.cache.ITSLCache; +import com.fastbee.iot.domain.Device; +import com.fastbee.iot.model.ThingsModels.ValueItem; import com.fastbee.iot.platform.domain.PlatformDeviceMapping; import com.fastbee.iot.platform.domain.PlatformDeviceSubMapping; import com.fastbee.iot.platform.domain.PlatformFlow; @@ -22,6 +26,7 @@ import com.fastbee.iot.platform.mapper.PlatformInterfaceMapper; import com.fastbee.iot.platform.service.IPlatformDeviceSubMappingService; import com.fastbee.iot.platform.service.IPlatformInstanceService; import com.fastbee.iot.platform.service.IPlatformPushLogService; +import com.fastbee.iot.service.IDeviceService; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; @@ -33,10 +38,13 @@ import java.time.OffsetDateTime; import java.time.ZoneId; import java.time.format.DateTimeFormatter; import java.time.format.DateTimeParseException; +import java.util.ArrayList; import java.util.Date; import java.util.HashMap; +import java.util.LinkedHashMap; import java.util.List; import java.util.Map; +import java.util.stream.Collectors; /** * DCP 平台实时数据推送适配器 @@ -73,6 +81,10 @@ public class DcpPlatformAdapter { private PlatformInterfaceMapper interfaceMapper; @Autowired private IPlatformDeviceSubMappingService subMappingService; + @Autowired + private ITSLCache itslCache; + @Autowired + private IDeviceService deviceService; /** * 推送设备上报数据到 DCP（直连设备） @@ -130,6 +142,12 @@ public class DcpPlatformAdapter { return; } + // 判断是否批量推送 + if (pushInterface.getIsBatchPush() != null && pushInterface.getIsBatchPush() == 1) { + pushBatchSubDevicesData(platform, mapping, subMapping, reportSerialNumber, payload, pushPath, interfaceCode, requestSchema); + return; + } + // 无数组字段 → 简单推送，使用子设备 platformDeviceId 和主表 fieldMapping if (!hasArrayField(requestSchema)) { // 如果只有一个设备完全可以走 @@ -1057,6 +1075,218 @@ public class DcpPlatformAdapter { return value == null ? null : value.trim(); } + // ==================== 批量推送 ==================== + + /** + * 批量推送子设备数据（整孔推送） + *

+ * 查询所有监测点子设备，截取到当前上报子设备， + * 复用已有 fieldMapping + requestSchema 组装每个子设备数据，一次 HTTP 请求推送。 + *

+ * 当前上报的子设备使用传入的 payload（最新数据），其余子设备从 Redis 缓存读取。 + * 推送顺序：从浅到深（与绑定时从深到浅相反）。 + */ + private void pushBatchSubDevicesData(PlatformInstance platform, PlatformDeviceMapping mapping, + PlatformDeviceSubMapping currentSubMapping, String reportSerialNumber, + String payload, String pushPath, String interfaceCode, + JSONArray requestSchema) { + try { + // 1. 查询所有子设备。批量推送只依赖子映射顺序和推送接口 schema， + // 不再从推送接口 request_schema 读取入网用的 sensorDepths。 + List allSubMappings = subMappingService.selectByParentId(mapping.getId()); + if (allSubMappings == null || allSubMappings.isEmpty()) { + log.warn("[DCP] 网关设备 {} 无子设备映射", mapping.getSerialNumber()); + return; + } + List monitors = allSubMappings.stream() + .filter(PlatformDeviceSubMapping::isMonitor) + .collect(Collectors.toList()); + + if (monitors.isEmpty()) { + log.warn("[DCP] 无监测点子设备 mappingId={}", mapping.getId()); + return; + } + + // 2. 找到当前子设备的索引位置 + int currentIndex = -1; + for (int i = 0; i < monitors.size(); i++) { + if (monitors.get(i).getId().equals(currentSubMapping.getId())) { + currentIndex = i; + break; + } + } + + if (currentIndex == -1) { + log.warn("[DCP] 当前子设备 {} 不在监测点列表中 mappingId={}", + reportSerialNumber, mapping.getId()); + return; + } + + // 3. 准备公共参数（复用已有 fieldMapping + requestSchema） + JSONArray fieldMappings = parseJsonArray(mapping.getFieldMapping()); + String fieldName = resolveArrayFieldName(requestSchema, null); + JSONObject extra = null; + if (StringUtils.hasText(mapping.getExtraMapping())) { + extra = JSON.parseObject(mapping.getExtraMapping()); + if (StringUtils.hasText(extra.getString("fieldName"))) { + fieldName = extra.getString("fieldName"); + } + } + JSONArray itemFieldMappings = extra != null ? extra.getJSONArray("itemFieldMappings") : null; + + // 解析当前上报子设备的 payload + JSONObject currentReportData = parsePayloadToMap(payload); + + // 4. 先按子设备顺序准备所有 monitor 的最新数据与 sensorData item。 + // 当前上报子设备使用 payload，其余从 Redis 缓存读取。 + Map reportDataBySubId = new LinkedHashMap<>(); + Map sensorItemBySubId = new LinkedHashMap<>(); + for (PlatformDeviceSubMapping sub : monitors) { + String subSn = sub.getSubSerialNumber(); + JSONObject reportData; + if (sub.getId() != null && sub.getId().equals(currentSubMapping.getId()) + && currentReportData != null && !currentReportData.isEmpty()) { + reportData = currentReportData; + } else { + reportData = buildReportDataFromCache(resolveProductId(mapping, subSn), subSn); + } + if (reportData == null || reportData.isEmpty()) { + log.warn("[DCP] 子设备 {} 无数据（缓存为空），跳过其 sensorData", subSn); + continue; + } + + String sensorId = resolveSubDeviceSensorId(sub); + JSONObject sensorDataItem = buildArrayItem( + platform, mapping, requestSchema, reportData, fieldName, itemFieldMappings, sensorId); + if (sensorDataItem == null || sensorDataItem.isEmpty()) { + log.warn("[DCP] 子设备 {} 数据映射为空，跳过其 sensorData", subSn); + continue; + } + reportDataBySubId.put(sub.getId(), reportData); + sensorItemBySubId.put(sub.getId(), sensorDataItem); + } + + // 5. 每个 monitor 生成一个对象：deviceId 使用当前 monitor 平台设备ID， + // sensorData 使用当前 monitor 到最深 monitor 的后缀数组，顺序保持从浅到深。 + JSONArray pushBodyArray = new JSONArray(); + for (int i = 0; i < monitors.size(); i++) { + PlatformDeviceSubMapping sub = monitors.get(i); + String subSn = sub.getSubSerialNumber(); + if (!StringUtils.hasText(sub.getPlatformDeviceId())) { + log.warn("[DCP] 子设备 {} 未分配平台设备ID，跳过", subSn); + continue; + } + JSONObject reportData = reportDataBySubId.get(sub.getId()); + if (reportData == null || reportData.isEmpty()) { + log.warn("[DCP] 子设备 {} 无自身数据，跳过该设备对象", subSn); + continue; + } + + JSONArray sensorDataArr = new JSONArray(); + for (int j = i; j < monitors.size(); j++) { + PlatformDeviceSubMapping sensorSub = monitors.get(j); + JSONObject sensorDataItem = sensorItemBySubId.get(sensorSub.getId()); + if (sensorDataItem != null && !sensorDataItem.isEmpty()) { + sensorDataArr.add(sensorDataItem); + } else { + log.warn("[DCP] 子设备 {} 缺少 sensorData，设备 {} 的后缀数组不包含该点", + sensorSub.getSubSerialNumber(), subSn); + } + } + if (sensorDataArr.isEmpty()) { + log.warn("[DCP] 子设备 {} 后缀 sensorData 为空，跳过", subSn); + continue; + } + + JSONObject bodyFields = buildObjectBySchema(platform, mapping, requestSchema, reportData, fieldMappings); + JSONArray deviceBodyArr = buildArrayPushBody(mapping, bodyFields, fieldName, + sensorDataArr, sub.getPlatformDeviceId()); + if (!deviceBodyArr.isEmpty()) { + pushBodyArray.add(deviceBodyArr.get(0)); + } + } + + // 6. 空数据保护 + if (pushBodyArray.isEmpty()) { + log.warn("[DCP] 批量推送无有效子设备数据，跳过 mappingId={}", mapping.getId()); + return; + } + + // 7. 一次 HTTP 请求推送 + String baseUrl = platform.getBaseUrl() + pushPath; + PlatformPushLog logEntry = createLogEntry(platform, mapping, interfaceCode, reportSerialNumber); + doHttpPost(platform, baseUrl, JSON.toJSONString(pushBodyArray), logEntry); + pushLogService.asyncSave(logEntry); + + log.info("[DCP] 批量推送完成 mappingId={} 当前子设备={} 推送设备对象数={}/{}", + mapping.getId(), reportSerialNumber, pushBodyArray.size(), monitors.size()); + + } catch (Exception e) { + log.error("[DCP] 批量推送失败 device={}", reportSerialNumber, e); + PlatformPushLog logEntry = createLogEntry(platform, mapping, interfaceCode, reportSerialNumber); + logEntry.setStatus(2); + logEntry.setErrorMsg(e.getMessage()); + pushLogService.asyncSave(logEntry); + } + } + + /** + * 获取上报设备所属产品。网关批量推送读取其它子设备缓存时，必须使用子设备产品ID。 + */ + private Long resolveProductId(PlatformDeviceMapping mapping, String serialNumber) { + if (StringUtils.hasText(serialNumber)) { + Device device = deviceService.selectDeviceBySerialNumber(serialNumber); + if (device != null && device.getProductId() != null) { + return device.getProductId(); + } + } + return mapping.getProductId(); + } + + /** + * 从 Redis 缓存构建 reportData（与 parsePayloadToMap 输出格式一致） + */ + private JSONObject buildReportDataFromCache(Long productId, String subSerialNumber) { + if (productId == null || !StringUtils.hasText(subSerialNumber)) { + return null; + } + String redisKey = RedisKeyBuilder.buildTSLVCacheKey(productId, subSerialNumber.toUpperCase()); + Map cacheMap = itslCache.getSimpleThingsModelValueItemByCache(redisKey); + if (cacheMap == null || cacheMap.isEmpty()) { + return null; + } + JSONObject reportData = new JSONObject(); + for (Map.Entry entry : cacheMap.entrySet()) { + String identifier = entry.getKey(); + ValueItem valueItem = entry.getValue(); + reportData.put(identifier, valueItem.getValue()); + if (valueItem.getTs() != null) { + reportData.put(identifier + "__ts", valueItem.getTs().getTime()); + } + } + return reportData; + } + + /** + * 获取子设备的 sensorId + *

+ * 优先从 onboard_config 中取 sensorId，否则使用子设备序列号。 + */ + private String resolveSubDeviceSensorId(PlatformDeviceSubMapping subMapping) { + if (StringUtils.hasText(subMapping.getOnboardConfig())) { + try { + JSONObject config = JSON.parseObject(subMapping.getOnboardConfig()); + String sensorId = config.getString("sensorId"); + if (StringUtils.hasText(sensorId)) { + return sensorId; + } + } catch (Exception e) { + log.debug("[DCP] 解析 onboardConfig 失败 subSn={}", subMapping.getSubSerialNumber()); + } + } + return subMapping.getSubSerialNumber(); + } + // ==================== Token 管理 ==================== private String ensureToken(PlatformInstance platform) { diff --git a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/domain/PlatformDeviceMapping.java b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/domain/PlatformDeviceMapping.java index d6098a85..fc3d64ab 100644 --- a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/domain/PlatformDeviceMapping.java +++ b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/domain/PlatformDeviceMapping.java @@ -1,6 +1,7 @@ package com.fastbee.iot.platform.domain; import com.baomidou.mybatisplus.annotation.IdType; +import com.baomidou.mybatisplus.annotation.TableField; import com.baomidou.mybatisplus.annotation.TableId; import com.baomidou.mybatisplus.annotation.TableName; import com.fasterxml.jackson.annotation.JsonFormat; @@ -30,10 +31,14 @@ public class PlatformDeviceMapping implements Serializable { @ApiModelProperty("FastBee设备ID") private Long deviceId; + @ApiModelProperty("FastBee产品ID（非持久化，JOIN查询）") + @TableField(exist = false) + private Long productId; + @ApiModelProperty("FastBee设备序列号") private String serialNumber; - @ApiModelProperty("FastBee设备名称") + @ApiModelProperty("默认为本平台设备名称，可以修改为对方平台需要的名称") private String deviceName; @ApiModelProperty("DCP侧deviceId，入网后自动回填") @@ -63,6 +68,16 @@ public class PlatformDeviceMapping implements Serializable { @ApiModelProperty("是否启用") private String enable; + @ApiModelProperty("推送模式：1=实时 2=定时（默认1）") + private Integer pushMode; + + @ApiModelProperty("定时推送间隔（分钟），0=不自动推送") + private Integer pushInterval; + + @ApiModelProperty("最近推送时间") + @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss") + private Date lastPushTime; + @ApiModelProperty("备注") private String remark; diff --git a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/domain/PlatformInterface.java b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/domain/PlatformInterface.java index 34e3ab6d..10f47e0b 100644 --- a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/domain/PlatformInterface.java +++ b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/domain/PlatformInterface.java @@ -51,6 +51,9 @@ public class PlatformInterface implements Serializable { @ApiModelProperty("排序") private Integer sortOrder; + @ApiModelProperty("是否批量推送（0=否，1=是）") + private Integer isBatchPush; + @ApiModelProperty("备注") private String remark; diff --git a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/flow/PlatformFlowExecutor.java b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/flow/PlatformFlowExecutor.java index 9cba00fa..ff17374d 100644 --- a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/flow/PlatformFlowExecutor.java +++ b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/flow/PlatformFlowExecutor.java @@ -14,6 +14,7 @@ import com.fastbee.iot.mapper.DeviceMapper; import com.fastbee.iot.mapper.SubGatewayMapper; import com.fastbee.iot.platform.domain.*; import com.fastbee.iot.platform.mapper.PlatformDeviceMappingMapper; +import com.fastbee.iot.platform.mapper.PlatformDeviceSubMappingMapper; import com.fastbee.iot.platform.mapper.PlatformFlowStepMapper; import com.fastbee.iot.platform.mapper.PlatformInstanceMapper; import com.fastbee.iot.platform.mapper.PlatformInterfaceMapper; @@ -48,6 +49,8 @@ public class PlatformFlowExecutor { private SubGatewayMapper subGatewayMapper; @Autowired private DeviceMapper deviceMapper; + @Autowired + private PlatformDeviceSubMappingMapper subMappingMapper; /** * 执行流程 @@ -315,6 +318,10 @@ public class PlatformFlowExecutor { return resolveOnboardConfigValue(ctx, fd); case "subDeviceSensorIds": return collectSubDeviceSensorIds(ctx); + case "cascadedSubDeviceSensorIds": + return collectCascadedSubDeviceSensorIds(ctx); + case "cascadedSubDeviceDepths": + return collectCascadedSubDeviceDepths(ctx, fd); case "referenceSubDeviceSensorIdPairs": return collectReferenceSubDeviceSensorIdPairs(ctx); default: @@ -545,6 +552,87 @@ public class PlatformFlowExecutor { .collect(Collectors.toList()); } + private List collectCascadedSubDeviceSensorIds(FlowContext ctx) { + List suffix = collectCascadedSubDeviceSuffix(ctx); + Collections.reverse(suffix); + return suffix.stream() + .map(sub -> resolveOnboardSensorId(sub.getOnboardConfig(), sub.getSubSerialNumber())) + .filter(StringUtils::hasText) + .collect(Collectors.toList()); + } + + private List collectCascadedSubDeviceDepths(FlowContext ctx, JSONObject fieldDef) { + List monitors = collectCascadedSubDeviceMonitors(ctx); + int currentIndex = resolveCurrentMonitorIndex(ctx, monitors); + List allDepths = resolveConfiguredDepths(fieldDef); + if (allDepths.size() < monitors.size()) { + throw new RuntimeException("串联子设备深度数组配置数量不足，监测点数量=" + + monitors.size() + "，深度数量=" + allDepths.size()); + } + List suffix = new ArrayList<>(allDepths.subList(currentIndex, monitors.size())); + Collections.reverse(suffix); + return suffix; + } + + private List collectCascadedSubDeviceSuffix(FlowContext ctx) { + List monitors = collectCascadedSubDeviceMonitors(ctx); + int currentIndex = resolveCurrentMonitorIndex(ctx, monitors); + return new ArrayList<>(monitors.subList(currentIndex, monitors.size())); + } + + private List collectCascadedSubDeviceMonitors(FlowContext ctx) { + PlatformDeviceMapping mapping = ctx.getDeviceMapping(); + if (mapping == null || mapping.getDeviceType() == null || mapping.getDeviceType() != 1) { + throw new RuntimeException("cascadedSubDeviceSensorIds/cascadedSubDeviceDepths 仅支持网关串联子设备入网流程"); + } + if (mapping.getId() == null) { + throw new RuntimeException("串联子设备入网缺少映射ID"); + } + List subMappings = subMappingMapper.selectByParentId(mapping.getId()); + List monitors = subMappings == null ? Collections.emptyList() : subMappings.stream() + .filter(PlatformDeviceSubMapping::isMonitor) + .collect(Collectors.toList()); + if (monitors.isEmpty()) { + throw new RuntimeException("串联子设备入网未配置监测点子设备"); + } + return monitors; + } + + private int resolveCurrentMonitorIndex(FlowContext ctx, List monitors) { + String currentSerialNumber = ctx.getCurrentMonitorSerialNumber(); + if (StringUtils.hasText(currentSerialNumber)) { + for (int i = 0; i < monitors.size(); i++) { + if (sameText(currentSerialNumber, monitors.get(i).getSubSerialNumber())) { + return i; + } + } + } + Integer index = ctx.getCurrentMonitorIndex(); + if (index != null && index >= 0 && index < monitors.size()) { + return index; + } + throw new RuntimeException("串联子设备入网无法确定当前监测点序号"); + } + + private List resolveConfiguredDepths(JSONObject fieldDef) { + if (!fieldDef.containsKey("value")) { + throw new RuntimeException("串联子设备未配置深度数组"); + } + Object value = fieldDef.get("value"); + List list = toList(value); + if (list == null && value instanceof String) { + try { + list = JSON.parseArray(((String) value).trim()); + } catch (Exception ignored) { + list = null; + } + } + if (list == null || list.isEmpty()) { + throw new RuntimeException("串联子设备深度数组为空"); + } + return new ArrayList<>(list); + } + /** * 网关入网时：优先按子表中当前 reference 行取 [基准点sensorId, 监测点sensorId] 一对。 * 直连设备/旧逻辑：从 extra_mapping 解析所有配对。 diff --git a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/job/PlatformPushScheduledJob.java b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/job/PlatformPushScheduledJob.java new file mode 100644 index 00000000..ca01420e --- /dev/null +++ b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/job/PlatformPushScheduledJob.java @@ -0,0 +1,323 @@ +package com.fastbee.iot.platform.job; + +import com.alibaba.fastjson2.JSONArray; +import com.alibaba.fastjson2.JSONObject; +import com.baomidou.mybatisplus.core.toolkit.Wrappers; +import com.fastbee.common.core.redis.RedisKeyBuilder; +import com.fastbee.iot.cache.ITSLCache; +import com.fastbee.iot.domain.Device; +import com.fastbee.iot.model.ThingsModels.ValueItem; +import com.fastbee.iot.platform.adapter.dcp.DcpPlatformAdapter; +import com.fastbee.iot.platform.domain.PlatformDeviceMapping; +import com.fastbee.iot.platform.domain.PlatformDeviceSubMapping; +import com.fastbee.iot.platform.domain.PlatformInstance; +import com.fastbee.iot.platform.domain.PlatformInterface; +import com.fastbee.iot.platform.mapper.PlatformInterfaceMapper; +import com.fastbee.iot.platform.service.IPlatformDeviceMappingService; +import com.fastbee.iot.platform.service.IPlatformDeviceSubMappingService; +import com.fastbee.iot.platform.service.IPlatformInstanceService; +import com.fastbee.iot.service.IDeviceService; +import lombok.extern.slf4j.Slf4j; +import org.apache.commons.collections4.CollectionUtils; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.scheduling.annotation.Scheduled; +import org.springframework.stereotype.Component; +import org.springframework.util.StringUtils; + +import java.util.Date; +import java.util.List; +import java.util.Map; +import java.util.stream.Collectors; + +/** + * 平台定时推送调度器 + *

+ * 每5分钟检查所有启用的定时推送配置，按照 push_interval 决定是否触发本次推送。 + * 支持两种推送模式： + * 1. 实时推送（pushMode=1）：设备上报即刻推送，不走此定时任务 + * 2. 定时推送（pushMode=2）：仅按时间间隔推送最新数据 + *

+ * 批量推送（isBatchPush=1）场景：取最深层监测点一次性推送整孔数据。 + * + * @author Sumuxi + */ +@Slf4j +@Component +public class PlatformPushScheduledJob { + + @Autowired + private IPlatformDeviceMappingService mappingService; + @Autowired + private IPlatformInstanceService instanceService; + @Autowired + private IPlatformDeviceSubMappingService subMappingService; + @Autowired + private DcpPlatformAdapter dcpAdapter; + @Autowired + private ITSLCache itslCache; + @Autowired + private IDeviceService deviceService; + @Autowired + private PlatformInterfaceMapper interfaceMapper; + + /** + * 每5分钟执行一次，检查并触发到期的推送任务 + */ + @Scheduled(fixedDelay = 300_000) + public void execute() { + // 查询所有启用的定时推送配置（push_mode=2） + List mappings = mappingService.list( + Wrappers.lambdaQuery() + .eq(PlatformDeviceMapping::getEnable, "1") + .eq(PlatformDeviceMapping::getPushMode, 2) + .gt(PlatformDeviceMapping::getPushInterval, 0) + ); + + if (CollectionUtils.isEmpty(mappings)) { + return; + } + + long now = System.currentTimeMillis(); + for (PlatformDeviceMapping mapping : mappings) { + try { + if (isDue(mapping, now)) { + log.info("[PlatformPushJob] 触发定时推送 mappingId={} sn={}", mapping.getId(), mapping.getSerialNumber()); + pushLatestData(mapping); + } + } catch (Exception e) { + log.error("[PlatformPushJob] 定时推送异常 mappingId={}: {}", mapping.getId(), e.getMessage(), e); + } + } + } + + /** + * 判断配置是否到达推送时间 + */ + private boolean isDue(PlatformDeviceMapping mapping, long nowMs) { + if (mapping.getPushInterval() == null || mapping.getPushInterval() <= 0) { + return false; + } + Date lastPush = mapping.getLastPushTime(); + if (lastPush == null) { + // 从未推送过，立即触发 + return true; + } + long intervalMs = (long) mapping.getPushInterval() * 60 * 1000; + return (nowMs - lastPush.getTime()) >= intervalMs; + } + + /** + * 推送最新数据 + * 一次推送下，只存在有且只有一个 productId + */ + private void pushLatestData(PlatformDeviceMapping mapping) { + // 如果 productId 为空，通过 deviceId 查询 + if (mapping.getProductId() == null && mapping.getDeviceId() != null) { + Long productId = deviceService.selectProductIdByDeviceId(mapping.getDeviceId()); + if (productId == null) { + log.warn("[PlatformPushJob] 设备 {} 不存在或无产品ID", mapping.getSerialNumber()); + return; + } + mapping.setProductId(productId); + } + + // 获取平台实例 + PlatformInstance platform = instanceService.getById(mapping.getPlatformId()); + if (platform == null || !"1".equals(platform.getEnable())) { + log.warn("[PlatformPushJob] 平台不存在或已禁用 platformId={}", mapping.getPlatformId()); + return; + } + + Date dataTime = resolveLatestDataTime(mapping); + if (dataTime == null) { + log.warn("[PlatformPushJob] 设备 {} 无有效数据时间，跳过推送", mapping.getSerialNumber()); + return; + } + if (mapping.getLastPushTime() != null && !dataTime.after(mapping.getLastPushTime())) { + log.info("[PlatformPushJob] 数据未变化，跳过重复推送 mappingId={} dataTime={} lastPushTime={}", + mapping.getId(), dataTime, mapping.getLastPushTime()); + return; + } + + if (mapping.getDeviceType() != null && mapping.getDeviceType() == 1) { + // 网关设备：区分批量推送和逐个子设备推送 + PlatformInterface pushInterface = loadPushInterface(mapping); + if (pushInterface != null && pushInterface.getIsBatchPush() != null && pushInterface.getIsBatchPush() == 1) { + // 批量推送 + pushBatchGatewayData(platform, mapping); + } else { + // 非批量：逐个子设备推送 + pushGatewayLatestData(platform, mapping); + } + } else { + // 直连设备 + String serialNumber = mapping.getSerialNumber().toUpperCase(); + Map cacheMap = loadCacheMap(mapping.getProductId(), serialNumber); + if (cacheMap == null || cacheMap.isEmpty()) { + log.warn("[PlatformPushJob] 设备 {} 无缓存数据，跳过推送", serialNumber); + return; + } + String payload = buildPayloadFromCache(cacheMap); + dcpAdapter.pushData(platform, mapping, serialNumber, payload); + } + + // lastPushTime 记录最近已推送的数据时间，用于幂等判断 + mappingService.updateLastPushTime(mapping.getId(), dataTime); + } + + /** + * 批量推送网关设备整孔数据 + *

+ * 取最深层（最后一个）监测点作为触发点，构造模拟 payload， + * 调用 pushGatewayData 进入批量推送路径，推送从浅到深的全量子设备数据。 + */ + private void pushBatchGatewayData(PlatformInstance platform, PlatformDeviceMapping mapping) { + List subMappings = subMappingService.selectByParentId(mapping.getId()); + if (CollectionUtils.isEmpty(subMappings)) { + log.warn("[PlatformPushJob] 网关设备 {} 无子设备映射", mapping.getSerialNumber()); + return; + } + + // 取最后一个监测点作为触发点（最深层，批量推送会包含所有子设备） + List monitors = subMappings.stream() + .filter(PlatformDeviceSubMapping::isMonitor) + .collect(Collectors.toList()); + if (monitors.isEmpty()) { + log.warn("[PlatformPushJob] 网关设备 {} 无监测点子设备", mapping.getSerialNumber()); + return; + } + PlatformDeviceSubMapping lastMonitor = monitors.get(monitors.size() - 1); + if (!StringUtils.hasText(lastMonitor.getPlatformDeviceId())) { + log.warn("[PlatformPushJob] 最深层监测点 {} 未分配平台设备ID", lastMonitor.getSubSerialNumber()); + return; + } + + // 从 Redis 缓存构造模拟 payload（最深层监测点的数据） + Long productId = resolveProductId(mapping, lastMonitor.getSubSerialNumber()); + Map cacheMap = loadCacheMap(productId, lastMonitor.getSubSerialNumber()); + if (cacheMap == null || cacheMap.isEmpty()) { + log.warn("[PlatformPushJob] 最深层监测点 {} 无缓存数据 productId={}", lastMonitor.getSubSerialNumber(), productId); + return; + } + String payload = buildPayloadFromCache(cacheMap); + + dcpAdapter.pushGatewayData(platform, mapping, lastMonitor, lastMonitor.getSubSerialNumber(), payload); + log.info("[PlatformPushJob] 批量推送完成 mappingId={} 最深层监测点={} 子设备总数={}", mapping.getId(), lastMonitor.getSubSerialNumber(), monitors.size()); + } + + /** + * 推送网关设备最新数据（非批量，逐个子设备推送） + */ + private void pushGatewayLatestData(PlatformInstance platform, PlatformDeviceMapping mapping) { + List subMappings = subMappingService.selectByParentId(mapping.getId()); + if (CollectionUtils.isEmpty(subMappings)) { + log.warn("[PlatformPushJob] 网关设备 {} 无子设备映射", mapping.getSerialNumber()); + return; + } + + for (PlatformDeviceSubMapping subMapping : subMappings) { + if (subMapping == null || subMapping.getPlatformDeviceId() == null) { + continue; + } + try { + String subSerialNumber = subMapping.getSubSerialNumber(); + Map subCacheMap = loadCacheMap(resolveProductId(mapping, subSerialNumber), subSerialNumber); + + if (subCacheMap == null || subCacheMap.isEmpty()) { + log.debug("[PlatformPushJob] 子设备 {} 无缓存数据，跳过", subSerialNumber); + continue; + } + + String payload = buildPayloadFromCache(subCacheMap); + dcpAdapter.pushGatewayData(platform, mapping, subMapping, subSerialNumber, payload); + } catch (Exception e) { + log.error("[PlatformPushJob] 网关子设备推送失败 subSn={}", subMapping.getSubSerialNumber(), e); + } + } + } + + /** + * 解析当前最新数据时间。网关多子设备按第一个监测点子设备的数据时间判断幂等。 + */ + private Date resolveLatestDataTime(PlatformDeviceMapping mapping) { + if (mapping.getDeviceType() != null && mapping.getDeviceType() == 1) { + List subMappings = subMappingService.selectByParentId(mapping.getId()); + if (CollectionUtils.isEmpty(subMappings)) { + return null; + } + List monitors = subMappings.stream() + .filter(PlatformDeviceSubMapping::isMonitor) + .collect(Collectors.toList()); + PlatformDeviceSubMapping firstSub = monitors.isEmpty() ? subMappings.get(0) : monitors.get(0); + String subSerialNumber = firstSub.getSubSerialNumber(); + return resolveLatestDataTime(loadCacheMap(resolveProductId(mapping, subSerialNumber), subSerialNumber)); + } + String serialNumber = mapping.getSerialNumber(); + return resolveLatestDataTime(loadCacheMap(mapping.getProductId(), serialNumber)); + } + + private Date resolveLatestDataTime(Map cacheMap) { + if (cacheMap == null || cacheMap.isEmpty()) { + return null; + } + Date latest = null; + for (ValueItem item : cacheMap.values()) { + if (item == null || item.getTs() == null) { + continue; + } + if (latest == null || item.getTs().after(latest)) { + latest = item.getTs(); + } + } + return latest; + } + + /** + * 获取上报设备所属产品。网关批量/子设备推送时，Redis 缓存按子设备产品ID+子设备编号存储。 + */ + private Long resolveProductId(PlatformDeviceMapping mapping, String serialNumber) { + if (StringUtils.hasText(serialNumber)) { + Device device = deviceService.selectDeviceBySerialNumber(serialNumber); + if (device != null && device.getProductId() != null) { + return device.getProductId(); + } + } + return mapping.getProductId(); + } + + /** + * 从 Redis 缓存加载子设备数据 + */ + private Map loadCacheMap(Long productId, String serialNumber) { + if (productId == null || !StringUtils.hasText(serialNumber)) { + return null; + } + String redisKey = RedisKeyBuilder.buildTSLVCacheKey(productId, serialNumber.toUpperCase()); + return itslCache.getSimpleThingsModelValueItemByCache(redisKey); + } + + /** + * 从缓存构建payload（模拟设备上报格式） + */ + private String buildPayloadFromCache(Map cacheMap) { + JSONArray arr = new JSONArray(); + for (Map.Entry entry : cacheMap.entrySet()) { + JSONObject item = new JSONObject(); + item.put("identifier", entry.getKey()); + item.put("value", entry.getValue().getValue()); + item.put("ts", entry.getValue().getTs()); + arr.add(item); + } + return arr.toJSONString(); + } + + /** + * 加载推送接口配置 + */ + private PlatformInterface loadPushInterface(PlatformDeviceMapping mapping) { + if (mapping.getPushInterfaceId() == null) { + return null; + } + return interfaceMapper.selectById(mapping.getPushInterfaceId()); + } +} diff --git a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/service/IPlatformDeviceMappingService.java b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/service/IPlatformDeviceMappingService.java index acb3087b..eac312ab 100644 --- a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/service/IPlatformDeviceMappingService.java +++ b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/service/IPlatformDeviceMappingService.java @@ -3,6 +3,7 @@ package com.fastbee.iot.platform.service; import com.baomidou.mybatisplus.extension.service.IService; import com.fastbee.iot.platform.domain.PlatformDeviceMapping; +import java.util.Date; import java.util.List; public interface IPlatformDeviceMappingService extends IService { @@ -14,4 +15,7 @@ public interface IPlatformDeviceMappingService extends IService selectSerialNumberByProductId(Long productId); + /** + * 根据设备ID查询产品ID + * + * @param deviceId 设备ID + * @return 产品ID + */ + public Long selectProductIdByDeviceId(Long deviceId); + /** * 获取产品下面的设备数量 * diff --git a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/service/impl/DeviceServiceImpl.java b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/service/impl/DeviceServiceImpl.java index 8ea974a0..3e8311a2 100644 --- a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/service/impl/DeviceServiceImpl.java +++ b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/service/impl/DeviceServiceImpl.java @@ -970,6 +970,15 @@ public class DeviceServiceImpl implements IDeviceService { return deviceMapper.selectSerialNumberByProductId(productId); } + @Override + public Long selectProductIdByDeviceId(Long deviceId) { + if (deviceId == null) { + return null; + } + Device device = deviceMapper.selectDeviceByDeviceId(deviceId); + return device != null ? device.getProductId() : null; + } + @Override public int selectDeviceCountByProductId(Long productId) { return deviceMapper.selectDeviceCountByProductId(productId); diff --git a/fastbee-service/fastbee-iot-service/src/main/resources/mapper/iot/platform/PlatformDeviceMappingMapper.xml b/fastbee-service/fastbee-iot-service/src/main/resources/mapper/iot/platform/PlatformDeviceMappingMapper.xml index d3a267d8..e17e37f9 100644 --- a/fastbee-service/fastbee-iot-service/src/main/resources/mapper/iot/platform/PlatformDeviceMappingMapper.xml +++ b/fastbee-service/fastbee-iot-service/src/main/resources/mapper/iot/platform/PlatformDeviceMappingMapper.xml @@ -17,6 +17,9 @@ + + + diff --git a/fastbee-service/fastbee-iot-service/src/main/resources/mapper/iot/platform/PlatformInterfaceMapper.xml b/fastbee-service/fastbee-iot-service/src/main/resources/mapper/iot/platform/PlatformInterfaceMapper.xml index 107fe3ae..62daa885 100644 --- a/fastbee-service/fastbee-iot-service/src/main/resources/mapper/iot/platform/PlatformInterfaceMapper.xml +++ b/fastbee-service/fastbee-iot-service/src/main/resources/mapper/iot/platform/PlatformInterfaceMapper.xml @@ -13,6 +13,7 @@ + diff --git a/sql/platform-integration.sql b/sql/platform-integration.sql index 8799fa2c..dbe83e3a 100644 --- a/sql/platform-integration.sql +++ b/sql/platform-integration.sql @@ -61,6 +61,7 @@ CREATE TABLE IF NOT EXISTS `platform_interface` ( `name` varchar(100) NOT NULL COMMENT '接口名称', `code` varchar(100) NOT NULL COMMENT '接口标识', `method` varchar(10) NOT NULL COMMENT 'HTTP方法: GET/POST/PUT/DELETE', + `is_batch_push` tinyint(1) NOT NULL DEFAULT 0 COMMENT '是否批量推送(0=否,1=是)，用于深部位移等整孔数据推送', `path` varchar(255) NOT NULL COMMENT '接口路径，支持路径变量 {userId}', `content_type` varchar(100) DEFAULT 'application/json' COMMENT 'Content-Type', `request_schema` json DEFAULT NULL COMMENT '入参字段定义 [{name,type,source,value,field,contextKey,required,desc}]', @@ -75,6 +76,8 @@ CREATE TABLE IF NOT EXISTS `platform_interface` ( KEY `idx_platform_id` (`platform_id`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='平台接口定义'; +ALTER TABLE `platform_interface` ADD COLUMN `is_batch_push` tinyint(1) NOT NULL DEFAULT 0 COMMENT '是否批量推送(0=否,1=是)，用于深部位移等整孔数据推送' AFTER `method` + -- ---------------------------- -- 3. 流程定义 -- ---------------------------- @@ -119,7 +122,7 @@ CREATE TABLE IF NOT EXISTS `platform_device_mapping` ( `platform_id` bigint(20) NOT NULL COMMENT '所属平台ID', `device_id` bigint(20) NOT NULL COMMENT 'FastBee设备ID', `serial_number` varchar(64) NOT NULL COMMENT 'FastBee设备序列号', - `device_name` varchar(64) DEFAULT NULL COMMENT 'FastBee设备名称', + `device_name` varchar(64) DEFAULT NULL COMMENT '默认为本平台设备名称，可以修改为对方平台需要的名称', `platform_device_id` varchar(100) DEFAULT NULL COMMENT 'DCP侧deviceId，入网后自动回填', `platform_point_id` varchar(100) DEFAULT NULL COMMENT 'DCP侧pointId，入网后自动回填', `sensor_type` varchar(50) NOT NULL COMMENT '传感器类型，来自字典 platform_sensor_type', @@ -128,6 +131,9 @@ CREATE TABLE IF NOT EXISTS `platform_device_mapping` ( `field_mapping` json DEFAULT NULL COMMENT '字段映射 [{identifier,platformField,transform}]', `extra_mapping` json DEFAULT NULL COMMENT '扩展映射配置，如静力水准子设备编号与DCP sensorId映射', `net_status` tinyint(1) NOT NULL DEFAULT 0 COMMENT '入网状态(0=未入网,1=已入网)', + `push_mode` TINYINT(1) NOT NULL DEFAULT 1 COMMENT '推送模式：1=实时 2=定时（默认1）', + `push_interval` INT DEFAULT 0 COMMENT '定时推送间隔（分钟），0=不自动推送', + `last_push_time` DATETIME DEFAULT NULL COMMENT '最近推送时间', `enable` char(1) NOT NULL DEFAULT '1' COMMENT '是否启用', `remark` varchar(255) DEFAULT NULL COMMENT '备注', `create_by` varchar(64) DEFAULT '' COMMENT '创建者', @@ -147,8 +153,23 @@ ALTER TABLE `platform_device_mapping` COMMENT '设备类型: 0=直连设备 1=网关设备(多子设备)' AFTER `sensor_type`; ALTER TABLE `platform_device_mapping` - ADD COLUMN IF NOT EXISTS `device_name` varchar(64) DEFAULT NULL - COMMENT 'FastBee设备名称' AFTER `serial_number`; + ADD COLUMN `device_name` varchar(64) DEFAULT NULL COMMENT '默认为本平台设备名称，可以修改为对方平台需要的名称' AFTER `serial_number`; + +-- 添加推送模式字段 +ALTER TABLE platform_device_mapping + ADD COLUMN `push_mode` TINYINT DEFAULT 1 COMMENT '推送模式：1=实时 2=定时（默认1）' AFTER `net_status`; + +-- 添加定时推送间隔字段 +ALTER TABLE platform_device_mapping + ADD COLUMN `push_interval` INT DEFAULT 0 COMMENT '定时推送间隔（分钟），0=不自动推送' after `push_mode`; + +-- 添加最近推送时间字段 +ALTER TABLE platform_device_mapping + ADD COLUMN `last_push_time` DATETIME DEFAULT NULL COMMENT '最近推送时间' after `push_interval`; + +-- 添加索引优化查询性能 +CREATE INDEX idx_push_mode ON platform_device_mapping(push_mode); +CREATE INDEX idx_last_push_time ON platform_device_mapping(last_push_time); -- ---------------------------- -- 5.2 子设备映射（网关多子设备通用） diff --git a/yack-web/src/assets/logo/yack.png b/yack-web/src/assets/logo/yack.png index fb703c31..6640463a 100644 Binary files a/yack-web/src/assets/logo/yack.png and b/yack-web/src/assets/logo/yack.png differ diff --git a/yack-web/src/views/platform/components/InterfaceDrawer.vue b/yack-web/src/views/platform/components/InterfaceDrawer.vue index a98f5ecd..bb6da62c 100644 --- a/yack-web/src/views/platform/components/InterfaceDrawer.vue +++ b/yack-web/src/views/platform/components/InterfaceDrawer.vue @@ -27,6 +27,11 @@ + + + @@ -89,6 +94,14 @@ + + +

开启后一次请求推送全量子设备数据
+ + + + + @@ -99,7 +112,7 @@ 入参定义（request_schema）
定义接口需要的入参字段。参数来源使用业务化名称： - 固定值、认证参数、平台参数、上一步结果、当前设备、自动UUID、子设备传感器列表、基准点成对传感器列表、设备物模型、上报时间。数组字段可配置组包模式。 + 固定值、认证参数、平台参数、上一步结果、当前设备、自动UUID、子设备传感器列表、串联子设备传感器列表、串联子设备深度数组、基准点成对传感器列表、设备物模型、上报时间。数组字段可配置组包模式。
@@ -129,6 +142,8 @@ + + @@ -159,8 +174,10 @@ + 每次请求自动生成UUID 自动取设备绑定页子设备映射中的 sensorId 列表 + 按当前节点到末端节点收集 sensorId，并倒序输出 按基准点与每个监测点成对拆分请求 设备绑定页面选择物模型字段 系统自动取设备上报时间 @@ -391,7 +408,7 @@ export default { const arr = JSON.parse(json) return arr.map(f => ({ ...f, - value: f.source === 'onboard_config' ? this.stringifySchemaValue(f.value) : f.value, + value: (f.source === 'onboard_config' || f.source === 'cascadedSubDeviceDepths') ? this.stringifySchemaValue(f.value) : f.value, itemSchema: f.itemSchema || [], required: !!f.required })) @@ -480,7 +497,7 @@ export default { }) } if (r.source === 'fixed') field.value = r.value - if (r.source === 'onboard_config') field.value = this.parseSchemaValue(r.value) + if (r.source === 'onboard_config' || r.source === 'cascadedSubDeviceDepths') field.value = this.parseSchemaValue(r.value) if (r.source === 'auth' || r.source === 'extra' || r.source === 'device') field.field = r.field if (r.source === 'context') field.contextKey = r.contextKey return field diff --git a/yack-web/src/views/platform/device-mapping/index.vue b/yack-web/src/views/platform/device-mapping/index.vue index 0f7c1d7f..c853fab7 100644 --- a/yack-web/src/views/platform/device-mapping/index.vue +++ b/yack-web/src/views/platform/device-mapping/index.vue @@ -97,6 +97,16 @@ + + + @@ -490,6 +494,8 @@ export default { return `当前设备 / ${deviceLabels[field.field] || field.field || ''}` } if (field.source === 'subDeviceSensorIds') return '子设备传感器列表' + if (field.source === 'cascadedSubDeviceSensorIds') return '串联子设备传感器列表' + if (field.source === 'cascadedSubDeviceDepths') return `串联子设备深度数组 / ${this.stringifySchemaValue(field.value)}` if (field.source === 'referenceSubDeviceSensorIdPairs') return '基准点成对传感器列表' if (field.source === 'uuid') return '自动UUID' if (field.source === 'thingsModel') return '设备物模型' @@ -568,7 +574,7 @@ export default { if (!row.name || !row.source) return const item = { source: row.source } if (row.source === 'fixed') item.value = row.value - if (row.source === 'onboard_config') item.value = this.parseSchemaValue(row.value) + if (row.source === 'onboard_config' || row.source === 'cascadedSubDeviceDepths') item.value = this.parseSchemaValue(row.value) if (row.source === 'auth' || row.source === 'extra' || row.source === 'device') item.field = row.field if (row.source === 'context') item.contextKey = row.contextKey mappings[row.name] = item