自来水数据采集至iot平台机制

dc3-gateway/src/main/java/io/github/pnoker/gateway/bizmgr/KprJiangjinWaterBizfun.java

@@ -223,7 +223,7 @@ public class KprJiangjinWaterBizfun {
             return;
         }
         try{
-            //TODO 按每天六小时从起始时间循环到今天的0点
+            //TODO
             LocalDateTime todayMidnight = LocalDateTime.now().toLocalDate().atTime(1, 0);
 
             // 循环从 startDateTime 开始，每 6 小时一次，直到今天 0 点

dc3-gateway/src/main/java/io/github/pnoker/gateway/bizmgr/KprZilaishuiWaterBizFun.java

@@ -0,0 +1,241 @@
+package io.github.pnoker.gateway.bizmgr;
+
+import io.github.pnoker.gateway.bizmgr.baseInit.KprBaseInitFun;
+import io.github.pnoker.gateway.comtool.TimeTool;
+import io.github.pnoker.gateway.dbdao.DBMgrProxy;
+import io.github.pnoker.gateway.utils.InfulxJiangjinDbUtil;
+import io.github.pnoker.gateway.utils.InfulxZilaishuiDbUtil;
+import org.influxdb.dto.Point;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.util.CollectionUtils;
+import org.springframework.util.ObjectUtils;
+
+import java.math.BigDecimal;
+import java.time.Instant;
+import java.time.LocalDateTime;
+import java.time.ZoneId;
+import java.time.ZonedDateTime;
+import java.time.format.DateTimeFormatter;
+import java.time.format.DateTimeFormatterBuilder;
+import java.time.temporal.ChronoField;
+import java.time.temporal.ChronoUnit;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+import static io.github.pnoker.gateway.comtool.jiangjinThread.KafkaConsumer.convertCubicMetersToLiters;
+
+/**
+ * @ClassName KprZilaishuiWaterBizFun
+ * @Description: TODO 市自来水采集
+ * @Author LX
+ * @Date 2024/12/3
+ * @Version V1.0
+ **/
+public class KprZilaishuiWaterBizFun {
+    private static final Logger log = LoggerFactory.getLogger(KprZilaishuiWaterBizFun.class);
+
+    private final static String mStrClassName = "KprZilaishuiWaterBizFun";
+    private final static String EMPTY_NULL = "NULL";
+
+    public static InfulxZilaishuiDbUtil infulxZilaishuiDbUtil = null;//infulx工具类对象
+
+    public static Point createPointFromJson(String measurement, String standardCode, String cmCode,Map<String,Object> mapEntity) {
+        // 获取时间戳
+        long nanoTimestamp = convertToNanoTimestamp(mapEntity.get("QCQUISITION_TIME").toString());
+        if(nanoTimestamp==0L){
+            return null;
+        }
+        // 创建 Point.Builder 对象
+        Point.Builder pointBuilder = Point.measurement(measurement)
+                .tag("dev_id",cmCode)
+                .time(nanoTimestamp, TimeUnit.NANOSECONDS);
+
+        // 遍历 JSON 对象的键值对，并添加到 Point.Builder 中
+
+        processField(pointBuilder,standardCode, mapEntity.get("VAL"));
+        if(!pointBuilder.hasFields()){
+            return null;
+        }
+
+        // 构建 Point 对象
+        try {
+            return pointBuilder.build();
+        }catch(Exception ex){
+            ex.printStackTrace();
+        }
+        return null;
+    }
+
+    //指定列处理
+    private static void processField(Point.Builder pointBuilder, String standardCode,Object value) {
+        if (value == null) {
+            return; // 跳过空值
+        }
+        String newKey = standardCode;
+        if (value instanceof Integer) {
+            pointBuilder.addField(newKey, (Integer) value);
+        } else if (value instanceof Long) {
+            pointBuilder.addField(newKey, (Long) value);
+        } else if (value instanceof Boolean) {
+            pointBuilder.addField(newKey, ((Boolean)value)?1:0);
+        } else if (value instanceof String) {
+            if (((String)value).equals("01")||((String)value).equals("00")){
+                pointBuilder.addField(newKey, ((String) value).replace("0",""));
+            }else {
+                pointBuilder.addField(newKey, (String) value);
+            }
+        }else if (value instanceof BigDecimal) {
+            pointBuilder.addField(newKey, ((BigDecimal) value).doubleValue());
+        }else{
+            pointBuilder.addField(newKey, (String.valueOf(value)) );
+        }
+        if(newKey.equals("flow_l_cur")||newKey.equals("flow_l_total_pos")){
+            pointBuilder.addField(newKey,  Double.valueOf(ObjectUtils.isEmpty(value)?"0.0":value.toString()));
+            if(newKey.equals("flow_l_cur")){
+                pointBuilder.addField(newKey,  Double.valueOf(ObjectUtils.isEmpty(value)?"0.0"
+                        :convertCubicMetersToLiters(Double.valueOf(value.toString()))));
+                pointBuilder.addField("flow_cur",  Double.valueOf(ObjectUtils.isEmpty(value)?"0.0":value.toString()));
+            }
+            if(newKey.equals("flow_l_total_pos")){
+                pointBuilder.addField(newKey,  Double.valueOf(ObjectUtils.isEmpty(value)?"0.0"
+                        :convertCubicMetersToLiters(Double.valueOf(value.toString()))));
+                pointBuilder.addField("flow_total_pos",  Double.valueOf(ObjectUtils.isEmpty(value)?"0.0":value.toString()));
+            }
+        }
+    }
+
+
+    //获取CST 纳秒时间戳
+    private static long convertToNanoTimestamp(String dateTimeString) {
+        // 定义时间格式，支持可选的小数秒部分
+        DateTimeFormatter formatter = new DateTimeFormatterBuilder()
+                .appendPattern("yyyy-MM-dd HH:mm:ss")
+                .optionalStart()
+                .appendFraction(ChronoField.NANO_OF_SECOND, 0, 9, true)
+                .optionalEnd()
+                .toFormatter();
+
+        // 解析字符串为 LocalDateTime 对象
+        LocalDateTime dateTime = LocalDateTime.parse(dateTimeString, formatter);
+
+        // 指定时区为CST（Asia/Shanghai）
+        ZoneId shanghaiZoneId = ZoneId.of("Asia/Shanghai");
+
+        // 将 LocalDateTime 对象转换为 ZonedDateTime 对象
+        ZonedDateTime zonedDateTime = dateTime.atZone(shanghaiZoneId);
+
+        // 将 ZonedDateTime 对象转换为 Instant 对象（UTC时间点）
+        Instant instant = zonedDateTime.toInstant();
+
+        // 计算从1970年1月1日00:00:00 UTC以来的纳秒数
+        long nanosecondsSinceEpoch = ChronoUnit.NANOS.between(Instant.EPOCH, instant);
+
+        // 每分钟的纳秒数
+        long nanosPerMinute = 60L * 1_000_000_000L;
+
+        // 四舍五入到最接近的整分的纳秒值
+        long roundedNanos = Math.round((double) nanosecondsSinceEpoch / nanosPerMinute) * nanosPerMinute;
+
+        // 输出调试信息
+//        System.out.println("解析的时间: " + dateTime);
+//        System.out.println("时区转换后的时间: " + zonedDateTime);
+//        System.out.println("计算的纳秒数: " + nanosecondsSinceEpoch);
+//        System.out.println("最接近整分的纳秒数: " + roundedNanos);
+
+        return roundedNanos;
+    }
+
+    //TODO 数据库视图采集
+    public static void initHistoryDb(LocalDateTime startDateTime){
+        DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
+        System.out.println("自来水历史数据执行:"+startDateTime.format(formatter));
+        if(startDateTime==null){
+            return;
+        }
+        try{
+            //TODO 按起始时间循环到当前时间
+            LocalDateTime todayNow = LocalDateTime.now();
+
+            // 循环从 startDateTime 开始，每 6 小时一次，直到今天 1 点
+            LocalDateTime currentDateTime = startDateTime;//每次循环查询的起始时间
+            //TODO 获取所有已配置的采集标签集合
+            List<String> content1List = KprBaseInitFun.getInstance().zilaishuiParams.values().stream()
+                    .flatMap(List::stream)
+                    .map(s -> s.split("___")[0])
+                    .collect(Collectors.toList());
+            // TODO 标签对应的采集字段
+            List<String> content2List = KprBaseInitFun.getInstance().zilaishuiParams.values().stream()
+                    .flatMap(List::stream)
+                    .map(s -> s.split("___")[1])
+                    .collect(Collectors.toList());
+
+            // TODO 标签对应的设施设备号
+            List<String> content3List = KprBaseInitFun.getInstance().zilaishuiParams.values().stream()
+                    .flatMap(List::stream)
+                    .map(s -> s.split("___")[2])
+                    .collect(Collectors.toList());
+            while(currentDateTime.isBefore(todayNow)) {
+                String startStr = currentDateTime.format(formatter);
+                String endStr = currentDateTime.plusHours(6).format(formatter);
+                String extend = "AND QCQUISITION_TIME >= TO_DATE('" + startStr + "', 'YYYY-MM-DD HH24:MI:SS') " +
+                        " AND QCQUISITION_TIME <= TO_DATE('" + endStr + "', 'YYYY-MM-DD HH24:MI:SS') " +
+                        " AND TAG_CODE IN " + "("+content1List.stream()
+                        .map(item -> "'" + item + "'")
+                        .collect(Collectors.joining(", "))+")";
+                int count = DBMgrProxy.getInstance().applyZilaishuiDbApi().getTabWaterHistoryCount(extend);
+                if(count>0){
+                    int pageNum = count % 5000 == 0 ? count / 5000 : (count / 5000) + 1;//总页数
+
+                    Integer limit = 5000;
+                    if (pageNum <= 1) {
+                        limit = count;//说明总数比第一页小
+                    }
+                    for (int i = 0; i < pageNum; i++) {
+                        Integer offset = i * limit;
+                        List<Map<String,Object>> mapList = DBMgrProxy.getInstance().applyZilaishuiDbApi()
+                                .getPageZILAISHUI_HISTORY2(offset+limit,offset,extend);
+                        if(!CollectionUtils.isEmpty(mapList)){
+                            for (Map<String,Object> map:mapList){
+                                //TODO 业务过程
+                                int tagIndex = content1List.indexOf(map.get("TAG_CODE").toString());
+                                if(tagIndex>0){
+                                    //TODO 那对应的字段和设备编号
+                                    String param = content2List.get(tagIndex);
+                                    String devId = content3List.get(tagIndex);
+                                    List<String> matchingKeys = KprBaseInitFun.getInstance().zilaishuiParams.entrySet().stream()
+                                            .filter(entry -> entry.getValue().stream()
+                                                    .anyMatch(s -> s.split("___")[0].equals(map.get("TAG_CODE"))))
+                                            .map(Map.Entry::getKey)
+                                            .collect(Collectors.toList());
+                                    if(!CollectionUtils.isEmpty(matchingKeys)) {
+                                        //比如key=Entity_WaterMeter
+                                        Point pointNanos = createPointFromJson(matchingKeys.get(0).split("_")[1],
+                                                param,
+                                                devId
+                                                , map);
+                                        if (pointNanos != null) {
+                                            KprZilaishuiWaterBizFun.infulxZilaishuiDbUtil.insert(pointNanos);
+                                        }
+                                    }
+                                }
+                            }
+                        }else{
+                            log.error(mStrClassName+";历史数据采集为空:"+ TimeTool.convertUTC2DateStr(TimeTool.getCurMsUTC(),TimeTool.TIMESTAMP_FORMAT)
+                                    +"initHistoryDbNullError");
+                        }
+                    }
+                    System.out.println(startStr+"到"+endStr+"数据整理完成:"+count);
+                }
+                currentDateTime = currentDateTime.plusHours(6); // 增加 6 小时
+            }
+        }catch(Exception ex){
+            ex.printStackTrace();
+            log.error(mStrClassName+";initHistoryDbError:"+ex.getLocalizedMessage());
+        }
+    }
+}

dc3-gateway/src/main/java/io/github/pnoker/gateway/bizmgr/baseInit/KprBaseInitFun.java

@@ -28,6 +28,10 @@ public class KprBaseInitFun {
     public Map<String,List<String>> jiangjinDeviceList = new HashMap<>();
     public List<String> closeTag = new ArrayList<>();//忽略的标签
 
+    //TODO 自来水相关
+    public Map<String,List<String>> zilaishuiDeviceList = new HashMap<>();
+    public Map<String,List<String>> zilaishuiParams = new HashMap<>();
+
     public Map<String,String> jiangjinMeasurementMap = new HashMap<>();
 
     private static volatile KprBaseInitFun msInstance = null;
@@ -50,6 +54,7 @@ public class KprBaseInitFun {
             initDangyang();
 //            initJiangjin();
             initJiangjinDb();
+            initZilaishuiDb();
         }catch(Exception ex){
             ex.printStackTrace();
         }
@@ -84,96 +89,96 @@ public class KprBaseInitFun {
         measurementMap.put("ELEC_CTRL_VALVE",properties.getProperty("ELEC_CTRL_VALVE"));
         measurementMap.put("LIQUID_LEVEL_METER",properties.getProperty("LIQUID_LEVEL_METER"));
     }
-    public void initJiangjin()throws Exception{
-        InputStream inputStream = getConfigStream("application-config.yml");
-        Properties properties = new Properties();
-        properties.load(inputStream);
-
-        List<String> VoltageSwitchgear=Arrays.asList(properties.getProperty("VoltageSwitchgear").split(","));
-        List<String> WaterValve=Arrays.asList(properties.getProperty("WaterValve").split(","));
-        List<String> VariableFrequencyDrive=Arrays.asList(properties.getProperty("VariableFrequencyDrive").split(","));
-        List<String> WaterQuality=Arrays.asList(properties.getProperty("WaterQuality").split(","));
-        List<String> DifferentialPressure=Arrays.asList(properties.getProperty("DifferentialPressure").split(","));
-        List<String> ElectricMotor=Arrays.asList(properties.getProperty("ElectricMotor").split(","));
-        List<String> WaterPump=Arrays.asList(properties.getProperty("WaterPump").split(","));
-        List<String> AirCompressor=Arrays.asList(properties.getProperty("AirCompressor").split(","));
-        List<String> RootsBlower=Arrays.asList(properties.getProperty("RootsBlower").split(","));
-        List<String> ThMeter=Arrays.asList(properties.getProperty("ThMeter").split(","));
-        List<String> SludgeIntfMeter=Arrays.asList(properties.getProperty("SludgeIntfMeter").split(","));
-        List<String> InformationSoftware=Arrays.asList(properties.getProperty("InformationSoftware").split(","));
-        List<String> WaterMeter=Arrays.asList(properties.getProperty("WaterMeter").split(","));
-        List<String> SmokeDetector=Arrays.asList(properties.getProperty("SmokeDetector").split(","));
-        List<String> LiquidLevelDiffer=Arrays.asList(properties.getProperty("LiquidLevelDiffer").split(","));
-        List<String> LiquidLevel=Arrays.asList(properties.getProperty("LiquidLevel").split(","));
-
-        jiangjinParams.put("VoltageSwitchgear",VoltageSwitchgear);
-        jiangjinParams.put("WaterValve",WaterValve);
-        jiangjinParams.put("VariableFrequencyDrive",VariableFrequencyDrive);
-        jiangjinParams.put("WaterQuality",WaterQuality);
-        jiangjinParams.put("DifferentialPressure",DifferentialPressure);
-        jiangjinParams.put("ElectricMotor",ElectricMotor);
-        jiangjinParams.put("WaterPump",WaterPump);
-        jiangjinParams.put("AirCompressor",AirCompressor);
-        jiangjinParams.put("RootsBlower",RootsBlower);
-        jiangjinParams.put("ThMeter",ThMeter);
-        jiangjinParams.put("SludgeIntfMeter",SludgeIntfMeter);
-        jiangjinParams.put("InformationSoftware",InformationSoftware);
-        jiangjinParams.put("WaterMeter",WaterMeter);
-        jiangjinParams.put("SmokeDetector",SmokeDetector);
-        jiangjinParams.put("LiquidLevelDiffer",LiquidLevelDiffer);
-        jiangjinParams.put("LiquidLevel",LiquidLevel);
-
-        List<String> DeviceVoltageSwitchgear=Arrays.asList(properties.getProperty("VoltageSwitchgear_device").split(","));
-        List<String> DeviceWaterValve=Arrays.asList(properties.getProperty("WaterValve_device").split(","));
-        List<String> DeviceVariableFrequencyDrive=Arrays.asList(properties.getProperty("VariableFrequencyDrive_device").split(","));
-        List<String> DeviceWaterQuality=Arrays.asList(properties.getProperty("WaterQuality_device").split(","));
-        List<String> DeviceDifferentialPressure=Arrays.asList(properties.getProperty("DifferentialPressure_device").split(","));
-        List<String> DeviceElectricMotor=Arrays.asList(properties.getProperty("ElectricMotor_device").split(","));
-        List<String> DeviceWaterPump=Arrays.asList(properties.getProperty("WaterPump_device").split(","));
-        List<String> DeviceAirCompressor=Arrays.asList(properties.getProperty("AirCompressor_device").split(","));
-        List<String> DeviceRootsBlower=Arrays.asList(properties.getProperty("RootsBlower_device").split(","));
-        List<String> DeviceThMeter=Arrays.asList(properties.getProperty("ThMeter_device").split(","));
-        List<String> DeviceSludgeIntfMeter=Arrays.asList(properties.getProperty("SludgeIntfMeter_device").split(","));
-        List<String> DeviceInformationSoftware=Arrays.asList(properties.getProperty("InformationSoftware_device").split(","));
-        List<String> DeviceWaterMeter=Arrays.asList(properties.getProperty("WaterMeter_device").split(","));
-        List<String> DeviceSmokeDetector=Arrays.asList(properties.getProperty("SmokeDetector_device").split(","));
-        List<String> DeviceLiquidLevelDiffer=Arrays.asList(properties.getProperty("LiquidLevelDiffer_device").split(","));
-        List<String> DeviceLiquidLevel=Arrays.asList(properties.getProperty("LiquidLevel_device").split(","));
-
-        jiangjinDeviceList.put("Device_VoltageSwitchgear",DeviceVoltageSwitchgear);
-        jiangjinDeviceList.put("Device_WaterValve",DeviceWaterValve);
-        jiangjinDeviceList.put("Device_VariableFrequencyDrive",DeviceVariableFrequencyDrive);
-        jiangjinDeviceList.put("Device_WaterQuality",DeviceWaterQuality);
-        jiangjinDeviceList.put("Device_DifferentialPressure",DeviceDifferentialPressure);
-        jiangjinDeviceList.put("Device_ElectricMotor",DeviceElectricMotor);
-        jiangjinDeviceList.put("Device_WaterPump",DeviceWaterPump);
-        jiangjinDeviceList.put("Device_AirCompressor",DeviceAirCompressor);
-        jiangjinDeviceList.put("Device_RootsBlower",DeviceRootsBlower);
-        jiangjinDeviceList.put("Device_ThMeter",DeviceThMeter);
-        jiangjinDeviceList.put("Device_SludgeIntfMeter",DeviceSludgeIntfMeter);
-        jiangjinDeviceList.put("Device_InformationSoftware",DeviceInformationSoftware);
-        jiangjinDeviceList.put("Device_WaterMeter",DeviceWaterMeter);
-        jiangjinDeviceList.put("Device_SmokeDetector",DeviceSmokeDetector);
-        jiangjinDeviceList.put("Device_LiquidLevelDiffer",DeviceLiquidLevelDiffer);
-        jiangjinDeviceList.put("Device_LiquidLevel",DeviceLiquidLevel);
-
-        jiangjinMeasurementMap.put("VoltageSwitchgear","VoltageSwitchgear");
-        jiangjinMeasurementMap.put("WaterValve","WaterValve");
-        jiangjinMeasurementMap.put("VariableFrequencyDrive","VariableFrequencyDrive");
-        jiangjinMeasurementMap.put("WaterQuality","WaterQuality");
-        jiangjinMeasurementMap.put("DifferentialPressure","DifferentialPressure");
-        jiangjinMeasurementMap.put("ElectricMotor","ElectricMotor");
-        jiangjinMeasurementMap.put("WaterPump","WaterPump");
-        jiangjinMeasurementMap.put("AirCompressor","AirCompressor");
-        jiangjinMeasurementMap.put("RootsBlower","RootsBlower");
-        jiangjinMeasurementMap.put("ThMeter","ThMeter");
-        jiangjinMeasurementMap.put("SludgeIntfMeter","SludgeIntfMeter");
-        jiangjinMeasurementMap.put("InformationSoftware","InformationSoftware");
-        jiangjinMeasurementMap.put("WaterMeter","WaterMeter");
-        jiangjinMeasurementMap.put("SmokeDetector","SmokeDetector");
-        jiangjinMeasurementMap.put("LiquidLevelDiffer","LiquidLevelDiffer");
-        jiangjinMeasurementMap.put("LiquidLevel","LiquidLevel");
-    }
+//    public void initJiangjin()throws Exception{
+//        InputStream inputStream = getConfigStream("application-config.yml");
+//        Properties properties = new Properties();
+//        properties.load(inputStream);
+//
+//        List<String> VoltageSwitchgear=Arrays.asList(properties.getProperty("VoltageSwitchgear").split(","));
+//        List<String> WaterValve=Arrays.asList(properties.getProperty("WaterValve").split(","));
+//        List<String> VariableFrequencyDrive=Arrays.asList(properties.getProperty("VariableFrequencyDrive").split(","));
+//        List<String> WaterQuality=Arrays.asList(properties.getProperty("WaterQuality").split(","));
+//        List<String> DifferentialPressure=Arrays.asList(properties.getProperty("DifferentialPressure").split(","));
+//        List<String> ElectricMotor=Arrays.asList(properties.getProperty("ElectricMotor").split(","));
+//        List<String> WaterPump=Arrays.asList(properties.getProperty("WaterPump").split(","));
+//        List<String> AirCompressor=Arrays.asList(properties.getProperty("AirCompressor").split(","));
+//        List<String> RootsBlower=Arrays.asList(properties.getProperty("RootsBlower").split(","));
+//        List<String> ThMeter=Arrays.asList(properties.getProperty("ThMeter").split(","));
+//        List<String> SludgeIntfMeter=Arrays.asList(properties.getProperty("SludgeIntfMeter").split(","));
+//        List<String> InformationSoftware=Arrays.asList(properties.getProperty("InformationSoftware").split(","));
+//        List<String> WaterMeter=Arrays.asList(properties.getProperty("WaterMeter").split(","));
+//        List<String> SmokeDetector=Arrays.asList(properties.getProperty("SmokeDetector").split(","));
+//        List<String> LiquidLevelDiffer=Arrays.asList(properties.getProperty("LiquidLevelDiffer").split(","));
+//        List<String> LiquidLevel=Arrays.asList(properties.getProperty("LiquidLevel").split(","));
+//
+//        jiangjinParams.put("VoltageSwitchgear",VoltageSwitchgear);
+//        jiangjinParams.put("WaterValve",WaterValve);
+//        jiangjinParams.put("VariableFrequencyDrive",VariableFrequencyDrive);
+//        jiangjinParams.put("WaterQuality",WaterQuality);
+//        jiangjinParams.put("DifferentialPressure",DifferentialPressure);
+//        jiangjinParams.put("ElectricMotor",ElectricMotor);
+//        jiangjinParams.put("WaterPump",WaterPump);
+//        jiangjinParams.put("AirCompressor",AirCompressor);
+//        jiangjinParams.put("RootsBlower",RootsBlower);
+//        jiangjinParams.put("ThMeter",ThMeter);
+//        jiangjinParams.put("SludgeIntfMeter",SludgeIntfMeter);
+//        jiangjinParams.put("InformationSoftware",InformationSoftware);
+//        jiangjinParams.put("WaterMeter",WaterMeter);
+//        jiangjinParams.put("SmokeDetector",SmokeDetector);
+//        jiangjinParams.put("LiquidLevelDiffer",LiquidLevelDiffer);
+//        jiangjinParams.put("LiquidLevel",LiquidLevel);
+//
+//        List<String> DeviceVoltageSwitchgear=Arrays.asList(properties.getProperty("VoltageSwitchgear_device").split(","));
+//        List<String> DeviceWaterValve=Arrays.asList(properties.getProperty("WaterValve_device").split(","));
+//        List<String> DeviceVariableFrequencyDrive=Arrays.asList(properties.getProperty("VariableFrequencyDrive_device").split(","));
+//        List<String> DeviceWaterQuality=Arrays.asList(properties.getProperty("WaterQuality_device").split(","));
+//        List<String> DeviceDifferentialPressure=Arrays.asList(properties.getProperty("DifferentialPressure_device").split(","));
+//        List<String> DeviceElectricMotor=Arrays.asList(properties.getProperty("ElectricMotor_device").split(","));
+//        List<String> DeviceWaterPump=Arrays.asList(properties.getProperty("WaterPump_device").split(","));
+//        List<String> DeviceAirCompressor=Arrays.asList(properties.getProperty("AirCompressor_device").split(","));
+//        List<String> DeviceRootsBlower=Arrays.asList(properties.getProperty("RootsBlower_device").split(","));
+//        List<String> DeviceThMeter=Arrays.asList(properties.getProperty("ThMeter_device").split(","));
+//        List<String> DeviceSludgeIntfMeter=Arrays.asList(properties.getProperty("SludgeIntfMeter_device").split(","));
+//        List<String> DeviceInformationSoftware=Arrays.asList(properties.getProperty("InformationSoftware_device").split(","));
+//        List<String> DeviceWaterMeter=Arrays.asList(properties.getProperty("WaterMeter_device").split(","));
+//        List<String> DeviceSmokeDetector=Arrays.asList(properties.getProperty("SmokeDetector_device").split(","));
+//        List<String> DeviceLiquidLevelDiffer=Arrays.asList(properties.getProperty("LiquidLevelDiffer_device").split(","));
+//        List<String> DeviceLiquidLevel=Arrays.asList(properties.getProperty("LiquidLevel_device").split(","));
+//
+//        jiangjinDeviceList.put("Device_VoltageSwitchgear",DeviceVoltageSwitchgear);
+//        jiangjinDeviceList.put("Device_WaterValve",DeviceWaterValve);
+//        jiangjinDeviceList.put("Device_VariableFrequencyDrive",DeviceVariableFrequencyDrive);
+//        jiangjinDeviceList.put("Device_WaterQuality",DeviceWaterQuality);
+//        jiangjinDeviceList.put("Device_DifferentialPressure",DeviceDifferentialPressure);
+//        jiangjinDeviceList.put("Device_ElectricMotor",DeviceElectricMotor);
+//        jiangjinDeviceList.put("Device_WaterPump",DeviceWaterPump);
+//        jiangjinDeviceList.put("Device_AirCompressor",DeviceAirCompressor);
+//        jiangjinDeviceList.put("Device_RootsBlower",DeviceRootsBlower);
+//        jiangjinDeviceList.put("Device_ThMeter",DeviceThMeter);
+//        jiangjinDeviceList.put("Device_SludgeIntfMeter",DeviceSludgeIntfMeter);
+//        jiangjinDeviceList.put("Device_InformationSoftware",DeviceInformationSoftware);
+//        jiangjinDeviceList.put("Device_WaterMeter",DeviceWaterMeter);
+//        jiangjinDeviceList.put("Device_SmokeDetector",DeviceSmokeDetector);
+//        jiangjinDeviceList.put("Device_LiquidLevelDiffer",DeviceLiquidLevelDiffer);
+//        jiangjinDeviceList.put("Device_LiquidLevel",DeviceLiquidLevel);
+//
+//        jiangjinMeasurementMap.put("VoltageSwitchgear","VoltageSwitchgear");
+//        jiangjinMeasurementMap.put("WaterValve","WaterValve");
+//        jiangjinMeasurementMap.put("VariableFrequencyDrive","VariableFrequencyDrive");
+//        jiangjinMeasurementMap.put("WaterQuality","WaterQuality");
+//        jiangjinMeasurementMap.put("DifferentialPressure","DifferentialPressure");
+//        jiangjinMeasurementMap.put("ElectricMotor","ElectricMotor");
+//        jiangjinMeasurementMap.put("WaterPump","WaterPump");
+//        jiangjinMeasurementMap.put("AirCompressor","AirCompressor");
+//        jiangjinMeasurementMap.put("RootsBlower","RootsBlower");
+//        jiangjinMeasurementMap.put("ThMeter","ThMeter");
+//        jiangjinMeasurementMap.put("SludgeIntfMeter","SludgeIntfMeter");
+//        jiangjinMeasurementMap.put("InformationSoftware","InformationSoftware");
+//        jiangjinMeasurementMap.put("WaterMeter","WaterMeter");
+//        jiangjinMeasurementMap.put("SmokeDetector","SmokeDetector");
+//        jiangjinMeasurementMap.put("LiquidLevelDiffer","LiquidLevelDiffer");
+//        jiangjinMeasurementMap.put("LiquidLevel","LiquidLevel");
+//    }
 
     public void initJiangjinDb()throws Exception{
         InputStream inputStream = getConfigStream("application-jiangjin.yml");
@@ -268,6 +273,27 @@ public class KprBaseInitFun {
         jiangjinMeasurementMap.put("LiquidLevel","LiquidLevel");
     }
 
+    public void initZilaishuiDb()throws Exception{
+        InputStream inputStream = getConfigStream("application-zilaishui.yml");
+        Properties properties = new Properties();
+        properties.load(inputStream);
+        List<String> WaterMeterDevice =Arrays.asList(properties.getProperty("Device_WaterMeter").split(","));
+        List<String> WaterQualityDevice =Arrays.asList(properties.getProperty("Device_WaterQuality").split(","));
+        List<String> LiquidLevelDevice =Arrays.asList(properties.getProperty("Device_LiquidLevel").split(","));
+
+        zilaishuiDeviceList.put("Device_WaterMeter",WaterMeterDevice);
+        zilaishuiDeviceList.put("Device_WaterQuality",WaterQualityDevice);
+        zilaishuiDeviceList.put("Device_LiquidLevel",LiquidLevelDevice);
+
+        List<String> WaterMeter =Arrays.asList(properties.getProperty("Entity_WaterMeter").split(","));
+        List<String> WaterQuality =Arrays.asList(properties.getProperty("Entity_WaterQuality").split(","));
+        List<String> LiquidLevel =Arrays.asList(properties.getProperty("Entity_LiquidLevel").split(","));
+
+        zilaishuiParams.put("Entity_WaterMeter",WaterMeter);
+        zilaishuiParams.put("Entity_WaterQuality",WaterQuality);
+        zilaishuiParams.put("Entity_LiquidLevel",LiquidLevel);
+    }
+
     public InputStream getConfigStream(String configFileName) {
         InputStream inputStream = null;
         try {

dc3-gateway/src/main/java/io/github/pnoker/gateway/comtool/ScheduleTaskMgr.java

@@ -4,9 +4,11 @@ import com.alibaba.fastjson.JSONArray;
 import com.alibaba.fastjson.JSONObject;
 import io.github.pnoker.gateway.bizmgr.KprDangyangWaterBizFun;
 import io.github.pnoker.gateway.bizmgr.KprJiangjinWaterBizfun;
+import io.github.pnoker.gateway.bizmgr.KprZilaishuiWaterBizFun;
 import io.github.pnoker.gateway.dbdao.DBMgrProxy;
 import io.github.pnoker.gateway.utils.InfulxDbUtil;
 import io.github.pnoker.gateway.utils.InfulxJiangjinDbUtil;
+import io.github.pnoker.gateway.utils.InfulxZilaishuiDbUtil;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Value;
@@ -21,6 +23,7 @@ import javax.annotation.Resource;
 import java.sql.Time;
 import java.time.LocalDate;
 import java.time.LocalDateTime;
+import java.time.LocalTime;
 import java.util.Timer;
 import java.util.TimerTask;
 import java.util.concurrent.Executors;
@@ -247,7 +250,7 @@ public class ScheduleTaskMgr {
                 }
             }
         },7000);
-        //TODO 启动是同步当前月一号到当前时间的所有历史数据
+        //TODO 启动时同步当前月一号到当前时间的所有历史数据
         new Timer().schedule(new TimerTask() {
             @Override
             public void run() {
@@ -274,12 +277,78 @@ public class ScheduleTaskMgr {
     }
 
     //TODO 每个月的1号凌晨2点执行
-    @Scheduled(cron = "0 0 2 1 * ?")
-    public void executeMonthlyTask() {
-        LocalDateTime firstDayOfLastMonthMidnight = LocalDate.now()
-                .minusMonths(1)  // 减去一个月
-                .withDayOfMonth(1)  // 设置为该月的第一天
-                .atStartOfDay();  // 设置时间为凌晨0点
-        KprJiangjinWaterBizfun.initHistoryDb(firstDayOfLastMonthMidnight);
+//    @Scheduled(cron = "0 0 2 1 * ?")
+//    public void executeMonthlyTask() {
+//        LocalDateTime firstDayOfLastMonthMidnight = LocalDate.now()
+//                .minusMonths(1)  // 减去一个月
+//                .withDayOfMonth(1)  // 设置为该月的第一天
+//                .atStartOfDay();  // 设置时间为凌晨0点
+//        KprJiangjinWaterBizfun.initHistoryDb(firstDayOfLastMonthMidnight);
+//    }
+
+
+    /**
+     * TODO 自来水相关
+     */
+    @Resource(name = "infulxZilaishuiDbUtil")
+    private InfulxZilaishuiDbUtil infulxZilaishuiDbUtil;
+
+    //TODO 启动后5秒初始化所有配置参数
+    @PostConstruct
+    public void initZilaishuiApplication(){
+        new Timer().schedule(new TimerTask() {
+            @Override
+            public void run() {
+                try {
+                    infulxZilaishuiDbUtil.initInfluxDataBase();
+                    KprZilaishuiWaterBizFun.infulxZilaishuiDbUtil= infulxZilaishuiDbUtil;
+                }catch(Exception ex){
+                    log.error("自来水启动时初始化配置参数失败:"+ex.getLocalizedMessage());
+                }
+            }
+        },5000);
     }
+
+    @PostConstruct
+    public void initZilaishuiHisData(){
+        //TODO 启动时同步当前月一号到当前时间的所有历史数据
+        new Timer().schedule(new TimerTask() {
+            @Override
+            public void run() {
+                try {
+                    KprZilaishuiWaterBizFun.initHistoryDb(LocalDate.now().withDayOfMonth(1).atStartOfDay());
+                }catch(Exception ex){
+
+                }
+            }
+        },7000);
+        new Timer().schedule(new TimerTask() {
+            @Override
+            public void run() {
+                try {
+                }catch(Exception ex){
+
+                }
+            }
+        },8000);
+    }
+
+    //TODO 每小时的第 0, 5, 10, 15, ..., 55 分钟执行一次
+    @Scheduled(cron = "0 0/5 * * * ?")
+    public void executeZilaishuiTask() {
+        KprZilaishuiWaterBizFun.initHistoryDb(LocalDateTime.now()
+                .withMinute(0)
+                .withSecond(0)
+                .withNano(0));
+    }
+
+    //TODO 每天凌晨1点执行前一天的
+    @Scheduled(cron = "0 0 1 * * ?")
+    public void executeZilaishuiHistoryTask() {
+        //TODO 执行前一天的数据到当前
+        KprZilaishuiWaterBizFun.initHistoryDb(LocalDate.now()
+                .minusDays(1)
+                .atStartOfDay());
+    }
+
 }

dc3-gateway/src/main/java/io/github/pnoker/gateway/config/OrcaleDataSourceConfig.java

@@ -33,27 +33,27 @@ public class OrcaleDataSourceConfig {
         return new HikariDataSource(config);
     }
 
-//    @Bean(name = "secondaryDataSource")
-//    public DataSource secondaryDataSource(@Value("${spring.datasource.secondary.url}") String url,
-//                                          @Value("${spring.datasource.secondary.username}") String username,
-//                                          @Value("${spring.datasource.secondary.password}") String password,
-//                                          @Value("${spring.datasource.secondary.driver-class-name}") String driverClassName) {
-//        HikariConfig config = new HikariConfig();
-//        config.setJdbcUrl(url);
-//        config.setUsername(username);
-//        config.setPassword(password);
-//        config.setDriverClassName(driverClassName);
-//        return new HikariDataSource(config);
-//    }
+    @Bean(name = "zilaishuiDataSource")
+    public DataSource secondaryDataSource(@Value("${spring.datasource.zilaishui.url}") String url,
+                                          @Value("${spring.datasource.zilaishui.username}") String username,
+                                          @Value("${spring.datasource.zilaishui.password}") String password,
+                                          @Value("${spring.datasource.zilaishui.driver-class-name}") String driverClassName) {
+        HikariConfig config = new HikariConfig();
+        config.setJdbcUrl(url);
+        config.setUsername(username);
+        config.setPassword(password);
+        config.setDriverClassName(driverClassName);
+        return new HikariDataSource(config);
+    }
 
     @Bean(name = "oneTemplate")
     public JdbcTemplate primaryJdbcTemplate(@Qualifier("primaryDataSource") DataSource dataSource) {
         return new JdbcTemplate(dataSource);
     }
 
-//    @Bean(name = "twoTemplate")
-//    public JdbcTemplate secondaryJdbcTemplate(@Qualifier("secondaryDataSource") DataSource dataSource) {
-//        return new JdbcTemplate(dataSource);
-//    }
+    @Bean(name = "twoTemplate")
+    public JdbcTemplate zilaishuiJdbcTemplate(@Qualifier("zilaishuiDataSource") DataSource dataSource) {
+        return new JdbcTemplate(dataSource);
+    }
 
 }

dc3-gateway/src/main/java/io/github/pnoker/gateway/dbdao/DBMgrProxy.java

@@ -3,6 +3,7 @@ package io.github.pnoker.gateway.dbdao;
 import io.github.pnoker.gateway.SpringContextUtil;
 import io.github.pnoker.gateway.dbdao.jiangjinSource.services.JiangjinService;
 import io.github.pnoker.gateway.dbdao.services.intef.*;
+import io.github.pnoker.gateway.dbdao.zilaishuiSource.service.ZilaishuiRealListService;
 
 /**
  * @ClassName DBMgrProxy
@@ -20,6 +21,7 @@ public class DBMgrProxy {
     private volatile TypeDefineService typeDefineService= null;
 
     private volatile JiangjinService jiangjinService = null;
+    private volatile ZilaishuiRealListService zilaishuiRealListService = null;
 
     private static volatile DBMgrProxy msInstance = null;
     public static DBMgrProxy getInstance(){
@@ -48,6 +50,8 @@ public class DBMgrProxy {
             typeDefineService = (TypeDefineService)SpringContextUtil.getBean(TypeDefineService.class);
         if (jiangjinService == null)
             jiangjinService = (JiangjinService)SpringContextUtil.getBean(JiangjinService.class);
+        if (zilaishuiRealListService == null)
+            zilaishuiRealListService = (ZilaishuiRealListService)SpringContextUtil.getBean(ZilaishuiRealListService.class);
     }
 
     public XuChangCustomerWaterConfigService applyXuchangCustomerWaterConfigApi() { return xuchangCustomerWaterConfigService; }
@@ -56,4 +60,5 @@ public class DBMgrProxy {
     public DeviceKindService applyDeviceKineApi() { return deviceKindService; }
     public TypeDefineService applyTypeDefineApi() { return typeDefineService; }
     public JiangjinService applyJiangjinDbApi() { return jiangjinService; }
+    public ZilaishuiRealListService applyZilaishuiDbApi() { return zilaishuiRealListService; }
 }

dc3-gateway/src/main/java/io/github/pnoker/gateway/dbdao/zilaishuiSource/ZilaishuiRealListDao.java

@@ -0,0 +1,62 @@
+package io.github.pnoker.gateway.dbdao.zilaishuiSource;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;
+import org.springframework.stereotype.Repository;
+import org.springframework.util.StringUtils;
+
+import javax.annotation.Resource;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * @ClassName JiangjinRealListDao
+ * @Description: TODO
+ * @Author LX
+ * @Date 2024/11/28
+ * @Version V1.0
+ **/
+@Slf4j
+@Repository
+public class ZilaishuiRealListDao {
+    @Resource
+    private JdbcTemplate twoTemplate;
+
+    //TODO orcale相关
+    public Integer getTabWaterHistoryCount(String extend){
+        try {
+            String sql = "select count(1)  from cqda.V_SHIZILAISHUI_HISTORY2 WHERE 1=1 ";
+            if(!StringUtils.isEmpty(extend)){
+                sql += extend;
+            }
+            int count = twoTemplate.queryForObject(sql, Integer.class);
+            return count;
+        }catch(Exception ex){
+            return null;
+        }
+    }
+    //TODO 分页获取历史水量数据
+    public List<Map<String,Object>> getPageZILAISHUI_HISTORY2(int limit,int offset,String extend){
+        try {
+            String sql = "select " +
+                    " a.TAG_CODE,a.NAME,a.VAL,TO_CHAR(a.QCQUISITION_TIME,'yyyy-mm-dd hh24:mi:ss') AS QCQUISITION_TIME," +
+                    " TO_CHAR(a.UPDATE_TIME,'yyyy-mm-dd hh24:mi:ss') AS UPDATE_TIME " +
+                    " from ( select t.*,rownum AS rownumber from cqda.V_SHIZILAISHUI_HISTORY2  t where rownum <= "+limit+") a " +
+                    " where rownumber > "+offset;
+            if (!StringUtils.isEmpty(extend)) {
+                sql = "select " +
+                        " a.TAG_CODE,a.NAME,a.VAL,TO_CHAR(a.QCQUISITION_TIME,'yyyy-mm-dd hh24:mi:ss') AS QCQUISITION_TIME," +
+                        " TO_CHAR(a.UPDATE_TIME,'yyyy-mm-dd hh24:mi:ss') AS UPDATE_TIME " +
+                        " from ( select t.*,rownum AS rownumber from cqda.V_SHIZILAISHUI_HISTORY2  t where rownum <= "+limit+" "+extend+") a " +
+                        " where rownumber >"+offset;
+            }
+            List<Map<String, Object>> tableData = twoTemplate.queryForList(sql);
+            return tableData;
+        }catch(Exception ex){
+            return null;
+        }
+    }
+
+}

dc3-gateway/src/main/java/io/github/pnoker/gateway/dbdao/zilaishuiSource/service/ZilaishuiRealListService.java

@@ -0,0 +1,10 @@
+package io.github.pnoker.gateway.dbdao.zilaishuiSource.service;
+
+import java.util.List;
+import java.util.Map;
+
+public interface ZilaishuiRealListService {
+    public Integer getTabWaterHistoryCount(String extend);
+
+    public List<Map<String,Object>> getPageZILAISHUI_HISTORY2(int limit, int offset, String extend);
+}

dc3-gateway/src/main/java/io/github/pnoker/gateway/dbdao/zilaishuiSource/service/impl/ZilaishuiRealListServiceImpl.java

@@ -0,0 +1,33 @@
+package io.github.pnoker.gateway.dbdao.zilaishuiSource.service.impl;
+
+import io.github.pnoker.gateway.dbdao.zilaishuiSource.ZilaishuiRealListDao;
+import io.github.pnoker.gateway.dbdao.zilaishuiSource.service.ZilaishuiRealListService;
+import org.springframework.stereotype.Service;
+
+import javax.annotation.Resource;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * @ClassName ZilaishuiRealListServiceImpl
+ * @Description: TODO
+ * @Author LX
+ * @Date 2024/12/3
+ * @Version V1.0
+ **/
+@Service
+public class ZilaishuiRealListServiceImpl implements ZilaishuiRealListService{
+
+    @Resource
+    private ZilaishuiRealListDao zilaishuiRealListDao;
+
+    @Override
+    public Integer getTabWaterHistoryCount(String extend) {
+        return zilaishuiRealListDao.getTabWaterHistoryCount(extend);
+    }
+
+    @Override
+    public List<Map<String, Object>> getPageZILAISHUI_HISTORY2(int limit, int offset, String extend) {
+        return zilaishuiRealListDao.getPageZILAISHUI_HISTORY2(limit, offset, extend);
+    }
+}

dc3-gateway/src/main/java/io/github/pnoker/gateway/utils/InfulxZilaishuiDbUtil.java

@@ -0,0 +1,71 @@
+package io.github.pnoker.gateway.utils;
+
+import com.alibaba.cloud.commons.lang.StringUtils;
+import io.github.pnoker.gateway.comtool.ResponseCode;
+import org.influxdb.InfluxDB;
+import org.influxdb.InfluxDBFactory;
+import org.influxdb.dto.BatchPoints;
+import org.influxdb.dto.Point;
+import org.influxdb.dto.Query;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.stereotype.Component;
+
+/**
+ * @ClassName InfulxDbUtil
+ * @Description: TODO influxDb连接工具类
+ * @Author LX
+ * @Date 2024/9/3
+ * @Version V1.0
+ **/
+@Component("infulxZilaishuiDbUtil")
+public class InfulxZilaishuiDbUtil {
+    @Value("${zilaishui.inluxdb.serverUrl:}")
+    private String serverUrl = ""; // InfluxDB 服务器地址
+
+    @Value("${zilaishui.inluxdb.username:}")
+    private String username = ""; // InfluxDB 用户名
+
+    @Value("${zilaishui.inluxdb.password:}")
+    private String password = ""; // InfluxDB 密码
+
+    @Value("${zilaishui.inluxdb.database:}")
+    private String database = ""; // 数据库名称
+
+    private InfluxDB influxDB = null;
+
+    public void initInfluxDataBase(){
+        try {
+            if (!StringUtils.isEmpty(username) && !StringUtils.isEmpty(password)) {
+                influxDB = InfluxDBFactory.connect(serverUrl, username, password);
+            } else {
+                influxDB = InfluxDBFactory.connect(serverUrl);
+            }
+            // 创建数据库
+            influxDB.query(new Query("CREATE DATABASE " + database));
+
+            // 设置要使用的数据库
+            influxDB.setDatabase(database);
+        }catch (Exception ex){
+            ex.printStackTrace();
+        }
+    }
+
+    public ResponseCode insert(Point point){
+        try {
+            influxDB.write(point);
+            return ResponseCode.RESULT_NORMAL;
+        }catch(Exception ex){
+            return ResponseCode.BUSINESS_DB_REQ_FAILED;
+        }
+    }
+
+    public ResponseCode insertBatch(BatchPoints point){
+        try {
+            influxDB.write(point);
+            return ResponseCode.RESULT_NORMAL;
+        }catch(Exception ex){
+            return ResponseCode.BUSINESS_DB_REQ_FAILED;
+        }
+    }
+
+}

dc3-gateway/src/main/resources/application.yml

@@ -144,6 +144,11 @@ spring:
             username: v_jiangjin
             password: JiangJin@1127
             driver-class-name: oracle.jdbc.driver.OracleDriver
+      zilaishui:
+            url: jdbc:oracle:thin:@10.127.16.117:1521/ORCLPDB1
+            username: v_shizilaishui
+            password: ShiZiLaiShui@0811
+            driver-class-name: oracle.jdbc.driver.OracleDriver
   jpa:
     show-sql: false
 
@@ -232,11 +237,10 @@ jiangjin:
 #    username: kpr
 #    password: kpr.2024dy.influxdb
 
-# oracle数据源
-#spring:
-#  datasource:
-#    primary:
-#      url: jdbc:oracle:thin:@10.127.16.117:1521/ORCLPDB1
-#      username: v_jiangjin
-#      password: JiangJin@1127
-#      driver-class-name: oracle.jdbc.driver.OracleDriver
+#自来水influxdb配置
+zilaishui:
+  #influxdb相关配置
+  inluxdb:
+    database: iot_test
+  #  database: iot
+    serverUrl: http://10.101.5.201:8086