4. Modbus TCP实践:使用Python的pymodbus库实现数据读写

好,咱们进入动手环节。

前面聊了那么多协议原理,说实话,不亲手写几行代码,总觉得心里没底。我个人习惯是,学任何工业协议,先跑通一个最简单的读写例子。Modbus TCP 在这方面特别友好——它简单、直接,而且 Python 的 pymodbus 库封装得相当成熟。

4.1 环境准备:装好 pymodbus

先别急着写代码,把工具备齐。

我建议你用 Python 3.8 以上版本。pymodbus 目前最新稳定版是 3.x 系列,跟 2.x 相比,API 变化挺大。如果你在网上搜到老教程,注意看版本号。

pip install pymodbus

就这么一行。装完之后,在 Python 里试一下:

import pymodbus
print(pymodbus.__version__)

能打印出版本号,说明环境没问题。

嗯,这里要注意:如果你用的是 Linux 系统,可能还需要装一个 pyserial,不过咱们只做 TCP 通信,用不到串口。

4.2 第一个程序:读取保持寄存器

Modbus 里最常用的操作就是读保持寄存器。说白了,就是从设备里拿数据。

我记得第一次在现场调试时,PLC 死活连不上,折腾了半天才发现是 IP 地址写错了。所以,写代码前,先确认好对方的 IP 和端口号。Modbus TCP 默认端口是 502。

from pymodbus.client import ModbusTcpClient

# 连接设备
client = ModbusTcpClient('192.168.1.100', port=502)
client.connect()

# 读取保持寄存器,从地址0开始,读10个
result = client.read_holding_registers(0, 10, slave=1)

if not result.isError():
    print("读取成功,数据:", result.registers)
else:
    print("读取失败:", result)

client.close()

这段代码做了三件事:连接、读取、关闭。你想想看,是不是比想象中简单?

这里有个坑:slave=1 这个参数。在 Modbus TCP 里,理论上不需要从站地址,但很多设备还是要求你填。我曾经遇到过一台变频器,不填 slave 参数就报错,填了反而正常。所以,我建议你每次都加上。

4.3 写数据:写单个和写多个寄存器

读数据只是第一步,工业现场更多时候是要写数据。比如控制电机启停、设定温度值。

pymodbus 提供了两种写操作:

  • 写单个寄存器write_register()
  • 写多个寄存器write_registers()

看个例子:

# 写单个寄存器(地址0,值100)
result = client.write_register(0, 100, slave=1)

# 写多个寄存器(从地址10开始,写入[200, 300, 400])
result = client.write_registers(10, [200, 300, 400], slave=1)

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

我个人习惯是,写完之后马上读回来验证一下。别问我为什么,吃过亏的人都知道。

⚠️ 注意: 写操作是有风险的。尤其是写单个线圈或寄存器时,一旦地址写错,可能触发设备误动作。我在项目里见过有人把「停止」写成了「启动」,结果传送带突然跑起来,差点出事故。所以,写操作前一定要再三确认地址和数据类型。

4.4 实战案例:读取温控仪表的实时数据

光讲理论没意思,咱们来一个真实场景。

假设现场有一台温控仪表,Modbus TCP 通信。它的寄存器定义如下:

寄存器地址 含义 数据类型
0 当前温度 16位无符号整数(单位0.1℃)
1 设定温度 16位无符号整数(单位0.1℃)
2 运行状态 0=停止,1=运行

好,现在我们要读取当前温度,并判断设备是否在运行。

client = ModbusTcpClient('192.168.1.200', port=502)
client.connect()

# 读取3个寄存器
result = client.read_holding_registers(0, 3, slave=1)

if not result.isError():
    temp_raw = result.registers[0]   # 原始值,比如 255 表示 25.5℃
    set_temp_raw = result.registers[1]
    status = result.registers[2]

    # 转换成实际温度
    current_temp = temp_raw / 10.0
    set_temp = set_temp_raw / 10.0

    print(f"当前温度:{current_temp}℃")
    print(f"设定温度:{set_temp}℃")
    print(f"运行状态:{'运行中' if status == 1 else '已停止'}")

client.close()

你看,就这么几行代码,就能把现场设备的数据读回来。是不是很有成就感?

💡 小技巧: 实际项目中,我习惯把 Modbus 通信封装成一个类。把连接、读取、写入、异常处理都包进去。这样主程序里调用起来特别清爽,而且便于后期维护。

4.5 异常处理:别让程序崩了

工业现场的网络环境,说实话,没那么稳定。设备重启、网线松动、干扰……什么情况都可能发生。

所以,代码里一定要加异常处理。我见过太多人写 Modbus 程序时,连 try-except 都不加,一断网程序直接崩溃。

import time
from pymodbus.client import ModbusTcpClient
from pymodbus.exceptions import ModbusException

def read_temperature(ip, port=502, slave=1, retries=3):
    for attempt in range(retries):
        try:
            client = ModbusTcpClient(ip, port=port, timeout=3)
            client.connect()
            result = client.read_holding_registers(0, 1, slave=slave)
            client.close()

            if result.isError():
                raise ModbusException(f"读取错误:{result}")

            return result.registers[0] / 10.0

        except Exception as e:
            print(f"第{attempt+1}次尝试失败:{e}")
            time.sleep(1)

    return None  # 所有重试都失败

# 使用
temp = read_temperature('192.168.1.200')
if temp is not None:
    print(f"温度:{temp}℃")
else:
    print("读取失败,请检查网络和设备")

这段代码加了重试机制和超时控制。我曾经在一个项目里,就是因为没加超时,程序卡在 connect() 那里整整 30 秒,导致整个产线监控系统响应缓慢。加了 timeout=3 之后,问题立刻解决。

4.6 避坑指南:我踩过的几个坑

最后,分享几个我实际项目中遇到的坑,希望能帮你少走弯路。

  • 字节序问题:Modbus 默认是大端模式,但有些设备用的小端。如果你读出来的数据是乱码,先检查字节序。pymodbus 里可以用 word_order 参数调整。
  • 寄存器地址偏移:有些设备的文档里写的是「地址 40001」,对应到代码里其实是 0。因为 Modbus 协议里地址是从 0 开始的,而 40001 是 PLC 的映射地址。这个坑我至少见过三次。
  • 同时读写冲突:如果你在循环里频繁读写,记得加一点延时。有些设备处理不过来,会返回异常码 06(从站设备忙)。

好了,这一章的内容就到这里。代码不多,但都是实打实能用的。下一章咱们会聊更高级的话题——如何用 pymodbus 搭建一个 Modbus TCP 服务器,模拟设备来调试程序。