diff --git a/intc-modules/intc-fishery/src/main/java/com/intc/fishery/controller/DeviceController.java b/intc-modules/intc-fishery/src/main/java/com/intc/fishery/controller/DeviceController.java index 8b60985..81160ed 100644 --- a/intc-modules/intc-fishery/src/main/java/com/intc/fishery/controller/DeviceController.java +++ b/intc-modules/intc-fishery/src/main/java/com/intc/fishery/controller/DeviceController.java @@ -80,6 +80,16 @@ public class DeviceController extends BaseController { return deviceService.queryPageList(bo, pageQuery); } + /** + * 查询太阳能网控设备列表 + */ + @SaCheckPermission("fishery:solarControl:list") + @GetMapping("/solar/list") + public TableDataInfo solarList(DeviceBo bo, PageQuery pageQuery) { + bo.setDeviceType(3); + return deviceService.queryPageList(bo, pageQuery); + } + /** * 导出设备管理列表 */ diff --git a/intc-modules/intc-iot/src/main/java/com/intc/iot/controller/IotController.java b/intc-modules/intc-iot/src/main/java/com/intc/iot/controller/IotController.java index 41f9282..63897e4 100644 --- a/intc-modules/intc-iot/src/main/java/com/intc/iot/controller/IotController.java +++ b/intc-modules/intc-iot/src/main/java/com/intc/iot/controller/IotController.java @@ -32,6 +32,7 @@ import com.intc.iot.domain.bo.AddDeviceDetectorBo; import com.intc.iot.domain.bo.DeviceCalibrateBo; import com.intc.iot.domain.bo.DeviceRealtimeDataBo; import com.intc.iot.domain.bo.DeviceSalinityCompensationBo; +import com.intc.iot.domain.bo.DeviceSolarPropertyCommandBo; import com.intc.iot.domain.bo.DeviceVoltageTypeBo; import com.intc.iot.domain.vo.DeviceRealtimeDataVo; import com.intc.iot.service.DeviceDataService; @@ -66,8 +67,11 @@ import java.time.LocalDateTime; import java.time.ZoneId; import java.time.format.DateTimeFormatter; import java.time.Duration; +import java.math.BigDecimal; +import java.util.LinkedHashMap; import java.util.List; import java.util.Map; +import java.util.Set; /** * 阿里云生活物联网平台(飞燕平台)控制器 @@ -82,6 +86,31 @@ import java.util.Map; public class IotController extends BaseController { private static final DateTimeFormatter DATETIME_FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"); + private static final int DEVICE_TYPE_SOLAR_CONTROLLER = 3; + private static final int DEVICE_STATUS_ONLINE = 1; + private static final int DEVICE_STATUS_OFFLINE = 3; + private static final Set SOLAR_WRITABLE_PROPERTY_KEYS = Set.of( + "correct", + "salinitySet", + "ammonia_nitrogen", + "phasedif_compensation", + "phase_compensation", + "temperature_compensation", + "stdsatphase", + "stdphaseinterval", + "SaturationPoint", + "AnaerobicPoint", + "LowtempCOEF", + "HightempCOEF", + "updatatime", + "PowerMode", + "WakeUpTime", + "BootToSleepTimeout", + "CorrectCharge", + "TempPobeOffSet1", + "TempPobeOffSet2", + "FirstBootToSleepTimeout" + ); private final DeviceTypeProperties deviceTypeProperties; @@ -2043,7 +2072,15 @@ public class IotController extends BaseController { boolean isSolarDevice = Integer.valueOf(3).equals(device.getDeviceType()); if (isSolarDevice) { - if (!cachePendingSolarSalinity(device, request.getSalinityCompensation())) { + if (device.getIotId() == null || device.getIotId().isBlank()) { + return R.fail("设备 iotId 为空，无法缓存盐度指令"); + } + if (device.getSerialNum() == null || device.getSerialNum().isBlank()) { + return R.fail("设备编号为空，无法缓存盐度指令"); + } + Map properties = new LinkedHashMap<>(); + properties.put("salinitySet", request.getSalinityCompensation()); + if (!cachePendingSolarProperties(device, properties)) { return R.fail("盐度缓存失败，请稍后重试"); } updateDeviceSalinityCompensation(device.getId(), request.getSalinityCompensation()); @@ -2083,7 +2120,6 @@ public class IotController extends BaseController { } updateDeviceSalinityCompensation(device.getId(), request.getSalinityCompensation()); - clearPendingSolarSalinity(device.getSerialNum()); log.info("设置盐度补偿成功: userId={}, deviceId={}, deviceName={}, serialNum={}, salinityCompensation={}‰", userId, device.getId(), device.getDeviceName(), device.getSerialNum(), request.getSalinityCompensation()); @@ -2103,34 +2139,6 @@ public class IotController extends BaseController { return setResult != null && Boolean.TRUE.equals(setResult.get("success")); } - private boolean cachePendingSolarSalinity(Device device, Double salinityCompensation) { - if (device.getSerialNum() == null || device.getSerialNum().isBlank()) { - log.warn("太阳能网控盐度缓存失败: deviceId={}, serialNum为空", device.getId()); - return false; - } - try { - Map cacheValue = new java.util.HashMap<>(); - cacheValue.put("deviceId", device.getId()); - cacheValue.put("iotId", device.getIotId()); - cacheValue.put("serialNum", device.getSerialNum()); - cacheValue.put("salinityCompensation", salinityCompensation); - cacheValue.put("updatedAt", System.currentTimeMillis()); - String cacheJson = cn.hutool.json.JSONUtil.toJsonStr(cacheValue); - String cacheKey = DeviceDataHandler.PENDING_SOLAR_SALINITY_KEY_PREFIX + device.getSerialNum(); - // 忽略租户前缀：HTTP线程写入带租户前缀，AMQP消费线程读取不带前缀，需统一 - TenantHelper.ignore(() -> - com.intc.common.redis.utils.RedisUtils.setCacheObject(cacheKey, cacheJson, Duration.ofDays(7)) - ); - log.info("太阳能网控盐度已缓存，等待上线补发: deviceId={}, serialNum={}, salinityCompensation={}", - device.getId(), device.getSerialNum(), salinityCompensation); - return true; - } catch (Exception e) { - log.error("缓存太阳能网控盐度失败: deviceId={}, serialNum={}, err={}", - device.getId(), device.getSerialNum(), e.getMessage(), e); - return false; - } - } - private void updateDeviceSalinityCompensation(Long deviceId, Double salinityCompensation) { deviceMapper.update(null, new com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper() @@ -2139,15 +2147,398 @@ public class IotController extends BaseController { ); } - private void clearPendingSolarSalinity(String serialNum) { - if (serialNum == null || serialNum.isBlank()) { - return; + @Operation(summary = "太阳能网控属性下发") + @SaCheckPermission("fishery:solarControl:command") + @PostMapping("/device/solar/property-command") + public R> setSolarPropertyCommand(@Validated @RequestBody DeviceSolarPropertyCommandBo request) { + try { + Long userId = LoginHelper.getUserId(); + if (userId == null) { + return R.fail("未登录或登录已过期"); + } + + if (deviceMapper == null) { + return R.fail("设备服务未启用"); + } + + Device device = deviceMapper.selectOne( + new com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper() + .eq(Device::getId, request.getId()) + .select(Device::getId, Device::getUserId, Device::getIotId, Device::getSerialNum, + Device::getDeviceName, Device::getDeviceType, Device::getDeadTime) + ); + if (device == null) { + return R.fail("设备不存在"); + } + if (!Integer.valueOf(DEVICE_TYPE_SOLAR_CONTROLLER).equals(device.getDeviceType())) { + return R.fail("仅支持太阳能网控设备"); + } + if (device.getDeadTime() != null && new Date().after(device.getDeadTime())) { + return R.fail("设备已到期"); + } + if (device.getIotId() == null || device.getIotId().isBlank()) { + return R.fail("设备 iotId 为空，无法下发指令"); + } + if (device.getSerialNum() == null || device.getSerialNum().isBlank()) { + return R.fail("设备编号为空，无法缓存指令"); + } + + Map properties = normalizeSolarProperties(request.getProperties()); + boolean cache = Boolean.TRUE.equals(request.getCache()); + Integer statusCode = queryDeviceStatusCode(device.getIotId()); + boolean online = Integer.valueOf(DEVICE_STATUS_ONLINE).equals(statusCode); + + Map result = new LinkedHashMap<>(); + result.put("deviceId", device.getId()); + result.put("serialNum", device.getSerialNum()); + result.put("online", online); + result.put("statusCode", statusCode); + result.put("properties", properties); + + if (cache) { + if (online) { + return R.fail("设备当前在线，请使用直接下发"); + } + if (!cachePendingSolarProperties(device, properties)) { + return R.fail("指令缓存失败，请稍后重试"); + } + syncDeviceFieldsAfterSolarCommand(device.getId(), properties); + result.put("mode", "cached"); + result.put("cached", true); + result.put("sent", false); + MessageOpRecordUtil.addMessageOpRecordUser( + device.getUserId(), + userId, + "设备操作", + String.format("%s(%s)，太阳能网控属性缓存下发：%s。", + device.getDeviceName(), device.getSerialNum(), properties) + ); + log.info("太阳能网控属性已缓存: userId={}, deviceId={}, serialNum={}, properties={}", + userId, device.getId(), device.getSerialNum(), properties); + return R.ok("缓存成功，设备上线后自动下发", result); + } + + if (!online) { + return R.fail("设备离线或未激活，请使用缓存下发"); + } + if (iotCloudService == null) { + return R.fail("飞燕平台配置未启用"); + } + + boolean success = iotCloudService.setProperty(device.getIotId(), properties, true, 2); + if (!success) { + return R.fail("属性下发失败，请重试"); + } + + syncDeviceFieldsAfterSolarCommand(device.getId(), properties); + removePendingSolarProperties(device, properties.keySet()); + + result.put("mode", "direct"); + result.put("cached", false); + result.put("sent", true); + MessageOpRecordUtil.addMessageOpRecordUser( + device.getUserId(), + userId, + "设备操作", + String.format("%s(%s)，太阳能网控属性直接下发：%s。", + device.getDeviceName(), device.getSerialNum(), properties) + ); + log.info("太阳能网控属性下发成功: userId={}, deviceId={}, serialNum={}, properties={}", + userId, device.getId(), device.getSerialNum(), properties); + return R.ok("下发成功", result); + } catch (IllegalArgumentException e) { + return R.fail(e.getMessage()); + } catch (Exception e) { + log.error("太阳能网控属性下发失败: {}", e.getMessage(), e); + return R.fail("太阳能网控属性下发失败: " + e.getMessage()); } - TenantHelper.ignore(() -> - com.intc.common.redis.utils.RedisUtils.deleteObject(DeviceDataHandler.PENDING_SOLAR_SALINITY_KEY_PREFIX + serialNum) - ); } + private Map normalizeSolarProperties(Map source) { + if (source == null || source.isEmpty()) { + throw new IllegalArgumentException("下发属性不能为空"); + } + Map properties = new LinkedHashMap<>(); + for (Map.Entry entry : source.entrySet()) { + String identifier = entry.getKey(); + if (identifier == null || identifier.isBlank()) { + throw new IllegalArgumentException("属性标识符不能为空"); + } + if (!SOLAR_WRITABLE_PROPERTY_KEYS.contains(identifier)) { + throw new IllegalArgumentException("属性不支持下发: " + identifier); + } + properties.put(identifier, normalizeSolarPropertyValue(identifier, entry.getValue())); + } + return properties; + } + + private Object normalizeSolarPropertyValue(String identifier, Object value) { + if (value == null || (value instanceof String str && str.isBlank())) { + throw new IllegalArgumentException(identifier + " 的值不能为空"); + } + return switch (identifier) { + case "correct" -> { + Integer intValue = parseInteger(value, identifier); + if (!Set.of(1, 2, 11, 12, 13, 14, 31, 32).contains(intValue)) { + throw new IllegalArgumentException("校准指令值不合法"); + } + yield intValue; + } + case "PowerMode" -> requireIntegerInRange(value, identifier, 0, 2); + case "updatatime" -> requireIntegerInRange(value, identifier, 0, 65535); + case "WakeUpTime" -> requireIntegerInRange(value, identifier, 1, Integer.MAX_VALUE); + case "BootToSleepTimeout" -> requireIntegerInRange(value, identifier, 2, Integer.MAX_VALUE); + case "FirstBootToSleepTimeout" -> requireIntegerInRange(value, identifier, 3, Integer.MAX_VALUE); + case "CorrectCharge" -> normalizeCorrectCharge(value); + case "salinitySet" -> requireDecimalInRange(value, identifier, "0", "100"); + case "ammonia_nitrogen" -> requireDecimalInRange(value, identifier, "0", "1000"); + case "phasedif_compensation" -> requireDecimalInRange(value, identifier, "-10", "100"); + case "phase_compensation" -> requireDecimalInRange(value, identifier, "0", "10"); + case "temperature_compensation" -> requireDecimalInRange(value, identifier, "-10", "10"); + case "stdsatphase", "stdphaseinterval" -> requireDecimalInRange(value, identifier, "0", "360"); + case "SaturationPoint", "AnaerobicPoint", "LowtempCOEF", "HightempCOEF" -> + requireDecimalInRange(value, identifier, "-1000", "1000"); + case "TempPobeOffSet1", "TempPobeOffSet2" -> requireDecimalInRange(value, identifier, "-12.7", "12.7"); + default -> throw new IllegalArgumentException("属性不支持下发: " + identifier); + }; + } + + @SuppressWarnings("unchecked") + private Map normalizeCorrectCharge(Object value) { + if (!(value instanceof Map rawMap)) { + throw new IllegalArgumentException("电量校准参数格式不正确"); + } + Map map = (Map) rawMap; + Map result = new LinkedHashMap<>(); + result.put("lowvoltage", requireDecimalInRange(map.get("lowvoltage"), "lowvoltage", "0", "10")); + result.put("highvoltage", requireDecimalInRange(map.get("highvoltage"), "highvoltage", "0", "10")); + return result; + } + + private Integer requireIntegerInRange(Object value, String identifier, int min, int max) { + Integer intValue = parseInteger(value, identifier); + if (intValue < min || intValue > max) { + throw new IllegalArgumentException(identifier + " 超出范围: " + min + " - " + max); + } + return intValue; + } + + private Integer parseInteger(Object value, String identifier) { + try { + if (value instanceof Number number) { + double doubleValue = number.doubleValue(); + if (doubleValue % 1 != 0) { + throw new IllegalArgumentException(identifier + " 必须为整数"); + } + return number.intValue(); + } + return Integer.valueOf(value.toString()); + } catch (IllegalArgumentException e) { + throw e; + } catch (Exception e) { + throw new IllegalArgumentException(identifier + " 必须为整数"); + } + } + + private Double requireDecimalInRange(Object value, String identifier, String min, String max) { + BigDecimal decimal = parseDecimal(value, identifier); + BigDecimal minDecimal = new BigDecimal(min); + BigDecimal maxDecimal = new BigDecimal(max); + if (decimal.compareTo(minDecimal) < 0 || decimal.compareTo(maxDecimal) > 0) { + throw new IllegalArgumentException(identifier + " 超出范围: " + min + " - " + max); + } + return decimal.doubleValue(); + } + + private BigDecimal parseDecimal(Object value, String identifier) { + if (value == null || (value instanceof String str && str.isBlank())) { + throw new IllegalArgumentException(identifier + " 的值不能为空"); + } + try { + return new BigDecimal(value.toString()); + } catch (Exception e) { + throw new IllegalArgumentException(identifier + " 必须为数字"); + } + } + + private Integer queryDeviceStatusCode(String iotId) { + if (iotDeviceService == null || iotId == null || iotId.isBlank()) { + return null; + } + try { + Map deviceDetail = iotDeviceService.queryDeviceInfo(iotId); + return extractDeviceStatusCode(deviceDetail); + } catch (Exception e) { + log.warn("查询设备状态失败: iotId={}, err={}", iotId, e.getMessage()); + return null; + } + } + + private Integer extractDeviceStatusCode(Map deviceDetail) { + if (deviceDetail == null || deviceDetail.get("data") == null) { + return 0; + } + Object detailData = deviceDetail.get("data"); + try { + java.lang.reflect.Method getStatusMethod = detailData.getClass().getMethod("getStatus"); + Object statusObj = getStatusMethod.invoke(detailData); + if (statusObj == null) { + return 0; + } + String statusStr = statusObj.toString(); + if ("ONLINE".equalsIgnoreCase(statusStr) || "online".equalsIgnoreCase(statusStr)) { + return DEVICE_STATUS_ONLINE; + } + if ("OFFLINE".equalsIgnoreCase(statusStr) || "offline".equalsIgnoreCase(statusStr)) { + return DEVICE_STATUS_OFFLINE; + } + if ("DISABLE".equalsIgnoreCase(statusStr) || "disable".equalsIgnoreCase(statusStr)) { + return 8; + } + return 0; + } catch (Exception ex) { + log.warn("无法从 SDK 对象中获取 status: {}", ex.getMessage()); + return 0; + } + } + + private boolean cachePendingSolarProperties(Device device, Map properties) { + try { + Map mergedProperties = new LinkedHashMap<>(); + String cacheKey = DeviceDataHandler.PENDING_SOLAR_PROPERTY_KEY_PREFIX + device.getSerialNum(); + + TenantHelper.ignore(() -> { + Object rawPending = com.intc.common.redis.utils.RedisUtils.getCacheObject(cacheKey); + if (rawPending != null) { + try { + cn.hutool.json.JSONObject pending = cn.hutool.json.JSONUtil.parseObj(rawPending); + Object propertiesObj = pending.get("properties"); + if (propertiesObj != null) { + cn.hutool.json.JSONObject propertyJson = cn.hutool.json.JSONUtil.parseObj(propertiesObj); + for (String key : propertyJson.keySet()) { + mergedProperties.put(key, propertyJson.get(key)); + } + } + } catch (Exception e) { + log.warn("解析旧太阳能属性缓存失败，将用新值覆盖: serialNum={}, err={}", + device.getSerialNum(), e.getMessage()); + } + } + }); + + mergedProperties.putAll(properties); + Map cacheValue = new LinkedHashMap<>(); + cacheValue.put("deviceId", device.getId()); + cacheValue.put("iotId", device.getIotId()); + cacheValue.put("serialNum", device.getSerialNum()); + cacheValue.put("properties", mergedProperties); + cacheValue.put("updatedAt", System.currentTimeMillis()); + String cacheJson = cn.hutool.json.JSONUtil.toJsonStr(cacheValue); + TenantHelper.ignore(() -> + com.intc.common.redis.utils.RedisUtils.setCacheObject(cacheKey, cacheJson, Duration.ofDays(7)) + ); + return true; + } catch (Exception e) { + log.error("缓存太阳能网控属性失败: deviceId={}, serialNum={}, err={}", + device.getId(), device.getSerialNum(), e.getMessage(), e); + return false; + } + } + + private void removePendingSolarProperties(Device device, Set identifiers) { + if (device == null || device.getSerialNum() == null || device.getSerialNum().isBlank() + || identifiers == null || identifiers.isEmpty()) { + return; + } + try { + String cacheKey = DeviceDataHandler.PENDING_SOLAR_PROPERTY_KEY_PREFIX + device.getSerialNum(); + TenantHelper.ignore(() -> { + Object rawPending = com.intc.common.redis.utils.RedisUtils.getCacheObject(cacheKey); + if (rawPending == null) { + return; + } + + cn.hutool.json.JSONObject pending = cn.hutool.json.JSONUtil.parseObj(rawPending); + Object propertiesObj = pending.get("properties"); + if (propertiesObj == null) { + com.intc.common.redis.utils.RedisUtils.deleteObject(cacheKey); + return; + } + + cn.hutool.json.JSONObject propertyJson = cn.hutool.json.JSONUtil.parseObj(propertiesObj); + Map remainingProperties = new LinkedHashMap<>(); + for (String key : propertyJson.keySet()) { + if (!identifiers.contains(key)) { + remainingProperties.put(key, propertyJson.get(key)); + } + } + + if (remainingProperties.isEmpty()) { + com.intc.common.redis.utils.RedisUtils.deleteObject(cacheKey); + return; + } + + Map cacheValue = new LinkedHashMap<>(); + cacheValue.put("deviceId", device.getId()); + cacheValue.put("iotId", device.getIotId()); + cacheValue.put("serialNum", device.getSerialNum()); + cacheValue.put("properties", remainingProperties); + cacheValue.put("updatedAt", System.currentTimeMillis()); + com.intc.common.redis.utils.RedisUtils.setCacheObject( + cacheKey, + cn.hutool.json.JSONUtil.toJsonStr(cacheValue), + Duration.ofDays(7) + ); + }); + } catch (Exception e) { + log.warn("清理太阳能网控已下发属性缓存失败: serialNum={}, identifiers={}, err={}", + device.getSerialNum(), identifiers, e.getMessage()); + } + } + + private void syncDeviceFieldsAfterSolarCommand(Long deviceId, Map properties) { + if (deviceId == null || properties == null || properties.isEmpty()) { + return; + } + Device updateDevice = new Device(); + updateDevice.setId(deviceId); + boolean needUpdate = false; + + if (properties.containsKey("salinitySet")) { + updateDevice.setSalinityCompensation(toDouble(properties.get("salinitySet"))); + needUpdate = true; + } + if (properties.containsKey("phase_compensation")) { + updateDevice.setPhaseCompensation(toDouble(properties.get("phase_compensation"))); + needUpdate = true; + } + if (properties.containsKey("phasedif_compensation")) { + updateDevice.setPhasedifCompensation(toDouble(properties.get("phasedif_compensation"))); + needUpdate = true; + } + if (properties.containsKey("temperature_compensation")) { + updateDevice.setTemperatureCompensation(toDouble(properties.get("temperature_compensation"))); + needUpdate = true; + } + + if (needUpdate) { + deviceMapper.updateById(updateDevice); + } + } + + private Double toDouble(Object value) { + if (value == null) { + return null; + } + if (value instanceof Number number) { + return number.doubleValue(); + } + try { + return Double.valueOf(value.toString()); + } catch (Exception e) { + return null; + } + } /** * 修改设备输入电压类型 diff --git a/intc-modules/intc-iot/src/main/java/com/intc/iot/domain/bo/DeviceSolarPropertyCommandBo.java b/intc-modules/intc-iot/src/main/java/com/intc/iot/domain/bo/DeviceSolarPropertyCommandBo.java new file mode 100644 index 0000000..4f6cdb7 --- /dev/null +++ b/intc-modules/intc-iot/src/main/java/com/intc/iot/domain/bo/DeviceSolarPropertyCommandBo.java @@ -0,0 +1,43 @@ +package com.intc.iot.domain.bo; + +import io.swagger.v3.oas.annotations.media.Schema; +import jakarta.validation.constraints.NotEmpty; +import jakarta.validation.constraints.NotNull; +import lombok.Data; + +import java.io.Serial; +import java.io.Serializable; +import java.util.Map; + +/** + * 太阳能网控属性下发请求对象 + * + * @author intc + */ +@Data +@Schema(description = "太阳能网控属性下发请求对象") +public class DeviceSolarPropertyCommandBo implements Serializable { + + @Serial + private static final long serialVersionUID = 1L; + + /** + * 设备ID + */ + @Schema(description = "设备ID") + @NotNull(message = "设备ID不能为空") + private Long id; + + /** + * 待下发属性，key 为物模型属性标识符 + */ + @Schema(description = "待下发属性") + @NotEmpty(message = "下发属性不能为空") + private Map properties; + + /** + * 是否缓存下发。true=写入缓存，设备上线后自动补发；false=立即下发 + */ + @Schema(description = "是否缓存下发") + private Boolean cache; +} diff --git a/intc-modules/intc-iot/src/main/java/com/intc/iot/handler/DeviceDataHandler.java b/intc-modules/intc-iot/src/main/java/com/intc/iot/handler/DeviceDataHandler.java index 5a19fad..6718ed3 100644 --- a/intc-modules/intc-iot/src/main/java/com/intc/iot/handler/DeviceDataHandler.java +++ b/intc-modules/intc-iot/src/main/java/com/intc/iot/handler/DeviceDataHandler.java @@ -45,8 +45,11 @@ import java.time.ZoneId; import java.time.format.DateTimeFormatter; import java.util.Date; import java.util.HashMap; +import java.util.LinkedHashMap; import java.util.List; import java.util.Map; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; /** * 设备数据处理器 @@ -59,9 +62,15 @@ import java.util.Map; @ConditionalOnBean(IDeviceSensorDataService.class) public class DeviceDataHandler { - public static final String PENDING_SOLAR_SALINITY_KEY_PREFIX = "device:pending:solar:salinity:"; + public static final String PENDING_SOLAR_PROPERTY_KEY_PREFIX = "device:pending:solar:property:"; private static final int DEVICE_TYPE_SOLAR_CONTROLLER = 3; + /** + * 太阳能网控上线后缓存属性下发的延迟时间（毫秒） + * 设备上线后 MQTT 订阅需要一定时间完成，延迟后再下发设置避免失败 + */ + private static final long SOLAR_PROPERTY_DELAY_MS = 400L; + /** * TDengine 服务，用于存储设备时序数据 */ @@ -127,6 +136,11 @@ public class DeviceDataHandler { */ private final AliyunIotProperties aliyunIotProperties; + /** + * 调度线程池，用于延迟执行任务 + */ + private final ScheduledExecutorService scheduledExecutorService; + private static final DateTimeFormatter DATETIME_FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"); /** @@ -246,7 +260,7 @@ public class DeviceDataHandler { if (params != null) { handleSalinitySet(deviceName, params); } -// flushPendingSolarSalinity(deviceName, "属性上报"); +// flushPendingSolarProperties(deviceName, "属性上报"); if (params != null && params.size() > 0) { try { @@ -960,71 +974,16 @@ public class DeviceDataHandler { log.info("[离线告警] 设备已上线，取消离线告警: {}", deviceName); } - // 设备上线后直接补发缓存盐度 - flushPendingSolarSalinity(deviceName, "生命周期上线"); + // 太阳能网控上线后，延迟下发缓存属性（设备MQTT订阅需要一定时间完成，立即下发可能失败） + scheduledExecutorService.schedule( + () -> flushPendingSolarProperties(deviceName, "生命周期上线"), + SOLAR_PROPERTY_DELAY_MS, TimeUnit.MILLISECONDS); // 同步更新 Device.warnCode，清除离线和休眠标志位 setDeviceWarnCodeBit(deviceName, DefineDeviceWarnCode.DeviceOffline, false); setDeviceWarnCodeBit(deviceName, DefineDeviceWarnCode.DeviceSleeping, false); } - private void flushPendingSolarSalinity(String deviceName, String trigger) { - Device device = findDeviceByOnlineName(deviceName); - if (device == null) { - log.debug("[太阳能盐度] 上线设备未绑定，跳过补发: deviceName={}", deviceName); - return; - } - if (!Integer.valueOf(DEVICE_TYPE_SOLAR_CONTROLLER).equals(device.getDeviceType())) { - log.debug("[太阳能盐度] 非太阳能网控设备，跳过补发: deviceName={}, deviceId={}, deviceType={}", - deviceName, device.getId(), device.getDeviceType()); - return; - } - - try { - PendingSolarSalinity pendingSalinity = findPendingSolarSalinity(deviceName, device); - if (pendingSalinity == null) { - logNoPendingSolarSalinity(trigger, deviceName, device); - return; - } - - JSONObject pending = JSONUtil.parseObj(pendingSalinity.rawPending); - Double salinityCompensation = pending.getDouble("salinityCompensation"); - String iotId = pending.getStr("iotId"); - Long deviceId = pending.getLong("deviceId"); - if (salinityCompensation == null || iotId == null || deviceId == null) { - log.warn("[太阳能盐度] 缓存数据不完整，跳过补发: deviceName={}, pending={}", deviceName, pending); - return; - } - - log.info("[太阳能盐度] 太阳能网控{}，准备补发缓存盐度: deviceName={}, cacheKey={}, deviceId={}, salinityCompensation={}", - trigger, deviceName, pendingSalinity.redisKey, deviceId, salinityCompensation); - - Map properties = new HashMap<>(); - properties.put("salinitySet", salinityCompensation); - boolean success = iotCloudService.setProperty(iotId, properties, true, 2); - if (!success) { - log.warn("[太阳能盐度] {}补发失败，保留缓存: deviceName={}, deviceId={}, salinityCompensation={}", - trigger, deviceName, deviceId, salinityCompensation); - return; - } - - Device updateDevice = new Device(); - updateDevice.setId(deviceId); - updateDevice.setSalinityCompensation(salinityCompensation); - int updated = deviceMapper.updateById(updateDevice); - if (updated > 0) { - clearPendingSolarSalinity(deviceName, device, pendingSalinity.redisKey); - log.info("[太阳能盐度] {}补发成功并清除缓存: deviceName={}, cacheKey={}, deviceId={}, salinityCompensation={}", - trigger, deviceName, pendingSalinity.redisKey, deviceId, salinityCompensation); - } else { - log.warn("[太阳能盐度] {}补发成功但设备表更新失败，保留缓存以便重试: deviceName={}, deviceId={}", - trigger, deviceName, deviceId); - } - } catch (Exception e) { - log.error("[太阳能盐度] {}补发异常: deviceName={}, err={}", trigger, deviceName, e.getMessage(), e); - } - } - private Device findDeviceByOnlineName(String deviceName) { if (StrUtil.isBlank(deviceName)) { return null; @@ -1041,47 +1000,196 @@ public class DeviceDataHandler { ); } - private PendingSolarSalinity findPendingSolarSalinity(String deviceName, Device device) { + private void flushPendingSolarProperties(String deviceName, String trigger) { + Device device = findDeviceByOnlineName(deviceName); + if (device == null) { + log.debug("[太阳能属性] 上线设备未绑定，跳过补发: deviceName={}", deviceName); + return; + } + if (!Integer.valueOf(DEVICE_TYPE_SOLAR_CONTROLLER).equals(device.getDeviceType())) { + log.debug("[太阳能属性] 非太阳能网控设备，跳过补发: deviceName={}, deviceId={}, deviceType={}", + deviceName, device.getId(), device.getDeviceType()); + return; + } + + try { + PendingSolarProperty pendingProperty = findPendingSolarProperty(device); + if (pendingProperty == null) { + logNoPendingSolarProperty(trigger, deviceName, device); + return; + } + + if (pendingProperty.iotId == null || pendingProperty.deviceId == null || pendingProperty.properties == null) { + log.warn("[太阳能属性] 缓存数据不完整，跳过补发: deviceName={}, cacheKey={}", + deviceName, pendingProperty.redisKey); + return; + } + + if (pendingProperty.properties.isEmpty()) { + log.warn("[太阳能属性] 缓存属性为空，跳过补发: deviceName={}, cacheKey={}", + deviceName, pendingProperty.redisKey); + return; + } + + log.info("[太阳能属性] 太阳能网控{}，准备补发缓存属性: deviceName={}, cacheKey={}, deviceId={}, properties={}", + trigger, deviceName, pendingProperty.redisKey, pendingProperty.deviceId, pendingProperty.properties); + + boolean success = iotCloudService.setProperty(pendingProperty.iotId, pendingProperty.properties, true, 2); + if (!success) { + log.warn("[太阳能属性] {}补发失败，保留缓存: deviceName={}, deviceId={}, properties={}", + trigger, deviceName, pendingProperty.deviceId, pendingProperty.properties); + return; + } + + syncDeviceFieldsAfterPropertySet(pendingProperty.deviceId, pendingProperty.properties); + clearPendingSolarProperty(pendingProperty); + log.info("[太阳能属性] {}补发成功并清除缓存: deviceName={}, cacheKey={}, deviceId={}, properties={}", + trigger, deviceName, pendingProperty.redisKey, pendingProperty.deviceId, pendingProperty.properties); + } catch (Exception e) { + log.error("[太阳能属性] {}补发异常: deviceName={}, err={}", trigger, deviceName, e.getMessage(), e); + } + } + + private PendingSolarProperty findPendingSolarProperty(Device device) { String serialNum = device.getSerialNum(); if (StrUtil.isBlank(serialNum)) { return null; } - // 太阳能盐度缓存为跨上下文共享数据（HTTP线程写入 + AMQP消费线程读取）， - // 必须忽略租户前缀，避免写入带租户前缀而读取不带前缀导致 key 不匹配 return TenantHelper.ignore(() -> { - String redisKey = PENDING_SOLAR_SALINITY_KEY_PREFIX + serialNum; + String redisKey = PENDING_SOLAR_PROPERTY_KEY_PREFIX + serialNum; Object rawPending = RedisUtils.getCacheObject(redisKey); - return rawPending != null ? new PendingSolarSalinity(redisKey, rawPending) : null; + if (rawPending == null) { + return null; + } + + Long deviceId = device.getId(); + String iotId = device.getIotId(); + Map properties = new LinkedHashMap<>(); + + try { + JSONObject pending = JSONUtil.parseObj(rawPending); + Long cachedDeviceId = pending.getLong("deviceId"); + String cachedIotId = pending.getStr("iotId"); + if (cachedDeviceId != null) { + deviceId = cachedDeviceId; + } + if (StrUtil.isNotBlank(cachedIotId)) { + iotId = cachedIotId; + } + Object propertiesObj = pending.get("properties"); + if (propertiesObj != null) { + JSONObject propertiesJson = JSONUtil.parseObj(propertiesObj); + for (String key : propertiesJson.keySet()) { + properties.put(key, propertiesJson.get(key)); + } + } + } catch (Exception e) { + log.warn("[太阳能属性] 解析缓存属性失败: serialNum={}, err={}", serialNum, e.getMessage()); + } + + return new PendingSolarProperty(redisKey, deviceId, iotId, properties); }); } - private boolean hasPendingSolarSalinity(String deviceName, Device device) { - return findPendingSolarSalinity(deviceName, device) != null; + private boolean hasPendingSolarProperty(Device device, String identifier) { + if (device == null || StrUtil.isBlank(identifier)) { + return false; + } + PendingSolarProperty pendingProperty = findPendingSolarProperty(device); + return pendingProperty != null + && pendingProperty.properties != null + && pendingProperty.properties.containsKey(identifier); } - private void logNoPendingSolarSalinity(String trigger, String deviceName, Device device) { + private void logNoPendingSolarProperty(String trigger, String deviceName, Device device) { if ("生命周期上线".equals(trigger)) { - log.info("[太阳能盐度] 太阳能网控{}，未找到待补发缓存: deviceName={}, deviceId={}, serialNum={}, iotId={}", + log.info("[太阳能属性] 太阳能网控{}，未找到待补发缓存: deviceName={}, deviceId={}, serialNum={}, iotId={}", trigger, deviceName, device.getId(), device.getSerialNum(), device.getIotId()); return; } - log.debug("[太阳能盐度] 太阳能网控{}，未找到待补发缓存: deviceName={}, deviceId={}, serialNum={}, iotId={}", + log.debug("[太阳能属性] 太阳能网控{}，未找到待补发缓存: deviceName={}, deviceId={}, serialNum={}, iotId={}", trigger, deviceName, device.getId(), device.getSerialNum(), device.getIotId()); } - private void clearPendingSolarSalinity(String deviceName, Device device, String matchedRedisKey) { - // 忽略租户前缀，确保与写入侧 key 一致（写入在 HTTP 线程，删除在 AMQP 线程） - TenantHelper.ignore(() -> RedisUtils.deleteObject(matchedRedisKey)); + private void clearPendingSolarProperty(PendingSolarProperty pendingProperty) { + if (pendingProperty == null) { + return; + } + TenantHelper.ignore(() -> { + if (StrUtil.isNotBlank(pendingProperty.redisKey)) { + RedisUtils.deleteObject(pendingProperty.redisKey); + } + }); } - private static class PendingSolarSalinity { - private final String redisKey; - private final Object rawPending; + private void syncDeviceFieldsAfterPropertySet(Long deviceId, Map properties) { + if (deviceId == null || properties == null || properties.isEmpty()) { + return; + } + Device updateDevice = new Device(); + updateDevice.setId(deviceId); + boolean needUpdate = false; - private PendingSolarSalinity(String redisKey, Object rawPending) { + if (properties.containsKey("salinitySet")) { + Double value = toDouble(properties.get("salinitySet")); + if (value != null) { + updateDevice.setSalinityCompensation(value); + needUpdate = true; + } + } + if (properties.containsKey("phase_compensation")) { + Double value = toDouble(properties.get("phase_compensation")); + if (value != null) { + updateDevice.setPhaseCompensation(value); + needUpdate = true; + } + } + if (properties.containsKey("phasedif_compensation")) { + Double value = toDouble(properties.get("phasedif_compensation")); + if (value != null) { + updateDevice.setPhasedifCompensation(value); + needUpdate = true; + } + } + if (properties.containsKey("temperature_compensation")) { + Double value = toDouble(properties.get("temperature_compensation")); + if (value != null) { + updateDevice.setTemperatureCompensation(value); + needUpdate = true; + } + } + + if (needUpdate) { + deviceMapper.updateById(updateDevice); + } + } + + private Double toDouble(Object value) { + if (value == null) { + return null; + } + if (value instanceof Number number) { + return number.doubleValue(); + } + try { + return Double.valueOf(value.toString()); + } catch (Exception e) { + return null; + } + } + + private static class PendingSolarProperty { + private final String redisKey; + private final Long deviceId; + private final String iotId; + private final Map properties; + + private PendingSolarProperty(String redisKey, Long deviceId, String iotId, Map properties) { this.redisKey = redisKey; - this.rawPending = rawPending; + this.deviceId = deviceId; + this.iotId = iotId; + this.properties = properties; } } @@ -2748,15 +2856,15 @@ public class DeviceDataHandler { Device device = deviceMapper.selectOne( new com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper() .eq(Device::getSerialNum, deviceName) - .select(Device::getId) + .select(Device::getId, Device::getIotId, Device::getSerialNum) .last("LIMIT 1") ); if (device == null) { log.debug("设备不存在，无法更新盐度补偿设定值: {}", deviceName); return; } - if (hasPendingSolarSalinity(deviceName, device)) { - log.debug("[太阳能盐度] 存在待补发盐度，跳过设备旧salinitySet同步: deviceName={}, reportSalinitySet={}", + if (hasPendingSolarProperty(device, "salinitySet")) { + log.debug("[太阳能属性] 存在待补发salinitySet，跳过设备旧salinitySet同步: deviceName={}, reportSalinitySet={}", deviceName, salinitySet); return; }