package com.zksy.gas.utils; import static com.zksy.common.utils.CollectionSideEffects.*; import cn.hutool.core.lang.UUID; import com.alibaba.fastjson.JSONObject; import com.zksy.api.domain.WarningThreshold; import com.zksy.api.service.EarlyWarningAutoService; import com.zksy.api.service.WarningThresholdService; import com.zksy.api.domain.AlarmData; import com.zksy.api.service.AlarmDataService; import com.zksy.api.utils.AlarmLevelUtil; import com.zksy.api.utils.SmsUtil; import com.zksy.common.exception.InvalidMessageException; import com.zksy.gas.domain.GasMonitorData; import com.zksy.gas.service.GasMonitorDataService; import com.zksy.api.service.DeviceContactService; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import io.netty.channel.ChannelHandler; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelInboundHandlerAdapter; import io.netty.handler.timeout.ReadTimeoutException; import io.netty.util.ReferenceCountUtil; import lombok.extern.slf4j.Slf4j; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; import org.springframework.http.HttpEntity; import org.springframework.http.HttpHeaders; import org.springframework.http.MediaType; import org.springframework.stereotype.Component; import org.springframework.web.client.RestTemplate; import java.math.BigDecimal; import java.time.LocalDateTime; import java.util.HashMap; import java.util.List; import java.util.Map; @ChannelHandler.Sharable @Slf4j @Component public class MessageHandler extends ChannelInboundHandlerAdapter { private static Logger logger = LoggerFactory.getLogger(MessageHandler.class); private final GasMonitorDataService service; @Autowired private WarningThresholdService warningThresholdService; @Autowired private AlarmDataService alarmDataService; @Autowired private EarlyWarningAutoService earlyWarningAutoService; @Autowired private SmsUtil smsUtil; @Autowired private DeviceContactService devicePhoneFetchUtil; @Autowired private AlarmUtil alarmUtil; @Autowired private RestTemplate restTemplate; @Autowired private DeviceOfflineCheckTask deviceOfflineCheckTask; @Value("${device.status.api.url:http://zk-api-service/equipmentStatus/updateAlarmStatus}") private String deviceStatusApiUrl; @Autowired public MessageHandler(GasMonitorDataService firefightingPressureService) { this.service = firefightingPressureService; } @Override public void channelActive(ChannelHandlerContext ctx) throws Exception { super.channelActive(ctx); //sendDataToDevice(ctx); } @Override public void channelRead(ChannelHandlerContext ctx, Object msg) { try { ByteBuf msgByteBuf = (ByteBuf) msg; if (msgByteBuf == null || !msgByteBuf.isReadable()) { logger.warn("接收到无效的消息"); return; } int readable = msgByteBuf.readableBytes(); byte[] msgBytes = new byte[readable]; msgByteBuf.readBytes(msgBytes); logger.info("接收到 {} 字节的数据: {}, 来自: {}", readable, ProtocolUtils.bytesToHex(msgBytes), ctx.channel().remoteAddress()); int checksumTotalLength = 2; // 末尾2字节校验码(如ED34) int bodyStartOffset = 16; // 校验和07的主体起始索引 int fullDataStartOffset = 3; // 校验和16的完整数据起始索引 // 校验报文长度合法性 if (readable < fullDataStartOffset + checksumTotalLength) { throw new InvalidMessageException("报文长度不足,无法完成校验"); } int bodyEndOffset = msgBytes.length - checksumTotalLength; // 避开末尾2字节校验码 int bodyLength = bodyEndOffset - bodyStartOffset; if (bodyLength <= 0) { throw new InvalidMessageException("主体数据长度无效,无法计算校验和07"); } int checksum07 = DataCheckUtil.calcSumOnly(msgBytes, bodyStartOffset, bodyLength); int fullDataEndOffset = msgBytes.length - 1; // 包含倒数第二个字节 int fullDataLength = fullDataEndOffset - fullDataStartOffset; if (fullDataLength <= 0) { throw new InvalidMessageException("完整数据长度无效,无法计算校验和16"); } int checksum16 = DataCheckUtil.calcChecksum(msgBytes, fullDataStartOffset, fullDataLength); int calculatedChecksum = (checksum07 << 8) | checksum16; int receivedChecksum = ((msgBytes[msgBytes.length - 2] & 0xFF) << 8) | (msgBytes[msgBytes.length - 1] & 0xFF); // 打印调试信息,确认拼接结果 logger.debug("校验和07(高8位,仅求和): {}", Integer.toHexString(checksum07).toUpperCase()); logger.debug("校验和16(低8位,FF-求和): {}", Integer.toHexString(checksum16).toUpperCase()); logger.debug("修改后拼接校验码: {}", Integer.toHexString(calculatedChecksum).toUpperCase()); logger.debug("报文携带校验码: {}", Integer.toHexString(receivedChecksum).toUpperCase()); if (calculatedChecksum != receivedChecksum) { throw new InvalidMessageException( String.format("数据校验失败:本地计算(07+16)=%04X(07=%02X, 16=%02X),报文携带=%04X", calculatedChecksum, checksum07, checksum16, receivedChecksum)); } com.zksy.gas.domain.GasMonitorData resultData = DataParser.parseMessage(msgBytes); resultData.setId(java.util.UUID.randomUUID().toString()); requireSaved(service.save(resultData)); // 更新设备最后接收时间,并标记设备在线 run(resultData.getMacAddress(), "接收时间更新", () -> DeviceOfflineCheckTask.deviceLastReceiveTimeMap.put(resultData.getMacAddress(), new java.util.Date())); run(resultData.getMacAddress(), "在线状态同步", () -> deviceOfflineCheckTask.markDeviceOnline(resultData.getMacAddress())); run(null, "报警预警评估", () -> checkIfSmsAlertNeeded(resultData)); logger.info("数据解析入库成功: {}", resultData); } catch (InvalidMessageException e) { logger.error("数据校验失败: {}", e.getMessage()); ctx.writeAndFlush(Unpooled.copiedBuffer("数据校验失败".getBytes())); } catch (Exception e) { logger.error("数据解析/入库异常", e); ctx.writeAndFlush(Unpooled.copiedBuffer("数据处理异常".getBytes())); } finally { ReferenceCountUtil.release(msg); } } private String printHexBinary(byte[] bytes) { StringBuilder sb = new StringBuilder(); for (byte b : bytes) { sb.append(String.format("%02X ", b)); } System.out.println("Received raw data: " + sb.toString()); logger.info("Received raw data: " + sb.toString()); return sb.toString(); } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { if (cause instanceof ReadTimeoutException) { logger.info("来自" + ctx.channel().remoteAddress() + "的连接超时断开"); } else { logger.error("设备连接异常", cause); logger.info("来自" + ctx.channel().remoteAddress() + "的连接异常断开"); ctx.close(); } } @Override public void channelReadComplete(ChannelHandlerContext ctx) throws Exception { ctx.flush(); } @Override public void channelUnregistered(ChannelHandlerContext ctx) throws Exception { logger.info("来自" + ctx.channel().remoteAddress() + "的连接主动断开"); ctx.fireChannelUnregistered(); } // 主动向设备发送数据 public void sendDataToDevice(ChannelHandlerContext ctx) { try { // 将字符串形式的十六进制数据转换为字节数组 String hexData = "01 03 00 00 00 01 84 0A "; String[] hexArray = hexData.split(" "); byte[] dataBytes = new byte[hexArray.length]; for (int i = 0; i < hexArray.length; i++) { dataBytes[i] = (byte) Integer.parseInt(hexArray[i], 16); } ByteBuf byteBuf = Unpooled.copiedBuffer(dataBytes); ctx.writeAndFlush(byteBuf); logger.info("已向设备发送数据: {}", printHexBinary(dataBytes)); } catch (Exception e) { logger.error("发送数据失败: {}", e.getMessage()); } } public void checkIfSmsAlertNeeded(GasMonitorData resultData){ String deviceId = resultData == null ? null : resultData.getMacAddress(); try { if (resultData == null || deviceId == null || deviceId.trim().isEmpty()) { log.warn("可燃气体报警数据缺少设备编码,跳过状态覆盖"); return; } boolean hasAlarm = evaluateAlarmStatus(resultData); updateDeviceAlarmStatus(deviceId, hasAlarm ? 1 : 0); } catch (Exception e) { // 评估不完整时不覆盖上一状态,避免异常数据误清除报警 log.error("可燃气体设备报警处理失败,保留原报警状态:deviceCode={}", deviceId, e); } } /** 评估单条采集数据是否存在报警。只有完整评估成功才允许同步设备状态。 */ private boolean evaluateAlarmStatus(GasMonitorData resultData){ //用MAC地址作为设备的设备编号 String deviceId = resultData.getMacAddress(); boolean hasAlarm = false; //获取气体浓度阈值(按气体类型只监测浓度,当前仅CH4;且仅按上限判定) String gasWarningCode = "WARN-GAS-CONCENTRATION"; WarningThreshold gasThreshold = checkThreshold(deviceId, gasWarningCode); Double gasMaxValue = gasThreshold != null ? gasThreshold.getMaxValue() : null; String gasWarningType = "气体浓度预警"; String gasRemark = null; //判断气体浓度:仅当浓度达到或超过上限时视为越限;浓度低(接近0)属正常,不告警 BigDecimal gasConcentration = resultData.getGasConcentration(); if (gasConcentration == null) { throw new IllegalArgumentException("缺少气体浓度值"); } if (gasConcentration != null && gasMaxValue != null) { BigDecimal gasMaxBd = BigDecimal.valueOf(gasMaxValue); if (gasConcentration.floatValue() >= gasMaxValue) { // 越限报警:等级计算以0为下限基线(气体浓度不会低于0),保证仅上限时也能正确分级 int alarmLevel = saveAlarmData(deviceId, gasWarningType, gasWarningCode, BigDecimal.ZERO, gasMaxBd, gasConcentration, gasRemark != null ? gasRemark : "可燃气体浓度报警"); run(deviceId, gasWarningCode + ":关闭预警", () -> earlyWarningAutoService.closeAutoWarning(deviceId, gasWarningCode, "已达报警阈值,转报警处理")); hasAlarm = true; if (alarmLevel == 1) { run(deviceId, gasWarningCode + ":短信通知", () -> sendMessage(deviceId, gasWarningType, resultData.getLongitude(), resultData.getLatitude())); } } else { // 未越限:接近上限时生成自动预警(min传null,仅按上限评估预警区) run(deviceId, gasWarningCode + ":告警恢复", () -> alarmDataService.recoverThresholdAlarm( deviceId, gasWarningCode, null, gasMaxBd, gasConcentration)); run(deviceId, gasWarningCode + ":生成预警", () -> earlyWarningAutoService.checkAndGenerate(deviceId, gasWarningType, gasWarningCode, null, gasMaxBd, gasConcentration, gasRemark)); } } return hasAlarm; } private WarningThreshold checkThreshold(String deviceCode, String warningCode) { return warningThresholdService.getWarningThresholdByDeviceAndCode(deviceCode, warningCode); } private int saveAlarmData(String deviceCode, String warningType, String warningCode, BigDecimal minValue, BigDecimal maxValue, BigDecimal actualValue, String remark) { int alarmLevel = 4; try { AlarmData alarmData = new AlarmData(); alarmData.setDeviceCode(deviceCode); alarmData.setWarningCode(warningCode); alarmData.setMinValue(minValue); alarmData.setMaxValue(maxValue); alarmData.setActualValue(actualValue); alarmLevel = AlarmLevelUtil.calculateLevel(actualValue, minValue, maxValue); BigDecimal deviationRatio = AlarmLevelUtil.calculateDeviationRatio(actualValue, minValue, maxValue); alarmData.setAlarmLevel(alarmLevel); alarmData.setDeviationRatio(deviationRatio); alarmData.setAlarmStatus(0); alarmData.setCreateTime(LocalDateTime.now()); alarmData.setCreateTime(LocalDateTime.now()); boolean saved = run(deviceCode, warningCode + ":报警入库", () -> requireSaved(alarmDataService.saveAlarmData(alarmData))); if (alarmLevel == 1) { log.warn("【1级紧急告警】deviceCode={}, warningCode={}, deviationRatio={}%", deviceCode, warningCode, deviationRatio); } if (saved) { log.info("告警数据入库成功: deviceCode={}, warningCode={}, level={}级", deviceCode, warningCode, alarmLevel); } } catch (Exception e) { log.error("保存告警数据失败", e); } return alarmLevel; } //构造发送短信参数&&发送短信 public void sendMessage(String deviceNo, String alarmType, BigDecimal lng, BigDecimal lat){ // 构造短信参数 JSONObject params = new JSONObject(); params.put("deviceNo", deviceNo); params.put("alarmType", alarmType); // 经纬度处理(位置) params.put("location", String.format("经纬度:%.6f,%.8f", lng, lat)); List phoneListByDeviceId = devicePhoneFetchUtil.getPhoneListByDeviceCode(deviceNo); Map sendResults = smsUtil.sendBatchSms(phoneListByDeviceId, params.toJSONString()); log.info("发送信息完成"); } /** * 更新设备报警状态 * @param deviceCode 设备编码 * @param alarmStatus 报警状态:0-正常,1-报警 */ private void updateDeviceAlarmStatus(String deviceCode, int alarmStatus) { try { Map params = new HashMap<>(); params.put("deviceCode", deviceCode); params.put("alarmStatus", alarmStatus); HttpHeaders headers = new HttpHeaders(); headers.setContentType(MediaType.APPLICATION_JSON); HttpEntity> request = new HttpEntity<>(params, headers); Map response = restTemplate.postForObject(deviceStatusApiUrl, request, Map.class); if (response == null || !Boolean.TRUE.equals(response.get("success"))) { log.warn("设备报警状态更新未成功: deviceCode={}, alarmStatus={}, response={}", deviceCode, alarmStatus, response); return; } log.info("设备报警状态更新成功: deviceCode={}, alarmStatus={}", deviceCode, alarmStatus); } catch (Exception e) { log.error("更新设备报警状态失败: deviceCode={}, alarmStatus={}", deviceCode, alarmStatus, e); } } }