📚 课程目录第2章 Modbus TCP 客户端通信

2. Modbus TCP 客户端通信

任务目标

编写一个 Modbus TCP 客户端程序，实现以下功能：

连接指定的远程 Modbus TCP 服务器（IP + 端口）
读取保持寄存器（Holding Register）数据
写入单个保持寄存器数据
验证读写结果是否正确

环境依赖

本案例使用 Python 3.8+ 和 pymodbus 库（v3.x）。安装命令：

pip install pymodbus==3.6.9

完整代码

from pymodbus.client import ModbusTcpClient
import time

# 配置参数
SERVER_HOST = "192.168.1.100"   # 远程服务器地址
SERVER_PORT = 502               # Modbus TCP 默认端口
UNIT_ID = 1                     # 从站地址（通常为 1）
TIMEOUT = 5                     # 超时时间（秒）

def read_holding_register(client, address, count=1):
    """读取保持寄存器"""
    result = client.read_holding_registers(address, count, slave=UNIT_ID)
    if result.isError():
        raise RuntimeError(f"读取失败: {result}")
    return result.registers

def write_holding_register(client, address, value):
    """写入单个保持寄存器"""
    result = client.write_register(address, value, slave=UNIT_ID)
    if result.isError():
        raise RuntimeError(f"写入失败: {result}")
    return True

def main():
    # 建立连接
    client = ModbusTcpClient(SERVER_HOST, port=SERVER_PORT, timeout=TIMEOUT)
    connected = client.connect()
    if not connected:
        print(f"连接失败: {SERVER_HOST}:{SERVER_PORT}")
        return

    print(f"连接成功: {SERVER_HOST}:{SERVER_PORT}")

    try:
        # 1. 读取寄存器地址 0 的当前值
        reg_addr = 0
        original_value = read_holding_register(client, reg_addr)[0]
        print(f"寄存器 {reg_addr} 原始值: {original_value}")

        # 2. 写入新值（原值 + 100，确保变化可观察）
        new_value = original_value + 100
        write_holding_register(client, reg_addr, new_value)
        print(f"已写入寄存器 {reg_addr} = {new_value}")

        # 3. 回读验证
        time.sleep(0.2)  # 等待写入生效
        readback_value = read_holding_register(client, reg_addr)[0]
        print(f"回读寄存器 {reg_addr} 值: {readback_value}")

        # 4. 判定结果
        if readback_value == new_value:
            print("验证通过: 写入值与回读值一致")
        else:
            print(f"验证失败: 期望 {new_value}, 实际 {readback_value}")

    except Exception as e:
        print(f"操作异常: {e}")
    finally:
        client.close()
        print("连接已关闭")

if __name__ == "__main__":
    main()

运行方式

将代码保存为 modbus_client.py
修改 SERVER_HOST 为实际 Modbus TCP 服务器地址
执行命令：python modbus_client.py

验证方法

程序运行后观察输出：

若显示 连接成功，说明 TCP 连接建立正常
若显示 验证通过: 写入值与回读值一致，说明读写寄存器功能完整可用
若连接失败，检查服务器 IP、端口是否可达，以及防火墙是否放行 502 端口

无服务器时的本地模拟验证

若当前没有现成的 Modbus TCP 服务器，可用以下最小模拟器在本地验证客户端逻辑：

# 保存为 mock_server.py，先启动此文件再运行客户端
from pymodbus.server import StartTcpServer
from pymodbus.datastore import ModbusSlaveContext, ModbusServerContext
from pymodbus.datastore import ModbusSequentialDataBlock

# 初始化 100 个保持寄存器，初始值为 0
store = ModbusSlaveContext(
    di=ModbusSequentialDataBlock(0, [0]*100),
    co=ModbusSequentialDataBlock(0, [0]*100),
    hr=ModbusSequentialDataBlock(0, [0]*100),
    ir=ModbusSequentialDataBlock(0, [0]*100)
)
context = ModbusServerContext(slaves=store, single=True)

if __name__ == "__main__":
    print("模拟服务器启动: 0.0.0.0:502")
    StartTcpServer(context=context, address=("0.0.0.0", 502))

验证步骤：先运行 python mock_server.py，再运行 python modbus_client.py，客户端将连接本机 502 端口完成读写验证。

常见问题

连接超时：确认服务器 IP 可达、端口开放，且服务器程序已启动
从站地址不匹配：若服务器配置了非 1 的从站地址，修改 UNIT_ID 为对应值
寄存器地址越界：模拟器初始化了 100 个寄存器（地址 0-99），读写地址需在此范围内