package com.ard.utils.netty.tcp; import com.alibaba.fastjson2.JSON; import com.ard.alarm.radar.domain.ArdAlarmRadar; import com.ard.alarm.radar.domain.ArdEquipRadar; import com.ard.alarm.radar.domain.RadarAlarmData; import com.ard.utils.util.ByteUtils; import com.ard.utils.util.GisUtils; import com.ard.utils.mqtt.MqttProducer; import io.netty.buffer.ByteBuf; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelId; import io.netty.channel.SimpleChannelInboundHandler; import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.StringUtils; import javax.xml.bind.DatatypeConverter; import java.net.InetSocketAddress; import java.text.SimpleDateFormat; import java.util.*; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; import static com.ard.utils.util.ByteUtils.toLittleEndian; /** * @Description: 客户端处理器 * @ClassName: ClientHandler * @Author: 刘苏义 * @Date: 2023年07月05日13:13 * @Version: 1.0 **/ @Slf4j(topic = "netty") public class ClientHandler extends SimpleChannelInboundHandler { private ChannelHandlerContext context; private ScheduledFuture heartbeatTask; /** * 连接建立 * * @param ctx * @throws Exception */ @Override public void channelActive(ChannelHandlerContext ctx) { context = ctx; startHeartbeatTask();//开始发送心跳 } /** * 连接断开 * * @param ctx * @throws Exception */ @Override public void channelInactive(ChannelHandlerContext ctx) throws Exception { InetSocketAddress ipSocket = (InetSocketAddress) ctx.channel().remoteAddress(); String ipPort = ipSocket.getHostString() + ":" + ipSocket.getPort(); log.error("与设备" + ipPort + "连接断开!"); // 连接断开后的最后处理 ctx.pipeline().remove(this); ctx.deregister(); ctx.close(); // 将失败信息插入Set集合 ArdEquipRadar radar = ClientInitialize.trueConnectMap.get(ipPort); if (radar != null) { ClientInitialize.falseConnectSet.add(radar); ClientInitialize.trueConnectMap.remove(ipPort); } super.channelInactive(ctx); } /** * 通道数据读取 * 业务操作 * * @param ctx * @param msg * @throws Exception */ @Override public void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) { InetSocketAddress ipSocket = (InetSocketAddress) ctx.channel().remoteAddress(); String ipPort = ipSocket.getHostString() + ":" + ipSocket.getPort(); ArdEquipRadar radar = ClientInitialize.trueConnectMap.get(ipPort); if (radar == null) { return; } MessageHandler messageHandler = ClientInitialize.SucMessageHandlerMap.get(ipPort); if (messageHandler == null) { return; } // 处理接收到的消息 byte[] byteArray = new byte[msg.readableBytes()]; msg.getBytes(msg.readerIndex(), byteArray); byte[] bytes = messageHandler.receiveCompletePacket(byteArray); if (bytes != null) { processData(radar, bytes); } } /** * 通道数据处理完成 * * @param ctx * @throws Exception */ @Override public void channelReadComplete(ChannelHandlerContext ctx) throws Exception { super.channelReadComplete(ctx); } /** * 事件触发 * * @param ctx * @param evt * @throws Exception */ @Override public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception { super.userEventTriggered(ctx, evt); } /** * 异常触发 * * @param ctx * @param cause * @throws Exception */ @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { // 发生异常时的处理 cause.printStackTrace(); ctx.close(); stopHeartbeatTask();//停止心跳发送 } /** * 开始心跳任务 */ private void startHeartbeatTask() { heartbeatTask = context.executor().scheduleAtFixedRate(() -> { // 发送心跳消息 ByteBuf message = context.alloc().buffer(); byte[] header = {0x01, 0x02, 0x01}; byte[] payload = {0x10, 0x00, 0x00, 0x00}; byte[] payloadCrc32 = ByteUtils.parseCrc32(payload); byte[] footer = {0x01, 0x02, 0x00}; byte[] heart = ByteUtils.appendArrays(header, payload, payloadCrc32, footer); //byte[] heart = {0x01, 0x02, 0x01, 0x10, 0x00, 0x00, 0x00, (byte) 0x83, (byte) 0x88, 0x5d, 0x71, 0x01, 0x02, 0x00}; String hexString = DatatypeConverter.printHexBinary(heart); // log.debug("发送心跳:" + hexString); message.writeBytes(heart); context.writeAndFlush(message); }, 0, 5, TimeUnit.SECONDS); } /** * 停止心跳任务 */ private void stopHeartbeatTask() { if (heartbeatTask != null) { heartbeatTask.cancel(false); heartbeatTask = null; } } /** * 解析报警数据 */ public void processData(ArdEquipRadar radar, byte[] data) { try { String radarId = radar.getId(); String radarName = radar.getName(); Double radarLongitude = radar.getLongitude(); Double radarLagitude = radar.getLatitude(); Double radarAltitude = radar.getAltitude(); //region crc校验-目前仅用于显示校验结果 Boolean crc32Check = MessageHandler.CRC32Check(data); if (!crc32Check) { log.debug("CRC32校验不通过"); } else { //log.debug("CRC32校验通过"); } //endregion //log.info("原始数据:" + DatatypeConverter.printHexBinary(data)); //log.info("雷达信息:" + host + "【port】" + port + "【X】" + longitude + "【Y】" + lagitude + "【Z】" + altitude); data = MessageHandler.transferData(data);//去掉包头和包尾、校验及转义 //region 负载头解析 byte[] type = Arrays.copyOfRange(data, 0, 1);//命令类型 // log.info("命令类型:" + DatatypeConverter.printHexBinary(type)); byte[] cmdId = Arrays.copyOfRange(data, 1, 2);//命令ID String cmdIdStr = DatatypeConverter.printHexBinary(cmdId); //log.info("命令ID:" + DatatypeConverter.printHexBinary(cmdId)); byte[] payloadSize = Arrays.copyOfRange(data, 2, 4);//有效负载大小 payloadSize = toLittleEndian(payloadSize); //log.info("payloadSize:" + DatatypeConverter.printHexBinary(payloadSize)); int payloadSizeToDecimal = ByteUtils.bytesToDecimal(payloadSize); // log.info("有效负载大小(转整型):" + payloadSizeToDecimal); //endregion List radarAlarmInfos = new ArrayList<>(); List radarFollowInfos = new ArrayList<>(); //抽油机状态雷达推送集合 List well = new ArrayList<>(); String alarmTime = ""; Integer targetNum = 0; log.debug("处理雷达" + radarName + "数据-->命令ID:" + cmdIdStr); //雷达移动防火报警 if (Arrays.equals(cmdId, new byte[]{0x01})) { //region 告警信息反馈 byte[] dwTim = Arrays.copyOfRange(data, 4, 8); dwTim = toLittleEndian(dwTim); SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); long l = ByteUtils.bytesToDecimal(dwTim); alarmTime = sdf.format(l * 1000); // log.info("周视图像的出现时间(转date):" + alarmTime); byte[] wTargetNum = Arrays.copyOfRange(data, 8, 10); wTargetNum = toLittleEndian(wTargetNum); targetNum = ByteUtils.bytesToDecimal(wTargetNum); if (targetNum == 0) { return; } //log.debug("目标总点数(转整型):" + targetNum); //解析NET_TARGET_UNIT(64是NET_TARGET_HEAD的字节数) int uintSize = (payloadSizeToDecimal - 64) / targetNum; // log.info("单条报警大小:" + uintSize); for (int i = 0; i < targetNum; i++) { Integer index = 68 + uintSize * i; byte[] dwID = Arrays.copyOfRange(data, index, index + 4); // log.info("目标ID:" + DatatypeConverter.printHexBinary(cmdId)); dwID = toLittleEndian(dwID); int targetId = ByteUtils.bytesToDecimal(dwID); // log.info("目标ID号:" + targetId); byte[] iDistance = Arrays.copyOfRange(data, index + 8, index + 12); iDistance = toLittleEndian(iDistance); double Distance = ByteUtils.bytesToDecimal(iDistance); //log.debug("目标当前直线距离(m):" + Distance); //region 不需要的字段 // byte[] dwGSum = Arrays.copyOfRange(data, index + 4, index + 8); // dwGSum = toLittleEndian(dwGSum); // int GSum = byteArrayToDecimal(dwGSum); // log.info("目标当前像素灰度和:" + GSum); // byte[] iTw = Arrays.copyOfRange(data, index + 12, index + 16); // iTw = toLittleEndian(iTw); // int Tw = byteArrayToDecimal(iTw); // log.info("目标当前的像素宽度:" + Tw); // // byte[] iTh = Arrays.copyOfRange(data, index + 16, index + 20); // iTh = toLittleEndian(iTh); // int Th = byteArrayToDecimal(iTh); // log.info("目标当前的像素高度:" + Th); // // byte[] wPxlArea = Arrays.copyOfRange(data, index + 20, index + 22); // wPxlArea = toLittleEndian(wPxlArea); // int PxlArea = byteArrayToDecimal(wPxlArea); // log.info("目标当前像素面积:" + PxlArea); // // byte[] cTrkNum = Arrays.copyOfRange(data, index + 22, index + 23); // cTrkNum = toLittleEndian(cTrkNum); // int TrkNum = byteArrayToDecimal(cTrkNum); // log.info("轨迹点数:" + TrkNum); // byte[] sVx = Arrays.copyOfRange(data, index + 24, index + 26); // sVx = toLittleEndian(sVx); // int Vx = byteArrayToDecimal(sVx); // log.info("目标当前速度矢量(像素距离)X:" + Vx); // // byte[] sVy = Arrays.copyOfRange(data, index + 26, index + 28); // sVy = toLittleEndian(sVy); // int Vy = byteArrayToDecimal(sVy); // log.info("目标当前速度矢量(像素距离)Y:" + Vy); // // byte[] sAreaNo = Arrays.copyOfRange(data, index + 28, index + 30); // sAreaNo = toLittleEndian(sAreaNo); // int AreaNo = byteArrayToDecimal(sAreaNo); // log.info("目标归属的告警区域号:" + AreaNo); // // byte[] cGrp = Arrays.copyOfRange(data, index + 30, index + 31); // cGrp = toLittleEndian(cGrp); // int Grp = byteArrayToDecimal(cGrp); // log.info("所属组:" + Grp); //endregion String alarmType = ""; byte[] cStat = Arrays.copyOfRange(data, index + 23, index + 24); cStat = toLittleEndian(cStat); // 提取第4位至第6位的值 int extractedValue = (cStat[0] >> 4) & 0b00001111; // 判断提取的值 if (extractedValue == 0b0000) { alarmType = "运动目标检测"; } else if (extractedValue == 0b0001) { alarmType = "热源检测"; } // log.info("报警类型:" + alarmType); byte[] szName = Arrays.copyOfRange(data, index + 64, index + 96); String alarmPointName = ByteUtils.bytesToStringZh(szName); // log.info("所属告警区域名称:" + alarmPointName); byte[] afTx = Arrays.copyOfRange(data, index + 96, index + 100); afTx = toLittleEndian(afTx); float fTx = ByteUtils.bytesToFloat(afTx); // log.info("水平角度:" + fTx); byte[] afTy = Arrays.copyOfRange(data, index + 112, index + 116); afTy = toLittleEndian(afTy); float fTy = ByteUtils.bytesToFloat(afTy); //log.debug("垂直角度:" + fTy); // 将角度转换为弧度 double thetaRadians = Math.toRadians(fTy + 90); // 使用正弦函数计算对边长度 Distance = Math.sin(thetaRadians) * Distance; if (Distance < 0) { continue;//过滤距离小于0的脏数据 } //log.debug("目标投影距离(m):" + Distance); Double[] radarXY = {radarLongitude, radarLagitude}; Double[] alarmXY = GisUtils.CalculateCoordinates(radarXY, Distance, (double) fTx); log.debug("报警信息:" + "【radarName】" + radarName + "【targetId】" + targetId + "【alarmType】" + alarmType + "【alarmTime】" + alarmTime + "【name】" + alarmPointName + "【Distance】" + Distance); ArdAlarmRadar ardAlarmRadar = new ArdAlarmRadar(); ardAlarmRadar.setTargetId(targetId); ardAlarmRadar.setName(alarmPointName); ardAlarmRadar.setLongitude(alarmXY[0]); ardAlarmRadar.setLatitude(alarmXY[1]); ardAlarmRadar.setAlarmType(alarmType); radarAlarmInfos.add(ardAlarmRadar); int bit1 = (cStat[0] >> 1) & 0x1; //目标的B1=1 锁定 if (bit1 == 1) { radarFollowInfos.add(ardAlarmRadar); } } //endregion if (StringUtils.isEmpty(alarmTime)) { return; } if (targetNum == 0) { return; } RadarAlarmData radarAlarmData = new RadarAlarmData(); radarAlarmData.setRadarId(radarId); radarAlarmData.setRadarName(radarName); radarAlarmData.setAlarmTime(alarmTime); radarAlarmData.setArdAlarmRadars(radarAlarmInfos); MqttProducer.publish(2, false, "radar", JSON.toJSONString(radarAlarmData)); if (radarFollowInfos.size() > 0) { radarAlarmData.setArdFollowRadars(radarFollowInfos); //当前雷达扫描存在引导跟踪数据,只保留最后一次锁定的数据 MqttProducer.publish(2, false, "radarFollowGuide", JSON.toJSONString(radarAlarmData)); } //抽油机状态MQTT队列 radarAlarmData.setArdAlarmRadars(well); MqttProducer.publish(2, false, "radarWellData", JSON.toJSONString(radarAlarmData)); } //抽油机AI状态反馈 if (Arrays.equals(cmdId, new byte[]{0x04})) { //region抽油机AI状态反馈 String hexString = DatatypeConverter.printHexBinary(data); //log.info(hexString); byte[] dwTim = Arrays.copyOfRange(data, 4, 8); dwTim = toLittleEndian(dwTim); SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); long l = ByteUtils.bytesToDecimal(dwTim); alarmTime = sdf.format(l * 1000); //log.info("周视图像的出现时间(转date):" + alarmTime); byte[] wTargetNum = Arrays.copyOfRange(data, 8, 10); wTargetNum = toLittleEndian(wTargetNum); targetNum = ByteUtils.bytesToDecimal(wTargetNum); //log.debug("目标总点数(转整型):" + targetNum); if (targetNum == 0) { return; } //解析NET_TARGET_UNIT(64是NET_TARGET_HEAD的字节数) int uintSize = (payloadSizeToDecimal - 64) / targetNum; //log.info("单条报警大小:" + uintSize); for (int i = 0; i < targetNum; i++) { Integer index = 68 + uintSize * i; byte[] dwID = Arrays.copyOfRange(data, index, index + 4); //log.info("目标ID:" + DatatypeConverter.printHexBinary(dwID)); dwID = toLittleEndian(dwID); int targetId = ByteUtils.bytesToDecimal(dwID); //log.info("目标ID号:" + targetId); //region 不需要的字段 byte[] iTw = Arrays.copyOfRange(data, index + 4, index + 8); iTw = toLittleEndian(iTw); int Tw = ByteUtils.bytesToDecimal(iTw); // log.info("目标当前的像素宽度:" + Tw); byte[] iTh = Arrays.copyOfRange(data, index + 8, index + 12); iTh = toLittleEndian(iTh); int Th = ByteUtils.bytesToDecimal(iTh); //log.info("目标当前的像素高度:" + Th); byte[] fTx = Arrays.copyOfRange(data, index + 12, index + 16); fTx = toLittleEndian(fTx); float fTxAngle = ByteUtils.bytesToFloat(fTx); //log.debug("p角度:" + fTxAngle); byte[] fTy = Arrays.copyOfRange(data, index + 16, index + 20); fTy = toLittleEndian(fTy); float fTyAngle = ByteUtils.bytesToFloat(fTy); //log.debug("t角度:" + fTyAngle); byte[] sAreaNo = Arrays.copyOfRange(data, index + 20, index + 22); sAreaNo = toLittleEndian(sAreaNo); int AreaNo = ByteUtils.bytesToDecimal(sAreaNo); //log.debug("目标归属的告警区域号:" + AreaNo); byte[] cGrp = Arrays.copyOfRange(data, index + 22, index + 23); cGrp = toLittleEndian(cGrp); int Grp = ByteUtils.bytesToDecimal(cGrp); //log.info("所属组:" + Grp); //endregion String alarmType; //抽油机状态变量 String wellType; byte[] cStat = Arrays.copyOfRange(data, index + 23, index + 24); cStat = toLittleEndian(cStat); //String binaryString = String.format("%8s", Integer.toBinaryString(cStat[0] & 0xFF)).replace(' ', '0'); //log.info("目标当前状态:" + binaryString); // 提取第0位值 // 使用位运算操作判断第0位是否为1 boolean isB0 = (cStat[0] & 0x01) == 0x00; // 判断提取的值 if (isB0) { alarmType = "雷达抽油机停机"; byte[] szName = Arrays.copyOfRange(data, index + 32, index + 64); //log.info("所属告警区域名称:" + DatatypeConverter.printHexBinary(szName)); String alarmPointName = ByteUtils.bytesToStringZh(szName); // log.info("所属告警区域名称:" + alarmPointName); log.debug("报警信息:" + "【radarName】" + radarName + "【targetId】" + targetId + "【name】" + alarmPointName + "【alarmType】" + alarmType + "【alarmTime】" + alarmTime); ArdAlarmRadar ardAlarmRadar = new ArdAlarmRadar(); ardAlarmRadar.setTargetId(targetId); ardAlarmRadar.setName(alarmPointName); ardAlarmRadar.setAlarmType(alarmType); radarAlarmInfos.add(ardAlarmRadar); wellType = "停机"; } else { wellType = "运行"; } //抽油机状态集合中装入数据 byte[] szName = Arrays.copyOfRange(data, index + 32, index + 64); String alarmPointName = ByteUtils.bytesToStringZh(szName); //log.debug("状态信息:" + "【radarName】" + radarName + "【targetId】" + targetId + "【alarmTime】" + alarmTime + "【name】" + alarmPointName + "【alarmState】" + wellType); ArdAlarmRadar wellAlarm = new ArdAlarmRadar(); wellAlarm.setTargetId(targetId); wellAlarm.setName(alarmPointName); wellAlarm.setAlarmType(wellType); well.add(wellAlarm); } //endregion if (StringUtils.isEmpty(alarmTime)) { return; } if (targetNum == 0) { return; } RadarAlarmData radarAlarmData = new RadarAlarmData(); radarAlarmData.setRadarId(radarId); radarAlarmData.setRadarName(radarName); radarAlarmData.setAlarmTime(alarmTime); radarAlarmData.setArdAlarmRadars(radarAlarmInfos); MqttProducer.publish(2, false, "radar", JSON.toJSONString(radarAlarmData)); //抽油机状态MQTT队列 radarAlarmData.setArdAlarmRadars(well); MqttProducer.publish(2, false, "radarWellData", JSON.toJSONString(radarAlarmData)); } //强制引导 if (Arrays.equals(cmdId, new byte[]{0x02})) { //region 告警前端发送的强制引导信息 byte[] iDistance = Arrays.copyOfRange(data, 4, 8); iDistance = toLittleEndian(iDistance); long distance = ByteUtils.bytesToDecimal(iDistance); log.info("目标当前距离(m):" + distance); byte[] fTx = Arrays.copyOfRange(data, 8, 12); fTx = toLittleEndian(fTx); float tx = ByteUtils.bytesToFloat(fTx); log.debug("方位:" + tx); byte[] fTy = Arrays.copyOfRange(data, 12, 16); fTy = toLittleEndian(fTy); float ty = ByteUtils.bytesToFloat(fTy); if (ty < 0) { ty += 360; } log.debug("俯仰:" + ty); Map forceGuideMap = new HashMap<>(); forceGuideMap.put("distance", distance); forceGuideMap.put("p", tx); forceGuideMap.put("t", ty); forceGuideMap.put("radarId", radarId); log.debug("强制引导信息" + forceGuideMap); //endregion MqttProducer.publish(2, false, "radarForceGuide", JSON.toJSONString(forceGuideMap)); } } catch (Exception ex) { log.error("雷达报文解析异常:" + ex.getMessage()); } } }