forked from uav/ard-uav-server
ard-sample/src/main/java/com/ard/sample/wayline/service/impl/WaylineJobServiceImpl.java
@@ -1,6 +1,6 @@ package com.ard.sample.wayline.service.impl; import com.ard.oss.service.impl.OssServiceContext; import com.ard.ossdock.service.impl.OssDockServiceContext; import com.ard.route.domain.route.ArdRouteInfo; import com.ard.route.service.IArdRouteInfoService; import com.baomidou.mybatisplus.core.conditions.Wrapper; @@ -77,7 +77,7 @@ private IArdRouteInfoService _ardRouteInfoService; @Resource private OssServiceContext _ossService; private OssDockServiceContext _ossService; @Autowired private IDeviceService deviceService; @@ -133,6 +133,7 @@ .waylineType(param.getWaylineType()) .outOfControlAction(param.getOutOfControlAction()) .rthAltitude(param.getRthAltitude()) .rthMode(param.getRth_mode()) .mediaCount(0) .parentId(param.getParentId()) .loopSum(param.getLoopSum()) @@ -177,6 +178,7 @@ if (WaylineTaskTypeEnum.IMMEDIATE.getVal() == param.getTaskType()) { param.setExecuteTime(System.currentTimeMillis()); } param.setRth_mode(1); Optional<WaylineJobDTO> waylineJobOpt = this.createWaylineJob(param, loginUser.getUsername(), param.getExecuteTime(), param.getExecuteTime()); // 循环任务 @@ -210,14 +212,19 @@ // get file url // URL url = waylineFileService.getObjectUrl(waylineFile.get().getWaylineId()); URL url = null; String minioUrl = subStringMinioUrlName(ardRouteOpt.getPath()); String minioUrl = subStringMinioUrlName(ardRouteOpt.getPath1()); String md5 = ""; try (InputStream inputStream = _ossService.getObject(kmzPath, minioUrl)) { md5 = DigestUtils.md5Hex(inputStream); url = new URL(ardRouteOpt.getPath()); url = new URL(ardRouteOpt.getPath1()); log.info("机场任务航线文件url地址：{}", url); log.info("机场任务航线文件md5值：{}", md5); } catch (Exception e) { log.error("机场任务航线文件地址错误！{}", e.getMessage()); return ResponseResult.error("机场任务航线文件地址错误！" + e.getMessage()); } log.info("机场任务-rth_mode的值：{}", waylineJob.getRthMode()); WaylineTaskCreateDTO flightTask = WaylineTaskCreateDTO.builder() .flightId(waylineJob.getJobId()) @@ -226,16 +233,20 @@ .waylineType(waylineJob.getWaylineType()) .rthAltitude(waylineJob.getRthAltitude()) .outOfControlAction(waylineJob.getOutOfControlAction()) .rthMode(waylineJob.getRthMode()) .file(WaylineTaskFileDTO.builder() .url(url.toString()) // .url(url.toString().replace("https://192.168.10.103:9001", "http://192.168.10.103:9011")) // .fingerprint(waylineFile.get().getSign()) // .replace("https://", "http://") // .replace("9001", "9011") .url(url.toString()) .fingerprint(md5) .build()) .simulateMission(WaylineSimulateMissionDTO.builder() .isEnable(1) .latitude(46.558831038) .longitude(125.148072319) .build()) // .simulateMission(WaylineSimulateMissionDTO.builder() // .isEnable(1) // .latitude(46.55819013988211) // .longitude(125.14652370145113) // .build()) .build(); String topic = TopicConst.THING_MODEL_PRE + TopicConst.PRODUCT + @@ -248,8 +259,19 @@ .method(WaylineMethodEnum.FLIGHT_TASK_PREPARE.getMethod()) .build(); // 下发参数打印 try{ String json = objectMapper.writeValueAsString(response); log.info("机场下发航线任务的参数response：{}", json); } catch (Exception e){ log.error("机场下发航线任务的参数response异常：{}", e.getMessage()); } ServiceReply serviceReply = messageSender.publishWithReply(topic, response); log.info("下发任务结果： {}", serviceReply.getResult()); if (ResponseResult.CODE_SUCCESS != serviceReply.getResult()) { log.info("Prepare task ====> Error code: {}", serviceReply.getResult()); this.updateJob(WaylineJobDTO.builder() @@ -263,20 +285,20 @@ // 飞行命令先注释掉 // Issue an immediate task execution command. // if (WaylineTaskTypeEnum.IMMEDIATE.getVal() == waylineJob.getTaskType()) { // if (!executeFlightTask(waylineJob.getJobId())) { // return ResponseResult.error("Failed to execute job."); // } // } // // if (WaylineTaskTypeEnum.TIMED.getVal() == waylineJob.getTaskType()) { // boolean isAdd = RedisOpsUtils.zAdd(RedisConst.WAYLINE_JOB_TIMED_EXECUTE, // 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.getVal() == waylineJob.getTaskType()) { if (!executeFlightTask(waylineJob.getJobId())) { return ResponseResult.error("Failed to execute job."); } } if (WaylineTaskTypeEnum.TIMED.getVal() == waylineJob.getTaskType()) { boolean isAdd = RedisOpsUtils.zAdd(RedisConst.WAYLINE_JOB_TIMED_EXECUTE, waylineJob.getDockSn() + RedisConst.DELIMITER + waylineJob.getJobId(), waylineJob.getBeginTime().atZone(ZoneId.systemDefault()).toInstant().toEpochMilli()); if (!isAdd) { return ResponseResult.error("Failed to create scheduled job."); } } return ResponseResult.success(); } @@ -303,7 +325,10 @@ try (InputStream inputStream = _ossService.getObject(kmzPath, minioUrl)) { md5 = DigestUtils.md5Hex(inputStream); url = new URL(ardRouteOpt.getPath()); log.info("机场任务航线文件url地址：{}", url); log.info("机场任务航线文件md5值：{}", md5); } catch (Exception e) { log.error("机场任务航线文件地址错误！{}", e.getMessage()); return ResponseResult.error("机场任务航线文件地址错误！" + e.getMessage()); } @@ -319,12 +344,21 @@ // .fingerprint(waylineFile.get().getSign()) .fingerprint(md5) .build()) .simulateMission(WaylineSimulateMissionDTO.builder() .isEnable(1) .latitude(46.558831038) .longitude(125.148072319) .build()) // .simulateMission(WaylineSimulateMissionDTO.builder() // .isEnable(1) // .latitude(46.558831038) // .longitude(125.148072319) // .build()) .build(); // 下发参数打印 try{ String json = objectMapper.writeValueAsString(flightTask); log.info("机场下发航线任务的参数flightTask：{}", json); } catch (Exception e){ log.error("机场下发航线任务的参数flightTask异常：{}", e.getMessage()); } String topic = TopicConst.THING_MODEL_PRE + TopicConst.PRODUCT + waylineJob.getDockSn() + TopicConst.SERVICES_SUF; @@ -337,7 +371,11 @@ .build(); ServiceReply serviceReply = messageSender.publishWithReply(topic, response); log.info("下发任务结果： {}", serviceReply.getResult()); log.info("机场任务航线文件url地址：{}", url); log.info("机场任务航线文件md5值：{}", md5); if (ResponseResult.CODE_SUCCESS != serviceReply.getResult()) { log.info("Prepare task ====> Error code: {}", serviceReply.getResult()); this.updateJob(WaylineJobDTO.builder() @@ -351,20 +389,20 @@ // 飞行命令先注释掉 // Issue an immediate task execution command. // if (WaylineTaskTypeEnum.IMMEDIATE.getVal() == waylineJob.getTaskType()) { // if (!executeFlightTask(waylineJob.getJobId())) { // return ResponseResult.error("Failed to execute job."); // } // } // // if (WaylineTaskTypeEnum.TIMED.getVal() == waylineJob.getTaskType()) { // boolean isAdd = RedisOpsUtils.zAdd(RedisConst.WAYLINE_JOB_TIMED_EXECUTE, // 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.getVal() == waylineJob.getTaskType()) { if (!executeFlightTask(waylineJob.getJobId())) { return ResponseResult.error("Failed to execute job."); } } if (WaylineTaskTypeEnum.TIMED.getVal() == waylineJob.getTaskType()) { boolean isAdd = RedisOpsUtils.zAdd(RedisConst.WAYLINE_JOB_TIMED_EXECUTE, waylineJob.getDockSn() + RedisConst.DELIMITER + waylineJob.getJobId(), waylineJob.getBeginTime().atZone(ZoneId.systemDefault()).toInstant().toEpochMilli()); if (!isAdd) { return ResponseResult.error("Failed to create scheduled job."); } } return ResponseResult.success(); } @@ -392,7 +430,7 @@ WaylineJobDTO job = waylineJob.get(); WaylineTaskCreateDTO flightTask = WaylineTaskCreateDTO.builder().flightId(jobId).build(); // flightTask中，只有任务id是有值的，其他字段不返回 String topic = TopicConst.THING_MODEL_PRE + TopicConst.PRODUCT + job.getDockSn() + TopicConst.SERVICES_SUF; CommonTopicResponse<Object> response = CommonTopicResponse.builder() @@ -403,7 +441,19 @@ .method(WaylineMethodEnum.FLIGHT_TASK_EXECUTE.getMethod()) .build(); // 下发参数打印 try{ String json = objectMapper.writeValueAsString(response); log.info("机场下发-执行飞行任务-的参数response：{}", json); } catch (Exception e){ log.error("机场下发-执行飞行任务-的参数response异常：{}", e.getMessage()); } ServiceReply serviceReply = messageSender.publishWithReply(topic, response); log.info("下发-执行飞行任务-结果： {}", serviceReply.getResult()); if (ResponseResult.CODE_SUCCESS != serviceReply.getResult()) { log.info("Execute job ====> Error code: {}", serviceReply.getResult()); this.updateJob(WaylineJobDTO.builder() @@ -436,14 +486,14 @@ public void cancelFlightTask(Collection<String> jobIds) { List<WaylineJobDTO> waylineJobs = getJobsByConditions(jobIds, WaylineJobStatusEnum.PENDING); 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(); if (isErr) { throw new IllegalArgumentException("任务开始后，无法取消。" + Arrays.toString(jobIds.toArray())); // throw new IllegalArgumentException("These tasks have an incorrect status and cannot be canceled. " // + Arrays.toString(jobIds.toArray())); } // 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(); // if (isErr) { // throw new IllegalArgumentException("任务开始后，无法取消。" + Arrays.toString(jobIds.toArray())); //// throw new IllegalArgumentException("These tasks have an incorrect status and cannot be canceled. " //// + Arrays.toString(jobIds.toArray())); // } // Group job id by dock sn. Map<String, List<String>> dockJobs = waylineJobs.stream() @@ -534,6 +584,18 @@ return Optional.ofNullable(entity2Dto(jobEntity)); } public List<WaylineJobDTO> getJobByJobParentId(String parentId) { List<WaylineJobEntity> jobEntities = mapper.selectList( new LambdaQueryWrapper<WaylineJobEntity>() .eq(WaylineJobEntity::getParentId, parentId)); if (CollectionUtils.isEmpty(jobEntities)) { return Collections.emptyList(); // 返回不可变空列表，更安全 } return jobEntities.stream() .map(this::entity2Dto) // 调用原有的单个转换方法 .collect(Collectors.toList()); } /** * 更新作业信息 * @@ -581,6 +643,7 @@ Page<WaylineJobEntity> pageData = mapper.selectPage( new Page<WaylineJobEntity>(page, pageSize), new LambdaQueryWrapper<WaylineJobEntity>() .isNull(WaylineJobEntity::getParentId) // 实际开始时间筛选，允许为空，为空则不进行筛选 .ge(startTime != null, WaylineJobEntity::getBeginTime, startTime) .le(endTime != null, WaylineJobEntity::getBeginTime, endTime) @@ -593,10 +656,14 @@ .ne(isComplete != null && !isComplete, WaylineJobEntity::getStatus, 3) // 任务状态筛选，允许为空，为空则不进行筛选 .eq(status != null, WaylineJobEntity::getStatus, status) .orderByDesc(WaylineJobEntity::getId)); .orderByDesc(WaylineJobEntity::getBeginTime)); List<WaylineJobDTO> records = pageData.getRecords() .stream() .map(this::entity2Dto) .peek(job -> { List<WaylineJobDTO> parentOpt = this.getJobByJobParentId(job.getJobId()); job.setParentJobs(parentOpt); }) .collect(Collectors.toList()); return new PaginationData<WaylineJobDTO>(records, new Pagination(pageData)); @@ -631,6 +698,8 @@ Map<String, String> jobIdMap = objectMapper.convertValue(receiver.getData(), new TypeReference<Map<String, String>>() { }); String jobId = jobIdMap.get(MapKeyConst.FLIGHT_ID); log.info("机场通过flightTaskResourceGet获取航线文件地址，jobId: " + jobId); CommonTopicResponse.CommonTopicResponseBuilder<RequestsReply> builder = CommonTopicResponse.<RequestsReply>builder() .tid(receiver.getTid()) @@ -668,21 +737,28 @@ // get file url URL url = null; String minioUrl = subStringMinioUrlName(ardRouteOpt.getPath()); String minioUrl = subStringMinioUrlName(ardRouteOpt.getPath1()); String md5 = ""; try (InputStream inputStream = _ossService.getObject(kmzPath, minioUrl)) { md5 = DigestUtils.md5Hex(inputStream); url = new URL(ardRouteOpt.getPath()); url = new URL(ardRouteOpt.getPath1()); log.info("机场通过flightTaskResourceGet获取航线文件地址，jobId: " + jobId + " 返回信息为url：" + url.toString() + " md5：" + md5); // } // try { // url = waylineFileService.getObjectUrl(waylineFile.get().getWaylineId()); // url = new URL(ardRouteOpt.getUpPath()); // Map<String, Object> jsonResourceGet = new HashMap<>(); // jsonResourceGet.put("url", url.toString().replace("https//192.168.10.103:9001", "http//192.168.10.103:9002")); // jsonResourceGet.put("fingerprint", md5); // .replace("https://", "http://") // .replace("9001", "9011") builder.data(RequestsReply.success(WaylineTaskCreateDTO.builder() .file(WaylineTaskFileDTO.builder() .url(url.toString()) // .fingerprint(waylineFile.get().getSign()) .fingerprint(md5) .build()) .url(url.toString()) .fingerprint(md5) .build()) .build())); } catch (NullPointerException | IOException e) { @@ -690,6 +766,15 @@ builder.data(RequestsReply.error(CommonErrorEnum.ILLEGAL_ARGUMENT)); messageSender.publish(topic, builder.build()); return; } // 下发参数打印 try{ String json = objectMapper.writeValueAsString(builder.build()); log.info("机场通过flighttask_resource_get获取航线文件的参数flightTask：{}", json); } catch (Exception e){ log.error("机场通过flighttask_resource_get获取航线文件的参数flightTask异常：{}", e.getMessage()); } messageSender.publish(topic, builder.build()); @@ -819,6 +904,7 @@ .taskType(entity.getTaskType()) .waylineType(entity.getWaylineType()) .rthAltitude(entity.getRthAltitude()) .rthMode(entity.getRthMode()) .outOfControlAction(entity.getOutOfControlAction()) .mediaCount(entity.getMediaCount()) .loopSum(entity.getLoopSum());
