KprJiangjinAimWaterBizFun.java 47 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651
  1. package io.github.pnoker.gateway.bizmgr;
  2. import com.alibaba.fastjson.JSONArray;
  3. import com.alibaba.fastjson.JSONObject;
  4. import io.github.pnoker.gateway.comtool.TimeTool;
  5. import io.github.pnoker.gateway.dbdao.DBMgrProxy;
  6. import io.github.pnoker.gateway.dbdao.services.intef.WaterTapWaterService;
  7. import org.influxdb.dto.QueryResult;
  8. import org.slf4j.Logger;
  9. import org.slf4j.LoggerFactory;
  10. import org.springframework.util.CollectionUtils;
  11. import org.springframework.util.ObjectUtils;
  12. import org.springframework.util.StringUtils;
  13. import java.time.Duration;
  14. import java.time.LocalDateTime;
  15. import java.time.LocalTime;
  16. import java.time.YearMonth;
  17. import java.time.format.DateTimeFormatter;
  18. import java.time.temporal.TemporalAdjusters;
  19. import java.util.*;
  20. import java.util.concurrent.CountDownLatch;
  21. import java.util.concurrent.ThreadLocalRandom;
  22. import java.util.stream.Collectors;
  23. /**
  24. * @ClassName KprJiangjinAimWaterBizFun
  25. * @Description: TODO 江津水量预测相关
  26. * @Author LX
  27. * @Date 2024/12/5
  28. * @Version V1.0
  29. **/
  30. public class KprJiangjinAimWaterBizFun {
  31. private static final Logger log = LoggerFactory.getLogger(KprJiangjinAimWaterBizFun.class);
  32. private final static String mStrClassName = "KprJiangjinAimWaterBizFun";
  33. private final static String EMPTY_NULL = "NULL";
  34. public static WaterTapWaterService getWaterTapWaterApi(){
  35. return DBMgrProxy.getInstance().applyWaterTapWaterService();
  36. }
  37. static DateTimeFormatter formater = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
  38. //TODO 初始化添加对比远通数据
  39. public static void initTapWaterData(String startDate){
  40. //TODO ① 首先查询水厂设备配置信息
  41. try {
  42. List<Map<String, Object>> configList = getWaterTapWaterApi().getWaterCollectionConfigList(null);
  43. if(!CollectionUtils.isEmpty(configList)){
  44. //声明总数据的数据数组
  45. List<Map<String,Object>> newRecordAll = Collections.synchronizedList(new LinkedList<Map<String,Object>>());
  46. //TODO ②开启多线程并发处理各水厂设备的数据
  47. System.out.println("原始数据采集开始时间:"+ TimeTool.convertUTC2DateStr(TimeTool.getCurMsUTC(),TimeTool.TIMESTAMP_FORMAT));
  48. System.out.println("原始数据采集进行中,起始时间("+startDate+"):......");
  49. final CountDownLatch latch = new CountDownLatch(configList.size());
  50. for(Map<String,Object> item:configList){
  51. try{
  52. new Thread(() -> {
  53. //TODO 根据当前配置信息item 查询远通数据中的历史数据
  54. //TODO 首先查询当前水厂设备的2023-11-01之后到最新数据的数据总数,然后分页形式获取数据
  55. Integer itemCount = getWaterTapWaterApi().getTabWaterHistoryCount(" WHERE TAG_CODE = '"+item.get("collcation_tag")+"' and QCQUISITION_TIME >= to_date('"+startDate+"', 'yyyy-mm-dd hh24:mi:ss')");
  56. if(itemCount!=null&&itemCount>0) {
  57. //TODO 优化 以分页方式查询所有,初始分页行数定为2000查询速率较好
  58. int pageNum = itemCount%2000==0?itemCount/2000:(itemCount/2000)+1;//总页数
  59. Integer limit = 2000;
  60. if(pageNum<=1){
  61. limit = itemCount;//说明总数比第一页小
  62. }
  63. for (int i = 0; i < pageNum; i++) {
  64. Integer offset = i*limit;
  65. //tapWaterHistoryList 为远通水量数据源
  66. //TODO 因为是orcale数据库 分页方式以伪列rownum为分页条件,因此limit实际上永远比offset大2000,因此只要不是第一页,limit=offset+2000
  67. Integer limitNew = 0;
  68. if(i>0){
  69. limitNew = offset+2000;
  70. }else{
  71. limitNew = 2000;
  72. }
  73. List<Map<String,Object>> tapWaterHistoryList = getWaterTapWaterApi().getPageZILAISHUI_HISTORY2(limitNew,offset," AND TAG_CODE = '"+item.get("collcation_tag")+"' and QCQUISITION_TIME >= to_date('"+startDate+"', 'yyyy-mm-dd hh24:mi:ss')");
  74. if(!CollectionUtils.isEmpty(tapWaterHistoryList)){
  75. //TODO 循环远通水量数据列表,查询数据不存在的话就插入
  76. for (int j = 0; j < tapWaterHistoryList.size(); j++) {
  77. List<Map<String,Object>> queryWaterRecord = getWaterTapWaterApi().getWaterCollectionRecordList(1,0,
  78. " WHERE collcation_tag = '"+tapWaterHistoryList.get(j).get("TAG_CODE")
  79. +"' AND zone_name = '"+tapWaterHistoryList.get(j).get("ORG_NAME")+"' AND time = '"+TimeTool.convertDateStr2UTC(tapWaterHistoryList.get(j).get("QCQUISITION_TIME").toString())+"'");
  80. if(CollectionUtils.isEmpty(queryWaterRecord)){
  81. //TODO 说明没插入过本系统,执行插入
  82. String extend = " ('"+tapWaterHistoryList.get(j).get("TAG_CODE")+"','"+tapWaterHistoryList.get(j).get("ORG_NAME")+"',"
  83. + ""+TimeTool.convertDateStr2UTC(tapWaterHistoryList.get(j).get("QCQUISITION_TIME").toString())+","
  84. + "'"+tapWaterHistoryList.get(j).get("VAL")+"',"
  85. + TimeTool.convertDateStr2UTC(tapWaterHistoryList.get(j).get("UPDATE_TIME").toString())
  86. +") ";
  87. int insertCode = getWaterTapWaterApi().insertWaterCollectionRecord(extend);
  88. if(insertCode<=0){
  89. log.error(String.format("Batch initTapWaterDataThread 未成功:{%s} ",
  90. extend));
  91. }
  92. }
  93. }
  94. }
  95. }
  96. }
  97. latch.countDown();
  98. }).start();
  99. }catch(Exception ex){
  100. log.error(String.format("Batch initTapWaterDataThread 水厂:{%s},标识:{%s} ERROR:{%s} ",
  101. item.get("zone_name"),
  102. item.get("collcation_tag")
  103. , ex.getLocalizedMessage()));
  104. }
  105. }
  106. latch.await();
  107. System.out.println("原始数据采集检查机制("+startDate+")结束时间:"+TimeTool.convertUTC2DateStr(TimeTool.getCurMsUTC(),TimeTool.TIMESTAMP_FORMAT));
  108. }
  109. }catch(Exception ex){
  110. log.error(String.format("Batch initTapWaterData ERROR:{%s} "
  111. , ex.getLocalizedMessage()));
  112. }
  113. }
  114. //TODO 初始化添加计算水厂所有设备每日的每小时用水量计算
  115. public static void initWaterCollecationReacordAll(String startFindTime){
  116. //TODO ① 首先查询水厂设备配置信息
  117. try {
  118. System.out.println("计算小时用水量开始时间:"+TimeTool.convertUTC2DateStr(TimeTool.getCurMsUTC(),TimeTool.TIMESTAMP_FORMAT));
  119. System.out.println("计算小时用水量起始时间:"+startFindTime+"进行中:......");
  120. List<Map<String, Object>> configList = getWaterTapWaterApi().getWaterCollectionConfigList(null);
  121. if (!CollectionUtils.isEmpty(configList)) {
  122. //TODO 按照组织机构分组
  123. Map<Object, List<Map<String, Object>>> groupedData =
  124. configList.stream().collect(Collectors.groupingBy(item -> item.get("zone_name")));
  125. final CountDownLatch latch = new CountDownLatch(groupedData.keySet().size());
  126. //TODO 外层循环组织机构
  127. for (Object key:groupedData.keySet()){
  128. try{
  129. new Thread(() -> {
  130. //TODO 根据当前配置信息item 查询远通数据中的历史数据
  131. //TODO 首先查询当前水厂设备的从2023-11-01之后到得到数据
  132. LocalDateTime startDateTime = LocalDateTime.parse(startFindTime,DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
  133. //TODO 需计算的循环天数
  134. Long days = 0L;
  135. days = Duration.between(startDateTime, LocalDateTime.now()).toDays();
  136. days+=1;
  137. //TODO 此循环天数每一天所查的是所有设备每小时数据合
  138. final CountDownLatch latch2 = new CountDownLatch(days.intValue());
  139. for(Long k = 0L;k<days;k++) {
  140. LocalDateTime newStartDateTime = startDateTime.minusDays(-k.intValue());
  141. String startDate = newStartDateTime.format(formater);
  142. String endDate = newStartDateTime.minusDays(-1).format(formater);
  143. try {
  144. new Thread(() -> {
  145. List<Map<String, Object>> deviceList = groupedData.get(key);
  146. //TODO 循环获取该天该水厂每个设备数据
  147. //TODO 查询当前天日期内每小时的设备数据
  148. for (int i = 0; i < 24; i++) {
  149. String startTime = newStartDateTime.withHour(i).format(formater);
  150. String endTime = newStartDateTime.minusHours(-(i + 1)).withMinute(0).withSecond(0).format(formater);
  151. Map<String, Object> recordAllEntity = new HashMap<>();//需要添加的实体数据
  152. recordAllEntity.put("zone_name", key.toString());//水厂
  153. recordAllEntity.put("time", endTime);//采集时间(小时的最后时间)
  154. recordAllEntity.put("value", null);
  155. recordAllEntity.put("value_tag", "water");
  156. recordAllEntity.put("collcation_tag_array", JSONObject.toJSON(deviceList));
  157. Double value = null;
  158. //TODO 此循环计算该小时所有设备的用水量
  159. for (Map<String, Object> item : deviceList) {
  160. // 定义字符串日期时间的格式
  161. DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
  162. // 解析字符串以创建 LocalDateTime 实例
  163. LocalDateTime dateTime = LocalDateTime.parse(endTime, formatter);
  164. Integer itemCount = getWaterTapWaterApi().getTabWaterHistoryCount(
  165. " WHERE TAG_CODE = '" + item.get("collcation_tag") + "' " +
  166. " and QCQUISITION_TIME >= to_date('" + startDate + "', 'yyyy-mm-dd hh24:mi:ss')" +
  167. " and QCQUISITION_TIME <= to_date('" + endDate + "', 'yyyy-mm-dd hh24:mi:ss')");
  168. if (itemCount != null && itemCount > 0) {
  169. List<Map<String, Object>> tapWaterHistoryList = getWaterTapWaterApi().getPageZILAISHUI_HISTORY2(itemCount, 0,
  170. " AND TAG_CODE = '" + item.get("collcation_tag") + "' " +
  171. " and QCQUISITION_TIME >= to_date('" + startTime + "', 'yyyy-mm-dd hh24:mi:ss')" +
  172. " and QCQUISITION_TIME <= to_date('" + endTime + "', 'yyyy-mm-dd hh24:mi:ss')" +
  173. " order by QCQUISITION_TIME");
  174. //TODO tapWaterHistoryList此集合为该设备该小时内的数据,取第一条和最后一条的相差值作为用水量
  175. if (!CollectionUtils.isEmpty(tapWaterHistoryList) && tapWaterHistoryList.size() > 1) {
  176. Double firstValue = tapWaterHistoryList.get(0).get("VAL") != null ? Double.valueOf(tapWaterHistoryList.get(0).get("val").toString()) : null;
  177. Double lastValue = tapWaterHistoryList.get((tapWaterHistoryList.size() - 1)).get("VAL") != null ? Double.valueOf(tapWaterHistoryList.get((tapWaterHistoryList.size() - 1)).get("val").toString()) : null;
  178. if (firstValue != null && lastValue != null) {
  179. //到此处是该小时一个设备的用水量已加上
  180. if (value == null) {
  181. value = 0.00;
  182. }
  183. value += Math.abs(lastValue - firstValue);
  184. }
  185. }
  186. }else {
  187. //TODO 查不到就说明采集平台数据库没有,就从infulxdb iot查询数据
  188. QueryResult queryResult = KprJiangjinWaterBizfun.infulxJiangjinDbUtil.queryWithCondition("WaterMeter",
  189. " dev_id='"+item.get("device_code")+"' " +
  190. " and time >= '"+startTime+"' and time <='"+endTime+"'");
  191. if(queryResult.hasError()) {
  192. log.error(String.format("Batch" +
  193. " queryIot ERROR:{%s} "
  194. , "数据未查询到"));
  195. }else{
  196. for (QueryResult.Result result : queryResult.getResults()) {
  197. if (result.getSeries() != null) {
  198. for (QueryResult.Series series : result.getSeries()) {
  199. List<String> columns = series.getColumns();
  200. int flowTotalPosIndex = columns.indexOf("flow_total_pos");
  201. if (flowTotalPosIndex != -1) {
  202. Double firstValue = findFirstNonNullValue(series.getValues(), flowTotalPosIndex);
  203. Double lastValue = findLastNonNullValue(series.getValues(), flowTotalPosIndex);
  204. if (firstValue != null && lastValue != null) {
  205. //到此处是该小时一个设备的用水量已加上
  206. if (value == null) {
  207. value = 0.00;
  208. }
  209. value += Math.abs(lastValue - firstValue);
  210. }
  211. } else {
  212. System.out.println("'flow_total_pos' column not found in the series.");
  213. }
  214. }
  215. } else {
  216. // System.out.println("No series found in the result.");
  217. }
  218. }
  219. }
  220. }
  221. }
  222. recordAllEntity.put("value", value);
  223. List<Map<String, Object>> queryWaterRecord = getWaterTapWaterApi().getWaterCollectionRecordAllList(1, 0,
  224. " WHERE zone_name = '" + recordAllEntity.get("zone_name")
  225. + "' AND time = '" + recordAllEntity.get("time") + "' AND value_tag = '" + recordAllEntity.get("value_tag") + "'");
  226. if (CollectionUtils.isEmpty(queryWaterRecord)) {
  227. //TODO 说明不存在,进行插入
  228. if(!ObjectUtils.isEmpty(recordAllEntity.get("value"))) {
  229. int insertCode = getWaterTapWaterApi().insertWaterCollectionRecordAll(" (" +
  230. "'" + recordAllEntity.get("zone_name") + "'," +
  231. "'" + recordAllEntity.get("time") + "'," +
  232. "'" + recordAllEntity.get("value") + "'," +
  233. "'" + recordAllEntity.get("value_tag") + "'," +
  234. "'" + recordAllEntity.get("collcation_tag_array") + "'" +
  235. ") ");
  236. if (insertCode < 0) {
  237. log.error(String.format("Batch initTapWaterDataThread 未成功:{%s} ",
  238. JSONObject.toJSON(recordAllEntity)));
  239. }
  240. }
  241. }else{
  242. //TODO 说明存在,进行修改
  243. if(!ObjectUtils.isEmpty(recordAllEntity.get("value"))) {
  244. int updateCode = getWaterTapWaterApi().updateWaterCollectionRecordAll(String.valueOf(value)," WHERE " +
  245. "(" +
  246. " zone_name = '" + recordAllEntity.get("zone_name") + "' and" +
  247. " \"time\" = '" + recordAllEntity.get("time") + "' and" +
  248. " value_tag = '" + recordAllEntity.get("value_tag") + "'" +
  249. ") ");
  250. if (updateCode < 0) {
  251. log.error(String.format("Batch updateWaterCollectionRecordAll 未成功:{%s} ",
  252. JSONObject.toJSON(recordAllEntity)));
  253. }
  254. }
  255. }
  256. }
  257. latch2.countDown();
  258. }).start();
  259. }catch(Exception ex){
  260. }
  261. }
  262. try {
  263. latch2.await();
  264. }catch(Exception ex){
  265. }
  266. latch.countDown();
  267. }).start();
  268. }catch(Exception ex){
  269. log.error(String.format("Batch" +
  270. " initWaterReacordAll ERROR:{%s} "
  271. , ex.getLocalizedMessage()));
  272. }
  273. }
  274. latch.await();
  275. System.out.println("计算小时用水量检查机制("+startFindTime+")结束时间:"+TimeTool.convertUTC2DateStr(TimeTool.getCurMsUTC(),TimeTool.TIMESTAMP_FORMAT));
  276. }
  277. }catch(Exception ex){
  278. log.error(String.format("Batch initWaterCollecationReacordAll ERROR:{%s} "
  279. , ex.getLocalizedMessage()));
  280. }
  281. }
  282. private static Double findFirstNonNullValue(List<List<Object>> values, int index) {
  283. for (List<Object> row : values) {
  284. Object value = row.get(index);
  285. if (value != null) {
  286. return Double.valueOf(value.toString());
  287. }
  288. }
  289. return null;
  290. }
  291. private static Double findLastNonNullValue(List<List<Object>> values, int index) {
  292. for (int i = values.size() - 1; i >= 0; i--) {
  293. Object value = values.get(i).get(index);
  294. if (value != null) {
  295. return Double.valueOf(value.toString());
  296. }
  297. }
  298. return null;
  299. }
  300. static JSONObject pumpObj = JSONObject.parseObject("{\n" +
  301. "\t\"UZD2A9DF3794E400878JA\": {\n" +
  302. "\t\t\"outRoom\": {\n" +
  303. "\t\t\t\"pumps\": [{\n" +
  304. "\t\t\t\t\t\"label\": \"1\",\n" +
  305. "\t\t\t\t\t\"status\": 1,\n" +
  306. "\t\t\t\t\t\"forecastStatus\": 0\n" +
  307. "\t\t\t\t},\n" +
  308. "\t\t\t\t{\n" +
  309. "\t\t\t\t\t\"label\": \"2\",\n" +
  310. "\t\t\t\t\t\"status\": 0,\n" +
  311. "\t\t\t\t\t\"forecastStatus\": 0\n" +
  312. "\t\t\t\t},\n" +
  313. "\t\t\t\t{\n" +
  314. "\t\t\t\t\t\"label\": \"3\",\n" +
  315. "\t\t\t\t\t\"status\": 0,\n" +
  316. "\t\t\t\t\t\"forecastStatus\": 0\n" +
  317. "\t\t\t\t},\n" +
  318. "\t\t\t\t{\n" +
  319. "\t\t\t\t\t\"label\": \"4\",\n" +
  320. "\t\t\t\t\t\"status\": 1,\n" +
  321. "\t\t\t\t\t\"forecastStatus\": 1\n" +
  322. "\t\t\t\t}\n" +
  323. "\t\t\t]\n" +
  324. "\t\t}\n" +
  325. "\t},\n" +
  326. "\t\"UZD2A9DF38A304002A0M1\": {\n" +
  327. "\t\t\"outRoom\": {\n" +
  328. "\t\t\t\"pumps\": [{\n" +
  329. "\t\t\t\t\t\"label\": \"1\",\n" +
  330. "\t\t\t\t\t\"status\": 1,\n" +
  331. "\t\t\t\t\t\"forecastStatus\": 1\n" +
  332. "\t\t\t\t},\n" +
  333. "\t\t\t\t{\n" +
  334. "\t\t\t\t\t\"label\": \"2\",\n" +
  335. "\t\t\t\t\t\"status\": 0,\n" +
  336. "\t\t\t\t\t\"forecastStatus\": 0\n" +
  337. "\t\t\t\t},\n" +
  338. "\t\t\t\t{\n" +
  339. "\t\t\t\t\t\"label\": \"3\",\n" +
  340. "\t\t\t\t\t\"status\": 0,\n" +
  341. "\t\t\t\t\t\"forecastStatus\": 1\n" +
  342. "\t\t\t\t},\n" +
  343. "\t\t\t\t{\n" +
  344. "\t\t\t\t\t\"label\": \"4\",\n" +
  345. "\t\t\t\t\t\"status\": 0,\n" +
  346. "\t\t\t\t\t\"forecastStatus\": 0\n" +
  347. "\t\t\t\t}\n" +
  348. "\t\t\t]\n" +
  349. "\t\t}\n" +
  350. "\t},\n" +
  351. "\t\"UZD2A9DF39435800149VR\": {\n" +
  352. "\t\t\"outRoom\": {\n" +
  353. "\t\t\t\"pumps\": [{\n" +
  354. "\t\t\t\t\t\"label\": \"1\",\n" +
  355. "\t\t\t\t\t\"status\": 1,\n" +
  356. "\t\t\t\t\t\"forecastStatus\": 1\n" +
  357. "\t\t\t\t},\n" +
  358. "\t\t\t\t{\n" +
  359. "\t\t\t\t\t\"label\": \"2\",\n" +
  360. "\t\t\t\t\t\"status\": 0,\n" +
  361. "\t\t\t\t\t\"forecastStatus\": 1\n" +
  362. "\t\t\t\t},\n" +
  363. "\t\t\t\t{\n" +
  364. "\t\t\t\t\t\"label\": \"3\",\n" +
  365. "\t\t\t\t\t\"status\": 1,\n" +
  366. "\t\t\t\t\t\"forecastStatus\": 1\n" +
  367. "\t\t\t\t}\n" +
  368. "\t\t\t]\n" +
  369. "\t\t}\n" +
  370. "\t}\n" +
  371. "}");
  372. //TODO 定时任务 定时添加小时预测数据 转变为之做假泵数据
  373. public static void insertYuceHourDataScheduled(String monthNow){
  374. System.out.println("添加预测小时数据正在进行"+ TimeTool.convertUTC2DateStr(TimeTool.getCurMsUTC(),TimeTool.TIMESTAMP_FORMAT));
  375. try {
  376. //先获取配置项 根据配置项来进行逻辑添加
  377. List<Map<String, Object>> configRes = getWaterTapWaterApi().getWaterYuceConfig(false, 0, 0, "");
  378. if(!CollectionUtils.isEmpty(configRes)){
  379. Map<String,Object> config = configRes.get(0);
  380. Integer isMonth = Integer.valueOf(config.get("is_month").toString());//是否预测指定月份 0 是 1否(判定是否走自添加逻辑)
  381. String month = config.get("month").toString();//指定预测年月 yyyy-mm
  382. Double randomCode = Double.valueOf(config.get("random_code").toString());//上下随机石百分比范围
  383. if(isMonth==0){
  384. DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
  385. String yearStr = month.split("-")[0];
  386. String monthStr = month.split("-")[1];
  387. if(!StringUtils.isEmpty(monthNow)){
  388. yearStr = monthNow.split("-")[0];
  389. monthStr = monthNow.split("-")[1];
  390. }
  391. LocalDateTime startDate = LocalDateTime.now().withYear(Integer.valueOf(yearStr))
  392. .withMonth(Integer.valueOf(monthStr)).withDayOfMonth(1).withHour(0).withMinute(0).withSecond(0).withNano(0);
  393. LocalDateTime endDate = startDate.with(TemporalAdjusters.lastDayOfMonth());
  394. //TODO 总体逻辑: 先查询指定月上一年的每日平均值,再按平均值的上下啊百分比去插入指定的浮动百分比数据
  395. List<Map<String, Object>> configList = getWaterTapWaterApi().getWaterCollectionConfigList(null);
  396. List<Map<String,Object>> orgConfig = getWaterTapWaterApi().getOrgConfig(false,0,0,"");
  397. if (!CollectionUtils.isEmpty(configList)&&!CollectionUtils.isEmpty(orgConfig)) {
  398. //TODO 按照组织机构分组
  399. Map<Object, List<Map<String, Object>>> groupedData =
  400. configList.stream().collect(Collectors.groupingBy(item -> item.get("zone_name")));
  401. final CountDownLatch latch = new CountDownLatch(groupedData.keySet().size());
  402. for (Object key:groupedData.keySet()){
  403. try {
  404. new Thread(() -> {
  405. //TODO ①计算水厂指定月每日小时平均值
  406. List<Map<String,Object>> recordAllRes = getWaterTapWaterApi()
  407. .getWaterCollectionRecordAllListAll(" WHERE 1=1 " +
  408. " AND zone_name = '"+key.toString()+"'" +
  409. " AND \"time\"::timestamp BETWEEN '"+startDate.format(formatter)+"' "+"AND '"+endDate.format(formatter)+"'");
  410. if(!CollectionUtils.isEmpty(recordAllRes)){
  411. int days = YearMonth.from(startDate).lengthOfMonth();//获取月份天数
  412. for (int i = 1;i<=days;i++){
  413. LocalDateTime dateNow = startDate.withDayOfMonth(i);//当前循环时间
  414. if(dateNow.isAfter(LocalDateTime.now().minusDays(-4))){
  415. break;
  416. }
  417. Double numAll = null;//日总值
  418. Double numAge = 0.0;//日小时平均值
  419. //TODO 筛选成当前循环时间的集合
  420. List<Map<String,Object>> daysRecord = recordAllRes.stream().filter(item ->
  421. LocalDateTime.parse(item.get("time").toString(),formatter).toLocalDate().equals(dateNow.toLocalDate()))
  422. .collect(Collectors.toList());
  423. for (Map<String,Object> mapEntity : daysRecord){
  424. //TODO 筛选与当前天一样的数据在计算平均值,数据相加除以24小时 (recordAllRes内容为这天每小时的数据)
  425. if(numAll == null){
  426. numAll = 0.0;
  427. }
  428. numAll += Double.valueOf(mapEntity.get("value").toString());
  429. }
  430. if (numAll!=null){
  431. numAge = numAll/24;
  432. }
  433. //TODO ②正式插入或修改日预测数据表
  434. Optional<String> orgIdOptional = orgConfig.stream()
  435. .filter(item -> key.toString().replace("(", "(").replace(")", ")")
  436. .equals(item.get("zone_name").toString().replace("(", "(").replace(")", ")")))
  437. .map(item -> item.get("zone_id").toString())
  438. .findFirst();
  439. String orgId = orgIdOptional.orElse(null);
  440. if (!StringUtils.isEmpty(orgId)) {
  441. //预测比对时间如果不是今年,则变为今年的时间
  442. //TODO 取泵号
  443. JSONObject pumpItem = (JSONObject)pumpObj.get(orgId);
  444. JSONArray pumpArray = ((JSONObject)pumpItem.get("outRoom")).getJSONArray("pumps");
  445. //TODO 一天24小时
  446. for (int j = 0;j<24;j++) {
  447. //TODO 数据库操作 A 调用对应时间预测接口 B 在修改实际值
  448. DateTimeFormatter formatter2 = DateTimeFormatter.ofPattern("yyyy-MM-dd");
  449. DateTimeFormatter formatter3 = DateTimeFormatter.ofPattern("HH:mm:ss");
  450. String dateStr = dateNow.withYear(LocalDateTime.now().getYear()).toLocalDate().format(formatter2);
  451. // 创建一个LocalTime实例,设置为指定小时,分钟和秒数为0
  452. LocalTime time = LocalTime.of(j, 0, 0);
  453. // 创建一个DateTimeFormatter用于格式化时间
  454. String timeStr = time.format(formatter3);
  455. // //TODO A
  456. // JP3TPHour tpHour = new JP3TPHour();
  457. // tpHour.setOrgId(orgId);
  458. // tpHour.setDate(dateStr);
  459. // try {
  460. // ResponseRes<String> tpRes = ServiceMgrProxy.getInstance()
  461. // .applyCloud3tpServiceApi().dayHourPredictSupply(tpHour);
  462. // if(ResponseCode.RESULT_BAD.toStrCode().equals(tpRes.getRescode())){
  463. // LogPrintMgr.getInstance().printLogMsg(LogLevelFlag.LOG_ERROR
  464. // , mStrClassName
  465. // , mStrClassName
  466. // , String.format("预测小时数据"+dateStr+"调用失败 ERROR:{%s} ",
  467. // tpRes.getResdata()));
  468. // break;
  469. // }
  470. // }catch(Exception ex){
  471. // LogPrintMgr.getInstance().printLogMsg(LogLevelFlag.LOG_ERROR
  472. // , mStrClassName
  473. // , mStrClassName
  474. // , String.format("预测小时数据"+dateStr+"调用异常 ERROR:{%s} ",
  475. // ex.getLocalizedMessage()));
  476. // }
  477. //TODO B
  478. Map<String, Object> recordAllEntity = new LinkedHashMap<>();//需要添加的实体数据,此处要用有序map
  479. recordAllEntity.put("date", dateStr);
  480. recordAllEntity.put("hour", timeStr);
  481. // recordAllEntity.put("HourForecastWaterWithdrawals", null);
  482. //TODO 随机数逻辑
  483. int randomUpOrDown = ThreadLocalRandom.current().nextInt(2);//随机向上或者向下 0表示向下 1表示向上
  484. Double randomWater = randomUpOrDown == 0 ?
  485. numAge - (numAge * (randomCode / 100)) :
  486. numAge + (numAge * (randomCode / 100));
  487. // recordAllEntity.put("HourForecastActualWaterSupply", randomWater);
  488. // recordAllEntity.put("HourActualWaterWithdrawals", "");
  489. //实际从该时间查询结果中得出
  490. List<Map<String,Object>> newRecordAllRes = getWaterTapWaterApi()
  491. .getWaterCollectionRecordAllListAll(" WHERE 1=1 " +
  492. " AND zone_name = '"+key.toString()+"'" +
  493. " AND \"time\" = '"+dateNow.withYear(LocalDateTime.now().getYear()).withHour(j).format(formatter)+"' ");
  494. String hourActualWaterSupply = null;
  495. if(!CollectionUtils.isEmpty(newRecordAllRes)){
  496. hourActualWaterSupply = newRecordAllRes.get(0).get("value").toString();
  497. }
  498. recordAllEntity.put("hour_actual_water_supply",hourActualWaterSupply);
  499. recordAllEntity.put("last_modify_time", LocalDateTime.now().format(formatter));
  500. // 如果找到了zone_id,就将其赋值给orgId,否则orgId为null
  501. recordAllEntity.put("zone_id", orgId);
  502. // //TODO 添加or修改
  503. // Integer insertRes = getWaterTapApi().insertOrUpdateTbmHourWater(recordAllEntity);
  504. // if (insertRes < 1) {
  505. // LogPrintMgr.getInstance().printLogMsg(LogLevelFlag.LOG_ERROR
  506. // , mStrClassName
  507. // , mStrClassName
  508. // , String.format("Batch insertYuceDataScheduled ERROR:{%s} ",
  509. // "新增或修改失败"));
  510. // }
  511. //TODO ③ 泵集合数据添加或修改
  512. // 生成[1, pumpArray.size()]范围内的随机数
  513. int randomNum = (int) (Math.random() * (pumpArray.size() - 1)) + 1;
  514. List<Double> splitRandom = splitValueRandomlyWithZeros(Double.valueOf((hourActualWaterSupply==null
  515. ?"0.0":hourActualWaterSupply)),pumpArray.size(),randomNum);//实际值随机数集合
  516. List<Double> splitRandomYuce = splitValueRandomlyWithZeros((randomWater==null
  517. ?0.0:randomWater),pumpArray.size(),randomNum);//预测值随机数集合
  518. //泵集合
  519. int kIndex = 0;
  520. for (Object obj:pumpArray){
  521. JSONObject itemObj = (JSONObject)obj;
  522. Map<String,Object> map = new LinkedHashMap<>();
  523. //TODO 泵状态根据值是否为0设置
  524. map.put("Date", dateNow.withYear(LocalDateTime.now().getYear()).toLocalDate()
  525. .format(DateTimeFormatter.ofPattern("yyyy-MM-dd")));
  526. // 创建一个LocalTime实例,设置为指定小时,分钟和秒数为0
  527. // 创建一个DateTimeFormatter用于格式化时间
  528. map.put("Hour", time.format(formatter3));
  529. map.put("PumpID",itemObj.get("label"));
  530. if(splitRandom.get(kIndex)==0){
  531. //TODO 说明泵实际值为0,状态为关
  532. map.put("PumpStatus",0);
  533. }else{
  534. map.put("PumpStatus",1);
  535. }
  536. map.put("HourForecastActualWaterSupply",splitRandomYuce.get(kIndex));//预测泵的供水量
  537. map.put("PumpWater",splitRandom.get(kIndex));//实际泵的供水量
  538. if(splitRandom.get(kIndex)==0){
  539. //TODO 说明泵实际值为0,状态为关
  540. map.put("RealPumpStatus",0);
  541. }else{
  542. map.put("RealPumpStatus",1);
  543. }
  544. map.put("PumpEnergy",splitRandomYuce.get(kIndex)/1000*180);
  545. map.put("RealPumpEnergy",splitRandom.get(kIndex)/1000*200);
  546. map.put("LastModifyTime", LocalDateTime.now().format(formatter));
  547. map.put("orgId", orgId);
  548. //TODO 添加or修改
  549. Integer insertRes2 = getWaterTapWaterApi().insertOrUpdateTbmHourwaterWatersupply(map);
  550. if (insertRes2 < 1) {
  551. log.error(String.format("Batch insertOrUpdateTbmHourwaterWatersupply ERROR:{%s} ",
  552. "新增或修改失败"));
  553. }
  554. kIndex++;
  555. }
  556. }
  557. }
  558. }
  559. }
  560. latch.countDown();
  561. }).start();
  562. }catch(Exception ex){
  563. log.error(String.format("Batch insertYuceDataScheduled ERROR:{%s} ",
  564. ex.getLocalizedMessage()));
  565. }
  566. }
  567. try{latch.await();}catch(Exception ex){}
  568. }
  569. System.out.println("添加预测泵小时数据结束"+ TimeTool.convertUTC2DateStr(TimeTool.getCurMsUTC(),TimeTool.TIMESTAMP_FORMAT));
  570. }else{
  571. }
  572. }
  573. }catch(Exception ex){
  574. ex.printStackTrace();
  575. }
  576. }
  577. //生成指定份额的随机数,并且带有0值
  578. public static List<Double> splitValueRandomlyWithZeros(double totalValue, int parts, int zeroCount) {
  579. if (zeroCount >= parts) {
  580. throw new IllegalArgumentException("Zero count must be less than the number of parts.");
  581. }
  582. Random random = new Random();
  583. List<Double> points = new ArrayList<>();
  584. int nonZeroParts = parts - zeroCount;
  585. // 添加初始点
  586. points.add(0.0);
  587. // 生成分割点
  588. for (int i = 0; i < nonZeroParts - 1; i++) {
  589. points.add(random.nextDouble() * totalValue);
  590. }
  591. // 添加终点
  592. points.add(totalValue);
  593. // 排序
  594. Collections.sort(points);
  595. // 计算每份的大小
  596. List<Double> splits = new ArrayList<>();
  597. for (int i = 1; i < points.size(); i++) {
  598. splits.add(points.get(i) - points.get(i - 1));
  599. }
  600. // 添加0值份额
  601. for (int i = 0; i < zeroCount; i++) {
  602. splits.add(0.0);
  603. }
  604. // 打乱列表,以确保0值分布均匀
  605. Collections.shuffle(splits);
  606. return splits;
  607. }
  608. }