modbus-tcp-cython 0.1.2

A Cython Modbus TCP client with heartbeat and multi-connection management.

modbus-tcp-cython

一个 Cython 实现的 Modbus TCP 客户端模块。

支持：

• Modbus TCP MBAP 组包和响应解析
• 读保持寄存器，功能码 03
• 读输入寄存器，功能码 04
• 写单个保持寄存器，功能码 06
• 写多个保持寄存器，功能码 10
• 自定义 PDU 请求
• 普通请求失败后自动重连重试
• 后台心跳和自动重连
• 多连接管理

安装

本地源码安装：

python -m pip install -U pip setuptools wheel Cython
python -m pip install .

强制重新安装本地源码：

python -m pip install --force-reinstall --no-build-isolation .

基本用法

from modbus_tcp import ModbusTcpClient

client = ModbusTcpClient(
    "192.168.180.18",
    26,
    timeout=5.0,
    debug=True,
    request_retries=1,
    request_retry_delay=0.2,
)

try:
    client.connect()
    values = client.read_holding_registers(
        unit_id=3,
        start_address=2,
        quantity=1,
    )
    print(values)
finally:
    client.close()

ModbusTcpClient

构造函数：

ModbusTcpClient(
    host,
    port=502,
    timeout=5.0,
    debug=False,
    request_retries=1,
    request_retry_delay=0.2,
)

参数：

• host：Modbus TCP 服务端 IP
• port：Modbus TCP 服务端端口，标准端口通常是 502
• timeout：socket 连接和读写超时时间，单位秒
• debug：是否打印 TX hex / RX hex
• request_retries：普通请求失败后的自动重试次数，默认 1
• request_retry_delay：重连后再次发送前的等待时间，单位秒

常用状态：

client.host
client.port
client.timeout
client.debug
client.is_connected
client.request_retries
client.request_retry_delay

连接管理

client.connect()

建立 TCP 连接。如果已经连接，会直接返回。

client.close()

关闭 TCP 连接，并停止心跳线程。

普通请求自动重连重试

普通读写请求默认支持自动重连重试。

流程：

发送请求
读取响应
如果超时或 socket 错误
关闭旧 socket
重新 connect
等待 request_retry_delay
重发请求
仍失败才抛异常

关闭自动重试：

client = ModbusTcpClient(
    "192.168.180.18",
    26,
    request_retries=0,
)

增加重试次数：

client = ModbusTcpClient(
    "192.168.180.18",
    26,
    request_retries=2,
    request_retry_delay=0.5,
)

注意：读请求重试通常安全。写请求如果设备已经执行但响应丢失，重试可能造成重复写入或重复动作。

读保持寄存器 03

values = client.read_holding_registers(
    unit_id=3,
    start_address=2,
    quantity=1,
)

对应 PDU：

03 00 02 00 01

返回：

[396]

读输入寄存器 04

values = client.read_input_registers(
    unit_id=3,
    start_address=0,
    quantity=2,
)

对应 PDU：

04 00 00 00 02

返回：

[123, 456]

写单个保持寄存器 06

value = client.write_single_register(
    unit_id=3,
    address=0x0006,
    value=20,
)

对应 PDU：

06 00 06 00 14

成功时返回写入值：

20

写多个保持寄存器 10

count = client.write_multiple_registers(
    unit_id=3,
    start_address=0x0010,
    values=[1, 2, 3],
)

成功时返回写入寄存器数量：

3

自定义 PDU

厂家自定义报文可以直接发送 PDU：

response_pdu = client.request_pdu(
    unit_id=0xFF,
    pdu=bytes.fromhex("03 00 70 00 00"),
)

例如 RTU 文档写：

FF 03 00 70 00 00 CRC_L CRC_H

Modbus TCP 侧只需要传：

unit_id = FF
pdu     = 03 00 70 00 00

网关会把 TCP 的 Unit ID 转成 RTU 地址码，并自动补 RTU CRC。

心跳

启动后台心跳：

client.start_heartbeat(
    unit_id=3,
    address=2,
    quantity=1,
    interval=10.0,
    max_failures=3,
    reconnect_attempts=3,
    reconnect_delay=2.0,
)

心跳会定时读取一个寄存器。连续失败达到 max_failures 后，会自动关闭连接并重连。

停止心跳：

client.stop_heartbeat()

查看心跳状态：

client.is_heartbeat_running()

如果心跳读取的是输入寄存器：

client.start_heartbeat(
    unit_id=3,
    address=0,
    quantity=1,
    use_input_registers=True,
)

多连接管理

from modbus_tcp import ModbusTcpManager

with ModbusTcpManager() as manager:
    manager.add_client(
        name="alarm_1",
        host="192.168.180.18",
        port=26,
        unit_id=3,
        timeout=5.0,
        debug=True,
        request_retries=1,
        request_retry_delay=0.2,
        heartbeat_address=2,
        heartbeat_interval=5.0,
        start_heartbeat=True,
    )

    values = manager.read_holding_registers(
        "alarm_1",
        start_address=2,
        quantity=1,
    )
    print(values)

ModbusTcpManager 常用接口

注册一路连接：

manager.add_client(
    name,
    host,
    port=502,
    unit_id=1,
    timeout=5.0,
    debug=False,
    request_retries=1,
    request_retry_delay=0.2,
    heartbeat_address=0,
    heartbeat_quantity=1,
    heartbeat_interval=10.0,
    heartbeat_max_failures=3,
    reconnect_attempts=3,
    reconnect_delay=2.0,
    use_input_registers=False,
    autoconnect=False,
    start_heartbeat=False,
)

连接管理：

manager.connect(name)
manager.close(name)
manager.connect_all()
manager.close_all()
manager.remove_client(name)

心跳管理：

manager.start_heartbeat(name)
manager.stop_heartbeat(name)
manager.start_all(connect=True, heartbeat=True)
manager.stop_all_heartbeats()

状态查询：

states = manager.states()

返回示例：

{
    "alarm_1": {
        "connected": True,
        "heartbeat_running": True,
    }
}

按连接名读写：

manager.read_holding_registers("alarm_1", 2, 1)
manager.read_input_registers("alarm_1", 0, 2)
manager.write_single_register("alarm_1", 0x0006, 20)
manager.write_multiple_registers("alarm_1", 0x0010, [1, 2, 3])
manager.request_pdu("alarm_1", bytes.fromhex("03 00 70 00 00"), unit_id=0xFF)

如果调用时不传 unit_id，会使用 add_client() 注册时配置的默认 unit_id。

多路 Modbus TCP 转 RTU 网关

对于多路网关：

IP + TCP 端口 选择哪一路串口
Unit ID       选择这一路串口下面哪个 RTU 从站

例如：

192.168.180.18:5550 + Unit ID 1 -> HC0 串口上的 1 号从站
192.168.180.18:5551 + Unit ID 3 -> HC1 串口上的 3 号从站

异常

模块异常类：

from modbus_tcp import ModbusTcpError

示例：

try:
    client.read_holding_registers(3, 2, 1)
except ModbusTcpError as exc:
    print(exc)

常见异常：

• modbus exception: function=0x83 code=0x02
• connection closed by remote
• transaction id mismatch
• unit id mismatch
• invalid byte count

调试建议

1. 开启 debug=True 查看 TX hex 和 RX hex。
2. 偶发超时时，优先调整 timeout、request_retries、request_retry_delay。
3. 后端是 RTU 网关时，轮询间隔不要太短。
4. 同一个网关端口通常只允许一个 TCP 客户端连接。
5. 如果同时启用心跳和业务轮询，注意串口总线压力。