前言

工业现场90%以上的PLC、传感器、仪表还在跑Modbus RTU。搞工业物联网，Modbus是绕不过去的坎。本文从协议原理到代码实战，结合有人物联网USR-G780 DTU，彻底讲透。

---

一、Modbus RTU协议帧结构

┌─────────┬───────────┬──────────┬──────────┬───────┬───────┐ │ 地址码 │ 功能码 │ 数据区 │ CRC校验 │ │ │ │ 1 Byte │ 1 Byte │ N Bytes │ 2 Bytes │ │ │ └─────────┴───────────┴──────────┴──────────┴───────┴───────┘

地址码（Slave Address）

• 范围：1~247

• 每个从机唯一地址

• 0为广播地址（所有从机执行但不回复）

功能码（Function Code）

功能码	名称	操作对象
01	读线圈	开关量输出
02	读离散输入	开关量输入
03	读保持寄存器	模拟量（最常用）
04	读输入寄存器	模拟量输入
05	写单线圈	单路控制
06	写单寄存器	单参数设置
15	写多线圈	批量控制
16	写多寄存器	批量参数

CRC校验

Modbus RTU用CRC-16校验，多项式 0xA001。这保证了工业环境强干扰下的数据完整性。

def modbus_crc(data): """Modbus CRC-16校验""" crc = 0xFFFF for byte in data: crc ^= byte for _ in range(8): if crc & 0x0001: crc = (crc >> 1) ^ 0xA001 else: crc >>= 1 return crc.to_bytes(2, 'little')

---

二、实战一：Python读取Modbus温湿度传感器

""" 使用Python + RS485转USB读取Modbus RTU传感器数据 硬件：有人物联网USR-G780 DTU（透传模式）或USB转RS485模块 """ import serial import struct import time class ModbusRTUMaster: def __init__(self, port='COM3', baudrate=9600, timeout=1): self.ser = serial.Serial( port=port, baudrate=baudrate, bytesize=8, parity='N', stopbits=1, timeout=timeout ) def read_holding_registers(self, slave_id, start_addr, count): """功能码03：读保持寄存器""" # 构造请求帧 request = bytearray([ slave_id, # 地址码 0x03, # 功能码 (start_addr >> 8) & 0xFF, # 起始地址高字节 start_addr & 0xFF, # 起始地址低字节 (count >> 8) & 0xFF, # 寄存器数量高字节 count & 0xFF # 寄存器数量低字节 ]) # 追加CRC request += modbus_crc(request) # 发送 self.ser.write(request) time.sleep(0.1) # 接收响应 response = self.ser.read(5 + count * 2) if len(response) < 5: raise TimeoutError("从机无响应") # 验证CRC if modbus_crc(response[:-2]) != response[-2:]: raise ValueError("CRC校验失败") # 解析数据 data_bytes = response[3:-2] registers = [] for i in range(0, len(data_bytes), 2): reg = (data_bytes[i] << 8) | data_bytes[i+1] registers.append(reg) return registers def write_single_register(self, slave_id, addr, value): """功能码06：写单个寄存器""" request = bytearray([ slave_id, 0x06, (addr >> 8) & 0xFF, addr & 0xFF, (value >> 8) & 0xFF, value & 0xFF ]) request += modbus_crc(request) self.ser.write(request) time.sleep(0.1) response = self.ser.read(8) return len(response) == 8 def registers_to_float(high_reg, low_reg): """两个16位寄存器 → 32位浮点数（大端序）""" raw = (high_reg << 16) | low_reg return struct.unpack('>f', struct.pack('>I', raw))[0] # ===== 使用示例 ===== if __name__ == '__main__': master = ModbusRTUMaster(port='COM3', baudrate=9600) # 读取从机地址1的保持寄存器，从0x0000开始读4个 regs = master.read_holding_registers(slave_id=1, start_addr=0, count=4) # 解析数据（假设前两个存温度，后两个存湿度） temperature = registers_to_float(regs[0], regs[1]) humidity = registers_to_float(regs[2], regs[3]) print(f"温度: {temperature:.1f}°C") print(f"湿度: {humidity:.1f}%RH")

---

三、实战二：Modbus RTU → MQTT JSON 转换（DTU场景）

当你用4G DTU（如有人USR-G780）把Modbus数据发到云端时，需要做格式转换：

""" modbus_to_mqtt.py 运行在边缘网关或服务器上，将Modbus RTU数据转换为MQTT JSON """ import json import paho.mqtt.client as mqtt from datetime import datetime # Modbus点位映射表 POINT_MAP = { "温度传感器_01": { "slave_id": 1, "start_addr": 0, "count": 4, "fields": [ {"name": "temperature", "reg_index": 0, "type": "float32"}, {"name": "humidity", "reg_index": 2, "type": "float32"} ] }, "电表_01": { "slave_id": 2, "start_addr": 100, "count": 6, "fields": [ {"name": "voltage", "reg_index": 0, "type": "float32"}, {"name": "current", "reg_index": 2, "type": "float32"}, {"name": "power", "reg_index": 4, "type": "float32"} ] } } def collect_and_publish(master, mqtt_client): """采集所有设备数据并发布MQTT""" for device_name, config in POINT_MAP.items(): try: regs = master.read_holding_registers( slave_id=config['slave_id'], start_addr=config['start_addr'], count=config['count'] ) # 解析字段 payload = { "device": device_name, "timestamp": datetime.now().isoformat(), "data": {} } for field in config['fields']: if field['type'] == 'float32': value = registers_to_float( regs[field['reg_index']], regs[field['reg_index'] + 1] ) else: value = regs[field['reg_index']] payload['data'][field['name']] = round(value, 2) # 发布MQTT topic = f"/iot/{device_name}/data" mqtt_client.publish(topic, json.dumps(payload), qos=1) print(f"[OK] {device_name}: {payload['data']}") except Exception as e: print(f"[ERR] {device_name}: {e}") # MQTT连接 mqttc = mqtt.Client(client_id="modbus_gateway") mqttc.connect("broker.emqx.io", 1883, 60) mqttc.loop_start()

---

四、常见排查方法

问题1：读不到数据

# 用最基础的指令测试从机是否存在 test_frame = bytearray([0x01, 0x03, 0x00, 0x00, 0x00, 0x01]) test_frame += modbus_crc(test_frame) ser.write(test_frame) response = ser.read(7) print(response.hex()) # 有回复说明通信正常，没回复检查接线/波特率/地址

问题2：A/B线接反

• 万用表测量：A线通常3-5V，B线通常0-1V

• 交换A/B线试试是最快的方法

问题3：波特率不对

• 常见波特率：9600、19200、38400、115200

• 用串口助手逐个试，直到读到不乱码的数据

问题4：32位浮点数数值异常

• 注意字节序：大端/小端、ABCD/CDAB等排列

• 用已知值反推（比如发个25.5，看寄存器原始值）

---

五、有人物联网DTU的Modbus优化

有人物联网USR-G780等DTU内置了Modbus自动采集功能： 轮询间隔可配**：1-3600秒

• 批量读取：一次配置读取多个从机的多个寄存器

• 自动上报：采集后自动转MQTT JSON上报

• 无需写代码：通过配置工具即可完成