XW-AIOT/wvp-platform
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(aiot): 实现配置直推Edge - Redis Stream聚合配置下发

Browse Source 
- 新增 writeDeviceAggregatedConfig/buildFlatConfig 聚合配置写入
- pushConfig 增加设备聚合配置推送步骤
- rollback 三方法增加 Stream 通知和 deviceId 回填修复
- AiRoiServiceImpl 新增 resolveDeviceId 自动关联边缘设备
- Mapper 层新增设备查询/回填/清理方法
- 新增 backfill-device-id 数据修复端点

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
16337
2026-02-10 15:21:19 +08:00
parent e8c3b32249
commit 53d0b2bb1f
8 changed files with 338 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

67
src/main/java/com/genersoft/iot/vmp/aiot/controller/AiConfigController.java
View File

@@ -1,6 +1,9 @@
package com.genersoft.iot.vmp.aiot.controller;
import com.genersoft.iot.vmp.aiot.bean.AiConfigSnapshot;
import com.genersoft.iot.vmp.aiot.bean.AiEdgeDevice;
import com.genersoft.iot.vmp.aiot.dao.AiEdgeDeviceMapper;
import com.genersoft.iot.vmp.aiot.dao.AiRoiMapper;
import com.genersoft.iot.vmp.aiot.service.IAiConfigService;
import com.genersoft.iot.vmp.aiot.service.IAiConfigSnapshotService;
import com.github.pagehelper.PageInfo;
@@ -11,6 +14,8 @@ import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.*;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
@Slf4j
@@ -25,6 +30,12 @@ public class AiConfigController {
@Autowired
private IAiConfigSnapshotService snapshotService;
@Autowired
private AiRoiMapper roiMapper;
@Autowired
private AiEdgeDeviceMapper edgeDeviceMapper;
// ==================== 配置推送 ====================
@Operation(summary = "推送配置到边缘端（Redis+通知）")
@@ -95,4 +106,60 @@ public class AiConfigController {
@Parameter(description = "版本B") @RequestParam Integer versionB) {
return snapshotService.diff(scopeType, scopeId, versionA, versionB);
}
// ==================== 数据修复 ====================
@Operation(summary = "统一边缘设备为 'edge'，清理多余设备，更新所有ROI关联")
@PostMapping("/backfill-device-id")
public Map<String, Object> backfillDeviceId() {
String targetDeviceId = "edge";
Map<String, Object> result = new LinkedHashMap<>();
// 1. 查询是否已有 device_id="edge" 的设备
AiEdgeDevice targetDevice = edgeDeviceMapper.queryByDeviceId(targetDeviceId);
if (targetDevice == null) {
// 没有 "edge" 设备，取第一个已注册设备并改名
List<AiEdgeDevice> devices = edgeDeviceMapper.queryAll();
if (devices == null || devices.isEmpty()) {
// 没有任何设备，创建一个
AiEdgeDevice newDevice = new AiEdgeDevice();
newDevice.setDeviceId(targetDeviceId);
newDevice.setStatus("offline");
newDevice.setUptimeSeconds(0L);
newDevice.setFramesProcessed(0L);
newDevice.setAlertsGenerated(0L);
edgeDeviceMapper.add(newDevice);
result.put("device_action", "created");
log.info("[AiConfig] 无已注册设备，已创建 device_id={}", targetDeviceId);
} else {
// 将第一个设备改名为 "edge"
AiEdgeDevice first = devices.get(0);
String oldId = first.getDeviceId();
edgeDeviceMapper.renameDeviceId(first.getId(), targetDeviceId);
result.put("device_action", "renamed");
result.put("old_device_id", oldId);
log.info("[AiConfig] 已将设备 {} 改名为 {}", oldId, targetDeviceId);
}
} else {
result.put("device_action", "already_exists");
}
// 2. 删除除 "edge" 以外的所有边缘设备
int deleted = edgeDeviceMapper.deleteAllExcept(targetDeviceId);
result.put("devices_deleted", deleted);
if (deleted > 0) {
log.info("[AiConfig] 已删除 {} 个多余边缘设备", deleted);
}
// 3. 将所有 ROI 的 device_id 统一为 "edge"
int updatedRois = roiMapper.updateAllDeviceId(targetDeviceId);
result.put("rois_updated", updatedRois);
result.put("status", "success");
result.put("device_id", targetDeviceId);
log.info("[AiConfig] 统一 device_id 完成: deviceId={}, deletedDevices={}, updatedRois={}",
targetDeviceId, deleted, updatedRois);
return result;
}
}

6
src/main/java/com/genersoft/iot/vmp/aiot/dao/AiEdgeDeviceMapper.java
View File

@@ -30,4 +30,10 @@ public interface AiEdgeDeviceMapper {
@Update("UPDATE wvp_ai_edge_device SET status='offline', updated_at=#{now} " +
"WHERE status='online' AND last_heartbeat < #{threshold}")
int markOffline(@Param("threshold") String threshold, @Param("now") String now);
@Delete("DELETE FROM wvp_ai_edge_device WHERE device_id != #{keepDeviceId}")
int deleteAllExcept(@Param("keepDeviceId") String keepDeviceId);
@Update("UPDATE wvp_ai_edge_device SET device_id=#{newDeviceId} WHERE id=#{id}")
int renameDeviceId(@Param("id") Integer id, @Param("newDeviceId") String newDeviceId);
}

12
src/main/java/com/genersoft/iot/vmp/aiot/dao/AiRoiMapper.java
View File

@@ -46,4 +46,16 @@ public interface AiRoiMapper {
@Select("SELECT * FROM wvp_ai_roi WHERE camera_id=#{cameraId} ORDER BY priority DESC")
List<AiRoi> queryAllByCameraId(@Param("cameraId") String cameraId);
@Select("SELECT DISTINCT camera_id FROM wvp_ai_roi WHERE device_id=#{deviceId}")
List<String> queryDistinctCameraIdsByDeviceId(@Param("deviceId") String deviceId);
@Select("SELECT DISTINCT device_id FROM wvp_ai_roi WHERE camera_id=#{cameraId} AND device_id IS NOT NULL LIMIT 1")
String queryDeviceIdByCameraId(@Param("cameraId") String cameraId);
@Update("UPDATE wvp_ai_roi SET device_id=#{deviceId} WHERE device_id IS NULL OR device_id=''")
int backfillDeviceId(@Param("deviceId") String deviceId);
@Update("UPDATE wvp_ai_roi SET device_id=#{deviceId}")
int updateAllDeviceId(@Param("deviceId") String deviceId);
}

9
src/main/java/com/genersoft/iot/vmp/aiot/service/IAiRedisConfigService.java
View File

@@ -50,4 +50,13 @@ public interface IAiRedisConfigService {
* 全量同步摄像头配置到Redis
*/
void syncCameraConfigToRedis(String cameraId);
/**
* 构建设备聚合配置（扁平格式：cameras/rois/binds平级）并写入 device:{deviceId}:config，
* 递增 device:{deviceId}:version，发布 device_config_stream 事件
*
* @param deviceId 边缘设备ID
* @param action 触发动作（UPDATE / ROLLBACK）
*/
void writeDeviceAggregatedConfig(String deviceId, String action);
}

14
src/main/java/com/genersoft/iot/vmp/aiot/service/impl/AiConfigServiceImpl.java
View File

@@ -103,15 +103,23 @@ public class AiConfigServiceImpl implements IAiConfigService {
JSON.toJSONString(config),
"PUSH", "推送配置到边缘端", null);
// 3. 将每个ROI和Bind写入Redis
// 3. 将每个ROI和Bind写入Redis（旧格式，保持兼容）
redisConfigService.syncCameraConfigToRedis(cameraId);
// 4. 发布 config_update 到 Redis Pub/Sub 通知边缘端
// 4. 发布 config_update 到 Redis Pub/Sub 通知边缘端（旧通知方式）
List<String> ids = new ArrayList<>();
ids.add(cameraId);
redisConfigService.publishConfigUpdate("full", ids);
// 5. 返回推送结果
// 5. 写入设备聚合配置 + Stream 通知（新格式，对接 Edge config_sync）
String deviceId = roiMapper.queryDeviceIdByCameraId(cameraId);
if (deviceId != null && !deviceId.isEmpty()) {
redisConfigService.writeDeviceAggregatedConfig(deviceId, "UPDATE");
} else {
log.warn("[AiConfig] 摄像头 {} 未关联边缘设备，跳过聚合配置推送", cameraId);
}
// 6. 返回推送结果
Map<String, Object> result = new LinkedHashMap<>();
result.put("camera_id", cameraId);
result.put("version", snapshot.getVersion());

30
src/main/java/com/genersoft/iot/vmp/aiot/service/impl/AiConfigSnapshotServiceImpl.java
View File

@@ -86,6 +86,9 @@ public class AiConfigSnapshotServiceImpl implements IAiConfigSnapshotService {
JSONObject config = JSON.parseObject(snapshot.getSnapshot());
JSONArray rois = config.getJSONArray("rois");
// 在删除前保存 deviceId（删除后查不到了）
String deviceId = roiMapper.queryDeviceIdByCameraId(cameraId);
// 删除当前所有ROI和绑定
List<AiRoi> currentRois = roiMapper.queryAllByCameraId(cameraId);
for (AiRoi roi : currentRois) {
@@ -101,6 +104,7 @@ public class AiConfigSnapshotServiceImpl implements IAiConfigSnapshotService {
AiRoi roi = new AiRoi();
roi.setRoiId(roiJson.getString("roi_id"));
roi.setCameraId(cameraId);
roi.setDeviceId(deviceId);
roi.setRoiType(roiJson.getString("roi_type"));
roi.setName(roiJson.getString("name"));
roi.setCoordinates(roiJson.containsKey("coordinates") ?
@@ -139,12 +143,17 @@ public class AiConfigSnapshotServiceImpl implements IAiConfigSnapshotService {
createSnapshot("CAMERA", cameraId, cameraId, JSON.toJSONString(newConfig),
"ROLLBACK", "回滚到版本" + targetVersion, operator);
// 推送到Redis
// 推送到Redis（旧格式）
redisConfigService.syncCameraConfigToRedis(cameraId);
List<String> ids = new ArrayList<>();
ids.add(cameraId);
redisConfigService.publishConfigUpdate("full", ids);
// 写入设备聚合配置 + Stream 通知（新格式，对接 Edge config_sync）
if (deviceId != null && !deviceId.isEmpty()) {
redisConfigService.writeDeviceAggregatedConfig(deviceId, "ROLLBACK");
}
log.info("[AiSnapshot] 摄像头配置回滚完成: cameraId={}, targetVersion={}", cameraId, targetVersion);
}
@@ -169,6 +178,9 @@ public class AiConfigSnapshotServiceImpl implements IAiConfigSnapshotService {
roi = new AiRoi();
roi.setRoiId(roiId);
roi.setCameraId(cameraId);
// 填充 deviceId（新建 ROI 时需要）
String deviceId = roiMapper.queryDeviceIdByCameraId(cameraId);
roi.setDeviceId(deviceId);
roi.setCreateTime(now);
}
roi.setRoiType(roiJson.getString("roi_type"));
@@ -211,12 +223,18 @@ public class AiConfigSnapshotServiceImpl implements IAiConfigSnapshotService {
createSnapshot("ROI", roiId, cameraId, snapshot.getSnapshot(),
"ROLLBACK", "回滚到版本" + targetVersion, operator);
// 推送到Redis
// 推送到Redis（旧格式）
redisConfigService.syncCameraConfigToRedis(cameraId);
List<String> ids = new ArrayList<>();
ids.add(roiId);
redisConfigService.publishConfigUpdate("roi", ids);
// 写入设备聚合配置 + Stream 通知（新格式，对接 Edge config_sync）
String deviceId = roiMapper.queryDeviceIdByCameraId(cameraId);
if (deviceId != null && !deviceId.isEmpty()) {
redisConfigService.writeDeviceAggregatedConfig(deviceId, "ROLLBACK");
}
log.info("[AiSnapshot] ROI配置回滚完成: roiId={}, targetVersion={}", roiId, targetVersion);
}
@@ -257,12 +275,18 @@ public class AiConfigSnapshotServiceImpl implements IAiConfigSnapshotService {
createSnapshot("BIND", bindId, cameraId, snapshot.getSnapshot(),
"ROLLBACK", "回滚到版本" + targetVersion, operator);
// 推送到Redis
// 推送到Redis（旧格式）
redisConfigService.syncCameraConfigToRedis(cameraId);
List<String> ids = new ArrayList<>();
ids.add(bindId);
redisConfigService.publishConfigUpdate("bind", ids);
// 写入设备聚合配置 + Stream 通知（新格式，对接 Edge config_sync）
String deviceId = roiMapper.queryDeviceIdByCameraId(cameraId);
if (deviceId != null && !deviceId.isEmpty()) {
redisConfigService.writeDeviceAggregatedConfig(deviceId, "ROLLBACK");
}
log.info("[AiSnapshot] 绑定配置回滚完成: bindId={}, targetVersion={}", bindId, targetVersion);
}

171
src/main/java/com/genersoft/iot/vmp/aiot/service/impl/AiRedisConfigServiceImpl.java
View File

@@ -17,9 +17,12 @@ import com.genersoft.iot.vmp.media.bean.MediaServer;
import com.genersoft.iot.vmp.media.service.IMediaServerService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.RecordId;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.*;
@@ -422,4 +425,172 @@ public class AiRedisConfigServiceImpl implements IAiRedisConfigService {
}
return val.toString();
}
// ==================== 设备聚合配置（对接 Edge config_sync） ====================
@Override
public void writeDeviceAggregatedConfig(String deviceId, String action) {
if (deviceId == null || deviceId.isEmpty()) {
log.warn("[AiRedis] deviceId 为空，跳过聚合配置写入");
return;
}
try {
// 1. 查询该设备下所有摄像头
List<String> cameraIds = roiMapper.queryDistinctCameraIdsByDeviceId(deviceId);
if (cameraIds == null || cameraIds.isEmpty()) {
log.info("[AiRedis] 设备 {} 下无关联摄像头，写入空配置", deviceId);
cameraIds = Collections.emptyList();
}
// 2. 构建扁平格式 JSON
Map<String, Object> flatConfig = buildFlatConfig(deviceId, cameraIds);
// 3. 写入聚合 Redis Key
String configKey = "device:" + deviceId + ":config";
String versionKey = "device:" + deviceId + ":version";
stringRedisTemplate.opsForValue().set(configKey, JSON.toJSONString(flatConfig));
Long newVersion = stringRedisTemplate.opsForValue().increment(versionKey);
// 4. 发布 Stream 事件
Map<String, String> event = new LinkedHashMap<>();
event.put("device_id", deviceId);
event.put("version", String.valueOf(newVersion));
event.put("action", action != null ? action : "UPDATE");
event.put("timestamp", Instant.now().toString());
stringRedisTemplate.opsForStream().add("device_config_stream",
Collections.<String, String>unmodifiableMap(event));
// 保留最近 10000 条
stringRedisTemplate.opsForStream().trim("device_config_stream", 10000);
log.info("[AiRedis] 设备聚合配置写入完成: deviceId={}, cameras={}, version={}, action={}",
deviceId, cameraIds.size(), newVersion, action);
} catch (Exception e) {
log.error("[AiRedis] 设备聚合配置写入失败: deviceId={}, error={}", deviceId, e.getMessage(), e);
}
}
/**
* 构建扁平格式配置 JSON（Edge 期望的格式）
* 输出: {cameras: [...], rois: [...], binds: [...]}
*/
private Map<String, Object> buildFlatConfig(String deviceId, List<String> cameraIds) {
List<Map<String, Object>> cameras = new ArrayList<>();
List<Map<String, Object>> rois = new ArrayList<>();
List<Map<String, Object>> binds = new ArrayList<>();
com.fasterxml.jackson.databind.ObjectMapper objectMapper = new com.fasterxml.jackson.databind.ObjectMapper();
for (String cameraId : cameraIds) {
// 摄像头信息
Map<String, Object> cameraMap = new LinkedHashMap<>();
cameraMap.put("camera_id", cameraId);
// 获取 RTSP URL 和摄像头名称
String rtspUrl = "";
String cameraName = "";
List<AiRoi> cameraRois = roiMapper.queryAllByCameraId(cameraId);
if (!cameraRois.isEmpty()) {
AiRoi firstRoi = cameraRois.get(0);
if (firstRoi.getChannelDbId() != null) {
try {
CommonGBChannel channel = channelMapper.queryById(firstRoi.getChannelDbId());
if (channel != null) {
cameraName = channel.getGbName() != null ? channel.getGbName() : "";
}
} catch (Exception e) {
log.warn("[AiRedis] 查询通道信息失败: channelDbId={}", firstRoi.getChannelDbId());
}
}
}
// 构建 RTSP 代理地址
try {
MediaServer mediaServer = mediaServerService.getDefaultMediaServer();
if (mediaServer != null && mediaServer.getRtspPort() != 0) {
rtspUrl = String.format("rtsp://%s:%s/%s",
mediaServer.getStreamIp() != null ? mediaServer.getStreamIp() : mediaServer.getIp(),
mediaServer.getRtspPort(), cameraId);
} else if (mediaServer != null) {
rtspUrl = String.format("http://%s:%s/%s.live.flv",
mediaServer.getStreamIp() != null ? mediaServer.getStreamIp() : mediaServer.getIp(),
mediaServer.getHttpPort(), cameraId);
}
} catch (Exception e) {
log.warn("[AiRedis] 获取媒体服务器信息失败: {}", e.getMessage());
}
cameraMap.put("rtsp_url", rtspUrl);
cameraMap.put("camera_name", cameraName);
cameraMap.put("enabled", true);
cameraMap.put("location", "");
cameras.add(cameraMap);
// 该摄像头下的 ROI 和绑定
for (AiRoi roi : cameraRois) {
Map<String, Object> roiMap = new LinkedHashMap<>();
roiMap.put("roi_id", roi.getRoiId());
roiMap.put("camera_id", roi.getCameraId());
roiMap.put("roi_type", roi.getRoiType() != null ? roi.getRoiType() : "polygon");
roiMap.put("enabled", roi.getEnabled() != null && roi.getEnabled() == 1);
roiMap.put("priority", roi.getPriority() != null ? roi.getPriority() : 0);
// coordinates: 解析为标准数组格式 [[x,y], ...]
try {
Object coords = objectMapper.readValue(roi.getCoordinates(), Object.class);
// 如果是 [{x:..., y:...}] 格式，转为 [[x,y], ...]
if (coords instanceof List) {
List<?> coordList = (List<?>) coords;
if (!coordList.isEmpty() && coordList.get(0) instanceof Map) {
List<List<Object>> converted = new ArrayList<>();
for (Object item : coordList) {
Map<?, ?> point = (Map<?, ?>) item;
List<Object> pair = new ArrayList<>();
pair.add(point.get("x"));
pair.add(point.get("y"));
converted.add(pair);
}
roiMap.put("coordinates", converted);
} else {
roiMap.put("coordinates", coords);
}
} else {
roiMap.put("coordinates", coords);
}
} catch (Exception e) {
roiMap.put("coordinates", new ArrayList<>());
}
rois.add(roiMap);
// 算法绑定
List<AiRoiAlgoBind> roiBinds = bindMapper.queryByRoiId(roi.getRoiId());
for (AiRoiAlgoBind bind : roiBinds) {
Map<String, Object> bindMap = new LinkedHashMap<>();
bindMap.put("bind_id", bind.getBindId());
bindMap.put("roi_id", bind.getRoiId());
bindMap.put("algo_code", bind.getAlgoCode());
bindMap.put("enabled", bind.getEnabled() != null && bind.getEnabled() == 1);
bindMap.put("priority", bind.getPriority() != null ? bind.getPriority() : 0);
// params: 解析为标准 JSON 对象（非 Python eval 字符串）
String effectiveParams = resolveEffectiveParams(bind);
try {
bindMap.put("params", objectMapper.readValue(effectiveParams, Object.class));
} catch (Exception e) {
bindMap.put("params", new LinkedHashMap<>());
}
binds.add(bindMap);
}
}
}
Map<String, Object> flatConfig = new LinkedHashMap<>();
flatConfig.put("cameras", cameras);
flatConfig.put("rois", rois);
flatConfig.put("binds", binds);
return flatConfig;
}
}

35
src/main/java/com/genersoft/iot/vmp/aiot/service/impl/AiRoiServiceImpl.java
View File

@@ -2,6 +2,7 @@ package com.genersoft.iot.vmp.aiot.service.impl;
import com.genersoft.iot.vmp.aiot.bean.*;
import com.genersoft.iot.vmp.aiot.dao.AiAlgorithmMapper;
import com.genersoft.iot.vmp.aiot.dao.AiEdgeDeviceMapper;
import com.genersoft.iot.vmp.aiot.dao.AiRoiAlgoBindMapper;
import com.genersoft.iot.vmp.aiot.dao.AiRoiMapper;
import com.genersoft.iot.vmp.aiot.service.IAiConfigLogService;
@@ -33,6 +34,9 @@ public class AiRoiServiceImpl implements IAiRoiService {
@Autowired
private AiAlgorithmMapper algorithmMapper;
@Autowired
private AiEdgeDeviceMapper edgeDeviceMapper;
@Autowired
private IAiConfigLogService configLogService;
@@ -43,6 +47,16 @@ public class AiRoiServiceImpl implements IAiRoiService {
public void save(AiRoi roi) {
String now = LocalDateTime.now().format(FORMATTER);
roi.setUpdateTime(now);
// 自动填充 deviceId（边缘设备关联）
if (ObjectUtils.isEmpty(roi.getDeviceId()) && !ObjectUtils.isEmpty(roi.getCameraId())) {
String resolvedDeviceId = resolveDeviceId(roi.getCameraId());
if (resolvedDeviceId != null) {
roi.setDeviceId(resolvedDeviceId);
log.info("[AiRoi] 自动关联边缘设备: cameraId={}, deviceId={}", roi.getCameraId(), resolvedDeviceId);
}
}
if (roi.getId() != null && roi.getId() > 0) {
AiRoi old = roiMapper.queryById(roi.getId());
roiMapper.update(roi);
@@ -63,6 +77,27 @@ public class AiRoiServiceImpl implements IAiRoiService {
}
}
/**
* 解析 deviceId：
* 1. 优先从同摄像头的已有 ROI 继承
* 2. 否则取系统中唯一的已注册边缘设备（单边缘场景）
*/
private String resolveDeviceId(String cameraId) {
// 从同摄像头已有 ROI 继承
String deviceId = roiMapper.queryDeviceIdByCameraId(cameraId);
if (deviceId != null && !deviceId.isEmpty()) {
return deviceId;
}
// 取系统中第一个已注册的边缘设备
List<AiEdgeDevice> devices = edgeDeviceMapper.queryAll();
if (devices != null && !devices.isEmpty()) {
log.info("[AiRoi] 使用默认边缘设备: {}", devices.get(0).getDeviceId());
return devices.get(0).getDeviceId();
}
log.warn("[AiRoi] 无已注册边缘设备，deviceId 为空");
return null;
}
@Override
@Transactional
public void delete(String roiId) {