Administrator
2024-07-16 ee711d38f66222915bc81332c98af677fa5cd96c
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
package com.dy.rtuMw.server.rtuData.p206V1_0_0;
 
import com.dy.common.mw.protocol.Data;
import com.dy.common.mw.protocol.p206V1_0_0.DataV1_0_1;
import com.dy.common.mw.protocol.p206V1_0_0.upVos.DataCd83CloseVo;
import com.dy.common.mw.protocol.p206V1_0_0.upVos.DataCd83OpenVo;
import com.dy.common.mw.protocol.p206V1_0_0.upVos.DataCd84Vo;
import com.dy.common.mw.protocol.p206V1_0_0.upVos.DataCdC0Vo;
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.pojoRm.RmIntakeAmountDay;
import com.dy.pipIrrGlobal.pojoRm.RmIntakeAmountDayLast;
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_0_1 dV1_0_1 = (DataV1_0_1) d.getSubData();//前面任务已经判断不为null
        Object cdObj = dV1_0_1.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_0_1, (DataCd83CloseVo) cdObj);
            } catch (Exception e) {
                log.error("保存取水口日用水量和漏损量数据时发生异常", e);
            }
        }
    }
 
    /**
     * 业务处理
     *
     * @param sv         服务
     * @param clientVo   农户对象
     * @param controller 控制器对象
     * @param rtuAddr    控制器地址
     * @param dV1_0_1    上报数据
     * @param cdData     上报关阀数据对象
     */
    private void doDeal(DbSv sv, SeClient clientVo, PrController controller, String rtuAddr, DataV1_0_1 dV1_0_1, DataCd83CloseVo cdData) throws Exception {
        RmClientAmountDayLast poLast = sv.getRmClientAmountLast(clientVo.getId());
        if (poLast == null) {
            //数据库中不存在该农户的日取水量数据
            //首先生成最新数据及历史数据,并先保存
            poLast = this.newRmClientAmountLast(clientVo, dV1_0_1, cdData);
            this.newAndSaveHistoryDataDeal(sv, clientVo, dV1_0_1, 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_0_1, cdData, poLast);
                    RmClientAmountDay poHistory = null ;
                    if(poLast.lastHistoryId != null){
                        poHistory = sv.getRmClientAmountDay(poLast.lastHistoryId) ;
                    }
                    if(poHistory == null){
                        this.newAndSaveHistoryDataDeal(sv, clientVo, dV1_0_1, cdData, poLast);
                    }else{
                        poHistory = this.updateRmClientAmountBySameDateNewData(clientVo, poHistory, poLast, dV1_0_1, cdData);
                        sv.updateRmClientAmountDay(poHistory);
                    }
                }else if(upYmd.after(poLast.dt)){
                    //新的日期
                    poLast = this.updateRmClientAmountLastByNewDateNewData(clientVo, dV1_0_1, cdData, poLast);
                    this.newAndSaveHistoryDataDeal(sv, clientVo, dV1_0_1, 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_0_1, cdData, null);
                }else{
                    poHistory = this.updateOldRmClientAmountBySupplyData(clientVo, poHistory, dV1_0_1, cdData);
                    sv.updateRmClientAmountDay(poHistory);
                }
            }
        }
    }
 
 
    /**
     * 生成新的控制器漏损日统计最新记录
     * @param clientVo
     * @param dV1_0_1
     * @param cdData
     * @return
     * @throws Exception
     */
    private RmClientAmountDayLast newRmClientAmountLast(SeClient clientVo, DataV1_0_1 dV1_0_1, 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_0_1
     * @param cdData
     * @param lastPo
     * @return
     * @throws Exception
     */
    private RmClientAmountDay newRmClientAmountDay(SeClient clientVo, DataV1_0_1 dV1_0_1, 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.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.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_0_1
     * @param cdData
     * @param lastPo
     * @throws Exception
     * @return RmIntakeAmountDay
     */
    private RmClientAmountDay newAndSaveHistoryDataDeal(DbSv sv, SeClient clientVo, DataV1_0_1 dV1_0_1, DataCd83CloseVo cdData, RmClientAmountDayLast lastPo)throws Exception {
        RmClientAmountDay poHistory = this.newRmClientAmountDay(clientVo, dV1_0_1, cdData, lastPo) ;
        sv.saveRmClientAmountDay(poHistory);
        //由最新数据持有历史数据中的最新记录ID,以方便快速查询
        lastPo.lastHistoryId = poHistory == null ? null: poHistory.id ;
        return poHistory ;
    }
 
    /**
     * 更新最新数据
     * @param clientVo
     * @param dV1_0_1
     * @param cdData
     * @param lastPo
     * @return
     * @throws Exception
     */
    private RmClientAmountDayLast updateRmClientAmountLastBySameDateNewData(SeClient clientVo, DataV1_0_1 dV1_0_1, 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_0_1
     * @param cdData
     * @param lastPo
     * @return
     * @throws Exception
     */
    private RmClientAmountDayLast updateRmClientAmountLastByNewDateNewData(SeClient clientVo, DataV1_0_1 dV1_0_1, 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_0_1
     * @param cdData
     * @param lastPo
     * @return
     * @throws Exception
     */
    private RmClientAmountDay updateRmClientAmountBySameDateNewData(SeClient clientVo, RmClientAmountDay poHistory, RmClientAmountDayLast lastPo, DataV1_0_1 dV1_0_1, 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 ;
        }
        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_0_1
     * @param cdData
     * @return
     * @throws Exception
     */
    private RmClientAmountDay updateOldRmClientAmountBySupplyData(SeClient clientVo, RmClientAmountDay poHistory, DataV1_0_1 dV1_0_1, 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 ;
        }
        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 ;
    }
 
 
 
 
}