forked from drone/drone-java
src/main/java/com/dji/sample/wayline/service/impl/WaylineJobServiceImpl.java
@@ -1,22 +1,40 @@ package com.dji.sample.wayline.service.impl; import com.alibaba.fastjson.JSON; import com.alibaba.fastjson.JSONObject; import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper; import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper; import com.baomidou.mybatisplus.core.toolkit.Wrappers; import com.baomidou.mybatisplus.extension.conditions.query.LambdaQueryChainWrapper; import com.baomidou.mybatisplus.extension.conditions.update.LambdaUpdateChainWrapper; import com.baomidou.mybatisplus.extension.plugins.pagination.Page; import com.dji.sample.common.error.CommonErrorEnum; import com.dji.sample.common.model.CustomClaim; import com.dji.sample.common.model.Pagination; import com.dji.sample.common.model.PaginationData; import com.dji.sample.common.model.ResponseResult; import com.dji.sample.common.util.MinioUrlUtils; import com.dji.sample.common.util.SpringBeanUtils; import com.dji.sample.component.mqtt.model.*; import com.dji.sample.component.mqtt.service.IMessageSenderService; import com.dji.sample.component.redis.RedisConst; import com.dji.sample.component.redis.RedisOpsUtils; import com.dji.sample.control.model.param.DrcModeParam; import com.dji.sample.control.model.dto.PointDTO; import com.dji.sample.control.model.enums.CameraModeEnum; import com.dji.sample.control.model.enums.DroneAuthorityEnum; import com.dji.sample.control.model.enums.PayloadCommandsEnum; import com.dji.sample.control.model.param.*; import com.dji.sample.control.service.IControlService; import com.dji.sample.control.service.IDrcService; import com.dji.sample.geo.entity.GeoJson; import com.dji.sample.log.dao.IDroneFlightLogMapper; import com.dji.sample.log.model.entity.DroneFlightLogEntity; import com.dji.sample.manage.model.dto.DeviceDTO; import com.dji.sample.manage.model.enums.DeviceModeCodeEnum; import com.dji.sample.manage.model.enums.DockModeCodeEnum; import com.dji.sample.manage.model.enums.DroneRcLostActionEnum; import com.dji.sample.manage.model.enums.WaylineRcLostActionEnum; import com.dji.sample.manage.model.receiver.OsdDockReceiver; import com.dji.sample.manage.model.receiver.OsdSubDeviceReceiver; import com.dji.sample.manage.service.IDeviceRedisService; @@ -24,16 +42,15 @@ import com.dji.sample.media.model.MediaFileCountDTO; import com.dji.sample.media.model.MediaMethodEnum; import com.dji.sample.media.service.IFileService; import com.dji.sample.geo.utils.GeoUtils; import com.dji.sample.wayline.dao.IWaylineJobMapper; import com.dji.sample.wayline.model.dto.*; import com.dji.sample.wayline.model.entity.WaylineJobBreakPointEntity; import com.dji.sample.wayline.model.entity.WaylineJobEntity; import com.dji.sample.wayline.model.enums.WaylineErrorCodeEnum; import com.dji.sample.wayline.model.enums.WaylineJobStatusEnum; import com.dji.sample.wayline.model.enums.WaylineMethodEnum; import com.dji.sample.wayline.model.enums.WaylineTaskTypeEnum; import com.dji.sample.wayline.model.param.CreateJobParam; import com.dji.sample.wayline.model.param.UpdateJobParam; import com.dji.sample.wayline.model.enums.*; import com.dji.sample.wayline.model.param.*; import com.dji.sample.wayline.service.IWaylineFileService; import com.dji.sample.wayline.service.IWaylineJobBreakPointService; import com.dji.sample.wayline.service.IWaylineJobService; import com.dji.sample.wayline.service.IWaylineRedisService; import com.fasterxml.jackson.core.type.TypeReference; @@ -49,6 +66,7 @@ import org.springframework.util.CollectionUtils; import org.springframework.util.StringUtils; import java.io.File; import java.net.URL; import java.sql.SQLException; import java.time.*; @@ -92,6 +110,15 @@ @Autowired private IWaylineRedisService waylineRedisService; @Autowired private IControlService controlService; @Autowired private IDroneFlightLogMapper flightLogMapper; @Autowired private IWaylineJobBreakPointService waylineJobBreakPointService; private Optional<WaylineJobDTO> insertWaylineJob(WaylineJobEntity jobEntity) { int id = mapper.insert(jobEntity); if (id <= 0) { @@ -122,7 +149,9 @@ .rthAltitude(param.getRthAltitude()) .mediaCount(0) .build(); if (StringUtils.hasText(param.getJobId())){ jobEntity.setJobId(param.getJobId()); } return insertWaylineJob(jobEntity); } @@ -144,99 +173,172 @@ } /** * For immediate tasks, the server time shall prevail. * 对于即时任务，以服务器时间为准。 * * @param param */ private void fillImmediateTime(CreateJobParam param) { if (WaylineTaskTypeEnum.IMMEDIATE != param.getTaskType()) { return; } long now = System.currentTimeMillis() / 1000; if (CollectionUtils.isEmpty(param.getTaskDays())) { param.setTaskDays(List.of(now)); } if (CollectionUtils.isEmpty(param.getTaskPeriods())) { param.setTaskPeriods(List.of(List.of(now))); } long now = System.currentTimeMillis() / 1000 - 10; // param.setTaskDays(Collections.singletonList(now)); // param.setTaskPeriods(Collections.singletonList(Collections.singletonList(now))); param.setTaskDays(List.of(now)); param.setTaskPeriods(List.of(List.of(now))); } @Override public ResponseResult publishFlightTask(CreateJobParam param, CustomClaim customClaim) throws SQLException { fillImmediateTime(param); // param.getTaskDays().sort((a, b) -> (int) (a - b)); // param.getTaskPeriods().sort((a, b) -> (int) (a.get(0) - b.get(0))); for (Long taskDay : param.getTaskDays()) { LocalDate date = LocalDate.ofInstant(Instant.ofEpochSecond(taskDay), ZoneId.systemDefault()); for (List<Long> taskPeriod : param.getTaskPeriods()) { long beginTime = LocalDateTime.of(date, LocalTime.ofInstant(Instant.ofEpochSecond(taskPeriod.get(0)), ZoneId.systemDefault())) .atZone(ZoneId.systemDefault()).toInstant().toEpochMilli(); long endTime = taskPeriod.size() > 1 ? long endTime = taskPeriod.size() > 1 && Objects.nonNull(taskPeriod.get(1)) ? LocalDateTime.of(date, LocalTime.ofInstant(Instant.ofEpochSecond(taskPeriod.get(1)), ZoneId.systemDefault())) .atZone(ZoneId.systemDefault()).toInstant().toEpochMilli() : beginTime; if (WaylineTaskTypeEnum.IMMEDIATE != param.getTaskType() && endTime < System.currentTimeMillis()) { return ResponseResult.error("任务已过期"); } Optional<WaylineJobDTO> waylineJobOpt = this.createWaylineJob(param, customClaim.getWorkspaceId(), customClaim.getUsername(), beginTime, endTime); if (waylineJobOpt.isEmpty()) { throw new SQLException("Failed to create wayline job."); throw new SQLException("任务创建失败"); } WaylineJobDTO waylineJob = waylineJobOpt.get(); // If it is a conditional task type, add conditions to the job parameters. addConditions(waylineJob, param, beginTime, endTime); //如果是条件任务类型，需要在任务参数中添加条件。 addPreparedJob(waylineJob, param, beginTime, endTime); return this.publishOneFlightTask(waylineJob); ResponseResult response = this.publishOneFlightTask(waylineJob); if (ResponseResult.CODE_SUCCESS != response.getCode()) { return response; } } } return ResponseResult.error(); return ResponseResult.success(); } private void addConditions(WaylineJobDTO waylineJob, CreateJobParam param, Long beginTime, Long endTime) { if (WaylineTaskTypeEnum.CONDITION != param.getTaskType()) { return; @Override public ResponseResult publishFlightTaskCondition(CreateJobParam param, CustomClaim customClaim) throws SQLException { //开始日期 LocalDate startDate = LocalDate.ofInstant(Instant.ofEpochSecond(param.getTaskDays().get(0)), ZoneId.systemDefault()); long start = LocalDateTime.of(startDate, LocalTime.ofInstant(Instant.ofEpochSecond(param.getExecuteStartTimeArr().get(0).get(0)), ZoneId.systemDefault())) .atZone(ZoneId.systemDefault()).toInstant().toEpochMilli(); //结束日期 LocalDate endDate = LocalDate.ofInstant(Instant.ofEpochSecond(param.getTaskDays().get(1)), ZoneId.systemDefault()); long end = LocalDateTime.of(endDate, LocalTime.ofInstant(Instant.ofEpochSecond(param.getExecuteStartTimeArr().get(param.getExecuteStartTimeArr().size() - 1).get(0)), ZoneId.systemDefault())) .atZone(ZoneId.systemDefault()).toInstant().toEpochMilli(); //保存数据 WaylineJobEntity waylineJobEntity = WaylineJobEntity.builder() .jobId(UUID.randomUUID().toString()) .name(param.getName()) .dockSn(param.getDockSn()) .fileId(param.getFileId()) .username(customClaim.getUsername()) .workspaceId(customClaim.getWorkspaceId()) .beginTime(start) .endTime(end) .status(WaylineJobStatusEnum.PENDING.getVal()) .taskType(param.getTaskType().getVal()) .waylineType(param.getWaylineType().getVal()) .outOfControlAction(param.getOutOfControlAction()) .batteryCapacity(param.getMinBatteryCapacity()) .rthAltitude(param.getRthAltitude()) .mediaCount(0) .repFreVal(param.getRepFreVal()) .repFreType(param.getRepFreType()) .repRuleType(param.getRepRuleType()) .repRuleVal(param.getRepRuleVal()) .executeTimeArr(param.getTaskPeriods()) .executeStartTimeArr(param.getExecuteStartTimeArr()) .build(); Optional<WaylineJobDTO> waylineJobOpt = insertWaylineJob(waylineJobEntity); if (waylineJobOpt.isEmpty()) { throw new SQLException("任务创建失败"); } waylineJob.setConditions( WaylineTaskConditionDTO.builder() .executableConditions(Objects.nonNull(param.getMinStorageCapacity()) ? WaylineTaskExecutableConditionDTO.builder().storageCapacity(param.getMinStorageCapacity()).build() : null) .readyConditions(WaylineTaskReadyConditionDTO.builder() .batteryCapacity(param.getMinBatteryCapacity()) .beginTime(beginTime) .endTime(endTime) .build()) .build()); WaylineJobDTO waylineJob = waylineJobOpt.get(); waylineRedisService.setConditionalWaylineJob(waylineJob); // key: wayline_job_condition, value: {workspace_id}:{dock_sn}:{job_id} boolean isAdd = waylineRedisService.addPrepareConditionalWaylineJob(waylineJob); //存一条记录, List<Long> timeArr = param.getExecuteStartTimeArr().get(0); LocalDate date = LocalDate.ofInstant(Instant.ofEpochSecond(timeArr.get(0)), ZoneId.systemDefault()); LocalDateTime beginTime = LocalDateTime.of(date, LocalTime.ofInstant(Instant.ofEpochSecond(timeArr.get(0)), ZoneId.systemDefault())); LocalDateTime endTime = timeArr.size() > 1 && Objects.nonNull(timeArr.get(1)) ? LocalDateTime.of(date, LocalTime.ofInstant(Instant.ofEpochSecond(timeArr.get(1)), ZoneId.systemDefault())) : beginTime; if (WaylineTaskTypeEnum.IMMEDIATE != param.getTaskType() && endTime.atZone(ZoneId.systemDefault()).toInstant().toEpochMilli() < System.currentTimeMillis()) { return ResponseResult.error("任务已过期"); } //条件任务 if (param.getTaskType() == WaylineTaskTypeEnum.CONDITION) { //如果是条件任务类型，需要在任务参数中添加条件。 waylineJob.setConditions( WaylineTaskConditionDTO.builder() .executableConditions(Objects.nonNull(param.getMinStorageCapacity()) ? WaylineTaskExecutableConditionDTO.builder().storageCapacity(param.getMinStorageCapacity()).build() : null) .readyConditions(WaylineTaskReadyConditionDTO.builder() .batteryCapacity(param.getMinBatteryCapacity()) .beginTime(beginTime.atZone(ZoneId.systemDefault()).toInstant().toEpochMilli()) .endTime(endTime.atZone(ZoneId.systemDefault()).toInstant().toEpochMilli()) .build()) .build()); } else { waylineJob.setBeginTime(beginTime); } ResponseResult response = this.publishOneFlightTask(waylineJob); if (ResponseResult.CODE_SUCCESS != response.getCode()) { return response; } return ResponseResult.success(); } private void addPreparedJob(WaylineJobDTO waylineJob, CreateJobParam param, Long beginTime, Long endTime) { if (WaylineTaskTypeEnum.CONDITION == param.getTaskType()) { waylineJob.setConditions( WaylineTaskConditionDTO.builder() .executableConditions(Objects.nonNull(param.getMinStorageCapacity()) ? WaylineTaskExecutableConditionDTO.builder().storageCapacity(param.getMinStorageCapacity()).build() : null) .readyConditions(WaylineTaskReadyConditionDTO.builder() .batteryCapacity(param.getMinBatteryCapacity()) .beginTime(beginTime) .endTime(endTime) .build()) .build()); waylineRedisService.setConditionalWaylineJob(waylineJob); } // value: {workspace_id}:{dock_sn}:{job_id} boolean isAdd = waylineRedisService.addPreparedWaylineJob(waylineJob); if (!isAdd) { throw new RuntimeException("Failed to create conditional job."); throw new RuntimeException("创建任务失败。"); } } public ResponseResult publishOneFlightTask(WaylineJobDTO waylineJob) throws SQLException { boolean isOnline = deviceRedisService.checkDeviceOnline(waylineJob.getDockSn()); if (!isOnline) { throw new RuntimeException("Dock is offline."); } boolean isSuccess = this.prepareFlightTask(waylineJob); if (!isSuccess) { return ResponseResult.error("Failed to prepare job."); return ResponseResult.error("任务准备失败"); } // Issue an immediate task execution command. if (WaylineTaskTypeEnum.IMMEDIATE.getVal() == waylineJob.getTaskType()) { if (!executeFlightTask(waylineJob.getWorkspaceId(), waylineJob.getJobId())) { return ResponseResult.error("Failed to execute job."); } } if (WaylineTaskTypeEnum.TIMED.getVal() == waylineJob.getTaskType()) { // key: wayline_job_timed, value: {workspace_id}:{dock_sn}:{job_id} boolean isAdd = RedisOpsUtils.zAdd(RedisConst.WAYLINE_JOB_TIMED_EXECUTE, waylineJob.getWorkspaceId() + RedisConst.DELIMITER + waylineJob.getDockSn() + RedisConst.DELIMITER + waylineJob.getJobId(), waylineJob.getBeginTime().atZone(ZoneId.systemDefault()).toInstant().toEpochMilli()); if (!isAdd) { return ResponseResult.error("Failed to create scheduled job."); //发出立即任务执行命令 if (WaylineTaskTypeEnum.IMMEDIATE == waylineJob.getTaskType()) { boolean isExecuted = executeFlightTask(waylineJob.getWorkspaceId(), waylineJob.getJobId()); if (!isExecuted) { return ResponseResult.error("当前状态不支持飞行，请检查是否已有正在执行的飞行任务！"); } } @@ -244,29 +346,45 @@ } private Boolean prepareFlightTask(WaylineJobDTO waylineJob) throws SQLException { /*boolean isOnline = deviceRedisService.checkDeviceOnline(waylineJob.getDockSn()); if (!isOnline) { throw new RuntimeException("设备离线。"); }*/ // get wayline file Optional<WaylineFileDTO> waylineFile = waylineFileService.getWaylineByWaylineId(waylineJob.getWorkspaceId(), waylineJob.getFileId()); if (waylineFile.isEmpty()) { throw new SQLException("Wayline file doesn't exist."); throw new SQLException("航线文件不存在。"); } // get file url //获取航线文件地址 URL url = waylineFileService.getObjectUrl(waylineJob.getWorkspaceId(), waylineFile.get().getWaylineId()); WaylineTaskCreateDTO flightTask = WaylineTaskCreateDTO.builder() .flightId(waylineJob.getJobId()) WaylineTaskCreateDTO.WaylineTaskCreateDTOBuilder flightTaskBuilder = WaylineTaskCreateDTO.builder().flightId(waylineJob.getJobId()) .executeTime(waylineJob.getBeginTime().atZone(ZoneId.systemDefault()).toInstant().toEpochMilli()) .taskType(waylineJob.getTaskType()) .waylineType(waylineJob.getWaylineType()) .rthAltitude(waylineJob.getRthAltitude()) .outOfControlAction(waylineJob.getOutOfControlAction()) .file(WaylineTaskFileDTO.builder() .url(url.toString()) .url(MinioUrlUtils.getUrl(url)) .fingerprint(waylineFile.get().getSign()) .build()) .build(); .build()); WaylineJobBreakPointEntity entity = waylineJob.getWaylineJobBreakPointEntity(); if (null != entity) { flightTaskBuilder.breakPoint(BreakPointJobDTO.builder() .index(entity.getBpIndex()) .state(entity.getState()) .progress(entity.getProgress()) .waylineId(entity.getWaylineId()) .build()); } if (WaylineTaskTypeEnum.CONDITION.getVal() == waylineJob.getTaskType()) { WaylineTaskCreateDTO flightTask = flightTaskBuilder.build(); //当任务类型为条件时 if (WaylineTaskTypeEnum.CONDITION == waylineJob.getTaskType()) { if (Objects.isNull(waylineJob.getConditions())) { throw new IllegalArgumentException(); } @@ -274,10 +392,12 @@ flightTask.setExecutableConditions(waylineJob.getConditions().getExecutableConditions()); } //发布飞行指令 ServiceReply serviceReply = messageSender.publishServicesTopic( waylineJob.getDockSn(), WaylineMethodEnum.FLIGHT_TASK_PREPARE.getMethod(), flightTask, waylineJob.getJobId()); if (ResponseResult.CODE_SUCCESS != serviceReply.getResult()) { log.info("Prepare task ====> Error code: {}", serviceReply.getResult()); //飞行失败，更新数据库信息 this.updateJob(WaylineJobDTO.builder() .workspaceId(waylineJob.getWorkspaceId()) .jobId(waylineJob.getJobId()) @@ -295,12 +415,12 @@ // get job Optional<WaylineJobDTO> waylineJob = this.getJobByJobId(workspaceId, jobId); if (waylineJob.isEmpty()) { throw new IllegalArgumentException("Job doesn't exist."); throw new IllegalArgumentException("任务不存在"); } boolean isOnline = deviceRedisService.checkDeviceOnline(waylineJob.get().getDockSn()); if (!isOnline) { throw new RuntimeException("Dock is offline."); throw new RuntimeException("设备离线"); } WaylineJobDTO job = waylineJob.get(); @@ -316,8 +436,8 @@ .status(WaylineJobStatusEnum.FAILED.getVal()) .completedTime(LocalDateTime.now()) .code(serviceReply.getResult()).build()); // The conditional task fails and enters the blocking status. if (WaylineTaskTypeEnum.CONDITION.getVal() == job.getTaskType() // 条件任务失败，进入阻塞状态。 if (WaylineTaskTypeEnum.CONDITION == job.getTaskType() && WaylineErrorCodeEnum.find(serviceReply.getResult()).isBlock()) { waylineRedisService.setBlockedWaylineJob(job.getDockSn(), jobId); } @@ -339,9 +459,9 @@ Set<String> waylineJobIds = waylineJobs.stream().map(WaylineJobDTO::getJobId).collect(Collectors.toSet()); // Check if the task status is correct. boolean isErr = !jobIds.removeAll(waylineJobIds) || !jobIds.isEmpty() ; boolean isErr = !jobIds.removeAll(waylineJobIds) || !jobIds.isEmpty(); if (isErr) { throw new IllegalArgumentException("These tasks have an incorrect status and cannot be canceled. " + Arrays.toString(jobIds.toArray())); throw new IllegalArgumentException("以下任务状态不正确，不能取消" + Arrays.toString(jobIds.toArray())); } // Group job id by dock sn. @@ -355,14 +475,14 @@ public void publishCancelTask(String workspaceId, String dockSn, List<String> jobIds) { boolean isOnline = deviceRedisService.checkDeviceOnline(dockSn); if (!isOnline) { throw new RuntimeException("Dock is offline."); throw new RuntimeException("设备离线"); } ServiceReply serviceReply = messageSender.publishServicesTopic( dockSn, WaylineMethodEnum.FLIGHT_TASK_CANCEL.getMethod(), Map.of(MapKeyConst.FLIGHT_IDS, jobIds)); if (ResponseResult.CODE_SUCCESS != serviceReply.getResult()) { log.info("Cancel job ====> Error code: {}", serviceReply.getResult()); throw new RuntimeException("Failed to cancel the wayline job of " + dockSn); throw new RuntimeException("航路作业取消失败 " + dockSn); } for (String jobId : jobIds) { @@ -372,18 +492,16 @@ .status(WaylineJobStatusEnum.CANCEL.getVal()) .completedTime(LocalDateTime.now()) .build()); RedisOpsUtils.zRemove(RedisConst.WAYLINE_JOB_TIMED_EXECUTE, workspaceId + RedisConst.DELIMITER + dockSn + RedisConst.DELIMITER + jobId); } } public List<WaylineJobDTO> getJobsByConditions(String workspaceId, Collection<String> jobIds, WaylineJobStatusEnum status) { return mapper.selectList( new LambdaQueryWrapper<WaylineJobEntity>() .eq(WaylineJobEntity::getWorkspaceId, workspaceId) .eq(Objects.nonNull(status), WaylineJobEntity::getStatus, status.getVal()) .and(!CollectionUtils.isEmpty(jobIds), wrapper -> jobIds.forEach(id -> wrapper.eq(WaylineJobEntity::getJobId, id).or()))) new LambdaQueryWrapper<WaylineJobEntity>() .eq(WaylineJobEntity::getWorkspaceId, workspaceId) .eq(Objects.nonNull(status), WaylineJobEntity::getStatus, status.getVal()) .in(!CollectionUtils.isEmpty(jobIds), WaylineJobEntity::getJobId, jobIds)) .stream() .map(this::entity2Dto) .collect(Collectors.toList()); @@ -399,19 +517,45 @@ } @Override public Optional<WaylineJobDTO> getJobByJobId(String workspaceId, String jobId,Boolean isBreakPoint) { WaylineJobEntity jobEntity = mapper.selectOne( new LambdaQueryWrapper<WaylineJobEntity>() .eq(WaylineJobEntity::getWorkspaceId, workspaceId) .eq(WaylineJobEntity::getJobId, jobId)); WaylineJobDTO waylineJobDTO = entity2Dto(jobEntity); if (isBreakPoint) { QueryWrapper queryWrapper = new QueryWrapper(); queryWrapper.eq("job_id", jobId); WaylineJobBreakPointEntity waylineJobBreakPointEntity = waylineJobBreakPointService.getOne(queryWrapper); waylineJobDTO.setWaylineJobBreakPointEntity(waylineJobBreakPointEntity); } //设置当前时间为执行时间 waylineJobDTO.setBeginTime(LocalDateTime.now()); return Optional.ofNullable(waylineJobDTO); } @Override public Boolean updateJob(WaylineJobDTO dto) { try { if (dto.getStatus() == 3) { this.checkNextJob(dto); } } catch (SQLException e) { throw new RuntimeException(e); } return mapper.update(this.dto2Entity(dto), new LambdaUpdateWrapper<WaylineJobEntity>() .eq(WaylineJobEntity::getJobId, dto.getJobId())) > 0; } @Override public PaginationData<WaylineJobDTO> getJobsByWorkspaceId(String workspaceId, long page, long pageSize) { Page<WaylineJobEntity> pageData = mapper.selectPage( new Page<WaylineJobEntity>(page, pageSize), new LambdaQueryWrapper<WaylineJobEntity>() .eq(WaylineJobEntity::getWorkspaceId, workspaceId) .orderByDesc(WaylineJobEntity::getId)); public PaginationData<WaylineJobDTO> getJobsByWorkspaceId(String workspaceId, long page, long pageSize, WaylineJobQueryParam waylineJobQueryParam) { Page<WaylineJobEntity> pageData = mapper.getPage(new Page<WaylineJobEntity>(page, pageSize), waylineJobQueryParam, workspaceId); List<WaylineJobDTO> records = pageData.getRecords() .stream() .map(this::entity2Dto) @@ -420,12 +564,42 @@ return new PaginationData<WaylineJobDTO>(records, new Pagination(pageData)); } @Override public PaginationData<WaylineJobDTO> getJobsByWorkspaceIdNew(String workspaceId, long page, long pageSize, WaylineJobQueryParam waylineJobQueryParam, String order) { Page<WaylineJobEntity> pageData = mapper.getPageNew(new Page<WaylineJobEntity>(page, pageSize), waylineJobQueryParam, workspaceId,order); List<WaylineJobDTO> records = pageData.getRecords() .stream() .map(this::entity2Dto) .collect(Collectors.toList()); // 是否需要断点续飞 records.forEach(wjd -> wjd.setBreakPoint(waylineJobBreakPointService.count( new LambdaQueryWrapper<WaylineJobBreakPointEntity>().eq(WaylineJobBreakPointEntity::getJobId,wjd.getJobId())) > 0)); return new PaginationData<WaylineJobDTO>(records, new Pagination(pageData)); } @Override public List<WaylineJobDTO> getChildrenJobs(String workspaceId, WaylineJobQueryParam waylineJobQueryParam) { List<WaylineJobEntity> list = mapper.getJobs(workspaceId, waylineJobQueryParam); List<WaylineJobDTO> records = list .stream() .map(this::entity2Dto) .collect(Collectors.toList()); return records; } @Override @ServiceActivator(inputChannel = ChannelName.INBOUND_REQUESTS_FLIGHT_TASK_RESOURCE_GET, outputChannel = ChannelName.OUTBOUND) @Transactional(isolation = Isolation.READ_UNCOMMITTED) public void flightTaskResourceGet(CommonTopicReceiver receiver, MessageHeaders headers) { Map<String, String> jobIdMap = objectMapper.convertValue(receiver.getData(), new TypeReference<Map<String, String>>() {}); Map<String, String> jobIdMap = objectMapper.convertValue(receiver.getData(), new TypeReference<Map<String, String>>() { }); String jobId = jobIdMap.get(MapKeyConst.FLIGHT_ID); CommonTopicResponse.CommonTopicResponseBuilder<RequestsReply> builder = CommonTopicResponse.<RequestsReply>builder() @@ -463,7 +637,7 @@ url = waylineFileService.getObjectUrl(waylineJob.getWorkspaceId(), waylineFile.get().getWaylineId()); builder.data(RequestsReply.success(WaylineTaskCreateDTO.builder() .file(WaylineTaskFileDTO.builder() .url(url.toString()) .url(MinioUrlUtils.getUrl(url)) .fingerprint(waylineFile.get().getSign()) .build()) .build())); @@ -495,7 +669,7 @@ ServiceReply reply = messageSender.publishServicesTopic( dockSn, MediaMethodEnum.UPLOAD_FLIGHT_TASK_MEDIA_PRIORITIZE.getMethod(), Map.of(MapKeyConst.FLIGHT_ID, jobId)); if (ResponseResult.CODE_SUCCESS != reply.getResult()) { throw new RuntimeException("Failed to set media job upload priority. Error Code: " + reply.getResult()); throw new RuntimeException("设置媒体作业上传优先级失败. 错误码: " + reply.getResult()); } } @@ -524,8 +698,8 @@ .fileId(dto.getFileId()) .dockSn(dto.getDockSn()) .workspaceId(dto.getWorkspaceId()) .taskType(dto.getTaskType()) .waylineType(dto.getWaylineType()) .taskType(Optional.ofNullable(dto.getTaskType()).map(WaylineTaskTypeEnum::getVal).orElse(null)) .waylineType(Optional.ofNullable(dto.getWaylineType()).map(WaylineTemplateTypeEnum::getVal).orElse(null)) .username(dto.getUsername()) .rthAltitude(dto.getRthAltitude()) .outOfControlAction(dto.getOutOfControlAction()) @@ -537,12 +711,12 @@ public void updateJobStatus(String workspaceId, String jobId, UpdateJobParam param) { Optional<WaylineJobDTO> waylineJobOpt = this.getJobByJobId(workspaceId, jobId); if (waylineJobOpt.isEmpty()) { throw new RuntimeException("The job does not exist."); throw new RuntimeException("任务不存在"); } WaylineJobDTO waylineJob = waylineJobOpt.get(); WaylineJobStatusEnum statusEnum = this.getWaylineState(waylineJob.getDockSn()); if (statusEnum.getEnd() || WaylineJobStatusEnum.PENDING == statusEnum) { throw new RuntimeException("The wayline job status does not match, and the operation cannot be performed."); throw new RuntimeException("航路线作业状态不匹配，无法执行操作."); } switch (param.getStatus()) { @@ -581,6 +755,236 @@ return WaylineJobStatusEnum.UNKNOWN; } @Override public WaylineJobEntity getLatestJob(String workspaceId, WaylineJobQueryParam waylineJobQueryParam) { WaylineJobEntity waylineJobEntity = mapper.getLatest(workspaceId, waylineJobQueryParam); return waylineJobEntity; } @Override public ResponseResult flyByArea(String sn, FlyAreaParam flyAreaParam, String deviceSn) throws Exception { //获取所有中心点 // List<PointPOJO> targetList = GeoUtils.caculatePointList(flyAreaParam); PointPOJO dockPoint = flyAreaParam.getDockPoint(); GeoJson geoJson = GeoUtils.readJsonFile(flyAreaParam.getJsonPath()); List<PointPOJO> targetList = GeoUtils.caculatePointList(geoJson, dockPoint, flyAreaParam.getRadius()); //一键起飞 TakeoffToPointParam takeoffToPointParam = buildTakeoffToPointParam(dockPoint); ResponseResult takeoffToPointRes = controlService.takeoffToPoint(sn, takeoffToPointParam); //设置飞向第一个点 while (takeoffToPointRes.getCode() == ResponseResult.CODE_SUCCESS) { //获取无人机状态 DeviceModeCodeEnum deviceMode = deviceService.getDeviceMode(deviceSn); //当无人机状态为手动飞行 if (deviceMode == DeviceModeCodeEnum.MANUAL) { //飞向目标点 FlyToPointParam flyToPointParam = new FlyToPointParam(); flyToPointParam.setMaxSpeed(14); List<PointDTO> pointDTOS = new ArrayList<>(); PointDTO pointDTO = new PointDTO(); pointDTO.setHeight(120.0); pointDTO.setLongitude(targetList.get(0).getLon()); pointDTO.setLatitude(targetList.get(0).getLat()); pointDTOS.add(pointDTO); flyToPointParam.setPoints(pointDTOS); ResponseResult flyToRes = controlService.flyToPoint(sn, flyToPointParam); if (flyToRes.getCode() == ResponseResult.CODE_SUCCESS) { //第一个点指令飞行成功后，把数组存到redis中 JSONObject jsonObject = new JSONObject(); jsonObject.put("targetList",targetList); jsonObject.put("payloadIndex",flyAreaParam.getPayloadIndex()); jsonObject.put("curIndex",0); RedisOpsUtils.set("tuban:"+sn,jsonObject); } break; } } return ResponseResult.success(targetList); } public TakeoffToPointParam buildTakeoffToPointParam(PointPOJO dockPoint) { TakeoffToPointParam takeoffToPointParam = new TakeoffToPointParam(); takeoffToPointParam.setTargetLatitude(dockPoint.getLat()); takeoffToPointParam.setTargetLongitude(dockPoint.getLon()); //设置飞行高度 takeoffToPointParam.setTargetHeight(120.0); //设置安全起飞高度 takeoffToPointParam.setSecurityTakeoffHeight(100.0); //设置返航高度 takeoffToPointParam.setRthAltitude(100.0); //设置失控操作 takeoffToPointParam.setRcLostAction(DroneRcLostActionEnum.RETURN_HOME); //设置起飞速度 takeoffToPointParam.setMaxSpeed(10.0); takeoffToPointParam.setExitWaylineWhenRcLost(WaylineRcLostActionEnum.EXECUTE_RC_LOST_ACTION); return takeoffToPointParam; } @Override public ResponseResult checkNextJob(WaylineJobDTO job) throws SQLException { WaylineJobEntity params = new WaylineJobEntity(); params.setJobId(job.getJobId()); WaylineJobEntity job1 = mapper.selectOne(Wrappers.query(params)); WaylineJobEntity currentJob = JSON.parseObject(JSON.toJSONString(job1), WaylineJobEntity.class); if (currentJob == null) { return ResponseResult.error("该任务不存在"); } //该任务没有多次时间 if (currentJob.getExecuteStartTimeArr() == null) { return ResponseResult.success(); } List<List<Long>> executeStartTimeArr = currentJob.getExecuteStartTimeArr(); int indexTime = -1; long currentTime = System.currentTimeMillis(); for (int i = 0; i < executeStartTimeArr.size(); i++) { Long startTime = executeStartTimeArr.get(i).get(0) * 1000; //如果当前时间小于开始时间 if (currentTime < startTime) { //当前索引就是下一次要执行的时间 indexTime = i; break; } } if (indexTime == -1) { return ResponseResult.success(); } List<Long> timeArr = currentJob.getExecuteStartTimeArr().get(indexTime); LocalDate date = LocalDate.ofInstant(Instant.ofEpochSecond(timeArr.get(0)), ZoneId.systemDefault()); LocalDateTime beginDate = LocalDateTime.of(date, LocalTime.ofInstant(Instant.ofEpochSecond(timeArr.get(0)), ZoneId.systemDefault())); LocalDateTime endDate = timeArr.size() > 1 && Objects.nonNull(timeArr.get(1)) ? LocalDateTime.of(date, LocalTime.ofInstant(Instant.ofEpochSecond(timeArr.get(1)), ZoneId.systemDefault())) : beginDate; long beginTime = beginDate .atZone(ZoneId.systemDefault()).toInstant().toEpochMilli(); long endTime = endDate .atZone(ZoneId.systemDefault()).toInstant().toEpochMilli(); //添加一条新数据 WaylineJobEntity waylineJobEntity = WaylineJobEntity.builder() .jobId(UUID.randomUUID().toString()) //大于1则拿前面的parentId,为1则拿第0个的id .parentId(indexTime > 1 ? currentJob.getParentId().toString() : currentJob.getId().toString()) .name(currentJob.getName()) .dockSn(currentJob.getDockSn()) .fileId(currentJob.getFileId()) .username(currentJob.getUsername()) .workspaceId(currentJob.getWorkspaceId()) .beginTime(beginTime) .endTime(endTime) .status(WaylineJobStatusEnum.PENDING.getVal()) .taskType(currentJob.getTaskType()) .waylineType(currentJob.getWaylineType()) .outOfControlAction(currentJob.getOutOfControlAction()) .batteryCapacity(currentJob.getBatteryCapacity()) .rthAltitude(currentJob.getRthAltitude()) .mediaCount(0) .repFreVal(currentJob.getRepFreVal()) .repFreType(currentJob.getRepFreType()) .repRuleType(currentJob.getRepRuleType()) .repRuleVal(currentJob.getRepRuleVal()) .executeTimeArr(currentJob.getExecuteTimeArr()) .executeStartTimeArr(currentJob.getExecuteStartTimeArr()) .build(); Optional<WaylineJobDTO> waylineJobOpt = insertWaylineJob(waylineJobEntity); if (waylineJobOpt.isEmpty()) { return ResponseResult.error("任务创建失败"); } WaylineJobDTO waylineJob = waylineJobOpt.get(); LocalDateTime beginTimeNext = LocalDateTime.of(date, LocalTime.ofInstant(Instant.ofEpochSecond(timeArr.get(0)), ZoneId.systemDefault())); LocalDateTime endTimeNext = timeArr.size() > 1 && Objects.nonNull(timeArr.get(1)) ? LocalDateTime.of(date, LocalTime.ofInstant(Instant.ofEpochSecond(timeArr.get(1)), ZoneId.systemDefault())) : beginTimeNext; //条件任务 if (currentJob.getTaskType() == 2) { //如果是条件任务类型，需要在任务参数中添加条件。 waylineJob.setConditions( WaylineTaskConditionDTO.builder() .executableConditions(null) .readyConditions(WaylineTaskReadyConditionDTO.builder() .batteryCapacity(currentJob.getBatteryCapacity()) .beginTime(beginTime) .endTime(endTime) .build()) .build()); } else { waylineJob.setBeginTime(beginDate); } ResponseResult response = this.publishOneFlightTask(waylineJob); if (ResponseResult.CODE_SUCCESS != response.getCode()) { return response; } return ResponseResult.success(); } @Override public WaylineJobCountDTO patrolStatistics(String workspaceId, String queryTime,String deviceSn) { WaylineJobCountDTO waylineJobCountDTO = new WaylineJobCountDTO(); List<DroneFlightLogEntity> list = flightLogMapper.patrolStatistics(workspaceId,queryTime,deviceSn); if (!CollectionUtils.isEmpty(list)) { waylineJobCountDTO.setTotalNumber(list.size()); long totalTime = list.stream().filter(task -> task.getEndTime()!= null && task.getStartTime()!= null).mapToLong(s -> s.getEndTime() - s.getStartTime()).sum() / 1000; long h = totalTime / 3600; long m = (totalTime % 3600) / 60; waylineJobCountDTO.setTotalDuration( String.format("%02d h %02d min", h, m)); List<String> deviceSns = list.stream().map(DroneFlightLogEntity::getDeviceSn).distinct().collect(Collectors.toList()); double sum = 0.0; for (String sn : deviceSns) { Double totalFlightDistance = new LambdaQueryChainWrapper<>(flightLogMapper) .eq(DroneFlightLogEntity::getDeviceSn, sn).orderByDesc(DroneFlightLogEntity::getEndTime).last("limit 1").one().getTotalFlightDistance(); if (totalFlightDistance!= null){ sum+=totalFlightDistance; } } waylineJobCountDTO.setTotalDistance((int) sum); } return waylineJobCountDTO; } @Override public void updateJobCollect(WaylineJobEntity waylineJob) { new LambdaUpdateChainWrapper<>(mapper) .eq(WaylineJobEntity::getJobId, waylineJob.getJobId()) .set(WaylineJobEntity::getCollectStatus, waylineJob.getCollectStatus()) .set(WaylineJobEntity::getUserId, waylineJob.getUserId()) .update(); } private void pauseJob(String workspaceId, String dockSn, String jobId, WaylineJobStatusEnum statusEnum) { if (WaylineJobStatusEnum.PAUSED == statusEnum && jobId.equals(waylineRedisService.getPausedWaylineJobId(dockSn))) { waylineRedisService.setPausedWaylineJob(dockSn, jobId); @@ -590,7 +994,7 @@ ServiceReply reply = messageSender.publishServicesTopic( dockSn, WaylineMethodEnum.FLIGHT_TASK_PAUSE.getMethod(), "", jobId); if (ResponseResult.CODE_SUCCESS != reply.getResult()) { throw new RuntimeException("Failed to pause wayline job. Error Code: " + reply.getResult()); throw new RuntimeException("未能恢复航路作业。错误码： " + reply.getResult()); } waylineRedisService.delRunningWaylineJob(dockSn); waylineRedisService.setPausedWaylineJob(dockSn, jobId); @@ -605,7 +1009,7 @@ ServiceReply reply = messageSender.publishServicesTopic( dockSn, WaylineMethodEnum.FLIGHT_TASK_RESUME.getMethod(), "", jobId); if (ResponseResult.CODE_SUCCESS != reply.getResult()) { throw new RuntimeException("Failed to resume wayline job. Error Code: " + reply.getResult()); throw new RuntimeException("未能恢复航路作业。错误码：: " + reply.getResult()); } runningDataOpt.ifPresent(runningData -> waylineRedisService.setRunningWaylineJob(dockSn, runningData)); @@ -636,7 +1040,7 @@ .workspaceId(entity.getWorkspaceId()) .status(WaylineJobStatusEnum.IN_PROGRESS.getVal() == entity.getStatus() && entity.getJobId().equals(waylineRedisService.getPausedWaylineJobId(entity.getDockSn())) ? WaylineJobStatusEnum.PAUSED.getVal() : entity.getStatus()) WaylineJobStatusEnum.PAUSED.getVal() : entity.getStatus()) .code(entity.getErrorCode()) .beginTime(LocalDateTime.ofInstant(Instant.ofEpochMilli(entity.getBeginTime()), ZoneId.systemDefault())) .endTime(Objects.nonNull(entity.getEndTime()) ? @@ -645,11 +1049,12 @@ LocalDateTime.ofInstant(Instant.ofEpochMilli(entity.getExecuteTime()), ZoneId.systemDefault()) : null) .completedTime(WaylineJobStatusEnum.find(entity.getStatus()).getEnd() ? LocalDateTime.ofInstant(Instant.ofEpochMilli(entity.getUpdateTime()), ZoneId.systemDefault()) : null) .taskType(entity.getTaskType()) .waylineType(entity.getWaylineType()) .taskType(WaylineTaskTypeEnum.find(entity.getTaskType())) .waylineType(WaylineTemplateTypeEnum.find(entity.getWaylineType())) .rthAltitude(entity.getRthAltitude()) .outOfControlAction(entity.getOutOfControlAction()) .mediaCount(entity.getMediaCount()); .mediaCount(entity.getMediaCount()) .hasChildren(entity.getHasChildren()); if (Objects.nonNull(entity.getEndTime())) { builder.endTime(LocalDateTime.ofInstant(Instant.ofEpochMilli(entity.getEndTime()), ZoneId.systemDefault())); @@ -672,7 +1077,7 @@ Object mediaFileCount = RedisOpsUtils.hashGet(countKey, entity.getJobId()); if (Objects.nonNull(mediaFileCount)) { builder.uploadedCount(((MediaFileCountDTO) mediaFileCount).getUploadedCount()) .uploading(RedisOpsUtils.checkExist(key) && entity.getJobId().equals(((MediaFileCountDTO)RedisOpsUtils.get(key)).getJobId())); .uploading(RedisOpsUtils.checkExist(key) && entity.getJobId().equals(((MediaFileCountDTO) RedisOpsUtils.get(key)).getJobId())); return builder.build(); }
