diff --git a/bin/send_modbus_and_register.py b/bin/send_modbus_and_register.py index a0bfe35f..3f9626c1 100644 --- a/bin/send_modbus_and_register.py +++ b/bin/send_modbus_and_register.py @@ -8,12 +8,14 @@ TARGET_PORT = 8888 # 请根据实际情况修改 # 元数组 modbus 指令 MODBUS_DATA_HEX = ( -"7e80D55746B29F687e", -"010310c07ffffe45217c3d41dfe9964140686641f4" +"7e80D21F6C448E8D7e", + "01031c41b6dcec418b0000402d2148000000004017e0b8400c19efc2ad8a8acb2b", + "02031c41babecc419d8000402be3560000000040137c30bfb2f4d0c2ae9bc1b582", + "03031c41bd1c1941810000402bb38600000000bf4501bdc027dfcac2ae8853eaef", + "04031c41babecc419d8000402be3560000000040137c30bfb2f4d0c2ae9bc1D581", + "05031c41bd1c1941810000402bb38600000000bf4501bdc027dfcac2ae88538AEC" ) -# "01031c4268442f41d30000418b6685000000004026f4cac00d1fd3c2ad2a9d6311", -# "02031c42488fd041c5800041865ab7000000003f364458c03f765ec2add9722da5", -# "03031c42ffdad541c78000421529b8000000003d5f38f94082080dc2abdf4f1f94" + # 01 03 1c 4268 442f 41d3 0000 418b 6685 0000 0000 4026 f4ca c00d 1fd3 c2ad 2a9d 6311 # 从机地址 功能码 字节长度 2个字节 CRC校验 @@ -45,7 +47,7 @@ def main(): # 2. 等待0.5秒 print("等待 0.1 秒...") - time.sleep(0.5) + time.sleep(1) # (可选)接收最终响应 try: diff --git a/fastbee-common/src/main/java/com/fastbee/common/utils/gateway/CRC16Utils.java b/fastbee-common/src/main/java/com/fastbee/common/utils/gateway/CRC16Utils.java index 51525bcc..f0fdc6ae 100644 --- a/fastbee-common/src/main/java/com/fastbee/common/utils/gateway/CRC16Utils.java +++ b/fastbee-common/src/main/java/com/fastbee/common/utils/gateway/CRC16Utils.java @@ -133,7 +133,7 @@ public class CRC16Utils { public static void main(String[] args) throws Exception { - String hex = "0103028000"; + String hex = "006000000000"; byte[] bytes = ByteUtils.hexToByte(hex); String crc = getCRC(bytes); System.out.println(crc); diff --git a/fastbee-common/src/main/java/com/fastbee/common/utils/gateway/protocol/ByteUtils.java b/fastbee-common/src/main/java/com/fastbee/common/utils/gateway/protocol/ByteUtils.java index c8e0a69a..c6ce73e0 100644 --- a/fastbee-common/src/main/java/com/fastbee/common/utils/gateway/protocol/ByteUtils.java +++ b/fastbee-common/src/main/java/com/fastbee/common/utils/gateway/protocol/ByteUtils.java @@ -303,9 +303,9 @@ public class ByteUtils { return data.getBytes(charset); } - /* + /** * 把16进制字符串转换成字节数组 - * @param hex + * @param hexStr 16进制字符 * @return */ public static byte[] hexToByte(String hexStr) { 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 635167e4..1ad9b427 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 @@ -19,6 +19,7 @@ import com.fastbee.iot.platform.flow.FlowContext; import com.fastbee.iot.platform.flow.PlatformFlowExecutor; import com.fastbee.iot.platform.mapper.PlatformFlowMapper; 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 lombok.extern.slf4j.Slf4j; @@ -34,6 +35,7 @@ import java.time.format.DateTimeFormatter; import java.time.format.DateTimeParseException; import java.util.Date; import java.util.HashMap; +import java.util.List; import java.util.Map; /** @@ -66,6 +68,8 @@ public class DcpPlatformAdapter { private StaticLevelBufferService bufferService; @Autowired private PlatformInterfaceMapper interfaceMapper; + @Autowired + private IPlatformDeviceSubMappingService subMappingService; /** 传感器类型 → 推送接口路径 */ private static final Map PUSH_PATH = new HashMap<>(); @@ -184,10 +188,12 @@ public class DcpPlatformAdapter { JSONObject currentBodyFields = buildObjectBySchema(platform, mapping, requestSchema, reportData, fieldMappings); if (subMapping.isReference()) { - // 基准点上报:仅缓存,不推送 + // 基准点上报:缓存当前基准点,并补推此前已缓存的监测点 bufferService.saveReferenceData( platform.getId(), fieldName, reportSerialNumber, currentItem, reportIntervalSec); - log.info("[DCP] 基准点 {} 已缓存,等待监测点上报", reportSerialNumber); + int replayCount = replayPendingMonitorData(platform, mapping, fieldName, currentItem, + reportIntervalSec, baseUrl, interfaceCode); + log.info("[DCP] 基准点 {} 已缓存,补推监测点 {} 条", reportSerialNumber, replayCount); return; } @@ -198,7 +204,7 @@ public class DcpPlatformAdapter { return; } - // 获取基准点序列号(从 onboard_config 中取 refSensorId 对应的原始序列号) + // 获取基准点序列号,优先使用子映射中当前标记的 reference 行 String refSerialNumber = resolveRefSerialNumber(subMapping, mapping); JSONObject refData = bufferService.getReferenceDataIfFresh( @@ -243,12 +249,18 @@ public class DcpPlatformAdapter { return false; } - /** - * 从子映射的 onboard_config 解析基准点序列号 - */ + /** 解析基准点序列号 */ private String resolveRefSerialNumber(PlatformDeviceSubMapping subMapping, PlatformDeviceMapping mapping) { - // 从 onboard_config 取 refSensorId 对应的子设备序列号 - // 通常基准点的 sensorId == 基准点的 serialNumber + try { + PlatformDeviceSubMapping reference = subMappingService.findReference(mapping.getId()); + if (reference != null && StringUtils.hasText(reference.getSubSerialNumber())) { + return reference.getSubSerialNumber(); + } + } catch (Exception e) { + log.warn("[DCP] 查询静力水准基准点失败 mappingId={}, error={}", mapping.getId(), e.getMessage()); + } + + // 兼容历史配置:从 onboard_config 取基准点原始序列号 if (StringUtils.hasText(subMapping.getOnboardConfig())) { try { JSONObject config = JSON.parseObject(subMapping.getOnboardConfig()); @@ -262,6 +274,67 @@ public class DcpPlatformAdapter { return mapping.getSerialNumber(); } + /** + * 基准点后到时,补推此前缓存的监测点数据。 + */ + private int replayPendingMonitorData(PlatformInstance platform, PlatformDeviceMapping mapping, + String fieldName, JSONObject referenceItem, + int reportIntervalSec, String baseUrl, String interfaceCode) { + List subMappings = subMappingService.selectByParentId(mapping.getId()); + if (subMappings == null || subMappings.isEmpty()) { + return 0; + } + + int replayCount = 0; + for (PlatformDeviceSubMapping monitor : subMappings) { + if (monitor == null || !monitor.isMonitor() + || !StringUtils.hasText(monitor.getSubSerialNumber())) { + continue; + } + JSONObject monitorData = bufferService.getMonitorDataIfFresh( + platform.getId(), fieldName, monitor.getSubSerialNumber(), reportIntervalSec); + if (monitorData == null) { + continue; + } + if (!StringUtils.hasText(monitor.getPlatformDeviceId())) { + log.warn("[DCP] 监测点 {} 有缓存但未分配平台设备ID,暂不补推", + monitor.getSubSerialNumber()); + continue; + } + + JSONObject monitorItem = monitorData.getJSONObject("itemData"); + if (monitorItem == null || monitorItem.isEmpty()) { + bufferService.removeMonitorData(platform.getId(), fieldName, monitor.getSubSerialNumber()); + continue; + } + JSONObject bodyFields = monitorData.getJSONObject("bodyFields"); + if (bodyFields == null) { + bodyFields = new JSONObject(); + } + + JSONArray sensorData = new JSONArray(); + sensorData.add(referenceItem); + sensorData.add(monitorItem); + JSONArray pushBody = buildArrayPushBody(mapping, bodyFields, fieldName, + sensorData, monitor.getPlatformDeviceId()); + + PlatformPushLog logEntry = createLogEntry(platform, mapping, interfaceCode, + monitor.getSubSerialNumber()); + doHttpPost(platform, baseUrl, JSON.toJSONString(pushBody), logEntry); + pushLogService.asyncSave(logEntry); + if (logEntry.getStatus() != null && logEntry.getStatus() == 1) { + bufferService.removeMonitorData(platform.getId(), fieldName, monitor.getSubSerialNumber()); + replayCount++; + log.info("[DCP] 基准点后到,已补推监测点 {} deviceId={}", + monitor.getSubSerialNumber(), monitor.getPlatformDeviceId()); + } else { + log.warn("[DCP] 基准点后到,补推监测点 {} 失败,保留缓存等待下次基准点上报", + monitor.getSubSerialNumber()); + } + } + return replayCount; + } + // ==================== 普通推送 ==================== private void handleNormalPush(PlatformInstance platform, PlatformDeviceMapping mapping, diff --git a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/flow/FlowContext.java b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/flow/FlowContext.java index 9263bbe0..4dd879af 100644 --- a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/flow/FlowContext.java +++ b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/flow/FlowContext.java @@ -21,6 +21,10 @@ public class FlowContext { private String currentMonitorSerialNumber; /** 网关子设备入网时,当前子映射行的 onboard_config JSON(平台相关参数) */ private String subMappingOnboardConfig; + /** 网关子设备入网时,当前基准点序列号 */ + private String currentReferenceSerialNumber; + /** 网关子设备入网时,基准点子映射行的 onboard_config JSON */ + private String referenceSubMappingOnboardConfig; public FlowContext(PlatformInstance platform, PlatformDeviceMapping deviceMapping, String payload) { this.platform = platform; @@ -64,6 +68,22 @@ public class FlowContext { this.subMappingOnboardConfig = subMappingOnboardConfig; } + public String getCurrentReferenceSerialNumber() { + return currentReferenceSerialNumber; + } + + public void setCurrentReferenceSerialNumber(String currentReferenceSerialNumber) { + this.currentReferenceSerialNumber = currentReferenceSerialNumber; + } + + public String getReferenceSubMappingOnboardConfig() { + return referenceSubMappingOnboardConfig; + } + + public void setReferenceSubMappingOnboardConfig(String referenceSubMappingOnboardConfig) { + this.referenceSubMappingOnboardConfig = referenceSubMappingOnboardConfig; + } + @SuppressWarnings("unchecked") public Object getFromContext(String contextKey) { // contextKey 格式: "step1.access_token" 或 "step1" 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 f2e05633..546316d2 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 @@ -429,6 +429,20 @@ public class PlatformFlowExecutor { } private List collectSubDeviceSensorIds(FlowContext ctx) { + if (StringUtils.hasText(ctx.getCurrentMonitorSerialNumber()) + && StringUtils.hasText(ctx.getCurrentReferenceSerialNumber())) { + String refSensorId = resolveOnboardSensorId( + ctx.getReferenceSubMappingOnboardConfig(), ctx.getCurrentReferenceSerialNumber()); + String monitorSensorId = resolveOnboardSensorId( + ctx.getSubMappingOnboardConfig(), ctx.getCurrentMonitorSerialNumber()); + List ids = new ArrayList<>(); + if (StringUtils.hasText(refSensorId)) ids.add(refSensorId); + if (StringUtils.hasText(monitorSensorId)) ids.add(monitorSensorId); + if (!ids.isEmpty()) { + return ids.stream().distinct().collect(Collectors.toList()); + } + } + // 网关入网时,从子表 onboard_config 获取 sensorId if (StringUtils.hasText(ctx.getSubMappingOnboardConfig())) { try { @@ -450,10 +464,32 @@ public class PlatformFlowExecutor { } /** - * 网关入网时:从当前子映射的 onboard_config 取 [refSensorId, sensorId] 一对。 + * 网关入网时:优先按子表中当前 reference 行取 [基准点sensorId, 监测点sensorId] 一对。 * 直连设备/旧逻辑:从 extra_mapping 解析所有配对。 */ private List> collectReferenceSubDeviceSensorIdPairs(FlowContext ctx) { + if (StringUtils.hasText(ctx.getCurrentMonitorSerialNumber())) { + String sensorId = resolveOnboardSensorId( + ctx.getSubMappingOnboardConfig(), ctx.getCurrentMonitorSerialNumber()); + String refSensorId = resolveOnboardSensorId( + ctx.getReferenceSubMappingOnboardConfig(), ctx.getCurrentReferenceSerialNumber()); + if (!StringUtils.hasText(refSensorId)) { + refSensorId = resolveOnboardConfigText(ctx.getSubMappingOnboardConfig(), "refSensorId"); + } + if (StringUtils.hasText(sensorId) && StringUtils.hasText(refSensorId)) { + if (sameText(refSensorId, sensorId)) { + throw new RuntimeException("静力水准 sensorIds 配置无效:基准点和监测点 sensorId 相同"); + } + List> pairs = new ArrayList<>(); + pairs.add(Arrays.asList(refSensorId, sensorId)); + return pairs; + } + if (isStaticLevel(ctx.getDeviceMapping())) { + throw new RuntimeException("静力水准监测点 " + ctx.getCurrentMonitorSerialNumber() + + " 缺少基准点或监测点 sensorId,请检查子设备映射"); + } + } + // 网关入网:onboardConfig 已指定当前监测点 if (StringUtils.hasText(ctx.getSubMappingOnboardConfig())) { try { @@ -494,6 +530,27 @@ public class PlatformFlowExecutor { return pairs; } + private String resolveOnboardSensorId(String onboardConfig, String fallbackSerialNumber) { + String sensorId = resolveOnboardConfigText(onboardConfig, "sensorId"); + return StringUtils.hasText(sensorId) ? sensorId : normalizeText(fallbackSerialNumber); + } + + private String resolveOnboardConfigText(String onboardConfig, String field) { + if (!StringUtils.hasText(onboardConfig) || !StringUtils.hasText(field)) { + return null; + } + try { + JSONObject config = JSON.parseObject(onboardConfig); + return normalizeText(config.getString(field)); + } catch (Exception ignored) { + return null; + } + } + + private boolean isStaticLevel(PlatformDeviceMapping mapping) { + return mapping != null && Objects.equals("static_level", mapping.getSensorType()); + } + private List collectSubDeviceSensorBindings(PlatformDeviceMapping mapping) { if (mapping == null) { return Collections.emptyList(); diff --git a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/service/impl/PlatformDeviceMappingServiceImpl.java b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/service/impl/PlatformDeviceMappingServiceImpl.java index 9998485c..4cd95611 100644 --- a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/service/impl/PlatformDeviceMappingServiceImpl.java +++ b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/service/impl/PlatformDeviceMappingServiceImpl.java @@ -21,6 +21,7 @@ import org.springframework.util.StringUtils; import java.util.ArrayList; import java.util.List; +import java.util.Objects; import java.util.stream.Collectors; @Slf4j @@ -105,6 +106,7 @@ public class PlatformDeviceMappingServiceImpl extends ServiceImpl subMappings = subMappingService.selectByParentId(mapping.getId()); + PlatformDeviceSubMapping reference = resolveGatewayReference(mapping, subMappings); List monitors = subMappings.stream() .filter(s -> s.isMonitor() && (s.getNetStatus() == null || s.getNetStatus() == 0)) .collect(Collectors.toList()); @@ -120,6 +122,10 @@ public class PlatformDeviceMappingServiceImpl extends ServiceImpl subMappings) { + List references = subMappings.stream() + .filter(PlatformDeviceSubMapping::isReference) + .collect(Collectors.toList()); + boolean staticLevel = Objects.equals("static_level", mapping.getSensorType()); + if (references.size() > 1) { + throw new RuntimeException("子表中只能指定一个基准点,请检查子设备映射"); + } + if (references.isEmpty()) { + if (staticLevel) { + throw new RuntimeException("静力水准网关入网需在子设备映射中指定一个基准点"); + } + return null; + } + return references.get(0); + } } diff --git a/fastbee-service/fastbee-iot-service/src/main/resources/mapper/iot/platform/PlatformDeviceSubMappingMapper.xml b/fastbee-service/fastbee-iot-service/src/main/resources/mapper/iot/platform/PlatformDeviceSubMappingMapper.xml index 7f34adf9..783cc576 100644 --- a/fastbee-service/fastbee-iot-service/src/main/resources/mapper/iot/platform/PlatformDeviceSubMappingMapper.xml +++ b/fastbee-service/fastbee-iot-service/src/main/resources/mapper/iot/platform/PlatformDeviceSubMappingMapper.xml @@ -5,7 +5,7 @@