2. Modbus RTU从站数据采集
任务目标
通过RS485接口连接Modbus RTU从站设备，周期读取保持寄存器数据并完成解析。本案例以标准Modbus功能码03（读保持寄存器）为核心，实现从站地址、寄存器起始地址及数量的可配置读取，并将原始寄存器值转换为有符号/无符号整数。
环境依赖
- Python 3.8+
- pyserial库：
pip install pyserial - modbus-tk库：
pip install modbus-tk - 硬件：RS485转USB适配器，Modbus RTU从站设备（或使用从站模拟器）
完整实现代码
import serial
import modbus_tk.defines as cst
import modbus_tk.modbus_rtu as modbus_rtu
import time
import struct
class ModbusRTUReader:
def __init__(self, port, baudrate=9600, bytesize=8, parity='N', stopbits=1, timeout=1):
"""初始化RS485串口连接"""
self.serial = serial.Serial(
port=port,
baudrate=baudrate,
bytesize=bytesize,
parity=parity,
stopbits=stopbits,
timeout=timeout
)
self.master = modbus_rtu.RtuMaster(self.serial)
self.master.set_timeout(timeout)
self.master.set_verbose(False)
def read_holding_registers(self, slave_id, start_addr, quantity):
"""
读取保持寄存器（功能码03）
:param slave_id: 从站地址 (1-247)
:param start_addr: 起始寄存器地址 (0-based)
:param quantity: 读取寄存器数量 (1-125)
:return: 寄存器原始值列表
"""
if not (1 <= slave_id <= 247):
raise ValueError("从站地址必须在1-247之间")
if not (1 <= quantity <= 125):
raise ValueError("读取数量必须在1-125之间")
try:
result = self.master.execute(
slave_id,
cst.READ_HOLDING_REGISTERS,
start_addr,
quantity
)
return list(result)
except Exception as e:
raise RuntimeError(f"Modbus读取失败: {e}")
def parse_registers(self, raw_values, data_format='uint16'):
"""
解析寄存器原始值
:param raw_values: 寄存器原始值列表
:param data_format: 数据格式
'uint16' - 无符号16位整数
'int16' - 有符号16位整数
'uint32' - 无符号32位整数（两个寄存器组合，大端序）
'int32' - 有符号32位整数（两个寄存器组合，大端序）
'float32' - 32位浮点数（两个寄存器组合，大端序）
:return: 解析后的数值列表
"""
parsed = []
i = 0
while i < len(raw_values):
if data_format in ('uint16', 'int16'):
val = raw_values[i]
if data_format == 'int16' and val > 0x7FFF:
val -= 0x10000
parsed.append(val)
i += 1
elif data_format in ('uint32', 'int32', 'float32'):
if i + 1 >= len(raw_values):
raise ValueError("寄存器数量不足，无法组合32位数据")
# 大端序组合两个寄存器
combined = (raw_values[i] << 16) | raw_values[i+1]
if data_format == 'uint32':
parsed.append(combined)
elif data_format == 'int32':
if combined > 0x7FFFFFFF:
combined -= 0x100000000
parsed.append(combined)
else: # float32
packed = struct.pack('>I', combined)
parsed.append(struct.unpack('>f', packed)[0])
i += 2
else:
raise ValueError(f"不支持的数据格式: {data_format}")
return parsed
def close(self):
"""关闭串口连接"""
if self.serial and self.serial.is_open:
self.serial.close()
def main():
# 配置参数 - 根据实际设备修改
PORT = 'COM3' # Windows串口号，Linux下为'/dev/ttyUSB0'
SLAVE_ID = 1 # 从站地址
START_ADDR = 0 # 起始寄存器地址
QUANTITY = 4 # 读取寄存器数量
DATA_FORMAT = 'int16' # 数据解析格式
READ_INTERVAL = 2 # 读取间隔（秒）
reader = ModbusRTUReader(PORT)
try:
print(f"开始采集: 从站{SLAVE_ID}, 起始地址{START_ADDR}, 数量{QUANTITY}, 格式{DATA_FORMAT}")
print("按Ctrl+C停止采集\n")
while True:
try:
raw_values = reader.read_holding_registers(SLAVE_ID, START_ADDR, QUANTITY)
parsed_values = reader.parse_registers(raw_values, DATA_FORMAT)
print(f"[{time.strftime('%H:%M:%S')}] 原始值: {raw_values}")
print(f"解析值: {parsed_values}\n")
except Exception as e:
print(f"采集错误: {e}")
time.sleep(READ_INTERVAL)
except KeyboardInterrupt:
print("\n采集已停止")
finally:
reader.close()
if __name__ == "__main__":
main()运行与验证
- 准备从站设备：将RS485从站设备连接到USB转485适配器，确认设备地址和寄存器映射。若无实体设备，可使用Modbus Slave模拟软件（如Modbus Poll的从站模式）进行测试。
- 确认串口参数：修改代码中
PORT为实际串口号，并确保波特率、校验位等参数与从站设备一致。 - 执行采集：运行脚本，观察终端输出。
通过条件：程序能持续输出寄存器原始值，且解析值与从站设备中预设的数据一致。例如从站地址1的寄存器0-3分别存储10、20、30、40，则预期输出原始值[10, 20, 30, 40]，解析值（int16格式）同样为[10, 20, 30, 40]。
常见问题
- 串口打开失败：检查串口号是否正确，或是否有其他程序占用该串口。
- 超时无响应：确认从站地址、波特率、校验位设置正确；检查RS485的A/B线是否接反，以及终端电阻是否匹配。
- CRC校验错误：通常由通信参数不一致或线路干扰导致，检查接线质量和参数配置。