获取计划最近一条

guoshilong

2023-10-23 dd73bbc7132f7b126b8fc2faabf2827b49622d21

 src/main/java/com/dji/sample/wayline/service/impl/WaylineJobServiceImpl.java

@@ -143,6 +143,7 @@

    /**
     * 对于即时任务，以服务器时间为准。
     *
     * @param param
     */
    private void fillImmediateTime(CreateJobParam param) {
@@ -171,14 +172,15 @@
                        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("The task has expired.");
                    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.
                //如果是条件任务类型，需要在任务参数中添加条件。
                addPreparedJob(waylineJob, param, beginTime, endTime);

                ResponseResult response = this.publishOneFlightTask(waylineJob);
@@ -190,12 +192,96 @@
        return ResponseResult.success();
    }

    @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("任务创建失败");
        }

        WaylineJobDTO waylineJob = waylineJobOpt.get();

        //循环执行时间
        for (List<Long> timeArr : param.getExecuteStartTimeArr()) {
            LocalDate date = LocalDate.ofInstant(Instant.ofEpochSecond(timeArr.get(0)), ZoneId.systemDefault());
            long beginTime = LocalDateTime.of(date, LocalTime.ofInstant(Instant.ofEpochSecond(timeArr.get(0)), ZoneId.systemDefault()))
                    .atZone(ZoneId.systemDefault()).toInstant().toEpochMilli();

            long endTime = timeArr.size() > 1 && Objects.nonNull(timeArr.get(1)) ?
                    LocalDateTime.of(date, LocalTime.ofInstant(Instant.ofEpochSecond(timeArr.get(1)), ZoneId.systemDefault()))
                            .atZone(ZoneId.systemDefault()).toInstant().toEpochMilli() : beginTime;

            if (WaylineTaskTypeEnum.IMMEDIATE != param.getTaskType() && endTime < System.currentTimeMillis()) {
                //中断当前循环执行下一个
                continue;
            }


            //条件任务
            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)
                                        .endTime(endTime)
                                        .build())
                                .build());
            }

            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)
                            .executableConditions(Objects.nonNull(param.getMinStorageCapacity()) ? WaylineTaskExecutableConditionDTO.builder().storageCapacity(param.getMinStorageCapacity()).build() : null)
                            .readyConditions(WaylineTaskReadyConditionDTO.builder()
                                    .batteryCapacity(param.getMinBatteryCapacity())
                                    .beginTime(beginTime)
@@ -208,7 +294,7 @@
        // value: {workspace_id}:{dock_sn}:{job_id}
        boolean isAdd = waylineRedisService.addPreparedWaylineJob(waylineJob);
        if (!isAdd) {
            throw new RuntimeException("Failed to create prepare job.");
            throw new RuntimeException("创建任务失败。");
        }
    }

@@ -216,14 +302,15 @@

        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 == waylineJob.getTaskType()) {
            boolean isExecuted = executeFlightTask(waylineJob.getWorkspaceId(), waylineJob.getJobId());
            if (!isExecuted) {
                return ResponseResult.error("Failed to execute job.");
                return ResponseResult.error("执行任务失败");
            }
        }

@@ -244,6 +331,7 @@
        }

        // get file url
        //获取航线文件地址
        URL url = waylineFileService.getObjectUrl(waylineJob.getWorkspaceId(), waylineFile.get().getWaylineId());

        WaylineTaskCreateDTO flightTask = WaylineTaskCreateDTO.builder()
@@ -259,6 +347,7 @@
                        .build())
                .build();

        //当任务类型为条件时
        if (WaylineTaskTypeEnum.CONDITION == waylineJob.getTaskType()) {
            if (Objects.isNull(waylineJob.getConditions())) {
                throw new IllegalArgumentException();
@@ -267,10 +356,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())
@@ -288,12 +379,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();
@@ -332,9 +423,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.
@@ -348,14 +439,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) {
@@ -371,10 +462,10 @@

    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())
                        .in(!CollectionUtils.isEmpty(jobIds), WaylineJobEntity::getJobId, jobIds))
                        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());
@@ -409,8 +500,7 @@
//
//                        .orderByDesc(WaylineJobEntity::getId));

        Page<WaylineJobEntity> pageData = mapper.getPage(new Page<WaylineJobEntity>(page, pageSize),waylineJobQueryParam,workspaceId);

        Page<WaylineJobEntity> pageData = mapper.getPage(new Page<WaylineJobEntity>(page, pageSize), waylineJobQueryParam, workspaceId);


        List<WaylineJobDTO> records = pageData.getRecords()
@@ -426,7 +516,8 @@
    @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()
@@ -496,7 +587,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());
        }
    }

@@ -538,12 +629,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()) {
@@ -582,6 +673,22 @@
        return WaylineJobStatusEnum.UNKNOWN;
    }

    @Override
    public WaylineJobEntity getLatestJob(String workspaceId, WaylineJobQueryParam waylineJobQueryParam) {
//        List<WaylineJobEntity> waylineJobEntities = mapper.selectList(new LambdaQueryWrapper<>(WaylineJobEntity.class)
//                .eq(WaylineJobEntity::getWorkspaceId, workspaceId)
//                .eq(WaylineJobEntity::getDockSn, waylineJobQueryParam.getDockSn())
//                //获取状态为待执行
//                        .eq(WaylineJobEntity::getStatus,1)
//                .orderByDesc(WaylineJobEntity::getBeginTime)
//        );

        WaylineJobEntity waylineJobEntity = mapper.getLatest(workspaceId,waylineJobQueryParam);


       return waylineJobEntity;
    }

    private void pauseJob(String workspaceId, String dockSn, String jobId, WaylineJobStatusEnum statusEnum) {
        if (WaylineJobStatusEnum.PAUSED == statusEnum && jobId.equals(waylineRedisService.getPausedWaylineJobId(dockSn))) {
            waylineRedisService.setPausedWaylineJob(dockSn, jobId);
@@ -591,7 +698,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);
@@ -606,7 +713,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));
@@ -637,7 +744,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()) ?
@@ -673,7 +780,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();
        }