From c696dc477d33f2d650537ea72df0878c4e1251d5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=B0=8F=E9=B1=BC=E5=B9=B2?= <1810377322@163.com> Date: Mon, 7 Jul 2025 18:13:37 +0800 Subject: [PATCH] =?UTF-8?q?=E7=87=83=E6=B0=94=E6=8A=A5=E8=AD=A6=E5=99=A8?= =?UTF-8?q?=E8=A7=A3=E6=9E=90\=E5=8A=A8=E7=81=AB=E7=A6=BB=E4=BA=BA?= =?UTF-8?q?=E8=A7=A3=E6=9E=90?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- emqx-plugin/pom.xml | 2 +- .../plugins/emqx/conf/CRC16ModbusUtil.java | 35 ++- .../plugins/emqx/conf/EventTypeMapper.java | 39 +-- .../emqx/conf/FireSafetyDataParser.java | 110 ++++++++ .../emqx/conf/FireSafetyDataReceiver.java | 187 ++++++++++++++ .../plugins/emqx/conf/GasAlarmDataParser.java | 163 +++++++++--- .../plugins/emqx/conf/GasDetectorParser.java | 44 ++++ .../plugins/emqx/conf/IoTConfigProtocol.java | 111 ++++++-- .../plugins/emqx/conf/MqttDataProcessor.java | 105 ++++++++ .../plugins/emqx/conf/MqttHexProcessor.java | 71 ++++++ .../plugins/emqx/conf/ProtocolTest.java | 180 ++++++++++++- .../plugins/emqx/service/EmqxPlugin.java | 228 ++++++++++++++--- .../plugins/emqx/service/MqttDevice.java | 240 ++++++++++++++---- pom.xml | 2 +- 14 files changed, 1337 insertions(+), 180 deletions(-) create mode 100644 emqx-plugin/src/main/java/cc/iotkit/plugins/emqx/conf/FireSafetyDataParser.java create mode 100644 emqx-plugin/src/main/java/cc/iotkit/plugins/emqx/conf/FireSafetyDataReceiver.java create mode 100644 emqx-plugin/src/main/java/cc/iotkit/plugins/emqx/conf/GasDetectorParser.java create mode 100644 emqx-plugin/src/main/java/cc/iotkit/plugins/emqx/conf/MqttDataProcessor.java create mode 100644 emqx-plugin/src/main/java/cc/iotkit/plugins/emqx/conf/MqttHexProcessor.java diff --git a/emqx-plugin/pom.xml b/emqx-plugin/pom.xml index 5f5b31a..54ad1bc 100644 --- a/emqx-plugin/pom.xml +++ b/emqx-plugin/pom.xml @@ -5,7 +5,7 @@ iot-iita-plugins cc.iotkit.plugins - 2.10.19 + 2.10.18 4.0.0 diff --git a/emqx-plugin/src/main/java/cc/iotkit/plugins/emqx/conf/CRC16ModbusUtil.java b/emqx-plugin/src/main/java/cc/iotkit/plugins/emqx/conf/CRC16ModbusUtil.java index cbb875c..46691e3 100644 --- a/emqx-plugin/src/main/java/cc/iotkit/plugins/emqx/conf/CRC16ModbusUtil.java +++ b/emqx-plugin/src/main/java/cc/iotkit/plugins/emqx/conf/CRC16ModbusUtil.java @@ -6,10 +6,21 @@ public class CRC16ModbusUtil { public static String calculate(String hexStr) { byte[] data = hexToByteArray(hexStr); - int crc = calculateCrc(data); + int crc = calculateCrcWithSwap(data); return String.format("%04X", crc); } - + private static int calculateCrcWithSwap(byte[] data) { + int crc = 0xFFFF; // INIT_VALUE应为0xFFFF + for (byte b : data) { + crc ^= (b & 0xFF); + for (int i = 0; i < 8; i++) { + boolean lsb = (crc & 1) == 1; + crc >>>= 1; + if (lsb) crc ^= 0xA001; // POLY应为0xA001 + } + } + return ((crc & 0xFF) << 8) | ((crc >> 8) & 0xFF); // 关键交换逻辑 + } private static int calculateCrc(byte[] data) { int crc = INIT_VALUE; for (byte b : data) { @@ -53,12 +64,24 @@ public class CRC16ModbusUtil { String mid, String cmd, String dataHex ) { String headerLength = "AA" + type + version + mid + cmd + dataHex +"crc1"+ "55"; - Integer length =headerLength.length()/2; - String header = "AA" + type + version + length + mid + cmd; - String crcData = length + mid + cmd + dataHex; + //额外加上长度的两个字节 + Integer length =headerLength.length()/2 + 2; + String lengthHex = String.format("%04X", length); + String header = "AA" + type + version + lengthHex + mid + cmd; + // 转换为2字节16进制（大端序） + System.out.println("dataHex+++++++++++++++++++++++++++++++" + dataHex); + System.out.println("header+++++++++++++++++++++++++++++++" + header); + String crcData = lengthHex + mid + cmd + dataHex; + System.out.println("crcData+++++++++++++++++++++++++++++++" + crcData); String crc = calculate(crcData); return header + dataHex + crc + "55"; } - + public static byte[] toBigEndianBytes(int length) { + if(length > 65535) throw new IllegalArgumentException("长度超过2字节范围"); + return new byte[] { + (byte)((length >> 8) & 0xFF), // 高字节 + (byte)(length & 0xFF) // 低字节 + }; + } } diff --git a/emqx-plugin/src/main/java/cc/iotkit/plugins/emqx/conf/EventTypeMapper.java b/emqx-plugin/src/main/java/cc/iotkit/plugins/emqx/conf/EventTypeMapper.java index 6bb2980..397c573 100644 --- a/emqx-plugin/src/main/java/cc/iotkit/plugins/emqx/conf/EventTypeMapper.java +++ b/emqx-plugin/src/main/java/cc/iotkit/plugins/emqx/conf/EventTypeMapper.java @@ -18,28 +18,31 @@ public class EventTypeMapper { EVENT_TYPE_MAP.put(8, "节点通讯断网故障"); EVENT_TYPE_MAP.put(9, "探头传感器故障恢复"); EVENT_TYPE_MAP.put(10, "节点通讯断网故障恢复"); - EVENT_TYPE_MAP.put(11, "自检/模块联动"); - EVENT_TYPE_MAP.put(12, "阀门动作"); + EVENT_TYPE_MAP.put(11, "节点自检"); + EVENT_TYPE_MAP.put(12, "节点自检完成"); + EVENT_TYPE_MAP.put(13, "模块联动"); + EVENT_TYPE_MAP.put(14, "阀门动作"); + EVENT_TYPE_MAP.put(15, "浓度变化上报"); // 控制器相关事件 - EVENT_TYPE_MAP.put(13, "自检"); - EVENT_TYPE_MAP.put(14, "备电故障"); - EVENT_TYPE_MAP.put(15, "主电故障"); - EVENT_TYPE_MAP.put(16, "主电欠压"); - EVENT_TYPE_MAP.put(17, "控制器复位"); - EVENT_TYPE_MAP.put(18, "控制器开机"); - EVENT_TYPE_MAP.put(19, "主电故障恢复"); - EVENT_TYPE_MAP.put(20, "备电故障恢复"); - EVENT_TYPE_MAP.put(21, "主电欠压恢复"); + EVENT_TYPE_MAP.put(30, "自检"); + EVENT_TYPE_MAP.put(31, "备电故障"); + EVENT_TYPE_MAP.put(32, "主电故障"); + EVENT_TYPE_MAP.put(33, "主电欠压"); + EVENT_TYPE_MAP.put(34, "控制器复位"); + EVENT_TYPE_MAP.put(35, "控制器开机"); + EVENT_TYPE_MAP.put(36, "主电故障恢复"); + EVENT_TYPE_MAP.put(37, "备电故障恢复"); + EVENT_TYPE_MAP.put(38, "主电欠压恢复"); // 动火离人相关事件 - EVENT_TYPE_MAP.put(22, "设备开机"); - EVENT_TYPE_MAP.put(23, "动火离人报警"); - EVENT_TYPE_MAP.put(24, "动火离人报警恢复"); - EVENT_TYPE_MAP.put(25, "设备故障"); - EVENT_TYPE_MAP.put(26, "设备故障恢复"); - EVENT_TYPE_MAP.put(27, "断网故障"); - EVENT_TYPE_MAP.put(28, "断网故障恢复"); + EVENT_TYPE_MAP.put(50, "设备开机"); + EVENT_TYPE_MAP.put(51, "动火离人报警"); + EVENT_TYPE_MAP.put(52, "动火离人报警恢复"); + EVENT_TYPE_MAP.put(53, "设备故障"); + EVENT_TYPE_MAP.put(54, "设备故障恢复"); + EVENT_TYPE_MAP.put(55, "断网故障"); + EVENT_TYPE_MAP.put(56, "断网故障恢复"); // 正常上报事件 EVENT_TYPE_MAP.put(128, "燃气报警器正常数据上报"); diff --git a/emqx-plugin/src/main/java/cc/iotkit/plugins/emqx/conf/FireSafetyDataParser.java b/emqx-plugin/src/main/java/cc/iotkit/plugins/emqx/conf/FireSafetyDataParser.java new file mode 100644 index 0000000..245a737 --- /dev/null +++ b/emqx-plugin/src/main/java/cc/iotkit/plugins/emqx/conf/FireSafetyDataParser.java @@ -0,0 +1,110 @@ +package cc.iotkit.plugins.emqx.conf; + +import java.nio.ByteBuffer; +import java.util.HashMap; +import java.util.Map; + +public class FireSafetyDataParser { + // 事件类型常量 + public static final int EVENT_POWER_ON = 50; + public static final int EVENT_FIRE_ALARM = 51; + public static final int EVENT_ALARM_RESET = 52; + public static final int EVENT_DEVICE_FAIL = 53; + public static final int EVENT_FAIL_RESET = 54; + public static final int EVENT_NET_DOWN = 55; + public static final int EVENT_NET_RECOVER = 56; + + // 数据项ID常量 + public static final int ITEM_TEMP_DISTANCE = 0x0040; + public static final int ITEM_ALARM_DURATION = 0x0041; + public static final int ITEM_SOUP_ALARM = 0x0042; + public static final int ITEM_VOLUME = 0x0043; + public static final int ITEM_SOUP_MODE = 0x0044; + public static final int ITEM_ALARM_STATUS = 0x0045; + public static final int ITEM_FLAME_DETECT = 0x0046; + public static final int ITEM_HUMAN_DETECT = 0x0047; + public static final int ITEM_ERROR_CODE = 0x0048; + + public static Map parseData(byte[] rawData) { + ByteBuffer buffer = ByteBuffer.wrap(rawData); + Map result = new HashMap<>(); + result.put("dataType", "HOT_WORK_DATA"); + // 解析事件头（1字节事件类型 + 6字节BCD时间） + // int eventType = buffer.get() & 0xFF; + int eventType = buffer.getShort() & 0xFFFF; + // buffer.getShort() & 0xFFFF + result.put("eventType", eventType); + result.put("timestamp", GasAlarmDataParser.parseBcdTime(buffer)); + result.put("eventTypeValue", EventTypeMapper.getEventTypeName(Integer.parseInt(result.get("eventType").toString()))); + // 解析数据域 + if (buffer.remaining() > 0) { + int dataLength = buffer.get() & 0xFF; + while (buffer.remaining() >= 4) { // 每个数据项至少4字节（2字节ID+2字节数据） + int itemId = buffer.getShort() & 0xFFFF; + int itemValue = buffer.getShort() & 0xFFFF; + result.put(getItemName(itemId), parseItemValue(itemId, itemValue)); + } + } + return result; + } + + /* private static String parseBcdTime(ByteBuffer buffer) { + byte[] bcdTime = new byte[6]; + buffer.get(bcdTime); + return String.format("20%02d-%02d-%02d %02d:%02d:%02d", + bcdTime[0], bcdTime[1], bcdTime[2], bcdTime[3], bcdTime[4], bcdTime[5]); + }*/ + + private static String getItemName(int itemId) { + switch (itemId) { + case ITEM_TEMP_DISTANCE: return "temperatureDistance"; + case ITEM_ALARM_DURATION: return "alarmDuration"; + case ITEM_SOUP_ALARM: return "soupAlarmTime"; + case ITEM_VOLUME: return "volumeLevel"; + case ITEM_SOUP_MODE: return "soupMode"; + case ITEM_ALARM_STATUS: return "alarmStatus"; + case ITEM_FLAME_DETECT: return "flameDetected"; + case ITEM_HUMAN_DETECT: return "humanDetected"; + case ITEM_ERROR_CODE: return "errorCode"; + default: return "unknown_" + Integer.toHexString(itemId); + } + } + + private static Object parseItemValue(int itemId, int value) { + switch (itemId) { + + // 测温距离(0x0040) + case ITEM_TEMP_DISTANCE: + return new String[]{"2米","2-3米","3-3.5米"}[value-1]; + // 动火报警时间(0x0041) + case ITEM_ALARM_DURATION: + return value + "分钟"; + + // 煲汤报警时间(0x0042) + case ITEM_SOUP_ALARM: + return (value > 60) ? + (value/60 + "小时" + value%60 + "分钟") : + (value + "分钟"); + // 音量控制(0x0043) + case ITEM_VOLUME: + return new String[]{"静音","小","中","大"}[value]; + // 煲汤模式(0x0044) + case ITEM_SOUP_MODE: + return value == 0 ? "普通模式" : "煲汤模式"; + // 报警状态(0x0045) + case ITEM_ALARM_STATUS: + return value == 0 ? "未报警" : "报警中"; + // 火苗检测(0x0046) + case ITEM_FLAME_DETECT: + return value == 0 ? "未检测到火苗" : "检测到火苗"; + // 人体检测(0x0047) + case ITEM_HUMAN_DETECT: + return value == 0 ? "无人员" : "有人员"; + // 故障码(预留) + case ITEM_ERROR_CODE: + return String.format("故障码0x%04X", value); + default: + return value; + } + } +} diff --git a/emqx-plugin/src/main/java/cc/iotkit/plugins/emqx/conf/FireSafetyDataReceiver.java b/emqx-plugin/src/main/java/cc/iotkit/plugins/emqx/conf/FireSafetyDataReceiver.java new file mode 100644 index 0000000..2376a2b --- /dev/null +++ b/emqx-plugin/src/main/java/cc/iotkit/plugins/emqx/conf/FireSafetyDataReceiver.java @@ -0,0 +1,187 @@ +package cc.iotkit.plugins.emqx.conf; + +import java.io.InputStream; +import java.nio.ByteBuffer; +import java.util.HashMap; +import java.util.Map; + + +public class FireSafetyDataReceiver { + public static void main(String[] args) { + // 测试16进制报文：17(事件类型) + 250616123456(时间) + 02(数据长度) + 0040(测温距离ID) + 0002(2-3米) + String hexData = "17 25 06 16 12 34 56 02 00 40 00 02"; + Map result = handleHexData(hexData); + + System.out.println("解析结果：" + result); + result.forEach((k, v) -> System.out.println(k + ": " + v)); + } + /*public void processDataStream(InputStream inputStream) throws Exception { + byte[] header = new byte[HEADER_SIZE]; + while (inputStream.read(header) == HEADER_SIZE) { + int dataLength = inputStream.read(); + byte[] payload = new byte[dataLength]; + inputStream.read(payload); + + byte[] fullData = new byte[HEADER_SIZE + 1 + dataLength]; + System.arraycopy(header, 0, fullData, 0, HEADER_SIZE); + fullData[HEADER_SIZE] = (byte)dataLength; + System.arraycopy(payload, 0, fullData, HEADER_SIZE+1, dataLength); + + Map parsedData = FireSafetyDataParser.parseData(fullData); + handleEvent(parsedData); + } + }*/ + public static Map handleHexData(String hexString) { + byte[] data = hexStringToBytes(hexString); + if (data == null || data.length < 1) { + return null; + } + + int eventType = Byte.toUnsignedInt(data[0]) + Byte.toUnsignedInt(data[1]); + if (eventType == 129) { + return handleEvent(data); + } + return OpenFireHandleEvent(data); + } + + private static byte[] hexStringToBytes(String hex) { + hex = hex.replaceAll("\\s", ""); + byte[] bytes = new byte[hex.length() / 2]; + for (int i = 0; i < bytes.length; i++) { + bytes[i] = (byte) Integer.parseInt(hex.substring(i * 2, i * 2 + 2), 16); + } + return bytes; + } + /* + * 动火离人告警事件 + * */ + public static Map OpenFireHandleEvent(byte[] data) { + Map result= new HashMap<>(); + ByteBuffer buffer = ByteBuffer.wrap(data); + result.put("dataType", "HOT_WORK_DATA"); + int eventType = buffer.getShort() & 0xFFFF; + // 解析事件头 + //int eventType = buffer.get() & 0xFF; + result.put("eventType", eventType); + result.put("timestamp", GasAlarmDataParser.parseBcdTime(buffer)); + result.put("eventTypeValue", EventTypeMapper.getEventTypeName(Integer.parseInt(result.get("eventType").toString()))); + // 完整事件处理 + /* switch (eventType) { + case FireSafetyDataParser.EVENT_POWER_ON: + result.put("eventName", "设备开机"); + // result.put("action", "记录启动日志"); + break; + case FireSafetyDataParser.EVENT_FIRE_ALARM: + result.put("eventName", "动火离人报警"); + //result.put("action", "触发声光报警+推送告警"); + // parseDataItems(buffer, result); + break; + case FireSafetyDataParser.EVENT_ALARM_RESET: + result.put("eventName", "动火离人报警恢复"); + // result.put("action", "停止报警+记录处置"); + break; + case FireSafetyDataParser.EVENT_DEVICE_FAIL: + result.put("eventName", "设备故障"); + // result.put("action", "启动自检+通知维护"); + // parseDataItems(buffer, result); + break; + case FireSafetyDataParser.EVENT_FAIL_RESET: + result.put("eventName", "设备故障恢复"); + // result.put("action", "更新设备状态"); + break; + case FireSafetyDataParser.EVENT_NET_DOWN: + result.put("eventName", "断网故障"); + // result.put("action", "启用本地存储"); + break; + case FireSafetyDataParser.EVENT_NET_RECOVER: + result.put("eventName", "断网故障恢复"); + // result.put("action", "同步缓存数据"); + break; + default: + result.put("eventName", "未知事件"); + // result.put("action", "需要人工核查"); + }*/ + System.out.println("解析后的数据" + result); + return result; + } + /* + * 动火离人正常上报数据解析 + * */ + public static Map handleEvent(byte[] data) { + // Map result = new HashMap<>(); + Map result= FireSafetyDataParser.parseData(data); + /* ByteBuffer buffer = ByteBuffer.wrap(data); + + // 解析事件头 + int eventType = buffer.get() & 0xFF; + result.put("eventType", eventType); + result.put("timestamp", GasAlarmDataParser.parseBcdTime(buffer)); + + // 完整事件处理 + switch (eventType) { + case FireSafetyDataParser.EVENT_POWER_ON: + result.put("eventName", "设备开机"); + // result.put("action", "记录启动日志"); + break; + case FireSafetyDataParser.EVENT_FIRE_ALARM: + result.put("eventName", "动火离人报警"); + //result.put("action", "触发声光报警+推送告警"); + // parseDataItems(buffer, result); + break; + case FireSafetyDataParser.EVENT_ALARM_RESET: + result.put("eventName", "动火离人报警恢复"); + // result.put("action", "停止报警+记录处置"); + break; + case FireSafetyDataParser.EVENT_DEVICE_FAIL: + result.put("eventName", "设备故障"); + // result.put("action", "启动自检+通知维护"); + // parseDataItems(buffer, result); + break; + case FireSafetyDataParser.EVENT_FAIL_RESET: + result.put("eventName", "设备故障恢复"); + // result.put("action", "更新设备状态"); + break; + case FireSafetyDataParser.EVENT_NET_DOWN: + result.put("eventName", "断网故障"); + // result.put("action", "启用本地存储"); + break; + case FireSafetyDataParser.EVENT_NET_RECOVER: + result.put("eventName", "断网故障恢复"); + // result.put("action", "同步缓存数据"); + break; + default: + result.put("eventName", "未知事件"); + // result.put("action", "需要人工核查"); + }*/ + System.out.println("解析后的数据" + result); + return result; + } + /* private static String parseBcdTime(byte[] data) { + return String.format("20%02d-%02d-%02d %02d:%02d:%02d", + bcdToInt(data[0]), bcdToInt(data[1]), bcdToInt(data[2]), + bcdToInt(data[3]), bcdToInt(data[4]), bcdToInt(data[5])); + } + private static int bcdToInt(byte b) { + return ((b >> 4) & 0x0F)*10 + (b & 0x0F); + }*/ +/* + private static String parseBcdTime(ByteBuffer buffer) { + byte[] timeBytes = new byte[6]; + buffer.get(timeBytes); + return String.format("20%02d-%02d-%02d %02d:%02d:%02d", + timeBytes[0], timeBytes[1], timeBytes[2], + timeBytes[3], timeBytes[4], timeBytes[5]); + } +*/ + + private static void parseDataItems(ByteBuffer buffer, Map result) { + if (buffer.remaining() > 0) { + int dataLength = buffer.get() & 0xFF; + while (buffer.remaining() >= 4) { + int itemId = buffer.getShort() & 0xFFFF; + int value = buffer.getShort() & 0xFFFF; + result.put("item_" + Integer.toHexString(itemId), value); + } + } + } +} diff --git a/emqx-plugin/src/main/java/cc/iotkit/plugins/emqx/conf/GasAlarmDataParser.java b/emqx-plugin/src/main/java/cc/iotkit/plugins/emqx/conf/GasAlarmDataParser.java index 6a6a3d8..d0b9210 100644 --- a/emqx-plugin/src/main/java/cc/iotkit/plugins/emqx/conf/GasAlarmDataParser.java +++ b/emqx-plugin/src/main/java/cc/iotkit/plugins/emqx/conf/GasAlarmDataParser.java @@ -2,10 +2,20 @@ package cc.iotkit.plugins.emqx.conf; import java.nio.ByteBuffer; import java.text.SimpleDateFormat; +import java.time.LocalDateTime; +import java.time.chrono.IsoChronology; +import java.time.format.DateTimeFormatter; +import java.time.format.ResolverStyle; import java.util.*; public class GasAlarmDataParser { public static void main(String[] args) { + // 测试数据：250707131758 -> 对应字节数组 {0x25, 0x07, 0x07, 0x13, 0x17, 0x58} + byte[] testData = new byte[]{0x25, 0x07, 0x07, 0x13, 0x17, 0x58}; + ByteBuffer buffer = ByteBuffer.wrap(testData); + + String result = parseBcdTime(buffer); + System.out.println("解析结果: " + result); // 节点相关事件测试数据 (1-12) // 节点相关事件测试数据 (1-12) - 数值部分改为低位在前 String[] nodeEvents = { @@ -63,10 +73,10 @@ public class GasAlarmDataParser { System.out.println("\n=== " + groupName + "测试 ==="); for (String hexData : testCases) { try { - List> result = GasAlarmDataParser.parseHexDataToList(hexData); + Map result = GasAlarmDataParser.parseHexDataToList(hexData); System.out.println("原始数据: " + hexData); - System.out.println("解析结果: " + result.get(0).get("eventType") - + " - " + result.get(0).get("dataType")); + System.out.println("解析结果: " + result.get("eventType") + + " - " + result.get("dataType")); System.out.println("rrr解析结果: " + result.get(0)); @@ -75,18 +85,21 @@ public class GasAlarmDataParser { } } } - public static List> parseHexDataToList(String hexData) { + + public static Map parseHexDataToList(String hexData) { + System.out.println("parseHexDataToList: " + hexData); byte[] data = hexStringToByteArray(hexData); return parseDataToList(data); } - public static List> parseDataToList(byte[] data) { - List> resultList = new ArrayList<>(); + public static Map parseDataToList(byte[] data) { + Map resultList = new HashMap<>(); if (data == null || data.length < 1) { return resultList; } int eventType = Byte.toUnsignedInt(data[0]) + Byte.toUnsignedInt(data[1]); + System.out.println("eventType: " + eventType); if (eventType == 128) { // // 正常数据上报 (NORMAL_DATA) //4.2.3 终端上报连接节点数据域格式 @@ -102,8 +115,8 @@ public class GasAlarmDataParser { return parseNodeEventToList(data); } - private static List> parseControllerEventToList(byte[] data) { - List> eventList = new ArrayList<>(); + private static Map parseControllerEventToList(byte[] data) { + // List> eventList = new ArrayList<>(); ByteBuffer buffer = ByteBuffer.wrap(data); Map eventMap = new LinkedHashMap<>(); @@ -111,29 +124,41 @@ public class GasAlarmDataParser { eventMap.put("eventType", buffer.getShort() & 0xFFFF); eventMap.put("eventTypeValue", EventTypeMapper.getEventTypeName(Integer.parseInt(eventMap.get("eventType").toString()))); eventMap.put("timestamp", parseBcdTime(buffer)); - eventList.add(eventMap); + // eventList.add(eventMap); - return eventList; + return eventMap; } - private static List> parseNodeEventToList(byte[] data) { - List> eventList = new ArrayList<>(); + private static Map parseNodeEventToList(byte[] data) { + // Map eventList = new HashMap<>(); ByteBuffer buffer = ByteBuffer.wrap(data); Map eventMap = new LinkedHashMap<>(); eventMap.put("dataType", "NODE_EVENT"); eventMap.put("eventType", buffer.getShort() & 0xFFFF); - eventMap.put("eventTypeValue", EventTypeMapper.getEventTypeName(Integer.parseInt(eventMap.get("eventType").toString()))); - eventMap.put("timestamp", parseBcdTime(buffer)); - eventMap.put("nodeId", buffer.getShort() & 0xFFFF); - eventMap.put("dataValue", buffer.getShort() & 0xFFFF); - eventList.add(eventMap); + if (!Objects.equals(eventMap.get("eventType"), 15)) { + eventMap.put("eventTypeValue", EventTypeMapper.getEventTypeName(Integer.parseInt(eventMap.get("eventType").toString()))); + eventMap.put("timestamp", parseBcdTime(buffer)); + eventMap.put("nodeId", buffer.getShort() & 0xFFFF); + eventMap.put("dataValue", buffer.getShort() & 0xFFFF); + } else { + eventMap.put("eventTypeValue", EventTypeMapper.getEventTypeName(Integer.parseInt(eventMap.get("eventType").toString()))); + eventMap.put("timestamp", parseBcdTime(buffer)); + eventMap.put("totalNodes", buffer.getShort() & 0xFF); + eventMap.put("nodeId", buffer.getShort() & 0xFFFF); + while (buffer.remaining() >= 2) { + eventMap.put("dataValue", buffer.getShort() & 0xFFFF); + } - return eventList; + } + System.out.println("eventMap+++++++++++: " + eventMap); + //eventList.add(eventMap); + + return eventMap; } - private static List> parseNormalDataToList(byte[] data) { - List> dataList = new ArrayList<>(); + private static Map parseNormalDataToList(byte[] data) { + // List> dataList = new ArrayList<>(); ByteBuffer buffer = ByteBuffer.wrap(data); // 基础信息 @@ -144,32 +169,91 @@ public class GasAlarmDataParser { baseInfo.put("timestamp", parseBcdTime(buffer)); baseInfo.put("totalNodes", buffer.getShort() & 0xFFFF); baseInfo.put("reportedNodes", buffer.get() & 0xFF); - dataList.add(baseInfo); - + // dataList.add(baseInfo); + System.out.println(buffer.remaining()); // 节点数据 - while (buffer.remaining() >= 8) { - Map nodeMap = new LinkedHashMap<>(); - nodeMap.put("nodeId", buffer.getShort() & 0xFFFF); - nodeMap.put("nodeType", buffer.get() & 0xFF); - nodeMap.put("unit", buffer.get() & 0xFF); - nodeMap.put("precision", buffer.get() & 0xFF); - nodeMap.put("gasType", buffer.get() & 0xFF); - nodeMap.put("dataValue", buffer.getShort() & 0xFFFF); - if(nodeMap.get("nodeType").equals("0")){ + // 节点数据（动态添加数字后缀） + int nodeCounter = 0; + while (buffer.remaining() >= 7) { + String suffix = String.valueOf(nodeCounter++); + // Map nodeMap = new LinkedHashMap<>(); + baseInfo.put("nodeId" + suffix, buffer.getShort() & 0xFFFF); + baseInfo.put("nodeType" + suffix, buffer.get() & 0xFF); + // nodeMap.put("unit", buffer.get() & 0xFF); // 节点节点参数 + // nodeMap.put("precision", buffer.get() & 0xFF); + baseInfo.put("gasType" + suffix, GasDetectorParser.parse(buffer.getShort() & 0xFFFF)); + baseInfo.put("dataValue" + suffix, buffer.getShort() & 0xFFFF); + if (Objects.equals(baseInfo.get("nodeType" + suffix), 0)) { // 0：节点为探测器 - nodeMap.put("dataTypeValue", nodeMap.get("dataValue")); - }else if (nodeMap.get("nodeType").equals("1")){ + baseInfo.put("dataTypeValue" + suffix, baseInfo.get("dataValue" + suffix)); + } else if (Objects.equals(baseInfo.get("nodeType" + suffix), 1)) { //1：节点为输出模块 - nodeMap.put("dataTypeValue",nodeMap.get("dataValue").equals("0")?"未动作":"已动作"); + baseInfo.put("dataTypeValue" + suffix, baseInfo.get("dataValue" + suffix).equals("0") ? "未动作" : "已动作"); } - dataList.add(nodeMap); + // dataList.add(nodeMap); } - return dataList; + return baseInfo; } + // 辅助方法：字节数组转十六进制字符串 + + private static String bytesToHex(byte[] bytes) { + + StringBuilder hexSb = new StringBuilder(); + for (byte b : bytes) { + hexSb.append(String.format("%02X", b)); + } + return hexSb.toString(); + } // 保留原有辅助方法 - private static String parseBcdTime(ByteBuffer buffer) { + /*public static String parseBcdTime(ByteBuffer buffer) { + // 打印ByteBuffer的十六进制表示 + System.out.println("ByteBuffer内容（十六进制）: " + bytesToHex(buffer.array())); + + byte[] bcdTime = new byte[6]; + + buffer.get(bcdTime); + + StringBuilder sb = new StringBuilder(); + + for (byte b : bcdTime) { + + // 提取高四位和低四位，并转换为十进制数字 + + int high = (b >> 4) & 0x0F; + + int low = b & 0x0F; + + // 组合成两位十进制字符串 + + sb.append(String.format("%1d%1d", high, low)); + + } + + String bcdStr = sb.toString(); + + try { + + SimpleDateFormat sdf = new SimpleDateFormat("yyMMddHHmmss"); + + sdf.set2DigitYearStart(new Date(946684800000L)); // 2000-01-01 + + Date date = sdf.parse(bcdStr); + + return new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(date); + + } catch (Exception e) { + + return bcdStr; // 返回原始BCD字符串（可能包含越界值） + + } + }*/ + + public static String parseBcdTime(ByteBuffer buffer) { + // 打印ByteBuffer的十六进制表示 + System.out.println("ByteBuffer内容（十六进制）: " + bytesToHex(buffer.array())); + byte[] bcdTime = new byte[6]; buffer.get(bcdTime); StringBuilder sb = new StringBuilder(); @@ -178,6 +262,7 @@ public class GasAlarmDataParser { } try { SimpleDateFormat sdf = new SimpleDateFormat("yyMMddHHmmss"); + sdf.set2DigitYearStart(new SimpleDateFormat("yyyy").parse("2000")); // 设置世纪转折点 Date date = sdf.parse(sb.toString()); return new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(date); } catch (Exception e) { @@ -190,12 +275,12 @@ public class GasAlarmDataParser { byte[] data = new byte[len / 2]; for (int i = 0; i < len; i += 2) { data[i / 2] = (byte) ((Character.digit(s.charAt(i), 16) << 4) - + Character.digit(s.charAt(i+1), 16)); + + Character.digit(s.charAt(i + 1), 16)); } return data; } private static boolean isControllerEvent(int eventType) { - return eventType >= 13 && eventType <= 21; + return eventType >= 30 && eventType <= 38; } } diff --git a/emqx-plugin/src/main/java/cc/iotkit/plugins/emqx/conf/GasDetectorParser.java b/emqx-plugin/src/main/java/cc/iotkit/plugins/emqx/conf/GasDetectorParser.java new file mode 100644 index 0000000..ea2172d --- /dev/null +++ b/emqx-plugin/src/main/java/cc/iotkit/plugins/emqx/conf/GasDetectorParser.java @@ -0,0 +1,44 @@ +package cc.iotkit.plugins.emqx.conf; + +import java.util.Map; + +public class GasDetectorParser { + + // 探测器精度映射 + private static final Map PRECISION_MAP = Map.of( + 0b000, "*1", + 0b001, "*0.1", + 0b010, "*0.01", + 0b011, "*0.001", + 0b100, "*0.0001" + ); + + // 浓度单位映射 + private static final Map UNIT_MAP = Map.of( + 0b000, "%LEL", + 0b001, "PPM", + 0b010, "%VOL" + ); + + // 气体类型映射 + private static final Map GAS_TYPE_MAP = Map.of( + 0b0000, "甲烷", + 0b0001, "一氧化碳", + 0b0010, "丙烷", + 0b0011, "氢气", + 0b0100, "苯", + 0b0101, "氧气" + ); + + public static String parse(int data) { + int precisionBits = (data >> 13) & 0b111; // 提取bit15-13 + int unitBits = (data >> 10) & 0b111; // 提取bit12-10 + int gasTypeBits = (data >> 6) & 0b1111; // 提取bit9-6 + + return String.format("探测器精度: %s, 浓度单位: %s, 测量气体: %s", + PRECISION_MAP.getOrDefault(precisionBits, "未知精度"), + UNIT_MAP.getOrDefault(unitBits, "未知单位"), + GAS_TYPE_MAP.getOrDefault(gasTypeBits, "未知气体") + ); + } +} diff --git a/emqx-plugin/src/main/java/cc/iotkit/plugins/emqx/conf/IoTConfigProtocol.java b/emqx-plugin/src/main/java/cc/iotkit/plugins/emqx/conf/IoTConfigProtocol.java index 85ad94c..507d86b 100644 --- a/emqx-plugin/src/main/java/cc/iotkit/plugins/emqx/conf/IoTConfigProtocol.java +++ b/emqx-plugin/src/main/java/cc/iotkit/plugins/emqx/conf/IoTConfigProtocol.java @@ -26,21 +26,48 @@ public class IoTConfigProtocol { public static final int TIME_SYNC = 0x0025; public static final int VERSION = 0x0026; // 生成读取指令 - public static String generateReadCommand(int terminalType, int configId) { - ByteBuffer buf = ByteBuffer.allocate(4); + // 反向映射Map + private static final Map CONFIG_ID_TO_KEY = new HashMap<>(); + + static { + // 初始化反向映射 + CONFIG_ID_TO_KEY.put(IP_REALM, "ip"); + CONFIG_ID_TO_KEY.put(PORT, "port"); + CONFIG_ID_TO_KEY.put(REPORT_AWAIT_TIME, "reportAwaitTime"); + CONFIG_ID_TO_KEY.put(REPORT_RETRY_TIMES, "reportRetryTimes"); + CONFIG_ID_TO_KEY.put(CARD, "card"); + CONFIG_ID_TO_KEY.put(USERNAME, "username"); + CONFIG_ID_TO_KEY.put(PASSWORD, "password"); + CONFIG_ID_TO_KEY.put(KEEP_ALIVE, "keepAlive"); + CONFIG_ID_TO_KEY.put(PUBLISH_TOPIC, "publishTopicPrefix"); + CONFIG_ID_TO_KEY.put(SUBSCRIBE_TOPIC, "subscribeTopicPrefix"); + CONFIG_ID_TO_KEY.put(REPORT_RATE_UPPER, "reportRateUpper"); + CONFIG_ID_TO_KEY.put(REPORT_CONCENTRATION_RANGE, "reportConcentrationRange"); + CONFIG_ID_TO_KEY.put(REPORT_RATE_LOWER, "reportRateLower"); + CONFIG_ID_TO_KEY.put(SIGNAL, "signal"); + CONFIG_ID_TO_KEY.put(TIME_SYNC, "timeSyncInternal"); + CONFIG_ID_TO_KEY.put(VERSION, "version"); + } + + public static String getKeyByConfigId(int configId) { + return CONFIG_ID_TO_KEY.get(configId); + } + + public static String generateReadCommand( int configId) { + ByteBuffer buf = ByteBuffer.allocate(2); // buf.putShort((short) 0xAA23); // 起始符+命令码 // buf.putShort((short) 4); // 数据长度 - buf.putShort((short) terminalType); + // buf.putShort((short) terminalType); buf.putShort((short) configId); return bytesToHex(buf.array()); } // 生成写入指令 - public static String generateWriteCommand(int terminalType, int configId, byte[] data) { + public static String generateWriteCommand(int configId, byte[] data) { - ByteBuffer buf = ByteBuffer.allocate(5 + data.length); + ByteBuffer buf = ByteBuffer.allocate(3 + data.length); //buf.putShort((short) (5 + data.length)); // 数据长度 - buf.putShort((short) terminalType); + // buf.putShort((short) terminalType); buf.putShort((short) configId); buf.put((byte) data.length); // 数据长度字节 buf.put(data); // 实际数据 @@ -62,6 +89,7 @@ public class IoTConfigProtocol { // 根据配置ID进行类型化解析 switch((int)result.get("configId")) { + case IP_REALM: case PUBLISH_TOPIC: case SUBSCRIBE_TOPIC: @@ -97,21 +125,38 @@ public class IoTConfigProtocol { default: result.put("value", bytesToHex(value)); } + // CONFIG_ID_TO_KEY.get((int)result.get("configId")); + result.put("key", getKeyByConfigId((int)result.get("configId"))); return result; } - // 特殊格式解析方法 +/* // 特殊格式解析方法 private static String parseTLVString(byte[] data) { int len = data[0] & 0xFF; - return new String(data, 1, Math.min(len, data.length-1), StandardCharsets.US_ASCII); + int maxAvailable = data.length - 1; // 可用最大长度 + if (maxAvailable <= 0 || len <= 0) { + return ""; // 或抛出自定义异常 + } + return new String(data, 1, Math.min(len, maxAvailable), StandardCharsets.US_ASCII); + }*/ +private static String parseTLVString(byte[] data) { + int maxAvailable = data.length; // 可用最大长度 + if (maxAvailable <= 0) { + return ""; // 或抛出自定义异常 } - + return new String(data, StandardCharsets.US_ASCII); +} private static String parseSimCard(byte[] data) { - int len = data[0] & 0xFF; - return new String(data, 1, Math.min(len, 20), StandardCharsets.US_ASCII); + // int len = data[0] & 0xFF; + int maxAvailable = data.length; // 可用最大长度 + if (maxAvailable <= 0) { + return ""; // 或抛出自定义异常 + } + return new String(data, StandardCharsets.US_ASCII); + } - private static String parseBcdTime(byte[] data) { + public static String parseBcdTime(byte[] data) { return String.format("20%02d-%02d-%02d %02d:%02d:%02d", bcdToInt(data[0]), bcdToInt(data[1]), bcdToInt(data[2]), bcdToInt(data[3]), bcdToInt(data[4]), bcdToInt(data[5])); @@ -144,6 +189,30 @@ public class IoTConfigProtocol { System.arraycopy(value, 0, data, 1, value.length); return data; } + /* + * ip地址转换成TLV数据 + * */ + public static byte[] convertToFixedAscii(String input, int fixedLength) { + // 验证输入格式 + if (!isValidIpOrDomain(input)) { + throw new IllegalArgumentException("Invalid IP/Domain format"); + } + + byte[] originBytes = input.getBytes(StandardCharsets.US_ASCII); + byte[] result = new byte[fixedLength]; + + // 复制原始数据并补0 + System.arraycopy(originBytes, 0, result, 0, Math.min(originBytes.length, fixedLength)); + return result; + } + private static boolean isValidIpOrDomain(String input) { + // IP地址正则验证 (IPv4) + String ipPattern = "^((25[0-5]|2[0-4]\\d|[01]?\\d\\d?)\\.){3}(25[0-5]|2[0-4]\\d|[01]?\\d\\d?)$"; + // 域名正则验证 (支持国际化域名IDN的ASCII形式) + String domainPattern = "^((?!-)[A-Za-z0-9-]{1,63}(?>> 1) ^ 0xA001; + } else { + crc >>>= 1; + } + } + } + return crc; + } + + +} diff --git a/emqx-plugin/src/main/java/cc/iotkit/plugins/emqx/conf/MqttHexProcessor.java b/emqx-plugin/src/main/java/cc/iotkit/plugins/emqx/conf/MqttHexProcessor.java new file mode 100644 index 0000000..23e690d --- /dev/null +++ b/emqx-plugin/src/main/java/cc/iotkit/plugins/emqx/conf/MqttHexProcessor.java @@ -0,0 +1,71 @@ +package cc.iotkit.plugins.emqx.conf; + +import org.apache.commons.codec.DecoderException; +import org.apache.commons.codec.binary.Hex; + +import java.nio.ByteBuffer; +import java.nio.charset.StandardCharsets; +import java.util.Arrays; + +public class MqttHexProcessor { + + + // 检测是否为HEX字符串格式 + private static boolean isHexString(byte[] data) { + if (data.length % 2 != 0) return false; + for (byte b : data) { + char c = (char) b; + if (!(Character.isDigit(c) || + (c >= 'A' && c <= 'F') || + (c >= 'a' && c <= 'f') || + Character.isWhitespace(c))) { // 新增空格检查 + return false; + } + } + return true; + } + + // HEX字符串转字节数组 + /* private static byte[] parseHexString(String s) { + int len = s.length(); + byte[] data = new byte[len / 2]; + for (int i = 0; i < len; i += 2) { + data[i / 2] = (byte) ((Character.digit(s.charAt(i), 16) << 4) + + Character.digit(s.charAt(i+1), 16)); + } + return data; + }*/ + + // 统一返回byte[]的入口方法 + public static byte[] processToBytes(byte[] payload) { + try { + String payloadStr = new String(payload, StandardCharsets.UTF_8); + System.out.println("payloadStr:" + payloadStr); + System.out.println("payloadStr1:" + isHexString(payload)); + return isHexString(payload) ? + Hex.decodeHex(payloadStr.replaceAll("\\s+", "")) : + payload; // 原始字节直接返回 + } catch (Exception e) { + return payload; // 异常时返回原始数据 + } + } + + + + /*public static void main(String[] args) throws DecoderException { + String input = "AA 01 01 00 35001100013836323537313037393339313237393839383631313235323037303833363037383837000000000111b12255"; // 原始字符串 + // byte[] result = convertHexStringToBytes(input); + byte[] bytes = Hex.decodeHex(input); + // 基础校验 + + + // 提取各字段 + ByteBuffer buffer = ByteBuffer.wrap(bytes); + byte type = buffer.get(1); + byte version = buffer.get(2); + System.out.println(String.format("%02X", bytes[0])); + System.out.println(type); + System.out.println(version); + + }*/ +} diff --git a/emqx-plugin/src/main/java/cc/iotkit/plugins/emqx/conf/ProtocolTest.java b/emqx-plugin/src/main/java/cc/iotkit/plugins/emqx/conf/ProtocolTest.java index 07d190f..7340762 100644 --- a/emqx-plugin/src/main/java/cc/iotkit/plugins/emqx/conf/ProtocolTest.java +++ b/emqx-plugin/src/main/java/cc/iotkit/plugins/emqx/conf/ProtocolTest.java @@ -1,14 +1,185 @@ package cc.iotkit.plugins.emqx.conf; +import io.vertx.core.buffer.Buffer; +import io.vertx.core.json.JsonObject; + +import java.net.URI; +import java.net.URISyntaxException; import java.nio.ByteBuffer; import java.text.SimpleDateFormat; import java.util.Date; +import java.util.HashMap; import java.util.Map; -import static cc.iotkit.plugins.emqx.conf.IoTConfigProtocol.buildTLVData; +import static cc.iotkit.plugins.emqx.conf.CRC16ModbusUtil.buildFrame; +import static cc.iotkit.plugins.emqx.conf.IoTConfigProtocol.*; +import static cc.iotkit.plugins.emqx.conf.MqttDataProcessor.parseFrame; public class ProtocolTest { + public static ByteBuffer hexStringToByteBuffer(String hex) { + int len = hex.length(); + byte[] data = new byte[len / 2]; + for (int i = 0; i < len; i += 2) { + data[i / 2] = (byte) ((Character.digit(hex.charAt(i), 16) << 4) + + Character.digit(hex.charAt(i+1), 16)); + } + return ByteBuffer.wrap(data); + } + public static byte[] parseFrame1(String hexString) { + if (hexString == null || hexString.length() % 2 != 0) { + throw new IllegalArgumentException("Invalid hex string"); + } + + byte[] bytes = new byte[hexString.length() / 2]; + for (int i = 0; i < hexString.length(); i += 2) { + String byteStr = hexString.substring(i, i + 2); + bytes[i / 2] = (byte) Integer.parseInt(byteStr, 16); + } + return bytes; + } + public static String asciiToHex(String asciiStr) { + StringBuilder hex = new StringBuilder(); + for (char ch : asciiStr.toCharArray()) { + hex.append(String.format("%02X ", (int) ch)); + } + return hex.toString().trim(); + } + public static String hexToAscii(String hexStr) { + StringBuilder output = new StringBuilder(); + // 移除所有空格 + hexStr = hexStr.replaceAll("\\s", ""); + + for (int i = 0; i < hexStr.length(); i += 2) { + String str = hexStr.substring(i, i + 2); + output.append((char) Integer.parseInt(str, 16)); + } + return output.toString(); + } + public static ByteBuffer createBcdTimeBuffer(String bcdTimeStr) { + // 验证输入字符串长度是否为12位(6字节BCD码) + if(bcdTimeStr == null || bcdTimeStr.length() != 12) { + throw new IllegalArgumentException("BCD时间字符串必须是12位数字"); + } + + byte[] bytes = new byte[6]; + for(int i = 0; i < 6; i++) { + // 每两位数字转换为一个BCD字节 + String twoDigits = bcdTimeStr.substring(i*2, i*2+2); + bytes[i] = (byte)Integer.parseInt(twoDigits); + } + + return ByteBuffer.wrap(bytes); + } + public static void main(String[] args) { + + // GasAlarmDataParser.parseBcdTime(createBcdTimeBuffer("250707131758")); + System.out.println(GasAlarmDataParser.parseBcdTime(createBcdTimeBuffer("250703103146"))); + //String url = "https://minio.dddd.com/feijialuo/20250702/lQLPKHYULtpjo8HM_Mz8sCPMoZZYPmEbCD1keBQpVgA_252_2521751416904105.png"; + try { + String url = "https://minio.dddd.com/feijialuo/20250702/lQLPKHYULtpjo8HM_Mz8sCPMoZZYPmEbCD1keBQpVgA_252_2521751416904105.png"; + + // 使用lastIndexOf和substring方法截取URL + int startIndex = url.lastIndexOf("/feijialuo/") + "/feijialuo/".length(); + String result = url.substring(startIndex); + System.out.println("Extracted File Name: " + result); + } catch (Exception e) { + e.printStackTrace(); + } + } + + public static String extractDateAndFileNameFromURL(String url) throws URISyntaxException { + // 解析URL并提取路径部分 + URI uri = new URI(url); + String path = uri.getPath(); + + // 分割路径为部分数组 + String[] parts = path.split("/"); + + // 遍历路径段，找到日期和文件名 + String date = null; + String fileName = null; + for (String part : parts) { + if (part.matches("\\d{8}")) { // 检查是否为8位数字日期格式 + date = part; + } else if (fileName == null && !part.isEmpty()) { // 找到第一个非空且非日期的部分作为文件名 + fileName = part; + // 由于我们已经找到了文件名，可以跳出循环（假设文件名之后没有其他重要信息） + // 但为了保持代码的通用性，这里不强制跳出循环 + } + } + + // 拼接日期和文件名 + if (date != null && fileName != null) { + return date + "/" + fileName; + } else { + throw new IllegalArgumentException("Invalid URL format: Unable to extract date and file name."); + } + } + /* public static void main(String[] args) { + String mid = "0x-102"; + String cmd= "0x-103"; + // 获取 mid 的最后两位字符 + if (mid.length() >= 2) { + mid = mid.substring(mid.length() - 2); + } else { + // 如果 mid 长度小于 2，可以根据需求设置默认值或进行其他处理 + // 例如，这里我们设置为空字符串，但实际应用中可能需要更合适的默认值或错误处理 + mid = ""; + } + System.out.println(mid); +// 获取 cmd 的最后两位字符 + if (cmd.length() >= 2) { + cmd = cmd.substring(cmd.length() - 2); + } else { + // 如果 cmd 长度小于 2，同样根据需求设置默认值或进行其他处理 + // 例如，这里我们设置为空字符串 + cmd = ""; + } + System.out.println(cmd); + System.out.println(asciiToHex("4141303130313030313330303234303032353036323530363235313034343539353434453535")); + System.out.println(hexToAscii("4141 3031 3031 3030 3133 3030 3234 3030 3235 3036 3235 3036 3235 3131 3035 3233 4234 3344 3535")); + byte[] bcdTime = buildBcdTimeData("2025-06-25 10:36:05"); + System.out.println(parseBcdTime(buildBcdTimeData("2025-06-25 10:36:05"))); + // 转换为16进制字符串输出 + System.out.print("16进制BCD时间: "); + for (byte b : bcdTime) { + System.out.print(String.format("%02X ", b)); + } + + Map original = new HashMap(); + original.put("test","tset"); + Map map = original; + map.put("test", "value"); + System.out.println(original.get("test")); // 如果输出value说明会影响 + + String replyMessage = buildFrame("01","01", + "00", "24", "0001002506250624154606"); + System.out.println("ddd" + replyMessage); + JsonObject payload = parseFrame(parseFrame1("aa01010023042300000415153839383631313235323037303833363037383837b31055")); + String hexData = "00002615153839383631313235323037303833363037383837"; + String data =payload.getString("data").replaceAll("\\s+", ""); ; + System.out.println(data); + IoTConfigProtocol.parseReadResponse(hexData); + + // 计算字节长度（每2字符=1字节） + int byteLength = hexData.length() / 2; + + // 转换为2字节16进制（大端序） + String lengthHex = String.format("%04X", byteLength); + + System.out.println("原始数据: " + lengthHex); + System.out.println("字节长度: " + byteLength + " (0x" + lengthHex + ")"); + System.out.println("2字节长度值: " + lengthHex.substring(0, 2) + " "+ lengthHex.substring(2)); + + // System.out.println("Buffer内容(HEX): " + buffer.toString("HEX")); + }*/ + + /*public static void main(String[] args) { + // 3. SIM卡号读取 (只读) + String readCardCmd = IoTConfigProtocol.generateReadCommand(0x0001, + IoTConfigProtocol.CARD); + System.out.println("读取SIM卡指令: " + readCardCmd); // 1. IP/域名配置 (TLV格式) String ip = "192.168.1.100"; byte[] ipData = buildTLVData(ip.getBytes()); @@ -23,10 +194,7 @@ public class ProtocolTest { IoTConfigProtocol.PORT, portData); System.out.println("写入端口指令: " + writePortCmd); - // 3. SIM卡号读取 (只读) - String readCardCmd = IoTConfigProtocol.generateReadCommand(0x01, - IoTConfigProtocol.CARD); - System.out.println("读取SIM卡指令: " + readCardCmd); + // 4. 时间同步设置 (BCD编码) byte[] timeData = new byte[6]; @@ -49,7 +217,7 @@ public class ProtocolTest { String readSignalCmd = IoTConfigProtocol.generateReadCommand(0x01, IoTConfigProtocol.SIGNAL); System.out.println("读取信号指令: " + readSignalCmd); - } + }*/ // 构建TLV格式数据 diff --git a/emqx-plugin/src/main/java/cc/iotkit/plugins/emqx/service/EmqxPlugin.java b/emqx-plugin/src/main/java/cc/iotkit/plugins/emqx/service/EmqxPlugin.java index 861048b..e4d0e28 100644 --- a/emqx-plugin/src/main/java/cc/iotkit/plugins/emqx/service/EmqxPlugin.java +++ b/emqx-plugin/src/main/java/cc/iotkit/plugins/emqx/service/EmqxPlugin.java @@ -6,18 +6,15 @@ import cc.iotkit.common.utils.UniqueIdUtil; import cc.iotkit.plugin.core.IPlugin; import cc.iotkit.plugin.core.IPluginConfig; import cc.iotkit.plugin.core.thing.IThingService; -import cc.iotkit.plugin.core.thing.actions.ActionResult; -import cc.iotkit.plugin.core.thing.actions.DeviceState; -import cc.iotkit.plugin.core.thing.actions.EventLevel; -import cc.iotkit.plugin.core.thing.actions.IDeviceAction; +import cc.iotkit.plugin.core.thing.actions.*; +import cc.iotkit.plugin.core.thing.actions.down.DeviceConfig; import cc.iotkit.plugin.core.thing.actions.up.*; import cc.iotkit.plugin.core.thing.model.ThingDevice; -import cc.iotkit.plugins.emqx.conf.DeviceStatusParser; -import cc.iotkit.plugins.emqx.conf.GasAlarmDataParser; -import cc.iotkit.plugins.emqx.conf.MqttConfig; +import cc.iotkit.plugins.emqx.conf.*; import cn.hutool.core.bean.BeanUtil; import cn.hutool.core.bean.copier.CopyOptions; import cn.hutool.core.util.IdUtil; +import cn.hutool.core.util.ObjectUtil; import com.gitee.starblues.bootstrap.annotation.AutowiredType; import com.gitee.starblues.bootstrap.realize.PluginCloseListener; import com.gitee.starblues.core.PluginCloseType; @@ -25,6 +22,7 @@ import com.gitee.starblues.core.PluginInfo; import io.netty.handler.codec.mqtt.MqttQoS; import io.vertx.core.Future; import io.vertx.core.Vertx; +import io.vertx.core.buffer.Buffer; import io.vertx.core.json.JsonArray; import io.vertx.core.json.JsonObject; import io.vertx.mqtt.MqttClient; @@ -36,6 +34,7 @@ import org.springframework.context.support.GenericApplicationContext; import org.springframework.stereotype.Service; import javax.annotation.PostConstruct; +import java.nio.charset.StandardCharsets; import java.util.*; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.CountDownLatch; @@ -45,6 +44,9 @@ import java.util.concurrent.TimeUnit; import static cc.iotkit.plugins.emqx.conf.BCDClockGenerator.formatBCDToHex; import static cc.iotkit.plugins.emqx.conf.BCDClockGenerator.generateBCDTime; import static cc.iotkit.plugins.emqx.conf.CRC16ModbusUtil.buildFrame; +import static cc.iotkit.plugins.emqx.conf.MqttDataProcessor.parseFrame; +import static cc.iotkit.plugins.emqx.conf.MqttHexProcessor.processToBytes; + /** * @author sjg @@ -97,8 +99,8 @@ public class EmqxPlugin implements PluginCloseListener, IPlugin, Runnable { String serverPassword = IdUtil.fastSimpleUUID(); MqttClientOptions options = new MqttClientOptions() - .setClientId("server") - .setUsername("server") + .setClientId("server11223") + .setUsername("server11223") .setPassword(serverPassword) .setCleanSession(true) .setMaxInflightQueue(100) @@ -175,16 +177,21 @@ public class EmqxPlugin implements PluginCloseListener, IPlugin, Runnable { }).publishHandler(msg -> { String topic = msg.topicName(); log.info("topic={}",topic); - if (topic.contains("/c/")) { + if (topic.contains("/request/")) { return; } - if (topic != null && !topic.startsWith("/sys/") && topic.endsWith("/rtdvalue/report")){ - String s = Arrays.asList(topic.split("/")).get(1); - topic = "/sys/*/"+s+"/s/event/property/post"; - log.info("Client received message on [{}] payload [{}] with QoS [{}]", topic, msg.payload().toJsonObject(), msg.qosLevel()); + if (topic != null && topic.startsWith("FIGARO/")){ + String s = Arrays.asList(topic.split("/")).get(2); + topic = "FIGARO/post/*/"+s; + // log.info("Client received message on [{}] payload [{}] with QoS [{}]", topic, msg.payload().toJsonObject(), msg.qosLevel()); } - JsonObject payload = msg.payload().toJsonObject(); - + // JsonObject payload =msg.payload().toJsonObject(); + // JsonObject payload = parseFrame(msg.payload().getBytes()); + JsonObject payload = parseFrame(processToBytes(msg.payload().getBytes())); + System.out.println("payload原始数据++++++++++++++++++++" + payload); + if(ObjectUtil.isEmpty(payload)){ + return; + } try { //客户端连接断开 if (topic.equals("/sys/client/disconnected")) { @@ -204,27 +211,37 @@ public class EmqxPlugin implements PluginCloseListener, IPlugin, Runnable { IDeviceAction action = null; String method = payload.getString("method", ""); if(StringUtils.isBlank(method)){ - method = "thing.event.property.post_reply"; + method = "thing.event.property.post"; } if (StringUtils.isBlank(method)) { return; } + boolean flag = true; JsonObject params = null; - + String replyHex = null; if(!StringUtils.isBlank(payload.getString("method", ""))){ params = payload.getJsonObject("params", defParams); }else{ String cmd =payload.getString("cmd"); - String data =payload.getString("data"); + String data =payload.getString("data").replaceAll("\\s+", ""); ; + String rawHex =payload.getString("rawHex").replaceAll("\\s+", ""); ; + replyHex = replyToInstructions(payload);//回复的数据 + Map mapRaw = new HashMap<>(); + mapRaw.put("rawHex", rawHex); + mapRaw.put("replyHex", replyHex); + Map result = new HashMap<>(mapRaw); if(ObjectUtils.isNotEmpty(cmd)&& ObjectUtils.isNotEmpty(data)){ - if(cmd.equals("11")){ + if(cmd.equals("0x11")){ + //4.1 状态信息上报（命令码0x11） - Map map = DeviceStatusParser.parseStatusReportToMap(data); + DeviceStatusParser.parseStatusReportToMap(data).forEach( + (key, value) -> result.merge(key, value, (v1, v2) -> v2) // 冲突时保留后者 + ); // params= payload.getJsonObject(DeviceStatusParser.parseStatusReportToMap(data),defParams); - payload.put("params",map); - device.setDeviceName(map.get("imei").toString()); - switch(map.get("deviceType").toString()) { + payload.put("params",result); + device.setDeviceName(result.get("imei").toString()); + switch(result.get("deviceType").toString()) { case "0": device.setProductKey("CEMpmANABN7Tt6Jh"); @@ -242,26 +259,73 @@ public class EmqxPlugin implements PluginCloseListener, IPlugin, Runnable { break; } - - - }else if(cmd.equals("12")){ + //注册 + IDeviceAction action1 = DeviceRegister.builder() + .model("FIGARO") + .id(UUID.randomUUID().toString()) + .deviceName(device.getDeviceName()) + .productKey(device.getProductKey()) + .params(result) + .time(System.currentTimeMillis()) + .build(); + thingService.post(pluginInfo.getPluginId(), action1); + + + }else if(cmd.equals("0x12")){ //4.2 燃气报警器物联网模块数据上报（0x12） - payload.put("params",GasAlarmDataParser.parseHexDataToList(data)); + GasAlarmDataParser.parseHexDataToList(data).forEach( + (key, value) -> result.merge(key, value, (v1, v2) -> v2)); + payload.put("params",result); // GasAlarmDataParser.parseHexDataToList(data); - }else if(cmd.equals("14")){ + }else if(cmd.equals("0x14")){ + //动火离人数据上报与事件上报 + FireSafetyDataReceiver.handleHexData(data).forEach( + (key, value) -> result.merge(key, value, (v1, v2) -> v2)); + payload.put("params",result); } + else if(cmd.equals("0x23")){ + flag=false; + IoTConfigProtocol.parseReadResponse(data).forEach( + (key, value) -> result.merge(key, value, (v1, v2) -> v2)); + //读取配置数据 + payload.put("params",result); + + //读取配置 + DeviceConfig config = DeviceConfig.builder() + .id(UUID.randomUUID().toString()) + .deviceName(device.getDeviceName()) + .productKey(device.getProductKey()) + .time(System.currentTimeMillis()) + .config(payload.getJsonObject("params", defParams).getMap()) + .build(); + thingService.post(pluginInfo.getPluginId(), config); + + + } + else if(cmd.equals("0x24")){ + flag=false; + //下发配置数据的回复 + // payload.put("params", FireSafetyDataReceiver.handleHexData(data)); + } + } // params = //解析设备数据 } + params = payload.getJsonObject("params", defParams); + System.out.println("params+++++++++-----------++++++++++++++++++++++++++" + params); + if (ObjectUtil.isNull(params) || params.isEmpty()){ + return; + } // System.out.println(payload.getJsonObject("data", defParams)); if (ObjectUtils.isNotEmpty(method) && "thing.lifetime.register".equalsIgnoreCase(method)) { //子设备注册 String subPk = params.getString("productKey"); String subDn = params.getString("deviceName"); String subModel = params.getString("model"); + ActionResult regResult = thingService.post( pluginInfo.getPluginId(), fillAction( @@ -308,7 +372,9 @@ public class EmqxPlugin implements PluginCloseListener, IPlugin, Runnable { } try { - reply(topic, payload, 0); + if(flag) { + reply1(replyHex,device.getDeviceName(), payload); + } }catch (Exception e){ log.error("reply error", e); } @@ -316,7 +382,7 @@ public class EmqxPlugin implements PluginCloseListener, IPlugin, Runnable { if (action == null) { return; } - action.setId(payload.getString("id")); + action.setId(UUID.randomUUID().toString()); action.setProductKey(device.getProductKey()); action.setDeviceName(device.getDeviceName()); action.setTime(System.currentTimeMillis()); @@ -334,11 +400,15 @@ public class EmqxPlugin implements PluginCloseListener, IPlugin, Runnable { } } + + + public ThingDevice getDevice(String topic) { String[] topicParts = topic.split("/"); - if (topicParts.length < 5) { + if (topicParts.length < 4) { return null; } + System.out.println(topicParts[2]); return ThingDevice.builder() .productKey(topicParts[2]) .deviceName(topicParts[3]) @@ -346,9 +416,9 @@ public class EmqxPlugin implements PluginCloseListener, IPlugin, Runnable { } public void online(String pk, String dn) { - if (Boolean.TRUE.equals(DEVICE_ONLINE.get(dn))) { + /* if (Boolean.TRUE.equals(DEVICE_ONLINE.get(dn))) { return; - } + }*/ //上线 thingService.post( @@ -387,7 +457,79 @@ public class EmqxPlugin implements PluginCloseListener, IPlugin, Runnable { action.setTime(System.currentTimeMillis()); return action; } - + private String replyToInstructions(JsonObject payload) { + String mid = payload.getString("messageId"); + String cmd = payload.getString("cmd"); + payload.getString("cmd"); + System.out.println("mid1" + mid); + System.out.println("cmd1" + cmd); + if (mid.toLowerCase().startsWith("0x")) { + mid = mid.substring(2); + } + if (cmd.toLowerCase().startsWith("0x")) { + cmd = cmd.substring(2); + } + cmd = cmd.replaceAll("\\D", ""); + mid = mid.replaceAll("\\D", ""); + if (mid.length() % 2 != 0) { + mid = "0" + mid; // 左侧补零 + } + if (cmd.length() % 2 != 0) { + cmd = "0" + cmd; // 左侧补零 + } + System.out.println("mid" + mid); + System.out.println("cmd" + cmd); + // System.out.println(mid + "re++++++++++++++++++++++++++++++++++++++++++++++++++" + cmd); + // System.out.println("BCD Time: " + formatBCDToHex(generateBCDTime())); + String replyMessage = buildFrame("01", "01", + mid, cmd, "00" + formatBCDToHex(generateBCDTime())); + return replyMessage; + } + /** + * 回复设备 + */ + private void reply1(String replyMessage,String deviceName, JsonObject payload) { + Map payloadReply = new HashMap<>(); + if(ObjectUtil.isNull(replyMessage)) { + String mid = payload.getString("messageId"); + String cmd = payload.getString("cmd"); + payload.getString("cmd"); + System.out.println("mid1" + mid); + System.out.println("cmd1" + cmd); + if (mid.toLowerCase().startsWith("0x")) { + mid = mid.substring(2); + } + if (cmd.toLowerCase().startsWith("0x")) { + cmd = cmd.substring(2); + } + cmd = cmd.replaceAll("\\D", ""); + mid = mid.replaceAll("\\D", ""); + if (mid.length() % 2 != 0) { + mid = "0" + mid; // 左侧补零 + } + if (cmd.length() % 2 != 0) { + cmd = "0" + cmd; // 左侧补零 + } + System.out.println("mid" + mid); + System.out.println("cmd" + cmd); + // System.out.println(mid + "re++++++++++++++++++++++++++++++++++++++++++++++++++" + cmd); + // System.out.println("BCD Time: " + formatBCDToHex(generateBCDTime())); + replyMessage = buildFrame("01", "01", + mid, cmd, "00" + formatBCDToHex(generateBCDTime())); + } + payloadReply.put("data", replyMessage); + String topic="FIGARO/request/" + deviceName; + /* byte[] bytes = replyMessage.getBytes(StandardCharsets.UTF_8); + Buffer buffer = Buffer.buffer(bytes);*/ + client.publish(topic, Buffer.buffer(replyMessage), MqttQoS.AT_LEAST_ONCE, false, false) + .onSuccess(h -> { + log.info("publish {} success", topic); + }); + /* client.publish(topic, JsonObject.mapFrom(payloadReply).toBuffer(), MqttQoS.AT_LEAST_ONCE, false, false) + .onSuccess(h -> { + log.info("publish {} success", finalTopic); + });*/ + } /** * 回复设备 */ @@ -396,9 +538,9 @@ public class EmqxPlugin implements PluginCloseListener, IPlugin, Runnable { // System.out.println("BCD Time: " + formatBCDToHex(generateBCDTime())); String replyMessage = buildFrame("01","01", - "01","01","01"+formatBCDToHex(generateBCDTime())); + "00","01","01"+formatBCDToHex(generateBCDTime())); - if( StringUtils.isBlank(payload.getString("method"))){ + if( StringUtils.isBlank(payload.getString("method"))){ payloadReply.put("id", payload.getString("messageID")); payloadReply.put("method", "thing.event.property.post_reply"); payloadReply.put("code", code); @@ -436,15 +578,17 @@ public class EmqxPlugin implements PluginCloseListener, IPlugin, Runnable { } payloadReply.put("data", replyMessage); String finalTopic = topic; - - client.publish(topic, JsonObject.mapFrom(payloadReply).toBuffer(), MqttQoS.AT_LEAST_ONCE, false, false) + /* String dadad="FIGARO/request/862571079391279"; + byte[] bytes = replyMessage.getBytes(StandardCharsets.UTF_8); + Buffer buffer = Buffer.buffer(bytes);*/ + /* client.publish(topic, payloadReply.get(""), MqttQoS.AT_LEAST_ONCE, false, false) .onSuccess(h -> { log.info("publish {} success", finalTopic); - }); - /* client.publish(topic, JsonObject.mapFrom(payloadReply).toBuffer(), MqttQoS.AT_LEAST_ONCE, false, false) + });*/ + client.publish(topic, JsonObject.mapFrom(payloadReply).toBuffer(), MqttQoS.AT_LEAST_ONCE, false, false) .onSuccess(h -> { log.info("publish {} success", finalTopic); - });*/ + }); } public Map getConfigMap(JsonObject payload) { diff --git a/emqx-plugin/src/main/java/cc/iotkit/plugins/emqx/service/MqttDevice.java b/emqx-plugin/src/main/java/cc/iotkit/plugins/emqx/service/MqttDevice.java index f03a3a3..76fca71 100644 --- a/emqx-plugin/src/main/java/cc/iotkit/plugins/emqx/service/MqttDevice.java +++ b/emqx-plugin/src/main/java/cc/iotkit/plugins/emqx/service/MqttDevice.java @@ -12,12 +12,14 @@ import cc.iotkit.plugin.core.thing.actions.down.ServiceInvoke; import cc.iotkit.plugins.emqx.conf.IoTConfigProtocol; import cn.hutool.core.util.ObjectUtil; import io.netty.handler.codec.mqtt.MqttQoS; +import io.vertx.core.buffer.Buffer; import io.vertx.core.json.JsonObject; import io.vertx.mqtt.MqttClient; import lombok.Setter; import org.springframework.stereotype.Service; import java.nio.ByteBuffer; +import java.nio.charset.StandardCharsets; import java.util.ArrayList; import java.util.HashMap; import java.util.List; @@ -40,6 +42,7 @@ public class MqttDevice implements IDevice { @Setter private MqttClient client; + public List processConfig(Map config) { if (config == null || config.isEmpty()) { throw new IllegalArgumentException("Config map cannot be null or empty"); @@ -49,47 +52,48 @@ public class MqttDevice implements IDevice { Map map = new HashMap(); String key = entry.getKey(); String value = entry.getValue().toString(); - Integer configId = null; + Integer configId = null; try { byte[] tlvData = null; - // int configId = getConfigId(key); - - /* if (configId == -1) { - System.err.println("Unknown config key: " + key); - continue; - } -*/ - switch(key.toLowerCase()) { - case "ip/realm": + switch (key) { + case "ip": configId = IP_REALM; - tlvData = buildTLVData(value.getBytes()); + tlvData = convertToFixedAscii(value,50); break; case "port": - case "keepalive": - case "reportrateupper": + case "keepAlive": + case "reportRateUpper": - configId = (key.toLowerCase().equals("port") ? PORT : key.toLowerCase().equals("keepalive") ?KEEP_ALIVE :REPORT_RATE_UPPER); - tlvData= ByteBuffer.allocate(2).putShort( Short.parseShort(value)).array(); - // tlvData = buildShortData(parseNumber(value)); + configId = (key.equals("port") ? PORT : key.equals("keepAlive") ? KEEP_ALIVE : REPORT_RATE_UPPER); + tlvData = ByteBuffer.allocate(2).putShort(Short.parseShort(value)).array(); + // tlvData = buildShortData(parseNumber(value)); break; - case "reportawaittime": - case "reportretrytimes": - configId = (key.toLowerCase().equals("reportawaittime") ? REPORT_AWAIT_TIME :REPORT_RETRY_TIMES); + case "reportAwaitTime": + case "reportRetryTimes": + configId = (key.equals("reportAwaitTime") ? REPORT_AWAIT_TIME : REPORT_RETRY_TIMES); tlvData = buildByteData(Integer.valueOf(value)); break; - // case "username": - // case "password": + // case "username": + // case "password": // tlvData = buildStringData(value.toString(), MAX_AUTH_LENGTH); - // break; - case "publishtopicprefix": - case "subscribetopicprefix": - configId = (key.toLowerCase().equals("publishtopicprefix") ? PUBLISH_TOPIC :SUBSCRIBE_TOPIC); + // break; + case "publishTopicPrefix": + case "subscribeTopicPrefix": + configId = (key.equals("subscribeTopicPrefix") ? SUBSCRIBE_TOPIC : PUBLISH_TOPIC); tlvData = buildStringData(value.toString(), 30); break; - case "timesyncinternal": + case "timeSyncInternal": configId = TIME_SYNC; tlvData = buildBcdTimeData(value); break; + case "reportConcentrationRange": + configId = REPORT_CONCENTRATION_RANGE; + tlvData = ByteBuffer.allocate(2).putShort(Short.parseShort(value)).array(); + break; + case "reportRateLower": + configId = REPORT_RATE_LOWER; + tlvData = ByteBuffer.allocate(2).putShort(Short.parseShort(value)).array(); + break; case "card": case "signal": case "version": @@ -99,39 +103,165 @@ public class MqttDevice implements IDevice { System.out.println("Unsupported config: " + key); continue; } - if(ObjectUtil.isNotNull(configId)) { - String command = IoTConfigProtocol.generateWriteCommand( - 0x01, configId, tlvData); - map.put("date",command); - System.out.printf("Generated command for %s (ID:0x%04X): %s%n", - key, configId, command); - } + if (ObjectUtil.isNotNull(configId)) { + String command = IoTConfigProtocol.generateWriteCommand( + configId, tlvData); + map.put("data", command); + list.add(map); + System.out.printf("Generated command for %s (ID:0x%04X): %s%n", + key, configId, command); + } + + } catch (Exception e) { + System.err.printf("Error processing %s: %s%n", key, e.getMessage()); + } + } + return list; + } + + public List processGetConfig(Map config) { + if (config == null || config.isEmpty()) { + throw new IllegalArgumentException("Config map cannot be null or empty"); + } + List list = new ArrayList<>(); + for (Map.Entry entry : config.entrySet()) { + Map map = new HashMap(); + String key = entry.getKey(); + Integer configId = null; + try { + switch (key) { + case "ip": + configId = IP_REALM; + break; + case "port": + configId = PORT; + break; + + case "reportAwaitTime": + configId = REPORT_AWAIT_TIME; + break; + + case "reportRetryTimes": + configId = REPORT_RETRY_TIMES; + break; + case "card": + configId = CARD; + break; + case "username": + configId = USERNAME; + break; + case "password": + configId = PASSWORD; + break; + case "keepAlive": + configId = KEEP_ALIVE; + break; + case "publishTopicPrefix": + configId = PUBLISH_TOPIC; + break; + case "subscribeTopicPrefix": + configId = SUBSCRIBE_TOPIC; + break; + case "reportRateUpper": + configId = REPORT_RATE_UPPER; + break; + case "reportConcentrationRange": + configId = REPORT_CONCENTRATION_RANGE; + break; + case "reportRateLower": + configId = REPORT_RATE_LOWER; + break; + case "signal": + configId = SIGNAL; + break; + case "timeSyncInternal": + configId = TIME_SYNC; + break; + case "version": + configId = VERSION; + break; + default: + System.out.println("Unsupported config: " + key); + continue; + } + if (ObjectUtil.isNotNull(configId)) { + String command = IoTConfigProtocol.generateReadCommand( + configId); + map.put("data", command); + list.add(map); + System.out.println("ddddeeeeeeeeeeeeeeddddd" + map); + /* System.out.printf("Generated command for %s (ID:0x%04X): %s%n", + key, configId, command);*/ + } } catch (Exception e) { System.err.printf("Error processing %s: %s%n", key, e.getMessage()); } } + System.out.println("ddddddddddddddddddddddddddddddddd" + list); return list; } + @Override public ActionResult config(DeviceConfig action) { - String topic = String.format("/sys/%s/%s/c/config/set", action.getProductKey(), action.getDeviceName()); - // Map data = JsonUtils.parseObject(action.getConfig(), Map.class); - List list= processConfig(action.getConfig()); - for (int i = 0; i < list.size(); i++) { - //循环发送配置指令 - String replyMessage = buildFrame("01","01", - "01","01","01"+ list.get(i).get("data")); - send( - topic, - new JsonObject() - .put("id", action.getId()) - .put("method", "thing.config.set") - .put("params", action.getConfig()) - .put("data", replyMessage) - ); - } + String topic = "FIGARO/request/"+ action.getDeviceName(); + // Map data = JsonUtils.parseObject(action.getConfig(), Map.class); + /* int type = 0x0000; + switch (action.getProductKey()) { + case "CEMpmANABN7Tt6Jh0": + type = 0x0000; + break; + case "XmXYxjzihseT76As": + type = 0x0001; + break; + case "bAASX8tBjYQjBGFP": + type = 0x0002; + break; + case "WfpZZFkMxxbGfRca": + type = 0x0003; + break; + }*/ + System.out.println("action.getProductKey()++++++++++++++++++++" + action.getProductKey()); + if (action.getModule().equals("set")) { + List list = processConfig(action.getConfig()); + System.out.println("listset++++++++++++++++++++" + list); + for (int i = 0; i < list.size(); i++) { + String replyMessage = buildFrame("01","01", + "00", "24", list.get(i).get("data").toString()); + //循环发送配置指令 + /* String replyMessage = buildFrame("01", "01", + "01", "24", "01" + list.get(i).get("data"));*/ + send( + topic,replyMessage + ); + try { + Thread.sleep(500); + }catch (Exception e){ + } + + } + } else if(action.getModule().equals("get")){ + System.out.println("action.getConfig()++++++++++++++++++++" + action.getConfig()); + //配置获取指令下发 + List list = processGetConfig(action.getConfig()); + System.out.println("list++++++++++++++++++++" + list); + for (int i = 0; i < list.size(); i++) { + //循环发送配置指令 + /* String replyMessage = buildFrame("01", "01", + "01", "23", "01" + list.get(i).get("data"));*/ + String replyMessage = buildFrame("01","01", + "00", "23",list.get(i).get("data").toString()); + + send( + topic,replyMessage + ); + try { + Thread.sleep(500); + }catch (Exception e){ + } + } + } return ActionResult.builder().code(0).reason("").build(); } @@ -182,5 +312,17 @@ public class MqttDevice implements IDevice { return ActionResult.builder().code(ErrCode.UNKNOWN_EXCEPTION.getKey()).reason(e.getMessage()).build(); } } + private ActionResult send(String topic, String payload) { + try { + /* byte[] bytes = payload.getBytes(StandardCharsets.UTF_8); + Buffer buffer = Buffer.buffer(bytes);*/ + client.publish(topic, Buffer.buffer(payload), MqttQoS.AT_LEAST_ONCE, false, false); + return ActionResult.builder().code(0).reason("").build(); + } catch (BizException e) { + return ActionResult.builder().code(e.getCode()).reason(e.getMessage()).build(); + } catch (Exception e) { + return ActionResult.builder().code(ErrCode.UNKNOWN_EXCEPTION.getKey()).reason(e.getMessage()).build(); + } + } } diff --git a/pom.xml b/pom.xml index e16fda1..b486840 100644 --- a/pom.xml +++ b/pom.xml @@ -20,7 +20,7 @@ 2.7.11 - 2.10.19 + 2.10.18 cc.iotkit.plugins iot-iita-plugins