diff --git a/wm-iot/pom.xml b/wm-iot/pom.xml
index 6ebe218a..81769730 100644
--- a/wm-iot/pom.xml
+++ b/wm-iot/pom.xml
@@ -13,5 +13,10 @@
org.springframework.bootspring-boot-starter-data-redis
org.postgresqlpostgresql
net.postgispostgis-jdbc
+
+ org.springframework.boot
+ spring-boot-starter-test
+ test
+
\ No newline at end of file
diff --git a/wm-iot/src/main/java/com/water/iot/config/RedisConfig.java b/wm-iot/src/main/java/com/water/iot/config/RedisConfig.java
new file mode 100644
index 00000000..0b25cde6
--- /dev/null
+++ b/wm-iot/src/main/java/com/water/iot/config/RedisConfig.java
@@ -0,0 +1,59 @@
+package com.water.iot.config;
+
+import com.fasterxml.jackson.annotation.JsonAutoDetect;
+import com.fasterxml.jackson.annotation.PropertyAccessor;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.SerializationFeature;
+import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.data.redis.connection.RedisConnectionFactory;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.data.redis.core.StringRedisTemplate;
+import org.springframework.data.redis.serializer.GenericJackson2JsonRedisSerializer;
+import org.springframework.data.redis.serializer.StringRedisSerializer;
+
+/**
+ * Redis配置
+ * - StringRedisTemplate: key/value均为String序列化(用于设备影子Hash存储)
+ * - RedisTemplate: key为String，value为JSON序列化(通用)
+ */
+@Configuration
+public class RedisConfig {
+
+ @Bean
+ public StringRedisTemplate stringRedisTemplate(RedisConnectionFactory factory) {
+ StringRedisTemplate template = new StringRedisTemplate();
+ template.setConnectionFactory(factory);
+ return template;
+ }
+
+ @Bean
+ public RedisTemplate redisTemplate(RedisConnectionFactory factory) {
+ RedisTemplate template = new RedisTemplate<>();
+ template.setConnectionFactory(factory);
+
+ // JSON序列化器
+ ObjectMapper mapper = new ObjectMapper();
+ mapper.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
+ mapper.activateDefaultTyping(
+ mapper.getPolymorphicTypeValidator(),
+ ObjectMapper.DefaultTyping.NON_FINAL
+ );
+ mapper.registerModule(new JavaTimeModule());
+ mapper.disable(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS);
+
+ GenericJackson2JsonRedisSerializer jsonSerializer = new GenericJackson2JsonRedisSerializer(mapper);
+ StringRedisSerializer stringSerializer = new StringRedisSerializer();
+
+ // key序列化
+ template.setKeySerializer(stringSerializer);
+ template.setHashKeySerializer(stringSerializer);
+ // value序列化
+ template.setValueSerializer(jsonSerializer);
+ template.setHashValueSerializer(jsonSerializer);
+
+ template.afterPropertiesSet();
+ return template;
+ }
+}
diff --git a/wm-iot/src/main/java/com/water/iot/controller/DeviceShadowController.java b/wm-iot/src/main/java/com/water/iot/controller/DeviceShadowController.java
new file mode 100644
index 00000000..6886e24d
--- /dev/null
+++ b/wm-iot/src/main/java/com/water/iot/controller/DeviceShadowController.java
@@ -0,0 +1,107 @@
+package com.water.iot.controller;
+
+import com.water.common.core.result.R;
+import com.water.iot.entity.DeviceShadow;
+import com.water.iot.service.DeviceShadowService;
+import io.swagger.v3.oas.annotations.Operation;
+import io.swagger.v3.oas.annotations.tags.Tag;
+import lombok.RequiredArgsConstructor;
+import org.springframework.web.bind.annotation.*;
+
+import java.util.List;
+import java.util.Map;
+
+@Tag(name = "设备影子")
+@RestController
+@RequestMapping("/api/iot/shadow")
+@RequiredArgsConstructor
+public class DeviceShadowController {
+
+ private final DeviceShadowService shadowService;
+
+ @Operation(summary = "获取设备影子")
+ @GetMapping("/{deviceSn}")
+ public R getShadow(@PathVariable String deviceSn) {
+ DeviceShadow shadow = shadowService.getShadow(deviceSn);
+ if (shadow == null) {
+ return R.fail(404, "设备影子不存在: " + deviceSn);
+ }
+ return R.ok(shadow);
+ }
+
+ @Operation(summary = "更新设备上报状态")
+ @PostMapping("/{deviceSn}/reported")
+ public R updateReported(@PathVariable String deviceSn,
+ @RequestBody Map state) {
+ shadowService.updateReported(deviceSn, state);
+ return R.ok("上报状态已更新");
+ }
+
+ @Operation(summary = "更新期望状态(云端下发)")
+ @PostMapping("/{deviceSn}/desired")
+ public R updateDesired(@PathVariable String deviceSn,
+ @RequestBody Map desired) {
+ try {
+ com.fasterxml.jackson.databind.ObjectMapper mapper = new com.fasterxml.jackson.databind.ObjectMapper();
+ String json = mapper.writeValueAsString(desired);
+ shadowService.updateDesired(deviceSn, json);
+ return R.ok("期望状态已下发");
+ } catch (Exception e) {
+ return R.fail("序列化失败: " + e.getMessage());
+ }
+ }
+
+ @Operation(summary = "获取上报状态")
+ @GetMapping("/{deviceSn}/reported")
+ public R getReportedState(@PathVariable String deviceSn) {
+ String state = shadowService.getReportedState(deviceSn);
+ return R.ok(state);
+ }
+
+ @Operation(summary = "获取期望状态")
+ @GetMapping("/{deviceSn}/desired")
+ public R getDesiredState(@PathVariable String deviceSn) {
+ String state = shadowService.getDesiredState(deviceSn);
+ return R.ok(state);
+ }
+
+ @Operation(summary = "获取差异状态(delta)")
+ @GetMapping("/{deviceSn}/delta")
+ public R getDeltaState(@PathVariable String deviceSn) {
+ String state = shadowService.getDeltaState(deviceSn);
+ return R.ok(state);
+ }
+
+ @Operation(summary = "批量查询设备影子")
+ @PostMapping("/batch")
+ public R> batchGetShadows(@RequestBody List deviceSns) {
+ return R.ok(shadowService.batchGetShadows(deviceSns));
+ }
+
+ @Operation(summary = "检查设备在线状态")
+ @GetMapping("/{deviceSn}/online")
+ public R checkOnline(@PathVariable String deviceSn,
+ @RequestParam(defaultValue = "30") int thresholdMinutes) {
+ return R.ok(shadowService.checkOnline(deviceSn, thresholdMinutes));
+ }
+
+ @Operation(summary = "批量检测在线状态")
+ @PostMapping("/online/batch")
+ public R> batchCheckOnline(@RequestBody List deviceSns,
+ @RequestParam(defaultValue = "30") int thresholdMinutes) {
+ return R.ok(shadowService.batchCheckOnline(deviceSns, thresholdMinutes));
+ }
+
+ @Operation(summary = "在线设备数量")
+ @GetMapping("/online/count")
+ public R countOnline() {
+ return R.ok(shadowService.countOnlineDevices());
+ }
+
+ @Operation(summary = "删除设备影子")
+ @DeleteMapping("/{deviceSn}")
+ public R deleteShadow(@PathVariable String deviceSn) {
+ shadowService.deleteShadow(deviceSn);
+ return R.ok("影子已删除");
+ }
+}
diff --git a/wm-iot/src/main/java/com/water/iot/controller/OtaController.java b/wm-iot/src/main/java/com/water/iot/controller/OtaController.java
new file mode 100644
index 00000000..e077a607
--- /dev/null
+++ b/wm-iot/src/main/java/com/water/iot/controller/OtaController.java
@@ -0,0 +1,179 @@
+package com.water.iot.controller;
+
+import com.water.common.core.result.R;
+import com.water.iot.entity.OtaFirmware;
+import com.water.iot.entity.OtaTask;
+import com.water.iot.entity.OtaUpgradeRecord;
+import com.water.iot.service.OtaService;
+import io.swagger.v3.oas.annotations.Operation;
+import io.swagger.v3.oas.annotations.tags.Tag;
+import lombok.RequiredArgsConstructor;
+import org.springframework.web.bind.annotation.*;
+
+import java.util.List;
+import java.util.Map;
+
+@Tag(name = "OTA固件升级")
+@RestController
+@RequestMapping("/api/iot/ota")
+@RequiredArgsConstructor
+public class OtaController {
+
+ private final OtaService otaService;
+
+ // ========== 固件管理 ==========
+
+ @Operation(summary = "创建固件版本")
+ @PostMapping("/firmware")
+ public R createFirmware(@RequestBody OtaFirmware firmware) {
+ return R.ok(otaService.createFirmware(firmware));
+ }
+
+ @Operation(summary = "发布固件")
+ @PostMapping("/firmware/{id}/publish")
+ public R publishFirmware(@PathVariable Long id,
+ @RequestParam String publishedBy) {
+ otaService.publishFirmware(id, publishedBy);
+ return R.ok("固件已发布");
+ }
+
+ @Operation(summary = "废弃固件")
+ @PostMapping("/firmware/{id}/deprecate")
+ public R deprecateFirmware(@PathVariable Long id) {
+ otaService.deprecateFirmware(id);
+ return R.ok("固件已废弃");
+ }
+
+ @Operation(summary = "查询固件详情")
+ @GetMapping("/firmware/{id}")
+ public R getFirmware(@PathVariable Long id) {
+ OtaFirmware firmware = otaService.getFirmware(id);
+ if (firmware == null) {
+ return R.fail(404, "固件不存在: " + id);
+ }
+ return R.ok(firmware);
+ }
+
+ @Operation(summary = "按模型查询固件列表")
+ @GetMapping("/firmware/model/{modelId}")
+ public R> listByModel(@PathVariable Long modelId) {
+ return R.ok(otaService.listFirmwareByModel(modelId));
+ }
+
+ @Operation(summary = "查询所有固件(分页)")
+ @GetMapping("/firmware")
+ public R> listAllFirmware(@RequestParam(defaultValue = "1") int page,
+ @RequestParam(defaultValue = "10") int size) {
+ return R.ok(otaService.listAllFirmware(page, size));
+ }
+
+ @Operation(summary = "设备查询可用固件")
+ @GetMapping("/firmware/available/{deviceSn}")
+ public R getAvailableFirmware(@PathVariable String deviceSn) {
+ OtaFirmware firmware = otaService.getAvailableFirmware(deviceSn);
+ return R.ok(firmware);
+ }
+
+ // ========== 升级任务管理 ==========
+
+ @Operation(summary = "创建升级任务(指定设备列表)")
+ @PostMapping("/task")
+ public R createUpgradeTask(@RequestBody Map body) {
+ Long firmwareId = ((Number) body.get("firmwareId")).longValue();
+ @SuppressWarnings("unchecked")
+ List deviceIdNums = (List) body.get("deviceIds");
+ List deviceIds = deviceIdNums.stream().map(Number::longValue).toList();
+ int batchSize = body.containsKey("batchSize") ? ((Number) body.get("batchSize")).intValue() : 10;
+ String createdBy = (String) body.getOrDefault("createdBy", "system");
+
+ return R.ok(otaService.createUpgradeTask(firmwareId, deviceIds, batchSize, createdBy));
+ }
+
+ @Operation(summary = "按条件创建升级任务")
+ @PostMapping("/task/filter")
+ public R createTaskByFilter(@RequestBody Map body) {
+ Long firmwareId = ((Number) body.get("firmwareId")).longValue();
+ String deviceType = (String) body.get("deviceType");
+ String area = (String) body.get("area");
+ int batchSize = body.containsKey("batchSize") ? ((Number) body.get("batchSize")).intValue() : 10;
+ String createdBy = (String) body.getOrDefault("createdBy", "system");
+
+ return R.ok(otaService.createUpgradeTaskByFilter(firmwareId, deviceType, area, batchSize, createdBy));
+ }
+
+ @Operation(summary = "启动升级任务")
+ @PostMapping("/task/{id}/start")
+ public R startTask(@PathVariable Long id) {
+ otaService.startTask(id);
+ return R.ok("任务已启动");
+ }
+
+ @Operation(summary = "取消升级任务")
+ @PostMapping("/task/{id}/cancel")
+ public R cancelTask(@PathVariable Long id) {
+ otaService.cancelTask(id);
+ return R.ok("任务已取消");
+ }
+
+ @Operation(summary = "查询升级任务详情")
+ @GetMapping("/task/{id}")
+ public R getTask(@PathVariable Long id) {
+ OtaTask task = otaService.getTask(id);
+ if (task == null) {
+ return R.fail(404, "任务不存在: " + id);
+ }
+ return R.ok(task);
+ }
+
+ @Operation(summary = "查询升级任务列表")
+ @GetMapping("/task")
+ public R> listTasks(@RequestParam(defaultValue = "1") int page,
+ @RequestParam(defaultValue = "10") int size) {
+ return R.ok(otaService.listTasks(page, size));
+ }
+
+ // ========== 进度追踪 ==========
+
+ @Operation(summary = "更新升级进度")
+ @PostMapping("/progress/{recordId}")
+ public R updateProgress(@PathVariable Long recordId,
+ @RequestParam int progress) {
+ otaService.updateProgress(recordId, progress);
+ return R.ok("进度已更新");
+ }
+
+ @Operation(summary = "标记升级成功")
+ @PostMapping("/record/{recordId}/success")
+ public R markSuccess(@PathVariable Long recordId,
+ @RequestParam Long deviceId,
+ @RequestParam String toVersion) {
+ otaService.markSuccess(recordId, deviceId, toVersion);
+ return R.ok("已标记成功");
+ }
+
+ @Operation(summary = "标记升级失败")
+ @PostMapping("/record/{recordId}/fail")
+ public R markFailed(@PathVariable Long recordId,
+ @RequestParam String reason) {
+ otaService.markFailed(recordId, reason);
+ return R.ok("已标记失败");
+ }
+
+ @Operation(summary = "查询任务统计")
+ @GetMapping("/task/{id}/statistics")
+ public R> getTaskStatistics(@PathVariable Long id) {
+ return R.ok(otaService.getTaskStatistics(id));
+ }
+
+ @Operation(summary = "查询任务升级记录")
+ @GetMapping("/task/{id}/records")
+ public R> getTaskRecords(@PathVariable Long id) {
+ return R.ok(otaService.getTaskRecords(id));
+ }
+
+ @Operation(summary = "查询设备升级历史")
+ @GetMapping("/device/{deviceSn}/history")
+ public R> getDeviceHistory(@PathVariable String deviceSn) {
+ return R.ok(otaService.getDeviceUpgradeHistory(deviceSn));
+ }
+}
diff --git a/wm-iot/src/main/java/com/water/iot/entity/DeviceShadow.java b/wm-iot/src/main/java/com/water/iot/entity/DeviceShadow.java
new file mode 100644
index 00000000..77f87b79
--- /dev/null
+++ b/wm-iot/src/main/java/com/water/iot/entity/DeviceShadow.java
@@ -0,0 +1,40 @@
+package com.water.iot.entity;
+
+import lombok.Data;
+import java.time.LocalDateTime;
+
+/**
+ * 设备影子实体
+ * Redis Hash存储：key=shadow:{deviceSn}
+ * fields: reported(JSON), desired(JSON), lastReportTime, online
+ */
+@Data
+public class DeviceShadow {
+
+ /** 设备SN(主键) */
+ private String deviceSn;
+
+ /** 设备ID */
+ private Long deviceId;
+
+ /** 设备名称 */
+ private String deviceName;
+
+ /** 上报状态JSON */
+ private String reportedState;
+
+ /** 期望状态JSON(云端下发) */
+ private String desiredState;
+
+ /** 差异状态JSON */
+ private String deltaState;
+
+ /** 最后上报时间 */
+ private LocalDateTime lastReportTime;
+
+ /** 在线状态 */
+ private Boolean online;
+
+ /** 影子版本号 */
+ private Long version;
+}
diff --git a/wm-iot/src/main/java/com/water/iot/entity/OtaFirmware.java b/wm-iot/src/main/java/com/water/iot/entity/OtaFirmware.java
new file mode 100644
index 00000000..7b6599d7
--- /dev/null
+++ b/wm-iot/src/main/java/com/water/iot/entity/OtaFirmware.java
@@ -0,0 +1,44 @@
+package com.water.iot.entity;
+
+import lombok.Data;
+import java.time.LocalDateTime;
+
+/**
+ * OTA固件版本实体
+ */
+@Data
+public class OtaFirmware {
+
+ /** 固件ID */
+ private Long id;
+
+ /** 设备模型ID */
+ private Long modelId;
+
+ /** 固件版本号 */
+ private String firmwareVersion;
+
+ /** 固件文件URL */
+ private String fileUrl;
+
+ /** 固件描述 */
+ private String description;
+
+ /** 状态: draft/published/deprecated */
+ private String status;
+
+ /** 文件MD5 */
+ private String md5;
+
+ /** 文件大小(bytes) */
+ private Long fileSize;
+
+ /** 发布人 */
+ private String publishedBy;
+
+ /** 发布时间 */
+ private LocalDateTime publishedAt;
+
+ private LocalDateTime createdAt;
+ private LocalDateTime updatedAt;
+}
diff --git a/wm-iot/src/main/java/com/water/iot/entity/OtaTask.java b/wm-iot/src/main/java/com/water/iot/entity/OtaTask.java
new file mode 100644
index 00000000..58d8df8b
--- /dev/null
+++ b/wm-iot/src/main/java/com/water/iot/entity/OtaTask.java
@@ -0,0 +1,59 @@
+package com.water.iot.entity;
+
+import lombok.Data;
+import java.time.LocalDateTime;
+
+/**
+ * OTA升级任务实体
+ */
+@Data
+public class OtaTask {
+
+ /** 任务ID */
+ private Long id;
+
+ /** 固件ID */
+ private Long firmwareId;
+
+ /** 固件版本号 */
+ private String firmwareVersion;
+
+ /** 目标设备ID列表(JSON) */
+ private String targetDeviceIds;
+
+ /** 任务状态: pending/executing/completed/cancelled */
+ private String taskStatus;
+
+ /** 每批大小 */
+ private Integer batchSize;
+
+ /** 目标设备总数 */
+ private Integer totalDevices;
+
+ /** 成功数 */
+ private Integer successCount;
+
+ /** 失败数 */
+ private Integer failedCount;
+
+ /** 执行中数量 */
+ private Integer executingCount;
+
+ /** 目标设备类型(可选) */
+ private String targetType;
+
+ /** 目标区域(可选) */
+ private String targetArea;
+
+ /** 创建人 */
+ private String createdBy;
+
+ /** 执行开始时间 */
+ private LocalDateTime executedAt;
+
+ /** 完成时间 */
+ private LocalDateTime completedAt;
+
+ private LocalDateTime createdAt;
+ private LocalDateTime updatedAt;
+}
diff --git a/wm-iot/src/main/java/com/water/iot/entity/OtaUpgradeRecord.java b/wm-iot/src/main/java/com/water/iot/entity/OtaUpgradeRecord.java
new file mode 100644
index 00000000..b3995906
--- /dev/null
+++ b/wm-iot/src/main/java/com/water/iot/entity/OtaUpgradeRecord.java
@@ -0,0 +1,46 @@
+package com.water.iot.entity;
+
+import lombok.Data;
+import java.time.LocalDateTime;
+
+/**
+ * OTA升级记录实体(每台设备一条)
+ */
+@Data
+public class OtaUpgradeRecord {
+
+ /** 记录ID */
+ private Long id;
+
+ /** 任务ID */
+ private Long taskId;
+
+ /** 设备ID */
+ private Long deviceId;
+
+ /** 设备SN */
+ private String deviceSn;
+
+ /** 升级前版本 */
+ private String fromVersion;
+
+ /** 升级目标版本 */
+ private String toVersion;
+
+ /** 状态: pending/executing/success/failed */
+ private String status;
+
+ /** 进度(0-100) */
+ private Integer progress;
+
+ /** 失败原因 */
+ private String failReason;
+
+ /** 开始时间 */
+ private LocalDateTime startedAt;
+
+ /** 完成时间 */
+ private LocalDateTime completedAt;
+
+ private LocalDateTime createdAt;
+}
diff --git a/wm-iot/src/main/java/com/water/iot/service/DeviceShadowService.java b/wm-iot/src/main/java/com/water/iot/service/DeviceShadowService.java
index cee5dc52..54f38e7d 100644
--- a/wm-iot/src/main/java/com/water/iot/service/DeviceShadowService.java
+++ b/wm-iot/src/main/java/com/water/iot/service/DeviceShadowService.java
@@ -2,15 +2,25 @@ package com.water.iot.service;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
+import com.water.iot.entity.DeviceShadow;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.StringRedisTemplate;
+import org.springframework.jdbc.core.BeanPropertyRowMapper;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Service;
-import java.util.Map;
+import java.time.LocalDateTime;
+import java.time.temporal.ChronoUnit;
+import java.util.*;
import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+/**
+ * 设备影子服务
+ * Redis Hash存储：key=shadow:{deviceSn}，TTL 24h
+ * fields: reported(JSON), desired(JSON), lastReportTime, online, version
+ */
@Slf4j
@Service
@RequiredArgsConstructor
@@ -18,32 +28,251 @@ public class DeviceShadowService {
private final StringRedisTemplate redisTemplate;
private final JdbcTemplate jdbcTemplate;
- private final ObjectMapper mapper = new ObjectMapper();
+ private final ObjectMapper objectMapper;
- private static final String SHADOW_PREFIX = "iot:shadow:";
+ private static final String SHADOW_PREFIX = "shadow:";
private static final long SHADOW_TTL_HOURS = 24;
- /** 更新设备上报状态 */
+ /**
+ * 更新设备上报状态
+ * @param deviceSn 设备SN
+ * @param state 上报的状态数据(Map)
+ */
public void updateReported(String deviceSn, Map state) {
try {
String key = SHADOW_PREFIX + deviceSn;
- String json = mapper.writeValueAsString(state);
- redisTemplate.opsForHash().put(key, "reported", json);
+ String json = objectMapper.writeValueAsString(state);
+
+ Map fields = new HashMap<>();
+ fields.put("reported", json);
+ fields.put("lastReportTime", LocalDateTime.now().toString());
+ fields.put("online", "true");
+
+ // 递增版本号
+ Long version = redisTemplate.opsForHash().increment(key, "version", 1);
+ fields.put("version", String.valueOf(version));
+
+ redisTemplate.opsForHash().putAll(key, fields);
redisTemplate.expire(key, SHADOW_TTL_HOURS, TimeUnit.HOURS);
- // 同步更新数据库设备最后上报时间
- jdbcTemplate.update("UPDATE iot_device SET last_report_time = NOW() WHERE device_sn = ?", deviceSn);
+
+ // 计算delta(期望与上报的差异)
+ computeDelta(deviceSn);
+
+ // 同步更新数据库设备最后上报时间和在线状态
+ jdbcTemplate.update(
+ "UPDATE iot_device SET last_report_time = NOW(), status = 'online' WHERE device_sn = ?",
+ deviceSn
+ );
+
+ log.debug("Shadow reported updated: deviceSn={}, version={}", deviceSn, version);
} catch (JsonProcessingException e) {
- log.error("Shadow update error: {}", e.getMessage());
+ log.error("Shadow update error for device {}: {}", deviceSn, e.getMessage());
}
}
- /** 获取设备影子 */
- public Map getShadow(String deviceSn) {
- return redisTemplate.opsForHash().entries(SHADOW_PREFIX + deviceSn);
+ /**
+ * 更新期望状态(云端→设备)
+ * @param deviceSn 设备SN
+ * @param desiredJson 期望状态JSON字符串
+ */
+ public void updateDesired(String deviceSn, String desiredJson) {
+ String key = SHADOW_PREFIX + deviceSn;
+ Map fields = new HashMap<>();
+ fields.put("desired", desiredJson);
+ Long version = redisTemplate.opsForHash().increment(key, "version", 1);
+ fields.put("version", String.valueOf(version));
+
+ redisTemplate.opsForHash().putAll(key, fields);
+ redisTemplate.expire(key, SHADOW_TTL_HOURS, TimeUnit.HOURS);
+
+ computeDelta(deviceSn);
+ log.debug("Shadow desired updated: deviceSn={}, version={}", deviceSn, version);
}
- /** 更新期望状态(云端→设备) */
- public void updateDesired(String deviceSn, String desiredJson) {
- redisTemplate.opsForHash().put(SHADOW_PREFIX + deviceSn, "desired", desiredJson);
+ /**
+ * 获取设备影子完整数据
+ */
+ public DeviceShadow getShadow(String deviceSn) {
+ String key = SHADOW_PREFIX + deviceSn;
+ Map entries = redisTemplate.opsForHash().entries(key);
+ if (entries.isEmpty()) {
+ return null;
+ }
+ return mapToShadow(deviceSn, entries);
+ }
+
+ /**
+ * 获取设备上报状态
+ */
+ public String getReportedState(String deviceSn) {
+ Object reported = redisTemplate.opsForHash().get(SHADOW_PREFIX + deviceSn, "reported");
+ return reported != null ? reported.toString() : null;
+ }
+
+ /**
+ * 获取设备期望状态
+ */
+ public String getDesiredState(String deviceSn) {
+ Object desired = redisTemplate.opsForHash().get(SHADOW_PREFIX + deviceSn, "desired");
+ return desired != null ? desired.toString() : null;
+ }
+
+ /**
+ * 获取差异状态(delta = desired - reported)
+ */
+ public String getDeltaState(String deviceSn) {
+ Object delta = redisTemplate.opsForHash().get(SHADOW_PREFIX + deviceSn, "delta");
+ return delta != null ? delta.toString() : null;
+ }
+
+ /**
+ * 批量查询设备影子
+ * @param deviceSns 设备SN列表
+ */
+ public List batchGetShadows(List deviceSns) {
+ if (deviceSns == null || deviceSns.isEmpty()) {
+ return Collections.emptyList();
+ }
+ return deviceSns.stream()
+ .map(this::getShadow)
+ .filter(Objects::nonNull)
+ .collect(Collectors.toList());
+ }
+
+ /**
+ * 离线检测：检查指定设备是否在线(基于Redis TTL)
+ * @param deviceSn 设备SN
+ * @param offlineThresholdMinutes 离线阈值(分钟)，超过该时间无上报视为离线
+ * @return true=在线, false=离线
+ */
+ public boolean checkOnline(String deviceSn, int offlineThresholdMinutes) {
+ String key = SHADOW_PREFIX + deviceSn;
+ Object lastReportTimeStr = redisTemplate.opsForHash().get(key, "lastReportTime");
+
+ if (lastReportTimeStr == null) {
+ // 无上报记录，查DB
+ List dbTimes = jdbcTemplate.queryForList(
+ "SELECT last_report_time FROM iot_device WHERE device_sn = ?",
+ LocalDateTime.class, deviceSn
+ );
+ if (dbTimes.isEmpty() || dbTimes.get(0) == null) return false;
+ lastReportTimeStr = dbTimes.get(0).toString();
+ }
+
+ LocalDateTime lastReport = LocalDateTime.parse(lastReportTimeStr.toString());
+ long minutesSinceReport = ChronoUnit.MINUTES.between(lastReport, LocalDateTime.now());
+
+ boolean isOnline = minutesSinceReport <= offlineThresholdMinutes;
+
+ // 更新Redis在线状态
+ redisTemplate.opsForHash().put(key, "online", String.valueOf(isOnline));
+
+ // 同步到DB
+ String dbStatus = isOnline ? "online" : "offline";
+ jdbcTemplate.update("UPDATE iot_device SET status = ? WHERE device_sn = ?", dbStatus, deviceSn);
+
+ return isOnline;
+ }
+
+ /**
+ * 批量离线检测(使用默认阈值30分钟)
+ */
+ public Map batchCheckOnline(List deviceSns) {
+ return batchCheckOnline(deviceSns, 30);
+ }
+
+ /**
+ * 批量离线检测
+ */
+ public Map batchCheckOnline(List deviceSns, int offlineThresholdMinutes) {
+ Map result = new LinkedHashMap<>();
+ for (String sn : deviceSns) {
+ result.put(sn, checkOnline(sn, offlineThresholdMinutes));
+ }
+ return result;
+ }
+
+ /**
+ * 删除设备影子(设备注销时)
+ */
+ public void deleteShadow(String deviceSn) {
+ redisTemplate.delete(SHADOW_PREFIX + deviceSn);
+ }
+
+ /**
+ * 查询在线设备数量(Redis中有记录且online=true)
+ */
+ public long countOnlineDevices() {
+ // 使用DB统计更可靠，这里提供Redis快速统计
+ List> result = jdbcTemplate.queryForList(
+ "SELECT COUNT(*) as cnt FROM iot_device WHERE status = 'online'"
+ );
+ return result.isEmpty() ? 0 : ((Number) result.get(0).get("cnt")).longValue();
+ }
+
+ // ===== 私有方法 =====
+
+ /**
+ * 计算delta状态(期望与上报的差异)
+ */
+ @SuppressWarnings("unchecked")
+ private void computeDelta(String deviceSn) {
+ String key = SHADOW_PREFIX + deviceSn;
+ try {
+ Object reportedObj = redisTemplate.opsForHash().get(key, "reported");
+ Object desiredObj = redisTemplate.opsForHash().get(key, "desired");
+
+ if (desiredObj == null) {
+ // 无期望状态，delta为空
+ redisTemplate.opsForHash().put(key, "delta", "{}");
+ return;
+ }
+
+ Map reported = reportedObj != null ?
+ objectMapper.readValue(reportedObj.toString(), Map.class) : Collections.emptyMap();
+ Map desired = objectMapper.readValue(desiredObj.toString(), Map.class);
+
+ // delta = desired中不同于reported的字段
+ Map delta = new LinkedHashMap<>();
+ for (Map.Entry entry : desired.entrySet()) {
+ Object reportedVal = reported.get(entry.getKey());
+ if (!Objects.equals(entry.getValue(), reportedVal)) {
+ delta.put(entry.getKey(), entry.getValue());
+ }
+ }
+
+ String deltaJson = objectMapper.writeValueAsString(delta);
+ redisTemplate.opsForHash().put(key, "delta", deltaJson);
+ } catch (Exception e) {
+ log.warn("Delta compute error for device {}: {}", deviceSn, e.getMessage());
+ }
+ }
+
+ /**
+ * Redis Hash → DeviceShadow
+ */
+ private DeviceShadow mapToShadow(String deviceSn, Map entries) {
+ DeviceShadow shadow = new DeviceShadow();
+ shadow.setDeviceSn(deviceSn);
+ shadow.setReportedState(entries.get("reported") != null ? entries.get("reported").toString() : null);
+ shadow.setDesiredState(entries.get("desired") != null ? entries.get("desired").toString() : null);
+ shadow.setDeltaState(entries.get("delta") != null ? entries.get("delta").toString() : null);
+ shadow.setOnline(entries.get("online") != null && "true".equals(entries.get("online").toString()));
+ shadow.setVersion(entries.get("version") != null ? Long.parseLong(entries.get("version").toString()) : 0L);
+
+ if (entries.get("lastReportTime") != null) {
+ shadow.setLastReportTime(LocalDateTime.parse(entries.get("lastReportTime").toString()));
+ }
+
+ // 从DB补充设备信息
+ List> deviceInfo = jdbcTemplate.queryForList(
+ "SELECT id, device_name FROM iot_device WHERE device_sn = ?", deviceSn
+ );
+ if (!deviceInfo.isEmpty()) {
+ shadow.setDeviceId(((Number) deviceInfo.get(0).get("id")).longValue());
+ shadow.setDeviceName((String) deviceInfo.get(0).get("device_name"));
+ }
+
+ return shadow;
}
}
diff --git a/wm-iot/src/main/java/com/water/iot/service/OtaService.java b/wm-iot/src/main/java/com/water/iot/service/OtaService.java
index c3ccc02f..2b868f67 100644
--- a/wm-iot/src/main/java/com/water/iot/service/OtaService.java
+++ b/wm-iot/src/main/java/com/water/iot/service/OtaService.java
@@ -1,12 +1,30 @@
package com.water.iot.service;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.water.iot.entity.OtaFirmware;
+import com.water.iot.entity.OtaTask;
+import com.water.iot.entity.OtaUpgradeRecord;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
+import org.springframework.jdbc.core.BeanPropertyRowMapper;
import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.jdbc.support.GeneratedKeyHolder;
+import org.springframework.jdbc.support.KeyHolder;
import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
-import java.util.Map;
+import java.sql.PreparedStatement;
+import java.sql.Statement;
+import java.time.LocalDateTime;
+import java.util.*;
+import java.util.stream.Collectors;
+/**
+ * OTA固件升级服务
+ * 支持固件上传/版本管理/升级任务创建(按批次)/进度追踪/结果统计/设备查询可用固件
+ */
@Slf4j
@Service
@RequiredArgsConstructor
@@ -14,21 +32,475 @@ public class OtaService {
private final JdbcTemplate jdbcTemplate;
private final DeviceShadowService shadowService;
+ private final ObjectMapper objectMapper;
- /** 创建 OTA 升级任务 */
- public void createUpgrade(Long modelId, String firmwareVersion, String firmwareUrl, String checkMd5) {
- jdbcTemplate.update(
- "INSERT INTO iot_device_event (device_id, device_sn, event_type, event_data) " +
- "SELECT id, device_sn, 'ota', json_build_object('version',?, 'url',?, 'md5',?) " +
- "FROM iot_device WHERE model_id = ? AND status = 'online'",
- firmwareVersion, firmwareUrl, checkMd5, modelId);
- log.info("OTA task created for model {}: version={}", modelId, firmwareVersion);
+ // ========== 固件管理 ==========
+
+ /**
+ * 创建固件版本
+ */
+ @Transactional
+ public OtaFirmware createFirmware(OtaFirmware firmware) {
+ firmware.setStatus("draft");
+ firmware.setCreatedAt(LocalDateTime.now());
+ firmware.setUpdatedAt(LocalDateTime.now());
+
+ KeyHolder keyHolder = new GeneratedKeyHolder();
+ jdbcTemplate.update(connection -> {
+ var ps = connection.prepareStatement(
+ "INSERT INTO iot_ota_firmware (model_id, firmware_version, file_url, description, status, md5, file_size, created_at, updated_at) " +
+ "VALUES (?, ?, ?, ?, ?, ?, ?, NOW(), NOW())",
+ Statement.RETURN_GENERATED_KEYS
+ );
+ ps.setLong(1, firmware.getModelId());
+ ps.setString(2, firmware.getFirmwareVersion());
+ ps.setString(3, firmware.getFileUrl());
+ ps.setString(4, firmware.getDescription());
+ ps.setString(5, firmware.getStatus());
+ ps.setString(6, firmware.getMd5());
+ ps.setLong(7, firmware.getFileSize() != null ? firmware.getFileSize() : 0);
+ return ps;
+ }, keyHolder);
+
+ firmware.setId(keyHolder.getKey().longValue());
+ log.info("Firmware created: id={}, version={}", firmware.getId(), firmware.getFirmwareVersion());
+ return firmware;
}
- /** 设备查询是否有待升级固件 */
- public Map checkUpgrade(String deviceSn, String currentVersion) {
- return jdbcTemplate.queryForMap(
- "SELECT * FROM iot_device_event WHERE device_sn = ? AND event_type = 'ota' ORDER BY created_at DESC LIMIT 1",
- deviceSn);
+ /**
+ * 发布固件
+ */
+ @Transactional
+ public void publishFirmware(Long firmwareId, String publishedBy) {
+ int updated = jdbcTemplate.update(
+ "UPDATE iot_ota_firmware SET status = 'published', published_by = ?, published_at = NOW(), updated_at = NOW() WHERE id = ?",
+ publishedBy, firmwareId
+ );
+ if (updated == 0) throw new RuntimeException("固件不存在: " + firmwareId);
+ log.info("Firmware {} published by {}", firmwareId, publishedBy);
+ }
+
+ /**
+ * 废弃固件
+ */
+ @Transactional
+ public void deprecateFirmware(Long firmwareId) {
+ jdbcTemplate.update(
+ "UPDATE iot_ota_firmware SET status = 'deprecated', updated_at = NOW() WHERE id = ?",
+ firmwareId
+ );
+ }
+
+ /**
+ * 查询固件详情
+ */
+ public OtaFirmware getFirmware(Long firmwareId) {
+ List list = jdbcTemplate.query(
+ "SELECT * FROM iot_ota_firmware WHERE id = ?",
+ new BeanPropertyRowMapper<>(OtaFirmware.class),
+ firmwareId
+ );
+ return list.isEmpty() ? null : list.get(0);
+ }
+
+ /**
+ * 按模型查询固件版本列表
+ */
+ public List listFirmwareByModel(Long modelId) {
+ return jdbcTemplate.query(
+ "SELECT * FROM iot_ota_firmware WHERE model_id = ? ORDER BY created_at DESC",
+ new BeanPropertyRowMapper<>(OtaFirmware.class),
+ modelId
+ );
+ }
+
+ /**
+ * 查询所有固件(分页)
+ */
+ public List listAllFirmware(int page, int size) {
+ int offset = (page - 1) * size;
+ return jdbcTemplate.query(
+ "SELECT * FROM iot_ota_firmware ORDER BY created_at DESC LIMIT ? OFFSET ?",
+ new BeanPropertyRowMapper<>(OtaFirmware.class),
+ size, offset
+ );
+ }
+
+ /**
+ * 设备查询可用固件（基于设备SN查找最新发布的固件）
+ */
+ public OtaFirmware getAvailableFirmware(String deviceSn) {
+ // 查找设备对应的模型
+ List modelIds = jdbcTemplate.queryForList(
+ "SELECT model_id FROM iot_device WHERE device_sn = ? AND model_id IS NOT NULL",
+ Long.class, deviceSn
+ );
+ if (modelIds.isEmpty()) return null;
+
+ Long modelId = modelIds.get(0);
+ List list = jdbcTemplate.query(
+ "SELECT * FROM iot_ota_firmware WHERE model_id = ? AND status = 'published' ORDER BY created_at DESC LIMIT 1",
+ new BeanPropertyRowMapper<>(OtaFirmware.class),
+ modelId
+ );
+ return list.isEmpty() ? null : list.get(0);
+ }
+
+ // ========== 升级任务管理 ==========
+
+ /**
+ * 创建OTA升级任务（按批次）
+ * @param firmwareId 固件ID
+ * @param deviceIds 目标设备ID列表
+ * @param batchSize 每批大小
+ * @param createdBy 创建人
+ */
+ @Transactional
+ public OtaTask createUpgradeTask(Long firmwareId, List deviceIds, int batchSize, String createdBy) {
+ OtaFirmware firmware = getFirmware(firmwareId);
+ if (firmware == null) throw new RuntimeException("固件不存在: " + firmwareId);
+ if (!"published".equals(firmware.getStatus())) throw new RuntimeException("固件未发布: " + firmwareId);
+
+ // 序列化设备ID列表
+ String deviceIdsJson;
+ try {
+ deviceIdsJson = objectMapper.writeValueAsString(deviceIds);
+ } catch (JsonProcessingException e) {
+ throw new RuntimeException("设备ID序列化失败", e);
+ }
+
+ OtaTask task = new OtaTask();
+ task.setFirmwareId(firmwareId);
+ task.setFirmwareVersion(firmware.getFirmwareVersion());
+ task.setTargetDeviceIds(deviceIdsJson);
+ task.setTaskStatus("pending");
+ task.setBatchSize(batchSize > 0 ? batchSize : 10);
+ task.setTotalDevices(deviceIds.size());
+ task.setSuccessCount(0);
+ task.setFailedCount(0);
+ task.setExecutingCount(0);
+ task.setCreatedBy(createdBy);
+ task.setCreatedAt(LocalDateTime.now());
+ task.setUpdatedAt(LocalDateTime.now());
+
+ KeyHolder keyHolder = new GeneratedKeyHolder();
+ jdbcTemplate.update(connection -> {
+ var ps = connection.prepareStatement(
+ "INSERT INTO iot_ota_task (firmware_id, firmware_version, target_device_ids, task_status, " +
+ "batch_size, total_devices, success_count, failed_count, executing_count, created_by, created_at, updated_at) " +
+ "VALUES (?, ?, ?::jsonb, ?, ?, ?, 0, 0, 0, ?, NOW(), NOW())",
+ Statement.RETURN_GENERATED_KEYS
+ );
+ ps.setLong(1, firmwareId);
+ ps.setString(2, firmware.getFirmwareVersion());
+ ps.setString(3, deviceIdsJson);
+ ps.setString(4, "pending");
+ ps.setInt(5, task.getBatchSize());
+ ps.setInt(6, deviceIds.size());
+ ps.setString(7, createdBy);
+ return ps;
+ }, keyHolder);
+
+ task.setId(keyHolder.getKey().longValue());
+
+ // 创建升级记录(每台设备一条)
+ createUpgradeRecords(task.getId(), deviceIds, firmware.getFirmwareVersion());
+
+ log.info("Upgrade task created: id={}, firmware={}, devices={}, batchSize={}",
+ task.getId(), firmware.getFirmwareVersion(), deviceIds.size(), task.getBatchSize());
+
+ return task;
+ }
+
+ /**
+ * 按设备类型/区域创建升级任务
+ */
+ @Transactional
+ public OtaTask createUpgradeTaskByFilter(Long firmwareId, String deviceType, String area,
+ int batchSize, String createdBy) {
+ // 查询符合条件的在线设备
+ String sql = "SELECT id FROM iot_device WHERE model_id = (SELECT model_id FROM iot_ota_firmware WHERE id = ?) " +
+ "AND status = 'online'";
+ List