package com.dy.rtuMw.server.rtuData.p206V1; import com.dy.common.mw.protocol.Data; import com.dy.common.mw.protocol.p206V1.DataV1; import com.dy.common.mw.protocol.p206V1.upVos.DataCd83CloseVo; import com.dy.common.util.DateTime; import com.dy.pipIrrGlobal.pojoPr.PrController; import com.dy.pipIrrGlobal.pojoRm.RmClientAmountDay; import com.dy.pipIrrGlobal.pojoRm.RmClientAmountDayLast; import com.dy.pipIrrGlobal.pojoSe.SeClient; 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; import java.util.Date; /** * @Author: liurunyu * @Date: 2024/7/9 10:00 * @Description 农户日用水量统计 */ public class TkDealClientAmountDay extends TaskSurpport { private static final Logger log = LogManager.getLogger(TkDealClientAmountDay.class.getName()); //类ID,一定与Tree.xml配置文件中配置一致 public static final String taskId = "TkDealClientAmountDay"; /** * 执行节点任务: 取水口日用水量和漏损量 * * @param data 需要处理的数据 */ @Override public void execute(Object data) { Data d = (Data) data; DataV1 dV1 = (DataV1) d.getSubData();//前面任务已经判断不为null Object cdObj = dV1.subData; if (cdObj != null && cdObj instanceof DataCd83CloseVo) { /* 只能采用DataCd83CloseVo来计算农户日用水量的原因: 1、如果采用DataCd84Vo(开阀工作报)来计算农户日用水量,必须采用本次累计流量减上次累流量的差值作为阶段农户用水量, 然后各个阶段相加,得到本次农户用水量,这个用水量还没加到农户日用水量中,因为只有收不到DataCd83CloseVo(关阀报) 上报数据时才能加到农户用水量中,否则就会重复计算,但什么时候及能否收到DataCd83CloseVo不可预知。 2、如果农户有多张IC卡,同时在多个取水口取水,将会是更复杂的计算,许多情形更不可预测,且时间复杂度 必然很高,这样程序运行时间更长,将会造成上行数据堵塞现象。 3、所以只采用上报数据DataCd83CloseVo作为计算农户用水量的数据来源,这时必然会有漏报缺失情况发生,采用补报机制可以弥补一些。 */ /* 补报DataCd83CloseVo处理方法: 采用DataCd83CloseVo数据中的关阀时间closeDt计算出农户用水日期,不能用数据接收日期时间作为农户用水日期, 但这种计算方法可能会有重复上报情况发生,即重复计算了,如果要判断重复,必然要增加一个关阀时间数据库表,增加时间 复杂度,计算用时将会长,本系统假设只要RTU上报上来数据了,说明当前网络较好,必然能收到下行应答,不会发生重复补报。 */ Object[] objs = this.getTaskResults(TkPreGenObjs.taskId); DbSv sv = (DbSv) objs[0]; PrController controller = (PrController) objs[1]; SeClient clientVo = (SeClient)objs[3] ;//这个值对象中只有id和name会有值 if(clientVo != null && clientVo.getId() != null) try { this.doDeal(sv, clientVo, controller, d.getRtuAddr(), dV1, (DataCd83CloseVo) cdObj); } catch (Exception e) { log.error("保存取水口日用水量和漏损量数据时发生异常", e); } } } /** * 业务处理 * * @param sv 服务 * @param clientVo 农户对象 * @param controller 控制器对象 * @param rtuAddr 控制器地址 * @param dV1 上报数据 * @param cdData 上报关阀数据对象 */ private void doDeal(DbSv sv, SeClient clientVo, PrController controller, String rtuAddr, DataV1 dV1, DataCd83CloseVo cdData) throws Exception { RmClientAmountDayLast poLast = sv.getRmClientAmountLast(clientVo.getId()); if (poLast == null) { //数据库中不存在该农户的日取水量数据 //首先生成最新数据及历史数据,并先保存 poLast = this.newRmClientAmountLast(clientVo, dV1, cdData); this.newAndSaveHistoryDataDeal(sv, clientVo, dV1, cdData, poLast); sv.saveRmClientAmountLast(poLast); } else { if(DateTime.dateFrom_yyyy_MM_dd_HH_mm_ss(cdData.rtuDt).equals(poLast.rtuDtLast)){ //RTU时钟等于本地最新数据中的RTU时钟,重复上报数据,不进行任何处理 }else if(DateTime.dateFrom_yyyy_MM_dd_HH_mm_ss(cdData.rtuDt).after(poLast.rtuDtLast)){ //RTU时钟晚于本地最新数据中的RTU时钟,是新上报数据 Date upYmd = DateTime.dateFrom_yyyy_MM_dd1(cdData.closeDt);//采用关阀日期作为统计日期 if(upYmd.equals(poLast.dt)){ //同一天数据 poLast = this.updateRmClientAmountLastBySameDateNewData(clientVo, dV1, cdData, poLast); RmClientAmountDay poHistory = null ; if(poLast.lastHistoryId != null){ poHistory = sv.getRmClientAmountDay(poLast.lastHistoryId) ; } if(poHistory == null){ this.newAndSaveHistoryDataDeal(sv, clientVo, dV1, cdData, poLast); }else{ poHistory = this.updateRmClientAmountBySameDateNewData(clientVo, poHistory, poLast, dV1, cdData); sv.updateRmClientAmountDay(poHistory); } }else if(upYmd.after(poLast.dt)){ //新的日期 poLast = this.updateRmClientAmountLastByNewDateNewData(clientVo, dV1, cdData, poLast); this.newAndSaveHistoryDataDeal(sv, clientVo, dV1, cdData, poLast); }else{ //这种情况不存在 } sv.updateRmClientAmountLast(poLast); }else{ //RTU时钟早于本地最新数据中的RTU时钟,是补报数据 Date upYmd = DateTime.dateFrom_yyyy_MM_dd1(cdData.closeDt);//采用关阀日期作为统计日期 RmClientAmountDay poHistory = sv.getRmClientAmountByClientAndDate(clientVo.getId(), upYmd) ; if(poHistory == null){ //无历史数据 this.newAndSaveHistoryDataDeal(sv, clientVo, dV1, cdData, null); }else{ poHistory = this.updateOldRmClientAmountBySupplyData(clientVo, poHistory, dV1, cdData); sv.updateRmClientAmountDay(poHistory); } } } } /** * 生成新的控制器漏损日统计最新记录 * @param clientVo * @param dV1 * @param cdData * @return * @throws Exception */ private RmClientAmountDayLast newRmClientAmountLast(SeClient clientVo, DataV1 dV1, DataCd83CloseVo cdData)throws Exception { RmClientAmountDayLast po = new RmClientAmountDayLast() ; po.clientId = clientVo.getId(); po.amount = cdData.thisAmount ; po.money = cdData.thisMoney ; po.dt = DateTime.dateFrom_yyyy_MM_dd1(cdData.closeDt);//采用关阀日期作为统计日期 po.openDtLast = DateTime.dateFrom_yyyy_MM_dd_HH_mm_ss(cdData.openDt); po.closeDtLast = DateTime.dateFrom_yyyy_MM_dd_HH_mm_ss(cdData.closeDt); po.thisAmountLast = cdData.thisAmount ; po.thisMoneyLast = cdData.thisMoney ; po.thisTimeLast = cdData.thisTime ; po.rtuDtLast = DateTime.dateFrom_yyyy_MM_dd_HH_mm_ss(cdData.rtuDt); return po ; } /** * 生成新的控制器漏损日统计历史记录 * @param clientVo * @param dV1 * @param cdData * @param lastPo * @return * @throws Exception */ private RmClientAmountDay newRmClientAmountDay(SeClient clientVo, DataV1 dV1, DataCd83CloseVo cdData, RmClientAmountDayLast lastPo)throws Exception { RmClientAmountDay poHistory = new RmClientAmountDay() ; poHistory.clientId = clientVo.getId(); if(lastPo != null){ if(poHistory.amount != null){ poHistory.amount += lastPo.amount ; }else{ poHistory.amount = lastPo.amount ; } if(poHistory.money != null){ poHistory.money += lastPo.money ; }else{ poHistory.money = lastPo.money ; } poHistory.times = 1 ; poHistory.dt = lastPo.dt ; poHistory.openDtLast = lastPo.openDtLast ; poHistory.closeDtLast = lastPo.closeDtLast ; poHistory.thisAmountLast = lastPo.thisAmountLast ; poHistory.thisMoneyLast = lastPo.thisMoneyLast ; poHistory.thisTimeLast = lastPo.thisTimeLast ; poHistory.rtuDtLast = lastPo.rtuDtLast ; }else{ poHistory.amount = cdData.thisAmount ; poHistory.money = cdData.thisMoney ; poHistory.times = 1 ; poHistory.dt = DateTime.dateFrom_yyyy_MM_dd1(cdData.closeDt);//采用关阀日期作为统计日期 poHistory.openDtLast = DateTime.dateFrom_yyyy_MM_dd_HH_mm_ss(cdData.openDt); poHistory.closeDtLast = DateTime.dateFrom_yyyy_MM_dd_HH_mm_ss(cdData.closeDt); poHistory.thisAmountLast = cdData.thisAmount ; poHistory.thisMoneyLast = cdData.thisMoney ; poHistory.thisTimeLast = cdData.thisTime ; poHistory.rtuDtLast = DateTime.dateFrom_yyyy_MM_dd_HH_mm_ss(cdData.rtuDt); } return poHistory ; } /** * 保存新的历史数据记录,并把ID赋值给最新记录的 lastHistoryId * @param sv * @param clientVo * @param dV1 * @param cdData * @param lastPo * @throws Exception * @return RmIntakeAmountDay */ private RmClientAmountDay newAndSaveHistoryDataDeal(DbSv sv, SeClient clientVo, DataV1 dV1, DataCd83CloseVo cdData, RmClientAmountDayLast lastPo)throws Exception { RmClientAmountDay poHistory = this.newRmClientAmountDay(clientVo, dV1, cdData, lastPo) ; sv.saveRmClientAmountDay(poHistory); //由最新数据持有历史数据中的最新记录ID,以方便快速查询 lastPo.lastHistoryId = poHistory == null ? null: poHistory.id ; return poHistory ; } /** * 更新最新数据 * @param clientVo * @param dV1 * @param cdData * @param lastPo * @return * @throws Exception */ private RmClientAmountDayLast updateRmClientAmountLastBySameDateNewData(SeClient clientVo, DataV1 dV1, DataCd83CloseVo cdData, RmClientAmountDayLast lastPo)throws Exception { lastPo.clientId = clientVo.getId(); if(lastPo.amount != null){ lastPo.amount += cdData.thisAmount ; }else{ lastPo.amount = cdData.thisAmount ; } if(lastPo.money != null){ lastPo.money += cdData.thisMoney ; }else{ lastPo.money = cdData.thisMoney ; } lastPo.dt = DateTime.dateFrom_yyyy_MM_dd1(cdData.closeDt);//采用关阀日期作为统计日期 lastPo.openDtLast = DateTime.dateFrom_yyyy_MM_dd_HH_mm_ss(cdData.openDt); lastPo.closeDtLast = DateTime.dateFrom_yyyy_MM_dd_HH_mm_ss(cdData.closeDt); lastPo.thisAmountLast = cdData.thisAmount ; lastPo.thisMoneyLast = cdData.thisMoney ; lastPo.thisTimeLast = cdData.thisTime ; lastPo.rtuDtLast = DateTime.dateFrom_yyyy_MM_dd_HH_mm_ss(cdData.rtuDt); return lastPo ; } /** * 更新最新数据 * @param clientVo * @param dV1 * @param cdData * @param lastPo * @return * @throws Exception */ private RmClientAmountDayLast updateRmClientAmountLastByNewDateNewData(SeClient clientVo, DataV1 dV1, DataCd83CloseVo cdData, RmClientAmountDayLast lastPo)throws Exception { lastPo.clientId = clientVo.getId(); lastPo.amount = cdData.thisAmount ; lastPo.money = cdData.thisMoney ; lastPo.dt = DateTime.dateFrom_yyyy_MM_dd1(cdData.closeDt);//采用关阀日期作为统计日期 lastPo.openDtLast = DateTime.dateFrom_yyyy_MM_dd_HH_mm_ss(cdData.openDt); lastPo.closeDtLast = DateTime.dateFrom_yyyy_MM_dd_HH_mm_ss(cdData.closeDt); lastPo.thisAmountLast = cdData.thisAmount ; lastPo.thisMoneyLast = cdData.thisMoney ; lastPo.thisTimeLast = cdData.thisTime ; lastPo.rtuDtLast = DateTime.dateFrom_yyyy_MM_dd_HH_mm_ss(cdData.rtuDt); return lastPo ; } /** * 更新最新数据 * @param clientVo * @param poHistory * @param dV1 * @param cdData * @param lastPo * @return * @throws Exception */ private RmClientAmountDay updateRmClientAmountBySameDateNewData(SeClient clientVo, RmClientAmountDay poHistory, RmClientAmountDayLast lastPo, DataV1 dV1, DataCd83CloseVo cdData)throws Exception { poHistory.clientId = clientVo.getId(); if(poHistory.amount != null){ poHistory.amount += cdData.thisAmount ; }else{ poHistory.amount = cdData.thisAmount ; } if(poHistory.money != null){ poHistory.money += cdData.thisMoney ; }else{ poHistory.money = cdData.thisMoney ; } if(poHistory.times != null){ poHistory.times += 1 ; }else{ poHistory.times = 1 ; } poHistory.dt = lastPo.dt ; poHistory.openDtLast = lastPo.openDtLast ; poHistory.closeDtLast = lastPo.closeDtLast ; poHistory.thisAmountLast = lastPo.thisAmountLast ; poHistory.thisMoneyLast = lastPo.thisMoneyLast ; poHistory.thisTimeLast = lastPo.thisTimeLast ; poHistory.rtuDtLast = lastPo.rtuDtLast ; return poHistory ; } /** * 更新最新数据 * @param clientVo * @param poHistory * @param dV1 * @param cdData * @return * @throws Exception */ private RmClientAmountDay updateOldRmClientAmountBySupplyData(SeClient clientVo, RmClientAmountDay poHistory, DataV1 dV1, DataCd83CloseVo cdData)throws Exception { poHistory.clientId = clientVo.getId(); if(poHistory.amount != null){ poHistory.amount += cdData.thisAmount ; }else{ poHistory.amount = cdData.thisAmount ; } if(poHistory.money != null){ poHistory.money += cdData.thisMoney ; }else{ poHistory.money = cdData.thisMoney ; } if(poHistory.times != null){ poHistory.times += 1 ; }else{ poHistory.times = 1 ; } poHistory.dt = DateTime.dateFrom_yyyy_MM_dd1(cdData.closeDt);//采用关阀日期作为统计日期 poHistory.openDtLast = DateTime.dateFrom_yyyy_MM_dd_HH_mm_ss(cdData.openDt) ; poHistory.closeDtLast = DateTime.dateFrom_yyyy_MM_dd_HH_mm_ss(cdData.closeDt) ; poHistory.thisAmountLast = cdData.thisAmount ; poHistory.thisMoneyLast = cdData.thisMoney ; poHistory.thisTimeLast = cdData.thisTime ; poHistory.rtuDtLast = DateTime.dateFrom_yyyy_MM_dd_HH_mm_ss(cdData.rtuDt) ; return poHistory ; } }