|
|
@@ -0,0 +1,165 @@
|
|
|
+package org.jeecg.modules.watch;
|
|
|
+
|
|
|
+import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
|
|
|
+import lombok.extern.slf4j.Slf4j;
|
|
|
+import org.jeecg.common.util.DateUtils;
|
|
|
+import org.jeecg.modules.productionConsume.entity.ProductionConsume;
|
|
|
+import org.jeecg.modules.productionConsume.entity.ProductionDevice;
|
|
|
+import org.jeecg.modules.productionConsume.mapper.ProductionConsumeMapper;
|
|
|
+import org.jeecg.modules.productionConsume.mapper.ProductionDeviceMapper;
|
|
|
+import org.jeecg.modules.utils.PostgreSQLUtil;
|
|
|
+import org.jeecg.modules.utils.ProductPostgreSQLUtil;
|
|
|
+import org.springframework.beans.factory.annotation.Autowired;
|
|
|
+import org.springframework.scheduling.annotation.Async;
|
|
|
+import org.springframework.scheduling.annotation.EnableAsync;
|
|
|
+import org.springframework.scheduling.annotation.Scheduled;
|
|
|
+import org.springframework.stereotype.Component;
|
|
|
+
|
|
|
+import java.math.BigDecimal;
|
|
|
+import java.math.RoundingMode;
|
|
|
+import java.sql.Connection;
|
|
|
+import java.sql.PreparedStatement;
|
|
|
+import java.sql.ResultSet;
|
|
|
+import java.sql.Timestamp;
|
|
|
+import java.util.*;
|
|
|
+import java.util.stream.Collectors;
|
|
|
+
|
|
|
+@Slf4j
|
|
|
+@EnableAsync
|
|
|
+@Component
|
|
|
+public class ProductPostgreSQLWatch {
|
|
|
+
|
|
|
+ @Autowired
|
|
|
+ ProductionConsumeMapper productionConsumeMapper;
|
|
|
+
|
|
|
+ @Autowired
|
|
|
+ ProductionDeviceMapper productionDeviceMapper;
|
|
|
+
|
|
|
+ @Async
|
|
|
+ @Scheduled(fixedDelay = 1000 * 60 * 10)
|
|
|
+ public void getProductionData() {
|
|
|
+ Connection connection = null;
|
|
|
+ PreparedStatement statement = null;
|
|
|
+ ResultSet resultSet = null;
|
|
|
+ log.info("抓取生产消耗数据");
|
|
|
+ try {
|
|
|
+ // 建立链接
|
|
|
+ connection = ProductPostgreSQLUtil.getConnection();
|
|
|
+ ;
|
|
|
+ boolean isClosed = connection.isClosed();
|
|
|
+ LambdaQueryWrapper<ProductionDevice> deviceLambdaQueryWrapper = new LambdaQueryWrapper<>();
|
|
|
+ List<ProductionDevice> devices = productionDeviceMapper.selectList(deviceLambdaQueryWrapper);
|
|
|
+ // 遍历所有设备
|
|
|
+ for (ProductionDevice device : devices) {
|
|
|
+ // 查询 productionConsume 表中 该设备下的createTime 最新的一条记录
|
|
|
+ LambdaQueryWrapper<ProductionConsume> productionLambdaQueryWrapper = new LambdaQueryWrapper<>();
|
|
|
+ productionLambdaQueryWrapper.eq(ProductionConsume::getDeviceTitle, device.getDeviceTitle());
|
|
|
+ productionLambdaQueryWrapper
|
|
|
+ .orderByDesc(ProductionConsume::getCreateTime) // 按 createTime 降序排序
|
|
|
+ .last("LIMIT 1"); // 只取一条记录
|
|
|
+ // 执行查询
|
|
|
+ ProductionConsume latestData = productionConsumeMapper.selectOne(productionLambdaQueryWrapper);
|
|
|
+ // 获取最新记录的 createTime
|
|
|
+ Date latestCreateTime1 = null;
|
|
|
+ if (latestData != null) {
|
|
|
+ latestCreateTime1 = latestData.getCreateTime();
|
|
|
+ } else {
|
|
|
+ // 如果没有找到记录，设置为当前时间前30分钟
|
|
|
+ Calendar calendar = Calendar.getInstance();
|
|
|
+ calendar.add(Calendar.MINUTE, -30); // 当前时间减去30分钟
|
|
|
+ latestCreateTime1 = calendar.getTime();
|
|
|
+ }
|
|
|
+
|
|
|
+ // 使用 split 方法分割字符串
|
|
|
+ String[] strArray = device.getDeviceVariable().split(",");
|
|
|
+
|
|
|
+ // 将数组转换为 List
|
|
|
+ List<String> list = Arrays.asList(strArray);
|
|
|
+
|
|
|
+ // 将 List 转换为正确的 SQL IN 子句格式
|
|
|
+ String deviceSnInClause = list.stream()
|
|
|
+ .map(String::trim) // 去掉可能的空格
|
|
|
+ .map(sn -> "'" + sn + "'") // 每个值加上单引号
|
|
|
+ .collect(Collectors.joining(",")); // 用逗号拼接
|
|
|
+
|
|
|
+ // 日期转字符串
|
|
|
+ String formattedTime = DateUtils.date2Str(latestCreateTime1, DateUtils.datetimeFormat.get());
|
|
|
+
|
|
|
+ // 构造 SQL
|
|
|
+ String sql = String.format("select key,ts,upload_time,device_sn from sys_device_delivery_t\n" +
|
|
|
+ "where device_sn in (%s) and \"key\" = 'ep' and upload_time > '%s' " +
|
|
|
+ "order by upload_time asc", deviceSnInClause, formattedTime);
|
|
|
+ statement = connection.prepareStatement(sql);
|
|
|
+ resultSet = statement.executeQuery();
|
|
|
+
|
|
|
+ // 解析结果集
|
|
|
+ List<Map<String, Object>> resultList = new ArrayList<>();
|
|
|
+ while (resultSet.next()) {
|
|
|
+ Map<String, Object> row = new HashMap<>();
|
|
|
+ row.put("key", resultSet.getString("key"));
|
|
|
+ row.put("ts", resultSet.getBigDecimal("ts"));
|
|
|
+ row.put("upload_time", resultSet.getTimestamp("upload_time"));
|
|
|
+ row.put("device_sn", resultSet.getString("device_sn"));
|
|
|
+ resultList.add(row);
|
|
|
+ }
|
|
|
+
|
|
|
+ // 按 upload_time 分组
|
|
|
+ Map<Timestamp, List<Map<String, Object>>> groupedData = resultList.stream()
|
|
|
+ .collect(Collectors.groupingBy(row -> (Timestamp) row.get("upload_time")));
|
|
|
+
|
|
|
+ // 对分组后的数据进行处理
|
|
|
+ for (Map.Entry<Timestamp, List<Map<String, Object>>> entry : groupedData.entrySet()) {
|
|
|
+ ProductionConsume productionConsume = new ProductionConsume();
|
|
|
+ productionConsume.setRegionTitle(device.getDeviceRegion());
|
|
|
+ productionConsume.setDeviceTitle(device.getDeviceTitle());
|
|
|
+
|
|
|
+ Timestamp uploadTime = entry.getKey();
|
|
|
+ List<Map<String, Object>> group = entry.getValue();
|
|
|
+ productionConsume.setCreateTime(uploadTime);
|
|
|
+
|
|
|
+ // 查询最新记录
|
|
|
+ LambdaQueryWrapper<ProductionConsume> consumeQueryWrapper = new LambdaQueryWrapper<ProductionConsume>()
|
|
|
+ .eq(ProductionConsume::getRegionTitle, device.getDeviceRegion())
|
|
|
+ .eq(ProductionConsume::getDeviceTitle, device.getDeviceTitle())
|
|
|
+ .orderByDesc(ProductionConsume::getCreateTime)
|
|
|
+ .last("limit 1");
|
|
|
+
|
|
|
+ ProductionConsume lastConsume = productionConsumeMapper.selectOne(consumeQueryWrapper);
|
|
|
+
|
|
|
+ log.info("处理分组数据: upload_time = {}, 数据量 = {}", uploadTime, group.size());
|
|
|
+
|
|
|
+ // 累加 ts 的值，保留为 BigDecimal 类型
|
|
|
+ BigDecimal totalTs = group.stream()
|
|
|
+ .map(row -> (BigDecimal) row.get("ts")) // 提取 ts 值为 BigDecimal
|
|
|
+ .reduce(BigDecimal.ZERO, BigDecimal::add); // 累加
|
|
|
+
|
|
|
+
|
|
|
+ // 将累加后的值设置到 activePower
|
|
|
+ productionConsume.setActivePower(totalTs.setScale(2, RoundingMode.HALF_UP));
|
|
|
+
|
|
|
+
|
|
|
+ // 计算 consumePower
|
|
|
+ if (lastConsume != null && totalTs != null && lastConsume.getActivePower() != null) {
|
|
|
+ productionConsume.setConsumePower(totalTs.subtract(lastConsume.getActivePower()).setScale(2, RoundingMode.HALF_UP));
|
|
|
+ } else if (lastConsume == null) {
|
|
|
+ productionConsume.setConsumePower(BigDecimal.ZERO);
|
|
|
+ }
|
|
|
+
|
|
|
+ // 保存或处理生产消耗记录
|
|
|
+ productionConsumeMapper.insert(productionConsume);
|
|
|
+ }
|
|
|
+
|
|
|
+ // 关闭连接
|
|
|
+ resultSet.close();
|
|
|
+ statement.close();
|
|
|
+ }
|
|
|
+ } catch (Exception e) {
|
|
|
+ // 关闭连接
|
|
|
+ log.info(e.getMessage());
|
|
|
+ } finally {
|
|
|
+ PostgreSQLUtil.close(connection);
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+
|
|
|
+}
|
