4. Modbus TCP客户端开发：Python实现连接、读写保持寄存器

好，咱们进入实战环节。

前面讲了Modbus TCP的协议原理，说白了就是主站问、从站答。那在Python里怎么实现？我个人的习惯是用 pymodbus 这个库，成熟、稳定，坑也少。今天咱们就手把手把客户端搭起来，连上设备，读写保持寄存器。

4.1 环境准备与库安装

先装库。别用pip直接装最新版，我踩过坑——有些新版本改了API，老代码直接跑不起来。

pip install pymodbus==3.6.8

嗯，3.6.8这个版本我用了两年，没出过问题。你想想看，生产环境稳定比什么都重要。

4.2 建立连接——三步走

连接Modbus TCP设备，其实就三步：

  1. 创建客户端对象
  2. 指定IP和端口（默认502）
  3. 调用connect()方法

看代码：

from pymodbus.client import ModbusTcpClient

# 创建客户端
client = ModbusTcpClient('192.168.1.100', port=502)

# 建立连接
if client.connect():
    print("连接成功！")
else:
    print("连接失败，检查网络或设备地址。")

这里有个细节——connect() 返回的是布尔值。我在项目里见过有人不判断返回值就直接读写，结果程序卡死。记住：连接成功再操作，这是铁律。

4.3 读取保持寄存器

保持寄存器是Modbus里最常用的数据区。储能系统的SOC、电压、电流，一般都放在这里。

读取用 read_holding_registers() 方法：

# 读取从地址0开始的10个寄存器
result = client.read_holding_registers(address=0, count=10, unit=1)

if result.isError():
    print("读取失败：", result)
else:
    print("寄存器值：", result.registers)

参数说明：

  • address：起始地址，从0开始
  • count：读取数量
  • unit：从站地址，默认1

我曾经遇到过一个坑——某厂家设备地址从1开始，但Modbus协议里地址是从0算的。结果读出来的数据全不对。后来我养成了习惯：先读一个寄存器看看值是否合理，再批量读。

4.4 写入保持寄存器

写入分两种：写单个和写多个。

4.4.1 写单个寄存器

# 将地址0的寄存器写入100
result = client.write_register(address=0, value=100, unit=1)

if result.isError():
    print("写入失败")
else:
    print("写入成功")

4.4.2 写多个寄存器

# 从地址0开始，连续写入3个值
values = [100, 200, 300]
result = client.write_registers(address=0, values=values, unit=1)

这里要注意：写入操作一定要确认返回值。我见过有人写寄存器不检查结果，设备没响应还以为写成功了，最后数据对不上，排查了半天。

4.5 完整示例——读取储能柜数据

咱们来个完整的。假设一个储能柜，保持寄存器地址分配如下：

地址含义数据类型
0SOC（%）uint16
1总电压（V）uint16
2总电流（A）int16
3-4累计电量（kWh）uint32（高16位+低16位）
from pymodbus.client import ModbusTcpClient

client = ModbusTcpClient('192.168.1.100', port=502)

if client.connect():
    # 读取SOC、电压、电流
    result = client.read_holding_registers(address=0, count=3, unit=1)
    if not result.isError():
        soc = result.registers[0]
        voltage = result.registers[1] / 10.0  # 假设精度0.1V
        current = result.registers[2] if result.registers[2] < 32768 else result.registers[2] - 65536
        print(f"SOC: {soc}%, 电压: {voltage}V, 电流: {current}A")
    
    # 读取累计电量（32位）
    result = client.read_holding_registers(address=3, count=2, unit=1)
    if not result.isError():
        energy = (result.registers[0] << 16) | result.registers[1]
        print(f"累计电量: {energy} kWh")
    
    client.close()
else:
    print("连接失败")

小技巧：int16转负数的处理，用 struct.unpack('<h', bytes) 更规范。但我习惯用三目运算，简单直接，现场调试时一眼能看懂。

4.6 异常处理与重连机制

工业现场网络不稳定，设备偶尔重启。代码里一定要加重连。

import time

def read_with_retry(client, address, count, unit=1, retries=3):
    for i in range(retries):
        try:
            result = client.read_holding_registers(address, count, unit)
            if not result.isError():
                return result.registers
        except Exception as e:
            print(f"第{i+1}次读取失败：{e}")
            time.sleep(1)
            # 尝试重连
            client.connect()
    return None

注意：重连间隔不要太短。我曾经设了0.1秒重连，设备还没恢复就疯狂发请求，把从站CPU占满了。建议至少1秒以上。

4.7 断开连接

用完记得关。虽然Python有垃圾回收，但TCP连接不释放，端口会被占满。

client.close()

嗯，就这么简单。但很多人会忘。我习惯在 finally 块里关连接，确保万无一失。

4.8 本章核心逻辑图

下面这张图，把整个流程串起来了：

Modbus TCP客户端开发流程 1. 创建客户端 2. 建立连接 3. 读写寄存器 read_holding_registers write_register(s) 4. 关闭连接

说白了，就是这四步。但实际项目中，第3步和第4步之间往往要加循环——不断读取、判断、写入。嗯，这个后面章节会展开讲。

核心要点：

  • 连接前先判断返回值
  • 读写后检查isError()
  • 异常时重连，间隔至少1秒
  • 用完关闭连接

好了，这一章就到这儿。代码不多，但都是实战里反复验证过的。你拿回去改改IP和寄存器地址，就能直接用。

公众号:蓝海资料掘金营，微信deep3321

← 上一章 📖 返回目录 下一章 →