panbaolin/SourceTermAnalysisSystem_java
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix:1.完善输运模拟功能2.完善下载gfs数据功能3.完善天气预报功能

Browse Source
This commit is contained in:
panbaolin 2026-08-03 18:12:39 +08:00
parent e477e90099
commit 919cb12c7a
36 changed files with 1207 additions and 327 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
jeecg-boot-base-core/src/main/java/org/jeecg/common/constant/CommonConstant.java
View File

@ -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";
}

26
jeecg-boot-base-core/src/main/java/org/jeecg/common/constant/enums/TransportTaskGenGifTypeEnum.java Normal file
View File

@ -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;
}
}

8
jeecg-boot-base-core/src/main/java/org/jeecg/common/constant/enums/WeatherTypeEnum.java
View File

@ -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;
}
}

45
jeecg-boot-base-core/src/main/java/org/jeecg/common/constant/enums/WeatherVariableNameEnum.java
View File

@ -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;

9
jeecg-boot-base-core/src/main/java/org/jeecg/common/properties/ServerProperties.java
View File

@ -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
*/

5
jeecg-boot-base-core/src/main/java/org/jeecg/common/properties/TransportSimulationProperties.java
View File

@ -174,4 +174,9 @@ public class TransportSimulationProperties {
* 粒子数量
*/
private Integer particleCount;
/**
* 气象数据检查最大失败天数（等待天数），超过就设置检查未通过
*/
private Integer metCheckMaxFailureDays;
}

4
jeecg-boot-base-core/src/main/java/org/jeecg/common/util/DownloadUtils.java
View File

@ -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");

50
jeecg-boot-base-core/src/main/java/org/jeecg/modules/base/entity/TransportGenGifLog.java Normal file
View File

@ -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;
}

7
jeecg-boot-base-core/src/main/java/org/jeecg/modules/base/entity/WeatherDownGFSDataLog.java
View File

@ -40,4 +40,11 @@ public class WeatherDownGFSDataLog {
*/
@TableField(value = "log_content")
private String logContent;
/**
* 日志类型
*/
@TableField(value = "log_type")
private Integer logType;
}

8
jeecg-boot-base-core/src/main/java/org/jeecg/modules/base/mapper/TransportGenGifLogMapper.java Normal file
View File

@ -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<TransportGenGifLog> {
}

2
jeecg-boot-base-core/src/main/java/org/jeecg/modules/base/mapper/xml/TransportTaskMapper.xml
View File

@ -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
<where>

26
jeecg-model-consumer/src/main/java/org/jeecg/rebuild/consumer/RebuildTaskConsumerHandler.java
View File

@ -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);
}
}
}

26
jeecg-model-consumer/src/main/java/org/jeecg/transport/consumer/TranTaskConsumerHandler.java
View File

@ -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);
}
}
}

62
jeecg-model-consumer/src/main/java/org/jeecg/transport/consumer/china/AbstractTaskMsgHandler.java
View File

@ -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);
}
}
}

8
jeecg-model-consumer/src/main/java/org/jeecg/transport/consumer/china/Server11TaskHandler.java
View File

@ -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);
}

8
jeecg-model-consumer/src/main/java/org/jeecg/transport/consumer/china/Server12TaskHandler.java
View File

@ -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);
}

8
jeecg-model-consumer/src/main/java/org/jeecg/transport/consumer/china/Server13TaskHandler.java
View File

@ -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);
}

86
jeecg-model-consumer/src/main/java/org/jeecg/transport/flexparttask/AbstractTaskExec.java
View File

@ -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<String> msgList){
if(WeatherDataSourceEnum.GFS.getKey().equals(this.transportTask.getUseMetType())){
return checkGfsMetData(msgList);
}else {
return checkOtherData(msgList);
}
}
/**
* 校验GFS数据
* @param msgList
* @return
*/
private boolean checkGfsMetData(List<String> msgList){
boolean checkFlag = true;
LocalDateTime startTime = this.transportTask.getStartTime();
LocalDateTime endTime = this.transportTask.getEndTime();
if(TransportTaskModeEnum.BACK_FORWARD.getKey().equals(this.transportTask.getTaskMode())){
//生成批次信息
List<String> 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<WeatherData> 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<WeatherData> 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<String> msgList){
String msg = "检查气象数据";
ProgressQueue.getInstance().offer(new ProgressEvent(this.transportTask.getId(),msg));
LocalDateTime startTime = this.transportTask.getStartTime();
LocalDateTime endTime = this.transportTask.getEndTime();
LambdaQueryWrapper<WeatherData> 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 {

52
jeecg-model-consumer/src/main/java/org/jeecg/transport/flexparttask/BackwardTaskExec.java
View File

@ -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();
}
}
}
}

13
jeecg-model-consumer/src/main/java/org/jeecg/transport/flexparttask/BuildNcToSrsFile.java
View File

@ -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<String> 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;
}

12
jeecg-model-consumer/src/main/java/org/jeecg/transport/flexparttask/ForwardTaskExec.java
View File

@ -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->{

26
jeecg-model-consumer/src/main/java/org/jeecg/weather/consumer/WeatherTaskConsumerHandler.java
View File

@ -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);
}
}
}

14
jeecg-model-consumer/src/main/java/org/jeecg/weather/task/WeatherForecastTaskExec.java
View File

@ -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<WeatherData> 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());

8
jeecg-module-source-rebuild/src/main/java/org/jeecg/controller/TaskResultDataController.java
View File

@ -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);
}
}

3
jeecg-module-transport/src/main/java/org/jeecg/consumer/SampleMessageConsumer.java
View File

@ -58,7 +58,7 @@ public class SampleMessageConsumer implements RocketMQListener<GardsSampleResult
LocalDateTime acqStartTime = sampleResultDTO.getAcquisitionStart().toInstant().atZone(ZoneId.systemDefault()).toLocalDateTime();
LocalDateTime acqEndTime = sampleResultDTO.getAcquisitionStop().toInstant().atZone(ZoneId.systemDefault()).toLocalDateTime();
//反演时间向前推14天作为开始时间
LocalDateTime startTime = acqEndTime.minusDays(14);
LocalDateTime startTime = acqEndTime.minusDays(5);
//设置基础参数
TransportTask transportTask = new TransportTask();
transportTask.setTaskName(taskName);
@ -72,6 +72,7 @@ public class SampleMessageConsumer implements RocketMQListener<GardsSampleResult
transportTask.setZ1(simulationProperties.getZ1());
transportTask.setZ2(simulationProperties.getZ2());
transportTask.setParticleCount(simulationProperties.getParticleCount());
transportTask.setUpdateTime(LocalDateTime.now());
//设置站点参数
List<TransportTaskBackwardChild> backwardChild = new ArrayList<>();
TransportTaskBackwardChild taskBackwardChild = new TransportTaskBackwardChild();

39
jeecg-module-transport/src/main/java/org/jeecg/controller/TransportResultDataController.java
View File

@ -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));
}
}

34
jeecg-module-transport/src/main/java/org/jeecg/service/TransportResultDataService.java
View File

@ -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<TransportGenGifLog> getGenGiflog(Integer taskId, Integer speciesId);
}

254
jeecg-module-transport/src/main/java/org/jeecg/service/impl/TransportResultDataServiceImpl.java
View File

@ -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<TransportTaskBackwardChild> queryWrapper = new LambdaQueryWrapper<>();
queryWrapper.eq(TransportTaskBackwardChild::getTaskId,taskId);
List<TransportTaskBackwardChild> 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<TransportGenGifLog> 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<TransportTaskBackwardChild> queryWrapper = new LambdaQueryWrapper<>();
queryWrapper.eq(TransportTaskBackwardChild::getTaskId,taskId);
List<TransportTaskBackwardChild> 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<TransportTaskBackwardChild> queryWrapper = new LambdaQueryWrapper<>();
queryWrapper.eq(TransportTaskBackwardChild::getTaskId,taskId);
List<TransportTaskBackwardChild> 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<TransportGenGifLog> getGenGiflog(Integer taskId, Integer speciesId) {
LambdaQueryWrapper<TransportGenGifLog> 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();
}

2
jeecg-module-transport/src/main/java/org/jeecg/service/impl/TransportTaskServiceImpl.java
View File

@ -1127,7 +1127,7 @@ public class TransportTaskServiceImpl extends ServiceImpl<TransportTaskMapper,Tr
transportTask.setStartTime(ExcelUtils.getDateValue(startTimeCell));
//设置结束时间
Cell endTimeCell = sheet.getRow(3).getCell(3);
transportTask.setEndTime(ExcelUtils.getDateValue(startTimeCell));
transportTask.setEndTime(ExcelUtils.getDateValue(endTimeCell));
//设置释放数据来源
Cell releaseDataSourceCell = sheet.getRow(3).getCell(5);
Integer releaseDataSourceCellValue = ExcelUtils.getIntValue(releaseDataSourceCell);

26
jeecg-module-weather/src/main/java/org/jeecg/controller/WeatherDataController.java
View File

@ -9,7 +9,6 @@ import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.jeecg.common.api.vo.Result;
import org.jeecg.common.aspect.annotation.AutoLog;
import org.jeecg.common.constant.enums.WeatherDataSourceEnum;
import org.jeecg.common.system.query.PageRequest;
import org.jeecg.job.DownloadT1hJob;
import org.jeecg.modules.base.entity.WeatherData;
@ -41,10 +40,10 @@ public class WeatherDataController {
@AutoLog(value = "分页查询气象文件数据")
@Operation(summary = "分页查询气象文件数据")
@GetMapping("page")
public Result<?> 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<WeatherData> page = weatherDataService.page(pageRequest,fileName,fileExt,dataSource,startDate,endDate);
IPage<WeatherData> page = weatherDataService.page(pageRequest,fileName,batchTime,dataSource,startDate,endDate);
Map<String, Object> rspData = new HashMap<>();
rspData.put("rows", page.getRecords());
rspData.put("total", page.getTotal());
@ -62,7 +61,8 @@ public class WeatherDataController {
List<String> timeBatchList = weatherDataService
.list(new LambdaQueryWrapper<WeatherData>()
.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();
}
/**

97
jeecg-module-weather/src/main/java/org/jeecg/job/CacheWeatherDataJob.java Normal file
View File

@ -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);
}
}
}
}

99
jeecg-module-weather/src/main/java/org/jeecg/job/CleanDataJob.java
View File

@ -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<Path> targetDirs = new ArrayList<>();
Map<String,Path> targetDirsMap = new HashMap<>();
if(t1hIsFirstRun){
Path t1hDataDir = Paths.get(this.t1hDataPath);
try (Stream<Path> 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<WeatherData> 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<Path> targetDirs = new ArrayList<>();
Map<String,Path> targetDirsMap = new HashMap<>();
if(gfsIsFirstRun){
Path gfsDataDir = Paths.get(this.gfsDataPath);
try (Stream<Path> 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<WeatherData> queryWrapper = new LambdaQueryWrapper<>();
queryWrapper.in(WeatherData::getTimeBatch,targetDirsMap.keySet());
weatherDataMapper.delete(queryWrapper);
}
this.cleanGFSDataLog();
log.info("完成{}目录清除超出保留期限气象数据",gfsDataPath);

140
jeecg-module-weather/src/main/java/org/jeecg/job/DownloadGFSJob.java
View File

@ -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<String> 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<File> files = FileUtil.loopFiles(new File(gfsBatchDataPath), file -> file.getName().startsWith(DATA_FILE_PREFIX));
//根据业务规则生成预期的文件后缀集合 (f000 ~ f384)
Set<String> 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<String> 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<File> files = List.of(FileUtil.listFiles(new File(gfsBatchDataPath), file -> file.getName().startsWith(DATA_FILE_PREFIX)));
List<File> 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<File> fileList = List.of(FileUtil.listFiles(new File(gfsBatchDataPath), file -> file.getName().startsWith(DATA_FILE_PREFIX)));
List<File> 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<WeatherDownGFSDataLog> queryWrapper = new LambdaQueryWrapper<>();
queryWrapper.eq(WeatherDownGFSDataLog::getLogType, MANUAL_DOWN_TYPE);
weatherDownGFSDataLogMapper.delete(queryWrapper);
}
}

28
jeecg-module-weather/src/main/java/org/jeecg/service/WeatherDataService.java
View File

@ -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<WeatherData> {
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<WindRoseData> getWindRose(Integer dataType, String timeBatch, LocalDateTime startTime, LocalDateTime endTime,double longitude, double latitude);
@ -20,13 +20,13 @@ public interface WeatherDataService extends IService<WeatherData> {
/**
* 分页查询气象数据
* @param pageRequest
* @param fileExt
* @param batchTime
* @param dataSource
* @param startDate
* @param endDate
* @return
*/
IPage<WeatherData> page(PageRequest pageRequest,String fileName, String fileExt, String dataSource, LocalDate startDate, LocalDate endDate);
IPage<WeatherData> page(PageRequest pageRequest,String fileName, String batchTime, String dataSource, LocalDate startDate, LocalDate endDate);
/**
* 删除气象数据
@ -41,4 +41,24 @@ public interface WeatherDataService extends IService<WeatherData> {
* @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);
}

292
jeecg-module-weather/src/main/java/org/jeecg/service/impl/WeatherDataServiceImpl.java
View File

@ -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<WeatherDataMapper, Weath
private final WeatherDataMapper weatherDataMapper;
private final WeatherLinkedDataLogService weatherLinkedDataLogService;
private final SystemStorageProperties systemStorageProperties;
private final RedisUtil redisUtil;
/**
* 根据类型和小时数获取天气数据
@ -59,32 +60,17 @@ public class WeatherDataServiceImpl extends ServiceImpl<WeatherDataMapper, Weath
* @return 天气数据列表
*/
@Override
public WeatherResultVO getWeatherData(Integer dataType, Integer weatherType, String timeBatch, LocalDateTime startTime, int hour) {
validateInputParams(weatherType, startTime, hour);
public WeatherResultVO getWeatherData(Integer dataType, Integer weatherType, String timeBatch, LocalDateTime startTime,LocalDateTime endTime, int hour) {
WeatherDataSourceEnum dataSourceEnum = WeatherDataSourceEnum.getInfoByKey(dataType);
WeatherTypeEnum weatherTypeEnum = WeatherTypeEnum.getInfoByKey(weatherType);
LocalDateTime targetTime = startTime.plusHours(hour);
try {
if (WeatherDataSourceEnum.PANGU.getKey() == dataType) {
return processWeatherData(weatherType, null, targetTime, WeatherDataSourceEnum.PANGU);
} else if (WeatherDataSourceEnum.CRA40.getKey() == dataType){
return processWeatherData(weatherType, null, targetTime, WeatherDataSourceEnum.CRA40);
} else if (WeatherDataSourceEnum.NCEP.getKey() == dataType){
return processWeatherData(weatherType, null, targetTime, WeatherDataSourceEnum.NCEP);
} else if (WeatherDataSourceEnum.FNL.getKey() == dataType){
return processWeatherData(weatherType,null, targetTime, WeatherDataSourceEnum.FNL);
} else if (WeatherDataSourceEnum.T1H.getKey() == dataType){
return processWeatherData(weatherType, timeBatch, targetTime, WeatherDataSourceEnum.T1H);
}else if (WeatherDataSourceEnum.GFS.getKey() == dataType){
return processWeatherData(weatherType, timeBatch, targetTime, WeatherDataSourceEnum.GFS);
}
} catch (JeecgBootException e) {
throw e;
} catch (Exception e) {
log.error("处理天气数据失败", e);
throw new JeecgBootException("处理天气数据失败", e);
String key = String.format(CommonConstant.WEATHER_DATA_CACHE, dataSourceEnum.getValue(), weatherTypeEnum.getValue());
startTime = startTime.plusHours(hour);
String item = startTime.format(DateTimeFormatter.ofPattern("yyyyMMddHH"));
if (!redisUtil.hHasKey(key,item)) {
throw new RuntimeException(item+"气象数据不存在");
}
throw new JeecgBootException("没有该类型的气象数据！");
return (WeatherResultVO)redisUtil.hget(key, item);
}
/**
@ -102,6 +88,8 @@ public class WeatherDataServiceImpl extends ServiceImpl<WeatherDataMapper, Weath
try {
if (WeatherDataSourceEnum.PANGU.getKey() == dataType) {
return processWeatherData(weatherType, null, targetTime, WeatherDataSourceEnum.PANGU);
}else if (WeatherDataSourceEnum.GRAPHCAST.getKey() == dataType){
return processWeatherData(weatherType, null, targetTime, WeatherDataSourceEnum.GRAPHCAST);
} else if (WeatherDataSourceEnum.CRA40.getKey() == dataType){
return processWeatherData(weatherType, null, targetTime, WeatherDataSourceEnum.CRA40);
} else if (WeatherDataSourceEnum.NCEP.getKey() == dataType){
@ -110,6 +98,8 @@ public class WeatherDataServiceImpl extends ServiceImpl<WeatherDataMapper, Weath
return processWeatherData(weatherType, null, targetTime, WeatherDataSourceEnum.FNL);
} else if (WeatherDataSourceEnum.T1H.getKey() == dataType){
return processWeatherData(weatherType, null, targetTime, WeatherDataSourceEnum.T1H);
}else if (WeatherDataSourceEnum.GFS.getKey() == dataType){
return processWeatherData(weatherType, null, targetTime, WeatherDataSourceEnum.GFS);
}
} catch (JeecgBootException e) {
throw e;
@ -283,33 +273,28 @@ public class WeatherDataServiceImpl extends ServiceImpl<WeatherDataMapper, Weath
* 分页查询气象数据
*
* @param pageRequest
* @param fileExt
* @param batchTime
* @param dataSource
* @param startDate
* @param endDate
* @return
*/
@Override
public IPage<WeatherData> page(PageRequest pageRequest,String fileName, String fileExt, String dataSource, LocalDate startDate, LocalDate endDate) {
public IPage<WeatherData> 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<WeatherData> queryWrapper = new LambdaQueryWrapper<>();
List<Integer> 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<WeatherDataMapper, Weath
*/
@Override
public void linkedData(Integer dataType, LocalDate startDate, LocalDate endDate) {
//先清空表
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<File> 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<File> 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<WeatherData> 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<WeatherData> 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<WeatherData> 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<WeatherDataMapper, Weath
variables.put("humidity", WeatherVariableNameEnum.PANGU_H.getValue());
variables.put("windU", WeatherVariableNameEnum.PANGU_U.getValue());
variables.put("windV", WeatherVariableNameEnum.PANGU_V.getValue());
} else if (WeatherDataSourceEnum.GRAPHCAST.getKey() == dataType){
variables.put("temperature", WeatherVariableNameEnum.GRAPHCAST_T.getValue());
variables.put("pressure", WeatherVariableNameEnum.GRAPHCAST_P.getValue());
variables.put("humidity", WeatherVariableNameEnum.GRAPHCAST_H.getValue());
variables.put("windU", WeatherVariableNameEnum.GRAPHCAST_U.getValue());
variables.put("windV", WeatherVariableNameEnum.GRAPHCAST_V.getValue());
} else if (WeatherDataSourceEnum.CRA40.getKey() == dataType){
variables.put("temperature", WeatherVariableNameEnum.CRA40_T.getValue());
variables.put("pressure", WeatherVariableNameEnum.CRA40_P.getValue());
@ -555,9 +608,14 @@ public class WeatherDataServiceImpl extends ServiceImpl<WeatherDataMapper, Weath
}
/**
* 处理天气数据
* 处理气象数据
* @param weatherType
* @param timeBatch
* @param targetTime
* @param dataTypeEnum
* @return
*/
private WeatherResultVO processWeatherData(Integer weatherType, String timeBatch, LocalDateTime targetTime, WeatherDataSourceEnum dataTypeEnum) {
public WeatherResultVO processWeatherData(Integer weatherType, String timeBatch, LocalDateTime targetTime, WeatherDataSourceEnum dataTypeEnum) {
LambdaQueryWrapper<WeatherData> 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<WeatherDataMapper, Weath
processResultDataInternal(weatherResultVO, dataList, null, null, lonData, latData, converter);
}
} catch (IOException e) {
log.error("NetCDF文件处理失败: {}", filePath, e);
throw new JeecgBootException("文件读取失败", e);
throw new JeecgBootException("NetCDF文件处理失败", e);
}
return weatherResultVO;
@ -887,17 +944,6 @@ public class WeatherDataServiceImpl extends ServiceImpl<WeatherDataMapper, Weath
return Arrays.asList(max, min);
}
/**
* 验证输入参数
*/
private void validateInputParams(Integer weatherType, LocalDateTime startTime, int hour) {
Objects.requireNonNull(weatherType, "天气类型不能为空");
Objects.requireNonNull(startTime, "开始时间不能为空");
if (hour < 0) {
throw new IllegalArgumentException("小时数必须大于等于0");
}
}
/**
* 验证文件
*/

2
jeecg-module-weather/src/main/java/org/jeecg/service/impl/WeatherLinkedDataLogServiceImpl.java
View File

@ -39,7 +39,7 @@ public class WeatherLinkedDataLogServiceImpl extends ServiceImpl<WeatherLinkedDa
public List<WeatherLinkedDataLog> getLinkedDataLog(Integer lastId) {
LambdaQueryWrapper<WeatherLinkedDataLog> 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);