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

Merge branch 'master' of http://git.hivekion.com:3000/panbaolin/SourceTermAnalysisSystem_java

Browse Source
This commit is contained in:
duwenyuan 2026-07-21 13:34:26 +08:00
commit 3eac85bf93
38 changed files with 1007 additions and 107 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
jeecg-boot-base-core/src/main/java/org/jeecg/common/constant/WeatherPrefixConstants.java
View File

@ -1,10 +0,0 @@
package org.jeecg.common.constant;
public class WeatherPrefixConstants {
public static final String PANGU_PREFIX = "panguweather_";
public static final String CRA40_PREFIX = "CRA40_";
public static final String NCEP_PREFIX = "cdas1";
public static final String T1H_PREFIX = "T1H_";
public static final String FNL_PREFIX = "fnl_";
}

7
jeecg-boot-base-core/src/main/java/org/jeecg/common/constant/WeatherSuffixConstants.java
View File

@ -1,7 +0,0 @@
package org.jeecg.common.constant;
public class WeatherSuffixConstants {
public static final String CRA40_SUFFIX = "_GLB_0P25_HOUR_V1_0_0";
public static final String NCEP_SUFFIX = "pgrbh";
}

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

@ -31,7 +31,11 @@ public enum WeatherDataSourceEnum {
/**
* T1H
*/
T1H(6,"T1H");
T1H(6,"T1H"),
/**
* GFS
*/
GFS(7,"GFS");
private Integer key;
private String value;

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

@ -30,7 +30,12 @@ public enum WeatherVariableNameEnum {
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");
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");
private Integer type;

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

@ -34,6 +34,11 @@ public class SystemStorageProperties {
*/
private String t1hDataPath;
/**
* gfs数据存储路径
*/
private String gfsDataPath;
/**
* fnl数据存储路径
*/

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

@ -53,6 +53,10 @@ public class TransportSimulationProperties {
* 参数配置文件路径（和T1H气象数据有关系）
*/
private String t1hParamConfigPath;
/**
* 参数配置文件路径（和GFS气象数据有关系）
*/
private String gfsParamConfigPath;
/**
* 台站数据配置文件路径（和fnl气象数据有关系）
@ -78,6 +82,10 @@ public class TransportSimulationProperties {
* 台站数据配置文件路径（和T1H气象数据有关系）
*/
private String t1hStationsConfigPath;
/**
* 台站数据配置文件路径（和GFS气象数据有关系）
*/
private String gfsStationsConfigPath;
/**
* 正演脚本路径（和fnl气象数据有关系）
@ -103,6 +111,10 @@ public class TransportSimulationProperties {
* 正演脚本路径（和T1H气象数据有关系）
*/
private String t1hForwardScriptPath;
/**
* 正演脚本路径（和GFS气象数据有关系）
*/
private String gfsForwardScriptPath;
/**
* 反演脚本路径（和fnl气象数据有关系）
@ -129,11 +141,25 @@ public class TransportSimulationProperties {
*/
private String t1hBackwardScriptPath;
/**
* 正演文件路径
* 反演脚本路径（和GFS气象数据有关系）
*/
private String gfsBackwardScriptPath;
/**
* nc转换gif配置
* 脚本路径
*/
private String gifConvertScriptPath;
/**
* python环境路径
*/
private String pythonEnvPath;
/**
* 正演任务模版文件路径
*/
private String forwardTemplatePath;
/**
* 反演文件路径
* 反演任务模版文件路径
*/
private String backwardTemplatePath;
/**

2
jeecg-boot-base-core/src/main/java/org/jeecg/common/util/Grib2TimeReader.java
View File

@ -70,7 +70,7 @@ public class Grib2TimeReader {
}
public static void main(String[] args) throws Exception {
LocalDateTime validTime = readValidTime("C:\\Users\\cnndc\\Desktop\\pangu_20260605_12_00.grib2");
LocalDateTime validTime = readValidTime("C:\\Users\\cnndc\\Desktop\\gfs.t00z.pgrb2.0p25.f004");
String result = validTime.format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
System.out.println(result);

13
jeecg-model-consumer/src/main/java/org/jeecg/jsch/JSchRemoteRunner.java → jeecg-boot-base-core/src/main/java/org/jeecg/common/util/JSchRemoteRunner.java
View File

@ -1,4 +1,4 @@
package org.jeecg.jsch;
package org.jeecg.common.util;
import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.util.StrUtil;
@ -6,7 +6,8 @@ import com.jcraft.jsch.*;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import java.io.*;
import java.io.ByteArrayInputStream;
import java.io.File;
import java.nio.charset.StandardCharsets;
import java.util.*;
@ -193,7 +194,9 @@ public class JSchRemoteRunner {
List<File> matchedFiles = new ArrayList<>();
for (ChannelSftp.LsEntry entry : fileList) {
String fileName = entry.getFilename();
if (fileName.endsWith(matchingSuffix)) {
if (StrUtil.isNotBlank(matchingSuffix) && fileName.endsWith(matchingSuffix)) {
matchedFiles.add(new File(path+File.separator+fileName));
}else if(StrUtil.isBlank(matchingSuffix)){
matchedFiles.add(new File(path+File.separator+fileName));
}
}
@ -223,7 +226,9 @@ public class JSchRemoteRunner {
Vector<ChannelSftp.LsEntry> fileList = sftp.ls(path);
for (ChannelSftp.LsEntry entry : fileList) {
String fileName = entry.getFilename();
if (fileName.endsWith(matchingSuffix)) {
if (StrUtil.isNotBlank(matchingSuffix) && fileName.endsWith(matchingSuffix)) {
matchedFiles.add(path+File.separator+fileName);
}else if(StrUtil.isBlank(matchingSuffix)){
matchedFiles.add(path+File.separator+fileName);
}
}

2
jeecg-boot-base-core/src/main/java/org/jeecg/modules/base/entity/WeatherData.java
View File

@ -47,7 +47,7 @@ public class WeatherData implements Serializable {
private LocalDateTime dataStartTime;
/**
* 数据来源（1-盘古模型，2-graphcast，3-cra40，4-ncep,5-fnl,6-t1h）
* 数据来源（1-盘古模型，2-graphcast，3-cra40，4-ncep,5-fnl,6-t1h,7-gfs）
*/
@TableField(value = "data_source")
private Integer dataSource;

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

@ -0,0 +1,43 @@
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;
/**
* 记录下载GFS气象数据日志数据表
*/
@Data
@TableName("stas_gfs_download_log")
public class WeatherDownGFSDataLog {
/**
* ID
*/
@TableId(type = IdType.AUTO)
private Integer id;
/**
* 批次时间
*/
@TableField(value = "batch_time")
private String batchTime;
/**
* 创建时间
*/
@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;
}

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

@ -0,0 +1,42 @@
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;
/**
* 记录下载T1h气象数据日志数据表
*/
@Data
@TableName("stas_t1h_download_log")
public class WeatherDownT1hDataLog {
/**
* ID
*/
@TableId(type = IdType.AUTO)
private Integer id;
/**
* 批次时间
*/
@TableField(value = "batch_time")
private String batchTime;
/**
* 创建时间
*/
@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;
}

2
jeecg-boot-base-core/src/main/java/org/jeecg/modules/base/entity/WeatherTask.java
View File

@ -41,7 +41,7 @@ public class WeatherTask{
private String taskName;
/**
* 任务状态（-1失败，0-未开始，1-等待中，2-执行中，3-已完成）
* 任务状态（-1失败，0-未开始，1-等待中，2-执行中，3-已完成,4-检查失败）
*/
@Null(message = "任务状态必须为空",groups = {InsertGroup.class, UpdateGroup.class})
@TableField(value = "task_status")

8
jeecg-boot-base-core/src/main/java/org/jeecg/modules/base/mapper/WeatherDownGFSDataLogMapper.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.WeatherDownGFSDataLog;
public interface WeatherDownGFSDataLogMapper extends BaseMapper<WeatherDownGFSDataLog> {
}

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

@ -0,0 +1,7 @@
package org.jeecg.modules.base.mapper;
import com.baomidou.mybatisplus.core.mapper.BaseMapper;
import org.jeecg.modules.base.entity.WeatherDownT1hDataLog;
public interface WeatherDownT1hDataLogMapper extends BaseMapper<WeatherDownT1hDataLog> {
}

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

@ -11,11 +11,13 @@
t.task_status as taskStatus,
t.use_met_type as useMetType,
t.time_consuming as timeConsuming,
t.create_by as createBy,
t.create_time as createTime,
t.top_task as topTask
from stas_transport_task t
<where>
<if test="taskName !=null and taskName !=''">
t.task_name like '%' || ${taskName} || '%'
</if>
<if test="taskMode !=null">
t.task_mode = #{taskMode}
</if>
@ -31,11 +33,12 @@
</where>
ORDER BY
CASE t.task_status
WHEN 1 THEN 0 <!-- 等待中 -->
WHEN -1 THEN 1 <!-- 执行失败 -->
WHEN 2 THEN 2
WHEN 0 THEN 3
ELSE 4 <!-- 其他状态 -->
WHEN -1 THEN 0 <!-- 执行失败 -->
WHEN 4 THEN 1 <!-- 检查未通过 -->
WHEN 1 THEN 2 <!-- 等待中 -->
WHEN 2 THEN 3 <!-- 执行中 -->
WHEN 3 THEN 4 <!-- 已完成 -->
ELSE 5 <!-- 其他状态 -->
END ASC,t.update_time desc
</select>
</mapper>

2
jeecg-model-consumer/src/main/java/org/jeecg/rebuild/task/SourceRebuildTaskExec.java
View File

@ -9,7 +9,7 @@ import org.jeecg.common.constant.enums.SourceRebuildReleaseSourceEnum;
import org.jeecg.common.constant.enums.SourceRebuildTaskStatusEnum;
import org.jeecg.common.properties.ServerProperties;
import org.jeecg.common.properties.SourceRebuildProperties;
import org.jeecg.jsch.JSchRemoteRunner;
import org.jeecg.common.util.JSchRemoteRunner;
import org.jeecg.modules.base.entity.SourceRebuildMonitoringData;
import org.jeecg.modules.base.entity.SourceRebuildTask;
import org.jeecg.modules.base.entity.SourceRebuildTaskLog;

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

@ -90,7 +90,6 @@ public abstract class AbstractTaskMsgHandler extends AbstractChain{
* @param transportTask
*/
protected void runTask(TransportTask transportTask,String ip){
log.info("收到任务："+transportTask.getTaskName());
boolean flag = false;
if (TransportTaskModeEnum.BACK_FORWARD.getKey().equals(transportTask.getTaskMode())){
AbstractTaskExec taskExec = new BackwardTaskExec();

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

@ -1,13 +1,16 @@
package org.jeecg.transport.flexparttask;
import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.date.LocalDateTimeUtil;
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.*;
import lombok.extern.slf4j.Slf4j;
import org.apache.logging.log4j.util.Strings;
import org.jeecg.common.constant.CommonConstant;
import org.jeecg.common.constant.enums.TransportTaskModeEnum;
import org.jeecg.common.constant.enums.WeatherDataSourceEnum;
import org.jeecg.common.properties.DataFusionProperties;
import org.jeecg.common.properties.ServerProperties;
@ -15,6 +18,8 @@ import org.jeecg.common.properties.SystemStorageProperties;
import org.jeecg.common.properties.TransportSimulationProperties;
import org.jeecg.common.util.RedisUtil;
import org.jeecg.modules.base.entity.TransportTask;
import org.jeecg.modules.base.entity.TransportTaskBackwardChild;
import org.jeecg.modules.base.entity.TransportTaskForwardSpecies;
import org.jeecg.modules.base.entity.WeatherData;
import org.jeecg.modules.base.mapper.*;
import org.jeecg.transport.service.StationDataService;
@ -22,7 +27,9 @@ import org.jeecg.transport.service.StationsModValService;
import org.jeecg.transport.service.TransportTaskService;
import java.io.*;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.time.temporal.ChronoUnit;
import java.util.ArrayList;
import java.util.List;
@Slf4j
@ -43,6 +50,8 @@ public abstract class AbstractTaskExec extends Thread{
protected TransportTask transportTask;
protected RedisUtil redisUtil;
protected boolean taskRunError;
protected static String FORWARD="forward";
protected static String BACK_FORWARD="backward";
/**
* 初始化
@ -109,13 +118,6 @@ public abstract class AbstractTaskExec extends Thread{
redisUtil.hset(CommonConstant.HOST_TASK_STATE,CommonConstant.TRAN_TASK_STATE_PRE+serverProperties.getHost(),transportTask.getId());
}
/**
* 任务运行失败，取消运行标记
*/
protected void cancelTaskRunFlag(){
redisUtil.hdel(CommonConstant.HOST_TASK_STATE,CommonConstant.TRAN_TASK_STATE_PRE+serverProperties.getHost());
}
/**
* 检查气象数据
*/
@ -140,6 +142,103 @@ public abstract class AbstractTaskExec extends Thread{
return true;
}
/**
* 用nc文件数据生成GIF动图
*/
protected void genNCToGif(Session session){
int mode = 0;
List<String> ncPaths = null;
if(TransportTaskModeEnum.FORWARD.getKey().equals(transportTask.getTaskMode())){
mode = TransportTaskModeEnum.FORWARD.getKey();
ncPaths = this.getForwardTaskNCPath();
}else if(TransportTaskModeEnum.BACK_FORWARD.getKey().equals(transportTask.getTaskMode())){
mode = TransportTaskModeEnum.BACK_FORWARD.getKey();
ncPaths = this.getBackForwardTaskNCPath();
}
try{
for(String ncPath : ncPaths){
File ncFile = new File(ncPath);
StringBuilder command = new StringBuilder();
command.append(simulationProperties.getPythonEnvPath());
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(ncFile.getParent());
command.append(File.separator);
command.append(ncFile.getParentFile().getName());
command.append(".gif");
this.execGenGIFCommand(command.toString(),session);
}
}catch (Exception e){
log.error("nc文件生成gif动图失败",e);
}
}
/**
* 获取反演NC结果文件路径
* @return
*/
private List<String> getBackForwardTaskNCPath(){
LambdaQueryWrapper<TransportTaskBackwardChild> queryWrapper = new LambdaQueryWrapper<>();
queryWrapper.eq(TransportTaskBackwardChild::getTaskId,this.transportTask.getId());
queryWrapper.select(TransportTaskBackwardChild::getId, TransportTaskBackwardChild::getStationCode);
List<TransportTaskBackwardChild> transportTaskChildren = this.taskBackwardChildMapper.selectList(queryWrapper);
if(CollUtil.isEmpty(transportTaskChildren)){
throw new RuntimeException("此任务站点信息不存在，请确认任务配置信息");
}
List<String> ncPaths = new ArrayList<>();
for (TransportTaskBackwardChild child : transportTaskChildren){
//拼接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(child.getStationCode());
path.append(File.separator);
path.append("grid_time_"+ DateUtil.format(transportTask.getEndTime(),"yyyyMMddHHmmss")+".nc");
ncPaths.add(path.toString());
}
return ncPaths;
}
/**
* 获取正演NC结果文件路径
* @return
*/
private List<String> getForwardTaskNCPath(){
LambdaQueryWrapper<TransportTaskForwardSpecies> queryWrapper = new LambdaQueryWrapper<>();
queryWrapper.eq(TransportTaskForwardSpecies::getTaskId,this.transportTask.getId());
queryWrapper.select(TransportTaskForwardSpecies::getSpeciesId);
List<TransportTaskForwardSpecies> transportTaskSpecies = this.taskForwardSpeciesMapper.selectList(queryWrapper);
if(CollUtil.isEmpty(transportTaskSpecies)){
throw new RuntimeException("此任务模拟核素信息不存在，请确认任务配置信息");
}
List<String> ncPaths = new ArrayList<>();
for (TransportTaskForwardSpecies species : transportTaskSpecies){
//拼接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(species.getSpeciesId());
path.append(File.separator);
path.append("grid_conc_"+DateUtil.format(transportTask.getStartTime(),"yyyyMMddHHmmss")+".nc");
ncPaths.add(path.toString());
}
return ncPaths;
}
/**
* 获取气象数据路径
* @param dataSource
@ -158,8 +257,27 @@ public abstract class AbstractTaskExec extends Thread{
}else if(WeatherDataSourceEnum.FNL.getKey().equals(dataSource)){
path.append(systemStorageProperties.getFnlDataPath());
}else if(WeatherDataSourceEnum.T1H.getKey().equals(dataSource)){
//T1H是中科天机的预报数据，正演只能按批次来，最大15天，每批次存储当前天往后15天的预报数据
if(TransportTaskModeEnum.FORWARD.getKey().equals(transportTask.getTaskMode())){
DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyyMMdd");
String format = formatter.format(transportTask.getStartTime());
path.append(File.separator);
path.append(format);
}else {
path.append(systemStorageProperties.getT1hDataPath());
}
}else if(WeatherDataSourceEnum.GFS.getKey().equals(dataSource)){
//GFS是美国的预报数据，正演只能按批次来，最大16天，每批次存储当前天往后16天的预报数据
//这里补一下数据所在批次的目录
if(TransportTaskModeEnum.FORWARD.getKey().equals(transportTask.getTaskMode())){
DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyyMMdd_HH");
String format = formatter.format(transportTask.getStartTime());
path.append(File.separator);
path.append("GFS_"+format);
}else {
path.append(systemStorageProperties.getGfsDataPath());
}
}
return path.toString();
}
@ -181,6 +299,8 @@ public abstract class AbstractTaskExec extends Thread{
return simulationProperties.getFnlParamConfigPath();
}else if(WeatherDataSourceEnum.T1H.getKey().equals(dataSource)){
return simulationProperties.getT1hParamConfigPath();
}else if(WeatherDataSourceEnum.GFS.getKey().equals(dataSource)){
return simulationProperties.getGfsParamConfigPath();
}
return Strings.EMPTY;
}
@ -195,7 +315,7 @@ public abstract class AbstractTaskExec extends Thread{
if(WeatherDataSourceEnum.PANGU.getKey().equals(dataSource)){
path.append(simulationProperties.getPanguStationsConfigPath());
}else if(WeatherDataSourceEnum.GRAPHCAST.getKey().equals(dataSource)){
path.append(simulationProperties.getCra40StationsConfigPath());
path.append(simulationProperties.getGraphcastStationsConfigPath());
}else if(WeatherDataSourceEnum.CRA40.getKey().equals(dataSource)){
path.append(simulationProperties.getCra40StationsConfigPath());
}else if(WeatherDataSourceEnum.NCEP.getKey().equals(dataSource)){
@ -204,14 +324,16 @@ public abstract class AbstractTaskExec extends Thread{
path.append(simulationProperties.getFnlStationsConfigPath());
}else if(WeatherDataSourceEnum.T1H.getKey().equals(dataSource)){
path.append(simulationProperties.getT1hStationsConfigPath());
}else if(WeatherDataSourceEnum.GFS.getKey().equals(dataSource)){
path.append(simulationProperties.getGfsStationsConfigPath());
}
return path.toString();
}
/**
* 持续读取日志
* 执行flexpart命令持续读取日志
*/
protected void execCommand(String command,Session session){
protected void execFlexpartCommand(String command,Session session){
ChannelExec channel = null;
try{
//打开一个执行通道
@ -241,4 +363,38 @@ public abstract class AbstractTaskExec extends Thread{
}
}
}
/**
* 执行生成GIF图片命令，持续读取日志
*/
protected void execGenGIFCommand(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)){
if(line.trim().startsWith("ERROR")){
taskRunError = true;
}
ProgressQueue.getInstance().offer(new ProgressEvent(transportTask.getId(),line));
}
}
}catch(JSchException |IOException e){
throw new RuntimeException(e);
}finally {
if (channel != null) {
channel.disconnect();
}
}
}
}

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

@ -8,8 +8,8 @@ import org.apache.logging.log4j.util.Strings;
import org.jeecg.common.constant.enums.FlexpartSpeciesType;
import org.jeecg.common.constant.enums.TransportTaskStatusEnum;
import org.jeecg.common.constant.enums.WeatherDataSourceEnum;
import org.jeecg.common.util.JSchRemoteRunner;
import org.jeecg.modules.base.entity.*;
import org.jeecg.jsch.JSchRemoteRunner;
import java.io.File;
import java.math.BigDecimal;
import java.math.RoundingMode;
@ -98,13 +98,10 @@ public class BackwardTaskExec extends AbstractTaskExec {
//生成SRS文件
// this.generateSRSFile();
}catch (Exception e){
String taskErrorLog = "任务执行失败，原因："+e.getMessage();
ProgressQueue.getInstance().offer(new ProgressEvent(super.transportTask.getId(),taskErrorLog));
//如果还未执行到flexpart，java业务代码报错，修改状态
super.transportTaskService.updateTaskStatus(super.transportTask.getId(),TransportTaskStatusEnum.FAILURE.getValue());
super.cancelTaskRunFlag();
log.error(taskErrorLog);
e.printStackTrace();
super.taskRunError = true;
String taskErrorLog = "任务执行失败，原因：";
ProgressQueue.getInstance().offer(new ProgressEvent(super.transportTask.getId(),taskErrorLog+e.getMessage()));
log.error(taskErrorLog,e);
}finally {
//添加任务耗时
stopWatch.stop();
@ -152,8 +149,7 @@ public class BackwardTaskExec extends AbstractTaskExec {
paramContent.append(super.transportTask.getZ1()).append("\n");
paramContent.append(super.transportTask.getZ2()).append("\n");
paramContent.append(metDataPath).append("\n");
paramContent.append(super.simulationProperties.getOutputPath()+ "/" +super.transportTask.getTaskName()).append("\n");
// paramContent.append(super.simulationProperties.getOutputPath()+ File.separator+super.transportTask.getTaskName()).append("\n");
paramContent.append(super.simulationProperties.getOutputPath()+ File.separator+super.transportTask.getTaskName()).append("\n");
paramContent.append(FlexpartSpeciesType.NOT_SPECIES.getId()).append("\n");//反演固定61，不显示具体核素，只显示Xe
jschRemoteRunner.writeFile(paramConfigPath,paramContent.toString());
//处理台站数据文件
@ -182,7 +178,9 @@ public class BackwardTaskExec extends AbstractTaskExec {
String execScriptMsg = "执行任务脚本，开始模拟，路径为："+scriptPath;
ProgressQueue.getInstance().offer(new ProgressEvent(super.transportTask.getId(),execScriptMsg));
//执行命令
super.execCommand(scriptPath,jschRemoteRunner.getSession());
super.execFlexpartCommand(scriptPath,jschRemoteRunner.getSession());
//生成gif动图
super.genNCToGif(jschRemoteRunner.getSession());
} catch (Exception e) {
throw new RuntimeException(e);
}finally {
@ -208,6 +206,8 @@ public class BackwardTaskExec extends AbstractTaskExec {
return simulationProperties.getFnlBackwardScriptPath();
}else if(WeatherDataSourceEnum.T1H.getKey().equals(dataSource)){
return simulationProperties.getT1hBackwardScriptPath();
}else if(WeatherDataSourceEnum.GFS.getKey().equals(dataSource)){
return simulationProperties.getGfsBackwardScriptPath();
}
return Strings.EMPTY;
}

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

@ -7,8 +7,8 @@ import org.apache.commons.lang3.time.StopWatch;
import org.apache.logging.log4j.util.Strings;
import org.jeecg.common.constant.enums.TransportTaskStatusEnum;
import org.jeecg.common.constant.enums.WeatherDataSourceEnum;
import org.jeecg.common.util.JSchRemoteRunner;
import org.jeecg.modules.base.entity.*;
import org.jeecg.jsch.JSchRemoteRunner;
import java.io.File;
import java.math.BigDecimal;
import java.math.RoundingMode;
@ -117,11 +117,10 @@ public class ForwardTaskExec extends AbstractTaskExec {
// stationDataService,simulationProperties,stationsModValService,transportTaskService);
// conformityAnalysis.exec();
}catch (Exception e){
String taskErrorLog = "任务执行失败，原因："+e.getMessage();
ProgressQueue.getInstance().offer(new ProgressEvent(super.transportTask.getId(),taskErrorLog));
this.transportTaskService.updateTaskStatus(super.transportTask.getId(),TransportTaskStatusEnum.FAILURE.getValue());
super.cancelTaskRunFlag();
log.error(taskErrorLog);
super.taskRunError = true;
String taskErrorLog = "任务执行失败，原因：";
ProgressQueue.getInstance().offer(new ProgressEvent(super.transportTask.getId(),taskErrorLog+e.getMessage()));
log.error(taskErrorLog,e);
}finally {
//添加任务耗时
stopWatch.stop();
@ -222,7 +221,7 @@ public class ForwardTaskExec extends AbstractTaskExec {
releaseInfoMap.put(station.getStationCode(),station.getForwardReleaseChild());
}
releaseInfoMap .forEach((stationCode,releaseInfoList)->{
String stationReleaseFile = super.simulationProperties.getInputSiteHourPath() + "/" + stationCode + ".txt";
String stationReleaseFile = super.simulationProperties.getInputSiteHourPath() + File.separator + stationCode + ".txt";
List<String> stationReleaseInfo = new ArrayList<>();
releaseInfoList.forEach(releaseInfo ->{
String format = "%s,%s,%s\n";
@ -243,7 +242,9 @@ public class ForwardTaskExec extends AbstractTaskExec {
String execScriptMsg = "执行任务脚本，开始模拟，路径为："+scriptPath;
ProgressQueue.getInstance().offer(new ProgressEvent(super.transportTask.getId(),execScriptMsg));
//执行命令
super.execCommand(scriptPath,jschRemoteRunner.getSession());
super.execFlexpartCommand(scriptPath,jschRemoteRunner.getSession());
//生成gif动图
super.genNCToGif(jschRemoteRunner.getSession());
} catch (Exception e) {
throw new RuntimeException(e);
}finally {
@ -268,6 +269,8 @@ public class ForwardTaskExec extends AbstractTaskExec {
return simulationProperties.getFnlForwardScriptPath();
}else if(WeatherDataSourceEnum.T1H.getKey().equals(dataSource)){
return simulationProperties.getT1hForwardScriptPath();
}else if(WeatherDataSourceEnum.GFS.getKey().equals(dataSource)){
return simulationProperties.getGfsForwardScriptPath();
}
return Strings.EMPTY;
}

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

@ -12,7 +12,7 @@ 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.jsch.JSchRemoteRunner;
import org.jeecg.common.util.JSchRemoteRunner;
import org.jeecg.modules.base.entity.WeatherData;
import org.springframework.http.MediaType;
import org.springframework.web.reactive.function.client.WebClient;

2
jeecg-module-source-rebuild/src/main/java/org/jeecg/service/impl/SourceRebuildTaskServiceImpl.java
View File

@ -59,7 +59,7 @@ public class SourceRebuildTaskServiceImpl extends ServiceImpl<SourceRebuildTaskM
queryWrapper.between(SourceRebuildTask::getCreateTime,startDateTime,endDateTime);
}
queryWrapper.like(StringUtils.isNotBlank(taskName),SourceRebuildTask::getTaskName,taskName);
queryWrapper.orderByDesc(SourceRebuildTask::getCreateTime);
queryWrapper.last("ORDER BY CASE task_status WHEN -1 THEN 0 WHEN 4 THEN 1 WHEN 1 THEN 2 WHEN 2 THEN 3 WHEN 3 THEN 4 ELSE 5 END ASC,update_time desc");
IPage<SourceRebuildTask> iPage = new Page<>(pageRequest.getPageNum(), pageRequest.getPageSize());
return this.page(iPage, queryWrapper);

11
jeecg-module-sync/src/main/java/org/jeecg/syncLog/controller/StasSyncLogController.java
View File

@ -5,11 +5,12 @@ import io.swagger.v3.oas.annotations.tags.Tag;
import org.jeecg.common.api.vo.Result;
import org.jeecg.modules.base.entity.StasSyncLog;
import org.jeecg.syncLog.service.IStasSyncLogService;
import com.baomidou.mybatisplus.core.metadata.IPage;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.*;
import java.util.List;
/**
* 同步日志信息
*/
@ -26,15 +27,13 @@ public class StasSyncLogController{
* 分页列表查询
*
* @param recordId
* @param pageNum
* @param pageSize
* @return
*/
@Operation(summary = "同步日志信息-分页列表查询")
@GetMapping(value = "/list")
public Result<IPage<StasSyncLog>> queryPageList(String recordId, Integer pageNum, Integer pageSize) {
IPage<StasSyncLog> pageList = stasSyncLogService.queryPageList(recordId, pageNum,pageSize);
return Result.OK(pageList);
public Result<?> queryList(String recordId) {
List<StasSyncLog> list = stasSyncLogService.queryList(recordId);
return Result.OK(list);
}
}

5
jeecg-module-sync/src/main/java/org/jeecg/syncLog/service/IStasSyncLogService.java
View File

@ -5,6 +5,7 @@ import org.jeecg.modules.base.entity.StasSyncLog;
import com.baomidou.mybatisplus.extension.service.IService;
import java.time.LocalDateTime;
import java.util.List;
/**
* 同步日志信息
@ -14,9 +15,7 @@ public interface IStasSyncLogService extends IService<StasSyncLog> {
/**
* 分页查询同步日志
* @param recordId
* @param pageNum
* @param pageSize
* @return
*/
IPage<StasSyncLog> queryPageList(String recordId, Integer pageNum, Integer pageSize);
List<StasSyncLog> queryList(String recordId);
}

10
jeecg-module-sync/src/main/java/org/jeecg/syncLog/service/impl/StasSyncLogServiceImpl.java
View File

@ -9,6 +9,8 @@ import org.jeecg.syncLog.service.IStasSyncLogService;
import org.springframework.stereotype.Service;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import java.util.List;
/**
* 同步日志信息
*/
@ -19,17 +21,13 @@ public class StasSyncLogServiceImpl extends ServiceImpl<StasSyncLogMapper, StasS
* 分页查询同步日志
*
* @param recordId
* @param pageNum
* @param pageSize
* @return
*/
@Override
public IPage<StasSyncLog> queryPageList(String recordId, Integer pageNum, Integer pageSize) {
public List<StasSyncLog> queryList(String recordId) {
LambdaQueryWrapper<StasSyncLog> queryWrapper = new LambdaQueryWrapper<>();
queryWrapper.eq(StasSyncLog::getRecordId, recordId);
queryWrapper.orderByAsc(StasSyncLog::getStartTime);
IPage<StasSyncLog> page = new Page<>(pageNum, pageSize);
return this.baseMapper.selectPage(page, queryWrapper);
return this.baseMapper.selectList(queryWrapper);
}
}

16
jeecg-module-sync/src/main/java/org/jeecg/taskConfig/job/AbstractSyncDataJob.java
View File

@ -29,6 +29,7 @@ public abstract class AbstractSyncDataJob implements Runnable{
protected Connection sourceConn;
protected Connection targetConn;
protected void init(StasTaskConfigMapper taskConfigMapper,
StasDataSourceMapper dataSourceMapper,
StasSyncStrategyMapper syncStrategyMapper,
@ -125,7 +126,6 @@ public abstract class AbstractSyncDataJob implements Runnable{
while (rs.next()) {
for (int i = 1; i <= columnCount; i++) {
Object value = rs.getObject(i);
// 特殊处理Oracle的TIMESTAMP类型到PostgreSQL
if (SourceDataTypeEnum.ORACLE.getKey().equals(this.sourceInfo.getType()) && SourceDataTypeEnum.POSTGRES.getKey().equals(this.targetInfo.getType())) {
String columnType = metaData.getColumnTypeName(i);
@ -153,6 +153,11 @@ public abstract class AbstractSyncDataJob implements Runnable{
pstmt.executeBatch();
this.targetConn.commit();
}
}catch (SQLException e){
// 回滚数据
this.targetConn.rollback();
log.error("保存数据出现错误，原因为",e);
throw e;
}
}
return totalRows;
@ -209,15 +214,10 @@ public abstract class AbstractSyncDataJob implements Runnable{
*
* @param e the exception
*/
public void uncaughtException(Throwable e) {
public void uncaughtException(Exception e) {
String tableName = this.syncStrategy.getSourceOwner()+"."+this.syncStrategy.getTableName();
log.error(tableName+"表数据同步出现错误，原因为",e);
String errLog = String.format("%s表数据同步出现错误，原因为：%s",tableName, e.getMessage());
log.error(errLog);
saveSyncLog(errLog);
try {
targetConn.rollback();
} catch (SQLException ex) {
log.error("数据回滚出现错误，原因为：{}",ex.getMessage());
}
}
}

26
jeecg-module-sync/src/main/java/org/jeecg/taskConfig/job/SyncDataByDateColumnJob.java
View File

@ -80,9 +80,15 @@ public class SyncDataByDateColumnJob extends AbstractSyncDataJob {
//保存表批次同步数量
super.saveSyncNum(rowsSynced);
//保存同步日志并修改同步位置
super.syncStrategy.setSyncOrigin(sdf.format(currentEnd));
Date finalSyncOrigin = null;
if (rowsSynced > 0) {
finalSyncOrigin = getLocalRealSyncOrigin();
}else {
finalSyncOrigin = currentEnd;
}
super.syncStrategy.setSyncOrigin(sdf.format(finalSyncOrigin));
super.syncStrategyMapper.updateById(super.syncStrategy);
super.saveSyncLog(String.format("%s表最终同步位置已更新为: %s",tableName, sdf.format(currentEnd)));
super.saveSyncLog(String.format("%s表最终同步位置已更新为: %s",tableName, sdf.format(finalSyncOrigin)));
}
/**
@ -100,4 +106,20 @@ public class SyncDataByDateColumnJob extends AbstractSyncDataJob {
}
throw new SQLException("无法获取日期范围");
}
/**
* 获取本地oracle数据表的本次同步数据对应字段的实际最新值
*/
private Date getLocalRealSyncOrigin() throws SQLException {
String sql = "MAX(" + this.syncStrategy.getColumnName() + ") as max_val " +
"FROM \"" + this.syncStrategy.getTargetOwner().toUpperCase() + "\".\"" + this.syncStrategy.getTableName() + "\"";
try (Statement stmt = this.targetConn.createStatement();
ResultSet rs = stmt.executeQuery(sql)) {
if (rs.next()) {
return rs.getTimestamp("max_val");
}
}
String errLog = this.syncStrategy.getTargetOwner().toUpperCase() + "\".\"" + this.syncStrategy.getTableName() + "\"";
throw new SQLException("无法获取"+errLog+"最新值");
}
}

27
jeecg-module-sync/src/main/java/org/jeecg/taskConfig/job/SyncDataByIncrementColumnJob.java
View File

@ -4,6 +4,7 @@ import cn.hutool.core.date.StopWatch;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.jeecg.vo.IdRangeVO;
import java.sql.*;
import java.util.Objects;
@ -83,9 +84,15 @@ public class SyncDataByIncrementColumnJob extends AbstractSyncDataJob {
// 更新同步位置
if (currentEnd > 0) {
super.syncStrategy.setSyncOrigin(String.valueOf(currentEnd));
long finalSyncOrigin;
if (rowsSynced > 0) {
finalSyncOrigin = getLocalRealSyncOrigin();
}else {
finalSyncOrigin = currentEnd;
}
super.syncStrategy.setSyncOrigin(String.valueOf(finalSyncOrigin));
super.syncStrategyMapper.updateById(syncStrategy);
super.saveSyncLog(String.format("%s表最终同步位置已更新为: %s",tableName, currentEnd));
super.saveSyncLog(String.format("%s表最终同步位置已更新为: %s",tableName, finalSyncOrigin));
}
}
@ -104,4 +111,20 @@ public class SyncDataByIncrementColumnJob extends AbstractSyncDataJob {
}
throw new SQLException("无法获取ID范围");
}
/**
* 获取本地oracle数据表的本次同步数据对应字段的实际最新值
*/
private long getLocalRealSyncOrigin() throws SQLException {
String sql = "SELECT MAX(" + this.syncStrategy.getColumnName() + ") as max_val " +
"FROM \"" + this.syncStrategy.getTargetOwner().toUpperCase() + "\".\"" + this.syncStrategy.getTableName() + "\"";
try (Statement stmt = this.targetConn.createStatement();
ResultSet rs = stmt.executeQuery(sql)) {
if (rs.next()) {
return rs.getLong("max_val");
}
}
String errLog = this.syncStrategy.getTargetOwner().toUpperCase() + "\".\"" + this.syncStrategy.getTableName() + "\"";
throw new SQLException("无法获取"+errLog+"最新值");
}
}

4
jeecg-module-sync/src/main/java/org/jeecg/taskConfig/job/SyncDataJob.java
View File

@ -120,8 +120,8 @@ public class SyncDataJob implements Job {
* @throws InterruptedException
*/
private void closeThreadPool(String recordId,String taskId) throws InterruptedException {
//再次多等待2分钟，如果还未全部关闭，则强制停止
if (!threadPoolExecutor.awaitTermination(120, TimeUnit.SECONDS)) {
//再次多等待10分钟，如果还未全部关闭，则强制停止
if (!threadPoolExecutor.awaitTermination(600, TimeUnit.SECONDS)) {
List<Runnable> droppedTasks = threadPoolExecutor.shutdownNow();
log.warn("超时！执行强制关闭，丢弃了 " + droppedTasks.size() + " 个未执行任务");
if (!threadPoolExecutor.awaitTermination(60, TimeUnit.SECONDS)) {

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

@ -115,4 +115,11 @@ public class TransportResultDataController {
public Result<?> getTaskStations(@NotNull(message = "任务id不能为空") Integer taskId) {
return Result.OK(transportResultDataService.getTaskStations(taskId));
}
@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));
}
}

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

@ -1,5 +1,6 @@
package org.jeecg.service;
import jakarta.validation.constraints.NotNull;
import org.jeecg.vo.ContributionAnalysisVO;
import org.jeecg.vo.QueryDiffusionVO;
import org.jeecg.vo.TaskStationsVO;
@ -108,4 +109,12 @@ public interface TransportResultDataService {
* @return
*/
Map<String,List<Map<String,Object>>> getConformityAnalysisScatterPlot(Integer taskId,String stationIds,Integer speciesId,String nuclideName);
/**
* 获取gif地址
* @param taskId
* @param speciesId
* @return
*/
String getTaskGifAddr(Integer taskId,Integer speciesId);
}

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

@ -4,6 +4,7 @@ import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.date.DateUtil;
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 lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
@ -674,6 +675,28 @@ public class TransportResultDataServiceImpl implements TransportResultDataServic
return resultMap;
}
/**
* 获取gif地址
* @param taskId
* @param speciesId
* @return
*/
@Override
public String getTaskGifAddr(Integer taskId,Integer speciesId) {
TransportTask transportTask = this.transportTaskMapper.selectById(taskId);
//nginx代理父路径就是/gif
String parentPath = "/gif";
if(TransportTaskModeEnum.FORWARD.getKey().equals(transportTask.getTaskMode())){
return this.getForwardTaskGIFPath(transportTask,speciesId,parentPath);
}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 StrUtil.EMPTY;
}
/**
* 获取正演NC结果文件路径
* @param transportTask
@ -694,6 +717,27 @@ public class TransportResultDataServiceImpl implements TransportResultDataServic
return path.toString();
}
/**
* 获取正演GIF文件路径
* @param transportTask
* @param speciesId
* @return
*/
private String getForwardTaskGIFPath(TransportTask transportTask,Integer speciesId,String parentPath){
//拼接GIF文件路径
StringBuilder path = new StringBuilder();
path.append(parentPath);
path.append(File.separator);
path.append(transportTask.getTaskName());
path.append(File.separator);
path.append(FORWARD);
path.append(File.separator);
path.append(speciesId.toString());
path.append(File.separator);
path.append(speciesId+".gif");
return path.toString();
}
/**
* 获取反演NC结果文件路径
* @param transportTask
@ -714,6 +758,27 @@ public class TransportResultDataServiceImpl implements TransportResultDataServic
return path.toString();
}
/**
* 获取反演GIF文件路径
* @param transportTask
* @param stationCode
* @return
*/
private String getBackForwardTaskGIFPath(TransportTask transportTask,String stationCode,String parentPath){
//拼接gif文件路径
StringBuilder path = new StringBuilder();
path.append(parentPath);
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+".gif");
return path.toString();
}
/**
* 在步长为 0.25° 的递增经纬度数组中，查找指定位置的索引
* - lons[0] 是起始经度

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

@ -216,6 +216,7 @@ public class TransportTaskServiceImpl extends ServiceImpl<TransportTaskMapper,Tr
taskBackwardChildMapper.delete(delQueryWrapper);
if(CollUtil.isNotEmpty(transportTask.getBackwardChild())){
transportTask.getBackwardChild().forEach(transportTaskChild -> {
transportTaskChild.setId(null);
transportTaskChild.setTaskId(transportTask.getId());
});
taskBackwardChildMapper.insert(transportTask.getBackwardChild());
@ -229,6 +230,7 @@ public class TransportTaskServiceImpl extends ServiceImpl<TransportTaskMapper,Tr
delForwardChildQueryWrapper.eq(TransportTaskForwardChild::getTaskId,checkIdResult.getId());
taskForwardChildMapper.delete(delForwardChildQueryWrapper);
transportTask.getForwardChild().forEach(transportTaskChild -> {
transportTaskChild.setId(null);
transportTaskChild.setTaskId(transportTask.getId());
});
taskForwardChildMapper.insert(transportTask.getForwardChild());
@ -1105,7 +1107,8 @@ public class TransportTaskServiceImpl extends ServiceImpl<TransportTaskMapper,Tr
!WeatherDataSourceEnum.T1H.getKey().equals(metDataSourceCellValue) &&
!WeatherDataSourceEnum.FNL.getKey().equals(metDataSourceCellValue) &&
!WeatherDataSourceEnum.NCEP.getKey().equals(metDataSourceCellValue) &&
!WeatherDataSourceEnum.CRA40.getKey().equals(metDataSourceCellValue)) {
!WeatherDataSourceEnum.CRA40.getKey().equals(metDataSourceCellValue) &&
!WeatherDataSourceEnum.GFS.getKey().equals(metDataSourceCellValue)) {
info.add("气象数据来源配置错误");
}else {
transportTask.setUseMetType(metDataSourceCellValue);

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

@ -58,11 +58,11 @@ public class WeatherDataController {
@AutoLog(value = "查询批次列表")
@Operation(summary = "查询批次列表")
@GetMapping(value = "getTimeBatch")
public Result<?> getTimeBatch() {
public Result<?> getTimeBatch(Integer dataType) {
List<String> timeBatchList = weatherDataService
.list(new LambdaQueryWrapper<WeatherData>()
.select(WeatherData::getTimeBatch) // 只查询需要的字段
.eq(WeatherData::getDataSource, WeatherDataSourceEnum.T1H.getKey()))
.eq(WeatherData::getDataSource,dataType))
.stream()
.map(WeatherData::getTimeBatch)
.distinct()

209
jeecg-module-weather/src/main/java/org/jeecg/job/CleanDataJob.java Normal file
View File

@ -0,0 +1,209 @@
package org.jeecg.job;
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.WeatherDownGFSDataLog;
import org.jeecg.modules.base.entity.WeatherDownT1hDataLog;
import org.jeecg.modules.base.mapper.WeatherDownGFSDataLogMapper;
import org.jeecg.modules.base.mapper.WeatherDownT1hDataLogMapper;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import java.io.File;
import java.io.IOException;
import java.nio.file.*;
import java.nio.file.attribute.BasicFileAttributes;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.List;
import java.util.stream.Stream;
/**
* 每天定时清除T1h和GFS历史数据
*/
@Slf4j
@Component
@RequiredArgsConstructor
public class CleanDataJob {
@Value("${gfs-download.cleanSwitch}")
private boolean gfsCleanSwitch;
@Value("${gfs-download.retentionPeriod}")
private Integer gfsDataRetentionPeriod;
@Value("${gfs-download.gfsDataPath}")
private String gfsDataPath;
@Value("${t1h-download.cleanSwitch}")
private boolean t1hCleanSwitch;
@Value("${t1h-download.retentionPeriod}")
private Integer t1hDataRetentionPeriod;
@Value("${t1h-download.t1hDataPath}")
private String t1hDataPath;
private final WeatherDownGFSDataLogMapper downGFSDataLogMapper;
private final WeatherDownT1hDataLogMapper downT1hDataLogMapper;
private boolean t1hIsFirstRun = true;
private boolean gfsIsFirstRun = true;
private final static String HOURS = "00";
private final static String DIRPREFIX = "GFS_";
/**
* 每天23点清除T1H和GFS历史预报数据
*/
@Scheduled(cron = "0 0 23 * * ?")
public void cleanDirectory(){
if(t1hCleanSwitch){
this.cleanT1hData();
if (t1hIsFirstRun){
t1hIsFirstRun = false;
}
}
if(gfsCleanSwitch){
this.cleanGFSData();
if (gfsIsFirstRun){
gfsIsFirstRun = false;
}
}
}
/**
* 清除t1h历史数据
*/
private void cleanT1hData(){
DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyyMMdd");
List<Path> targetDirs = new ArrayList<>();
if(t1hIsFirstRun){
Path t1hDataDir = Paths.get(this.t1hDataPath);
try (Stream<Path> stream = Files.walk(t1hDataDir)){
List<Path> dirs = stream.filter(Files::isDirectory)
.filter(path -> !path.equals(t1hDataDir))
.toList();
LocalDate retentionDate = LocalDate.now().minusDays(t1hDataRetentionPeriod);
for (Path dir : dirs) {
String name = dir.toFile().getName();
LocalDate dirDate = LocalDate.parse(name, formatter);
if (!dirDate.isAfter(retentionDate)) {
targetDirs.add(dir);
}
}
} catch (IOException e) {
throw new RuntimeException(e);
}
}else {
LocalDate localDate = LocalDate.now().minusDays(t1hDataRetentionPeriod);
String batchTime = formatter.format(localDate);
String targetDir = this.t1hDataPath+File.separator+batchTime;
targetDirs.add(Paths.get(targetDir));
}
try {
for (Path targetDirPath : targetDirs){
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;
}
});
}
this.cleanT1hDataLog();
log.info("完成{}目录清除超出保留期限气象数据",t1hDataPath);
} catch (IOException e) {
log.error("{}目录清除超出保留期限气象数据出现异常",t1hDataPath,e);
}
}
/**
* 清除gfs历史数据
*/
private void cleanGFSData(){
DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyyMMdd");
List<Path> targetDirs = new ArrayList<>();
if(gfsIsFirstRun){
Path gfsDataDir = Paths.get(this.gfsDataPath);
try (Stream<Path> stream = Files.walk(gfsDataDir)){
List<Path> dirs = stream.filter(Files::isDirectory)
.filter(path -> !path.equals(gfsDataDir))
.toList();
LocalDate retentionDate = LocalDate.now().minusDays(gfsDataRetentionPeriod);
for (Path dir : dirs) {
String name = dir.toFile().getName();
if (name.startsWith(DIRPREFIX)) {
String dateStr = name.substring(DIRPREFIX.length(),name.lastIndexOf(StringPool.UNDERSCORE));
LocalDate dirDate = LocalDate.parse(dateStr, formatter);
if (!dirDate.isAfter(retentionDate)) {
targetDirs.add(dir);
}
}
}
} catch (IOException e) {
throw new RuntimeException(e);
}
}else {
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));
}
try {
for (Path targetDirPath : targetDirs){
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;
}
});
}
this.cleanGFSDataLog();
log.info("完成{}目录清除超出保留期限气象数据",gfsDataPath);
} catch (IOException e) {
log.error("{}目录清除超出保留期限气象数据出现异常",gfsDataPath,e);
}
}
/**
* 清除gfs下载日志
*/
private void cleanGFSDataLog(){
LocalDateTime cleanDateTime = LocalDate.now().minusDays(gfsDataRetentionPeriod).atTime(23,59,59);
LambdaQueryWrapper<WeatherDownGFSDataLog> queryWrapper = new LambdaQueryWrapper<>();
queryWrapper.le(WeatherDownGFSDataLog::getCreateTime,cleanDateTime);
this.downGFSDataLogMapper.delete(queryWrapper);
}
/**
* 清除他t1h下载日志
*/
private void cleanT1hDataLog(){
LocalDateTime cleanDateTime = LocalDate.now().minusDays(gfsDataRetentionPeriod).atTime(23,59,59);
LambdaQueryWrapper<WeatherDownT1hDataLog> queryWrapper = new LambdaQueryWrapper<>();
queryWrapper.le(WeatherDownT1hDataLog::getCreateTime,cleanDateTime);
this.downT1hDataLogMapper.delete(queryWrapper);
}
}

274
jeecg-module-weather/src/main/java/org/jeecg/job/DownloadGFSJob.java Normal file
View File

@ -0,0 +1,274 @@
package org.jeecg.job;
import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.util.StrUtil;
import com.baomidou.mybatisplus.core.toolkit.StringPool;
import com.jcraft.jsch.ChannelExec;
import com.jcraft.jsch.JSchException;
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;
import org.jeecg.modules.base.entity.WeatherData;
import org.jeecg.modules.base.entity.WeatherDownGFSDataLog;
import org.jeecg.modules.base.mapper.WeatherDataMapper;
import org.jeecg.modules.base.mapper.WeatherDownGFSDataLogMapper;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import java.io.*;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.TimeUnit;
/**
* 下载GFS预报数据任务
*/
@Slf4j
@RequiredArgsConstructor
@Component
public class DownloadGFSJob {
@Value("${gfs-download.downloadSwitch:false}")
private boolean downloadSwitch;
@Value("${gfs-download.pythonEnvPath}")
private String pythonEnvPath;
@Value("${gfs-download.gfsDownloadScriptPath}")
private String downloadScriptPath;
@Value("${gfs-download.gfsDataPath}")
private String gfsDataPath;
private final WeatherDownGFSDataLogMapper weatherDownGFSDataLogMapper;
private final ServerProperties serverProperties;
private final WeatherDataMapper weatherDataMapper;
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() {
//开关为true才执行下载任务
if(downloadSwitch){
retryCount = 0;
//ssh连接宿主机调用flexpart
JSchRemoteRunner jschRemoteRunner = new JSchRemoteRunner();
try {
//登录ssh
jschRemoteRunner.login(this.serverProperties.getHost(),this.serverProperties.getPort(),
this.serverProperties.getUsername(),this.serverProperties.getPassword());
//获取批次时间和下载命令
DateTimeFormatter dateTimeFormatter = DateTimeFormatter.ofPattern("yyyyMMdd");
String batchTime = dateTimeFormatter.format(LocalDate.now());
this.execDownload(jschRemoteRunner,batchTime);
}catch (Exception e) {
log.error("GFS预报数据下载出现错误",e);
}
}
}
/**
* 执行下载
* @param jschRemoteRunner
*/
private void execDownload(JSchRemoteRunner jschRemoteRunner,String batchTime){
//获取批次时间和下载命令
String downloadCommand = this.getDownloadCommand(batchTime);
//执行下载命令
log.info("开始下载");
log.info("下载命令为{}",downloadCommand);
this.execCommand(batchTime,downloadCommand,jschRemoteRunner.getSession());
//检查结果数据
this.checkData(jschRemoteRunner,batchTime);
}
/**
* 检验下载的数据是否完整
* @param jschRemoteRunner
* @param batchTime
*/
private void checkData(JSchRemoteRunner jschRemoteRunner,String batchTime){
log.info("开始检查数据");
String gfsBatchDataPath = this.getGfsDataPath(batchTime);
List<File> files = List.of(FileUtil.listFiles(new File(gfsBatchDataPath), file -> file.getName().startsWith(DATA_FILE_PREFIX)));
try {
if(CollUtil.isEmpty(files)){
throw new RuntimeException(gfsBatchDataPath+"目录没有数据，下载可能出现错误");
}
//根据业务规则生成预期的文件后缀集合 (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)) {
String logContent = "缺失"+DATA_FILE_PREFIX+expected+"气象文件";
this.saveLog(batchTime,logContent);
flag = true;
}
}
if (flag){
throw new RuntimeException(gfsBatchDataPath+"目录批次数据存在缺失");
}
//保存数据入库
this.saveDataToDataBase(batchTime);
}catch (Exception e) {
this.saveLog(e.getMessage(),batchTime);
log.error(e.getMessage(),e);
try {
//间隔5分钟，重试200次
if(retryCount <= 200){
retryCount++;
TimeUnit.MINUTES.sleep(5);
log.info("{}批次数据，重试下载{}次,最大200次",gfsBatchDataPath,retryCount);
this.saveLog(batchTime,batchTime);
this.execDownload(jschRemoteRunner,batchTime);
}
} catch (InterruptedException ex) {
throw new RuntimeException(ex);
}
}
}
/**
* 保存已下载的数据入库
* @param batchTime
*/
private void saveDataToDataBase(String batchTime){
//本批次数据开始日期
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)));
if (CollUtil.isNotEmpty(fileList)) {
for (File file : fileList) {
try {
//获取文件数据时间
int hour = Integer.parseInt(file.getName().substring(DATA_FILE_PREFIX.length()+1));
LocalDateTime validTime = localDateTime.plusHours(hour);
//如果此文件存在则无需再次新增
String gribFileMD5 = this.getGribFileMD5(file.getAbsolutePath());
WeatherData weatherData = new WeatherData();
weatherData.setFileName(file.getName());
weatherData.setFileExt(file.getName().substring(file.getName().lastIndexOf(".")+1));
weatherData.setDataSource(WeatherDataSourceEnum.GFS.getKey());
weatherData.setFilePath(file.getAbsolutePath());
weatherData.setDataStartTime(validTime);
weatherData.setMd5Value(gribFileMD5);
weatherData.setTimeBatch(batchTime);
this.weatherDataMapper.insert(weatherData);
}catch (Exception e){
log.error("保存{}气象数据文件出现错误，原因为：",file.getAbsolutePath(),e);
String logContent = "保存{}气象数据文件出现错误，原因为："+e.getMessage();
this.saveLog(logContent,batchTime);
}
}
}
}
/**
* 获取GRIB文件的MD5唯一值
*/
private String getGribFileMD5(String filePath) throws IOException {
try (FileInputStream fis = new FileInputStream(filePath)) {
// 底层自动采用流式读取，内存占用极低
return DigestUtils.md5Hex(fis);
}
}
/**
* 获取下载命令
* @param batchTime
* @return
*/
private String getDownloadCommand(String batchTime){
StringBuilder command = new StringBuilder();
command.append(pythonEnvPath);
command.append(" -u ");
command.append(downloadScriptPath);
command.append(" --date ");
command.append(batchTime);
command.append(" --cycle ");
command.append(HOURS);
command.append(" --outdir ");
command.append(this.getGfsDataPath(batchTime));
return command.toString();
}
/**
* 获取GFS数据存储路径
* @param batchTime
* @return
*/
private String getGfsDataPath(String batchTime){
return gfsDataPath +
File.separator +
DIRPREFIX +
batchTime +
StringPool.UNDERSCORE +
HOURS;
}
/**
* 执行下载命令持续读取日志
*/
protected void execCommand(String batchTime,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)){
//保存日志
this.saveLog(line,batchTime);
}
}
}catch(JSchException | IOException e){
throw new RuntimeException(e);
}finally {
if (channel != null) {
channel.disconnect();
}
}
}
/**
* 保存过程日志
*/
private void saveLog(String log,String batchTime){
WeatherDownGFSDataLog dataLog = new WeatherDownGFSDataLog();
dataLog.setLogContent(log);
dataLog.setBatchTime(batchTime);
weatherDownGFSDataLogMapper.insert(dataLog);
}
}

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

@ -34,10 +34,8 @@ import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.text.DecimalFormat;
import java.time.Instant;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.stream.Collectors;
@ -77,6 +75,8 @@ public class WeatherDataServiceImpl extends ServiceImpl<WeatherDataMapper, Weath
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;
@ -313,6 +313,7 @@ public class WeatherDataServiceImpl extends ServiceImpl<WeatherDataMapper, Weath
queryWrapper.like(StringUtils.isNotBlank(fileName),WeatherData::getFileName, fileName);
queryWrapper.select(WeatherData::getId,WeatherData::getFileName,WeatherData::getDataSource,
WeatherData::getFileExt,WeatherData::getDataStartTime,WeatherData::getFilePath,WeatherData::getTimeBatch);
queryWrapper.orderByDesc(WeatherData::getDataStartTime);
IPage<WeatherData> iPage = new Page<>(pageRequest.getPageNum(),pageRequest.getPageSize());
return this.baseMapper.selectPage(iPage, queryWrapper);
}
@ -520,6 +521,12 @@ public class WeatherDataServiceImpl extends ServiceImpl<WeatherDataMapper, Weath
variables.put("humidity", WeatherVariableNameEnum.T1H_H.getValue());
variables.put("windU", WeatherVariableNameEnum.T1H_U.getValue());
variables.put("windV", WeatherVariableNameEnum.T1H_V.getValue());
}else if (WeatherDataSourceEnum.GFS.getKey() == dataType){
variables.put("temperature", WeatherVariableNameEnum.GFS_T.getValue());
variables.put("pressure", WeatherVariableNameEnum.GFS_P.getValue());
variables.put("humidity", WeatherVariableNameEnum.GFS_H.getValue());
variables.put("windU", WeatherVariableNameEnum.GFS_U.getValue());
variables.put("windV", WeatherVariableNameEnum.GFS_V.getValue());
}
return variables;
}

6
jeecg-module-weather/src/main/java/org/jeecg/service/impl/WeatherTaskServiceImpl.java
View File

@ -17,16 +17,13 @@ import org.jeecg.common.properties.SystemStorageProperties;
import org.jeecg.common.system.query.PageRequest;
import org.jeecg.modules.base.entity.WeatherTask;
import org.jeecg.modules.base.entity.WeatherTaskLog;
import org.jeecg.modules.base.mapper.WeatherDataMapper;
import org.jeecg.modules.base.mapper.WeatherTaskLogMapper;
import org.jeecg.modules.base.mapper.WeatherTaskMapper;
import org.jeecg.service.WeatherTaskService;
import org.reflections.vfs.Vfs;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.web.multipart.MultipartFile;
import java.io.File;
import java.io.FileFilter;
import java.io.IOException;
import java.time.LocalDate;
import java.time.LocalDateTime;
@ -43,7 +40,6 @@ public class WeatherTaskServiceImpl extends ServiceImpl<WeatherTaskMapper, Weath
private final WeatherTaskLogMapper weatherTaskLogMapper;
private final SystemStorageProperties systemStorageProperties;
private final WeatherDataMapper weatherDataMapper;
/**
* 分页查询任务列表
@ -64,7 +60,7 @@ public class WeatherTaskServiceImpl extends ServiceImpl<WeatherTaskMapper, Weath
queryWrapper.between(WeatherTask::getStartDate,startDateTime,endDateTime);
}
queryWrapper.like(StringUtils.isNotBlank(taskName),WeatherTask::getTaskName,taskName);
queryWrapper.orderByDesc(WeatherTask::getStartDate);
queryWrapper.last("ORDER BY CASE task_status WHEN -1 THEN 0 WHEN 4 THEN 1 WHEN 1 THEN 2 WHEN 2 THEN 3 WHEN 3 THEN 4 ELSE 5 END ASC,update_time desc");
IPage<WeatherTask> iPage = new Page<>(pageRequest.getPageNum(), pageRequest.getPageSize());
return this.page(iPage, queryWrapper);