diff --git a/jeecg-boot-base-core/src/main/java/org/jeecg/common/constant/CommonConstant.java b/jeecg-boot-base-core/src/main/java/org/jeecg/common/constant/CommonConstant.java index 065eb5b..10a3135 100644 --- a/jeecg-boot-base-core/src/main/java/org/jeecg/common/constant/CommonConstant.java +++ b/jeecg-boot-base-core/src/main/java/org/jeecg/common/constant/CommonConstant.java @@ -339,6 +339,9 @@ public interface CommonConstant { */ String BUILD_TASK_STATE_PRE = "build_task_"; - + /*** + * weather_data:数据类型（ncep、fnl）:变量类型(温、湿、压、风) + */ + String WEATHER_DATA_CACHE = "weather_data:%s:%s"; } diff --git a/jeecg-boot-base-core/src/main/java/org/jeecg/common/constant/enums/TransportTaskGenGifTypeEnum.java b/jeecg-boot-base-core/src/main/java/org/jeecg/common/constant/enums/TransportTaskGenGifTypeEnum.java new file mode 100644 index 0000000..99c86c0 --- /dev/null +++ b/jeecg-boot-base-core/src/main/java/org/jeecg/common/constant/enums/TransportTaskGenGifTypeEnum.java @@ -0,0 +1,26 @@ +package org.jeecg.common.constant.enums; + +/** + * 输运任务生成GIF动图类型说明枚举 + */ +public enum TransportTaskGenGifTypeEnum { + + /** + * 全球模式 + */ + GLOBAL("global"), + /** + * 区域模式 + */ + REGION("region"); + + private String value; + + TransportTaskGenGifTypeEnum(String key) { + this.value = key; + } + + public String getValue(){ + return this.value; + } +} diff --git a/jeecg-boot-base-core/src/main/java/org/jeecg/common/constant/enums/WeatherTypeEnum.java b/jeecg-boot-base-core/src/main/java/org/jeecg/common/constant/enums/WeatherTypeEnum.java index 9b3d949..953a760 100644 --- a/jeecg-boot-base-core/src/main/java/org/jeecg/common/constant/enums/WeatherTypeEnum.java +++ b/jeecg-boot-base-core/src/main/java/org/jeecg/common/constant/enums/WeatherTypeEnum.java @@ -28,4 +28,12 @@ public enum WeatherTypeEnum { return this.value; } + public static WeatherTypeEnum getInfoByKey(int key) { + for (WeatherTypeEnum info : WeatherTypeEnum.values()) { + if (info.getKey() == key) { + return info; + } + } + return null; + } } diff --git a/jeecg-boot-base-core/src/main/java/org/jeecg/common/constant/enums/WeatherVariableNameEnum.java b/jeecg-boot-base-core/src/main/java/org/jeecg/common/constant/enums/WeatherVariableNameEnum.java index 79bb873..bd70ce7 100644 --- a/jeecg-boot-base-core/src/main/java/org/jeecg/common/constant/enums/WeatherVariableNameEnum.java +++ b/jeecg-boot-base-core/src/main/java/org/jeecg/common/constant/enums/WeatherVariableNameEnum.java @@ -6,36 +6,41 @@ package org.jeecg.common.constant.enums; public enum WeatherVariableNameEnum { - PANGU_T(WeatherDataSourceEnum.PANGU.getKey(), 0, "Temperature_height_above_ground"), - PANGU_P(WeatherDataSourceEnum.PANGU.getKey(), 1, "Pressure_height_above_ground"), - PANGU_H(WeatherDataSourceEnum.PANGU.getKey(), 2, "Specific_humidity_height_above_ground"), - PANGU_U(WeatherDataSourceEnum.PANGU.getKey(), 3, "u-component_of_wind_height_above_ground"), - PANGU_V(WeatherDataSourceEnum.PANGU.getKey(), 4, "v-component_of_wind_height_above_ground"), + PANGU_T(WeatherDataSourceEnum.PANGU.getKey(), 0, "Temperature_height_above_ground"),//2米温度 + PANGU_P(WeatherDataSourceEnum.PANGU.getKey(), 1, "Pressure_msl"),//压力 + PANGU_H(WeatherDataSourceEnum.PANGU.getKey(), 2, "Specific_humidity_isobaric"),//5000米湿度 + PANGU_U(WeatherDataSourceEnum.PANGU.getKey(), 3, "u-component_of_wind_height_above_ground"),//10米高度层上的东西向风速分量（正值为西风，负值为东风）。 + PANGU_V(WeatherDataSourceEnum.PANGU.getKey(), 4, "v-component_of_wind_height_above_ground"),//10米高度层上的南北向风速分量（正值为南风，负值为北风）。 + GRAPHCAST_T(WeatherDataSourceEnum.GRAPHCAST.getKey(), 0, "Temperature_height_above_ground"),//2米温度 + GRAPHCAST_P(WeatherDataSourceEnum.GRAPHCAST.getKey(), 1, "Pressure_msl"),//压力 + GRAPHCAST_H(WeatherDataSourceEnum.GRAPHCAST.getKey(), 2, "Specific_humidity_isobaric"),//5000米湿度 + GRAPHCAST_U(WeatherDataSourceEnum.GRAPHCAST.getKey(), 3, "u-component_of_wind_height_above_ground"),//10米高度层上的东西向风速分量（正值为西风，负值为东风）。 + GRAPHCAST_V(WeatherDataSourceEnum.GRAPHCAST.getKey(), 4, "v-component_of_wind_height_above_ground"),//10米高度层上的南北向风速分量（正值为南风，负值为北风）。 CRA40_T(WeatherDataSourceEnum.CRA40.getKey(), 0, "Temperature_isobaric"), CRA40_P(WeatherDataSourceEnum.CRA40.getKey(), 1, "Vertical_velocity_pressure_isobaric"), CRA40_H(WeatherDataSourceEnum.CRA40.getKey(), 2, "Relative_humidity_isobaric"), CRA40_U(WeatherDataSourceEnum.CRA40.getKey(), 3, "u-component_of_wind_isobaric"), CRA40_V(WeatherDataSourceEnum.CRA40.getKey(), 4, "v-component_of_wind_isobaric"), - NCEP_T(WeatherDataSourceEnum.NCEP.getKey(), 0, "Temperature_height_above_ground"), - NCEP_P(WeatherDataSourceEnum.NCEP.getKey(), 1, "Pressure_msl"), - NCEP_H(WeatherDataSourceEnum.NCEP.getKey(), 2, "Relative_humidity_height_above_ground"), - NCEP_U(WeatherDataSourceEnum.NCEP.getKey(), 3, "u-component_of_wind_height_above_ground"), - NCEP_V(WeatherDataSourceEnum.NCEP.getKey(), 4, "v-component_of_wind_height_above_ground"), - FNL_T(WeatherDataSourceEnum.FNL.getKey(), 0, "Temperature_height_above_ground"), - FNL_P(WeatherDataSourceEnum.FNL.getKey(), 1, "Pressure_height_above_ground"), - FNL_H(WeatherDataSourceEnum.FNL.getKey(), 2, "Relative_humidity_height_above_ground"), - FNL_U(WeatherDataSourceEnum.FNL.getKey(), 3, "u-component_of_wind_height_above_ground"), - FNL_V(WeatherDataSourceEnum.FNL.getKey(), 4, "v-component_of_wind_height_above_ground"), + NCEP_T(WeatherDataSourceEnum.NCEP.getKey(), 0, "Temperature_height_above_ground"),//2米温度 + NCEP_P(WeatherDataSourceEnum.NCEP.getKey(), 1, "Pressure_msl"),//海平面气压 + NCEP_H(WeatherDataSourceEnum.NCEP.getKey(), 2, "Relative_humidity_height_above_ground"),//2米湿度 + NCEP_U(WeatherDataSourceEnum.NCEP.getKey(), 3, "u-component_of_wind_height_above_ground"),//10米高度层上的东西向风速分量（正值为西风，负值为东风）。 + NCEP_V(WeatherDataSourceEnum.NCEP.getKey(), 4, "v-component_of_wind_height_above_ground"),//10米高度层上的南北向风速分量（正值为南风，负值为北风）。 + FNL_T(WeatherDataSourceEnum.FNL.getKey(), 0, "Temperature_height_above_ground"),//2米温度 + FNL_P(WeatherDataSourceEnum.FNL.getKey(), 1, "Pressure_height_above_ground"),//80米高度气压 + FNL_H(WeatherDataSourceEnum.FNL.getKey(), 2, "Relative_humidity_height_above_ground"),//2米湿度 + FNL_U(WeatherDataSourceEnum.FNL.getKey(), 3, "u-component_of_wind_height_above_ground"),//10米高度层上的东西向风速分量（正值为西风，负值为东风）。 + FNL_V(WeatherDataSourceEnum.FNL.getKey(), 4, "v-component_of_wind_height_above_ground"),//10米高度层上的南北向风速分量（正值为南风，负值为北风）。 T1H_T(WeatherDataSourceEnum.T1H.getKey(), 0, "Temperature_height_above_ground"), T1H_P(WeatherDataSourceEnum.T1H.getKey(), 1, "Pressure_height_above_ground"), T1H_H(WeatherDataSourceEnum.T1H.getKey(), 2, "Relative_humidity_height_above_ground"), T1H_U(WeatherDataSourceEnum.T1H.getKey(), 3, "u-component_of_wind_height_above_ground"), T1H_V(WeatherDataSourceEnum.T1H.getKey(), 4, "v-component_of_wind_height_above_ground"), - GFS_T(WeatherDataSourceEnum.GFS.getKey(), 0, "Temperature_height_above_ground"), - GFS_P(WeatherDataSourceEnum.GFS.getKey(), 1, "Pressure_height_above_ground"), - GFS_H(WeatherDataSourceEnum.GFS.getKey(), 2, "Relative_humidity_height_above_ground"), - GFS_U(WeatherDataSourceEnum.GFS.getKey(), 3, "u-component_of_wind_height_above_ground"), - GFS_V(WeatherDataSourceEnum.GFS.getKey(), 4, "v-component_of_wind_height_above_ground"); + GFS_T(WeatherDataSourceEnum.GFS.getKey(), 0, "Temperature_height_above_ground"),//2米温度 + GFS_P(WeatherDataSourceEnum.GFS.getKey(), 1, "Pressure_height_above_ground"),//80米高度气压 + GFS_H(WeatherDataSourceEnum.GFS.getKey(), 2, "Relative_humidity_height_above_ground"),//2米湿度 + GFS_U(WeatherDataSourceEnum.GFS.getKey(), 3, "u-component_of_wind_height_above_ground"),//10米高度层上的东西向风速分量（正值为西风，负值为东风）。 + GFS_V(WeatherDataSourceEnum.GFS.getKey(), 4, "v-component_of_wind_height_above_ground");//10米高度层上的南北向风速分量（正值为南风，负值为北风）。 private Integer type; diff --git a/jeecg-boot-base-core/src/main/java/org/jeecg/common/properties/ServerProperties.java b/jeecg-boot-base-core/src/main/java/org/jeecg/common/properties/ServerProperties.java index 3ed2a8d..a6c8394 100644 --- a/jeecg-boot-base-core/src/main/java/org/jeecg/common/properties/ServerProperties.java +++ b/jeecg-boot-base-core/src/main/java/org/jeecg/common/properties/ServerProperties.java @@ -25,18 +25,27 @@ public class ServerProperties { * 密码 */ private String password; + + /** + * 服务器开关 + */ + private boolean consumerEnable; + /** * 172.21.170.11 ip */ private String ip11; + /** * 172.21.170.12 ip */ private String ip12; + /** * 172.21.170.13 ip */ private String ip13; + /** * 172.21.170.14 ip */ diff --git a/jeecg-boot-base-core/src/main/java/org/jeecg/common/properties/TransportSimulationProperties.java b/jeecg-boot-base-core/src/main/java/org/jeecg/common/properties/TransportSimulationProperties.java index e37260d..2483758 100644 --- a/jeecg-boot-base-core/src/main/java/org/jeecg/common/properties/TransportSimulationProperties.java +++ b/jeecg-boot-base-core/src/main/java/org/jeecg/common/properties/TransportSimulationProperties.java @@ -174,4 +174,9 @@ public class TransportSimulationProperties { * 粒子数量 */ private Integer particleCount; + + /** + * 气象数据检查最大失败天数（等待天数），超过就设置检查未通过 + */ + private Integer metCheckMaxFailureDays; } diff --git a/jeecg-boot-base-core/src/main/java/org/jeecg/common/util/DownloadUtils.java b/jeecg-boot-base-core/src/main/java/org/jeecg/common/util/DownloadUtils.java index ae5b992..bfbc14c 100644 --- a/jeecg-boot-base-core/src/main/java/org/jeecg/common/util/DownloadUtils.java +++ b/jeecg-boot-base-core/src/main/java/org/jeecg/common/util/DownloadUtils.java @@ -14,8 +14,8 @@ public class DownloadUtils { public static void downloadTxt(HttpServletResponse response,InputStream is,String fileName){ download(response,is,fileName,MediaType.TEXT_PLAIN_VALUE); } - - private static void download(HttpServletResponse response,InputStream is,String fileName,String contentType){ + + public static void download(HttpServletResponse response,InputStream is,String fileName,String contentType){ response.reset(); response.setContentType(contentType); response.setCharacterEncoding("utf-8"); diff --git a/jeecg-boot-base-core/src/main/java/org/jeecg/modules/base/entity/TransportGenGifLog.java b/jeecg-boot-base-core/src/main/java/org/jeecg/modules/base/entity/TransportGenGifLog.java new file mode 100644 index 0000000..411ff06 --- /dev/null +++ b/jeecg-boot-base-core/src/main/java/org/jeecg/modules/base/entity/TransportGenGifLog.java @@ -0,0 +1,50 @@ +package org.jeecg.modules.base.entity; + +import com.baomidou.mybatisplus.annotation.IdType; +import com.baomidou.mybatisplus.annotation.TableField; +import com.baomidou.mybatisplus.annotation.TableId; +import com.baomidou.mybatisplus.annotation.TableName; +import com.fasterxml.jackson.annotation.JsonFormat; +import lombok.Data; + +import java.time.LocalDateTime; + +/** + * 记录生成gif动图日志 + */ +@Data +@TableName("stas_gen_gif_log") +public class TransportGenGifLog { + + /** + * ID + */ + @TableId(type = IdType.AUTO) + private Integer id; + + /** + * 任务id + */ + @TableField(value = "task_id") + private Integer taskId; + + /** + * 创建时间 + */ + @JsonFormat(timezone = "GMT+8", pattern = "yyyy-MM-dd HH:mm:ss") + @TableField(value = "create_time") + private LocalDateTime createTime; + + /** + * 运行过程日志 + */ + @TableField(value = "log_content") + private String logContent; + + /** + * 核素id + */ + @TableField(value = "species_id") + private Integer speciesId; + +} diff --git a/jeecg-boot-base-core/src/main/java/org/jeecg/modules/base/entity/WeatherDownGFSDataLog.java b/jeecg-boot-base-core/src/main/java/org/jeecg/modules/base/entity/WeatherDownGFSDataLog.java index d7955c7..64a4493 100644 --- a/jeecg-boot-base-core/src/main/java/org/jeecg/modules/base/entity/WeatherDownGFSDataLog.java +++ b/jeecg-boot-base-core/src/main/java/org/jeecg/modules/base/entity/WeatherDownGFSDataLog.java @@ -40,4 +40,11 @@ public class WeatherDownGFSDataLog { */ @TableField(value = "log_content") private String logContent; + + /** + * 日志类型 + */ + @TableField(value = "log_type") + private Integer logType; + } diff --git a/jeecg-boot-base-core/src/main/java/org/jeecg/modules/base/mapper/TransportGenGifLogMapper.java b/jeecg-boot-base-core/src/main/java/org/jeecg/modules/base/mapper/TransportGenGifLogMapper.java new file mode 100644 index 0000000..8524619 --- /dev/null +++ b/jeecg-boot-base-core/src/main/java/org/jeecg/modules/base/mapper/TransportGenGifLogMapper.java @@ -0,0 +1,8 @@ +package org.jeecg.modules.base.mapper; + +import com.baomidou.mybatisplus.core.mapper.BaseMapper; +import org.jeecg.modules.base.entity.TransportGenGifLog; + +public interface TransportGenGifLogMapper extends BaseMapper { + +} diff --git a/jeecg-boot-base-core/src/main/java/org/jeecg/modules/base/mapper/xml/TransportTaskMapper.xml b/jeecg-boot-base-core/src/main/java/org/jeecg/modules/base/mapper/xml/TransportTaskMapper.xml index 6a72cee..68e2300 100644 --- a/jeecg-boot-base-core/src/main/java/org/jeecg/modules/base/mapper/xml/TransportTaskMapper.xml +++ b/jeecg-boot-base-core/src/main/java/org/jeecg/modules/base/mapper/xml/TransportTaskMapper.xml @@ -12,6 +12,8 @@ t.use_met_type as useMetType, t.time_consuming as timeConsuming, t.create_time as createTime, + t.start_time as startTime, + t.end_time as endTime, t.top_task as topTask from stas_transport_task t diff --git a/jeecg-model-consumer/src/main/java/org/jeecg/rebuild/consumer/RebuildTaskConsumerHandler.java b/jeecg-model-consumer/src/main/java/org/jeecg/rebuild/consumer/RebuildTaskConsumerHandler.java index b42bf69..bb3cad9 100644 --- a/jeecg-model-consumer/src/main/java/org/jeecg/rebuild/consumer/RebuildTaskConsumerHandler.java +++ b/jeecg-model-consumer/src/main/java/org/jeecg/rebuild/consumer/RebuildTaskConsumerHandler.java @@ -82,19 +82,31 @@ public class RebuildTaskConsumerHandler { private class MessageConsumerThread extends Thread{ @Override public void run() { + Boolean lastEnable = null; while (true) { try { - //获取本机项数据，如果为空，表示本机没有任务在运行 - boolean flag = redisUtil.hHasKey(CommonConstant.HOST_TASK_STATE,CommonConstant.BUILD_TASK_STATE_PRE+serverProperties.getHost()); - if (!flag) { - SourceRebuildTask topMessage = getTopMessage(); - if (Objects.nonNull(topMessage)) { - handlerTask(topMessage); + boolean currentEnable = serverProperties.isConsumerEnable(); + if (currentEnable) { + //获取本机项数据，如果为空，表示本机没有任务在运行 + boolean flag = redisUtil.hHasKey(CommonConstant.HOST_TASK_STATE,CommonConstant.BUILD_TASK_STATE_PRE+serverProperties.getHost()); + if (!flag) { + SourceRebuildTask topMessage = getTopMessage(); + if (Objects.nonNull(topMessage)) { + handlerTask(topMessage); + } } } + if(!Objects.equals(currentEnable, lastEnable)){ + if (currentEnable) { + log.info("源项重建任务执行开关已打开，开始接收并执行新的任务"); + }else { + log.info("源项重建任务执行开关已关闭，停止接收新的任务"); + } + lastEnable = currentEnable; + } TimeUnit.SECONDS.sleep(60); } catch (InterruptedException e) { - log.error("执行源项重建任务数据异常，原因为：",e); + log.error("执行源项重建任务出现异常，原因为：",e); } } } diff --git a/jeecg-model-consumer/src/main/java/org/jeecg/transport/consumer/TranTaskConsumerHandler.java b/jeecg-model-consumer/src/main/java/org/jeecg/transport/consumer/TranTaskConsumerHandler.java index b02b6ef..ff3f8db 100644 --- a/jeecg-model-consumer/src/main/java/org/jeecg/transport/consumer/TranTaskConsumerHandler.java +++ b/jeecg-model-consumer/src/main/java/org/jeecg/transport/consumer/TranTaskConsumerHandler.java @@ -91,19 +91,31 @@ public class TranTaskConsumerHandler{ @Override public void run() { + Boolean lastEnable = null; while (true) { try { - //获取本机项数据，如果为空，表示本机没有任务在运行 - boolean flag = redisUtil.hHasKey(CommonConstant.HOST_TASK_STATE,CommonConstant.TRAN_TASK_STATE_PRE+serverProperties.getHost()); - if (!flag) { - TransportTask topMessage = getTopMessage(); - if (Objects.nonNull(topMessage)) { - handlerTask(topMessage); + boolean currentEnable = serverProperties.isConsumerEnable(); + if (currentEnable) { + //获取本机项数据，如果为空，表示本机没有任务在运行 + boolean flag = redisUtil.hHasKey(CommonConstant.HOST_TASK_STATE,CommonConstant.TRAN_TASK_STATE_PRE+serverProperties.getHost()); + if (!flag) { + TransportTask topMessage = getTopMessage(); + if (Objects.nonNull(topMessage)) { + handlerTask(topMessage); + } } } + if(!Objects.equals(currentEnable, lastEnable)){ + if (currentEnable) { + log.info("大气输运任务执行开关已打开，开始接收并执行新的任务"); + }else { + log.info("大气输运任务执行开关已关闭，停止接收新的任务"); + } + lastEnable = currentEnable; + } TimeUnit.SECONDS.sleep(60); } catch (InterruptedException e) { - log.error("执行输运任务数据异常，原因为：",e); + log.error("执行输运任务出现异常，原因为：",e); } } } diff --git a/jeecg-model-consumer/src/main/java/org/jeecg/transport/consumer/china/AbstractTaskMsgHandler.java b/jeecg-model-consumer/src/main/java/org/jeecg/transport/consumer/china/AbstractTaskMsgHandler.java index 9f248b3..4544dd4 100644 --- a/jeecg-model-consumer/src/main/java/org/jeecg/transport/consumer/china/AbstractTaskMsgHandler.java +++ b/jeecg-model-consumer/src/main/java/org/jeecg/transport/consumer/china/AbstractTaskMsgHandler.java @@ -71,15 +71,6 @@ public abstract class AbstractTaskMsgHandler extends AbstractChain{ this.setChina(); } - /** - * 初始化下个节点 - * @param serverProperties - */ - protected void initNext(ServerProperties serverProperties){ - this.serverProperties = serverProperties; - this.setChina(); - } - /** * 处理消息 */ @@ -91,34 +82,39 @@ public abstract class AbstractTaskMsgHandler extends AbstractChain{ */ protected void runTask(TransportTask transportTask,String ip){ boolean flag = false; - if (TransportTaskModeEnum.BACK_FORWARD.getKey().equals(transportTask.getTaskMode())){ - AbstractTaskExec taskExec = new BackwardTaskExec(); - taskExec.init(weatherDataMapper,transportTaskService,transportTask, - simulationProperties,systemStorageProperties, - dataFusionProperties,serverProperties,taskBackwardChildMapper,redisUtil); - taskExec.setName("flexpart-backword-thread"); - //匹配成功并且检查成功才能设置为true - flag = taskExec.checkTask(); - if (flag){ - taskExec.start(); + try{ + if (TransportTaskModeEnum.BACK_FORWARD.getKey().equals(transportTask.getTaskMode())){ + AbstractTaskExec taskExec = new BackwardTaskExec(); + taskExec.init(weatherDataMapper,transportTaskService,transportTask, + simulationProperties,systemStorageProperties, + dataFusionProperties,serverProperties,taskBackwardChildMapper,redisUtil); + taskExec.setName("flexpart-backword-thread"); + //匹配成功并且检查成功才能设置为true + flag = taskExec.checkTask(); + if (flag){ + taskExec.start(); + } + }else if (TransportTaskModeEnum.FORWARD.getKey().equals(transportTask.getTaskMode())){ + AbstractTaskExec taskExec = new ForwardTaskExec(); + taskExec.init(weatherDataMapper,transportTaskService,transportTask, + simulationProperties,systemStorageProperties,serverProperties, + stationDataService,stationsModValService,taskForwardSpeciesMapper, + taskForwardChildMapper,taskForwardReleaseMapper,redisUtil); + taskExec.setName("flexpart-fword-thread"); + //匹配成功并且检查成功才能设置为true + flag = taskExec.checkTask(); + if (flag){ + taskExec.start(); + } } - }else if (TransportTaskModeEnum.FORWARD.getKey().equals(transportTask.getTaskMode())){ - AbstractTaskExec taskExec = new ForwardTaskExec(); - taskExec.init(weatherDataMapper,transportTaskService,transportTask, - simulationProperties,systemStorageProperties,serverProperties, - stationDataService,stationsModValService,taskForwardSpeciesMapper, - taskForwardChildMapper,taskForwardReleaseMapper,redisUtil); - taskExec.setName("flexpart-fword-thread"); - //匹配成功并且检查成功才能设置为true - flag = taskExec.checkTask(); if (flag){ - taskExec.start(); + log.info("任务：{}，检查通过，当前在主机：{}上执行。", transportTask.getTaskName(),ip); + }else { + transportTaskService.setInpectionFailed(transportTask); } - } - if (flag){ - log.info("任务：{}，检查通过，当前在主机：{}上执行。", transportTask.getTaskName(),ip); - }else { + }catch (Exception e){ transportTaskService.setInpectionFailed(transportTask); + log.error("任务检查失败",e); } } } diff --git a/jeecg-model-consumer/src/main/java/org/jeecg/transport/consumer/china/Server11TaskHandler.java b/jeecg-model-consumer/src/main/java/org/jeecg/transport/consumer/china/Server11TaskHandler.java index 495bb19..e46d68b 100644 --- a/jeecg-model-consumer/src/main/java/org/jeecg/transport/consumer/china/Server11TaskHandler.java +++ b/jeecg-model-consumer/src/main/java/org/jeecg/transport/consumer/china/Server11TaskHandler.java @@ -15,7 +15,13 @@ public class Server11TaskHandler extends AbstractTaskMsgHandler { @Override protected void setChina() { AbstractTaskMsgHandler taskHandler = new Server12TaskHandler(); - taskHandler.initNext(super.serverProperties); + taskHandler.init(super.transportTaskMapper,super.taskBackwardChildMapper, + super.weatherDataMapper,simulationProperties, + super.systemStorageProperties,super.dataFusionProperties, + super.serverProperties,super.taskForwardSpeciesMapper, + super.taskForwardChildMapper,super.transportTaskService, + super.taskForwardReleaseMapper,super.stationDataService, + super.stationsModValService,super.redisUtil); taskHandler.setPrevious(this); super.setNext(taskHandler); } diff --git a/jeecg-model-consumer/src/main/java/org/jeecg/transport/consumer/china/Server12TaskHandler.java b/jeecg-model-consumer/src/main/java/org/jeecg/transport/consumer/china/Server12TaskHandler.java index 91656e3..f136dca 100644 --- a/jeecg-model-consumer/src/main/java/org/jeecg/transport/consumer/china/Server12TaskHandler.java +++ b/jeecg-model-consumer/src/main/java/org/jeecg/transport/consumer/china/Server12TaskHandler.java @@ -13,7 +13,13 @@ public class Server12TaskHandler extends AbstractTaskMsgHandler { @Override protected void setChina() { AbstractTaskMsgHandler taskHandler = new Server13TaskHandler(); - taskHandler.initNext(super.serverProperties); + taskHandler.init(super.transportTaskMapper,super.taskBackwardChildMapper, + super.weatherDataMapper,simulationProperties, + super.systemStorageProperties,super.dataFusionProperties, + super.serverProperties,super.taskForwardSpeciesMapper, + super.taskForwardChildMapper,super.transportTaskService, + super.taskForwardReleaseMapper,super.stationDataService, + super.stationsModValService,super.redisUtil); taskHandler.setPrevious(this); super.setNext(taskHandler); } diff --git a/jeecg-model-consumer/src/main/java/org/jeecg/transport/consumer/china/Server13TaskHandler.java b/jeecg-model-consumer/src/main/java/org/jeecg/transport/consumer/china/Server13TaskHandler.java index 22bac0b..6151ef4 100644 --- a/jeecg-model-consumer/src/main/java/org/jeecg/transport/consumer/china/Server13TaskHandler.java +++ b/jeecg-model-consumer/src/main/java/org/jeecg/transport/consumer/china/Server13TaskHandler.java @@ -13,7 +13,13 @@ public class Server13TaskHandler extends AbstractTaskMsgHandler { @Override protected void setChina() { AbstractTaskMsgHandler taskHandler = new Server14TaskHandler(); - taskHandler.initNext(super.serverProperties); + taskHandler.init(super.transportTaskMapper,super.taskBackwardChildMapper, + super.weatherDataMapper,simulationProperties, + super.systemStorageProperties,super.dataFusionProperties, + super.serverProperties,super.taskForwardSpeciesMapper, + super.taskForwardChildMapper,super.transportTaskService, + super.taskForwardReleaseMapper,super.stationDataService, + super.stationsModValService,super.redisUtil); taskHandler.setPrevious(this); super.setNext(taskHandler); } diff --git a/jeecg-model-consumer/src/main/java/org/jeecg/transport/flexparttask/AbstractTaskExec.java b/jeecg-model-consumer/src/main/java/org/jeecg/transport/flexparttask/AbstractTaskExec.java index f4ca828..4b93366 100644 --- a/jeecg-model-consumer/src/main/java/org/jeecg/transport/flexparttask/AbstractTaskExec.java +++ b/jeecg-model-consumer/src/main/java/org/jeecg/transport/flexparttask/AbstractTaskExec.java @@ -26,6 +26,7 @@ import org.jeecg.transport.service.StationDataService; import org.jeecg.transport.service.StationsModValService; import org.jeecg.transport.service.TransportTaskService; import java.io.*; +import java.time.LocalDate; import java.time.LocalDateTime; import java.time.format.DateTimeFormatter; import java.time.temporal.ChronoUnit; @@ -122,9 +123,81 @@ public abstract class AbstractTaskExec extends Thread{ * 检查气象数据 */ protected boolean checkMetData(List msgList){ + if(WeatherDataSourceEnum.GFS.getKey().equals(this.transportTask.getUseMetType())){ + return checkGfsMetData(msgList); + }else { + return checkOtherData(msgList); + } + } + + + /** + * 校验GFS数据 + * @param msgList + * @return + */ + private boolean checkGfsMetData(List msgList){ + boolean checkFlag = true; + LocalDateTime startTime = this.transportTask.getStartTime(); + LocalDateTime endTime = this.transportTask.getEndTime(); + if(TransportTaskModeEnum.BACK_FORWARD.getKey().equals(this.transportTask.getTaskMode())){ + //生成批次信息 + List batches = new ArrayList<>(); + while (!startTime.isAfter(endTime)){ + batches.add(startTime.format(DateTimeFormatter.ofPattern("yyyyMMdd"))); + startTime = startTime.plusDays(1); + } + if(CollUtil.isNotEmpty(batches)){ + for (String batch : batches){ + //查询批次的第一天的气象数据是否有缺失 + LocalDate batchDate = LocalDate.parse(batch, DateTimeFormatter.ofPattern("yyyyMMdd")); + LocalDateTime batchStartTime = batchDate.atTime(0,0,0); + LocalDateTime batchEndTime = batchDate.atTime(23,59,59); + LambdaQueryWrapper queryWrapper = new LambdaQueryWrapper<>(); + queryWrapper.eq(WeatherData::getDataSource,this.transportTask.getUseMetType()); + queryWrapper.eq(WeatherData::getTimeBatch,batch); + queryWrapper.between(WeatherData::getDataStartTime,batchStartTime,batchEndTime); + Long dataCount = this.weatherDataMapper.selectCount(queryWrapper); + if(dataCount != 24){ + checkFlag = false; + String exceptionMsg = "%s批次气象数据前24小时逐小时数据有缺失，请确认"; + String formatMsg = String.format(exceptionMsg,batch); + msgList.add(formatMsg); + } + } + } + }else if(TransportTaskModeEnum.FORWARD.getKey().equals(this.transportTask.getTaskMode())){ + String batchTime = startTime.format(DateTimeFormatter.ofPattern("yyyyMMdd")); + long days = ChronoUnit.DAYS.between(startTime, endTime); + //gfs每批次数据是16天，所以如果正演使用gfs批次数据不能超过16天 + if(days > 16){ + String log = "模拟时间范围大于16天，已超出"+batchTime+"批次数据范围"; + ProgressQueue.getInstance().offer(new ProgressEvent(this.transportTask.getId(),log)); + throw new RuntimeException(log); + } + //gfs每批次数据是16天，下载文件是209个，如果不是说明有缺失，后续weather模块会自动检查并下载缺失文件 + LambdaQueryWrapper queryWrapper = new LambdaQueryWrapper<>(); + queryWrapper.eq(WeatherData::getDataSource,this.transportTask.getUseMetType()); + queryWrapper.eq(WeatherData::getTimeBatch,batchTime); + Long dataCount = this.weatherDataMapper.selectCount(queryWrapper); + if(dataCount < 209){ + checkFlag = false; + String exceptionMsg = "下载的%s批次气象数据有缺失，请确认"; + String formatMsg = String.format(exceptionMsg,batchTime); + msgList.add(formatMsg); + } + } + return checkFlag; + } + + /** + * 校验除GFS数据以外其他气象数据 + * @param msgList + * @return + */ + private boolean checkOtherData(List msgList){ String msg = "检查气象数据"; ProgressQueue.getInstance().offer(new ProgressEvent(this.transportTask.getId(),msg)); - LocalDateTime startTime = this.transportTask.getStartTime(); LocalDateTime endTime = this.transportTask.getEndTime(); LambdaQueryWrapper queryWrapper = new LambdaQueryWrapper<>(); @@ -160,17 +233,19 @@ public abstract class AbstractTaskExec extends Thread{ File ncFile = new File(ncPath); StringBuilder command = new StringBuilder(); command.append(simulationProperties.getPythonEnvPath()); + command.append(" -u "); command.append(StringPool.SPACE); command.append(simulationProperties.getGifConvertScriptPath()); - command.append(" --nc_file "); - command.append(ncPath); command.append(" --mode "); command.append(mode); - command.append(" --gif_file "); + command.append(" --area global ");//默认生成全球的 + command.append(" --ncdir "); + command.append(ncPath); + command.append(" --outputdir "); command.append(ncFile.getParent()); command.append(File.separator); command.append(ncFile.getParentFile().getName()); - command.append(".gif"); + command.append("_global.gif"); this.execGenGIFCommand(command.toString(),session); } }catch (Exception e){ @@ -272,6 +347,7 @@ public abstract class AbstractTaskExec extends Thread{ if(TransportTaskModeEnum.FORWARD.getKey().equals(transportTask.getTaskMode())){ DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyyMMdd_HH"); String format = formatter.format(transportTask.getStartTime()); + path.append(systemStorageProperties.getGfsDataPath()); path.append(File.separator); path.append("GFS_"+format); }else { diff --git a/jeecg-model-consumer/src/main/java/org/jeecg/transport/flexparttask/BackwardTaskExec.java b/jeecg-model-consumer/src/main/java/org/jeecg/transport/flexparttask/BackwardTaskExec.java index ca93d1c..712f91e 100644 --- a/jeecg-model-consumer/src/main/java/org/jeecg/transport/flexparttask/BackwardTaskExec.java +++ b/jeecg-model-consumer/src/main/java/org/jeecg/transport/flexparttask/BackwardTaskExec.java @@ -14,9 +14,11 @@ import java.io.File; import java.math.BigDecimal; import java.math.RoundingMode; import java.text.DecimalFormat; +import java.time.LocalDate; import java.time.format.DateTimeFormatter; import java.util.ArrayList; import java.util.List; +import java.util.Objects; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; @@ -50,23 +52,32 @@ public class BackwardTaskExec extends AbstractTaskExec { flag.set(false); }else { super.transportTask.setBackwardChild(transportTaskChildren); + transportTaskChildren.forEach(taskBackwardChild -> { + if(taskBackwardChild.getAcqStartTime().isAfter(taskBackwardChild.getAcqEndTime())){ + msgList.add(taskBackwardChild.getStationCode()+"站点，样品测量开始时间不能在结束时间之后，请确认"); + flag.set(false); + } + if(taskBackwardChild.getAcqStartTime().isAfter(super.transportTask.getEndTime())){ + msgList.add(taskBackwardChild.getStationCode()+"站点，样品测量结束时间不能在任务模拟结束时间之后，请确认"); + flag.set(false); + } + if(taskBackwardChild.getAcqStartTime().isBefore(super.transportTask.getStartTime())){ + msgList.add(taskBackwardChild.getStationCode()+"站点，样品测量开始时间不能在任务模拟开始时间之前，请确认"); + flag.set(false); + } + }); } - transportTaskChildren.forEach(taskBackwardChild -> { - if(taskBackwardChild.getAcqStartTime().isAfter(taskBackwardChild.getAcqEndTime())){ - msgList.add(taskBackwardChild.getStationCode()+"站点，样品测量开始时间不能在结束时间之后，请确认"); - flag.set(false); - } - if(taskBackwardChild.getAcqStartTime().isAfter(super.transportTask.getEndTime())){ - msgList.add(taskBackwardChild.getStationCode()+"站点，样品测量结束时间不能在任务模拟结束时间之后，请确认"); - flag.set(false); - } - if(taskBackwardChild.getAcqStartTime().isBefore(super.transportTask.getStartTime())){ - msgList.add(taskBackwardChild.getStationCode()+"站点，样品测量开始时间不能在任务模拟开始时间之前，请确认"); - flag.set(false); - } - }); //检查气象数据 - flag.set(super.checkMetData(msgList)); + boolean megCheckResult = super.checkMetData(msgList); + if(!megCheckResult){ + //如果气象数据连续5天检查都缺失，那么才真正设置气象数据的检查结果为false + LocalDate updateTime = this.transportTask.getUpdateTime().toLocalDate(); + updateTime = updateTime.plusDays(super.simulationProperties.getMetCheckMaxFailureDays()); + LocalDate now = LocalDate.now(); + if(updateTime.isBefore(now)){ + flag.set(false); + } + } if (!flag.get()) { if (CollUtil.isNotEmpty(msgList)) { msgList.forEach(msg->{ @@ -96,7 +107,7 @@ public class BackwardTaskExec extends AbstractTaskExec { //执行模拟 this.execSimulation(); //生成SRS文件 -// this.generateSRSFile(); + this.generateSRSFile(); }catch (Exception e){ super.taskRunError = true; String taskErrorLog = "任务执行失败，原因："; @@ -217,9 +228,12 @@ public class BackwardTaskExec extends AbstractTaskExec { */ private void generateSRSFile(){ for (TransportTaskBackwardChild transportTaskChild : super.transportTask.getBackwardChild()){ - BuildNcToSrsFile ncToSrsFile = new BuildNcToSrsFile(); - ncToSrsFile.init(dataFusionProperties,simulationProperties,transportTask,transportTaskChild,transportTaskChild.getReleaseAmount()); - ncToSrsFile.execute(); + //如果是分析系统来的反演数据，会有样品id和样品类型会生成srs文件，如果是氙本地源系统新增的，没有生成srs文件，因为没有样品id和类型参数 + if(Objects.nonNull(transportTaskChild.getSampleId()) && Objects.nonNull(transportTaskChild.getSampleType())){ + BuildNcToSrsFile ncToSrsFile = new BuildNcToSrsFile(); + ncToSrsFile.init(dataFusionProperties,simulationProperties,transportTask,transportTaskChild,transportTaskChild.getReleaseAmount()); + ncToSrsFile.execute(); + } } } } diff --git a/jeecg-model-consumer/src/main/java/org/jeecg/transport/flexparttask/BuildNcToSrsFile.java b/jeecg-model-consumer/src/main/java/org/jeecg/transport/flexparttask/BuildNcToSrsFile.java index ec74056..d1c9a60 100644 --- a/jeecg-model-consumer/src/main/java/org/jeecg/transport/flexparttask/BuildNcToSrsFile.java +++ b/jeecg-model-consumer/src/main/java/org/jeecg/transport/flexparttask/BuildNcToSrsFile.java @@ -16,7 +16,6 @@ import ucar.ma2.InvalidRangeException; import ucar.nc2.Attribute; import ucar.nc2.NetcdfFile; import ucar.nc2.Variable; - import java.io.File; import java.io.FileOutputStream; import java.io.IOException; @@ -27,8 +26,8 @@ import java.nio.file.Path; import java.nio.file.Paths; import java.text.DecimalFormat; import java.time.LocalDateTime; -import java.time.ZoneId; import java.time.format.DateTimeFormatter; +import java.time.temporal.ChronoUnit; import java.util.ArrayList; import java.util.List; import java.util.zip.GZIPOutputStream; @@ -103,9 +102,7 @@ public class BuildNcToSrsFile { //处理模拟总时长 LocalDateTime taskStartTime = transportTask.getStartTime(); LocalDateTime taskEndTime = transportTask.getEndTime(); - long startHour = taskStartTime.atZone(ZoneId.of("Asia/Shanghai")).getHour(); - long endHour = taskEndTime.atZone(ZoneId.of("Asia/Shanghai")).getHour(); - long totalHour = endHour - startHour; + long totalHour = ChronoUnit.HOURS.between(taskStartTime, taskEndTime); //处理输出频率[小时] 和 平均时间[小时]参数 Attribute loutstep = ncFile.findGlobalAttribute("loutstep"); Attribute loutaver = ncFile.findGlobalAttribute("loutaver"); @@ -149,7 +146,7 @@ public class BuildNcToSrsFile { double dyout = dyoutVar.getNumericValue().doubleValue(); //处理输出频率[小时] Attribute loutstep = ncFile.findGlobalAttribute("loutstep"); - double outputFreHour = loutstep.getNumericValue().doubleValue(); + double outputFreSecond = Math.abs(loutstep.getNumericValue().doubleValue()); List body = new ArrayList<>(); Variable spec001Mr = ncFile.findVariable("spec001_mr"); @@ -169,7 +166,7 @@ public class BuildNcToSrsFile { double dy_1d = radius * pi / 180; double area=dyout * dxout * dy_1d * dx_1d; double volume=area*100;//height = 100 米（固定的高度层厚度） - double factor = pointVal / (outputFreHour * 3600) / volume * factorMul; + double factor = pointVal / outputFreSecond / volume * factorMul; DecimalFormat scientificFormat = new DecimalFormat("0.0000000E0"); String result = scientificFormat.format(factor); String line = String.format(" %-7s %8s %-4s %s",currentLat,currentLon,t+1,result); @@ -239,6 +236,8 @@ public class BuildNcToSrsFile { return WeatherDataSourceEnum.FNL.name().toLowerCase(); }else if (WeatherDataSourceEnum.T1H.getKey().equals(transportTask.getUseMetType())){ return WeatherDataSourceEnum.T1H.name().toLowerCase(); + }else if (WeatherDataSourceEnum.GFS.getKey().equals(transportTask.getUseMetType())){ + return WeatherDataSourceEnum.GFS.name().toLowerCase(); } return Strings.EMPTY; } diff --git a/jeecg-model-consumer/src/main/java/org/jeecg/transport/flexparttask/ForwardTaskExec.java b/jeecg-model-consumer/src/main/java/org/jeecg/transport/flexparttask/ForwardTaskExec.java index 54b1f8d..c3f70b4 100644 --- a/jeecg-model-consumer/src/main/java/org/jeecg/transport/flexparttask/ForwardTaskExec.java +++ b/jeecg-model-consumer/src/main/java/org/jeecg/transport/flexparttask/ForwardTaskExec.java @@ -13,6 +13,7 @@ import java.io.File; import java.math.BigDecimal; import java.math.RoundingMode; import java.text.DecimalFormat; +import java.time.LocalDate; import java.time.format.DateTimeFormatter; import java.util.*; import java.util.concurrent.TimeUnit; @@ -82,7 +83,16 @@ public class ForwardTaskExec extends AbstractTaskExec { } } //检查气象数据 - flag.set(super.checkMetData(msgList)); + boolean megCheckResult = super.checkMetData(msgList); + if(!megCheckResult){ + //如果气象数据连续5天检查都缺失，那么才真正设置气象数据的检查结果为false + LocalDate updateTime = this.transportTask.getUpdateTime().toLocalDate(); + updateTime = updateTime.plusDays(super.simulationProperties.getMetCheckMaxFailureDays()); + LocalDate now = LocalDate.now(); + if(updateTime.isBefore(now)){ + flag.set(false); + } + } if (!flag.get()) { if (CollUtil.isNotEmpty(msgList)) { msgList.forEach(msg->{ diff --git a/jeecg-model-consumer/src/main/java/org/jeecg/weather/consumer/WeatherTaskConsumerHandler.java b/jeecg-model-consumer/src/main/java/org/jeecg/weather/consumer/WeatherTaskConsumerHandler.java index 1e8d7d4..48b7265 100644 --- a/jeecg-model-consumer/src/main/java/org/jeecg/weather/consumer/WeatherTaskConsumerHandler.java +++ b/jeecg-model-consumer/src/main/java/org/jeecg/weather/consumer/WeatherTaskConsumerHandler.java @@ -80,19 +80,31 @@ public class WeatherTaskConsumerHandler { private class MessageConsumerThread extends Thread{ @Override public void run() { + Boolean lastEnable = null; while (true) { try { - //获取本机项数据，如果为空，表示本机没有任务在运行 - boolean flag = redisUtil.hHasKey(CommonConstant.HOST_TASK_STATE,CommonConstant.WEATHER_TASK_STATE_PRE+serverProperties.getHost()); - if (!flag) { - WeatherTask topMessage = getTopMessage(); - if (Objects.nonNull(topMessage)) { - handlerTask(topMessage); + boolean currentEnable = serverProperties.isConsumerEnable(); + if (currentEnable) { + //获取本机项数据，如果为空，表示本机没有任务在运行 + boolean flag = redisUtil.hHasKey(CommonConstant.HOST_TASK_STATE,CommonConstant.WEATHER_TASK_STATE_PRE+serverProperties.getHost()); + if (!flag) { + WeatherTask topMessage = getTopMessage(); + if (Objects.nonNull(topMessage)) { + handlerTask(topMessage); + } } } + if(!Objects.equals(currentEnable, lastEnable)){ + if (currentEnable) { + log.info("天气预测任务执行开关已打开，开始接收并执行新的任务"); + }else { + log.info("天气预测任务执行开关已关闭，停止接收新的任务"); + } + lastEnable = currentEnable; + } TimeUnit.SECONDS.sleep(30); } catch (InterruptedException e) { - log.error("执行源项重建任务数据异常，原因为：",e); + log.error("执行天气预测任务出现异常，原因为：",e); } } } diff --git a/jeecg-model-consumer/src/main/java/org/jeecg/weather/task/WeatherForecastTaskExec.java b/jeecg-model-consumer/src/main/java/org/jeecg/weather/task/WeatherForecastTaskExec.java index 54252d7..99fb69d 100644 --- a/jeecg-model-consumer/src/main/java/org/jeecg/weather/task/WeatherForecastTaskExec.java +++ b/jeecg-model-consumer/src/main/java/org/jeecg/weather/task/WeatherForecastTaskExec.java @@ -11,7 +11,6 @@ import org.jeecg.common.constant.enums.WeatherDataSourceEnum; import org.jeecg.common.constant.enums.WeatherFileSuffixEnum; import org.jeecg.common.constant.enums.WeatherForecastDatasourceEnum; import org.jeecg.common.constant.enums.WeatherTaskStatusEnum; -import org.jeecg.common.util.Grib2TimeReader; import org.jeecg.common.util.JSchRemoteRunner; import org.jeecg.modules.base.entity.WeatherData; import org.springframework.http.MediaType; @@ -19,6 +18,7 @@ import org.springframework.web.reactive.function.client.WebClient; import java.io.*; import java.math.BigDecimal; import java.math.RoundingMode; +import java.time.LocalDateTime; import java.time.format.DateTimeFormatter; import java.util.*; import java.util.concurrent.TimeUnit; @@ -280,11 +280,21 @@ public class WeatherForecastTaskExec extends AbstractWeatherTask { List dataList = new ArrayList<>(); for(File sourceFile : sourceFiles){ try{ + LocalDateTime dateTime = null; + if(WeatherDataSourceEnum.PANGU.getKey().equals(weatherTask.getPredictionModel())){ + String dateTimeStr = sourceFile.getName().substring("pangu_".length(), sourceFile.getName().lastIndexOf(".")); + DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyyMMdd_HH_mm"); + dateTime = LocalDateTime.parse(dateTimeStr, formatter).withSecond(0); + }else if(WeatherDataSourceEnum.GRAPHCAST.getKey().equals(weatherTask.getPredictionModel())){ + String dateTimeStr = sourceFile.getName().substring("graphcast_".length(), sourceFile.getName().lastIndexOf(".")); + DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyyMMdd_HH_mm"); + dateTime = LocalDateTime.parse(dateTimeStr, formatter).withSecond(0); + } //构建文件信息 WeatherData weatherData = new WeatherData(); weatherData.setFileName(sourceFile.getName()); weatherData.setFileExt(sourceFile.getName().substring(sourceFile.getName().lastIndexOf(".")+1)); - weatherData.setDataStartTime(Grib2TimeReader.readValidTime(sourceFilesFinalPath+File.separator+sourceFile.getName())); + weatherData.setDataStartTime(dateTime); weatherData.setDataSource(weatherTask.getPredictionModel()); weatherData.setFilePath(sourceFilesFinalPath+File.separator+sourceFile.getName()); weatherData.setFormatFilePath(formatFilesFinalPath+File.separator+sourceFile.getName().substring(0,sourceFile.getName().lastIndexOf(".")+1)+ WeatherFileSuffixEnum.GRIB2.getValue()); diff --git a/jeecg-module-source-rebuild/src/main/java/org/jeecg/controller/TaskResultDataController.java b/jeecg-module-source-rebuild/src/main/java/org/jeecg/controller/TaskResultDataController.java index 8877524..6341267 100644 --- a/jeecg-module-source-rebuild/src/main/java/org/jeecg/controller/TaskResultDataController.java +++ b/jeecg-module-source-rebuild/src/main/java/org/jeecg/controller/TaskResultDataController.java @@ -58,28 +58,28 @@ public class TaskResultDataController { @AutoLog(value = "导出概率分布图数据") @Operation(summary = "导出概率分布图数据") @GetMapping("exportBayesProbLocTxt") - public void exportBayesProbLocTxt(@NotNull(message = "任务ID不能为空") Integer taskId,HttpServletResponse response) throws IOException { + public void exportBayesProbLocTxt(@NotNull(message = "任务ID不能为空") Integer taskId,HttpServletResponse response) { taskResultDataService.exportBayesProbLocTxt(taskId,response); } @AutoLog(value = "导出观测值和模拟值的活度浓度比较结果数据") @Operation(summary = "导出观测值和模拟值的活度浓度比较结果数据") @GetMapping("exportBayesAcTimeSeriesTxt") - public void exportBayesAcTimeSeriesTxt(@NotNull(message = "任务ID不能为空") Integer taskId,HttpServletResponse response) throws IOException { + public void exportBayesAcTimeSeriesTxt(@NotNull(message = "任务ID不能为空") Integer taskId,HttpServletResponse response) { taskResultDataService.exportBayesAcTimeSeriesTxt(taskId,response); } @AutoLog(value = "导出单变量后验分布结果数据") @Operation(summary = "导出单变量后验分布结果数据") @GetMapping("exportBayesMonovarPosterior") - public void exportBayesMonovarPosterior(@NotNull(message = "任务ID不能为空") Integer taskId,HttpServletResponse response) throws IOException { + public void exportBayesMonovarPosterior(@NotNull(message = "任务ID不能为空") Integer taskId,HttpServletResponse response) { taskResultDataService.exportBayesMonovarPosterior(taskId,response); } @AutoLog(value = "导出双变量后验分布结果数据") @Operation(summary = "导出双变量后验分布结果数据") @GetMapping("exportBayesBivarPosterior") - public void exportBayesBivarPosterior(@NotNull(message = "任务ID不能为空") Integer taskId,HttpServletResponse response) throws IOException { + public void exportBayesBivarPosterior(@NotNull(message = "任务ID不能为空") Integer taskId,HttpServletResponse response) { taskResultDataService.exportBayesBivarPosterior(taskId,response); } } diff --git a/jeecg-module-transport/src/main/java/org/jeecg/consumer/SampleMessageConsumer.java b/jeecg-module-transport/src/main/java/org/jeecg/consumer/SampleMessageConsumer.java index 68d3c80..b633b95 100644 --- a/jeecg-module-transport/src/main/java/org/jeecg/consumer/SampleMessageConsumer.java +++ b/jeecg-module-transport/src/main/java/org/jeecg/consumer/SampleMessageConsumer.java @@ -58,7 +58,7 @@ public class SampleMessageConsumer implements RocketMQListener backwardChild = new ArrayList<>(); TransportTaskBackwardChild taskBackwardChild = new TransportTaskBackwardChild(); diff --git a/jeecg-module-transport/src/main/java/org/jeecg/controller/TransportResultDataController.java b/jeecg-module-transport/src/main/java/org/jeecg/controller/TransportResultDataController.java index 39b8096..250662b 100644 --- a/jeecg-module-transport/src/main/java/org/jeecg/controller/TransportResultDataController.java +++ b/jeecg-module-transport/src/main/java/org/jeecg/controller/TransportResultDataController.java @@ -1,7 +1,9 @@ package org.jeecg.controller; import io.swagger.v3.oas.annotations.Operation; +import jakarta.servlet.http.HttpServletResponse; import jakarta.validation.constraints.NotBlank; +import jakarta.validation.constraints.NotEmpty; import jakarta.validation.constraints.NotNull; import lombok.RequiredArgsConstructor; import org.jeecg.common.api.vo.Result; @@ -119,7 +121,40 @@ public class TransportResultDataController { @AutoLog(value = "查询任务所属GIF文件地址") @Operation(summary = "查询任务所属GIF文件地址") @GetMapping("getTaskGifAddr") - public Result getTaskGifAddr(@NotNull(message = "任务id不能为空") Integer taskId,Integer speciesId) { - return Result.OK(transportResultDataService.getTaskGifAddr(taskId,speciesId)); + public Result getTaskGifAddr( + @NotNull(message = "任务id不能为空") Integer taskId, + Integer speciesId, + @NotBlank(message = "gif类型不能为空") String genGifType) { + return Result.OK(transportResultDataService.getTaskGifAddr(taskId,speciesId,genGifType)); + } + + @AutoLog(value = "下载任务所属GIF文件") + @Operation(summary = "下载任务所属GIF文件") + @GetMapping("downloadGif") + public void downloadGif(HttpServletResponse response, + @NotNull(message = "任务id不能为空") Integer taskId, + Integer speciesId, + @NotBlank(message = "生成GIF类型不能为空") String genGifType) { + transportResultDataService.downloadGif(response,taskId,speciesId,genGifType); + } + + @AutoLog(value = "生成任务所属GIF文件") + @Operation(summary = "生成任务所属GIF文件") + @PostMapping("genGif") + public Result genGif(@NotNull(message = "任务id不能为空") Integer taskId, + Integer speciesId, + @NotNull(message = "最小经度不能为空") Integer lonmin, + @NotNull(message = "最小纬度不能为空") Integer latmin, + @NotNull(message = "最大经度不能为空") Integer lonmax, + @NotNull(message = "最大纬度不能为空") Integer latmax) { + transportResultDataService.genGif(taskId,speciesId,lonmin,latmin,lonmax,latmax); + return Result.OK("任务执行成功，请关注执行过程日志"); + } + + @AutoLog(value = "获取任务生成GIF文件过程日志") + @Operation(summary = "获取任务生成GIF文件过程日志") + @GetMapping("getGenGiflog") + public Result getGenGiflog(@NotNull(message = "任务id不能为空") Integer taskId,Integer speciesId) { + return Result.OK(transportResultDataService.getGenGiflog(taskId,speciesId)); } } diff --git a/jeecg-module-transport/src/main/java/org/jeecg/service/TransportResultDataService.java b/jeecg-module-transport/src/main/java/org/jeecg/service/TransportResultDataService.java index 82b12fa..325a8b7 100644 --- a/jeecg-module-transport/src/main/java/org/jeecg/service/TransportResultDataService.java +++ b/jeecg-module-transport/src/main/java/org/jeecg/service/TransportResultDataService.java @@ -1,6 +1,7 @@ package org.jeecg.service; -import jakarta.validation.constraints.NotNull; +import jakarta.servlet.http.HttpServletResponse; +import org.jeecg.modules.base.entity.TransportGenGifLog; import org.jeecg.vo.ContributionAnalysisVO; import org.jeecg.vo.QueryDiffusionVO; import org.jeecg.vo.TaskStationsVO; @@ -114,7 +115,36 @@ public interface TransportResultDataService { * 获取gif地址 * @param taskId * @param speciesId + * @param genGifType * @return */ - String getTaskGifAddr(Integer taskId,Integer speciesId); + String getTaskGifAddr(Integer taskId,Integer speciesId,String genGifType); + + /** + * 生成gif + * @param taskId + * @param speciesId + * @param lonmin + * @param latmin + * @param lonmax + * @param latmax + * + */ + void genGif(Integer taskId, Integer speciesId, Integer lonmin,Integer latmin,Integer lonmax,Integer latmax); + + /** + * 下载gif + * @param response + * @param taskId + * @param speciesId + * @param genGifType + */ + void downloadGif(HttpServletResponse response,Integer taskId, Integer speciesId, String genGifType); + + /** + * 获取生成gif日志 + * @param taskId + * @param speciesId + */ + List getGenGiflog(Integer taskId, Integer speciesId); } diff --git a/jeecg-module-transport/src/main/java/org/jeecg/service/impl/TransportResultDataServiceImpl.java b/jeecg-module-transport/src/main/java/org/jeecg/service/impl/TransportResultDataServiceImpl.java index 4da1298..3092712 100644 --- a/jeecg-module-transport/src/main/java/org/jeecg/service/impl/TransportResultDataServiceImpl.java +++ b/jeecg-module-transport/src/main/java/org/jeecg/service/impl/TransportResultDataServiceImpl.java @@ -6,13 +6,22 @@ import cn.hutool.core.date.LocalDateTimeUtil; import cn.hutool.core.io.FileUtil; import cn.hutool.core.util.StrUtil; import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; +import com.baomidou.mybatisplus.core.toolkit.StringPool; +import com.jcraft.jsch.ChannelExec; +import com.jcraft.jsch.JSchException; +import com.jcraft.jsch.Session; +import jakarta.servlet.http.HttpServletResponse; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.jeecg.common.constant.CommonConstant; import org.jeecg.common.constant.enums.FlexpartSpeciesType; +import org.jeecg.common.constant.enums.TransportTaskGenGifTypeEnum; import org.jeecg.common.constant.enums.TransportTaskModeEnum; import org.jeecg.common.constant.enums.TransportTimingAnalysisEnum; +import org.jeecg.common.properties.ServerProperties; import org.jeecg.common.properties.TransportSimulationProperties; +import org.jeecg.common.util.DownloadUtils; +import org.jeecg.common.util.JSchRemoteRunner; import org.jeecg.common.util.NcUtil; import org.jeecg.common.util.RedisUtil; import org.jeecg.modules.base.entity.*; @@ -28,6 +37,7 @@ import org.jeecg.util.PreSortedTimeRangeQuery; import org.jeecg.vo.ContributionAnalysisVO; import org.jeecg.vo.QueryDiffusionVO; import org.jeecg.vo.TaskStationsVO; +import org.springframework.http.MediaType; import org.springframework.stereotype.Service; import ucar.ma2.Array; import ucar.ma2.DataType; @@ -35,8 +45,8 @@ import ucar.ma2.InvalidRangeException; import ucar.nc2.Attribute; import ucar.nc2.NetcdfFile; import ucar.nc2.Variable; -import java.io.File; -import java.io.IOException; + +import java.io.*; import java.math.BigDecimal; import java.math.RoundingMode; import java.time.LocalDateTime; @@ -50,6 +60,7 @@ import java.util.stream.Collectors; public class TransportResultDataServiceImpl implements TransportResultDataService { private final TransportTaskForwardSpeciesMapper taskForwardSpeciesMapper; + private final TransportGenGifLogMapper genGifLogMapper; private final TransportTaskMapper transportTaskMapper; private final TransportSimulationProperties simulationProperties; private final TransportTaskBackwardChildMapper backwardChildMapper; @@ -57,8 +68,11 @@ public class TransportResultDataServiceImpl implements TransportResultDataServic private final StationDataService stationDataService; private final StationsModValService stationsModValService; private final RedisUtil redisUtil; + private final ServerProperties serverProperties; private final static String FORWARD="forward"; private final static String BACK_FORWARD="backward"; + private final static String GLOBAL_GIF_SUFFIX="_global.gif"; + private final static String REGION_GIF_SUFFIX="_region.gif"; /** * 获取扩散数据 @@ -679,24 +693,182 @@ public class TransportResultDataServiceImpl implements TransportResultDataServic * 获取gif地址 * @param taskId * @param speciesId + * @param genGifType * @return */ @Override - public String getTaskGifAddr(Integer taskId,Integer speciesId) { + public String getTaskGifAddr(Integer taskId,Integer speciesId,String genGifType) { TransportTask transportTask = this.transportTaskMapper.selectById(taskId); //nginx代理父路径就是/gif String parentPath = "/gif"; if(TransportTaskModeEnum.FORWARD.getKey().equals(transportTask.getTaskMode())){ - return this.getForwardTaskGIFPath(transportTask,speciesId,parentPath); + return this.getForwardTaskGIFProxyPath(transportTask,speciesId,parentPath,genGifType); }else if(TransportTaskModeEnum.BACK_FORWARD.getKey().equals(transportTask.getTaskMode())){ LambdaQueryWrapper queryWrapper = new LambdaQueryWrapper<>(); queryWrapper.eq(TransportTaskBackwardChild::getTaskId,taskId); List taskBackwardChildren = this.backwardChildMapper.selectList(queryWrapper); - return this.getBackForwardTaskGIFPath(transportTask,taskBackwardChildren.get(0).getStationCode(),parentPath); + return this.getBackForwardTaskGIFProxyPath(transportTask,taskBackwardChildren.get(0).getStationCode(),parentPath,genGifType); } return StrUtil.EMPTY; } + /** + * 生成gif + * + * @param taskId + * @param speciesId + * @param lonmin + * @param latmin + * @param lonmax + * @param latmax + */ + @Override + public void genGif(Integer taskId, Integer speciesId, Integer lonmin, Integer latmin, Integer lonmax, Integer latmax) { + Thread genThread = new Thread(()->{ + LambdaQueryWrapper gifLogQueryWrapper = new LambdaQueryWrapper<>(); + gifLogQueryWrapper.eq(TransportGenGifLog::getTaskId,taskId); + gifLogQueryWrapper.eq(Objects.nonNull(speciesId),TransportGenGifLog::getSpeciesId,speciesId); + genGifLogMapper.delete(gifLogQueryWrapper); + + TransportTask transportTask = this.transportTaskMapper.selectById(taskId); + + int mode = 0; + String ncPath = null; + if(TransportTaskModeEnum.FORWARD.getKey().equals(transportTask.getTaskMode())){ + mode = TransportTaskModeEnum.FORWARD.getKey(); + ncPath = this.getForwardTaskNCPath(transportTask,speciesId); + }else if(TransportTaskModeEnum.BACK_FORWARD.getKey().equals(transportTask.getTaskMode())){ + mode = TransportTaskModeEnum.BACK_FORWARD.getKey(); + LambdaQueryWrapper queryWrapper = new LambdaQueryWrapper<>(); + queryWrapper.eq(TransportTaskBackwardChild::getTaskId,taskId); + List taskBackwardChildren = this.backwardChildMapper.selectList(queryWrapper); + ncPath = this.getBackForwardTaskNCPath(transportTask,taskBackwardChildren.get(0).getStationCode()); + } + JSchRemoteRunner jschRemoteRunner = new JSchRemoteRunner(); + try{ + File ncFile = new File(ncPath); + StringBuilder command = new StringBuilder(); + command.append(simulationProperties.getPythonEnvPath()); + command.append(" -u "); + command.append(StringPool.SPACE); + command.append(simulationProperties.getGifConvertScriptPath()); + command.append(" --mode "); + command.append(mode); + command.append(" --area region "); + command.append(" --lonmin "); + command.append(lonmin); + command.append(" --latmin "); + command.append(latmin); + command.append(" --lonmax "); + command.append(lonmax); + command.append(" --latmax "); + command.append(latmax); + command.append(" --ncdir "); + command.append(ncPath); + command.append(" --outputdir "); + command.append(ncFile.getParent()); + command.append(File.separator); + command.append(ncFile.getParentFile().getName()); + command.append(REGION_GIF_SUFFIX); + + //登录ssh + jschRemoteRunner.login(this.serverProperties.getHost(),this.serverProperties.getPort(), + this.serverProperties.getUsername(),this.serverProperties.getPassword()); + this.execGenGIFCommand(taskId,speciesId,command.toString(),jschRemoteRunner.getSession()); + }catch (Exception e){ + log.error("nc文件生成gif动图失败",e); + }finally { + jschRemoteRunner.close(); + } + }); + genThread.setName("genGifThread"); + genThread.start(); + } + + /** + * 执行生成GIF图片命令，持续读取日志 + */ + protected void execGenGIFCommand(Integer taskId,Integer speciesId,String command, Session session){ + ChannelExec channel = null; + try{ + //打开一个执行通道 + channel = (ChannelExec) session.openChannel("exec"); + String fullCommand = command + " 2>&1"; + channel.setCommand(fullCommand); + // 获取脚本的标准输出流，包含错误输出流 + InputStream in = channel.getInputStream(); + BufferedReader reader = new BufferedReader(new InputStreamReader(in)); + // 连接通道 + channel.connect(); + + String line; + while ((line = reader.readLine()) != null) { + if(StrUtil.isNotBlank(line)){ + TransportGenGifLog genGifLog = new TransportGenGifLog(); + genGifLog.setTaskId(taskId); + genGifLog.setLogContent(line); + if(Objects.nonNull(speciesId)){ + genGifLog.setSpeciesId(speciesId); + } + genGifLogMapper.insert(genGifLog); + } + } + }catch(JSchException | IOException e){ + throw new RuntimeException(e); + }finally { + if (channel != null) { + channel.disconnect(); + } + } + } + + /** + * 下载gif + * + * @param taskId + * @param speciesId + * @param genGifType + */ + @Override + public void downloadGif(HttpServletResponse response, Integer taskId, Integer speciesId, String genGifType) { + TransportTask transportTask = this.transportTaskMapper.selectById(taskId); + String localPath = ""; + if(TransportTaskModeEnum.FORWARD.getKey().equals(transportTask.getTaskMode())){ + localPath = this.getForwardTaskGIFLocalPath(transportTask,speciesId,genGifType); + }else if(TransportTaskModeEnum.BACK_FORWARD.getKey().equals(transportTask.getTaskMode())){ + LambdaQueryWrapper queryWrapper = new LambdaQueryWrapper<>(); + queryWrapper.eq(TransportTaskBackwardChild::getTaskId,taskId); + List taskBackwardChildren = this.backwardChildMapper.selectList(queryWrapper); + localPath = this.getBackForwardTaskGIFLocalPath(transportTask,taskBackwardChildren.get(0).getStationCode(),genGifType); + } + try { + File gifFile = new File(localPath); + if(!gifFile.exists()){ + throw new RuntimeException("文件不存在，需提前生成GIF文件"); + } + InputStream inputStream = new FileInputStream(gifFile); + DownloadUtils.download(response,inputStream,gifFile.getName(), MediaType.IMAGE_GIF_VALUE); + } catch (FileNotFoundException e) { + throw new RuntimeException("下载gif文件出现错误",e); + } + } + + /** + * 获取生成gif日志 + * + * @param taskId + * @param speciesId + */ + @Override + public List getGenGiflog(Integer taskId, Integer speciesId) { + LambdaQueryWrapper gifLogQueryWrapper = new LambdaQueryWrapper<>(); + gifLogQueryWrapper.eq(TransportGenGifLog::getTaskId,taskId); + gifLogQueryWrapper.eq(Objects.nonNull(speciesId),TransportGenGifLog::getSpeciesId,speciesId); + gifLogQueryWrapper.select(TransportGenGifLog::getCreateTime,TransportGenGifLog::getLogContent); + gifLogQueryWrapper.orderByAsc(TransportGenGifLog::getCreateTime); + return genGifLogMapper.selectList(gifLogQueryWrapper); + } + /** * 获取正演NC结果文件路径 * @param transportTask @@ -723,7 +895,7 @@ public class TransportResultDataServiceImpl implements TransportResultDataServic * @param speciesId * @return */ - private String getForwardTaskGIFPath(TransportTask transportTask,Integer speciesId,String parentPath){ + private String getForwardTaskGIFProxyPath(TransportTask transportTask,Integer speciesId,String parentPath,String genGifType){ //拼接GIF文件路径 StringBuilder path = new StringBuilder(); path.append(parentPath); @@ -732,9 +904,40 @@ public class TransportResultDataServiceImpl implements TransportResultDataServic path.append(File.separator); path.append(FORWARD); path.append(File.separator); - path.append(speciesId.toString()); + path.append(speciesId); path.append(File.separator); - path.append(speciesId+".gif"); + path.append(speciesId); + if(TransportTaskGenGifTypeEnum.GLOBAL.getValue().equals(genGifType)){ + path.append(GLOBAL_GIF_SUFFIX); + } else if (TransportTaskGenGifTypeEnum.REGION.getValue().equals(genGifType)) { + path.append(REGION_GIF_SUFFIX); + } + return path.toString(); + } + + /** + * 获取正演GIF文件路径 + * @param transportTask + * @param speciesId + * @return + */ + private String getForwardTaskGIFLocalPath(TransportTask transportTask,Integer speciesId,String genGifType){ + //拼接nc文件路径 + StringBuilder path = new StringBuilder(); + path.append(simulationProperties.getOutputPath()); + path.append(File.separator); + path.append(transportTask.getTaskName()); + path.append(File.separator); + path.append(FORWARD); + path.append(File.separator); + path.append(speciesId); + path.append(File.separator); + path.append(speciesId); + if(TransportTaskGenGifTypeEnum.GLOBAL.getValue().equals(genGifType)){ + path.append(GLOBAL_GIF_SUFFIX); + } else if (TransportTaskGenGifTypeEnum.REGION.getValue().equals(genGifType)) { + path.append(REGION_GIF_SUFFIX); + } return path.toString(); } @@ -764,7 +967,7 @@ public class TransportResultDataServiceImpl implements TransportResultDataServic * @param stationCode * @return */ - private String getBackForwardTaskGIFPath(TransportTask transportTask,String stationCode,String parentPath){ + private String getBackForwardTaskGIFProxyPath(TransportTask transportTask,String stationCode,String parentPath,String genGifType){ //拼接gif文件路径 StringBuilder path = new StringBuilder(); path.append(parentPath); @@ -775,7 +978,38 @@ public class TransportResultDataServiceImpl implements TransportResultDataServic path.append(File.separator); path.append(stationCode); path.append(File.separator); - path.append(stationCode+".gif"); + path.append(stationCode); + if(TransportTaskGenGifTypeEnum.GLOBAL.getValue().equals(genGifType)){ + path.append(GLOBAL_GIF_SUFFIX); + } else if (TransportTaskGenGifTypeEnum.REGION.getValue().equals(genGifType)) { + path.append(REGION_GIF_SUFFIX); + } + return path.toString(); + } + + /** + * 获取反演GIF文件路径 + * @param transportTask + * @param stationCode + * @return + */ + private String getBackForwardTaskGIFLocalPath(TransportTask transportTask,String stationCode,String genGifType){ + //拼接nc文件路径 + StringBuilder path = new StringBuilder(); + path.append(simulationProperties.getOutputPath()); + path.append(File.separator); + path.append(transportTask.getTaskName()); + path.append(File.separator); + path.append(BACK_FORWARD); + path.append(File.separator); + path.append(stationCode); + path.append(File.separator); + path.append(stationCode); + if(TransportTaskGenGifTypeEnum.GLOBAL.getValue().equals(genGifType)){ + path.append(GLOBAL_GIF_SUFFIX); + } else if (TransportTaskGenGifTypeEnum.REGION.getValue().equals(genGifType)) { + path.append(REGION_GIF_SUFFIX); + } return path.toString(); } diff --git a/jeecg-module-transport/src/main/java/org/jeecg/service/impl/TransportTaskServiceImpl.java b/jeecg-module-transport/src/main/java/org/jeecg/service/impl/TransportTaskServiceImpl.java index f3d2f86..3d7025a 100644 --- a/jeecg-module-transport/src/main/java/org/jeecg/service/impl/TransportTaskServiceImpl.java +++ b/jeecg-module-transport/src/main/java/org/jeecg/service/impl/TransportTaskServiceImpl.java @@ -1127,7 +1127,7 @@ public class TransportTaskServiceImpl extends ServiceImpl page(PageRequest pageRequest, String fileName,String fileExt, String dataSource, + public Result page(PageRequest pageRequest, String fileName,String batchTime, String dataSource, @DateTimeFormat(iso = DateTimeFormat.ISO.DATE) LocalDate startDate, @DateTimeFormat(iso = DateTimeFormat.ISO.DATE) LocalDate endDate) { - IPage page = weatherDataService.page(pageRequest,fileName,fileExt,dataSource,startDate,endDate); + IPage page = weatherDataService.page(pageRequest,fileName,batchTime,dataSource,startDate,endDate); Map rspData = new HashMap<>(); rspData.put("rows", page.getRecords()); rspData.put("total", page.getTotal()); @@ -62,7 +61,8 @@ public class WeatherDataController { List timeBatchList = weatherDataService .list(new LambdaQueryWrapper() .select(WeatherData::getTimeBatch) // 只查询需要的字段 - .eq(WeatherData::getDataSource,dataType)) + .eq(WeatherData::getDataSource,dataType) + .orderByAsc(WeatherData::getCreateTime)) .stream() .map(WeatherData::getTimeBatch) .distinct() @@ -81,8 +81,24 @@ public class WeatherDataController { Integer weatherType, String timeBatch, @DateTimeFormat(pattern = "yyyy-MM-dd HH:mm:ss") LocalDateTime startTime, + @DateTimeFormat(pattern = "yyyy-MM-dd HH:mm:ss") LocalDateTime endTime, int hour) { - return Result.OK(weatherDataService.getWeatherData(dataType, weatherType, timeBatch, startTime, hour)); + return Result.OK(weatherDataService.getWeatherData(dataType, weatherType, timeBatch, startTime,endTime, hour)); + } + + /** + * 缓存气象数据 + * @return + */ + @AutoLog(value = "缓存气象数据") + @Operation(summary = "缓存气象数据") + @GetMapping(value = "cacheWeatherData") + public Result cacheWeatherData(Integer dataType, + String timeBatch, + @DateTimeFormat(pattern = "yyyy-MM-dd HH:mm:ss") LocalDateTime startTime, + @DateTimeFormat(pattern = "yyyy-MM-dd HH:mm:ss") LocalDateTime endTime) { + weatherDataService.cacheWeatherData(dataType, timeBatch, startTime,endTime); + return Result.OK(); } /** diff --git a/jeecg-module-weather/src/main/java/org/jeecg/job/CacheWeatherDataJob.java b/jeecg-module-weather/src/main/java/org/jeecg/job/CacheWeatherDataJob.java new file mode 100644 index 0000000..fa0855d --- /dev/null +++ b/jeecg-module-weather/src/main/java/org/jeecg/job/CacheWeatherDataJob.java @@ -0,0 +1,97 @@ +package org.jeecg.job; + +import lombok.extern.slf4j.Slf4j; +import org.jeecg.common.constant.CommonConstant; +import org.jeecg.common.constant.enums.WeatherDataSourceEnum; +import org.jeecg.common.constant.enums.WeatherTypeEnum; +import org.jeecg.common.util.RedisUtil; +import org.jeecg.service.WeatherDataService; +import org.jeecg.vo.WeatherResultVO; + +import java.time.LocalDateTime; +import java.time.format.DateTimeFormatter; +import java.util.Objects; + +/** + * 缓存天气数据 + */ +@Slf4j +public class CacheWeatherDataJob extends Thread { + + private WeatherDataService weatherDataService; + private RedisUtil redisUtil; + private Integer dataType; + private String timeBatch; + private LocalDateTime startTime; + private LocalDateTime endTime; + + + /** + * 初始化 + * @param weatherDataService + * @param redisUtil + * @param dataType + * @param timeBatch + * @param startTime + * @param endTime + */ + public void init(WeatherDataService weatherDataService, + RedisUtil redisUtil, + Integer dataType, + String timeBatch, + LocalDateTime startTime, + LocalDateTime endTime){ + this.weatherDataService = weatherDataService; + this.redisUtil = redisUtil; + this.dataType = dataType; + this.timeBatch = timeBatch; + this.startTime = startTime; + this.endTime = endTime; + } + + @Override + public void run() { + while (!startTime.isAfter(endTime)) { + try { + WeatherTypeEnum[] weatherTypes = WeatherTypeEnum.values(); + for (WeatherTypeEnum weatherTypeEnum : weatherTypes){ + WeatherResultVO weatherResultVO= null; + WeatherDataSourceEnum dataSourceEnum = null; + if (WeatherDataSourceEnum.PANGU.getKey().equals(dataType)) { + dataSourceEnum = WeatherDataSourceEnum.PANGU; + weatherResultVO = weatherDataService.processWeatherData(weatherTypeEnum.getKey(), null, startTime, WeatherDataSourceEnum.PANGU); + }else if (WeatherDataSourceEnum.GRAPHCAST.getKey().equals(dataType)){ + dataSourceEnum = WeatherDataSourceEnum.GRAPHCAST; + weatherResultVO = weatherDataService.processWeatherData(weatherTypeEnum.getKey(), timeBatch, startTime, WeatherDataSourceEnum.GRAPHCAST); + } else if (WeatherDataSourceEnum.CRA40.getKey().equals(dataType)){ + dataSourceEnum = WeatherDataSourceEnum.CRA40; + weatherResultVO = weatherDataService.processWeatherData(weatherTypeEnum.getKey(), null, startTime, WeatherDataSourceEnum.CRA40); + } else if (WeatherDataSourceEnum.NCEP.getKey().equals(dataType)){ + dataSourceEnum = WeatherDataSourceEnum.NCEP; + weatherResultVO = weatherDataService.processWeatherData(weatherTypeEnum.getKey(), null, startTime, WeatherDataSourceEnum.NCEP); + } else if (WeatherDataSourceEnum.FNL.getKey().equals(dataType)){ + dataSourceEnum = WeatherDataSourceEnum.FNL; + weatherResultVO = weatherDataService.processWeatherData(weatherTypeEnum.getKey(),null, startTime, WeatherDataSourceEnum.FNL); + } else if (WeatherDataSourceEnum.T1H.getKey().equals(dataType)){ + dataSourceEnum = WeatherDataSourceEnum.T1H; + weatherResultVO = weatherDataService.processWeatherData(weatherTypeEnum.getKey(), timeBatch, startTime, WeatherDataSourceEnum.T1H); + }else if (WeatherDataSourceEnum.GFS.getKey().equals(dataType)){ + dataSourceEnum = WeatherDataSourceEnum.GFS; + weatherResultVO = weatherDataService.processWeatherData(weatherTypeEnum.getKey(), timeBatch, startTime, WeatherDataSourceEnum.GFS); + } + if(Objects.nonNull(weatherResultVO)){ + String key = String.format(CommonConstant.WEATHER_DATA_CACHE, dataSourceEnum.getValue(), weatherTypeEnum.getValue()); + String item = startTime.format(DateTimeFormatter.ofPattern("yyyyMMddHH")); + if (!redisUtil.hHasKey(key,item)) { + //数据缓存1天过期 + redisUtil.hset(key,item,weatherResultVO,3600*24); + } + } + } + startTime = startTime.plusHours(6); + } catch (Exception e) { + log.error("处理天气数据失败", e); + } + } + } +} diff --git a/jeecg-module-weather/src/main/java/org/jeecg/job/CleanDataJob.java b/jeecg-module-weather/src/main/java/org/jeecg/job/CleanDataJob.java index 6407927..2c0703a 100644 --- a/jeecg-module-weather/src/main/java/org/jeecg/job/CleanDataJob.java +++ b/jeecg-module-weather/src/main/java/org/jeecg/job/CleanDataJob.java @@ -1,11 +1,14 @@ package org.jeecg.job; +import cn.hutool.core.collection.CollUtil; import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; import com.baomidou.mybatisplus.core.toolkit.StringPool; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; +import org.jeecg.modules.base.entity.WeatherData; import org.jeecg.modules.base.entity.WeatherDownGFSDataLog; import org.jeecg.modules.base.entity.WeatherDownT1hDataLog; +import org.jeecg.modules.base.mapper.WeatherDataMapper; import org.jeecg.modules.base.mapper.WeatherDownGFSDataLogMapper; import org.jeecg.modules.base.mapper.WeatherDownT1hDataLogMapper; import org.jetbrains.annotations.NotNull; @@ -22,7 +25,9 @@ import java.time.LocalDate; import java.time.LocalDateTime; import java.time.format.DateTimeFormatter; import java.util.ArrayList; +import java.util.HashMap; import java.util.List; +import java.util.Map; import java.util.stream.Stream; /** @@ -47,6 +52,7 @@ public class CleanDataJob { private String t1hDataPath; private final WeatherDownGFSDataLogMapper downGFSDataLogMapper; private final WeatherDownT1hDataLogMapper downT1hDataLogMapper; + private final WeatherDataMapper weatherDataMapper; private boolean t1hIsFirstRun = true; private boolean gfsIsFirstRun = true; @@ -77,7 +83,7 @@ public class CleanDataJob { */ private void cleanT1hData(){ DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyyMMdd"); - List targetDirs = new ArrayList<>(); + Map targetDirsMap = new HashMap<>(); if(t1hIsFirstRun){ Path t1hDataDir = Paths.get(this.t1hDataPath); try (Stream stream = Files.walk(t1hDataDir)){ @@ -89,7 +95,7 @@ public class CleanDataJob { String name = dir.toFile().getName(); LocalDate dirDate = LocalDate.parse(name, formatter); if (!dirDate.isAfter(retentionDate)) { - targetDirs.add(dir); + targetDirsMap.put(name,dir); } } } catch (IOException e) { @@ -99,27 +105,33 @@ public class CleanDataJob { LocalDate localDate = LocalDate.now().minusDays(t1hDataRetentionPeriod); String batchTime = formatter.format(localDate); String targetDir = this.t1hDataPath+File.separator+batchTime; - targetDirs.add(Paths.get(targetDir)); + targetDirsMap.put(batchTime,Paths.get(targetDir)); } try { - for (Path targetDirPath : targetDirs){ - if(!Files.exists(targetDirPath)){ - break; + if(CollUtil.isNotEmpty(targetDirsMap)){ + for (Path targetDirPath : targetDirsMap.values()){ + if(!Files.exists(targetDirPath)){ + break; + } + Files.walkFileTree(targetDirPath,new SimpleFileVisitor<>() { + @NotNull + @Override + public FileVisitResult visitFile(@NotNull Path file, @NotNull BasicFileAttributes attrs) throws IOException { + Files.delete(file); + return FileVisitResult.CONTINUE; + } + @NotNull + @Override + public FileVisitResult postVisitDirectory(@NotNull Path dir, @Nullable IOException exc) throws IOException { + Files.delete(dir); + return FileVisitResult.CONTINUE; + } + }); } - Files.walkFileTree(targetDirPath,new SimpleFileVisitor<>() { - @NotNull - @Override - public FileVisitResult visitFile(@NotNull Path file, @NotNull BasicFileAttributes attrs) throws IOException { - Files.delete(file); - return FileVisitResult.CONTINUE; - } - @NotNull - @Override - public FileVisitResult postVisitDirectory(@NotNull Path dir, @Nullable IOException exc) throws IOException { - Files.delete(dir); - return FileVisitResult.CONTINUE; - } - }); + //删除数据表数据 + LambdaQueryWrapper queryWrapper = new LambdaQueryWrapper<>(); + queryWrapper.in(WeatherData::getTimeBatch,targetDirsMap.keySet()); + weatherDataMapper.delete(queryWrapper); } this.cleanT1hDataLog(); log.info("完成{}目录清除超出保留期限气象数据",t1hDataPath); @@ -133,7 +145,7 @@ public class CleanDataJob { */ private void cleanGFSData(){ DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyyMMdd"); - List targetDirs = new ArrayList<>(); + Map targetDirsMap = new HashMap<>(); if(gfsIsFirstRun){ Path gfsDataDir = Paths.get(this.gfsDataPath); try (Stream stream = Files.walk(gfsDataDir)){ @@ -147,7 +159,7 @@ public class CleanDataJob { String dateStr = name.substring(DIRPREFIX.length(),name.lastIndexOf(StringPool.UNDERSCORE)); LocalDate dirDate = LocalDate.parse(dateStr, formatter); if (!dirDate.isAfter(retentionDate)) { - targetDirs.add(dir); + targetDirsMap.put(dateStr,dir); } } } @@ -158,27 +170,34 @@ public class CleanDataJob { LocalDate localDate = LocalDate.now().minusDays(gfsDataRetentionPeriod); String batchTime = formatter.format(localDate); String targetDir = this.gfsDataPath+ File.separator+DIRPREFIX+batchTime+StringPool.UNDERSCORE+HOURS; - targetDirs.add(Paths.get(targetDir)); + targetDirsMap.put(batchTime,Paths.get(targetDir)); } try { - for (Path targetDirPath : targetDirs){ - if(!Files.exists(targetDirPath)){ - break; + if(CollUtil.isNotEmpty(targetDirsMap)){ + //删除文件 + for (Path targetDirPath : targetDirsMap.values()){ + if(!Files.exists(targetDirPath)){ + break; + } + Files.walkFileTree(targetDirPath,new SimpleFileVisitor<>() { + @NotNull + @Override + public FileVisitResult visitFile(@NotNull Path file, @NotNull BasicFileAttributes attrs) throws IOException { + Files.delete(file); + return FileVisitResult.CONTINUE; + } + @NotNull + @Override + public FileVisitResult postVisitDirectory(@NotNull Path dir, @Nullable IOException exc) throws IOException { + Files.delete(dir); + return FileVisitResult.CONTINUE; + } + }); } - Files.walkFileTree(targetDirPath,new SimpleFileVisitor<>() { - @NotNull - @Override - public FileVisitResult visitFile(@NotNull Path file, @NotNull BasicFileAttributes attrs) throws IOException { - Files.delete(file); - return FileVisitResult.CONTINUE; - } - @NotNull - @Override - public FileVisitResult postVisitDirectory(@NotNull Path dir, @Nullable IOException exc) throws IOException { - Files.delete(dir); - return FileVisitResult.CONTINUE; - } - }); + //删除数据表数据 + LambdaQueryWrapper queryWrapper = new LambdaQueryWrapper<>(); + queryWrapper.in(WeatherData::getTimeBatch,targetDirsMap.keySet()); + weatherDataMapper.delete(queryWrapper); } this.cleanGFSDataLog(); log.info("完成{}目录清除超出保留期限气象数据",gfsDataPath); diff --git a/jeecg-module-weather/src/main/java/org/jeecg/job/DownloadGFSJob.java b/jeecg-module-weather/src/main/java/org/jeecg/job/DownloadGFSJob.java index b0071ea..4043d00 100644 --- a/jeecg-module-weather/src/main/java/org/jeecg/job/DownloadGFSJob.java +++ b/jeecg-module-weather/src/main/java/org/jeecg/job/DownloadGFSJob.java @@ -1,7 +1,10 @@ package org.jeecg.job; import cn.hutool.core.collection.CollUtil; +import cn.hutool.core.io.FileUtil; +import cn.hutool.core.util.ArrayUtil; import cn.hutool.core.util.StrUtil; +import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; import com.baomidou.mybatisplus.core.toolkit.StringPool; import com.jcraft.jsch.ChannelExec; import com.jcraft.jsch.JSchException; @@ -9,7 +12,6 @@ import com.jcraft.jsch.Session; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.apache.commons.codec.digest.DigestUtils; -import org.aspectj.util.FileUtil; import org.jeecg.common.constant.enums.WeatherDataSourceEnum; import org.jeecg.common.properties.ServerProperties; import org.jeecg.common.util.JSchRemoteRunner; @@ -24,6 +26,7 @@ import java.io.*; import java.time.LocalDate; import java.time.LocalDateTime; import java.time.format.DateTimeFormatter; +import java.util.ArrayList; import java.util.HashSet; import java.util.List; import java.util.Set; @@ -49,14 +52,18 @@ public class DownloadGFSJob { private final ServerProperties serverProperties; private final WeatherDataMapper weatherDataMapper; + private static final int AUTO_DOWN_TYPE = 1; + private static final int MANUAL_DOWN_TYPE = 2; + private final static String DATA_FILE_PREFIX = "gfs.t00z.pgrb2.0p25."; private final static String HOURS = "00"; private final static String DIRPREFIX = "GFS_"; + //下载失败重试次数 private int retryCount = 0; - @Scheduled(cron = "${gfs-download.cron}") - public void downloadT1HFile() { + @Scheduled(cron = "${gfs-download.downCron}") + public void downloadGFSFile() { //开关为true才执行下载任务 if(downloadSwitch){ retryCount = 0; @@ -69,26 +76,109 @@ public class DownloadGFSJob { //获取批次时间和下载命令 DateTimeFormatter dateTimeFormatter = DateTimeFormatter.ofPattern("yyyyMMdd"); String batchTime = dateTimeFormatter.format(LocalDate.now()); - this.execDownload(jschRemoteRunner,batchTime); + this.execDownload(jschRemoteRunner,batchTime,AUTO_DOWN_TYPE); }catch (Exception e) { log.error("GFS预报数据下载出现错误",e); } } } + /** + * 检查gfsDataPath目录下所有批次gfs气象数据，缺失则下载 + */ + @Scheduled(cron = "${gfs-download.checkCron}") + public void checkAllData(){ + //删除手动检查历史日志，手动检查的只保留一次 + cleanManualDownTypeLogs(); + File gfsDataDir = new File(this.gfsDataPath); + if (gfsDataDir.exists() && gfsDataDir.isDirectory()){ + File[] dirs = gfsDataDir.listFiles(file -> file.isDirectory() && file.getName().startsWith(DIRPREFIX)); + List batchTimes = new ArrayList<>(); + if (ArrayUtil.isEmpty(dirs)) { + String log = "gfs数据存储目录为空，检查完毕。"; + this.saveLog(log,"-1",MANUAL_DOWN_TYPE); + return; + } + for (File dirFile : dirs){ + String batchTime = dirFile.getName().split(StringPool.UNDERSCORE)[1]; + LocalDate batchTimeParseVal = LocalDate.parse(batchTime, DateTimeFormatter.ofPattern("yyyyMMdd")); + //当天批次数据不检查 + if(LocalDate.now().isEqual(batchTimeParseVal)){ + continue; + } + if(FileUtil.isDirEmpty(dirFile)){ + batchTimes.add(batchTime); + + String log = batchTime+"批次数据存储目录为空。"; + this.saveLog(log,batchTime,MANUAL_DOWN_TYPE); + + }else { + String gfsBatchDataPath = this.getGfsDataPath(batchTime); + List files = FileUtil.loopFiles(new File(gfsBatchDataPath), file -> file.getName().startsWith(DATA_FILE_PREFIX)); + //根据业务规则生成预期的文件后缀集合 (f000 ~ f384) + Set expectedSuffixes = new HashSet<>(); + // 前 120 步进为 1 + for (int i = 0; i <= 120; i++) { + expectedSuffixes.add(String.format("f%03d", i)); + } + // 120 之后步进为 3 + for (int i = 123; i <= 384; i += 3) { + expectedSuffixes.add(String.format("f%03d", i)); + } + //获取实际后缀 + Set actualSuffixes = new HashSet<>(); + for (File file : files){ + actualSuffixes.add(file.getName().substring(DATA_FILE_PREFIX.length())); + } + // 比对预期与实际文件，找出缺失的文件 + boolean flag = false; + for (String expected : expectedSuffixes) { + if (!actualSuffixes.contains(expected)) { + flag = true; + break; + } + } + //如果flag = true;说明此目录气象数据有缺失，需补充 + if (flag){ + batchTimes.add(batchTime); + + String log = batchTime+"批次数据下载有缺失。"; + this.saveLog(log,batchTime,MANUAL_DOWN_TYPE); + } + } + } + if(CollUtil.isNotEmpty(batchTimes)){ + JSchRemoteRunner jschRemoteRunner = new JSchRemoteRunner(); + //登录ssh + try { + jschRemoteRunner.login(this.serverProperties.getHost(),this.serverProperties.getPort(), + this.serverProperties.getUsername(),this.serverProperties.getPassword()); + } catch (JSchException e) { + throw new RuntimeException(e); + } + for (String batchTime : batchTimes){ + this.execDownload(jschRemoteRunner,batchTime,MANUAL_DOWN_TYPE); + } + }else { + String log = "数据检查完成，各批次数据完好。"; + this.saveLog(log,"-1",MANUAL_DOWN_TYPE); + } + } + } + /** * 执行下载 * @param jschRemoteRunner */ - private void execDownload(JSchRemoteRunner jschRemoteRunner,String batchTime){ + private void execDownload(JSchRemoteRunner jschRemoteRunner,String batchTime,int logType){ //获取批次时间和下载命令 String downloadCommand = this.getDownloadCommand(batchTime); //执行下载命令 log.info("开始下载"); log.info("下载命令为{}",downloadCommand); - this.execCommand(batchTime,downloadCommand,jschRemoteRunner.getSession()); + this.execCommand(batchTime,downloadCommand,jschRemoteRunner.getSession(),logType); //检查结果数据 - this.checkData(jschRemoteRunner,batchTime); + this.checkData(jschRemoteRunner,batchTime,logType); } /** @@ -96,10 +186,10 @@ public class DownloadGFSJob { * @param jschRemoteRunner * @param batchTime */ - private void checkData(JSchRemoteRunner jschRemoteRunner,String batchTime){ + private void checkData(JSchRemoteRunner jschRemoteRunner,String batchTime,int logType){ log.info("开始检查数据"); String gfsBatchDataPath = this.getGfsDataPath(batchTime); - List files = List.of(FileUtil.listFiles(new File(gfsBatchDataPath), file -> file.getName().startsWith(DATA_FILE_PREFIX))); + List files = FileUtil.loopFiles(new File(gfsBatchDataPath), file -> file.getName().startsWith(DATA_FILE_PREFIX)); try { if(CollUtil.isEmpty(files)){ throw new RuntimeException(gfsBatchDataPath+"目录没有数据，下载可能出现错误"); @@ -124,7 +214,7 @@ public class DownloadGFSJob { for (String expected : expectedSuffixes) { if (!actualSuffixes.contains(expected)) { String logContent = "缺失"+DATA_FILE_PREFIX+expected+"气象文件"; - this.saveLog(batchTime,logContent); + this.saveLog(batchTime,logContent,logType); flag = true; } } @@ -132,9 +222,9 @@ public class DownloadGFSJob { throw new RuntimeException(gfsBatchDataPath+"目录批次数据存在缺失"); } //保存数据入库 - this.saveDataToDataBase(batchTime); + this.saveDataToDataBase(batchTime,logType); }catch (Exception e) { - this.saveLog(e.getMessage(),batchTime); + this.saveLog(e.getMessage(),batchTime,logType); log.error(e.getMessage(),e); try { //间隔5分钟，重试200次 @@ -142,8 +232,8 @@ public class DownloadGFSJob { retryCount++; TimeUnit.MINUTES.sleep(5); log.info("{}批次数据，重试下载{}次,最大200次",gfsBatchDataPath,retryCount); - this.saveLog(batchTime,batchTime); - this.execDownload(jschRemoteRunner,batchTime); + this.saveLog(batchTime,batchTime,logType); + this.execDownload(jschRemoteRunner,batchTime,logType); } } catch (InterruptedException ex) { throw new RuntimeException(ex); @@ -155,11 +245,11 @@ public class DownloadGFSJob { * 保存已下载的数据入库 * @param batchTime */ - private void saveDataToDataBase(String batchTime){ + private void saveDataToDataBase(String batchTime,int logType){ //本批次数据开始日期 LocalDateTime localDateTime = LocalDate.parse(batchTime,DateTimeFormatter.ofPattern("yyyyMMdd")).atTime(Integer.parseInt(HOURS),0,0); String gfsBatchDataPath = this.getGfsDataPath(batchTime); - List fileList = List.of(FileUtil.listFiles(new File(gfsBatchDataPath), file -> file.getName().startsWith(DATA_FILE_PREFIX))); + List fileList = FileUtil.loopFiles(new File(gfsBatchDataPath), file -> file.getName().startsWith(DATA_FILE_PREFIX)); if (CollUtil.isNotEmpty(fileList)) { for (File file : fileList) { try { @@ -181,7 +271,7 @@ public class DownloadGFSJob { log.error("保存{}气象数据文件出现错误，原因为：",file.getAbsolutePath(),e); String logContent = "保存{}气象数据文件出现错误，原因为："+e.getMessage(); - this.saveLog(logContent,batchTime); + this.saveLog(logContent,batchTime,logType); } } } @@ -233,7 +323,7 @@ public class DownloadGFSJob { /** * 执行下载命令持续读取日志 */ - protected void execCommand(String batchTime,String command, Session session){ + protected void execCommand(String batchTime,String command, Session session,int logType){ ChannelExec channel = null; try{ //打开一个执行通道 @@ -250,7 +340,7 @@ public class DownloadGFSJob { while ((line = reader.readLine()) != null) { if(StrUtil.isNotBlank(line)){ //保存日志 - this.saveLog(line,batchTime); + this.saveLog(line,batchTime,logType); } } }catch(JSchException | IOException e){ @@ -265,10 +355,20 @@ public class DownloadGFSJob { /** * 保存过程日志 */ - private void saveLog(String log,String batchTime){ + private void saveLog(String log,String batchTime,int logType){ WeatherDownGFSDataLog dataLog = new WeatherDownGFSDataLog(); dataLog.setLogContent(log); dataLog.setBatchTime(batchTime); + dataLog.setLogType(logType); weatherDownGFSDataLogMapper.insert(dataLog); } + + /** + * 删除手动检查日志，手动日志只保留一次 + */ + private void cleanManualDownTypeLogs(){ + LambdaQueryWrapper queryWrapper = new LambdaQueryWrapper<>(); + queryWrapper.eq(WeatherDownGFSDataLog::getLogType, MANUAL_DOWN_TYPE); + weatherDownGFSDataLogMapper.delete(queryWrapper); + } } diff --git a/jeecg-module-weather/src/main/java/org/jeecg/service/WeatherDataService.java b/jeecg-module-weather/src/main/java/org/jeecg/service/WeatherDataService.java index 8a3a2aa..21047b6 100644 --- a/jeecg-module-weather/src/main/java/org/jeecg/service/WeatherDataService.java +++ b/jeecg-module-weather/src/main/java/org/jeecg/service/WeatherDataService.java @@ -2,9 +2,9 @@ package org.jeecg.service; import com.baomidou.mybatisplus.core.metadata.IPage; import com.baomidou.mybatisplus.extension.service.IService; +import org.jeecg.common.constant.enums.WeatherDataSourceEnum; import org.jeecg.common.system.query.PageRequest; import org.jeecg.modules.base.entity.WeatherData; -import org.jeecg.modules.base.entity.WeatherLinkedDataLog; import org.jeecg.vo.*; import java.time.LocalDate; import java.time.LocalDateTime; @@ -12,7 +12,7 @@ import java.util.List; public interface WeatherDataService extends IService { - WeatherResultVO getWeatherData(Integer dataType, Integer weatherType, String timeBatch, LocalDateTime startTime, int hour); + WeatherResultVO getWeatherData(Integer dataType, Integer weatherType, String timeBatch, LocalDateTime startTime,LocalDateTime endTime, int hour); WeatherResultVO getWeatherDataPreview(Integer weatherId, Integer weatherType); WindDataLineVO getDataLine(Integer dataType, String timeBatch, LocalDateTime startTime, LocalDateTime endTime, double longitude, double latitude); List getWindRose(Integer dataType, String timeBatch, LocalDateTime startTime, LocalDateTime endTime,double longitude, double latitude); @@ -20,13 +20,13 @@ public interface WeatherDataService extends IService { /** * 分页查询气象数据 * @param pageRequest - * @param fileExt + * @param batchTime * @param dataSource * @param startDate * @param endDate * @return */ - IPage page(PageRequest pageRequest,String fileName, String fileExt, String dataSource, LocalDate startDate, LocalDate endDate); + IPage page(PageRequest pageRequest,String fileName, String batchTime, String dataSource, LocalDate startDate, LocalDate endDate); /** * 删除气象数据 @@ -41,4 +41,24 @@ public interface WeatherDataService extends IService { * @param endDate */ void linkedData(Integer dataType,LocalDate startDate, LocalDate endDate); + + + /** + * 缓存气象数据 + * @param dataType + * @param timeBatch + * @param startTime + * @param endTime + */ + void cacheWeatherData(Integer dataType, String timeBatch, LocalDateTime startTime, LocalDateTime endTime); + + /** + * 处理气象数据 + * @param weatherType + * @param timeBatch + * @param targetTime + * @param dataTypeEnum + * @return + */ + WeatherResultVO processWeatherData(Integer weatherType, String timeBatch, LocalDateTime targetTime, WeatherDataSourceEnum dataTypeEnum); } diff --git a/jeecg-module-weather/src/main/java/org/jeecg/service/impl/WeatherDataServiceImpl.java b/jeecg-module-weather/src/main/java/org/jeecg/service/impl/WeatherDataServiceImpl.java index 26428ff..674bdd5 100644 --- a/jeecg-module-weather/src/main/java/org/jeecg/service/impl/WeatherDataServiceImpl.java +++ b/jeecg-module-weather/src/main/java/org/jeecg/service/impl/WeatherDataServiceImpl.java @@ -11,16 +11,16 @@ import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.apache.commons.codec.digest.DigestUtils; import org.apache.commons.lang3.StringUtils; +import org.jeecg.common.constant.CommonConstant; import org.jeecg.common.constant.WeatherStepConstants; import org.jeecg.common.constant.enums.*; import org.jeecg.common.exception.JeecgBootException; -import org.jeecg.common.exception.JeecgFileUploadException; import org.jeecg.common.properties.SystemStorageProperties; import org.jeecg.common.system.query.PageRequest; -import org.jeecg.common.util.Grib2TimeReader; import org.jeecg.common.util.NcUtil; +import org.jeecg.common.util.RedisUtil; +import org.jeecg.job.CacheWeatherDataJob; import org.jeecg.modules.base.entity.WeatherData; -import org.jeecg.modules.base.entity.WeatherLinkedDataLog; import org.jeecg.modules.base.mapper.WeatherDataMapper; import org.jeecg.service.WeatherDataService; import org.jeecg.service.WeatherLinkedDataLogService; @@ -49,6 +49,7 @@ public class WeatherDataServiceImpl extends ServiceImpl page(PageRequest pageRequest,String fileName, String fileExt, String dataSource, LocalDate startDate, LocalDate endDate) { + public IPage page(PageRequest pageRequest,String fileName, String batchTime, String dataSource, LocalDate startDate, LocalDate endDate) { LocalDateTime startTime = null; - if(Objects.nonNull(startDate)){ - startTime = LocalDateTime.of(startDate.getYear(), startDate.getMonth(), startDate.getDayOfMonth(), 0, 0, 0); - } LocalDateTime endTime = null; - if(Objects.nonNull(endDate)){ + if(Objects.nonNull(startDate) && Objects.nonNull(endDate)){ + startTime = LocalDateTime.of(startDate.getYear(), startDate.getMonth(), startDate.getDayOfMonth(), 0, 0, 0); endTime = LocalDateTime.of(endDate.getYear(), endDate.getMonth(), endDate.getDayOfMonth(), 23, 59, 59); } LambdaQueryWrapper queryWrapper = new LambdaQueryWrapper<>(); List dataSources = null; if (StringUtils.isNotBlank(dataSource)) { dataSources = Arrays.stream(dataSource.split(",")).map(Integer::parseInt).collect(Collectors.toList()); - }else { - dataSources = new ArrayList<>(); - dataSources.add(Integer.parseInt(fileExt)); } queryWrapper.in(CollUtil.isNotEmpty(dataSources),WeatherData::getDataSource,dataSources); queryWrapper.between((Objects.nonNull(startTime) && Objects.nonNull(endTime)),WeatherData::getDataStartTime,startTime,endTime); - queryWrapper.eq(StringUtils.isNotBlank(fileExt),WeatherData::getFileExt, fileExt); + queryWrapper.eq(StringUtils.isNotBlank(batchTime),WeatherData::getTimeBatch, batchTime); queryWrapper.like(StringUtils.isNotBlank(fileName),WeatherData::getFileName, fileName); queryWrapper.select(WeatherData::getId,WeatherData::getFileName,WeatherData::getDataSource, WeatherData::getFileExt,WeatherData::getDataStartTime,WeatherData::getFilePath,WeatherData::getTimeBatch); @@ -368,89 +353,151 @@ public class WeatherDataServiceImpl extends ServiceImpl fileList = FileUtil.loopFiles(dir, new FileFilter() { - @Override - public boolean accept(File file) { - if(StrUtil.endWith(file.getName(), WeatherFileSuffixEnum.GRIB.getValue()) - || StrUtil.endWith(file.getName(), WeatherFileSuffixEnum.GRB2.getValue()) - || StrUtil.endWith(file.getName(), WeatherFileSuffixEnum.GRIB2.getValue())){ - try{ - LocalDate localDate = null; + Thread linkedDataThread = new Thread(() -> { + //先清空表 + weatherLinkedDataLogService.cleanTable(); + String linkedAddr = ""; + if(WeatherDataSourceEnum.NCEP.getKey().equals(dataType)) { + linkedAddr = systemStorageProperties.getNcepDataPath(); + }else if(WeatherDataSourceEnum.FNL.getKey().equals(dataType)) { + linkedAddr = systemStorageProperties.getFnlDataPath(); + }else if(WeatherDataSourceEnum.CRA40.getKey().equals(dataType)) { + linkedAddr = systemStorageProperties.getCra40DataPath(); + } + File dir = new File(linkedAddr); + if (!dir.exists()) { + FileUtil.mkdir(dir); + } + List fileList = FileUtil.loopFiles(dir, new FileFilter() { + @Override + public boolean accept(File file) { + if(StrUtil.endWith(file.getName(), WeatherFileSuffixEnum.GRIB.getValue()) + || StrUtil.endWith(file.getName(), WeatherFileSuffixEnum.GRB2.getValue()) + || StrUtil.endWith(file.getName(), WeatherFileSuffixEnum.GRIB2.getValue())){ + try{ + LocalDate localDate = null; + if(WeatherDataSourceEnum.NCEP.getKey().equals(dataType)) { + String dateTimeStr = file.getName().substring(0, file.getName().indexOf(".")); + DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyyMMddHHmm"); + LocalDateTime dateTime = LocalDateTime.parse(dateTimeStr, formatter); + localDate = dateTime.toLocalDate(); + }else if(WeatherDataSourceEnum.FNL.getKey().equals(dataType)){ + String dateTimeStr = file.getName().split("_")[1]; + DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyyMMdd"); + localDate = LocalDate.parse(dateTimeStr, formatter); + }else if(WeatherDataSourceEnum.CRA40.getKey().equals(dataType)){ + String dateTimeStr = file.getName().split("_")[1]; + DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyyMMdd"); + localDate = LocalDate.parse(dateTimeStr, formatter); + } + if(Objects.nonNull(localDate) && !localDate.isBefore(startDate) && !localDate.isAfter(endDate)) { + return true; + } + }catch(Exception e){ + String msg = file.getAbsolutePath()+"文件命名格式错误，需确认"; + log.error(msg,e); + weatherLinkedDataLogService.create(dataType,msg); + } + } + return false; + } + }); + if (CollUtil.isNotEmpty(fileList)) { + for (File file : fileList) { + try { + //获取文件数据开始日期 + LocalDateTime dateTime = null; if(WeatherDataSourceEnum.NCEP.getKey().equals(dataType)) { String dateTimeStr = file.getName().substring(0, file.getName().indexOf(".")); DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyyMMddHHmm"); - LocalDateTime dateTime = LocalDateTime.parse(dateTimeStr, formatter); - localDate = dateTime.toLocalDate(); + dateTime = LocalDateTime.parse(dateTimeStr, formatter).withSecond(0); }else if(WeatherDataSourceEnum.FNL.getKey().equals(dataType)){ - String dateTimeStr = file.getName().split("_")[1]; - DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyyMMdd"); - localDate = LocalDate.parse(dateTimeStr, formatter); + String dateTimeStr = file.getName().substring("fnl_".length(),file.getName().lastIndexOf(".")); + DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyyMMdd_HH_mm"); + dateTime = LocalDateTime.parse(dateTimeStr, formatter).withSecond(0); }else if(WeatherDataSourceEnum.CRA40.getKey().equals(dataType)){ - String dateTimeStr = file.getName().split("_")[1]; - DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyyMMdd"); - localDate = LocalDate.parse(dateTimeStr, formatter); + String dateTimeStr = file.getName().substring("CRA40_".length(),file.getName().lastIndexOf(".")); + DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyyMMdd_HH_mm"); + dateTime = LocalDateTime.parse(dateTimeStr, formatter).withSecond(0); } - if(Objects.nonNull(localDate) && !localDate.isBefore(startDate) && !localDate.isAfter(endDate)) { - return true; + LocalDate localDate = dateTime.toLocalDate(); + //再次验证grib文件的有效数据时间在给定时间内，才关联保存 + if(!localDate.isBefore(startDate) && !localDate.isAfter(endDate)) { + //如果此文件存在则无需再次新增 + String gribFileMD5 = this.getGribFileMD5(file.getAbsolutePath()); + LambdaQueryWrapper queryWrapper = new LambdaQueryWrapper<>(); + queryWrapper.eq(WeatherData::getMd5Value, gribFileMD5); + WeatherData queryResult = this.baseMapper.selectOne(queryWrapper); + if (Objects.nonNull(queryResult)) { + weatherLinkedDataLogService.create(dataType,file.getAbsolutePath()+"已存在"); + continue; + } + WeatherData weatherData = new WeatherData(); + weatherData.setFileName(file.getName()); + weatherData.setFileExt(file.getName().substring(file.getName().lastIndexOf(".")+1)); + weatherData.setDataSource(dataType); + weatherData.setFilePath(file.getAbsolutePath()); + weatherData.setDataStartTime(dateTime); + weatherData.setMd5Value(gribFileMD5); + this.baseMapper.insert(weatherData); + weatherLinkedDataLogService.create(dataType,file.getAbsolutePath()+"关联成功"); } - }catch(Exception e){ - String msg = file.getAbsolutePath()+"文件命名格式错误，需确认"; - log.error(msg,e); - weatherLinkedDataLogService.create(dataType,msg); + }catch (Exception e){ + log.error("关联{}气象数据文件出现错误，原因为：",file.getAbsolutePath(),e); + weatherLinkedDataLogService.create(dataType,file.getAbsolutePath()+"关联失败，原因为："+e.getMessage()); } } - return false; + weatherLinkedDataLogService.create(dataType,"关联任务执行完毕！"); } }); - if (CollUtil.isNotEmpty(fileList)) { - for (File file : fileList) { - try { - //获取文件数据开始日期 - LocalDateTime localDateTime = Grib2TimeReader.readValidTime(file.getAbsolutePath()); - LocalDate localDate = localDateTime.toLocalDate(); - //再次验证grib文件的有效数据时间在给定时间内，才关联保存 - if(!localDate.isBefore(startDate) && !localDate.isAfter(endDate)) { - //如果此文件存在则无需再次新增 - String gribFileMD5 = this.getGribFileMD5(file.getAbsolutePath()); - LambdaQueryWrapper queryWrapper = new LambdaQueryWrapper<>(); - queryWrapper.eq(WeatherData::getMd5Value, gribFileMD5); - WeatherData queryResult = this.baseMapper.selectOne(queryWrapper); - if (Objects.nonNull(queryResult)) { - weatherLinkedDataLogService.create(dataType,file.getAbsolutePath()+"已存在"); - continue; - } - WeatherData weatherData = new WeatherData(); - weatherData.setFileName(file.getName()); - weatherData.setFileExt(file.getName().substring(file.getName().lastIndexOf(".")+1)); - weatherData.setDataSource(dataType); - weatherData.setFilePath(file.getAbsolutePath()); - weatherData.setDataStartTime(localDateTime); - weatherData.setMd5Value(gribFileMD5); - this.baseMapper.insert(weatherData); - weatherLinkedDataLogService.create(dataType,file.getAbsolutePath()+"关联成功"); - } - }catch (Exception e){ - log.error("关联{}气象数据文件出现错误，原因为：",file.getAbsolutePath(),e); - weatherLinkedDataLogService.create(dataType,file.getAbsolutePath()+"关联失败，原因为："+e.getMessage()); - } - } - weatherLinkedDataLogService.create(dataType,"关联任务执行完毕！"); - } + linkedDataThread.setName("linkedDataThread"); + linkedDataThread.start(); + linkedDataThread.setUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() { + @Override + public void uncaughtException(Thread t, Throwable e) { + String errLog = "关联气象数据出现错误，原因为："+e.getMessage(); + log.error(errLog); + weatherLinkedDataLogService.create(dataType,errLog); + } + }); + } + /** + * 缓存气象数据 + * 后续前台播放气象数据时加快速度 + * @param dataType + * @param timeBatch + * @param startTime + * @param endTime + */ + @Override + public void cacheWeatherData(Integer dataType, String timeBatch, LocalDateTime startTime, LocalDateTime endTime) { + Objects.requireNonNull(dataType, "数据类型不能为空"); + if(WeatherDataSourceEnum.T1H.getKey().equals(dataType) || WeatherDataSourceEnum.GFS.getKey().equals(dataType)){ + if (StrUtil.isBlank(timeBatch)){ + throw new IllegalArgumentException("时间批次不能为空"); + } + startTime = LocalDate.parse(timeBatch,DateTimeFormatter.ofPattern("yyyyMMdd")).atTime(0,0,0); + //如果是GFS数据，结束时间是批次时间固定+384小时（16）天的数据 + if(WeatherDataSourceEnum.GFS.getKey().equals(dataType)){ + endTime = startTime.plusHours(384); + } + }else{ + Objects.requireNonNull(startTime, "开始时间不能为空"); + Objects.requireNonNull(endTime, "结束时间不能为空"); + } + LambdaQueryWrapper queryWrapper = new LambdaQueryWrapper<>(); + queryWrapper.eq(WeatherData::getDataSource, dataType); + queryWrapper.eq(StrUtil.isNotBlank(timeBatch),WeatherData::getTimeBatch, timeBatch); + queryWrapper.between(Objects.nonNull(startTime) && Objects.nonNull(endTime),WeatherData::getDataStartTime, startTime,endTime); + Long count = this.baseMapper.selectCount(queryWrapper); + if(count == 0){ + throw new RuntimeException("此时间范围无气象数据"); + } + CacheWeatherDataJob weatherDataJob = new CacheWeatherDataJob(); + weatherDataJob.init(this,redisUtil,dataType,timeBatch,startTime,endTime); + weatherDataJob.setName("cacheWeatherDataJob"); + weatherDataJob.start(); } /** @@ -497,6 +544,12 @@ public class WeatherDataServiceImpl extends ServiceImpl queryWrapper = new LambdaQueryWrapper<>(); queryWrapper.eq(WeatherData::getDataStartTime, targetTime).eq(WeatherData::getDataSource,dataTypeEnum.getKey()); if(StringUtils.isNotBlank(timeBatch)){ @@ -595,8 +653,7 @@ public class WeatherDataServiceImpl extends ServiceImpl getLinkedDataLog(Integer lastId) { LambdaQueryWrapper queryWrapper = new LambdaQueryWrapper<>(); queryWrapper.gt(Objects.nonNull(lastId),WeatherLinkedDataLog::getId,lastId); - queryWrapper.select(WeatherLinkedDataLog::getCreateTime,WeatherLinkedDataLog::getLogContent); + queryWrapper.select(WeatherLinkedDataLog::getId,WeatherLinkedDataLog::getCreateTime,WeatherLinkedDataLog::getLogContent); queryWrapper.orderByAsc(WeatherLinkedDataLog::getId); queryWrapper.last("LIMIT 10"); return this.baseMapper.selectList(queryWrapper);