package com.zksy.pressure.utils; import static com.zksy.common.utils.CollectionSideEffects.*; import cn.hutool.core.lang.UUID; import com.alibaba.fastjson.JSONObject; import com.aliyun.apache.hc.client5.http.utils.Hex; import com.zksy.api.domain.AlarmData; import com.zksy.api.domain.WarningThreshold; import com.zksy.api.service.AlarmDataService; import com.zksy.api.service.EarlyWarningAutoService; import com.zksy.api.service.WarningThresholdService; import com.zksy.api.utils.AlarmLevelUtil; import com.zksy.api.utils.EarlyWarningLevelUtil; import com.zksy.api.utils.SmsUtil; import com.zksy.common.exception.InvalidMessageException; import com.zksy.pressure.domain.FirefightingPressure; import com.zksy.pressure.service.FirefightingPressureService; import com.zksy.api.service.DeviceContactService; import io.netty.buffer.ByteBuf; import io.netty.util.ReferenceCountUtil; 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 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.Date; 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); /** 压力值预警编码,阈值取自 app_user.warning_threshold(device_code + warning_code) */ private static final String PRESSURE_WARNING_CODE = "WARN-FIRE-PRESSURE"; private static final String PRESSURE_WARNING_TYPE = "消防压力监测"; private final FirefightingPressureService firefightingPressureService; @Autowired private AlarmDataService alarmDataService; @Autowired private WarningThresholdService warningThresholdService; @Autowired private EarlyWarningAutoService earlyWarningAutoService; @Autowired private SmsUtil smsUtil; @Autowired private DeviceContactService devicePhoneFetchUtil; @Autowired private RestTemplate restTemplate; @Autowired private DeviceOfflineCheckTask deviceOfflineCheckTask; @Value("${device.status.api.url}") private String deviceStatusApiUrl; @Autowired public MessageHandler(FirefightingPressureService firefightingPressureService) { this.firefightingPressureService = 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; } logger.info("接收到 {} 字节的数据,来自: {}", msgByteBuf.readableBytes(), ctx.channel().remoteAddress()); byte[] msgBytes = new byte[msgByteBuf.readableBytes()]; msgByteBuf.readBytes(msgBytes); String result = printHexBinary(msgBytes); String msgString = result.replaceAll("\\s", ""); String CRCString = msgString.substring(msgString.length() - 4); String bodyResult = msgString.substring(0, msgString.length() - 4); String codeString = DataCheckUtil.crc16Tall(bodyResult); System.out.println("校验码:" + CRCString + "-----------生成的校验码:" + codeString); if (!CRCString.equals(codeString)) { throw new InvalidMessageException("数据校验不成功"); } else { String hexString = Hex.encodeHexString(msgBytes).toUpperCase(); String uploadEndFlag = hexString.substring(hexString.length() - 6, hexString.length() - 4); logger.info("成功!!!接收到 {} 字节的数据,来自: {}", msgString, ctx.channel().remoteAddress()); FirefightingPressure resultData = DataParser.parseMessage(msgString); String functionCode = msgString.substring(20, 22); logger.info("生成的回复报文功能码: {}", functionCode); if("EF".equals(functionCode)){ String responseMsg = buildResponseMessage(resultData.getTelemeteringStation(), resultData.getCentralStation(), resultData.getPassword(),resultData.getFunctionCode(), "0000", uploadEndFlag); logger.info("生成的回复报文: {}", responseMsg); byte[] messageBytes = hexStringToByteArray(responseMsg); ctx.channel().writeAndFlush(Unpooled.copiedBuffer(messageBytes)); return; } resultData.setId(UUID.randomUUID().toString()); // 更新设备最后一次接收数据的时间,并标记设备在线 run(resultData.getTelemeteringStation(), "接收时间更新", () -> DeviceOfflineCheckTask.deviceLastReceiveTimeMap.put(resultData.getTelemeteringStation(), new Date())); requireSaved(firefightingPressureService.save(resultData)); run(resultData.getTelemeteringStation(), "采集应答", () -> { // 4. 构建回复报文 String responseMsg = buildResponseMessage(resultData.getTelemeteringStation(), resultData.getCentralStation(), resultData.getPassword(), resultData.getFunctionCode(), resultData.getSerialNumber(), uploadEndFlag); logger.info("生成的回复报文: {}", responseMsg); byte[] messageBytes = hexStringToByteArray(responseMsg); ctx.channel().writeAndFlush(Unpooled.copiedBuffer(messageBytes)); }); run(resultData.getTelemeteringStation(), "在线状态同步", () -> deviceOfflineCheckTask.markDeviceOnline(resultData.getTelemeteringStation())); // 阈值监测:报警 + 接近阈值自动预警 run(null, "报警预警评估", () -> checkIfSmsAlertNeeded(resultData)); } } catch (InvalidMessageException e) { logger.error("数据入库失败: {}", e.getMessage()); ctx.writeAndFlush(Unpooled.copiedBuffer("数据入库失败".getBytes())); } catch (Exception e) { logger.error("当前采集报文处理失败,连接继续接收后续数据", e); } finally { ReferenceCountUtil.release(msg); } } private byte[] hexStringToByteArray(String hexString) { // 校验输入合法性 if (hexString == null || hexString.length() % 2 != 0) { throw new IllegalArgumentException("十六进制字符串长度必须为偶数,输入:" + hexString); } int len = hexString.length(); byte[] data = new byte[len / 2]; // 每2个十六进制字符转换为1个字节 for (int i = 0; i < len; i += 2) { data[i / 2] = (byte) ((Character.digit(hexString.charAt(i), 16) << 4) + Character.digit(hexString.charAt(i + 1), 16)); } return data; } /** * 构建回复报文 * @param telemetryAddr 遥测站地址(来自上传报文) * @param centerAddr 中心站地址(来自上传报文) * @param password 密码(来自上传报文) * @param funcCode 功能码(来自上传报文) * @param serialNo 流水号(来自上传报文) * @param uploadEndFlag 上传报文结束符(03/17) * @return 完整的回复报文十六进制字符串(包含CRC) */ private String buildResponseMessage(String telemetryAddr, String centerAddr, String password, String funcCode, String serialNo, String uploadEndFlag) { // 1. 固定字段 String frameHeader = "7E7E"; // 帧头 String downLen = "8008"; // 下行/载荷长度 String msgStartFlag = "02"; // 报文起始符 // 2. 动态字段 - 发报时间(当前时间,格式:年月日时分秒,十六进制) String sendTime = DataCheckUtil.getCurrentTimeHex(); // 3. 回复报文结束符(上传是03则回复04,上传是17则回复06) String responseEndFlag = "04"; if ("17".equals(uploadEndFlag)) { responseEndFlag = "06"; } // 4. 拼接回复报文主体(不含CRC) StringBuilder bodyBuilder = new StringBuilder(); bodyBuilder.append(frameHeader) // 帧头 7E7E .append(telemetryAddr) // 遥测站地址 .append(centerAddr) // 中心站地址 .append(password) // 密码 .append(funcCode) // 功能码 .append(downLen) // 下行/载荷长度 .append(msgStartFlag) // 报文起始符 .append(serialNo) // 流水号 .append(sendTime) // 发报时间 .append(responseEndFlag); // 报文结束符 String body = bodyBuilder.toString(); System.out.println("回复报文主体:" + body); // 5. 计算CRC16校验码 String crc = DataCheckUtil.crc16Tall(body); // 6. 拼接最终报文(主体 + CRC) return body + crc; } 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()); } } /** * 压力值阈值评估:不再使用设备上报的 D14~D10 五种压力报警标志, * 统一按 pressure_value 与 WARN-FIRE-PRESSURE 配置阈值判定。 */ public void checkIfSmsAlertNeeded(FirefightingPressure data) { if (data == null) { log.warn("消防压力数据为空,跳过阈值预警评估"); return; } String deviceCode = data.getTelemeteringStation() == null ? null : data.getTelemeteringStation().trim(); if (deviceCode == null || deviceCode.isEmpty()) { log.warn("消防压力数据缺少遥测站地址,跳过阈值预警评估"); return; } // 评估失败(阈值未配置、压力值缺失)时返回 null,保留设备上一报警状态 Boolean status = evaluate(deviceCode, PRESSURE_WARNING_CODE + ":压力阈值评估", () -> evaluatePressure(deviceCode, data)); if (status != null) { updateDeviceAlarmStatus(deviceCode, status ? 1 : 0); } } /** * 按阈值判定本次压力值:越限写报警并关闭预警;未越限则处理告警恢复并生成接近阈值预警。 * * @return true 报警,false 正常/预警区 */ private boolean evaluatePressure(String deviceCode, FirefightingPressure data) { if (warningThresholdService == null || alarmDataService == null || earlyWarningAutoService == null) { throw new IllegalStateException("阈值预警依赖未注入"); } if (data.getPressureValue() == null) { throw new IllegalArgumentException("采集报文缺少压力值,无法判定报警状态"); } BigDecimal actual = BigDecimal.valueOf(data.getPressureValue()); WarningThreshold threshold = warningThresholdService .getWarningThresholdByDeviceAndCode(deviceCode, PRESSURE_WARNING_CODE); if (threshold == null || (threshold.getMinValue() == null && threshold.getMaxValue() == null)) { throw new IllegalStateException("消防压力阈值未配置,无法判定报警状态: deviceCode=" + deviceCode); } BigDecimal min = threshold.getMinValue() == null ? null : BigDecimal.valueOf(threshold.getMinValue()); BigDecimal max = threshold.getMaxValue() == null ? null : BigDecimal.valueOf(threshold.getMaxValue()); // 单侧阈值时补齐等级/预警区计算基线:压力不会为负,仅上限以 0 作下限;仅下限以两倍下限作上限, // 保证只配一侧时等级计算仍有可比区间(与可燃气体仅按上限判定的处理保持一致)。 BigDecimal levelMin = min != null ? min : BigDecimal.ZERO; BigDecimal levelMax = max != null ? max : min.multiply(BigDecimal.valueOf(2)); if (EarlyWarningLevelUtil.isOverThreshold(actual, min, max)) { int level = saveAlarmData(deviceCode, PRESSURE_WARNING_TYPE, PRESSURE_WARNING_CODE, levelMin, levelMax, actual, "消防压力值越限报警"); run(deviceCode, PRESSURE_WARNING_CODE + ":关闭预警", () -> earlyWarningAutoService.closeAutoWarning(deviceCode, PRESSURE_WARNING_CODE, "已达报警阈值,转报警处理")); if (level == 1) { run(deviceCode, PRESSURE_WARNING_CODE + ":一级告警短信", () -> sendMessage(deviceCode, PRESSURE_WARNING_TYPE, data.getLongitude(), data.getLatitude())); } return true; } run(deviceCode, PRESSURE_WARNING_CODE + ":告警恢复", () -> alarmDataService.recoverThresholdAlarm(deviceCode, PRESSURE_WARNING_CODE, levelMin, levelMax, actual)); run(deviceCode, PRESSURE_WARNING_CODE + ":生成预警", () -> earlyWarningAutoService.checkAndGenerate(deviceCode, PRESSURE_WARNING_TYPE, PRESSURE_WARNING_CODE, levelMin, levelMax, actual, null)); return false; } 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.setAlarmContent(buildAlarmContent(warningType, actualValue, minValue, maxValue, remark)); alarmData.setAlarmStatus(0); alarmData.setCreateTime(LocalDateTime.now()); alarmData.setUpdateTime(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; } /** 告警内容与其他采集服务保持一致:指标名称 + 越限方向,便于前端直接展示。 */ private String buildAlarmContent(String warningType, BigDecimal actualValue, BigDecimal minValue, BigDecimal maxValue, String remark) { String content = warningType == null || warningType.trim().isEmpty() ? remark : warningType.trim(); if (actualValue == null) { return content; } if (minValue != null && actualValue.compareTo(minValue) <= 0) { return content + "达到或低于下限"; } if (maxValue != null && actualValue.compareTo(maxValue) >= 0) { return content + "达到或超过上限"; } return content; } //构造发送短信参数&&发送短信 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", (lng != null && lat != null) ? 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); } } }