威尔普斯采集数据BUG处理

src/main/java/com/isu/gaswellwatch/modbus/data/listener/DynamicRabbitListener.java

@@ -5,6 +5,7 @@ package com.isu.gaswellwatch.modbus.data.listener;
  * 2024/11/23 0:32
  */
 
+import lombok.Getter;
 import org.springframework.amqp.core.AmqpAdmin;
 import org.springframework.amqp.core.Queue;
 import org.springframework.amqp.core.QueueBuilder;
@@ -22,6 +23,7 @@ import java.util.stream.IntStream;
 public class DynamicRabbitListener implements ApplicationRunner {
 
     private final AmqpAdmin amqpAdmin;
+    @Getter
     private final ComposeModbusMessageListener composeListener;
     private final ComposeModbusMessageListener businessMessageListener;
     private final SimpleMessageListenerContainer modbusMessageListenerContainer;

src/main/java/com/isu/gaswellwatch/modbus/data/listener/ModbusMessagePersistListener.java

@@ -207,8 +207,12 @@ public class ModbusMessagePersistListener implements BatchMessageListener {
             } else if (stepSize <= 1) {
                 messagePoint = modbusMessage.getMessagePointMap()
                         .get(StringUtils.leftPad(String.valueOf(startAddress), 4, '0'));
+                if (Objects.isNull(messagePoint)) {
+                    value = StringUtils.EMPTY;
+                } else {
                     decodeMessage(decodeName, point, messagePoint);
                     value = messagePoint.getValue();
+                }
             } else {
                 value = decodeStepCommandPoint(modbusMessage.getMessagePointMap(), decodeName, point, startAddress, stepSize);
             }

src/main/java/com/isu/gaswellwatch/service/impl/DeviceServiceImpl.java

@@ -38,6 +38,7 @@ import java.text.SimpleDateFormat;
 import java.util.Date;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 
 
 @Service("deviceService")
@@ -185,7 +186,7 @@ public class DeviceServiceImpl extends ServiceImpl<DeviceDao, Device> implements
         this.gasWellService.unbindDevice(gasWellId);
 
         //删除t_data_设备历史数据表  删除缓存和MQ
-        cacheService.cleanDeviceCache(id);
+        this.cacheService.cleanDeviceCache(id);
 
     }
 
@@ -221,13 +222,13 @@ public class DeviceServiceImpl extends ServiceImpl<DeviceDao, Device> implements
     @Override
     public List<DeviceHistoryVO> getPressureChartData(Long deviceId, String startTime, String endTime) {
         String tableName = Redis2DBPersistenceService.DEFAULT_DATA_TABLE + deviceId;
-        return deviceDao.getPressureChartData(deviceId,startTime,endTime,tableName);
+        return this.deviceDao.getPressureChartData(deviceId, startTime, endTime, tableName);
     }
 
     @Override
     public List<DeviceHistoryVO> getSwitchStatusData(Long deviceId, String startTime, String endTime, String deviceProduct) {
         String tableName = Redis2DBPersistenceService.DEFAULT_DATA_TABLE + deviceId;
-        return deviceDao.getSwitchStatusData(deviceId,startTime,endTime,tableName,deviceProduct);
+        return this.deviceDao.getSwitchStatusData(deviceId, startTime, endTime, tableName, deviceProduct);
     }
 
     @Override
@@ -256,6 +257,9 @@ public class DeviceServiceImpl extends ServiceImpl<DeviceDao, Device> implements
             Map<String, Dictionary> plugStatusMap = this.dictionaryService.getValueMapByType("plugStatus");
 
             for (DeviceHistoryVO deviceVO : list) {
+                if (Objects.equals(deviceVO.getType(), PersistenceHandler.WEPS_PLUG_MODBUS_TYPE)) {
+                    continue;
+                }
                 deviceVO.setRunMode(StringUtils.isEmpty(deviceVO.getRunMode()) ? "" : runModeMap.get(deviceVO.getRunMode()).getName());
                 deviceVO.setPlugStatus(StringUtils.isEmpty(deviceVO.getPlugStatus()) ? "" : plugStatusMap.get(deviceVO.getPlugStatus()).getName());
             }
@@ -277,7 +281,7 @@ public class DeviceServiceImpl extends ServiceImpl<DeviceDao, Device> implements
                     // 当然这里数据也可以用 List<List<String>> 去传入
                     .doWrite(export);
         } catch (Exception e) {
-            log.error(e.getMessage());
+            this.log.error(e.getMessage());
         }
 
     }
@@ -287,9 +291,9 @@ public class DeviceServiceImpl extends ServiceImpl<DeviceDao, Device> implements
         SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
 
         UserLoginInfoVO userLoginInfoVO = (UserLoginInfoVO) StpUtil.getTokenSession().get(UserConstant.TOKEN_SESSION);
-        UserVO userVO = userService.selectUserInfo(userLoginInfoVO.getUserVO().getUsername());
+        UserVO userVO = this.userService.selectUserInfo(userLoginInfoVO.getUserVO().getUsername());
 
-        List<SwitchStatusExport> list = switchStatusExportService.getSwitchStatusExport(startTime, endTime,userVO.getDepartment().getId());
+        List<SwitchStatusExport> list = this.switchStatusExportService.getSwitchStatusExport(startTime, endTime, userVO.getDepartment().getId());
 
         List<SwitchStatusExportVO> export = ConverterUtil.convert(list, SwitchStatusExportVO.class);
 
@@ -307,7 +311,7 @@ public class DeviceServiceImpl extends ServiceImpl<DeviceDao, Device> implements
                     // 当然这里数据也可以用 List<List<String>> 去传入
                     .doWrite(export);
         } catch (Exception e) {
-            log.error(e.getMessage());
+            this.log.error(e.getMessage());
         }
     }
 

src/main/java/com/isu/gaswellwatch/service/impl/SummaryServiceImpl.java

@@ -36,7 +36,7 @@ public class SummaryServiceImpl  implements SummaryService {
     public List<PieSummaryVO> getIndexPieSummary() {
         Map<String, String> onlineMap = new HashMap<>();
         try {
-            onlineMap = redisTemplate.opsForHash().entries(ONLINE_DEVICE_CACHE);
+            onlineMap = this.redisTemplate.opsForHash().entries(ONLINE_DEVICE_CACHE);
         } catch (RedisConnectionFailureException e) {
             log.error("获取在线设备缓存失败", e);
             return getPieSummaryVOS();
@@ -47,7 +47,7 @@ public class SummaryServiceImpl  implements SummaryService {
         }
 //        List<Long> deviceIdList = onlineMap.keySet().stream().map(Long::parseLong).toList();
 
-        List<DeviceVO> deviceVOList = deviceService.getDeviceVOByIds(null);
+        List<DeviceVO> deviceVOList = this.deviceService.getDeviceVOByIds(null);
 
         //将所有设备按不同的设备品牌进行分组
         Map<String, List<DeviceVO>> deviceGroup = new HashMap<>();
@@ -66,6 +66,8 @@ public class SummaryServiceImpl  implements SummaryService {
         addKNPCDeviceSummary(KNPCV1_MODBUS_TYPE, deviceGroup, onlineMap, pieSummaryVOList, "KNPCV1");
         addKNPCDeviceSummary(ETC_MODBUS_TYPE, deviceGroup, onlineMap, pieSummaryVOList, "ETC");
         addKNPCDeviceSummary(SCSS_MODBUS_TYPE, deviceGroup, onlineMap, pieSummaryVOList, "四川双晟");
+        addKNPCDeviceSummary(WEPS_PLUG_MODBUS_TYPE, deviceGroup, onlineMap, pieSummaryVOList, "维尔普斯");
+        addKNPCDeviceSummary(MI_WEPS_PLUG_MODBUS_TYPE, deviceGroup, onlineMap, pieSummaryVOList, "维尔普斯");
 
         //计算总数
         PieSummaryVO pieSummaryVO = new PieSummaryVO();
@@ -95,13 +97,13 @@ public class SummaryServiceImpl  implements SummaryService {
     @Override
     public LineSummaryVO getPressureChartData(String startTime, String endTime, Long deviceId) {
         //根据设备ID、时间范围查询设备历史数据表的油压 套压
-        DeviceVO deviceVO = deviceService.getDevice(deviceId);
+        DeviceVO deviceVO = this.deviceService.getDevice(deviceId);
         if (deviceVO == null) {
             throw new RuntimeException("设备不存在");
         }
         List<String> xAxisData = new ArrayList<>();
 
-        List<DeviceHistoryVO> dataList = deviceService.getPressureChartData(deviceId,startTime,endTime);
+        List<DeviceHistoryVO> dataList = this.deviceService.getPressureChartData(deviceId, startTime, endTime);
         //将查询到的结果组装成LineSummaryVO返回给前端
         LineSummaryVO lineSummaryVO = new LineSummaryVO();
         lineSummaryVO.setTitle(deviceVO.getGasWell().getName());
@@ -175,13 +177,13 @@ public class SummaryServiceImpl  implements SummaryService {
 
     @Override
     public LineSummaryVO getSwitchChartData(String startTime, String endTime, Long deviceId) {
-        DeviceVO deviceVO = deviceService.getDevice(deviceId);
+        DeviceVO deviceVO = this.deviceService.getDevice(deviceId);
         if (deviceVO == null) {
             throw new RuntimeException("设备不存在");
         }
         List<String> xAxisData = new ArrayList<>();
 
-        List<DeviceHistoryVO> dataList = deviceService.getSwitchStatusData(deviceId,startTime,endTime,deviceVO.getProduct().getCode());
+        List<DeviceHistoryVO> dataList = this.deviceService.getSwitchStatusData(deviceId, startTime, endTime, deviceVO.getProduct().getCode());
         LineSummaryVO lineSummaryVO = new LineSummaryVO();
         lineSummaryVO.setTitle(deviceVO.getGasWell().getName());
 

src/main/java/com/isu/gaswellwatch/vo/DeviceHistoryVO.java

@@ -21,37 +21,65 @@ public class DeviceHistoryVO extends Model<DeviceHistoryVO> {
 
     private static final long serialVersionUID = 1L;
 
-    /** 采集时间 */
+
+    /**
+     * 设备类型
+     */
+    private String type;
+
+    /**
+     * 采集时间
+     */
     private String collectionTime;
 
-    /** 油压 */
+    /**
+     * 油压
+     */
     private String oilPressure;
 
-    /** 套压 */
+    /**
+     * 套压
+     */
     private String casPressure;
 
-    /** 输压 */
+    /**
+     * 输压
+     */
     private String prePressure;
 
-    /** 运行模式 */
+    /**
+     * 运行模式
+     */
     private String runMode;
 
-    /** 柱塞状态 */
+    /**
+     * 柱塞状态
+     */
     private String plugStatus;
 
-    /** 气井状态 */
+    /**
+     * 气井状态
+     */
     private String wellStatus;
 
-    /** 倒计时 */
+    /**
+     * 倒计时
+     */
     private String statusEndTime;
 
-    /** 温度 */
+    /**
+     * 温度
+     */
     private String temperature;
 
-    /** 湿度 */
+    /**
+     * 湿度
+     */
     private String humidity;
 
-    /** 太阳能电压 */
+    /**
+     * 太阳能电压
+     */
     private String solarVoltage;
 
 

src/main/resources/application-dev.yaml

@@ -2,6 +2,6 @@ spring:
   datasource:
     type: com.alibaba.druid.pool.DruidDataSource
     url: jdbc:mysql://localhost:3306/gas_well_watch?characterEncoding=UTF-8&useUnicode=true&useSSL=false&allowPublicKeyRetrieval=true&serverTimezone=Asia/Shanghai
-    username: cq
+    username: dev
-    password: cq
+    password: 1qaz@WSX
     driver-class-name: com.mysql.cj.jdbc.Driver

src/main/resources/mapper/DeviceDao.xml

@@ -171,15 +171,26 @@
 
     <select id="getDeviceHistoryData" resultType="com.isu.gaswellwatch.vo.DeviceHistoryVO">
         select t.collection_time,t.oil_pressure,t.cas_pressure,t.pre_pressure
-        <if test="deviceProduct!=null and deviceProduct=='knpc'">
+        <choose>
-            ,t.run_mode,t.status_end_time,t.temperature,t.humidity,t.well_status,t.plug_status
+            <when test="deviceProduct!=null and deviceProduct=='knpc'">
-        </if>
+                ,'knpc' as type,t.run_mode,t.status_end_time,t.temperature,t.humidity,t.well_status,t.plug_status
-        <if test="deviceProduct!=null and deviceProduct=='etc'">,t.current_status_remaining_time as
+            </when>
-            statusEndTime,t.controller_current_status as runMode,t.solenoid_valve_status as wellStatus
+            <when test="deviceProduct!=null and deviceProduct=='etc'">
-        </if>
+                ,'etc' as type,t.current_status_remaining_time as statusEndTime
-        <if test="deviceProduct!=null and deviceProduct=='scss'">,t.ctl_model as runMode,t.remaining_time_action as
+                ,t.controller_current_status as runMode,t.solenoid_valve_status as wellStatus
-            statusEndTime,t.solar_voltage,t.first_solenoid_status as wellStatus
+            </when>
-        </if>
+            <when test="deviceProduct!=null and deviceProduct=='scss'">
+                ,'scss' as type,t.ctl_model as runMode,t.remaining_time_action as statusEndTime
+                ,t.solar_voltage,t.first_solenoid_status as wellStatus
+            </when>
+            <when test="deviceProduct!=null and deviceProduct=='weps_plug'">
+                ,'weps_plug' as type,case t.run_mode when 1 then '定时开关井模式'
+                when 2 then '压力优化微升模式' when 3 then '自动优化模式' else '未知' end as runMode
+                ,case t.solenoid_valve_status when 2 then t.open_well_remaining_time
+                when 1 then t.close_well_remaining_time else null end as statusEndTime
+                ,case t.solenoid_valve_status when 2 then '关井' when 1 then '开井' else null end as wellStatus
+            </when>
+        </choose>
         from ${tableName} t
         <where>
             t.device_id = #{deviceId}

src/main/resources/sql/INSERT_WEPS_PLUG.sql

@@ -8,37 +8,39 @@ INSERT INTO `$TableName$`(`id`, `device_id`, `created_time`, `collection_time`,
                           `close_well_time`, `open_well_cas_pressure`, `little_rise_cas_pressure`,
                           `check_stability_time`, `min_close_well_time`, `max_close_well_time`, `min_open_well_time`,
                           `max_open_well_time`, `well_depth`, `produce_mode`)
-VALUES (?, ?, NOW(), ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+VALUES (?, ?, NOW(), ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?,
-ON DUPLICATE KEY UPDATE receive_time=VALUES(receive_time),
+        ?) ON DUPLICATE KEY
-                        cas_pressure=VALUES(cas_pressure),
+UPDATE receive_time=
-                        oil_pressure=VALUES(oil_pressure),
+VALUES (receive_time), cas_pressure=
-                        alarm=VALUES(alarm),
+VALUES (cas_pressure), oil_pressure=
-                        solenoid_valve_status=VALUES(solenoid_valve_status),
+VALUES (oil_pressure), alarm=
-                        open_well_remaining_time=VALUES(open_well_remaining_time),
+VALUES (alarm), solenoid_valve_status=
-                        close_well_remaining_time=VALUES(close_well_remaining_time),
+VALUES (solenoid_valve_status), open_well_remaining_time=
-                        plug_arrival_status=VALUES(plug_arrival_status),
+VALUES (open_well_remaining_time), close_well_remaining_time=
-                        plug_too_fast_count=VALUES(plug_too_fast_count),
+VALUES (close_well_remaining_time), plug_arrival_status=
-                        plug_rising_speed=VALUES(plug_rising_speed),
+VALUES (plug_arrival_status), plug_too_fast_count=
-                        downstream_pressure=VALUES(downstream_pressure),
+VALUES (plug_too_fast_count), plug_rising_speed=
-                        solenoid_valve_opening=VALUES(solenoid_valve_opening),
+VALUES (plug_rising_speed), downstream_pressure=
-                        reserve=VALUES(reserve),
+VALUES (downstream_pressure), solenoid_valve_opening=
-                        pre_pressure=VALUES(pre_pressure),
+VALUES (solenoid_valve_opening), reserve=
-                        run_mode=VALUES(run_mode),
+VALUES (reserve), pre_pressure=
-                        well_status=VALUES(well_status),
+VALUES (pre_pressure), run_mode=
-                        gas_collection_method=VALUES(gas_collection_method),
+VALUES (run_mode), well_status=
-                        high_pressure_protection=VALUES(high_pressure_protection),
+VALUES (well_status), gas_collection_method=
-                        low_pressure_protection=VALUES(low_pressure_protection),
+VALUES (gas_collection_method), high_pressure_protection=
-                        too_fast_count=VALUES(too_fast_count),
+VALUES (high_pressure_protection), low_pressure_protection=
-                        dangerous_rise_time=VALUES(dangerous_rise_time),
+VALUES (low_pressure_protection), too_fast_count=
-                        too_fast_rise_time=VALUES(too_fast_rise_time),
+VALUES (too_fast_count), dangerous_rise_time=
-                        open_well_time=VALUES(open_well_time),
+VALUES (dangerous_rise_time), too_fast_rise_time=
-                        close_well_time=VALUES(close_well_time),
+VALUES (too_fast_rise_time), open_well_time=
-                        open_well_cas_pressure=VALUES(open_well_cas_pressure),
+VALUES (open_well_time), close_well_time=
-                        little_rise_cas_pressure=VALUES(little_rise_cas_pressure),
+VALUES (close_well_time), open_well_cas_pressure=
-                        check_stability_time=VALUES(check_stability_time),
+VALUES (open_well_cas_pressure), little_rise_cas_pressure=
-                        min_close_well_time=VALUES(min_close_well_time),
+VALUES (little_rise_cas_pressure), check_stability_time=
-                        max_close_well_time=VALUES(max_close_well_time),
+VALUES (check_stability_time), min_close_well_time=
-                        min_open_well_time=VALUES(min_open_well_time),
+VALUES (min_close_well_time), max_close_well_time=
-                        max_open_well_time=VALUES(max_open_well_time),
+VALUES (max_close_well_time), min_open_well_time=
-                        well_depth=VALUES(well_depth),
+VALUES (min_open_well_time), max_open_well_time=
-                        produce_mode=VALUES(produce_mode)
+VALUES (max_open_well_time), well_depth=
+VALUES (well_depth), produce_mode=
+VALUES (produce_mode)