3. 网关Modbus TCP客户端采集实现

任务目标

实现一个 Modbus TCP 客户端，连接远程 Modbus TCP 服务器，周期性读取线圈（Coils，功能码 01）与输入寄存器（Input Registers，功能码 04），并打印结果。使用 Python 的 pymodbus 库完成，无需自建服务器即可对接任意标准 Modbus TCP 从站。

环境与依赖

• Python 3.8+
• pip install pymodbus==3.6.9（3.x 版本 API 与 2.x 不兼容，本案例基于 3.x）
• 一个可访问的 Modbus TCP 服务器地址与端口（默认 502）

实现思路

客户端按固定周期执行一轮采集：先读线圈，再读输入寄存器，两者都成功则打印；任一失败则记录错误并继续下一轮，避免单次异常终止整个采集循环。使用 ModbusTcpClient 建立连接，read_coils 与 read_input_registers 分别对应功能码 01 和 04。

完整代码

import time
from pymodbus.client import ModbusTcpClient

HOST = "127.0.0.1"      # 替换为实际 Modbus TCP 服务器地址
PORT = 502
UNIT_ID = 1             # 从站单元号
COIL_ADDR = 0           # 线圈起始地址
COIL_COUNT = 8          # 读取线圈数量
IREG_ADDR = 0           # 输入寄存器起始地址
IREG_COUNT = 4          # 读取输入寄存器数量
INTERVAL = 2.0          # 采集周期（秒）

def poll_once(client):
    """执行一轮采集，返回 (coils, iregs)，任一失败返回 (None, None)。"""
    coils = client.read_coils(COIL_ADDR, count=COIL_COUNT, slave=UNIT_ID)
    if coils.isError():
        print(f"[ERROR] 读线圈失败: {coils}")
        return None, None

    iregs = client.read_input_registers(IREG_ADDR, count=IREG_COUNT, slave=UNIT_ID)
    if iregs.isError():
        print(f"[ERROR] 读输入寄存器失败: {iregs}")
        return None, None

    return coils.bits[:COIL_COUNT], iregs.registers

def main():
    client = ModbusTcpClient(HOST, port=PORT)
    if not client.connect():
        print(f"[FATAL] 无法连接 {HOST}:{PORT}")
        return

    print(f"已连接 {HOST}:{PORT}，开始周期采集（Ctrl+C 退出）")
    try:
        while True:
            coils, iregs = poll_once(client)
            if coils is not None:
                print(f"线圈[{COIL_ADDR}:{COIL_ADDR+COIL_COUNT}] = {coils}")
                print(f"输入寄存器[{IREG_ADDR}:{IREG_ADDR+IREG_COUNT}] = {iregs}")
            time.sleep(INTERVAL)
    except KeyboardInterrupt:
        print("已停止采集")
    finally:
        client.close()

if __name__ == "__main__":
    main()

运行方式

pip install pymodbus==3.6.9
python modbus_client.py

若本地没有 Modbus TCP 服务器，可用 pymodbus 自带的示例服务器快速验证：

python -m pymodbus.server --modbus-server tcp --host 127.0.0.1 --port 502

验证方法与通过条件

• 启动客户端后，终端应打印“已连接 …，开始周期采集”，随后每 2 秒输出一行线圈列表和一行输入寄存器列表。
• 线圈输出为长度 8 的布尔列表，输入寄存器输出为长度 4 的整数列表，与服务器端配置的地址范围一致。
• 若服务器未启动或地址错误，应打印 [FATAL] 无法连接 … 并退出；若某次读取失败，应打印 [ERROR] 并继续下一轮，不中断循环。
• 按 Ctrl+C 应打印“已停止采集”并正常关闭连接。

常见问题

• slave= 参数在 pymodbus 3.x 中用于指定从站号，2.x 使用 unit=，版本不匹配会报错。
• 部分服务器要求 UNIT_ID 与配置一致，否则返回异常码而非数据。
• 读取数量超过服务器实际寄存器范围会返回 IllegalAddress 错误，需按服务器实际映射调整地址与数量。