diff --git a/pom.xml b/pom.xml index 16c02e5..d8ccf78 100644 --- a/pom.xml +++ b/pom.xml @@ -98,6 +98,11 @@ org.aspectj aspectjweaver + + commons-io + commons-io + 2.18.0 + org.apache.commons commons-pool2 diff --git a/src/main/java/com/isu/gaswellwatch/modbus/data/ModbusMessage.java b/src/main/java/com/isu/gaswellwatch/modbus/data/ModbusMessage.java index d10de18..7bbc459 100644 --- a/src/main/java/com/isu/gaswellwatch/modbus/data/ModbusMessage.java +++ b/src/main/java/com/isu/gaswellwatch/modbus/data/ModbusMessage.java @@ -8,6 +8,7 @@ import lombok.experimental.SuperBuilder; import java.io.Serial; import java.io.Serializable; import java.util.Map; +import java.util.Objects; /** * 网关标识/设备标识/命令标识/采集下发时间/采集数据接收时间/采集数据体 @@ -63,7 +64,7 @@ public class ModbusMessage implements Serializable { @Setter @SuperBuilder @NoArgsConstructor - public static class MessagePoint implements Serializable { + public static class MessagePoint implements Serializable { @Serial private static final long serialVersionUID = -3118301653064611676L; /** @@ -73,10 +74,14 @@ public class ModbusMessage implements Serializable { /** * 初始解析值 */ - private T parseValue; + private String parseValue; /** * 修正解析值 */ - private T value; + private String value; + + public String getValue() { + return Objects.isNull(this.value) ? this.parseValue : this.value; + } } } diff --git a/src/main/java/com/isu/gaswellwatch/modbus/data/PersistenceHandler.java b/src/main/java/com/isu/gaswellwatch/modbus/data/PersistenceHandler.java index 691ce17..3b176ea 100644 --- a/src/main/java/com/isu/gaswellwatch/modbus/data/PersistenceHandler.java +++ b/src/main/java/com/isu/gaswellwatch/modbus/data/PersistenceHandler.java @@ -7,5 +7,11 @@ package com.isu.gaswellwatch.modbus.data; * 2024/11/23 11:53 */ public interface PersistenceHandler { + public static final String MODBUS_DEVICE_TYPE = "KNPCV1"; + String DEVICE_DATA_CACHE = "data:device:"; + + void createTable(String tableName); + + void insert(String tableName, String cacheKey); } diff --git a/src/main/java/com/isu/gaswellwatch/modbus/data/decode/DecodeHandler.java b/src/main/java/com/isu/gaswellwatch/modbus/data/decode/DecodeHandler.java index 98e4454..4d65ef9 100644 --- a/src/main/java/com/isu/gaswellwatch/modbus/data/decode/DecodeHandler.java +++ b/src/main/java/com/isu/gaswellwatch/modbus/data/decode/DecodeHandler.java @@ -12,9 +12,8 @@ import org.slf4j.LoggerFactory; */ public interface DecodeHandler { - String MODBUS_DEVICE_TYPE_FIELD_NAME = "modbusDeviceType"; Logger logger = LoggerFactory.getLogger("com.isu.gaswellwatch.decode"); - void decode(ModbusMessage modbusMessage); + void decode(ModbusMessage.MessagePoint point); } diff --git a/src/main/java/com/isu/gaswellwatch/modbus/data/decode/impl/Knpcv1DecodeHandler.java b/src/main/java/com/isu/gaswellwatch/modbus/data/decode/impl/Knpcv1DecodeHandler.java index 4bf9df7..8c2394b 100644 --- a/src/main/java/com/isu/gaswellwatch/modbus/data/decode/impl/Knpcv1DecodeHandler.java +++ b/src/main/java/com/isu/gaswellwatch/modbus/data/decode/impl/Knpcv1DecodeHandler.java @@ -3,8 +3,7 @@ package com.isu.gaswellwatch.modbus.data.decode.impl; import com.isu.gaswellwatch.modbus.data.ModbusMessage; import com.isu.gaswellwatch.modbus.data.decode.DecodeHandler; import jakarta.annotation.Resource; -import org.apache.commons.lang3.StringUtils; -import org.springframework.data.redis.core.HashOperations; +import lombok.extern.slf4j.Slf4j; import org.springframework.data.redis.core.RedisTemplate; import org.springframework.jdbc.core.JdbcTemplate; import org.springframework.stereotype.Component; @@ -15,10 +14,13 @@ import org.springframework.stereotype.Component; * @author 王仕龙 * 2024/11/23 11:23 */ -@Component(Knpcv1DecodeHandler.MODBUS_DEVICE_TYPE) +@Slf4j +@SuppressWarnings("all") +@Component(Knpcv1DecodeHandler.MODBUS_DEVICE_TYPE + "_1") public class Knpcv1DecodeHandler implements DecodeHandler { public static final String MODBUS_DEVICE_TYPE = "KNPCV1"; + private static final String COMMAND_POINT_SQL = "select * from command_points where command_id = ?"; @Resource private JdbcTemplate jdbcTemplate; @@ -27,12 +29,7 @@ public class Knpcv1DecodeHandler implements DecodeHandler { private RedisTemplate redisTemplate; @Override - public void decode(ModbusMessage modbusMessage) { - - HashOperations hashOperations = this.redisTemplate.opsForHash(); - String[] queuePath = StringUtils.split(modbusMessage.getQueueName(), "/"); - - logger.info("Decode {}, Queue {}, Message {}", MODBUS_DEVICE_TYPE, modbusMessage.getQueueName(), modbusMessage.getMessage()); + public void decode(ModbusMessage.MessagePoint point) { } diff --git a/src/main/java/com/isu/gaswellwatch/modbus/data/decode/listener/ModbusMessagePersistListener.java b/src/main/java/com/isu/gaswellwatch/modbus/data/decode/listener/ModbusMessagePersistListener.java index 8c89723..a821e5a 100644 --- a/src/main/java/com/isu/gaswellwatch/modbus/data/decode/listener/ModbusMessagePersistListener.java +++ b/src/main/java/com/isu/gaswellwatch/modbus/data/decode/listener/ModbusMessagePersistListener.java @@ -1,10 +1,10 @@ package com.isu.gaswellwatch.modbus.data.decode.listener; import cn.hutool.core.map.MapUtil; +import cn.hutool.json.JSONUtil; import com.isu.gaswellwatch.modbus.data.ModbusMessage; +import com.isu.gaswellwatch.modbus.data.PersistenceHandler; import com.isu.gaswellwatch.modbus.data.decode.DecodeHandler; -import com.serotonin.modbus4j.code.FunctionCode; -import com.serotonin.modbus4j.msg.ModbusResponse; import com.serotonin.modbus4j.msg.ReadInputRegistersResponse; import com.serotonin.modbus4j.serial.rtu.RtuMessageParser; import com.serotonin.modbus4j.serial.rtu.RtuMessageResponse; @@ -17,13 +17,12 @@ import org.apache.commons.lang3.StringUtils; import org.apache.commons.lang3.math.NumberUtils; import org.springframework.amqp.core.BatchMessageListener; import org.springframework.amqp.core.Message; +import org.springframework.data.redis.core.HashOperations; import org.springframework.data.redis.core.RedisTemplate; import org.springframework.jdbc.core.JdbcTemplate; -import java.util.HashMap; -import java.util.List; -import java.util.Map; -import java.util.Optional; +import java.time.Duration; +import java.util.*; /** * @author 王仕龙 @@ -31,9 +30,12 @@ import java.util.Optional; */ @Slf4j @RequiredArgsConstructor +@SuppressWarnings("all") public class ModbusMessagePersistListener implements BatchMessageListener { + public static final String COMMAND_CACHE = "modbus:command:"; private static final String COMMAND_SQL = "select * from commands where id = ?"; + private static final String COMMAND_POINT_SQL = "select * from command_points where command_id = ?"; @Resource private JdbcTemplate jdbcTemplate; @@ -49,89 +51,137 @@ public class ModbusMessagePersistListener implements BatchMessageListener { @Override public void onMessageBatch(List messages) { + Long commandId; + String[] messageSplit; + Map commandMap; + String messageString, collectionMessage, commandString; + Map messagePointMap; + // 网关标识/设备标识/命令标识/采集下发时间/采集数据接收时间/采集数据体 + // 4B454E454E4731343030303030333538/1/10000/1732365690000/1732365691000/01042C07D0000 + for (Message message : messages) { + messageString = new String(message.getBody()); + messageSplit = StringUtils.split(messageString, "/"); + if (messageSplit.length < 6) { + log.error("非法数据: {}", messageString); + continue; + } - try { - Long commandId; - String[] messageSplit; - Map commandMap; - String messageString, collectionMessage; - Map messagePointMap; -// 网关标识/设备标识/命令标识/采集下发时间/采集数据接收时间/采集数据体 -// 4B454E454E4731343030303030333538/1/10000/1732365690000/1732365691000/01042C07D000010003000C002B0029000300010000001A0018003303CD002300080DAC0DAC0000000000000959091DF066 - for (Message message : messages) { - messageString = new String(message.getBody()); - messageSplit = StringUtils.split(messageString, "/"); - if (messageSplit.length < 6) { - log.error("非法数据: {}", messageString); - continue; - } - - collectionMessage = messageSplit[5]; - commandId = NumberUtils.createLong(messageSplit[2]); + collectionMessage = messageSplit[5]; + commandId = NumberUtils.createLong(messageSplit[2]); + commandString = Objects.toString(this.redisTemplate.opsForValue().get(COMMAND_CACHE + commandId), ""); + commandMap = StringUtils.isNotEmpty(commandString) ? JSONUtil.parseObj(commandString) : null; + if (ObjectUtils.isEmpty(commandMap)) { commandMap = this.jdbcTemplate.queryForMap(COMMAND_SQL, commandId); if (ObjectUtils.isNotEmpty(commandMap)) { log.error("指令[{}]不存在，数据: {}", commandId, messageString); continue; } + this.redisTemplate.opsForValue() + .setIfAbsent(COMMAND_CACHE + commandId, JSONUtil.toJsonStr(commandMap), Duration.ofDays(1)); + } + DecodeHandler decodeHandler = Optional.ofNullable(MapUtil.getStr(commandMap, "decode_handler_name")) + .map(this.decodeHandlerMap::get) + .orElse(null); + if (Objects.isNull(decodeHandler)) { + log.error("指令[{}]未配置解析器，数据: {}", commandId, messageString); + continue; + } + messagePointMap = new HashMap<>(); + try { + String address; int index = MapUtil.getInt(commandMap, "start_address"), stepSize = 0; - - messagePointMap = new HashMap<>(); ByteQueue byteQueue = new ByteQueue(collectionMessage); RtuMessageParser masterParser = new RtuMessageParser(true); RtuMessageResponse response = (RtuMessageResponse) masterParser.parseMessage(byteQueue); - ModbusResponse modbusResponse = response.getModbusResponse(); - switch (modbusResponse.getFunctionCode()) { - case FunctionCode.READ_INPUT_REGISTERS: { - ReadInputRegistersResponse readInputRegistersResponse = (ReadInputRegistersResponse) modbusResponse; - for (short value : readInputRegistersResponse.getShortData()) { - stepSize = index * 4; - messagePointMap.put(StringUtils.leftPad(String.valueOf(index), 4, '0'), - ModbusMessage.MessagePoint.builder() - .parseValue(value) - .originalValue(StringUtils.substring(collectionMessage, 6 + stepSize, 10 + stepSize)) - .build()); - index++; - } - } -// case FunctionCode.READ_COILS: { -// ReadCoilsResponse readCoilsResponse = (ReadCoilsResponse) modbusResponse; -// } -// case FunctionCode.READ_DISCRETE_INPUTS: { -// ReadDiscreteInputsResponse readDiscreteInputsResponse = (ReadDiscreteInputsResponse) modbusResponse; -// } -// case FunctionCode.READ_HOLDING_REGISTERS: { -// ReadHoldingRegistersResponse readHoldingRegistersResponse = (ReadHoldingRegistersResponse) modbusResponse; -// } -// case FunctionCode.READ_EXCEPTION_STATUS: { -// ReadExceptionStatusResponse readExceptionStatusResponse = (ReadExceptionStatusResponse) modbusResponse; -// } -// case FunctionCode.REPORT_SLAVE_ID: { -// ReportSlaveIdResponse reportSlaveIdResponse = (ReportSlaveIdResponse) modbusResponse; -// } - default: { - log.error("FunctionCode {}, Message {}", modbusResponse.getFunctionCode(), collectionMessage); - } + ReadInputRegistersResponse readInputRegistersResponse = (ReadInputRegistersResponse) response.getModbusResponse(); + for (short value : readInputRegistersResponse.getShortData()) { + stepSize = index * 4; + messagePointMap.put(StringUtils.leftPad(String.valueOf(index), 4, '0'), + ModbusMessage.MessagePoint.builder() + .parseValue(String.valueOf(value)) + .originalValue(StringUtils.substring(collectionMessage, 6 + stepSize, 10 + stepSize)) + .build()); + index++; } - - - Optional.ofNullable(commandMap.get("decode_handler_name")) - .map(this.decodeHandlerMap::get) - .ifPresent(decodeHandler -> decodeHandler.decode(ModbusMessage.builder() - .queueName(message.getMessageProperties().getConsumerQueue()) - .commandId(commandId) - .message(collectionMessage) - .gatewayIdentifier(messageSplit[0]) - .deviceId(NumberUtils.createLong(messageSplit[1])) - .collectionTime(NumberUtils.createLong(messageSplit[3])) - .receiveTime(NumberUtils.createLong(messageSplit[4])) - .messagePointMap(messagePointMap) - .build())); - - + } catch (Exception e) { + log.error("初始数据解析异常: {}", messageString, e); + continue; + } + if (ObjectUtils.isNotEmpty(messagePointMap)) { + decode(ModbusMessage.builder() + .commandId(commandId) + .message(collectionMessage) + .messagePointMap(messagePointMap) + .gatewayIdentifier(messageSplit[0]) + .deviceId(NumberUtils.createLong(messageSplit[1])) + .receiveTime(NumberUtils.createLong(messageSplit[4])) + .collectionTime(NumberUtils.createLong(messageSplit[3])) + .queueName(message.getMessageProperties().getConsumerQueue()) + .build()); } - } catch (Exception e) { - throw new RuntimeException(e); - } } + + + public void decode(ModbusMessage modbusMessage) { + Long commandId = modbusMessage.getCommandId(); + String cacheName = ModbusMessagePersistListener.COMMAND_CACHE + commandId + ":points"; + String cachePoints = Objects.toString(this.redisTemplate.opsForValue().get(cacheName), ""); + + List> commandPointList = StringUtils.isNotEmpty(cachePoints) ? (List) JSONUtil.parseArray(cachePoints) : null; + if (ObjectUtils.isEmpty(commandPointList)) { + commandPointList = this.jdbcTemplate.queryForList(COMMAND_POINT_SQL, commandId); + if (ObjectUtils.isNotEmpty(commandPointList)) { + log.error("指令[{}]点表配置为空", commandId, cachePoints); + throw new RuntimeException("指令[" + commandId + "]点表未配置"); + } + this.redisTemplate.opsForValue().setIfAbsent(cacheName, JSONUtil.toJsonStr(commandPointList), Duration.ofDays(1)); + } + + DecodeHandler decodeHandler; + String fieldName, decodeName; + int startAddress = 0, stepSize = 0; + ModbusMessage.MessagePoint messagePoint; + String deviceDataCacheName = PersistenceHandler.DEVICE_DATA_CACHE + modbusMessage.getDeviceId(); + HashOperations hashOperations = this.redisTemplate.opsForHash(); + for (Map point : commandPointList) { + fieldName = MapUtil.getStr(point, "field"); + stepSize = MapUtil.getInt(point, "step_size"); + decodeName = MapUtil.getStr(point, "decode_name"); + startAddress = MapUtil.getInt(point, "start_address"); + String value; + if (stepSize <= 1) { + messagePoint = modbusMessage.getMessagePointMap() + .get(StringUtils.leftPad(String.valueOf(startAddress), 4, '0')); + decodeMessage(decodeName, messagePoint); + value = messagePoint.getValue(); + } else { + value = decodeCommandPoint(modbusMessage.getMessagePointMap(), decodeName, + MapUtil.getStr(point, "format"), startAddress, stepSize); + } + hashOperations.put(deviceDataCacheName, fieldName, value); + } + } + + private void decodeMessage(String decodeName, ModbusMessage.MessagePoint messagePoint) { + if (StringUtils.isNotBlank(decodeName)) { + DecodeHandler decodeHandler = this.decodeHandlerMap.get(decodeName); + if (Objects.nonNull(decodeHandler)) { + decodeHandler.decode(messagePoint); + } + } + } + + private String decodeCommandPoint(Map pointMap, + String decodeName, String format, int startAddress, int stepSize) { + String[] values = new String[stepSize]; + ModbusMessage.MessagePoint messagePoint; + for (int i = 0; i < stepSize; i++) { + messagePoint = pointMap.get(StringUtils.leftPad(String.valueOf(startAddress + i), 4, '0')); + decodeMessage(decodeName, messagePoint); + values[i] = messagePoint.getValue(); + } + return String.format(format, values); + } + } diff --git a/src/main/java/com/isu/gaswellwatch/modbus/data/impl/Knpcv1PersistenceHandler.java b/src/main/java/com/isu/gaswellwatch/modbus/data/impl/Knpcv1PersistenceHandler.java new file mode 100644 index 0000000..e0c2e27 --- /dev/null +++ b/src/main/java/com/isu/gaswellwatch/modbus/data/impl/Knpcv1PersistenceHandler.java @@ -0,0 +1,135 @@ +package com.isu.gaswellwatch.modbus.data.impl; + +import cn.hutool.core.map.MapUtil; +import com.isu.gaswellwatch.config.SnowflakeConfig; +import com.isu.gaswellwatch.modbus.data.PersistenceHandler; +import jakarta.annotation.Resource; +import org.apache.commons.io.IOUtils; +import org.apache.commons.lang3.StringUtils; +import org.springframework.dao.DataAccessException; +import org.springframework.data.redis.core.RedisTemplate; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.jdbc.core.PreparedStatementCallback; +import org.springframework.stereotype.Component; +import org.springframework.util.ResourceUtils; + +import java.io.File; +import java.io.FileInputStream; +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.sql.PreparedStatement; +import java.sql.SQLException; +import java.sql.Types; +import java.util.Map; +import java.util.Objects; + +/** + * @author 王仕龙 + * 2024/11/24 17:33 + */ +@Component(PersistenceHandler.MODBUS_DEVICE_TYPE) +public class Knpcv1PersistenceHandler implements PersistenceHandler { + + @Resource + private JdbcTemplate jdbcTemplate; + @Resource + private SnowflakeConfig snowflakeConfig; + @Resource(name = "redisTemplate") + private RedisTemplate redisTemplate; + + @Override + public void createTable(String tableName) { + String createTableSQL = getResource("sql/CREATE_KNPCV1.sql"); + this.jdbcTemplate.execute(StringUtils.replace(createTableSQL, "$TableName$", tableName)); + } + + @Override + public void insert(String tableName, String cacheKey) { + String insertTableSQL = getResource("sql/INSERT_KNPCV1.sql"); + insertTableSQL = StringUtils.replace(insertTableSQL, "$TableName$", tableName); + + Map row = this.redisTemplate.opsForHash().entries(cacheKey); + + this.jdbcTemplate.execute(insertTableSQL, new PreparedStatementCallback() { + @Override + public Object doInPreparedStatement(PreparedStatement ps) throws SQLException, DataAccessException { + ps.setLong(1, Knpcv1PersistenceHandler.this.snowflakeConfig.snowflakeId()); + Knpcv1PersistenceHandler.this.setValue(ps, row, 2, "deviceId", Types.BIGINT); + Knpcv1PersistenceHandler.this.setValue(ps, row, 3, "collectionTime", Types.BIGINT); + Knpcv1PersistenceHandler.this.setValue(ps, row, 4, "receiveTime", Types.BIGINT); + Knpcv1PersistenceHandler.this.setValue(ps, row, 5, "status", Types.BIGINT); + Knpcv1PersistenceHandler.this.setValue(ps, row, 6, "device_time", Types.BIGINT); + Knpcv1PersistenceHandler.this.setValue(ps, row, 7, "runMode", Types.BIGINT); + Knpcv1PersistenceHandler.this.setValue(ps, row, 8, "wellStatus", Types.BIGINT); + Knpcv1PersistenceHandler.this.setValue(ps, row, 9, "plugStatus", Types.BIGINT); + Knpcv1PersistenceHandler.this.setValue(ps, row, 10, "statusOpenTime", Types.BIGINT); + Knpcv1PersistenceHandler.this.setValue(ps, row, 11, "statusCloseTime", Types.BIGINT); + Knpcv1PersistenceHandler.this.setValue(ps, row, 12, "oilPressure", Types.BIGINT); + Knpcv1PersistenceHandler.this.setValue(ps, row, 13, "casPressure", Types.BIGINT); + Knpcv1PersistenceHandler.this.setValue(ps, row, 14, "prePressure", Types.BIGINT); + Knpcv1PersistenceHandler.this.setValue(ps, row, 15, "pipePressure", Types.BIGINT); + Knpcv1PersistenceHandler.this.setValue(ps, row, 16, "liquidLevel", Types.BIGINT); + Knpcv1PersistenceHandler.this.setValue(ps, row, 17, "temperature", Types.BIGINT); + Knpcv1PersistenceHandler.this.setValue(ps, row, 18, "humidity", Types.BIGINT); + Knpcv1PersistenceHandler.this.setValue(ps, row, 19, "opmode", Types.BIGINT); + Knpcv1PersistenceHandler.this.setValue(ps, row, 20, "well_ctl", Types.BIGINT); + Knpcv1PersistenceHandler.this.setValue(ps, row, 21, "timer1", Types.BIGINT); + Knpcv1PersistenceHandler.this.setValue(ps, row, 22, "timer2", Types.BIGINT); + Knpcv1PersistenceHandler.this.setValue(ps, row, 23, "timer1Open", Types.BIGINT); + Knpcv1PersistenceHandler.this.setValue(ps, row, 24, "timer1Close", Types.BIGINT); + Knpcv1PersistenceHandler.this.setValue(ps, row, 25, "timer2Open", Types.BIGINT); + Knpcv1PersistenceHandler.this.setValue(ps, row, 26, "timer2Close", Types.BIGINT); + Knpcv1PersistenceHandler.this.setValue(ps, row, 27, "timingOpen", Types.BIGINT); + Knpcv1PersistenceHandler.this.setValue(ps, row, 28, "timingClose", Types.BIGINT); + Knpcv1PersistenceHandler.this.setValue(ps, row, 29, "timingCelay", Types.BIGINT); + Knpcv1PersistenceHandler.this.setValue(ps, row, 30, "presource", Types.BIGINT); + Knpcv1PersistenceHandler.this.setValue(ps, row, 31, "pressureOpen", Types.BIGINT); + Knpcv1PersistenceHandler.this.setValue(ps, row, 32, "pressureClose", Types.BIGINT); + Knpcv1PersistenceHandler.this.setValue(ps, row, 33, "triggerType", Types.BIGINT); + Knpcv1PersistenceHandler.this.setValue(ps, row, 34, "stabilityTime", Types.BIGINT); + Knpcv1PersistenceHandler.this.setValue(ps, row, 35, "maxOpenWell", Types.BIGINT); + Knpcv1PersistenceHandler.this.setValue(ps, row, 36, "maxCloseWell", Types.BIGINT); + Knpcv1PersistenceHandler.this.setValue(ps, row, 37, "minOpenWell", Types.BIGINT); + Knpcv1PersistenceHandler.this.setValue(ps, row, 38, "minCloseWell", Types.BIGINT); + Knpcv1PersistenceHandler.this.setValue(ps, row, 39, "presproTect", Types.BIGINT); + Knpcv1PersistenceHandler.this.setValue(ps, row, 40, "presproSource", Types.BIGINT); + Knpcv1PersistenceHandler.this.setValue(ps, row, 41, "openWellLimitMax", Types.BIGINT); + Knpcv1PersistenceHandler.this.setValue(ps, row, 42, "openWellLimitMin", Types.BIGINT); + Knpcv1PersistenceHandler.this.setValue(ps, row, 43, "plugInitStatus", Types.BIGINT); + Knpcv1PersistenceHandler.this.setValue(ps, row, 44, "plugSustainTime", Types.BIGINT); + Knpcv1PersistenceHandler.this.setValue(ps, row, 45, "plugCloseTime", Types.BIGINT); + Knpcv1PersistenceHandler.this.setValue(ps, row, 46, "tpInitStatus", Types.BIGINT); + Knpcv1PersistenceHandler.this.setValue(ps, row, 47, "tpOpenSource", Types.BIGINT); + Knpcv1PersistenceHandler.this.setValue(ps, row, 48, "tpOpenTrigger", Types.BIGINT); + Knpcv1PersistenceHandler.this.setValue(ps, row, 49, "tpOpenPressure", Types.BIGINT); + Knpcv1PersistenceHandler.this.setValue(ps, row, 50, "tpOpenTime", Types.BIGINT); + Knpcv1PersistenceHandler.this.setValue(ps, row, 51, "tpCloseSource", Types.BIGINT); + Knpcv1PersistenceHandler.this.setValue(ps, row, 52, "tpCloseTrigger", Types.BIGINT); + Knpcv1PersistenceHandler.this.setValue(ps, row, 53, "tpClosePressure", Types.BIGINT); + Knpcv1PersistenceHandler.this.setValue(ps, row, 54, "tpCloseTime", Types.BIGINT); + Knpcv1PersistenceHandler.this.setValue(ps, row, 55, "tpStabilityTime", Types.BIGINT); + return ps.executeUpdate(); + } + }); + + } + + public void setValue(PreparedStatement ps, Map row, int index, String key, int sqlType) throws SQLException { + String value = MapUtil.getStr(row, key); + if (Objects.isNull(value)) { + ps.setNull(index, sqlType); + } else { + ps.setObject(index, value); + } + } + + public static String getResource(String classPath) { + try { + File file = ResourceUtils.getFile("classpath:" + classPath); + return IOUtils.toString(new FileInputStream(file), StandardCharsets.UTF_8); + } catch (IOException e) { + throw new RuntimeException(e); + } + } + +} diff --git a/src/main/java/com/isu/gaswellwatch/modbus/data/impl/Redis2DBPersistenceHandlerImpl.java b/src/main/java/com/isu/gaswellwatch/modbus/data/impl/Redis2DBPersistenceHandlerImpl.java deleted file mode 100644 index fe27be0..0000000 --- a/src/main/java/com/isu/gaswellwatch/modbus/data/impl/Redis2DBPersistenceHandlerImpl.java +++ /dev/null @@ -1,28 +0,0 @@ -package com.isu.gaswellwatch.modbus.data.impl; - -import com.isu.gaswellwatch.modbus.data.PersistenceHandler; -import jakarta.annotation.Resource; -import org.springframework.data.redis.core.RedisTemplate; -import org.springframework.jdbc.core.JdbcTemplate; -import org.springframework.scheduling.annotation.Scheduled; -import org.springframework.stereotype.Component; - -/** - * @author 王仕龙 - * 2024/11/23 11:55 - */ -@Component -public class Redis2DBPersistenceHandlerImpl implements PersistenceHandler { - - @Resource - private JdbcTemplate jdbcTemplate; - - @Resource(name = "redisTemplate") - private RedisTemplate redisTemplate; - - @Scheduled(cron = "0/10 * * * * ? ") - public void write() { - this.redisTemplate.getClientList(); - } - -} diff --git a/src/main/java/com/isu/gaswellwatch/modbus/data/impl/Redis2DBPersistenceService.java b/src/main/java/com/isu/gaswellwatch/modbus/data/impl/Redis2DBPersistenceService.java new file mode 100644 index 0000000..fae477d --- /dev/null +++ b/src/main/java/com/isu/gaswellwatch/modbus/data/impl/Redis2DBPersistenceService.java @@ -0,0 +1,87 @@ +package com.isu.gaswellwatch.modbus.data.impl; + +import cn.hutool.core.map.MapUtil; +import com.isu.gaswellwatch.modbus.data.PersistenceHandler; +import jakarta.annotation.Resource; +import lombok.extern.slf4j.Slf4j; +import org.apache.commons.lang3.ObjectUtils; +import org.apache.commons.lang3.StringUtils; +import org.springframework.core.annotation.Order; +import org.springframework.data.redis.core.Cursor; +import org.springframework.data.redis.core.HashOperations; +import org.springframework.data.redis.core.RedisTemplate; +import org.springframework.data.redis.core.ScanOptions; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.scheduling.annotation.Scheduled; +import org.springframework.stereotype.Component; +import org.springframework.web.client.RestTemplate; + +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +/** + * @author 王仕龙 + * 2024/11/23 11:55 + */ +@Slf4j +@Component +@SuppressWarnings("all") +public class Redis2DBPersistenceService { + + private static final String EXISTS_TABLE_SQL = "SELECT TABLE_NAME FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA='gaswellwatch' AND TABLE_NAME=?"; + @Resource + private JdbcTemplate jdbcTemplate; + @Resource(name = "redisTemplate") + private RedisTemplate redisTemplate; + @Resource + private Map persistenceHandlerMap; + private RestTemplate restTemplate = new RestTemplate(); + + @Order(100) + @Scheduled(cron = "5/30 * * * * ? ") + public void write() { + Map idGatewayMappingMap = null; + try { + String onlineGateways = this.restTemplate.getForObject("http://127.0.0.1:9999/modbus-tcp/online", String.class); + if (StringUtils.isBlank(onlineGateways)) { + return; + } + List> idGatewayMappingList = this.jdbcTemplate + .queryForList("select id, gateway_sn from devices where gateway_sn in (" + onlineGateways + ")"); + if (ObjectUtils.isEmpty(idGatewayMappingList)) { + return; + } + idGatewayMappingMap = new HashMap(); + for (Map map : idGatewayMappingList) { + idGatewayMappingMap.put(MapUtil.getLong(map, "id"), MapUtil.getStr(map, "gateway_sn")); + } + } catch (Exception e) { + log.error("Get online devices list fail", e); + } + + HashOperations operations = this.redisTemplate.opsForHash(); + try (Cursor cursor = this.redisTemplate.scan(ScanOptions.scanOptions() + .match(PersistenceHandler.DEVICE_DATA_CACHE + "*").build())) { + Long deviceId; + String cacheKey, tableName; + Map deviceMap, temp; + PersistenceHandler persistenceHandler; + while (cursor.hasNext()) { + deviceMap = operations.entries(cacheKey = cursor.next()); + deviceId = MapUtil.getLong(deviceMap, "deviceId"); + operations.put(PersistenceHandler.DEVICE_DATA_CACHE + deviceId, + "online", idGatewayMappingMap.containsKey(deviceId)); + + persistenceHandler = persistenceHandlerMap.get("KNPCV1"); + tableName = "t_device_data_" + deviceId; + temp = this.jdbcTemplate.queryForMap(EXISTS_TABLE_SQL, tableName); //$TableName$ + if (StringUtils.isBlank(MapUtil.getStr(temp, "TABLE_NAME"))) { + persistenceHandler.createTable(tableName); + } + persistenceHandler.insert(tableName, cacheKey); + } + } + } + +} diff --git a/src/main/resources/sql/CREATE_KNPCV1.sql b/src/main/resources/sql/CREATE_KNPCV1.sql new file mode 100644 index 0000000..69df33d --- /dev/null +++ b/src/main/resources/sql/CREATE_KNPCV1.sql @@ -0,0 +1,61 @@ +CREATE TABLE `$TableName$` +( + `id` bigint NOT NULL COMMENT '主键', + `device_id` int NOT NULL COMMENT '设备标识', + `created_time` datetime NOT NULL COMMENT '数据落库时间', + `collection_time` datetime NOT NULL COMMENT '采集指令下发时间', + `receive_time` datetime NOT NULL COMMENT '接收到数据时间', + `status` tinyint NULL DEFAULT NULL COMMENT '开关井。寄存器地址：0:0001，允许值：0：关井 ，1：开井', + `device_time` datetime NULL DEFAULT NULL COMMENT '设备时间。\r\n寄存器地址：3:0000，3:0001，3:0002，3:0003，3:0004，3:0005\r\n', + `run_mode` tinyint NULL DEFAULT NULL COMMENT '运行模式：\r\n0：手动模式 \r\n1：定时器模式 \r\n2：计时器模式 \r\n3：压力模式\r\n4：柱塞模式\r\n5：时压模式', + `gas_status` tinyint NULL DEFAULT NULL COMMENT '气井状态：\r\n0：关闭 \r\n1：打开', + `plug_status` tinyint NULL DEFAULT NULL COMMENT '柱塞状态：\r\n0：离开\r\n1：上升中\r\n2：到达', + `status_start_time` time NULL DEFAULT NULL COMMENT '当前状态开始时间', + `status_end_time` time NULL DEFAULT NULL COMMENT '当前状态结束时间', + `oil_pressure` decimal(10, 2) NULL DEFAULT NULL COMMENT '油压：单位Mpa', + `cas_pressure` decimal(10, 2) NULL DEFAULT NULL COMMENT '套压：单位Mpa', + `pre_pressure` decimal(10, 2) NULL DEFAULT NULL COMMENT '输压：单位Mpa', + `pipe_pressure` decimal(10, 2) NULL DEFAULT NULL COMMENT '管压：单位Mpa', + `liquid_level` decimal(10, 2) NULL DEFAULT NULL COMMENT '流量：单位mm', + `temperature` decimal(10, 2) NULL DEFAULT NULL COMMENT '温度：单位℃，*100，最高位表示正负，0正1负，即零下的温度加32768', + `humidity` decimal(10, 2) NULL DEFAULT NULL COMMENT '湿度：单位%', + `op_mode` tinyint NULL DEFAULT NULL COMMENT '运行模式：\r\n0： 手动模式 hand_mode\r\n1：定时器模式 timer_mode\r\n2：计时器模式 t2mode\r\n3：压力模式 pressure_mode\r\n4：柱塞模式 piston_mode\r\n5：时压模式 tp_mode', + `well_ctl` tinyint NULL DEFAULT NULL COMMENT '开关井，保持现有的模式，同显示屏按钮：\r\n0：关井 close_well\r\n1：开井 open_well', + `timer_able1` tinyint NULL DEFAULT NULL COMMENT '定时模式定时器1使能：\r\n0：禁止 disable\r\n1：使能 enable', + `timer_able2` tinyint NULL DEFAULT NULL COMMENT '定时模式定时器2使能：\r\n0：禁止 disable\r\n1：使能 ', + `timer_open1` time NULL DEFAULT NULL COMMENT '时0~23：分0~59：秒0~59', + `timer_close1` time NULL DEFAULT NULL COMMENT '时0~23：分0~59：秒0~59', + `timer_open2` time NULL DEFAULT NULL COMMENT '时0~23：分0~59：秒0~59', + `timer_close2` time NULL DEFAULT NULL COMMENT '时0~23：分0~59：秒0~59', + `timing_open` time NULL DEFAULT NULL COMMENT '时0~23：分0~59：秒0~59', + `timing_close` time NULL DEFAULT NULL COMMENT '时0~23：分0~59：秒0~59', + `timing_delay` time NULL DEFAULT NULL COMMENT '时0~23：分0~59：秒0~59', + `presource` tinyint NULL DEFAULT NULL COMMENT '触发压力源：\r\n0：油压 oil_pressure\r\n1：套压 cas_pressure\r\n2：输压 pre_pressure\r\n3：差压 diff_pressure', + `pressure_open` decimal(10, 2) NULL DEFAULT NULL COMMENT '开井压力：单位Mpa，*100', + `pressure_close` decimal(10, 2) NULL DEFAULT NULL COMMENT '关井压力： 单位Mpa，*100', + `trigger_type` tinyint NULL DEFAULT NULL COMMENT '压力触发类型：\n0：大于开井压力开井、小于关井压力关井 gre_less\n1：大于开井压力开井、大于关井压力关井 gre_gre\n2：小于开井压力开井、小于关井压力关井 less_less\n3：小于开井压力开井、大于关井压力关井 less_gre', + `stability_time` smallint NULL DEFAULT NULL COMMENT '传感器压力稳定时间：秒', + `max_open_well` time NULL DEFAULT NULL COMMENT '时0~999：分0~59：秒0~59', + `max_close_well` time NULL DEFAULT NULL COMMENT '时0~999：分0~59：秒0~59', + `min_open_well` time NULL DEFAULT NULL COMMENT '时0~999：分0~59：秒0~59', + `min_close_well` time NULL DEFAULT NULL COMMENT '时0~999：分0~59：秒0~59', + `prespro_tect` tinyint NULL DEFAULT NULL COMMENT '开井压力保护使能：\r\n0：禁止 disable\r\n1：使能 enable', + `prespro_source` tinyint NULL DEFAULT NULL COMMENT '开井压力保护源：\r\n0：油压 oil_pressure\r\n1：套压 cas_pressure\r\n2：输压 pre_pressure\r\n3：差压 diff_pressure', + `open_well_limit_max` decimal(10, 2) NULL DEFAULT NULL COMMENT '开井压力限制上限：单位MPa *100', + `open_well_limit_min` decimal(10, 2) NULL DEFAULT NULL COMMENT '开井压力限制下限：单位MPa *100', + `plug_init_status` tinyint NULL DEFAULT NULL COMMENT '柱塞模式初始气井状态：\r\n0：关井 close_well\r\n1：开井 open_well', + `plug_sustain_time` time NULL DEFAULT NULL COMMENT '时0~999：分0~59：秒0~59', + `plug_close_time` time NULL DEFAULT NULL COMMENT '时0~999：分0~59：秒0~59', + `tp_init_status` tinyint NULL DEFAULT NULL COMMENT '时压模式初始气井状态：\r\n0：关井 close_well\r\n1：开井 open_well', + `tp_open_source` tinyint NULL DEFAULT NULL COMMENT '时压模式开井源：\r\n0：油压 oil_pressure\r\n1：套压 cas_pressure\r\n2：输压 pre_pressure\r\n3：差压 diff_pressure\r\n4：时间 time_pressure', + `tp_open_trigger` tinyint NULL DEFAULT NULL COMMENT '时压模式开井压力触发模式：\r\n0：大于压力触发\r\n1：小于压力触发', + `tp_open_pressure` tinyint NULL DEFAULT NULL COMMENT '时压模式开井压力', + `tp_open_time` time NULL DEFAULT NULL COMMENT '时0~999：分0~59：秒0~59', + `tp_close_source` tinyint NULL DEFAULT NULL COMMENT '时压模式关井源：\r\n0：油压 oil_pressure\r\n1：套压 cas_pressure\r\n2：输压 pre_pressure\r\n3：差压 diff_pressure\r\n4：时间 time_pressure', + `tp_close_trigger` tinyint NULL DEFAULT NULL COMMENT '时压模式关井压力触发模式：\r\n0：大于压力触发 max_pressure\r\n1：小于压力触发 min_pressure', + `tp_close_pressure` smallint NULL DEFAULT NULL COMMENT '时压模式关井压力', + `tp_close_time` time NULL DEFAULT NULL COMMENT '时0~999：分0~59：秒0~59', + `tp_stability_time` tinyint NULL DEFAULT NULL COMMENT '时压模式压力稳定时长：秒', + PRIMARY KEY (`id`) USING BTREE, + UNIQUE INDEX `udx_device_create_time` (`device_id` ASC, `created_time` ASC) USING BTREE COMMENT '设备采集数据唯一键' +) ENGINE = InnoDB COMMENT = '设备ID：1的采集数据'; \ No newline at end of file diff --git a/src/main/resources/sql/INSERT_KNPCV1.sql b/src/main/resources/sql/INSERT_KNPCV1.sql new file mode 100644 index 0000000..459e9be --- /dev/null +++ b/src/main/resources/sql/INSERT_KNPCV1.sql @@ -0,0 +1,14 @@ +INSERT INTO `$TableName$` (`id`, `device_id`, `created_time`, `collection_time`, `receive_time`, `status`, + `device_time`, `run_mode`, `gas_status`, `plug_status`, `status_start_time`, + `status_end_time`, `oil_pressure`, `cas_pressure`, `pre_pressure`, `pipe_pressure`, + `liquid_level`, `temperature`, `humidity`, `op_mode`, `well_ctl`, `timer_able1`, + `timer_able2`, `timer_open1`, `timer_close1`, `timer_open2`, `timer_close2`, + `timing_open`, `timing_close`, `timing_delay`, `presource`, `pressure_open`, + `pressure_close`, `trigger_type`, `stability_time`, `max_open_well`, `max_close_well`, + `min_open_well`, `min_close_well`, `prespro_tect`, `prespro_source`, + `open_well_limit_max`, `open_well_limit_min`, `plug_init_status`, `plug_sustain_time`, + `plug_close_time`, `tp_init_status`, `tp_open_source`, `tp_open_trigger`, + `tp_open_pressure`, `tp_open_time`, `tp_close_source`, `tp_close_trigger`, + `tp_close_pressure`, `tp_close_time`, `tp_stability_time`) +VALUES (?, ?, NOW(), ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, + ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?); diff --git a/src/test/java/com/isu/gaswellwatch/GasWellWatchApplicationTests.java b/src/test/java/com/isu/gaswellwatch/GasWellWatchApplicationTests.java index 5368b93..c1ad583 100644 --- a/src/test/java/com/isu/gaswellwatch/GasWellWatchApplicationTests.java +++ b/src/test/java/com/isu/gaswellwatch/GasWellWatchApplicationTests.java @@ -1,13 +1,32 @@ package com.isu.gaswellwatch; +import org.apache.commons.io.IOUtils; import org.junit.jupiter.api.Test; -import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.util.ResourceUtils; -@SpringBootTest +import java.io.File; +import java.io.FileInputStream; +import java.io.IOException; +import java.nio.charset.StandardCharsets; + +//@SpringBootTest class GasWellWatchApplicationTests { @Test - void contextLoads() { + void contextLoads() throws Exception { + String fileName = "sql/CREATE_KNPCV1.sql"; + System.out.println(ResourceUtils.getURL(fileName)); + System.out.println(getResource(fileName)); + + } + + public static String getResource(String classPath) { + try { + File file = ResourceUtils.getFile("classpath:" + classPath); + return IOUtils.toString(new FileInputStream(file), StandardCharsets.UTF_8); + } catch (IOException e) { + throw new RuntimeException(e); + } } }