KtGate-MQ1 技术开发文档 （测试工具下载）

1. 产品概述

KtGate-MQ1是一款通用的 4G DTU 边缘网关，主要用于工业设备的数据采集与远程监控。本产品支持 Modbus RTU 协议通信，用户不需要关心具体通讯协议，网关实现数据采集和解析的业务逻辑，然后通过 MQTT 协议把最终解析好的数据上传用户Mqtt服务器。

1.1 主要功能

• Modbus RTU 通信：支持通过 RS485 接口与工业设备通信
• MQTT 协议通信：支持连接用户自有 MQTT 服务器
• 远程配置管理：支持通过 MQTT 消息或 AT 指令进行配置
• 定时任务：支持变化上报和定时上报
• 状态监控：实时监控 SIM 卡状态、网络连接状态和串口通信状态
• 指示灯状态指示：通过 LED 灯直观显示设备工作状态

1.2 硬件规格

• 通信接口：RS485（支持 Modbus RTU）
• 网络制式：4G CAT1
• 指示灯：
  • LINK 灯：指示网络连接状态
  • NET 灯：指示串口通信状态
  • 信号强度指示灯：3 个 LED 灯显示信号强度

2. 快速开始

本章节帮助开发人员在拿到网关后，在最短时间内跑通数据上报流程。

2.1 硬件准备

1. 插入有效 SIM 卡（确保未欠费且支持 4G 网络）。
2. 拧紧 4G 天线。
3. 通过 USB 转 RS485 模块 将网关与电脑连接。
4. 给网关供电，等待 LINK 灯 进入常亮状态（表示 MQTT 连接成功）。

2.2 串口连接与验证

使用串口调试工具（如 SSCOM、XCOM），配置以下参数：

发送以下指令验证通信：

AT+VER\r\n
正常响应示例：

AT+INFO={Ver:1.0.1,iccid:89860012345678901234,imei:861234567890123}

2.3 配置用户 MQTT 服务器

发送以下指令配置你自己的 MQTT 服务器（以 mqtt.example.com 为例）：

AT+SETMQTT=mqtt.example.com,1883,user1,password1,0\r\n
响应应为：

AT+INFO=set_mqtt_ok

2.4 验证数据上报

订阅设备的上行主题，等待数据上报：

• 主题：/KeterRtu/{设备ID}/up/change

如果收到类似以下 JSON 数据，则说明配置成功：

json

{
"devSn": "U4GC0000000000",
"comm_s": 0,
"varLst": {
"M01": "123.4",
"D01": "1"
}
}

注意：设备ID可通过 AT+GETRTUSN 指令获取，设备ID=UserGsn不为空，否则GSN，例如 U4GC0000000000。

3. AT指令配置网关指南

3.1 语法规范

AT指令的基本格式为：

AT+<命令>[=<参数>]

• 命令必须以 AT+ 开头
• 命令名称通常为大写字母
• 参数之间用逗号分隔
• 指令以 \r\n 结尾
• 设备响应格式：AT+INFO=<响应内容>\r\n

3.2 命令集说明

3.3 参数定义

3.3.1 AT+SETMQTT 参数

• host: MQTT服务器主机地址
• port: MQTT服务器端口
• name: MQTT用户名
• pwd: MQTT密码
• ssl: SSL标志（0：不使用SSL，1：使用SSL）

3.4 使用方法

3.4.1 基本操作流程

1. 连接设备：使用串口调试工具连接设备的RS485接口
2. 发送指令：按照AT指令格式发送命令
3. 接收响应：等待并接收设备的响应
4. 解析响应：根据响应内容进行相应处理

3.4.2 配置示例

示例1：获取设备信息

AT+VER

响应：
AT+INFO={Ver:1.0.1,iccid:89860012345678901234,imei:861234567890123}

示例2：设置用户MQTT配置

AT+SETMQTT=mqtt.example.com,1883,user1,password1,0

响应：
AT+INFO=set_mqtt_ok

示例3：上传CA证书

AT+SETMQTTCA

响应：
AT+INFO=ready_for_ca

然后发送证书内容：
-----BEGIN CERTIFICATE-----
MIICzjCCAbegAwIBAgIUWN...
...
-----END OF CERTIFICATE-----

响应：
AT+INFO=set_ca_ok

3.5 注意事项

1. 指令格式：确保AT指令以\r\n结尾，否则设备可能无法正确解析
2. 参数校验：发送指令前应验证参数的有效性，避免发送错误的参数
3. 响应等待：发送指令后应等待设备响应，避免连续发送导致缓冲区溢出
4. CA证书：上传CA证书时，确保证书格式正确，包含完整的开始和结束标记
5. 超时处理：设置合理的响应超时时间，避免长时间等待导致程序阻塞

3.6 常见问题解决方案

4. 设备变量描述表说明

重要提示：设备变量描述表是数据解析的基础，所有MQTT数据上报和变量操作都必须基于此表。在开发应用前，必须先获取并理解设备的变量描述表，确保后续数据传输的准确性和一致性。

4.1 获取变量描述表

客户应通过以下流程获取变量描述表信息：

1. 连接MQTT服务器：使用配置的参数连接到MQTT服务器
2. 订阅上行主题：订阅设备的上行主题，用于接收响应
3. 发送请求：向设备的下行主题发送包含 { "ParType": "DevDesp" } 的消息
4. 接收响应：接收设备返回的变量描述表信息
5. 解析并存储：解析变量描述表并在应用中存储，用于后续数据处理

4.1.1 获取变量描述表示例

请求：

• 主题：/KeterRtu/MyDevice001/down/readCon
• 消息：

{ "ParType": "DevDesp" }

响应：

• 主题：/KeterRtu/MyDevice001/up/readCon
• 消息：

{
  "ParType": "DevDesp",
  "Device": {
    "实时状态": {
      "M01": {
        "Desp": "温度",
        "Unit": "℃"
      },
      "M02": {
        "Desp": "湿度",
        "Unit": "%RH"
      },
      "M03": {
        "Desp": "系统故障"
      }
    },
    "远程控制": {
      "On/Off": {
        "Desp": "模式设置",
        "Remark": "8=关机模式 1=制冷模式 2=制热模式 4=水泵模式"
      },
      "M110": {
        "Desp": "制冷温度设定",
        "Unit": "℃",
        "IsWrite": 1
      }
    }
  }
}

4.2 变量描述表字段说明

4.2.1 一级字段

4.2.2 变量字段

4.2.3 变量编码规则

变量编码由系统根据以下规则生成：

• 如果 ClassNo > 0，格式为 No_Code（例如：1_M01）
• 如果 ClassNo = 0，格式为原始的 Code（例如：M01）

系统会确保变量编码的唯一性，避免重复。

4.3 变量描述表应用说明

4.3.1 数据解析流程

1. 获取变量描述表：通过MQTT发送 { "ParType": "DevDesp" } 获取设备的变量描述表
2. 解析变量描述表：提取"实时状态"和"远程控制"下的所有变量
3. 建立映射关系：建立变量编码与变量配置的映射关系
4. 解析数据：根据变量描述表解析MQTT上报的数据
5. 展示数据：结合变量描述和单位展示数据

4.3.2 变量写入流程

1. 查找变量配置：根据变量编码在变量描述表中查找配置
2. 检查可写性：确认变量的 IsWrite 字段为 1（可写）
3. 发送写入命令：向设备发送变量写入命令
4. 接收响应：确认写入是否成功

4.3.3 数据解析示例

假设收到以下数据上报：

{
  "devSn": "MyDevice001",
  "comm_s": 0,
  "varLst": {
    "M01": "123.4",
    "M02": "65.5",
    "M03": "1"
  }
}

根据变量描述表解析：

• M01：Desp=温度, Unit=℃
  • 显示值：123.4℃
• M02：Desp=湿度, Unit=%RH
  • 显示值：65.5%RH
• M03：Desp=系统故障
  • 显示值：1

5. MQTT协议配置网关指南

5.1 连接参数设置

5.1.1 用户MQTT配置

5.2 主题结构定义

5.2.1 用户主题

• 上行主题：/KeterRtu/{设备ID}/up/{消息类型}
• 下行主题：/KeterRtu/{设备ID}/down/{消息类型}

5.2.2 消息类型

• reg：设备注册信息
• data：定时上报数据
• readCon：读取配置
• writeCon：写入配置
• readVar：读取变量
• writeVar：写入变量
• Err：错误信息

5.3 消息格式规范

5.3.1 读取配置（readCon）

• 请求：

{ "ParType": "Getway" }

• 响应：

{
  "ParType": "Getway",
  "varLst": {
    "report_interval": 5
  }
}

5.3.2 写入配置（writeCon）

• polling_interval: 轮询周期（秒）【默认值：5秒，每5秒轮询一次下位机】

• response_timeout: 响应超时（毫秒）【默认值：1秒，等待1秒未收到下位机响应则认为指令未响应】

• delay_between_poll: 两条指令之间的间隔（毫秒）【默认值：200毫秒，两条指令之间间隔200毫秒】

• 485_uart: 波特率（如9600, 115200等），数据位（通常为8），校验位（None, Even, Odd），停止位（通常为1）

• 请求：

{
  "ParType": "Com",
  "varLst": {
    "polling_interval": 3,
    "response_timeout": 500,
    "delay_between_poll": 200,
    "485_uart": ["9600", 8, "None", 1]
  }
}

• 响应：

{
  "ParType": "Com",
  "varLst": {
    "polling_interval": 3,
    "response_timeout": 500,
    "delay_between_poll": 200,
    "485_uart": ["9600", 8, "None", 1],
    "flag": 1
  }
}

5.3.3 远程重启

• 请求：

{ "ParType": "RST" }

• 响应：无（设备直接重启）

5.4 认证方式

5.4.1 用户名密码认证

• 用户MQTT：使用GsnUser或设备ID作为用户名

5.4.2 SSL/TLS加密

• 支持SSL/TLS连接
• 支持服务器证书验证
• 通过配置文件中的MqttUser[5]控制是否启用
• 证书内容通过MqttUserCa配置

5.5 配置流程说明

1. 连接MQTT服务器：使用配置的参数连接到MQTT服务器
2. 订阅下行主题：订阅设备的下行主题，用于接收配置命令
3. 发送配置请求：向设备的下行主题发送配置命令
4. 接收配置响应：接收设备的配置响应，确认配置是否成功
5. 重启设备：某些配置需要重启设备才能生效

5.6 配置示例

5.6.1 示例1：修改上报间隔

1. 发送配置命令：

   • 主题：/KeterRtu/MyDevice001/down/writeCon

   • 消息：

     {
     "ParType": "Getway",
     "varLst": {
       "report_interval": 10
     }
     }

2. 接收响应：

   • 主题：/KeterRtu/MyDevice001/up/writeCon

   • 消息：

     {
     "ParType": "Getway",
     "varLst": {
       "report_interval": 10,
       "flag": 1
     }
     }

3. 重启设备：

   • 主题：/KeterRtu/MyDevice001/down/writeCon
   • 消息：{"ParType": "RST"}

5.6.2 示例2：修改串口配置

1. 发送配置命令：

   • 主题：/KeterRtu/MyDevice001/down/writeCon

   • 消息：

     {
     "ParType": "Com",
     "varLst": {
       "polling_interval": 5,
       "response_timeout": 1000,
       "delay_between_poll": 300,
       "485_uart": ["115200", 8, "None", 1]
     }
     }

2. 接收响应：

   • 主题：/KeterRtu/MyDevice001/up/writeCon

   • 消息：

     {
     "ParType": "Com",
     "varLst": {
       "polling_interval": 5,
       "response_timeout": 1000,
       "delay_between_poll": 300,
       "485_uart": ["115200", 8, "None", 1],
       "flag": 1
     }
     }

3. 重启设备：设备会自动重启以应用新配置

5. MQTT数据获取与应用集成指南

5.1 数据主题结构

5.1.1 上行数据主题

• 设备注册：/KeterRtu/{设备ID}/up/reg
• 定时数据：/KeterRtu/{设备ID}/up/data
• 变化上报：/KeterRtu/{设备ID}/up/change
• 变量读取响应：/KeterRtu/{设备ID}/up/readVar
• 错误信息：/KeterRtu/{设备ID}/up/Err

5.2 消息Payload格式解析

5.2.1 设备注册信息（reg）

{
  "csq": 25, // 信号强度
  "imei": "861234567890123", // 设备IMEI
  "iccid": "89860012345678901234", // SIM卡ICCID
  "ver": "1.0.1" // 固件版本
} 

5.2.2 定时/变化上报（data/change）

{
  "devSn": "MyDevice001", // 设备ID
  "comm_s": 0, // 通信状态（0：成功，12：超时，其他：错误码）
  "varLst": {
    "M01": "123.4", // 变量编码和值
    "D01": "1",
    "D01_0": "0"
  }
}

5.2.3 变量读取响应（readVar）

{
  "devSn": "MyDevice001", // 设备ID
  "comm_s": 0, // 通信状态
  "varLst": {
    "M01": "123.4", // 变量编码和值
    "D01": "1",
    "D01_0": "0"
  }
}

5.3 数据更新频率

• 定时上报：根据配置的report_interval参数，默认为5分钟
• 变化上报：当变量值发生变化时上报
• 手动读取：通过发送readVar命令主动读取

5.5 异常处理机制

5.5.1 通信状态码comm_s

5.6 应用集成实现指南

5.6.1 MQTT客户端连接

import paho.mqtt.client as mqtt

# 连接参数
broker = "mqtt.example.com"
port = 1883
username = "user1"
password = "password1"
client_id = "app_client"

# 回调函数
def on_connect(client, userdata, flags, rc):
    print(f"Connected with result code {rc}")
    # 订阅设备的上行主题
    client.subscribe("/KeterRtu/MyDevice001/up/#")

def on_message(client, userdata, msg):
    print(f"Topic: {msg.topic}")
    print(f"Payload: {msg.payload.decode()}")
    # 处理接收到的消息

# 创建客户端
client = mqtt.Client(client_id)
client.username_pw_set(username, password)
client.on_connect = on_connect
client.on_message = on_message

# 连接服务器
client.connect(broker, port, 60)
client.loop_forever()

5.6.2 数据解析与处理

import json

# 变量描述表缓存（实际应用中应从设备获取）
var_descriptions = {}

# 从设备获取变量描述表
def fetch_var_descriptions():
    message = {"ParType": "DevDesp"}
    client.publish("/KeterRtu/MyDevice001/down/readCon", json.dumps(message))

# 解析变量描述表
def parse_var_descriptions(payload):
    global var_descriptions
    try:
        data = json.loads(payload)
        if data.get("ParType") == "DevDesp" and "Device" in data:
            device_data = data.get("Device", {})
            # 合并实时状态和远程控制的变量
            all_vars = {}

            # 处理实时状态变量
            real_time_vars = device_data.get("实时状态", {})
            for var_name, var_config in real_time_vars.items():
                all_vars[var_name] = var_config

            # 处理远程控制变量
            remote_control_vars = device_data.get("远程控制", {})
            for var_name, var_config in remote_control_vars.items():
                all_vars[var_name] = var_config

            var_descriptions = all_vars
            print(f"Loaded {len(var_descriptions)} variables from device")
    except Exception as e:
        print(f"Error parsing var descriptions: {e}")

def process_data_message(payload):
    """处理数据上报消息"""
    try:
        data = json.loads(payload)
        device_id = data.get("devSn")
        comm_status = data.get("comm_s")
        var_list = data.get("varLst", {})

        print(f"Device: {device_id}")
        print(f"Communication status: {comm_status}")
        print("Variables:")

        parsed_data = {}
        for var_name, var_value in var_list.items():
            if var_name in var_descriptions:
                var_config = var_descriptions[var_name]
                var_desc = var_config.get("Desp", var_name)
                var_unit = var_config.get("Unit", "")
                var_is_write = var_config.get("IsWrite", 0)
                var_remark = var_config.get("Remark", "")

                parsed_data[var_name] = {
                    "value": var_value,
                    "desc": var_desc,
                    "unit": var_unit,
                    "is_write": var_is_write,
                    "remark": var_remark
                }
                print(f"  {var_desc}: {var_value}{var_unit}")
                if var_remark:
                    print(f"    Remark: {var_remark}")
            else:
                parsed_data[var_name] = {
                    "value": var_value,
                    "desc": var_name,
                    "unit": ""
                }
                print(f"  {var_name}: {var_value} (unknown variable)")

        return parsed_data

    except json.JSONDecodeError as e:
        print(f"JSON decode error: {e}")
        return None

def on_message(client, userdata, msg):
    """消息回调函数"""
    payload = msg.payload.decode()
    topic = msg.topic

    if "/up/data" in topic:
        process_data_message(payload)
    elif "/up/readCon" in topic and "DevDesp" in payload:
        parse_var_descriptions(payload)
    elif "/up/reg" in topic:
        # 处理注册消息
        pass
    elif "/up/readVar" in topic:
        # 处理变量读取响应
        pass
    elif "/up/Err" in topic:
        # 处理错误消息
        pass

5.6.3 主动读取变量

def read_all_variables():
    """读取所有变量"""
    message = {
        "devSn": "MyDevice001",
        "varID": "all"
    }
    client.publish("/KeterRtu/MyDevice001/down/readVar", json.dumps(message))

# 调用函数读取变量
read_all_variables()

5.6.4 写入变量

def write_variable(var_name, value):
    """写入变量"""
    # 检查变量是否存在且可写
    if var_name in var_descriptions and var_descriptions[var_name].get("IsWrite") == 1:
        message = {
            "devSn": "MyDevice001",
            "varLst": {
                var_name: value
            }
        }
        client.publish("/KeterRtu/MyDevice001/down/writeVar", json.dumps(message))
        print(f"Wrote {var_name}: {value}")
    else:
        print(f"Variable {var_name} not found or not writable")

# 调用函数写入变量
write_variable("M110", "25")  # 设置制冷温度为25℃
write_variable("On/Off", "1")  # 设置为制冷模式

5.7 集成示例

5.7.1 完整数据监控应用

import paho.mqtt.client as mqtt
import json
import time

# 连接参数
broker = "mqtt.example.com"
port = 1883
username = "user1"
password = "password1"
client_id = "monitor_app"
device_id = "MyDevice001"

# 变量描述表（实际应用中应从设备获取）
var_descriptions = {}
latest_data = {}

def fetch_var_descriptions():
    """从设备获取变量描述表"""
    message = {"ParType": "DevDesp"}
    client.publish(f"/KeterRtu/{device_id}/down/readCon", json.dumps(message))

def parse_var_descriptions(payload):
    """解析变量描述表"""
    global var_descriptions
    try:
        data = json.loads(payload)
        if data.get("ParType") == "DevDesp" and "Device" in data:
            device_data = data.get("Device", {})
            # 合并实时状态和远程控制的变量
            all_vars = {}

            # 处理实时状态变量
            real_time_vars = device_data.get("实时状态", {})
            for var_name, var_config in real_time_vars.items():
                all_vars[var_name] = var_config

            # 处理远程控制变量
            remote_control_vars = device_data.get("远程控制", {})
            for var_name, var_config in remote_control_vars.items():
                all_vars[var_name] = var_config

            var_descriptions = all_vars
            print(f"Loaded {len(var_descriptions)} variables from device")
    except Exception as e:
        print(f"Error parsing var descriptions: {e}")

def process_data_message(payload):
    """处理数据上报消息"""
    global latest_data
    try:
        data = json.loads(payload)
        var_list = data.get("varLst", {})

        parsed_data = {}
        for var_name, var_value in var_list.items():
            if var_name in var_descriptions:
                var_config = var_descriptions[var_name]
                var_desc = var_config.get("Desp", var_name)
                var_unit = var_config.get("Unit", "")
                var_remark = var_config.get("Remark", "")

                parsed_data[var_name] = {
                    "value": var_value,
                    "desc": var_desc,
                    "unit": var_unit,
                    "remark": var_remark
                }
            else:
                parsed_data[var_name] = {
                    "value": var_value,
                    "desc": var_name,
                    "unit": ""
                }

        latest_data = parsed_data

        print("\n=== New Data Received ===")
        print(f"Device: {data.get('devSn')}")
        print(f"Timestamp: {time.strftime('%Y-%m-%d %H:%M:%S')}")
        print(f"Communication status: {data.get('comm_s')}")
        print("Variables:")
        for var, info in parsed_data.items():
            print(f"  {info['desc']}: {info['value']}{info['unit']}")
            if info.get('remark'):
                print(f"    Remark: {info['remark']}")

        return parsed_data

    except json.JSONDecodeError as e:
        print(f"Error parsing data: {e}")
        return None

# 回调函数
def on_connect(client, userdata, flags, rc):
    print(f"Connected with result code {rc}")
    # 订阅设备的所有上行主题
    client.subscribe(f"/KeterRtu/{device_id}/up/#")
    # 获取变量描述表
    fetch_var_descriptions()

def on_message(client, userdata, msg):
    payload = msg.payload.decode()
    topic = msg.topic

    if "/up/data" in topic:
        process_data_message(payload)
    elif "/up/readCon" in topic and "DevDesp" in payload:
        parse_var_descriptions(payload)
    elif "/up/reg" in topic:
        print(f"\n=== Device Registered ===")
        print(payload)

# 创建客户端
client = mqtt.Client(client_id)
client.username_pw_set(username, password)
client.on_connect = on_connect
client.on_message = on_message

# 连接服务器
client.connect(broker, port, 60)

# 启动监控
print("Starting data monitoring...")
print("Press Ctrl+C to exit")
try:
    client.loop_forever()
except KeyboardInterrupt:
    print("\nExiting...")
    client.disconnect()

6. 开发注意事项

6.1 网络连接

1. 网络稳定性：确保设备所在区域有良好的4G网络覆盖
2. SIM卡管理：使用有效的SIM卡，确保不欠费
3. 网络切换：设备会自动切换网络，无需手动干预
4. 信号强度：通过信号强度指示灯或AT+GNET指令查看信号强度

6.2 数据安全

1. MQTT认证：使用强密码，避免使用默认密码
2. SSL/TLS：生产环境中建议启用SSL/TLS加密
3. 数据传输：敏感数据建议在应用层进行加密
4. 访问控制：限制MQTT服务器的访问权限

6.3 性能优化

1. 上报频率：根据实际需求设置合理的上报频率，避免过于频繁的数据上报
2. 消息大小：控制消息负载大小，避免超过MQTT broker的限制
3. 连接管理：合理管理MQTT连接，避免频繁重连
4. 资源使用：注意设备的内存和CPU使用情况

6.5 变量描述表使用注意事项

1. 获取时机：应用启动时应立即获取变量描述表
2. 缓存管理：将变量描述表缓存到本地，避免频繁获取
3. 错误处理：处理变量描述表获取失败的情况
4. 映射关系：建立变量编码与配置的明确映射关系
5. 数据一致性：确保所有数据处理都基于同一版本的变量描述表
6. 字段大小写：注意字段名称为大写（Desp、Unit、IsWrite、Remark）

7. 附录

7.1 设备状态指示灯说明

7.2 常见错误码

版本：1.0.2 日期：2026-03-06

深圳市科特时控技术有限公司