/**
 * Copyright 2019 DH
 * All right reserved.
 * 项目名称: 大恒泰山系统
 * 创建日期:2022/8/24
 */
package org.springblade.modules.business.warning.service.impl;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import lombok.extern.slf4j.Slf4j;
import org.springblade.core.mp.base.BaseServiceImpl;
import org.springblade.core.tool.jackson.JsonUtil;
import org.springblade.core.tool.utils.ConcurrentDateFormat;
import org.springblade.core.tool.utils.Func;
import org.springblade.enums.WarnKindEnum;
import org.springblade.enums.WarningActiveEnum;
import org.springblade.enums.WarningStatusEnum;
import org.springblade.etl.entity.EtlRainDataEntity;
import org.springblade.etl.entity.EtlRiverDataEntity;
import org.springblade.etl.entity.EtlRsvrDataEntity;
import org.springblade.modules.baseinfo.rtu.entity.RtuInfoEntity;
import org.springblade.modules.business.data.entity.RtuDataGroundEntity;
import org.springblade.modules.business.data.entity.RtuDataRainEntity;
import org.springblade.modules.business.data.entity.RtuDataRiverEntity;
import org.springblade.modules.business.data.entity.RtuDataRsvrEntity;
import org.springblade.modules.business.data.service.IRtuDataGroundService;
import org.springblade.modules.business.data.service.IRtuDataRainService;
import org.springblade.modules.business.data.service.IRtuDataRiverService;
import org.springblade.modules.business.data.service.IRtuDataRsvrService;
import org.springblade.modules.business.rtumanage.entity.RtuStatusEntity;
import org.springblade.modules.business.rtumanage.service.IRtuManageService;
import org.springblade.modules.business.warning.dto.RtuWarningInfoDTO;
import org.springblade.modules.business.warning.entity.RtuWarnSettingEntity;
import org.springblade.modules.business.warning.entity.RtuWarningInfoEntity;
import org.springblade.modules.business.warning.mapper.RtuWarningMapper;
import org.springblade.modules.business.warning.service.IRtuWarnSettingService;
import org.springblade.modules.business.warning.service.IRtuWarningService;
import org.springblade.modules.business.warning.vo.RtuWarningInfoVO;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.HashOperations;
import org.springframework.data.redis.core.ListOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ValueOperations;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.Calendar;
import java.util.Date;
import java.util.List;


/***
 * Date:2022/8/24
 * Title:预警服务接口实现模块
 * Description：对本文件的详细描述，原则上不能少于30字
 * @author dylan
 * @version 1.0
 * Remark：认为有必要的其他信息
 */
@Slf4j
@Service
public class RtuWarningServiceImpl extends BaseServiceImpl<RtuWarningMapper, RtuWarningInfoEntity> implements IRtuWarningService {

	@Resource
	private IRtuWarnSettingService rtuWarnSettingService;

	@Resource
	private IRtuManageService rtuManageService;

	@Resource
	private IRtuDataRainService rtuDataRainService;

	@Resource
	private IRtuDataRiverService rtuDataRiverService;

	@Resource
	private IRtuDataRsvrService rtuDataRsvrService;

	@Resource
	private IRtuDataGroundService rtuDataGroundService;

	@Resource
	private KafkaTemplate<String, String> kafkaTemplate;

	@Resource
	private RedisTemplate redisTemplate;

	@Value("${spring.mq-topic.ywxt-warning}")
	private String topicYwxtWarning;

	@Override
	public IPage<RtuWarningInfoVO> selectPage(IPage<RtuWarningInfoVO> page, RtuWarningInfoDTO rtuWarningInfoDTO) {
		return page.setRecords(baseMapper.selectPage(page, rtuWarningInfoDTO));
	}

	@Override
	public Long warningRtuCount(RtuWarningInfoDTO rtuWarningInfoDTO) {
		return baseMapper.warningRtuCount(rtuWarningInfoDTO);
	}

	/**
	 * 时钟预警
	 *
	 * @param rtuInfoEntity
	 * @param checkTm
	 * @param upTime
	 */
	private void clockWarnHappen(RtuInfoEntity rtuInfoEntity, Date checkTm, Date upTime) {
		LambdaQueryWrapper<RtuWarningInfoEntity> warningWrapper = Wrappers.<RtuWarningInfoEntity>query().lambda();
		warningWrapper.eq(RtuWarningInfoEntity::getIsDeleted, 0);
		warningWrapper.eq(RtuWarningInfoEntity::getRtuCode, rtuInfoEntity.getRtuCode());
		warningWrapper.eq(RtuWarningInfoEntity::getWarningStatus, WarningStatusEnum.STATUS_HAPPEN.getCode());
		warningWrapper.eq(RtuWarningInfoEntity::getWarningKind, WarnKindEnum.WARN_CLOCK.getCode());
		RtuWarningInfoEntity warningInfoEntity = this.getOne(warningWrapper);
		if (null != warningInfoEntity) {
			warningInfoEntity.setUpdateTime(checkTm);
			this.updateById(warningInfoEntity);
			kafkaTemplate.send(topicYwxtWarning, JsonUtil.toJson(warningInfoEntity));
		} else {
			RtuWarningInfoEntity entity = new RtuWarningInfoEntity();
			entity.setRtuCode(rtuInfoEntity.getRtuCode());
			entity.setWarningHappenTime(checkTm);
			entity.setWarningStatus(WarningStatusEnum.STATUS_HAPPEN.getCode());
			entity.setWarningKind(WarnKindEnum.WARN_CLOCK.getCode());
			entity.setWarningDesc("时钟异常，服务器时间：" + Func.formatDateTime(checkTm) + "," + "测站采集时间：" + Func.formatDateTime(upTime));
			this.save(entity);
			kafkaTemplate.send(topicYwxtWarning, JsonUtil.toJson(entity));
		}
	}

	/**
	 * 时钟预警恢复
	 *
	 * @param rtuInfoEntity
	 * @param checkTm
	 */
	private void clockWarnRecovery(RtuInfoEntity rtuInfoEntity, Date checkTm) {
		LambdaQueryWrapper<RtuWarningInfoEntity> warningWrapper = Wrappers.<RtuWarningInfoEntity>query().lambda();
		warningWrapper.eq(RtuWarningInfoEntity::getIsDeleted, 0);
		warningWrapper.eq(RtuWarningInfoEntity::getRtuCode, rtuInfoEntity.getRtuCode());
		warningWrapper.eq(RtuWarningInfoEntity::getWarningStatus, WarningStatusEnum.STATUS_HAPPEN.getCode());
		warningWrapper.eq(RtuWarningInfoEntity::getWarningKind, WarnKindEnum.WARN_CLOCK.getCode());
		RtuWarningInfoEntity warningInfoEntity = this.getOne(warningWrapper);
		if (null != warningInfoEntity) {
			//恢复
			warningInfoEntity.setWarningRecoveryTime(checkTm);
			warningInfoEntity.setWarningStatus(WarningStatusEnum.STATUS_CLOSE.getCode());
			warningInfoEntity.setUpdateTime(checkTm);
			this.updateById(warningInfoEntity);
			kafkaTemplate.send(topicYwxtWarning, JsonUtil.toJson(warningInfoEntity));
		}
	}

	/**
	 * 时钟预警检测
	 *
	 * @param checkTm
	 * @param upTime
	 * @param fromDate
	 */
	private boolean clockWarnCheck(Date checkTm, Date upTime, Date fromDate) {
		//检测时钟异常,情况1，上报时间比服务器时间晚，时钟错误
		if (null != upTime && checkTm.after(upTime)) {
			long tm = (upTime.getTime() - checkTm.getTime()) / 1000L;
			if (tm > 60 * 60) {
				return true;
			}
		} else {
			//检测时钟异常,情况2，上报时间比数据库服务器（东华）时间晚，时钟错误
			if (null != fromDate && checkTm.after(fromDate)) {
				long tm = (fromDate.getTime() - checkTm.getTime()) / 1000L;
				if (tm > 60 * 60) {
					return true;
				}
			}
		}
		return false;
	}

	@Override
	public boolean checkRtuClockStatus(RtuInfoEntity rtuInfoEntity) {
		Date checkTm = new Date();
		//判断测站类型
		boolean checkSiteKind = false;
		if (rtuInfoEntity.getIsRain() != null && rtuInfoEntity.getIsRain() == 1) {
			checkSiteKind = true;
		}
		if (rtuInfoEntity.getIsRiver() != null && rtuInfoEntity.getIsRiver() == 1) {
			checkSiteKind = true;
		}
		if (rtuInfoEntity.getIsRes() != null && rtuInfoEntity.getIsRes() == 1) {
			checkSiteKind = true;
		}
		if (rtuInfoEntity.getIsGround() != null && rtuInfoEntity.getIsGround() == 1) {
			checkSiteKind = true;
		}
		if (!checkSiteKind) {
			return false;
		}
		//测站检测参数
		int warnActive = 0;
		int clockWarn = 0;
		LambdaQueryWrapper<RtuWarnSettingEntity> warnSettingWrapper = Wrappers.<RtuWarnSettingEntity>query().lambda();
		warnSettingWrapper.eq(RtuWarnSettingEntity::getIsDeleted, 0);
		List<RtuWarnSettingEntity> warnSettingEntityList = rtuWarnSettingService.list(warnSettingWrapper);
		if (null != warnSettingEntityList && warnSettingEntityList.size() > 0) {
			if (warnSettingEntityList.get(0).getWarnActive() != null) {
				warnActive = warnSettingEntityList.get(0).getWarnActive();
			}
			if (warnSettingEntityList.get(0).getWarnClockActive() != null) {
				clockWarn = warnSettingEntityList.get(0).getWarnClockActive();
			}
		}
		if (warnActive == WarningActiveEnum.ACTIVE_OPEN.getCode()) {
			if (clockWarn == WarningActiveEnum.ACTIVE_OPEN.getCode()) {
				if (rtuInfoEntity.getIsRain() != null && rtuInfoEntity.getIsRain() == 1) {
					LambdaQueryWrapper<RtuDataRainEntity> dataWrapper = Wrappers.<RtuDataRainEntity>query().lambda();
					dataWrapper.eq(RtuDataRainEntity::getIsDeleted, 0);
					dataWrapper.eq(RtuDataRainEntity::getRtuCode, rtuInfoEntity.getRtuCode());
					RtuDataRainEntity rainEntity = rtuDataRainService.getOne(dataWrapper);
					if (null != rainEntity) {
						Date upTime = rainEntity.getTm();
						Date fromDate = rainEntity.getFromTime();
						if (this.clockWarnCheck(checkTm, upTime, fromDate)) {
							//大于一个小时，测站时钟有异常，产生预警
							clockWarnHappen(rtuInfoEntity, checkTm, upTime);
							return true;
						}
					}
				}
				if (rtuInfoEntity.getIsRiver() != null && rtuInfoEntity.getIsRiver() == 1) {
					LambdaQueryWrapper<RtuDataRiverEntity> dataWrapper = Wrappers.<RtuDataRiverEntity>query().lambda();
					dataWrapper.eq(RtuDataRiverEntity::getIsDeleted, 0);
					dataWrapper.eq(RtuDataRiverEntity::getRtuCode, rtuInfoEntity.getRtuCode());
					RtuDataRiverEntity riverEntity = rtuDataRiverService.getOne(dataWrapper);
					if (null != riverEntity) {
						Date upTime = riverEntity.getTm();
						Date fromDate = riverEntity.getFromTime();
						if (this.clockWarnCheck(checkTm, upTime, fromDate)) {
							//大于一个小时，测站时钟有异常，产生预警
							clockWarnHappen(rtuInfoEntity, checkTm, upTime);
							return true;
						}
					}
				}
				if (rtuInfoEntity.getIsRes() != null && rtuInfoEntity.getIsRes() == 1) {
					LambdaQueryWrapper<RtuDataRsvrEntity> dataWrapper = Wrappers.<RtuDataRsvrEntity>query().lambda();
					dataWrapper.eq(RtuDataRsvrEntity::getIsDeleted, 0);
					dataWrapper.eq(RtuDataRsvrEntity::getRtuCode, rtuInfoEntity.getRtuCode());
					RtuDataRsvrEntity rsvrEntity = rtuDataRsvrService.getOne(dataWrapper);
					if (null != rsvrEntity) {
						Date upTime = rsvrEntity.getTm();
						Date fromDate = rsvrEntity.getFromTime();
						if (this.clockWarnCheck(checkTm, upTime, fromDate)) {
							//大于一个小时，测站时钟有异常，产生预警
							clockWarnHappen(rtuInfoEntity, checkTm, upTime);
							return true;
						}
					}
				}
				if (rtuInfoEntity.getIsGround() != null && rtuInfoEntity.getIsGround() == 1) {
					LambdaQueryWrapper<RtuDataGroundEntity> dataWrapper = Wrappers.<RtuDataGroundEntity>query().lambda();
					dataWrapper.eq(RtuDataGroundEntity::getIsDeleted, 0);
					dataWrapper.eq(RtuDataGroundEntity::getRtuCode, rtuInfoEntity.getRtuCode());
					RtuDataGroundEntity groundEntity = rtuDataGroundService.getOne(dataWrapper);
					if (null != groundEntity) {
						Date upTime = groundEntity.getTm();
						Date fromDate = groundEntity.getFromTime();
						if (this.clockWarnCheck(checkTm, upTime, fromDate)) {
							//大于一个小时，测站时钟有异常，产生预警
							clockWarnHappen(rtuInfoEntity, checkTm, upTime);
							return true;
						}
					}
				}

				clockWarnRecovery(rtuInfoEntity, checkTm);
			}
		}
		return true;
	}

	@Override
	public boolean checkRtuRainUpHourDelayStatus(EtlRainDataEntity etlRainDataEntity) {
		Date checkTm = new Date();
		//HashOperations<String,String,String> hashOperations= redisTemplate.opsForHash();
		//测站检测参数
		int warnActive = 0;
		int delayWarn = 0;
		//默认延时，120分钟
		int warnDelayDuration = 120;
		//查询延时设置值
		LambdaQueryWrapper<RtuWarnSettingEntity> warnSettingWrapper = Wrappers.<RtuWarnSettingEntity>query().lambda();
		warnSettingWrapper.eq(RtuWarnSettingEntity::getIsDeleted, 0);
		List<RtuWarnSettingEntity> warnSettingEntityList = rtuWarnSettingService.list(warnSettingWrapper);
		if (null != warnSettingEntityList && warnSettingEntityList.size() > 0) {
			if (warnSettingEntityList.get(0).getWarnActive() != null) {
				warnActive = warnSettingEntityList.get(0).getWarnActive();
			}
			if (warnSettingEntityList.get(0).getWarnHourUpDelayActive() != null) {
				delayWarn = warnSettingEntityList.get(0).getWarnHourUpDelayActive();
			}
			if (warnSettingEntityList.get(0).getWarnRainDelayHourDuration() != null) {
				warnDelayDuration = warnSettingEntityList.get(0).getWarnRainDelayHourDuration();
			}
		}
		if (warnActive == WarningActiveEnum.ACTIVE_OPEN.getCode()) {
			if (delayWarn == WarningActiveEnum.ACTIVE_OPEN.getCode()) {
				Date upTime = etlRainDataEntity.getTm();
				Date fromDate = etlRainDataEntity.getFromDate();
				if (upTime.before(fromDate)) {
					//延时时长，分
					long delayDuration = (fromDate.getTime() - upTime.getTime()) / 1000L / 60L;
					if (delayDuration >= warnDelayDuration) {
						//延时异常
						LambdaQueryWrapper<RtuWarningInfoEntity> warningWrapper = Wrappers.<RtuWarningInfoEntity>query().lambda();
						warningWrapper.eq(RtuWarningInfoEntity::getIsDeleted, 0);
						warningWrapper.eq(RtuWarningInfoEntity::getRtuCode, etlRainDataEntity.getRtuCode());
						warningWrapper.eq(RtuWarningInfoEntity::getWarningStatus, WarningStatusEnum.STATUS_HAPPEN.getCode());
						warningWrapper.eq(RtuWarningInfoEntity::getWarningKind, WarnKindEnum.WARN_RAIN_UP_HOUR_DELAY.getCode());
						RtuWarningInfoEntity warningInfoEntity = this.getOne(warningWrapper);
						if (null != warningInfoEntity) {
							warningInfoEntity.setUpdateTime(checkTm);
							this.updateById(warningInfoEntity);
							kafkaTemplate.send(topicYwxtWarning, JsonUtil.toJson(warningInfoEntity));
						} else {
							RtuWarningInfoEntity entity = new RtuWarningInfoEntity();
							entity.setRtuCode(etlRainDataEntity.getRtuCode());
							entity.setWarningHappenTime(checkTm);
							entity.setWarningStatus(WarningStatusEnum.STATUS_HAPPEN.getCode());
							entity.setWarningKind(WarnKindEnum.WARN_RAIN_UP_HOUR_DELAY.getCode());
							entity.setWarningDesc("雨情整点上报延时，上报时间：" + Func.formatDateTime(fromDate) + "," + "测站采集时间：" + Func.formatDateTime(upTime) + ",上报延时设定参数：" + Func.toStr(warnDelayDuration) + "分钟");
							this.save(entity);
							kafkaTemplate.send(topicYwxtWarning, JsonUtil.toJson(entity));
						}

					} else {


						//没有检测到预警发生，进行预警恢复操作
						LambdaQueryWrapper<RtuWarningInfoEntity> warningWrapper = Wrappers.<RtuWarningInfoEntity>query().lambda();
						warningWrapper.eq(RtuWarningInfoEntity::getIsDeleted, 0);
						warningWrapper.eq(RtuWarningInfoEntity::getRtuCode, etlRainDataEntity.getRtuCode());
						warningWrapper.eq(RtuWarningInfoEntity::getWarningStatus, WarningStatusEnum.STATUS_HAPPEN.getCode());
						warningWrapper.eq(RtuWarningInfoEntity::getWarningKind, WarnKindEnum.WARN_RAIN_UP_HOUR_DELAY.getCode());
						RtuWarningInfoEntity warningInfoEntity = this.getOne(warningWrapper);
						if (null != warningInfoEntity) {

							//恢复
							warningInfoEntity.setWarningRecoveryTime(checkTm);
							warningInfoEntity.setWarningStatus(WarningStatusEnum.STATUS_CLOSE.getCode());
							warningInfoEntity.setUpdateTime(checkTm);
							this.updateById(warningInfoEntity);
							kafkaTemplate.send(topicYwxtWarning, JsonUtil.toJson(warningInfoEntity));
						}
					}
				}
			}
		}
		return true;
	}

	@Override
	public boolean checkRtuRiverUpHourDelayStatus(EtlRiverDataEntity etlRiverDataEntity) {
		Date checkTm = new Date();
		//HashOperations<String,String,String> hashOperations= redisTemplate.opsForHash();
		//测站检测参数
		int warnActive = 0;
		int delayWarn = 0;
		//默认延时，120分钟
		int warnDelayDuration = 120;
		//查询延时设置值
		LambdaQueryWrapper<RtuWarnSettingEntity> warnSettingWrapper = Wrappers.<RtuWarnSettingEntity>query().lambda();
		warnSettingWrapper.eq(RtuWarnSettingEntity::getIsDeleted, 0);
		List<RtuWarnSettingEntity> warnSettingEntityList = rtuWarnSettingService.list(warnSettingWrapper);
		if (null != warnSettingEntityList && warnSettingEntityList.size() > 0) {
			if (warnSettingEntityList.get(0).getWarnActive() != null) {
				warnActive = warnSettingEntityList.get(0).getWarnActive();
			}
			if (warnSettingEntityList.get(0).getWarnHourUpDelayActive() != null) {
				delayWarn = warnSettingEntityList.get(0).getWarnHourUpDelayActive();
			}
			if (warnSettingEntityList.get(0).getWarnRiverDelayHourDuration() != null) {
				warnDelayDuration = warnSettingEntityList.get(0).getWarnRiverDelayHourDuration();
			}
		}
		if (warnActive == WarningActiveEnum.ACTIVE_OPEN.getCode()) {
			if (delayWarn == WarningActiveEnum.ACTIVE_OPEN.getCode()) {
				Date upTime = etlRiverDataEntity.getTm();
				Date fromDate = etlRiverDataEntity.getFromDate();
				if (upTime.before(fromDate)) {
					//延时时长，分
					long delayDuration = (fromDate.getTime() - upTime.getTime()) / 1000L / 60L;
					if (delayDuration >= warnDelayDuration) {
						//延时异常
						LambdaQueryWrapper<RtuWarningInfoEntity> warningWrapper = Wrappers.<RtuWarningInfoEntity>query().lambda();
						warningWrapper.eq(RtuWarningInfoEntity::getIsDeleted, 0);
						warningWrapper.eq(RtuWarningInfoEntity::getRtuCode, etlRiverDataEntity.getRtuCode());
						warningWrapper.eq(RtuWarningInfoEntity::getWarningStatus, WarningStatusEnum.STATUS_HAPPEN.getCode());
						warningWrapper.eq(RtuWarningInfoEntity::getWarningKind, WarnKindEnum.WARN_WL_UP_HOUR_DELAY.getCode());
						RtuWarningInfoEntity warningInfoEntity = this.getOne(warningWrapper);
						if (null != warningInfoEntity) {
							warningInfoEntity.setUpdateTime(checkTm);
							this.updateById(warningInfoEntity);
							kafkaTemplate.send(topicYwxtWarning, JsonUtil.toJson(warningInfoEntity));
						} else {
							RtuWarningInfoEntity entity = new RtuWarningInfoEntity();
							entity.setRtuCode(etlRiverDataEntity.getRtuCode());
							entity.setWarningHappenTime(checkTm);
							entity.setWarningStatus(WarningStatusEnum.STATUS_HAPPEN.getCode());
							entity.setWarningKind(WarnKindEnum.WARN_WL_UP_HOUR_DELAY.getCode());
							entity.setWarningDesc("水位整点上报延时，上报时间：" + Func.formatDateTime(fromDate) + "," + "测站采集时间：" + Func.formatDateTime(upTime) + ",上报延时设定参数：" + Func.toStr(warnDelayDuration) + "分钟");
							this.save(entity);
							kafkaTemplate.send(topicYwxtWarning, JsonUtil.toJson(entity));
						}

					} else {


						//没有检测到预警发生，进行预警恢复操作
						LambdaQueryWrapper<RtuWarningInfoEntity> warningWrapper = Wrappers.<RtuWarningInfoEntity>query().lambda();
						warningWrapper.eq(RtuWarningInfoEntity::getIsDeleted, 0);
						warningWrapper.eq(RtuWarningInfoEntity::getRtuCode, etlRiverDataEntity.getRtuCode());
						warningWrapper.eq(RtuWarningInfoEntity::getWarningStatus, WarningStatusEnum.STATUS_HAPPEN.getCode());
						warningWrapper.eq(RtuWarningInfoEntity::getWarningKind, WarnKindEnum.WARN_WL_UP_HOUR_DELAY.getCode());
						RtuWarningInfoEntity warningInfoEntity = this.getOne(warningWrapper);
						if (null != warningInfoEntity) {

							//恢复
							warningInfoEntity.setWarningRecoveryTime(checkTm);
							warningInfoEntity.setWarningStatus(WarningStatusEnum.STATUS_CLOSE.getCode());
							warningInfoEntity.setUpdateTime(checkTm);
							this.updateById(warningInfoEntity);
							kafkaTemplate.send(topicYwxtWarning, JsonUtil.toJson(warningInfoEntity));
						}
					}
				}
			}
		}
		return true;
	}

	@Override
	public boolean checkRtuRsvrUpHourDelayStatus(EtlRsvrDataEntity etlRsvrDataEntity) {
		Date checkTm = new Date();
		//HashOperations<String,String,String> hashOperations= redisTemplate.opsForHash();
		//测站检测参数
		int warnActive = 0;
		int delayWarn = 0;
		//默认延时，120分钟
		int warnDelayDuration = 120;
		//查询延时设置值
		LambdaQueryWrapper<RtuWarnSettingEntity> warnSettingWrapper = Wrappers.<RtuWarnSettingEntity>query().lambda();
		warnSettingWrapper.eq(RtuWarnSettingEntity::getIsDeleted, 0);
		List<RtuWarnSettingEntity> warnSettingEntityList = rtuWarnSettingService.list(warnSettingWrapper);
		if (null != warnSettingEntityList && warnSettingEntityList.size() > 0) {
			if (warnSettingEntityList.get(0).getWarnActive() != null) {
				warnActive = warnSettingEntityList.get(0).getWarnActive();
			}
			if (warnSettingEntityList.get(0).getWarnHourUpDelayActive() != null) {
				delayWarn = warnSettingEntityList.get(0).getWarnHourUpDelayActive();
			}
			if (warnSettingEntityList.get(0).getWarnRsvrDelayHourDuration() != null) {
				warnDelayDuration = warnSettingEntityList.get(0).getWarnRsvrDelayHourDuration();
			}
		}
		if (warnActive == WarningActiveEnum.ACTIVE_OPEN.getCode()) {
			if (delayWarn == WarningActiveEnum.ACTIVE_OPEN.getCode()) {
				Date upTime = etlRsvrDataEntity.getTm();
				Date fromDate = etlRsvrDataEntity.getFromDate();
				if (upTime.before(fromDate)) {
					//延时时长，分
					long delayDuration = (fromDate.getTime() - upTime.getTime()) / 1000L / 60L;
					if (delayDuration >= warnDelayDuration) {
						//延时异常
						LambdaQueryWrapper<RtuWarningInfoEntity> warningWrapper = Wrappers.<RtuWarningInfoEntity>query().lambda();
						warningWrapper.eq(RtuWarningInfoEntity::getIsDeleted, 0);
						warningWrapper.eq(RtuWarningInfoEntity::getRtuCode, etlRsvrDataEntity.getRtuCode());
						warningWrapper.eq(RtuWarningInfoEntity::getWarningStatus, WarningStatusEnum.STATUS_HAPPEN.getCode());
						warningWrapper.eq(RtuWarningInfoEntity::getWarningKind, WarnKindEnum.WARN_WL_UP_HOUR_DELAY.getCode());
						RtuWarningInfoEntity warningInfoEntity = this.getOne(warningWrapper);
						if (null != warningInfoEntity) {
							warningInfoEntity.setUpdateTime(checkTm);
							this.updateById(warningInfoEntity);
							kafkaTemplate.send(topicYwxtWarning, JsonUtil.toJson(warningInfoEntity));
						} else {
							RtuWarningInfoEntity entity = new RtuWarningInfoEntity();
							entity.setRtuCode(etlRsvrDataEntity.getRtuCode());
							entity.setWarningHappenTime(checkTm);
							entity.setWarningStatus(WarningStatusEnum.STATUS_HAPPEN.getCode());
							entity.setWarningKind(WarnKindEnum.WARN_WL_UP_HOUR_DELAY.getCode());
							entity.setWarningDesc("水位整点上报延时，上报时间：" + Func.formatDateTime(fromDate) + "," + "测站采集时间：" + Func.formatDateTime(upTime) + ",上报延时设定参数：" + Func.toStr(warnDelayDuration) + "分钟");
							this.save(entity);
							kafkaTemplate.send(topicYwxtWarning, JsonUtil.toJson(entity));
						}

					} else {

						//没有检测到预警发生，进行预警恢复操作
						LambdaQueryWrapper<RtuWarningInfoEntity> warningWrapper = Wrappers.<RtuWarningInfoEntity>query().lambda();
						warningWrapper.eq(RtuWarningInfoEntity::getIsDeleted, 0);
						warningWrapper.eq(RtuWarningInfoEntity::getRtuCode, etlRsvrDataEntity.getRtuCode());
						warningWrapper.eq(RtuWarningInfoEntity::getWarningStatus, WarningStatusEnum.STATUS_HAPPEN.getCode());
						warningWrapper.eq(RtuWarningInfoEntity::getWarningKind, WarnKindEnum.WARN_WL_UP_HOUR_DELAY.getCode());
						RtuWarningInfoEntity warningInfoEntity = this.getOne(warningWrapper);
						if (null != warningInfoEntity) {

							//恢复
							warningInfoEntity.setWarningRecoveryTime(checkTm);
							warningInfoEntity.setWarningStatus(WarningStatusEnum.STATUS_CLOSE.getCode());
							warningInfoEntity.setUpdateTime(checkTm);
							this.updateById(warningInfoEntity);
							kafkaTemplate.send(topicYwxtWarning, JsonUtil.toJson(warningInfoEntity));
						}
					}
				}
			}
		}
		return true;
	}

	@Override
	public boolean checkRtuRainUpDelayStatus(String rtuCode, JSONObject beforeUpInfo, JSONObject nowUpInfo) {
		Date checkTm = new Date();
		//测站检测参数
		int warnActive = 0;
		int delayWarn = 0;
		//默认延时，10分钟
		int rainWarnDelayDuration = 10;
		//查询延时设置值
		LambdaQueryWrapper<RtuWarnSettingEntity> warnSettingWrapper = Wrappers.<RtuWarnSettingEntity>query().lambda();
		warnSettingWrapper.eq(RtuWarnSettingEntity::getIsDeleted, 0);
		List<RtuWarnSettingEntity> warnSettingEntityList = rtuWarnSettingService.list(warnSettingWrapper);
		if (null != warnSettingEntityList && warnSettingEntityList.size() > 0) {
			if (warnSettingEntityList.get(0).getWarnActive() != null) {
				warnActive = warnSettingEntityList.get(0).getWarnActive();
			}
			if (warnSettingEntityList.get(0).getWarnMinUpDelayActive() != null) {
				delayWarn = warnSettingEntityList.get(0).getWarnMinUpDelayActive();
			}
			if (warnSettingEntityList.get(0).getWarnRainDelayMinDuration() != null) {
				rainWarnDelayDuration = warnSettingEntityList.get(0).getWarnRainDelayMinDuration();
			}
		}
		if (warnActive == WarningActiveEnum.ACTIVE_OPEN.getCode()) {
			if (delayWarn == WarningActiveEnum.ACTIVE_OPEN.getCode()) {
				boolean isBeforeWarn = false;
				Date upTime = Func.parseDate(beforeUpInfo.getString("tm"), "yyyy-MM-dd HH:mm:ss");
				Date fromDate = Func.parseDate(beforeUpInfo.getString("from"), "yyyy-MM-dd HH:mm:ss");
				if (upTime.before(fromDate)) {
					//延时时长，秒
					long delayDuration = (fromDate.getTime() - upTime.getTime()) / 1000L;
					if (delayDuration > rainWarnDelayDuration * 60) {
						isBeforeWarn = true;
					}
				}
				boolean isNowWarn = false;
				upTime = Func.parseDate(nowUpInfo.getString("tm"), "yyyy-MM-dd HH:mm:ss");
				fromDate = Func.parseDate(nowUpInfo.getString("from"), "yyyy-MM-dd HH:mm:ss");
				if (upTime.before(fromDate)) {
					//延时时长，秒
					long delayDuration = (fromDate.getTime() - upTime.getTime()) / 1000L;
					if (delayDuration > rainWarnDelayDuration * 60) {
						isNowWarn = true;
					}
				}
				if (isBeforeWarn && isNowWarn) {
					//延时异常
					LambdaQueryWrapper<RtuWarningInfoEntity> warningWrapper = Wrappers.<RtuWarningInfoEntity>query().lambda();
					warningWrapper.eq(RtuWarningInfoEntity::getIsDeleted, 0);
					warningWrapper.eq(RtuWarningInfoEntity::getRtuCode, rtuCode);
					warningWrapper.eq(RtuWarningInfoEntity::getWarningStatus, WarningStatusEnum.STATUS_HAPPEN.getCode());
					warningWrapper.eq(RtuWarningInfoEntity::getWarningKind, WarnKindEnum.WARN_UP_MIN_DELAY.getCode());
					RtuWarningInfoEntity warningInfoEntity = this.getOne(warningWrapper);
					if (null != warningInfoEntity) {
						//warningInfoEntity.setWarningDesc("雨情5分钟上报延时，上报时间：" + Func.formatDateTime(fromDate) + "," + "测站采集时间：" + Func.formatDateTime(upTime) + ",上报延时设定参数：" + Func.toStr(rainWarnDelayDuration) + "分钟");
						warningInfoEntity.setUpdateTime(checkTm);
						this.updateById(warningInfoEntity);
						kafkaTemplate.send(topicYwxtWarning, JsonUtil.toJson(warningInfoEntity));
					} else {
						RtuWarningInfoEntity entity = new RtuWarningInfoEntity();
						entity.setRtuCode(rtuCode);
						entity.setWarningHappenTime(checkTm);
						entity.setWarningStatus(WarningStatusEnum.STATUS_HAPPEN.getCode());
						entity.setWarningKind(WarnKindEnum.WARN_UP_MIN_DELAY.getCode());
						entity.setWarningDesc("雨情5分钟上报延时，上报时间：" + Func.formatDateTime(fromDate) + "," + "测站采集时间：" + Func.formatDateTime(upTime) + ",上报延时设定参数：" + Func.toStr(rainWarnDelayDuration) + "分钟");
						this.save(entity);
						kafkaTemplate.send(topicYwxtWarning, JsonUtil.toJson(entity));
					}
				} else {
					//没有检测到预警发生，进行预警恢复操作
					LambdaQueryWrapper<RtuWarningInfoEntity> warningWrapper = Wrappers.<RtuWarningInfoEntity>query().lambda();
					warningWrapper.eq(RtuWarningInfoEntity::getIsDeleted, 0);
					warningWrapper.eq(RtuWarningInfoEntity::getRtuCode, rtuCode);
					warningWrapper.eq(RtuWarningInfoEntity::getWarningStatus, WarningStatusEnum.STATUS_HAPPEN.getCode());
					warningWrapper.eq(RtuWarningInfoEntity::getWarningKind, WarnKindEnum.WARN_UP_MIN_DELAY.getCode());
					RtuWarningInfoEntity warningInfoEntity = this.getOne(warningWrapper);
					if (null != warningInfoEntity) {
						//恢复
						warningInfoEntity.setWarningRecoveryTime(checkTm);
						warningInfoEntity.setWarningStatus(WarningStatusEnum.STATUS_CLOSE.getCode());
						warningInfoEntity.setUpdateTime(checkTm);
						this.updateById(warningInfoEntity);
						kafkaTemplate.send(topicYwxtWarning, JsonUtil.toJson(warningInfoEntity));
					}
				}
			}
		}
		return true;
	}

	@Override
	public boolean checkRtuOfflineStatus(RtuInfoEntity rtuInfoEntity) {
		Date checkTm = new Date();
		//判断测站类型
		boolean checkSiteKind = false;
		if (rtuInfoEntity.getIsRain() != null && rtuInfoEntity.getIsRain() == 1) {
			checkSiteKind = true;
		}
		if (rtuInfoEntity.getIsRiver() != null && rtuInfoEntity.getIsRiver() == 1) {
			checkSiteKind = true;
		}
		if (rtuInfoEntity.getIsRes() != null && rtuInfoEntity.getIsRes() == 1) {
			checkSiteKind = true;
		}
		if (rtuInfoEntity.getIsGround() != null && rtuInfoEntity.getIsGround() == 1) {
			checkSiteKind = true;
		}
		if (!checkSiteKind) {
			return false;
		}
		//测站检测参数
		int warnActive = 0;
		int offlineWarn = 0;
		int rainWarnOfflineDuration = 24;
		int riverWarnOfflineDuration = 24;
		int rsvrWarnOfflineDuration = 24;
		int groundWarnOfflineDuration = 24;
		LambdaQueryWrapper<RtuWarnSettingEntity> warnSettingWrapper = Wrappers.<RtuWarnSettingEntity>query().lambda();
		warnSettingWrapper.eq(RtuWarnSettingEntity::getIsDeleted, 0);
		List<RtuWarnSettingEntity> warnSettingEntityList = rtuWarnSettingService.list(warnSettingWrapper);
		if (null != warnSettingEntityList && warnSettingEntityList.size() > 0) {
			if (warnSettingEntityList.get(0).getWarnActive() != null) {
				warnActive = warnSettingEntityList.get(0).getWarnActive();
			}
			if (warnSettingEntityList.get(0).getWarnOfflineActive() != null) {
				offlineWarn = warnSettingEntityList.get(0).getWarnOfflineActive();
			}
			if (warnSettingEntityList.get(0).getWarnRainOfflineDuration() != null) {
				rainWarnOfflineDuration = warnSettingEntityList.get(0).getWarnRainOfflineDuration();
			}
			if (warnSettingEntityList.get(0).getWarnRiverOfflineDuration() != null) {
				riverWarnOfflineDuration = warnSettingEntityList.get(0).getWarnRiverOfflineDuration();
			}
			if (warnSettingEntityList.get(0).getWarnRsvrOfflineDuration() != null) {
				rsvrWarnOfflineDuration = warnSettingEntityList.get(0).getWarnRsvrOfflineDuration();
			}
			if (warnSettingEntityList.get(0).getWarnGroundOfflineDuration() != null) {
				groundWarnOfflineDuration = warnSettingEntityList.get(0).getWarnGroundOfflineDuration();
			}
		}
		if (warnActive == WarningActiveEnum.ACTIVE_OPEN.getCode()) {
			if (offlineWarn == WarningActiveEnum.ACTIVE_OPEN.getCode()) {
				LambdaQueryWrapper<RtuStatusEntity> wrapper = Wrappers.<RtuStatusEntity>query().lambda();
				wrapper.eq(RtuStatusEntity::getIsDeleted, 0);
				wrapper.eq(RtuStatusEntity::getRtuCode, rtuInfoEntity.getRtuCode());
				RtuStatusEntity rtuStatusEntity = rtuManageService.getOne(wrapper);
				if (null != rtuStatusEntity) {
					Date tm = rtuStatusEntity.getLastUpTime();
					if (null != tm) {
						if (tm.before(checkTm)) {
							int checkCount = 0;
							int warnCount = 0;
							//只处理正常时间内的数据
							long goTm = (checkTm.getTime() - tm.getTime()) / 1000L;
							//雨量站
							if (rtuInfoEntity.getIsRain() != null && rtuInfoEntity.getIsRain() == 1) {
								checkCount += 1;
								if (goTm > rainWarnOfflineDuration * 60 * 60) {
									warnCount += 1;
								}
							}
							//河道水位站
							if (rtuInfoEntity.getIsRiver() != null && rtuInfoEntity.getIsRiver() == 1) {
								checkCount += 1;
								if (goTm > riverWarnOfflineDuration * 60 * 60) {
									warnCount += 1;
								}
							}
							//水库水位站
							if (rtuInfoEntity.getIsRes() != null && rtuInfoEntity.getIsRes() == 1) {
								checkCount += 1;
								if (goTm > rsvrWarnOfflineDuration * 60 * 60) {
									warnCount += 1;
								}
							}
							//墒情站
							if (rtuInfoEntity.getIsGround() != null && rtuInfoEntity.getIsGround() == 1) {
								checkCount += 1;
								if (goTm > groundWarnOfflineDuration * 60 * 60) {
									warnCount += 1;
								}
							}
							//所有检测对像都已经离线时，产生离线预警
							if (checkCount == warnCount) {
								//触发预警
								LambdaQueryWrapper<RtuWarningInfoEntity> warningWrapper = Wrappers.<RtuWarningInfoEntity>query().lambda();
								warningWrapper.eq(RtuWarningInfoEntity::getIsDeleted, 0);
								warningWrapper.eq(RtuWarningInfoEntity::getRtuCode, rtuInfoEntity.getRtuCode());
								warningWrapper.eq(RtuWarningInfoEntity::getWarningStatus, WarningStatusEnum.STATUS_HAPPEN.getCode());
								warningWrapper.eq(RtuWarningInfoEntity::getWarningKind, WarnKindEnum.WARN_OFFLINE.getCode());
								RtuWarningInfoEntity warningInfoEntity = this.getOne(warningWrapper);
								if (null != warningInfoEntity) {
									warningInfoEntity.setUpdateTime(checkTm);
									//warningInfoEntity.setWarningDesc("测站离线异常" + "," + "判断时间：" + Func.formatDateTime(checkTm) + ",测站最后上线时间：" + Func.formatDateTime(tm) + ",离线延时设定参数：" + Func.toStr(groundWarnOfflineDuration) + "小时");
									this.updateById(warningInfoEntity);
									kafkaTemplate.send(topicYwxtWarning, JsonUtil.toJson(warningInfoEntity));
								} else {
									RtuWarningInfoEntity entity = new RtuWarningInfoEntity();
									entity.setRtuCode(rtuInfoEntity.getRtuCode());
									entity.setWarningHappenTime(checkTm);
									entity.setWarningStatus(WarningStatusEnum.STATUS_HAPPEN.getCode());
									entity.setWarningKind(WarnKindEnum.WARN_OFFLINE.getCode());
									entity.setWarningDesc("测站离线异常" + "," + "判断时间：" + Func.formatDateTime(checkTm) + ",测站最后上线时间：" + Func.formatDateTime(tm) + ",离线延时设定参数：" + Func.toStr(groundWarnOfflineDuration) + "小时");
									this.save(entity);
									kafkaTemplate.send(topicYwxtWarning, JsonUtil.toJson(entity));
								}
							} else {
								LambdaQueryWrapper<RtuWarningInfoEntity> warningWrapper = Wrappers.<RtuWarningInfoEntity>query().lambda();
								warningWrapper.eq(RtuWarningInfoEntity::getIsDeleted, 0);
								warningWrapper.eq(RtuWarningInfoEntity::getRtuCode, rtuInfoEntity.getRtuCode());
								warningWrapper.eq(RtuWarningInfoEntity::getWarningStatus, WarningStatusEnum.STATUS_HAPPEN.getCode());
								warningWrapper.eq(RtuWarningInfoEntity::getWarningKind, WarnKindEnum.WARN_OFFLINE.getCode());
								RtuWarningInfoEntity warningInfoEntity = this.getOne(warningWrapper);
								if (null != warningInfoEntity) {
									//恢复
									warningInfoEntity.setWarningRecoveryTime(checkTm);
									warningInfoEntity.setWarningStatus(WarningStatusEnum.STATUS_CLOSE.getCode());
									warningInfoEntity.setUpdateTime(checkTm);
									this.updateById(warningInfoEntity);
									kafkaTemplate.send(topicYwxtWarning, JsonUtil.toJson(warningInfoEntity));
								}
							}
						}
					} else {
						//没有找到数据，触发预警
						LambdaQueryWrapper<RtuWarningInfoEntity> warningWrapper = Wrappers.<RtuWarningInfoEntity>query().lambda();
						warningWrapper.eq(RtuWarningInfoEntity::getIsDeleted, 0);
						warningWrapper.eq(RtuWarningInfoEntity::getRtuCode, rtuInfoEntity.getRtuCode());
						warningWrapper.eq(RtuWarningInfoEntity::getWarningStatus, WarningStatusEnum.STATUS_HAPPEN.getCode());
						warningWrapper.eq(RtuWarningInfoEntity::getWarningKind, WarnKindEnum.WARN_OFFLINE.getCode());
						RtuWarningInfoEntity warningInfoEntity = this.getOne(warningWrapper);
						if (null != warningInfoEntity) {
							warningInfoEntity.setUpdateTime(checkTm);
							this.updateById(warningInfoEntity);
							kafkaTemplate.send(topicYwxtWarning, JsonUtil.toJson(warningInfoEntity));
						} else {
							RtuWarningInfoEntity entity = new RtuWarningInfoEntity();
							entity.setRtuCode(rtuInfoEntity.getRtuCode());
							entity.setWarningHappenTime(checkTm);
							entity.setWarningStatus(WarningStatusEnum.STATUS_HAPPEN.getCode());
							entity.setWarningKind(WarnKindEnum.WARN_OFFLINE.getCode());
							entity.setWarningDesc("未检测到历史上报数据，测站离线异常" + "," + "判断时间：" + Func.formatDateTime(checkTm) + ",离线延时设定参数：" + Func.toStr(groundWarnOfflineDuration) + "小时");
							this.save(entity);
							kafkaTemplate.send(topicYwxtWarning, JsonUtil.toJson(entity));
						}
					}
				} else {
					//没有找到数据，触发预警
					LambdaQueryWrapper<RtuWarningInfoEntity> warningWrapper = Wrappers.<RtuWarningInfoEntity>query().lambda();
					warningWrapper.eq(RtuWarningInfoEntity::getIsDeleted, 0);
					warningWrapper.eq(RtuWarningInfoEntity::getRtuCode, rtuInfoEntity.getRtuCode());
					warningWrapper.eq(RtuWarningInfoEntity::getWarningStatus, WarningStatusEnum.STATUS_HAPPEN.getCode());
					warningWrapper.eq(RtuWarningInfoEntity::getWarningKind, WarnKindEnum.WARN_OFFLINE.getCode());
					RtuWarningInfoEntity warningInfoEntity = this.getOne(warningWrapper);
					if (null != warningInfoEntity) {
						warningInfoEntity.setUpdateTime(checkTm);
						this.updateById(warningInfoEntity);
						kafkaTemplate.send(topicYwxtWarning, JsonUtil.toJson(warningInfoEntity));
					} else {
						RtuWarningInfoEntity entity = new RtuWarningInfoEntity();
						entity.setRtuCode(rtuInfoEntity.getRtuCode());
						entity.setWarningHappenTime(checkTm);
						entity.setWarningStatus(WarningStatusEnum.STATUS_HAPPEN.getCode());
						entity.setWarningKind(WarnKindEnum.WARN_OFFLINE.getCode());
						entity.setWarningDesc("未检测到历史上报数据，测站离线异常" + "," + "判断时间：" + Func.formatDateTime(checkTm) + ",离线延时设定参数：" + Func.toStr(groundWarnOfflineDuration) + "小时");
						this.save(entity);
						kafkaTemplate.send(topicYwxtWarning, JsonUtil.toJson(entity));
					}
				}
			}
		}
		return true;
	}

	@Override
	public boolean checkRtuRainMissOutStatus(EtlRainDataEntity etlRainDataEntity) {
		Date checkTm = new Date();
		//测站检测参数
		int warnActive = 0;
		int missOutWarn = 0;
		LambdaQueryWrapper<RtuWarnSettingEntity> warnSettingWrapper = Wrappers.<RtuWarnSettingEntity>query().lambda();
		warnSettingWrapper.eq(RtuWarnSettingEntity::getIsDeleted, 0);
		List<RtuWarnSettingEntity> warnSettingEntityList = rtuWarnSettingService.list(warnSettingWrapper);
		if (null != warnSettingEntityList && warnSettingEntityList.size() > 0) {
			if (warnSettingEntityList.get(0).getWarnActive() != null) {
				warnActive = warnSettingEntityList.get(0).getWarnActive();
			}
			if (warnSettingEntityList.get(0).getWarnMissOutActive() != null) {
				missOutWarn = warnSettingEntityList.get(0).getWarnMissOutActive();
			}
		}
		if (warnActive == WarningActiveEnum.ACTIVE_OPEN.getCode()) {
			if (missOutWarn == WarningActiveEnum.ACTIVE_OPEN.getCode()) {
				String key = "etl.rain." + etlRainDataEntity.getRtuCode() + ".up.hour";
				ValueOperations<String, String> ops = redisTemplate.opsForValue();
				if (redisTemplate.hasKey(key)) {
					String value = ops.get(key);
					JSONArray array = JSONArray.parseArray(value);
					if (array.size() > 0) {
						log.info(" up time size is {}", array.size());
						JSONObject lastobject = array.getJSONObject(array.size() - 1);
						Date hourTime = Func.parseDate(lastobject.getString("tm"), "yyyy-MM-dd HH:mm:ss");
						Calendar lastDate = Calendar.getInstance();
						lastDate.setTime(hourTime);
						lastDate.set(Calendar.SECOND, 0);
						lastDate.set(Calendar.MINUTE, 0);
						Calendar nowDate = Calendar.getInstance();
						nowDate.setTime(etlRainDataEntity.getTm());
						nowDate.set(Calendar.SECOND, 0);
						nowDate.set(Calendar.MINUTE, 0);
						//	if (!lastDate.equals(nowDate)) {
						if (lastDate.get(Calendar.YEAR) == nowDate.get(Calendar.YEAR) && lastDate.get(Calendar.MONTH) == nowDate.get(Calendar.MONTH) && lastDate.get(Calendar.DAY_OF_MONTH) == nowDate.get(Calendar.DAY_OF_MONTH) && lastDate.get(Calendar.HOUR_OF_DAY) == nowDate.get(Calendar.HOUR_OF_DAY)) {
							log.info(" up time is no process!!");
						} else {
							if (array.size() >= 24) {
								log.info(" up time remove!!");
								array.remove(0);
							}
							log.info("add  up time !");
							//添加上报时间
							JSONObject object = new JSONObject();
							object.put("tm", Func.formatDateTime(etlRainDataEntity.getTm()));
							object.put("from", Func.formatDateTime(etlRainDataEntity.getFromDate()));
							array.add(object);
						}
						if (array.size() >= 24) {
							log.info(" up time is 24");
							int upCount = 0;
							String upTimeMessage = "";
							Date lastUpTime = etlRainDataEntity.getTm();
							nowDate = Calendar.getInstance();
							nowDate.setTime(lastUpTime);
							nowDate.set(Calendar.SECOND, 0);
							nowDate.set(Calendar.MINUTE, 0);
							for (int i = 0; i < 24; i++) {
								for (int j = 1; j <= 24; j++) {
									JSONObject object = array.getJSONObject(24 - j);
									hourTime = Func.parseDate(object.getString("tm"), "yyyy-MM-dd HH:mm:ss");
									Calendar beforeDate = Calendar.getInstance();
									beforeDate.setTime(hourTime);
									beforeDate.set(Calendar.SECOND, 0);
									beforeDate.set(Calendar.MINUTE, 0);
									if (beforeDate.get(Calendar.YEAR) == nowDate.get(Calendar.YEAR) && beforeDate.get(Calendar.MONTH) == nowDate.get(Calendar.MONTH) && beforeDate.get(Calendar.DAY_OF_MONTH) == nowDate.get(Calendar.DAY_OF_MONTH) && beforeDate.get(Calendar.HOUR_OF_DAY) == nowDate.get(Calendar.HOUR_OF_DAY)) {
										log.info(" before time {} now time {}", beforeDate.getTime(), nowDate.getTime());
										upCount++;
										if (upTimeMessage.length() > 0) {
											upTimeMessage += ",";
										}
										upTimeMessage += Func.formatDateTime(beforeDate.getTime());
										break;
									}
								}
								nowDate.add(Calendar.HOUR_OF_DAY, -1);
							}
							log.info(" find size is {}", upCount);
							//漏报的报文数量
							int missCount = 24 - upCount;
							if (missCount >= 3) {
								//触发预警
								LambdaQueryWrapper<RtuWarningInfoEntity> warningWrapper = Wrappers.<RtuWarningInfoEntity>query().lambda();
								warningWrapper.eq(RtuWarningInfoEntity::getIsDeleted, 0);
								warningWrapper.eq(RtuWarningInfoEntity::getRtuCode, etlRainDataEntity.getRtuCode());
								warningWrapper.eq(RtuWarningInfoEntity::getWarningStatus, WarningStatusEnum.STATUS_HAPPEN.getCode());
								warningWrapper.eq(RtuWarningInfoEntity::getWarningKind, WarnKindEnum.WARN_RAIN_MISS_OUT.getCode());
								RtuWarningInfoEntity warningInfoEntity = this.getOne(warningWrapper);
								if (null != warningInfoEntity) {
									warningInfoEntity.setUpdateTime(checkTm);
									//warningInfoEntity.setWarningDesc("测站雨量漏报异常,判断时间：" + Func.formatDateTime(checkTm) + ",24小时采集时间：" + upTimeMessage + ",漏报数量:" + missCount);
									this.updateById(warningInfoEntity);
									kafkaTemplate.send(topicYwxtWarning, JsonUtil.toJson(warningInfoEntity));
								} else {
									RtuWarningInfoEntity entity = new RtuWarningInfoEntity();
									entity.setRtuCode(etlRainDataEntity.getRtuCode());
									entity.setWarningHappenTime(checkTm);
									entity.setWarningStatus(WarningStatusEnum.STATUS_HAPPEN.getCode());
									entity.setWarningKind(WarnKindEnum.WARN_RAIN_MISS_OUT.getCode());
									entity.setWarningDesc("测站雨量漏报异常,判断时间：" + Func.formatDateTime(checkTm) + ",24小时采集时间：" + upTimeMessage + ",漏报数量:" + missCount);
									this.save(entity);
									kafkaTemplate.send(topicYwxtWarning, JsonUtil.toJson(entity));
								}
							} else {
								//恢复
								LambdaQueryWrapper<RtuWarningInfoEntity> warningWrapper = Wrappers.<RtuWarningInfoEntity>query().lambda();
								warningWrapper.eq(RtuWarningInfoEntity::getIsDeleted, 0);
								warningWrapper.eq(RtuWarningInfoEntity::getRtuCode, etlRainDataEntity.getRtuCode());
								warningWrapper.eq(RtuWarningInfoEntity::getWarningStatus, WarningStatusEnum.STATUS_HAPPEN.getCode());
								warningWrapper.eq(RtuWarningInfoEntity::getWarningKind, WarnKindEnum.WARN_RAIN_MISS_OUT.getCode());
								RtuWarningInfoEntity warningInfoEntity = this.getOne(warningWrapper);
								if (null != warningInfoEntity) {
									//恢复
									warningInfoEntity.setWarningRecoveryTime(checkTm);
									warningInfoEntity.setWarningStatus(WarningStatusEnum.STATUS_CLOSE.getCode());
									warningInfoEntity.setUpdateTime(checkTm);
									this.updateById(warningInfoEntity);
									kafkaTemplate.send(topicYwxtWarning, JsonUtil.toJson(warningInfoEntity));
								}
							}

						}
						//更新缓存
						ops.set(key, array.toJSONString());
					} else {
						log.info(" up time size is 0");
						JSONObject object = new JSONObject();
						object.put("tm", Func.formatDateTime(etlRainDataEntity.getTm()));
						object.put("from", Func.formatDateTime(etlRainDataEntity.getFromDate()));
						array.add(object);
						ops.set(key, array.toJSONString());
					}
				} else {
					log.info("no up time");
					JSONArray array = new JSONArray();
					JSONObject object = new JSONObject();
					object.put("tm", Func.formatDateTime(etlRainDataEntity.getTm()));
					object.put("from", Func.formatDateTime(etlRainDataEntity.getFromDate()));
					array.add(object);
					ops.set(key, array.toJSONString());
				}
			}
		}
		return true;
	}

	@Override
	public boolean checkRtuRiverMissOutStatus(EtlRiverDataEntity etlRiverDataEntity) {
		Date checkTm = new Date();
		//测站检测参数
		int warnActive = 0;
		int missOutWarn = 0;
		LambdaQueryWrapper<RtuWarnSettingEntity> warnSettingWrapper = Wrappers.<RtuWarnSettingEntity>query().lambda();
		warnSettingWrapper.eq(RtuWarnSettingEntity::getIsDeleted, 0);
		List<RtuWarnSettingEntity> warnSettingEntityList = rtuWarnSettingService.list(warnSettingWrapper);
		if (null != warnSettingEntityList && warnSettingEntityList.size() > 0) {
			if (warnSettingEntityList.get(0).getWarnActive() != null) {
				warnActive = warnSettingEntityList.get(0).getWarnActive();
			}
			if (warnSettingEntityList.get(0).getWarnMissOutActive() != null) {
				missOutWarn = warnSettingEntityList.get(0).getWarnMissOutActive();
			}
		}
		if (warnActive == WarningActiveEnum.ACTIVE_OPEN.getCode()) {
			if (missOutWarn == WarningActiveEnum.ACTIVE_OPEN.getCode()) {
				String key = "etl.river." + etlRiverDataEntity.getRtuCode() + ".up.hour";
				ValueOperations<String, String> ops = redisTemplate.opsForValue();
				if (redisTemplate.hasKey(key)) {
					String value = ops.get(key);
					JSONArray array = JSONArray.parseArray(value);
					if (array.size() > 0) {
						log.info(" up time size is {}", array.size());
						JSONObject lastobject = array.getJSONObject(array.size() - 1);
						Date hourTime = Func.parseDate(lastobject.getString("tm"), "yyyy-MM-dd HH:mm:ss");
						Calendar lastDate = Calendar.getInstance();
						lastDate.setTime(hourTime);
						lastDate.set(Calendar.SECOND, 0);
						lastDate.set(Calendar.MINUTE, 0);
						Calendar nowDate = Calendar.getInstance();
						nowDate.setTime(etlRiverDataEntity.getTm());
						nowDate.set(Calendar.SECOND, 0);
						nowDate.set(Calendar.MINUTE, 0);
						if (lastDate.get(Calendar.YEAR) == nowDate.get(Calendar.YEAR) && lastDate.get(Calendar.MONTH) == nowDate.get(Calendar.MONTH) && lastDate.get(Calendar.DAY_OF_MONTH) == nowDate.get(Calendar.DAY_OF_MONTH) && lastDate.get(Calendar.HOUR_OF_DAY) == nowDate.get(Calendar.HOUR_OF_DAY)) {
							log.info(" up time is no process!!");
						} else {
							if (array.size() >= 24) {
								log.info(" up time remove!!");
								array.remove(0);
							}
							log.info("add  up time !");
							//添加上报时间
							JSONObject object = new JSONObject();
							object.put("tm", Func.formatDateTime(etlRiverDataEntity.getTm()));
							object.put("from", Func.formatDateTime(etlRiverDataEntity.getFromDate()));
							array.add(object);
						}
						if (array.size() >= 24) {
							log.info(" up time is 24");
							int upCount = 0;
							String upTimeMessage = "";
							Date lastUpTime = etlRiverDataEntity.getTm();
							nowDate = Calendar.getInstance();
							nowDate.setTime(lastUpTime);
							nowDate.set(Calendar.SECOND, 0);
							nowDate.set(Calendar.MINUTE, 0);
							for (int i = 0; i < 24; i++) {
								for (int j = 1; j <= 24; j++) {
									JSONObject object = array.getJSONObject(24 - j);
									hourTime = Func.parseDate(object.getString("tm"), "yyyy-MM-dd HH:mm:ss");
									Calendar beforeDate = Calendar.getInstance();
									beforeDate.setTime(hourTime);
									beforeDate.set(Calendar.SECOND, 0);
									beforeDate.set(Calendar.MINUTE, 0);
									if (beforeDate.get(Calendar.YEAR) == nowDate.get(Calendar.YEAR) && beforeDate.get(Calendar.MONTH) == nowDate.get(Calendar.MONTH) && beforeDate.get(Calendar.DAY_OF_MONTH) == nowDate.get(Calendar.DAY_OF_MONTH) && beforeDate.get(Calendar.HOUR_OF_DAY) == nowDate.get(Calendar.HOUR_OF_DAY)) {
										log.info(" before time {} now time {}", beforeDate.getTime(), nowDate.getTime());
										upCount++;
										if (upTimeMessage.length() > 0) {
											upTimeMessage += ",";
										}
										upTimeMessage += Func.formatDateTime(beforeDate.getTime());
										break;
									}
								}
								nowDate.add(Calendar.HOUR_OF_DAY, -1);
							}
							log.info(" find size is {}", upCount);
							//漏报的报文数量
							int missCount = 24 - upCount;
							if (missCount >= 3) {
								//触发预警
								LambdaQueryWrapper<RtuWarningInfoEntity> warningWrapper = Wrappers.<RtuWarningInfoEntity>query().lambda();
								warningWrapper.eq(RtuWarningInfoEntity::getIsDeleted, 0);
								warningWrapper.eq(RtuWarningInfoEntity::getRtuCode, etlRiverDataEntity.getRtuCode());
								warningWrapper.eq(RtuWarningInfoEntity::getWarningStatus, WarningStatusEnum.STATUS_HAPPEN.getCode());
								warningWrapper.eq(RtuWarningInfoEntity::getWarningKind, WarnKindEnum.WARN_WL_MISS_OUT.getCode());
								RtuWarningInfoEntity warningInfoEntity = this.getOne(warningWrapper);
								if (null != warningInfoEntity) {
									warningInfoEntity.setUpdateTime(checkTm);
									//warningInfoEntity.setWarningDesc("测站水位漏报异常,判断时间：" + Func.formatDateTime(checkTm) + ",24小时采集时间：" + upTimeMessage + ",漏报数量:" + missCount);
									this.updateById(warningInfoEntity);
									kafkaTemplate.send(topicYwxtWarning, JsonUtil.toJson(warningInfoEntity));
								} else {
									RtuWarningInfoEntity entity = new RtuWarningInfoEntity();
									entity.setRtuCode(etlRiverDataEntity.getRtuCode());
									entity.setWarningHappenTime(checkTm);
									entity.setWarningStatus(WarningStatusEnum.STATUS_HAPPEN.getCode());
									entity.setWarningKind(WarnKindEnum.WARN_WL_MISS_OUT.getCode());
									entity.setWarningDesc("测站水位漏报异常,判断时间：" + Func.formatDateTime(checkTm) + ",24小时采集时间：" + upTimeMessage + ",漏报数量:" + missCount);
									this.save(entity);
									kafkaTemplate.send(topicYwxtWarning, JsonUtil.toJson(entity));
								}
							} else {
								//恢复
								LambdaQueryWrapper<RtuWarningInfoEntity> warningWrapper = Wrappers.<RtuWarningInfoEntity>query().lambda();
								warningWrapper.eq(RtuWarningInfoEntity::getIsDeleted, 0);
								warningWrapper.eq(RtuWarningInfoEntity::getRtuCode, etlRiverDataEntity.getRtuCode());
								warningWrapper.eq(RtuWarningInfoEntity::getWarningStatus, WarningStatusEnum.STATUS_HAPPEN.getCode());
								warningWrapper.eq(RtuWarningInfoEntity::getWarningKind, WarnKindEnum.WARN_WL_MISS_OUT.getCode());
								RtuWarningInfoEntity warningInfoEntity = this.getOne(warningWrapper);
								if (null != warningInfoEntity) {
									//恢复
									warningInfoEntity.setWarningRecoveryTime(checkTm);
									warningInfoEntity.setWarningStatus(WarningStatusEnum.STATUS_CLOSE.getCode());
									warningInfoEntity.setUpdateTime(checkTm);
									this.updateById(warningInfoEntity);
									kafkaTemplate.send(topicYwxtWarning, JsonUtil.toJson(warningInfoEntity));
								}
							}

						}
						//更新缓存
						ops.set(key, array.toJSONString());
					} else {
						log.info(" up time size is 0");
						JSONObject object = new JSONObject();
						object.put("tm", Func.formatDateTime(etlRiverDataEntity.getTm()));
						object.put("from", Func.formatDateTime(etlRiverDataEntity.getFromDate()));
						array.add(object);
						ops.set(key, array.toJSONString());
					}
				} else {
					log.info("no up time");
					JSONArray array = new JSONArray();
					JSONObject object = new JSONObject();
					object.put("tm", Func.formatDateTime(etlRiverDataEntity.getTm()));
					object.put("from", Func.formatDateTime(etlRiverDataEntity.getFromDate()));
					array.add(object);
					ops.set(key, array.toJSONString());
				}
			}
		}
		return true;
	}

	@Override
	public boolean checkRtuRsvrMissOutStatus(EtlRsvrDataEntity etlRsvrDataEntity) {
		Date checkTm = new Date();
		//测站检测参数
		int warnActive = 0;
		int missOutWarn = 0;
		LambdaQueryWrapper<RtuWarnSettingEntity> warnSettingWrapper = Wrappers.<RtuWarnSettingEntity>query().lambda();
		warnSettingWrapper.eq(RtuWarnSettingEntity::getIsDeleted, 0);
		List<RtuWarnSettingEntity> warnSettingEntityList = rtuWarnSettingService.list(warnSettingWrapper);
		if (null != warnSettingEntityList && warnSettingEntityList.size() > 0) {
			if (warnSettingEntityList.get(0).getWarnActive() != null) {
				warnActive = warnSettingEntityList.get(0).getWarnActive();
			}
			if (warnSettingEntityList.get(0).getWarnMissOutActive() != null) {
				missOutWarn = warnSettingEntityList.get(0).getWarnMissOutActive();
			}
		}
		if (warnActive == WarningActiveEnum.ACTIVE_OPEN.getCode()) {
			if (missOutWarn == WarningActiveEnum.ACTIVE_OPEN.getCode()) {
				String key = "etl.rsvr." + etlRsvrDataEntity.getRtuCode() + ".up.hour";
				ValueOperations<String, String> ops = redisTemplate.opsForValue();
				if (redisTemplate.hasKey(key)) {
					String value = ops.get(key);
					JSONArray array = JSONArray.parseArray(value);
					if (array.size() > 0) {
						JSONObject lastobject = array.getJSONObject(array.size() - 1);
						Date hourTime = Func.parseDate(lastobject.getString("tm"), "yyyy-MM-dd HH:mm:ss");
						Calendar lastDate = Calendar.getInstance();
						lastDate.setTime(hourTime);
						lastDate.set(Calendar.SECOND, 0);
						lastDate.set(Calendar.MINUTE, 0);
						Calendar nowDate = Calendar.getInstance();
						nowDate.setTime(etlRsvrDataEntity.getTm());
						nowDate.set(Calendar.SECOND, 0);
						nowDate.set(Calendar.MINUTE, 0);
						//	if (!lastDate.equals(nowDate)) {
						if (lastDate.get(Calendar.YEAR) == nowDate.get(Calendar.YEAR) && lastDate.get(Calendar.MONTH) == nowDate.get(Calendar.MONTH) && lastDate.get(Calendar.DAY_OF_MONTH) == nowDate.get(Calendar.DAY_OF_MONTH) && lastDate.get(Calendar.HOUR_OF_DAY) == nowDate.get(Calendar.HOUR_OF_DAY)) {
							log.info(" up time is no process!!");
						} else {
							if (array.size() >= 24) {
								array.remove(0);
							}
							//添加上报时间
							JSONObject object = new JSONObject();
							object.put("tm", Func.formatDateTime(etlRsvrDataEntity.getTm()));
							object.put("from", Func.formatDateTime(etlRsvrDataEntity.getFromDate()));
							array.add(object);
						}
						if (array.size() >= 24) {
							int upCount = 0;
							String upTimeMessage = "";
							Date lastUpTime = etlRsvrDataEntity.getTm();
							nowDate = Calendar.getInstance();
							nowDate.setTime(lastUpTime);
							nowDate.set(Calendar.SECOND, 0);
							nowDate.set(Calendar.MINUTE, 0);
							for (int i = 0; i < 24; i++) {
								for (int j = 1; j <= 24; j++) {
									JSONObject object = array.getJSONObject(24 - j);
									hourTime = Func.parseDate(object.getString("tm"), "yyyy-MM-dd HH:mm:ss");
									Calendar beforeDate = Calendar.getInstance();
									beforeDate.setTime(hourTime);
									beforeDate.set(Calendar.SECOND, 0);
									beforeDate.set(Calendar.MINUTE, 0);
									if (beforeDate.get(Calendar.YEAR) == nowDate.get(Calendar.YEAR) && beforeDate.get(Calendar.MONTH) == nowDate.get(Calendar.MONTH) && beforeDate.get(Calendar.DAY_OF_MONTH) == nowDate.get(Calendar.DAY_OF_MONTH) && beforeDate.get(Calendar.HOUR_OF_DAY) == nowDate.get(Calendar.HOUR_OF_DAY)) {
										log.info(" before time {} now time {}", beforeDate.getTime(), nowDate.getTime());
										upCount++;
										if (upTimeMessage.length() > 0) {
											upTimeMessage += ",";
										}
										upTimeMessage += Func.formatDateTime(beforeDate.getTime());
										break;
									}
								}
								nowDate.add(Calendar.HOUR_OF_DAY, -1);
							}
							log.info(" find size is {}", upCount);
							//漏报的报文数量
							int missCount = 24 - upCount;
							if (missCount >= 3) {
								//触发预警
								LambdaQueryWrapper<RtuWarningInfoEntity> warningWrapper = Wrappers.<RtuWarningInfoEntity>query().lambda();
								warningWrapper.eq(RtuWarningInfoEntity::getIsDeleted, 0);
								warningWrapper.eq(RtuWarningInfoEntity::getRtuCode, etlRsvrDataEntity.getRtuCode());
								warningWrapper.eq(RtuWarningInfoEntity::getWarningStatus, WarningStatusEnum.STATUS_HAPPEN.getCode());
								warningWrapper.eq(RtuWarningInfoEntity::getWarningKind, WarnKindEnum.WARN_WL_MISS_OUT.getCode());
								RtuWarningInfoEntity warningInfoEntity = this.getOne(warningWrapper);
								if (null != warningInfoEntity) {
									warningInfoEntity.setUpdateTime(checkTm);
									//warningInfoEntity.setWarningDesc("测站水位漏报异常,判断时间：" + Func.formatDateTime(checkTm) + ",24小时采集时间：" + upTimeMessage + ",漏报数量:" + missCount);
									this.updateById(warningInfoEntity);
									kafkaTemplate.send(topicYwxtWarning, JsonUtil.toJson(warningInfoEntity));
								} else {
									RtuWarningInfoEntity entity = new RtuWarningInfoEntity();
									entity.setRtuCode(etlRsvrDataEntity.getRtuCode());
									entity.setWarningHappenTime(checkTm);
									entity.setWarningStatus(WarningStatusEnum.STATUS_HAPPEN.getCode());
									entity.setWarningKind(WarnKindEnum.WARN_WL_MISS_OUT.getCode());
									entity.setWarningDesc("测站水位漏报异常,判断时间：" + Func.formatDateTime(checkTm) + ",24小时采集时间：" + upTimeMessage + ",漏报数量:" + missCount);
									this.save(entity);
									kafkaTemplate.send(topicYwxtWarning, JsonUtil.toJson(entity));
								}
							} else {
								//恢复
								LambdaQueryWrapper<RtuWarningInfoEntity> warningWrapper = Wrappers.<RtuWarningInfoEntity>query().lambda();
								warningWrapper.eq(RtuWarningInfoEntity::getIsDeleted, 0);
								warningWrapper.eq(RtuWarningInfoEntity::getRtuCode, etlRsvrDataEntity.getRtuCode());
								warningWrapper.eq(RtuWarningInfoEntity::getWarningStatus, WarningStatusEnum.STATUS_HAPPEN.getCode());
								warningWrapper.eq(RtuWarningInfoEntity::getWarningKind, WarnKindEnum.WARN_WL_MISS_OUT.getCode());
								RtuWarningInfoEntity warningInfoEntity = this.getOne(warningWrapper);
								if (null != warningInfoEntity) {
									//恢复
									warningInfoEntity.setWarningRecoveryTime(checkTm);
									warningInfoEntity.setWarningStatus(WarningStatusEnum.STATUS_CLOSE.getCode());
									warningInfoEntity.setUpdateTime(checkTm);
									this.updateById(warningInfoEntity);
									kafkaTemplate.send(topicYwxtWarning, JsonUtil.toJson(warningInfoEntity));
								}
							}

						}
						//更新缓存
						ops.set(key, array.toJSONString());
					} else {
						JSONObject object = new JSONObject();
						object.put("tm", Func.formatDateTime(etlRsvrDataEntity.getTm()));
						object.put("from", Func.formatDateTime(etlRsvrDataEntity.getFromDate()));
						array.add(object);
						ops.set(key, array.toJSONString());
					}
				} else {
					JSONArray array = new JSONArray();
					JSONObject object = new JSONObject();
					object.put("tm", Func.formatDateTime(etlRsvrDataEntity.getTm()));
					object.put("from", Func.formatDateTime(etlRsvrDataEntity.getFromDate()));
					array.add(object);
					ops.set(key, array.toJSONString());
				}
			}
		}
		return true;
	}

	@Override
	public boolean checkRtuOutlierValueStatus(RtuInfoEntity rtuInfoEntity) {
		Date checkTm = new Date();
		String rtuCode = rtuInfoEntity.getRtuCode();
		HashOperations<String, String, String> ops = redisTemplate.opsForHash();
		String surfkey = "surf.data." + rtuCode;
		String surfTmValue = ops.get(surfkey, "upTime");
		if (null == surfTmValue) {
			return true;
		}
		Date surfTm = Func.parse(surfTmValue, ConcurrentDateFormat.of("yyyy-MM-dd HH:mm:ss"));
		Calendar calendar = Calendar.getInstance();
		calendar.setTime(surfTm);
		calendar.set(Calendar.MINUTE, 0);
		calendar.set(Calendar.SECOND, 0);

		//surfTm = calendar.getTime();

		ListOperations<String, String> listOps = redisTemplate.opsForList();
		String hourCacheKey = "etl.rain." + rtuCode + ".hour.cache";
		if (!redisTemplate.hasKey(hourCacheKey)) {
			return true;
		}
		long size = listOps.size(hourCacheKey);
		for (int i = 0; i < size; i++) {
			String d = listOps.index(hourCacheKey, i);
			EtlRainDataEntity etlRainDataEntity = JSONObject.parseObject(d, EtlRainDataEntity.class);
			Calendar calendar2 = Calendar.getInstance();
			calendar2.setTime(etlRainDataEntity.getTm());
			calendar2.set(Calendar.MINUTE, 0);
			calendar2.set(Calendar.SECOND, 0);

			if (calendar.get(Calendar.YEAR) == calendar2.get(Calendar.YEAR) && calendar.get(Calendar.MONTH) == calendar2.get(Calendar.MONTH) && calendar.get(Calendar.DAY_OF_MONTH) == calendar2.get(Calendar.DAY_OF_MONTH) && calendar.get(Calendar.HOUR_OF_DAY) == calendar2.get(Calendar.HOUR_OF_DAY) && calendar.get(Calendar.MINUTE) == calendar2.get(Calendar.MINUTE) && calendar.get(Calendar.SECOND) == calendar2.get(Calendar.SECOND)) {

				String value1 = ops.get(surfkey, "lefttop.value");
//		String ltLon = ops.get(key,"lefttop.lon");
//		String ltLat = ops.get(key,"lefttop.lat");

				String value2 = ops.get(surfkey, "leftbottom.value");
//		String lbLon = ops.get(key,"leftbottom.lon");
//		String lbLat = ops.get(key,"leftbottom.lat");

				String value3 = ops.get(surfkey, "righttop.value");
//		String rtLon = ops.get(key,"righttop.lon");
//		String rtLat = ops.get(key,"righttop.lat");

				String value4 = ops.get(surfkey, "rightbottom.value");
//		String rbLon = ops.get(key,"rightbottom.lon");
//		String rbLat = ops.get(key,"rightbottom.lat");

				if (null != value1 && null != value2 && null != value3 && null != value4) {
					float count = Func.toFloat(value1) + Func.toFloat(value2) + Func.toFloat(value3) + Func.toFloat(value4);
					if (count > 0) {
						count = count / 4;
					}
//			log.info("测站编码{} 纬度 {} 经度{}", rtuCode, 39.044694, 101.851139);
//			log.info("网格左上角 值  {} 纬度 {} 经度 {} ", value1, ltLat, ltLon);
//			log.info("网格左下角 值  {} 纬度 {} 经度 {} ", value2, lbLat, lbLon);
//			log.info("网格右上角 值  {} 纬度 {} 经度 {} ", value3, rtLat, rtLon);
//			log.info("网格右下角 值  {} 纬度 {} 经度 {} ", value4, rbLat, rbLon);
//			log.info("网格内降雨平均值 {}", count);

					Double drp = etlRainDataEntity.getDrp();
					if (drp != null) {
						double baseValue = drp.doubleValue() - count;
						if (baseValue >= 30.0) {
							//疑似异常值
							LambdaQueryWrapper<RtuWarningInfoEntity> warningWrapper = Wrappers.<RtuWarningInfoEntity>query().lambda();
							warningWrapper.eq(RtuWarningInfoEntity::getIsDeleted, 0);
							warningWrapper.eq(RtuWarningInfoEntity::getRtuCode, etlRainDataEntity.getRtuCode());
							warningWrapper.eq(RtuWarningInfoEntity::getWarningStatus, WarningStatusEnum.STATUS_HAPPEN.getCode());
							warningWrapper.eq(RtuWarningInfoEntity::getWarningKind, WarnKindEnum.WARN_OUTLIER_VALUE.getCode());
							RtuWarningInfoEntity warningInfoEntity = this.getOne(warningWrapper);
							if (null != warningInfoEntity) {
								warningInfoEntity.setUpdateTime(checkTm);
//						String desc="测站采集雨量大于面雨量20%，疑似异常值，";
//						desc += "雨量采集时间："+Func.formatDateTime(etlRainDataEntity.getTm());
//						desc += "，时段降雨："+Func.toStr(etlRainDataEntity.getDrp());
//						desc += "，平均面雨量："+Func.toStr(count);
//						warningInfoEntity.setWarningDesc(desc);
								this.updateById(warningInfoEntity);
								kafkaTemplate.send(topicYwxtWarning, JsonUtil.toJson(warningInfoEntity));
							} else {
								RtuWarningInfoEntity entity = new RtuWarningInfoEntity();
								entity.setRtuCode(etlRainDataEntity.getRtuCode());
								entity.setWarningHappenTime(checkTm);
								entity.setWarningStatus(WarningStatusEnum.STATUS_HAPPEN.getCode());
								entity.setWarningKind(WarnKindEnum.WARN_OUTLIER_VALUE.getCode());
								String desc = "测站采集雨量大于面雨量30mm，疑似异常值，";
								desc += "雨量采集时间：" + Func.formatDateTime(etlRainDataEntity.getTm());
								desc += "，时段降雨：" + Func.toStr(etlRainDataEntity.getDrp());
								desc += "，平均面雨量：" + Func.toStr(count);
								entity.setWarningDesc(desc);
								this.save(entity);
								//触发预警
								kafkaTemplate.send(topicYwxtWarning, JsonUtil.toJson(entity));
							}
						}
					}

				} else {
					log.info("no surf info");
				}
				break;
			}
		}


		return true;
	}
}
