forked from uav/ard-uav-server
ard-sample/src/main/java/com/ard/sample/manage/service/impl/DeviceServiceImpl.java
@@ -1,6 +1,14 @@ package com.ard.sample.manage.service.impl; import com.alibaba.fastjson2.JSONObject; import com.ard.common.core.domain.model.LoginUser; import com.ard.common.utils.SecurityUtils; import com.ard.device.camera.domain.ArdCamera; import com.ard.device.camera.service.IArdCameraService; import com.ard.device.channel.domain.ArdChannel; import com.ard.sample.component.websocket.enums.MessageType; import com.ard.sample.component.websocket.handler.DockWebSocketHandler; import com.ard.sample.component.websocket.handler.LiveStreamHandler; import com.ard.system.authorize.service.AuthorizeService; import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper; @@ -34,12 +42,14 @@ import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; import lombok.extern.slf4j.Slf4j; import org.apache.poi.ss.formula.functions.T; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.dao.DuplicateKeyException; import org.springframework.integration.annotation.ServiceActivator; import org.springframework.integration.mqtt.support.MqttHeaders; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHeaders; import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Service; import org.springframework.transaction.annotation.Transactional; import org.springframework.util.CollectionUtils; @@ -52,6 +62,7 @@ import java.time.ZoneId; import java.util.*; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.TimeUnit; import java.util.stream.Collectors; import static com.ard.sample.component.mqtt.model.TopicConst.*; @@ -65,6 +76,15 @@ @Slf4j @Transactional public class DeviceServiceImpl implements IDeviceService { @Autowired private DockWebSocketHandler _dockWebSocketHandler; @Autowired private LiveStreamHandler _liveStreamHandler; @Autowired private IArdCameraService _cameraService; @Autowired private IMessageSenderService messageSender; @@ -96,9 +116,9 @@ @Resource private AuthorizeService authorizeService; // 机场1：OSD 按机型分发（原为 @Qualifier("gatewayOSDServiceImpl") ITSAService tsaService，直连 dock2 链头） @Autowired @Qualifier("gatewayOSDServiceImpl") private ITSAService tsaService; private DockOsdDispatcher dockOsdDispatcher; @Resource private ISysDeptService sysDeptService; @@ -339,8 +359,13 @@ return; } DeviceDTO subDevice = getDevicesByParams(DeviceQueryParam.builder().deviceSn(gateway.getChildDeviceSn()).build()).get(0); List<DeviceDTO> subDevices = getDevicesByParams(DeviceQueryParam.builder().deviceSn(gateway.getChildDeviceSn()).build()); if (CollectionUtils.isEmpty(subDevices)) { log.warn("网关 {} 的子设备 {} 不存在，跳过拓扑拼接", gateway.getDeviceSn(), gateway.getChildDeviceSn()); return; } DeviceDTO subDevice = subDevices.get(0); subDevice.setStatus(RedisOpsUtils.checkExist(RedisConst.DEVICE_ONLINE_PREFIX + subDevice.getDeviceSn())); gateway.setChildren(subDevice); @@ -350,11 +375,30 @@ //region 单独写一个方法做权限验证，由于有好多定时任务也在实时获取设备列表，原方法不适用再添加权限注解 @Override public List<DeviceDTO> getDockList() { public List<DeviceDTO> getDockList(String nickname) { LoginUser loginUser = SecurityUtils.getLoginUser(); List<DeviceDTO> devicesList = this.getDockListByParams( DeviceQueryParam.builder() .domains(List.of(DeviceDomainEnum.GATEWAY.getVal(), DeviceDomainEnum.DOCK.getVal())) .nickname(nickname) .build()); // 在獲取设备列表时，顺便获取直播地址 for(DeviceDTO devices : devicesList){ // 暂时注释掉机场设备视频直址 2026-04-01 // _liveStreamHandler.StartLiveStreaming(devices.getDeviceSn(), loginUser.getUserId().toString()); // _liveStreamHandler.StartLiveStreaming(devices.getChildDeviceSn(), loginUser.getUserId().toString()); ArdCamera ardCamera = new ArdCamera(); ardCamera.setDockId(devices.getId()); List<ArdCamera> ardCameraList = _cameraService.selectArdCameraList(ardCamera); for(ArdCamera ardCamera1 : ardCameraList){ for(ArdChannel ardChannel : ardCamera1.getChannelList()) { if(ardChannel.getChanNo() == 1) { // 修改列表中外接相机地址 updateCameraWebrtcUrl(devices.getDeviceSn(), ardChannel.getArdVtdu().getWebrtcUrl(), ardCamera1.getState()); } } } } devicesList.stream() .filter(gateway -> DeviceDomainEnum.DOCK.getVal() == gateway.getDomain() || @@ -371,8 +415,13 @@ if (!StringUtils.hasText(gateway.getChildDeviceSn())) { return; } DeviceDTO subDevice = getDevicesByParams(DeviceQueryParam.builder().deviceSn(gateway.getChildDeviceSn()).build()).get(0); List<DeviceDTO> subDevices = getDevicesByParams(DeviceQueryParam.builder().deviceSn(gateway.getChildDeviceSn()).build()); if (CollectionUtils.isEmpty(subDevices)) { log.warn("网关 {} 的子设备 {} 不存在，跳过拓扑拼接", gateway.getDeviceSn(), gateway.getChildDeviceSn()); return; } DeviceDTO subDevice = subDevices.get(0); subDevice.setStatus(RedisOpsUtils.checkExist(RedisConst.DEVICE_ONLINE_PREFIX + subDevice.getDeviceSn())); gateway.setChildren(subDevice); } @@ -398,6 +447,7 @@ }) .eq(param.getBoundStatus() != null, DeviceEntity::getBoundStatus, param.getBoundStatus()) .like(StringUtils.hasText(param.getNickname()), DeviceEntity::getNickname, "%" + param.getNickname() + "%") .orderBy(param.isOrderBy(), param.isAsc(), DeviceEntity::getId)) .stream() @@ -405,6 +455,90 @@ .collect(Collectors.toList()); } //endregion // region 多路直播列表，按nickname筛选 // /** // * 多路直播列表，按nickname筛选 // * */ // @Override // public List<DeviceDTO> getDockCameraList(String nickname) { // // List<DeviceDTO> devicesList = this.getDockListByParams( // DeviceQueryParam.builder() // .domains(List.of(DeviceDomainEnum.GATEWAY.getVal(), DeviceDomainEnum.DOCK.getVal())) // .nickname(nickname) // .build()); // // // 在獲取设备列表时，顺便获取直播地址 // for(DeviceDTO devices : devicesList){ // _liveStreamHandler.StartLiveStreaming(devices.getDeviceSn()); // _liveStreamHandler.StartLiveStreaming(devices.getChildDeviceSn()); // ArdCamera ardCamera = new ArdCamera(); // ardCamera.setDockId(devices.getId()); // List<ArdCamera> ardCameraList = _cameraService.selectArdCameraList(ardCamera); // for(ArdCamera ardCamera1 : ardCameraList){ // for(ArdChannel ardChannel : ardCamera1.getChannelList()) { // if(ardChannel.getChanNo() == 1) { // // 修改列表中外接相机地址 // updateCameraWebrtcUrl(devices.getDeviceSn(), ardChannel.getArdVtdu().getWebrtcUrl()); // } // } // } // } // // devicesList.stream() // .filter(gateway -> DeviceDomainEnum.DOCK.getVal() == gateway.getDomain() || // RedisOpsUtils.checkExist(RedisConst.DEVICE_ONLINE_PREFIX + gateway.getDeviceSn())) // .forEach(this::spliceDeviceTopoPer); // // return devicesList; // } // @Override // public void spliceDeviceTopoCamera(DeviceDTO gateway) { // gateway.setStatus(RedisOpsUtils.checkExist(RedisConst.DEVICE_ONLINE_PREFIX + gateway.getDeviceSn())); // // sub device // if (!StringUtils.hasText(gateway.getChildDeviceSn())) { // return; // } // DeviceDTO subDevice = // getDevicesByParams(DeviceQueryParam.builder().deviceSn(gateway.getChildDeviceSn()).build()).get(0); // subDevice.setStatus(RedisOpsUtils.checkExist(RedisConst.DEVICE_ONLINE_PREFIX + subDevice.getDeviceSn())); // gateway.setChildren(subDevice); // } // /** // * 按nickname筛选列表 // * */ // @Override // public List<DeviceDTO> getDockCameraListByParams(DeviceQueryParam param) { // return mapper.selectListPer( // new LambdaQueryWrapper<DeviceEntity>() // .eq(StringUtils.hasText(param.getDeviceSn()), // DeviceEntity::getDeviceSn, param.getDeviceSn()) // .eq(param.getDeviceType() != null, // DeviceEntity::getDeviceType, param.getDeviceType()) // .eq(param.getSubType() != null, // DeviceEntity::getSubType, param.getSubType()) // .eq(StringUtils.hasText(param.getChildSn()), // DeviceEntity::getChildSn, param.getChildSn()) // .and(!CollectionUtils.isEmpty(param.getDomains()), wrapper -> { // for (Integer domain : param.getDomains()) { // wrapper.eq(DeviceEntity::getDomain, domain).or(); // } // }) // .eq(param.getBoundStatus() != null, DeviceEntity::getBoundStatus, // param.getBoundStatus()) // .like(param.getNickname()!=null, DeviceEntity::getNickname, "%" + param.getNickname() + "%") // .orderBy(param.isOrderBy(), // param.isAsc(), DeviceEntity::getId)) // .stream() // .map(this::deviceEntityConvertToDTO) // .collect(Collectors.toList()); // } // endregion @Override public Optional<TopologyDeviceDTO> getDeviceTopoForPilot(String sn) { @@ -538,7 +672,8 @@ .getValueAndUserType(UserTypeEnum.WEB.getVal()); tsaService.handleOSD(receiver, device, webSessions, wsMessage); // 机场1：OSD 按机型分发（dock1=67 → dock1 链；其余透传 dock2 原链） dockOsdDispatcher.dispatch(receiver, device, webSessions, wsMessage); } catch (NullPointerException e) { log.error("处理OSD消息NPE，headers={}, payload={}", @@ -622,7 +757,22 @@ } // 调用验证方法 return mapper.insert(entity) > 0 ? Optional.of(entity) : Optional.empty(); try { return mapper.insert(entity) > 0 ? Optional.of(entity) : Optional.empty(); } catch (DuplicateKeyException e) { // 机场首次上线时 MQTT QoS1 可能重复投递 update_topo（或多线程并发）， // 两个线程同时查到“设备不存在”后会并发 insert，后插入者触发唯一键冲突。 // 此时数据已被另一线程写入，退化为更新即可，避免整条状态消息处理失败。 DeviceEntity exist = mapper.selectOne( new LambdaQueryWrapper<DeviceEntity>() .eq(DeviceEntity::getDeviceSn, entity.getDeviceSn())); if (exist == null) { throw e; } entity.setId(exist.getId()); mapper.updateById(entity); return Optional.of(exist); } } /** @@ -753,6 +903,9 @@ .firmwareVersion(entity.getFirmwareVersion()) .deptId(entity.getDeptId()) .workType(entity.getWorkType()) .webrtcUrl(entity.getWebrtcUrl()) .cameraWebrtcUrl(entity.getCameraWebrtcUrl()) .cameraState(entity.getCameraState()) .workspaceName(entity.getDeptId() != null ? Optional.ofNullable(sysDeptService.selectDeptById(entity.getDeptId())) .map(SysDept::getDeptName) @@ -803,6 +956,27 @@ int update = mapper.update(null, lambdaUpdateWrapper); return update > 0; } @Override public Boolean updateWebrtcUrl(String sn, String webrtcUrl){ LambdaUpdateWrapper<DeviceEntity> lambdaUpdateWrapper = new LambdaUpdateWrapper<>(); lambdaUpdateWrapper.eq(DeviceEntity::getDeviceSn, sn) .set(DeviceEntity::getWebrtcUrl, webrtcUrl); int update = mapper.update(null, lambdaUpdateWrapper); return update > 0; } public Boolean updateCameraWebrtcUrl(String sn, String CameraWebrtcUrl, String state){ LambdaUpdateWrapper<DeviceEntity> lambdaUpdateWrapper = new LambdaUpdateWrapper<>(); lambdaUpdateWrapper.eq(DeviceEntity::getDeviceSn, sn) .set(DeviceEntity::getCameraWebrtcUrl, CameraWebrtcUrl) .set(DeviceEntity::getCameraState, state); int update = mapper.update(null, lambdaUpdateWrapper); return update > 0; } @Override public Boolean bindDevice(DeviceDTO device) { @@ -1041,34 +1215,41 @@ .data(Map.of(MapKeyConst.DEVICES, deviceOtaFirmwares)) .build()); if (serviceReply.getResult() != ResponseResult.CODE_SUCCESS) { return ResponseResult.error(serviceReply.getResult(), "Firmware Error Code: " + serviceReply.getResult()); String errorMsg = CommonErrorEnum.getMsgByCode(serviceReply.getResult()); return ResponseResult.error(serviceReply.getResult(), errorMsg); } // Record the device state that needs to be updated. deviceOtaFirmwares.forEach(deviceOta -> RedisOpsUtils.setWithExpire( RedisConst.FIRMWARE_UPGRADING_PREFIX + deviceOta.getSn(), bid, RedisConst.DEVICE_ALIVE_SECOND * RedisConst.DEVICE_ALIVE_SECOND)); bid,RedisConst.DEVICE_ALIVE_SECOND * RedisConst.DEVICE_ALIVE_SECOND)); return ResponseResult.success(); } /** * 属性设置 * */ @Override public void devicePropertySet(String dockSn, DeviceSetPropertyEnum propertyEnum, JsonNode param) { public ResponseResult devicePropertySet(String dockSn, DeviceSetPropertyEnum propertyEnum, JsonNode param) { boolean dockOnline = this.checkDeviceOnline(dockSn); if (!dockOnline) { throw new RuntimeException("Dock is offline."); return ResponseResult.error(CommonErrorEnum.DEVICE_OFFLINE.getErrorCode(),"机场" + CommonErrorEnum.DEVICE_OFFLINE.getErrorMsg()); //throw new RuntimeException(CommonErrorEnum.DEVICE_OFFLINE.getErrorMsg()); } DeviceDTO deviceDTO = (DeviceDTO) RedisOpsUtils.get(RedisConst.DEVICE_ONLINE_PREFIX + dockSn); boolean deviceOnline = this.checkDeviceOnline(deviceDTO.getChildDeviceSn()); if (!deviceOnline) { throw new RuntimeException("Device is offline."); return ResponseResult.error(2222,"无人机" + CommonErrorEnum.DEVICE_OFFLINE.getErrorMsg()); //throw new RuntimeException("Device is offline."); // CommonErrorEnum.DEVICE_OFFLINE.getErrorCode() } // Make sure the data is valid. BasicDeviceProperty basicDeviceProperty = objectMapper.convertValue(param, propertyEnum.getClazz()); boolean valid = basicDeviceProperty.valid(); if (!valid) { throw new IllegalArgumentException(CommonErrorEnum.ILLEGAL_ARGUMENT.getErrorMsg()); return ResponseResult.error(CommonErrorEnum.ILLEGAL_ARGUMENT.getErrorMsg()); // throw new IllegalArgumentException(CommonErrorEnum.ILLEGAL_ARGUMENT.getErrorMsg()); } String topic = THING_MODEL_PRE + PRODUCT + dockSn + PROPERTY_SUF + SET_SUF; @@ -1076,12 +1257,14 @@ (OsdSubDeviceReceiver) RedisOpsUtils.get(RedisConst.OSD_PREFIX + deviceDTO.getChildDeviceSn()); if (!param.isObject()) { this.deviceOnePropertySet(topic, propertyEnum, Map.entry(propertyEnum.getProperty(), param)); return; return ResponseResult.success(); // return; } if(param.size()==1){ this.deviceOnePropertySetSilent(topic,propertyEnum,param); return; return ResponseResult.success(); // return; } // if(Objects.equals(param.fieldNames().next(), "silent_mode")) // { @@ -1098,7 +1281,7 @@ this.deviceOnePropertySet(topic, propertyEnum, Map.entry(propertyEnum.getProperty(), node)); } return ResponseResult.success(); } /** @@ -1193,6 +1376,9 @@ .domain(dto.getDomain()) .firmwareVersion(dto.getFirmwareVersion()) .workType(dto.getWorkType()) .webrtcUrl(dto.getWebrtcUrl()) .cameraWebrtcUrl(dto.getCameraWebrtcUrl()) .cameraState(dto.getCameraState()) .compatibleStatus(dto.getFirmwareStatus() == null ? null : DeviceFirmwareStatusEnum.CONSISTENT_UPGRADE != DeviceFirmwareStatusEnum.find(dto.getFirmwareStatus())) .build(); @@ -1334,9 +1520,48 @@ messageSender.publish(topic, builder.build()); } // @DataScopePlus(deptAlias="d") // @Override // public List<DeviceEntity> selectList(Wrapper<DeviceEntity> queryWrapper) { // return mapper.selectList(queryWrapper); // } /** * 定时任务，每秒钟从redis中读取信息发送机场摄像头（舱内舱外）、飞行器摄像头模式（广角、变焦、红外） * */ @Override @Scheduled(initialDelay = 5, fixedRate = 1, timeUnit = TimeUnit.SECONDS) public void sendCameraInfo() { // 此方法修改为定时任务，每秒钟从redis中读取信息发送 // 发送机场摄像头位置信息 Map<Object, Object> map = RedisOpsUtils.getAllData(RedisConst.COVER_STATE_PREFIX+"*"); map.forEach((k, v) -> { _dockWebSocketHandler.broadcastMessageToAllClients(v.toString()); }); } // 机场直播摄像头信息存redis中，舱内、舱外 @Override public void setRedisCameraPosition(String sn, Integer cameraPosition) { String liveStatus = ""; Optional<DeviceDTO> device = getDeviceBySn(sn); if(device.get().getDomain().equals(DeviceDomainEnum.DOCK.getVal())){ CommonTopicReceiver<Map<String,Integer>> stateReceiver = new CommonTopicReceiver<>(); stateReceiver.setTid(UUID.randomUUID().toString()); stateReceiver.setBid(UUID.randomUUID().toString()); stateReceiver.setType(MessageType.LIVESTATUS.getType()); stateReceiver.setGateway(sn); stateReceiver.setTimestamp(System.currentTimeMillis()); Map<String,Integer> live_status = new HashMap<>(); live_status.put("camera_position", cameraPosition); stateReceiver.setData(live_status); try{ liveStatus=objectMapper.writeValueAsString(stateReceiver); } catch (Exception e){ log.error("机场直播设备状态反馈转换异常:{}",e); } } log.info("机场直播设备状态反馈存入redis中，key值：{}中:{}",RedisConst.COVER_STATE_PREFIX,liveStatus); RedisOpsUtils.set(RedisConst.COVER_STATE_PREFIX + sn, liveStatus); } public void setChildRedisCamera(String sn, String context){ // 保存子设备信息到redis中，便于前端获取 RedisOpsUtils.set(RedisConst.COVER_STATE_PREFIX + sn, context); } }
