package com.dy.rtuMw.server.rtuData.pSdV1; import com.dy.common.mw.protocol4Mqtt.MqttSubMsg; import com.dy.common.mw.protocol4Mqtt.pSdV1.upVos.WeatherVo; import com.dy.pipIrrGlobal.pojoPr.PrStWeather; import com.dy.pipIrrGlobal.pojoRm.RmWeatherHistory; import com.dy.pipIrrGlobal.pojoRm.RmWeatherLast; import com.dy.rtuMw.server.rtuData.TaskSurpport; import com.dy.rtuMw.server.rtuData.dbSv.DbSv; import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; /** * @Author: liurunyu * @Date: 2025/6/25 11:37 * @Description */ public class TkDealWeatherSdV1 extends TaskSurpport { private static Logger log = LogManager.getLogger(TkDealWeatherSdV1.class.getName()); //类ID,一定与Tree.xml配置文件中配置一致 public static final String taskId = "TkDealWeatherSdV1"; /** * 执行节点任务 * * @param data 需要处理的数据 */ @Override public void execute(Object data) { //前面的任务已经判断了data不为空且为气象数据 MqttSubMsg msg = (MqttSubMsg) data; WeatherVo stVo = (WeatherVo) msg.vo4Up; Object[] objs = this.getTaskResults(TkPreGenObjs4WeatherSdV1.taskId); DbSv sv = (DbSv) objs[0]; PrStWeather stPo = (PrStWeather) objs[1]; try{ this.doDeal(sv, stPo, msg, stVo); }catch (Exception e){ log.error("保存气象数据时发生异常", e); } } /** * 处理上行消息数据 * @param sv 服务 * @param stPo 实体对象 * @param msg 上行的订阅消息 * @param stVo 上行的设备数据 */ private void doDeal(DbSv sv, PrStWeather stPo, MqttSubMsg msg, WeatherVo stVo) throws Exception { RmWeatherHistory hpo = this.saveHistory(sv, stPo, msg, stVo); this.saveOrUpdateLast(sv, stPo, msg, stVo, hpo); } /** * 处理上行消息数据 * @param sv 服务 * @param stPo 实体对象 * @param msg 上行的订阅消息 * @param stVo 上行的设备数据 * @param hpo 历史记录最新数据 */ private void saveOrUpdateLast(DbSv sv, PrStWeather stPo, MqttSubMsg msg, WeatherVo stVo, RmWeatherHistory hpo) throws Exception { RmWeatherLast po = sv.getRmWeatherLast(stPo.id) ; if(po == null){ po = new RmWeatherLast(); po.valueFrom(msg, stVo); po.weatherId = stPo.id ; po.lastHistoryId = hpo==null?null:hpo.id ; sv.saveRmWeatherLast(po) ; }else{ po.valueFrom(msg, stVo); po.weatherId = stPo.id ; po.lastHistoryId = hpo==null?null:hpo.id ; sv.updateRmWeatherLast(po); } } /** * 处理上行消息数据 * @param sv 服务 * @param stPo 实体对象 * @param msg 上行的订阅消息 * @param stVo 上行的设备数据 */ private RmWeatherHistory saveHistory(DbSv sv, PrStWeather stPo, MqttSubMsg msg, WeatherVo stVo) throws Exception { RmWeatherHistory po = new RmWeatherHistory(); po.valueFrom(msg, stVo); po.weatherId = stPo.id ; sv.saveRmWeatherHistory(po); return po ; } }