package com.tsi.comm.server.protocol; import com.tsi.app.common.cpu.dto.TsiCvimAbnormal; import com.tsi.app.common.cpu.dto.TsiCvimControl; import com.tsi.app.common.cpu.dto.TsiCvimStatus; import com.tsi.app.common.cpu.enums.eLightsStatus; import com.tsi.app.common.cpu.enums.eLightsType; import com.tsi.app.common.cpu.enums.eOpCode; import com.tsi.app.common.cpu.enums.eTimeReliability; import com.tsi.app.common.utils.ByteUtils; import com.tsi.app.common.utils.CRC16Utils; import com.tsi.app.common.utils.HexString; import com.tsi.app.common.utils.TimeUtils; import com.tsi.app.common.xnet.NettyUtils; import com.tsi.comm.server.mongo.dto.TcsNodeStatus; import com.tsi.comm.server.repository.TsiNodeAddManager; import com.tsi.comm.server.vo.TsiNodeAddDetailVo; import com.tsi.comm.server.vo.TsiNodeAddVo; import com.tsi.comm.server.vo.TsiNodeVo; import io.netty.channel.Channel; import lombok.Getter; import lombok.Setter; import lombok.extern.slf4j.Slf4j; import java.nio.ByteBuffer; import java.nio.ByteOrder; import java.text.SimpleDateFormat; import java.util.*; @Slf4j @Getter @Setter public class TsiCpuPacket extends AbstractTsiPacket { // 헤더(10), 상태헤더(8), 상태정보(5)*상태정보건수, 체크섬(2) // CVIB Protocol Format(CPU-->CENTER) //TYPE STX1 STX2 LEN OPCODE DataVer NodeID DATA CHKSUM //Size Byte Byte 2Byte Byte 1Byte 4Byte NByte 2Byte //Value 0x7E 0x7E Size(LEN…CHKSUM) 0x13 0x01 … CRC16 //- LEN : LEN∼CHKSUM data length // - OPCODE: 0x13, 신호상태정보전송 // - NodeID: Node ID //- DATA : Opcode에 따른 데이터 영역 //- CRC16 : CCITT / ITU / CRC - 16, bits shift right, final little(x) big(o) endian encoding.from LEN to DATA /* // 10 byte uint8_t stx1; // stx 0x7E uint8_t stx2; // stx 0x7E uint8_t length[2]; // length, length byte ~ check sum uint8_t opcode; // op code, 0x13 uint8_t dataVer; // Data Version uint8_t nodeid[4]; // signal node id // 8 byte tsc_cvim_hdr_t hdr; // 8 byte, cvim signal status header // 5 byte * count tsc_cvim_stts_t stts[MAX_CVIM_STTS]; // 5 byte * 127 = 635 byte */ // cvim-raw packet header /* #define TIMESPEC_SIZE (sizeof(struct timespec)) // 16 bytes uint8_t pktTm[TIMESPEC_SIZE]; // little endian uint8_t ipaddr[4]; // big endian uint8_t port[2]; // big endian uint8_t connect; uint8_t nodeid[4]; // big endian */ public static final byte STX1 = 0x7E; public static final byte STX2 = 0x7E; public static final int SIZE_HEAD = 10; public static final int SIZE_STX1 = 1; public static final int SIZE_STX2 = 1; public static final int SIZE_LENGTH = 2; public static final int SIZE_OPCODE = 1; public static final int SIZE_VERSION = 1; public static final int SIZE_NODE_ID = 4; public static final int SIZE_STATUS_HDR = 8; public static final int SIZE_STATUS_DATA = 5; public static final int SIZE_CHECKSUM = 2; public static final int SIZE_PACKET_DATA = 18; // length, opcode, version, node_id, status_hdr, checksum public static final int INDEX_STX1 = 0; // 0 public static final int INDEX_STX2 = 1; // 1 public static final int INDEX_LENGTH = 2; // 2,3 public static final int INDEX_OPCODE = 4; // 4 public static final int INDEX_VERSION = 5; // 5 public static final int INDEX_NODE_ID = 6; // 6,7,8,9 public static final int INDEX_STATUS_HDR = 10; // 10,11,12,13,14,15,16,17 public static final int INDEX_STATUS_DATA = 18; // 18,19,20,21,22 public static final int INDEX_STATUS_DIR_ADD = 0; // 상태정보내 방향 추가 정보 인덱스 public static final int INDEX_STATUS_DIRECTION = 4; // 상태정보내 방향코드 public static final int SIZE_NODE_DUMMY = 1+1+2+1+1+2; // stx1, stx1, length(2), opcode, version, checksum(2) public static final int SIZE_NODE_HEAD = SIZE_NODE_ID + SIZE_STATUS_HDR; public static final int POS_NODE_HEAD_NODEID = 0; // node(4), public static final int POS_NODE_HEAD_COUNT = 7; // node(4), public static final int SIZE_IPC_SIZE = 27; public static final int SIZE_TIMESPEC = 16; public static final int POS_IPC_TIMESPEC = 0; public static final int POS_IPC_IPADDR = 16; public static final int POS_IPC_PORT = 20; public static final int POS_IPC_CONNECT = 22; public static final int POS_IPC_NODEID = 23; public static final int POS_IPC_PACKET = SIZE_IPC_SIZE; public static final byte CONNECT = 0x01; public static final byte DISCONNECT = 0x00; private Object obj; private int length; private byte dataVer; private int count; private int checkSum; protected byte[] cvimData; // for cvim-raw topic protected byte[] nodeData; // for node topic protected List addNodes; public TsiCpuPacket(long nodeId, long msec, long nsec, Channel channel) { super(nodeId, msec, nsec, NettyUtils.getRemoteIpAddressToLong(channel), NettyUtils.getRemotePort(channel)); setOpCode(eOpCode.TSI_CPU_SIGNAL_NOTIFY.getValue()); } public TsiCpuPacket(long nodeId, long msec, long nsec, long remoteIpAddressToLong, int remotePort) { super(nodeId, msec, nsec, remoteIpAddressToLong, remotePort); } // FOR CVIM packet public TsiCpuPacket(long nodeId, byte[] value) { super(nodeId, TimeUtils.currentTimeSeconds(), System.nanoTime(), 0, 0); this.buf = value; // CVIM Header 를 포함한 내부 IPC 형식 데이터임 int nodeLength = this.buf.length - TsiCpuPacket.SIZE_IPC_SIZE; this.cvimData = new byte[TsiCpuPacket.SIZE_IPC_SIZE]; System.arraycopy(this.buf, 0, this.cvimData, 0, TsiCpuPacket.SIZE_IPC_SIZE); if (nodeLength > 0) { this.nodeData = new byte[nodeLength]; System.arraycopy(this.buf, TsiCpuPacket.SIZE_IPC_SIZE, this.nodeData, 0, nodeLength); } } protected byte getStx1() { if (this.buf != null) return this.buf[INDEX_STX1]; return 0x00; } protected byte getStx2() { if (this.buf != null) return this.buf[INDEX_STX2]; return 0x00; } public byte[] getCvimData() { return this.cvimData; } public byte[] getTestData() { return this.buf; } public byte[] getNodeData() { return this.nodeData; } /* * Make cvim-raw packet */ protected void makeCvimPacket() { int length = this.buf == null ? 0 : this.buf.length; this.cvimData = new byte[SIZE_IPC_SIZE + length]; // cvim-raw header System.arraycopy(this.timespec.bytes(), 0, this.cvimData, POS_IPC_TIMESPEC, SIZE_TIMESPEC); ByteUtils.setUnsignedInt(this.cvimData, POS_IPC_IPADDR, getRemoteIp()); ByteUtils.setUnsignedShort(this.cvimData, POS_IPC_PORT, getRemotePort()); this.cvimData[POS_IPC_CONNECT] = opCode == (byte)eOpCode.TSI_CPU_SIGNAL_NOTIFY.getValue() ? CONNECT : DISCONNECT; ByteUtils.setUnsignedInt(this.cvimData, POS_IPC_NODEID, this.nodeId); // cvim-raw body if (length > 0) { System.arraycopy(this.buf, 0, this.cvimData, POS_IPC_PACKET, length); } //log.error("CVIM: {}", HexString.fromBytes(this.cvimData)); } /* * Make cvim-raw packet */ protected void makeAddNodeCvimPaket(TsiCpuPacket cpuPacket, byte[] packet) { int length = packet == null ? 0 : packet.length; final int headSize = SIZE_IPC_SIZE + 6; cpuPacket.cvimData = new byte[headSize + length + 2]; // cpu packet 6 byte(nodeid 제외), crc 2 byte // cvim-raw header System.arraycopy(timespec.bytes(), 0, cpuPacket.cvimData, POS_IPC_TIMESPEC, SIZE_TIMESPEC); ByteUtils.setUnsignedInt(cpuPacket.cvimData, POS_IPC_IPADDR, getRemoteIp()); ByteUtils.setUnsignedShort(cpuPacket.cvimData, POS_IPC_PORT, getRemotePort()); cpuPacket.cvimData[POS_IPC_CONNECT] = opCode == (byte)eOpCode.TSI_CPU_SIGNAL_NOTIFY.getValue() ? CONNECT : DISCONNECT; ByteUtils.setUnsignedInt(cpuPacket.cvimData, POS_IPC_NODEID, cpuPacket.nodeId); cpuPacket.cvimData[SIZE_IPC_SIZE+INDEX_STX1] = getStx1(); cpuPacket.cvimData[SIZE_IPC_SIZE+INDEX_STX2] = getStx2(); ByteUtils.setUnsignedShort(cpuPacket.cvimData, SIZE_IPC_SIZE+INDEX_LENGTH, length+2); cpuPacket.cvimData[SIZE_IPC_SIZE+INDEX_OPCODE] = (byte)getOpCode(); cpuPacket.cvimData[SIZE_IPC_SIZE+INDEX_VERSION] = getDataVer(); // cvim-raw body if (length > 0) { System.arraycopy(packet, 0, cpuPacket.cvimData, headSize, length); } // 체크섬 계산하지 않음 } protected boolean checkPacket() { // 0 단계. STX1, STX2 체크 if (this.buf[INDEX_STX1] != STX1 || this.buf[INDEX_STX2] != STX2) { log.info("Node: {}, STX Error: {}, {}", nodeId, this.buf[INDEX_STX1], this.buf[INDEX_STX2]); return false; } // 1 단계. 패킷 길이 체크 if (this.length != ( SIZE_PACKET_DATA + (SIZE_STATUS_DATA * this.count) ) ) { log.info("Node: {}, Length Error: {}, status count: {}, {}", nodeId, this.length, this.count, SIZE_PACKET_DATA + (SIZE_STATUS_DATA * this.count)); return false; } // 2단계. 체크섬 this.checkSum = ByteUtils.getUnsignedShort(this.buf, this.buf.length-2); int calcCheckSum = CRC16Utils.CRC16_ccitt_cvim(this.buf, INDEX_LENGTH, this.length-2); // 시작인덱스가 있으므로 전체길이로 계산 if (this.checkSum != calcCheckSum) { log.error("Node: {}, Check Sum Error: recv: {}, calc: {}", nodeId, this.checkSum, calcCheckSum); log.error("{}", HexString.fromBytes(this.buf)); return false; } return true; } public boolean parsing(TsiNodeVo obj) { this.opCode = this.buf[INDEX_OPCODE]; this.dataVer = this.buf[INDEX_VERSION]; this.length = ByteUtils.getUnsignedShort(this.buf, INDEX_LENGTH); this.count = (int)(this.buf[INDEX_STATUS_HDR+3] & 0x7F); makeCvimPacket(); if (!checkPacket()) { return false; } TsiNodeAddVo tsiNodeAddVo = TsiNodeAddManager.getInstance().get(this.nodeId); if (tsiNodeAddVo == null) { // 연등지 정보가 없는 경우 int length = this.buf.length-SIZE_NODE_DUMMY; this.nodeData = new byte[length]; System.arraycopy(this.buf, INDEX_NODE_ID, this.nodeData, 0, length); return true; } // 연등지 교차로 데이터 파싱 byte[] head = new byte[SIZE_NODE_HEAD]; List nodeStatus = new ArrayList<>(); Map> addStatus = new HashMap<>(); System.arraycopy(this.buf, INDEX_NODE_ID, head, 0, SIZE_NODE_HEAD); int loop = 0; for (int ii = INDEX_STATUS_DATA; loop < this.count; ii += SIZE_STATUS_DATA, loop++) { byte[] status = new byte[SIZE_STATUS_DATA]; System.arraycopy(this.buf, ii, status, 0, SIZE_STATUS_DATA); int dirAdd = (int)(status[INDEX_STATUS_DIR_ADD] & 0x0F); if (dirAdd == 0) { // 원천 노드 정보 nodeStatus.add(status); } else { // 추가 노드 정보 int directionCode = (int)status[INDEX_STATUS_DIRECTION]; status[INDEX_STATUS_DIR_ADD] = (byte) (status[INDEX_STATUS_DIR_ADD] & 0xF0); // 방향추가정보를 0 으로 초기화 TsiNodeAddDetailVo detailVo = tsiNodeAddVo.getAddNodeMap().get(directionCode * 1000 + dirAdd); if (detailVo != null) { //detailVo.getNodeObj(); for (int dirIdx = 0; dirIdx < 2; dirIdx++) { if (detailVo.getAddDirCode()[dirIdx] != 0x00) { List list = addStatus.get(detailVo.getNodeId()); if (list == null) { list = new ArrayList<>(); addStatus.put(detailVo.getNodeId(), list); } if (dirIdx == 0) { status[INDEX_STATUS_DIRECTION] = detailVo.getAddDirCode()[dirIdx]; list.add(status); } else { byte[] status2 = new byte[SIZE_STATUS_DATA]; System.arraycopy(status, 0, status2, 0, SIZE_STATUS_DATA); status2[INDEX_STATUS_DIRECTION] = detailVo.getAddDirCode()[dirIdx]; list.add(status2); } } } } } } byte splitFlag = 0x01; int statusCount = nodeStatus.size(); this.nodeData = new byte[SIZE_NODE_HEAD + (statusCount*SIZE_STATUS_DATA)]; System.arraycopy(head, 0, this.nodeData, 0, SIZE_NODE_HEAD); this.nodeData[POS_NODE_HEAD_COUNT] = (byte) (statusCount | (splitFlag << 7)); for (int ii = 0; ii < statusCount; ii++) { System.arraycopy(nodeStatus.get(ii), 0, this.nodeData, SIZE_NODE_HEAD+(ii*SIZE_STATUS_DATA), SIZE_STATUS_DATA); } makeAddNodeCvimPaket(this, this.nodeData); if (addStatus.size() == 0) { // 연등지 정보가 설정된게 없을까??? return true; } // 연등지 노드 카프카 패킷 생성 this.addNodes = new ArrayList<>(); for (Map.Entry> entry: addStatus.entrySet()) { statusCount = entry.getValue().size(); TsiCpuAddPacket addPacket = new TsiCpuAddPacket(entry.getKey(), this.timespec, this.remoteIp, this.remotePort); addPacket.setObj(TsiNodeAddManager.getInstance().get(entry.getKey())); addPacket.setNodeData(new byte[SIZE_NODE_HEAD + (statusCount*SIZE_STATUS_DATA)]); System.arraycopy(head, 0, addPacket.getNodeData(), 0, SIZE_NODE_HEAD); ByteUtils.setUnsignedInt(addPacket.getNodeData(), POS_NODE_HEAD_NODEID, entry.getKey()); addPacket.getNodeData()[POS_NODE_HEAD_COUNT] = (byte) (statusCount | (splitFlag << 7)); for (int ii = 0; ii < statusCount; ii++) { System.arraycopy(entry.getValue().get(ii), 0, addPacket.getNodeData(), SIZE_NODE_HEAD+(ii*SIZE_STATUS_DATA), SIZE_STATUS_DATA); } makeAddNodeCvimPaket(addPacket, addPacket.getNodeData()); addNodes.add(addPacket); } /* log.error(" RAW: {}, {}", this.nodeId, HexString.fromBytes(getBuf())); log.error("NODE: {}, {}", this.nodeId, HexString.fromBytes(getNodeData())); log.error("CVIM: {}, {}", this.nodeId, HexString.fromBytes(getCvimData())); for (int ii = 0; ii < addNodes.size(); ii++) { log.error("NODE: {}, {}", addNodes.get(ii).nodeId, HexString.fromBytes(addNodes.get(ii).getNodeData())); log.error("CVIM: {}, {}", addNodes.get(ii).nodeId, HexString.fromBytes(addNodes.get(ii).getCvimData())); }*/ return true; } public TcsNodeStatus getNodeStatus() { int nodeLength = this.buf.length - TsiCpuPacket.SIZE_IPC_SIZE; byte[] msec = new byte[8]; byte[] nsec = new byte[8]; System.arraycopy(this.cvimData, 0, msec, 0, 8); System.arraycopy(this.cvimData, 8, nsec, 0, 8); ByteBuffer byteMSec = ByteBuffer.wrap(msec); ByteBuffer byteNSec = ByteBuffer.wrap(nsec); byteMSec.order(ByteOrder.LITTLE_ENDIAN); byteNSec.order(ByteOrder.LITTLE_ENDIAN); long milliSeconds = byteMSec.getLong(); long nanoSeconds = byteNSec.getLong(); this.remoteIp = ByteUtils.getUnsignedInt(this.cvimData, TsiCpuPacket.POS_IPC_IPADDR); this.remotePort = ByteUtils.getUnsignedShort(this.cvimData, TsiCpuPacket.POS_IPC_PORT); ///// getTimespec().tv_nsec(nanoSeconds); // recv tcp, Date date = new Date(milliSeconds * 1000L); SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); sdf.setTimeZone(java.util.TimeZone.getTimeZone("GMT+9")); String collectTime = sdf.format(date); if (this.cvimData[TsiCpuPacket.POS_IPC_CONNECT] == TsiCpuPacket.DISCONNECT) { return TcsNodeStatus.builder() ._id(this.nodeId) .nodeId(this.nodeId) .isConnect(false) .collectTime(collectTime) .build(); } if (nodeLength < TsiCpuPacket.SIZE_HEAD + TsiCpuPacket.SIZE_STATUS_HDR) { // packet error return null; } String tscDateTime; int cycleElapsedTime; TsiCvimControl tscControlInfo = new TsiCvimControl(); TsiCvimAbnormal tscAbnormalInfo = new TsiCvimAbnormal(); int signalStatusInfoCount; int divFlag; long localTime; List signalStatusInfos = new ArrayList<>(); //this.opCode = this.nodeData[TsiCpuPacket.INDEX_OPCODE]; //byte dataVer = this.nodeData[TsiCpuPacket.INDEX_VERSION]; //int length = ByteUtils.getUnsignedShort(this.nodeData, TsiCpuPacket.INDEX_LENGTH); //int count = (int)(this.nodeData[TsiCpuPacket.INDEX_STATUS_HDR+3] & 0x7F); byte control = this.cvimData[TsiCpuPacket.SIZE_IPC_SIZE+TsiCpuPacket.INDEX_STATUS_HDR+0]; tscControlInfo.inManualControl = ((control ) & 0x01) == 0x01; //수동; tscControlInfo.inFlashingControl = ((control >> 1) & 0x01) == 0x01; //점멸; tscControlInfo.inLightsOutControl = ((control >> 2) & 0x01) == 0x01; //소등 tscControlInfo.inActuationControl = ((control >> 3) & 0x01) == 0x01; //감응; tscControlInfo.inTransitionControl = ((control >> 4) & 0x01) == 0x01; //전이; byte abnormal = this.cvimData[TsiCpuPacket.SIZE_IPC_SIZE+TsiCpuPacket.INDEX_STATUS_HDR+1]; tscAbnormalInfo.inSignalConflict = ((abnormal ) & 0x01) == 0x01; //모순상태; tscAbnormalInfo.inCenterComm = ((abnormal >> 1) & 0x01) == 0x01; //센터상태; tscAbnormalInfo.inScuComm = ((abnormal >> 2) & 0x01) == 0x01; //SCU 상태 cycleElapsedTime = (int)(this.cvimData[TsiCpuPacket.SIZE_IPC_SIZE+TsiCpuPacket.INDEX_STATUS_HDR+2] & 0xFF); byte stts = this.cvimData[TsiCpuPacket.SIZE_IPC_SIZE+TsiCpuPacket.INDEX_STATUS_HDR+3];; signalStatusInfoCount = (int)(stts & 0x7F); //divFlag = (int)((stts >> 7) & 0x01); localTime = ByteUtils.getUnsignedInt(this.cvimData, TsiCpuPacket.SIZE_IPC_SIZE+TsiCpuPacket.INDEX_STATUS_HDR+4); int ii = TsiCpuPacket.SIZE_IPC_SIZE+TsiCpuPacket.INDEX_STATUS_DATA; for (int idx = 0; idx < signalStatusInfoCount; idx++) { TsiCvimStatus status = new TsiCvimStatus(); //int dirAdd = (buffer.get(ii)) & 0x0F; //연등지 int lightsType = (int)((this.cvimData[ii ] >> 4) & 0x0F); int lightsStatus = (int)((this.cvimData[ii+1] ) & 0x07); int reliability = (int)((this.cvimData[ii+1] >> 7) & 0x01); boolean readyPedestrian = ((this.cvimData[ii+1] >> 6) & 0x01) == 0x01; boolean unProtected = ((this.cvimData[ii+1] >> 3) & 0x01) == 0x01; int totalSeconds = (int)(this.cvimData[ii+2] & 0xFF); int remainSeconds = (int)(this.cvimData[ii+3] & 0xFF); int directionCode = (int)(this.cvimData[ii+4] & 0xFF); status.setLightsType(eLightsType.getByValue(lightsType)); // 신호등정보 [직진,좌,보행] status.setLightsStatus(eLightsStatus.getByValue(lightsStatus)); //신호등상태 status.setTimeReliability(eTimeReliability.getByValue(reliability)); //시간정보신뢰성 status.setReadyPedestrianSignal(readyPedestrian); //보행자 status.setUnProtectedSignal(unProtected); //비보호 상태 status.setTotalSeconds(totalSeconds); //표출시간 status.setRemainingSeconds(remainSeconds); //잔여시간 status.setDirectionCode(directionCode); //방향코드 signalStatusInfos.add(status); ii += TsiCpuPacket.SIZE_STATUS_DATA; } date = new Date(localTime * 1000L); sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); sdf.setTimeZone(java.util.TimeZone.getTimeZone("GMT+9")); tscDateTime = sdf.format(date); //log.error("Packet Received: NodeId: {}, {}", this.nodeId, TimeUtils.elapsedTimeStr(System.nanoTime()-nanoSeconds)); return TcsNodeStatus.builder() ._id(this.nodeId) .nodeId(this.nodeId) .isConnect(true) .collectTime(collectTime) .tscDateTime(tscDateTime) .cycleElapsedTime(cycleElapsedTime) .tscControlInfo(tscControlInfo) .tscAbnormalInfo(tscAbnormalInfo) .signalStatusInfoCount(signalStatusInfoCount) .signalStatusInfos(signalStatusInfos) .build(); } }