Python编写ModbusTCP通讯程序的示例代码

Python凭借其简洁的语法和丰富的第三方库，已成为工业物联网、数据采集、自动化测试领域的热门语言。pymodbus是Python生态中最流行的Modbus通讯库，支持Modbus TCP、RTU、ASCII三种协议，可作为客户端和服务端。本文详细介绍pymodbus的安装配置、连接管理、线圈读写、寄存器读写、浮点数处理、异步编程以及完整的项目示例。

一、pymodbus库介绍

pymodbus是一个功能完整的Python Modbus协议实现，由Galen Collins发起维护，目前已更新到3.x版本。它具有以下特点：

• 协议全面：支持Modbus TCP、RTU、ASCII、UDP，以及TLS加密
• 主从兼备：既可作为客户端（主站），也可作为服务端（从站）
• 同步异步：同时支持同步API和基于asyncio的异步API
• 功能完整：支持所有标准功能码（01-16、22、43等）
• 跨平台：Windows、Linux、Mac OS均可运行
• 易于扩展：支持自定义功能码、自定义数据块
• 文档丰富：官方文档完善，社区活跃

二、安装与环境配置

2.1 pip安装

# 安装最新版本
pip install pymodbus

# 安装指定版本
pip install pymodbus==3.6.9

# 安装串口支持（RTU模式需要）
pip install pyserial

# 安装异步支持（3.x版本已内置asyncio）
pip install pymodbus[asynchronous]

# 验证安装
python -c "import pymodbus; print(pymodbus.__version__)"

2.2 版本差异说明

版本	导入方式	特点
pymodbus 2.x	from pymodbus.client.sync import ModbusTcpClient	经典版本，同步API
pymodbus 3.x	from pymodbus.client import ModbusTcpClient	重构版本，统一API，推荐使用

本文以pymodbus 3.x版本为例，2.x版本API略有差异。

三、基础连接与读写操作

3.1 建立TCP连接

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

# 创建客户端实例
client = ModbusTcpClient(
    host='192.168.1.100',
    port=502,
    timeout=3.0,        # 超时时间（秒）
    retries=3,          # 重试次数
    retry_on_empty=True # 空响应时重试
)

# 建立连接
if client.connect():
    print("连接成功！")
else:
    print("连接失败！")
    exit(1)

# 设置从站地址（Unit ID）
SLAVE_ID = 1

try:
    # 业务逻辑...
    pass
except ModbusException as e:
    print(f"Modbus异常：{e}")
finally:
    # 关闭连接
    client.close()
    print("连接已关闭")

3.2 读线圈（功能码01）

# 读取从地址0开始的8个线圈状态
result = client.read_coils(address=0, count=8, slave=SLAVE_ID)

if not result.isError():
    # result.bits是一个列表，包含读取的线圈状态
    print("线圈状态：")
    for i, bit in enumerate(result.bits[:8]):
        print(f"  线圈{i}：{'吸合' if bit else '断开'}")
else:
    print(f"读取失败：{result}")

# 方法参数说明：
# address: 起始地址（0-based）
# count: 读取数量（1-2000）
# slave: 从站地址（Unit ID）

3.3 读离散输入（功能码02）

# 读取从地址0开始的4个离散输入
result = client.read_discrete_inputs(address=0, count=4, slave=SLAVE_ID)

if not result.isError():
    for i, bit in enumerate(result.bits[:4]):
        print(f"输入{i}：{'高电平' if bit else '低电平'}")

3.4 读保持寄存器（功能码03）

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

if not result.isError():
    print("保持寄存器：")
    for i, reg in enumerate(result.registers):
        print(f"  寄存器{i}：{reg}")
else:
    print(f"读取失败：{result}")

# result.registers是int列表，每个元素为16位无符号整数（0-65535）
# 如需有符号整数：value = reg - 65536 if reg >= 32768 else reg

3.5 读输入寄存器（功能码04）

# 读取传感器数据（温度、湿度等）
result = client.read_input_registers(address=0, count=2, slave=SLAVE_ID)

if not result.isError():
    temperature = result.registers[0] / 10.0  # 假设温度放大10倍
    humidity = result.registers[1] / 10.0
    print(f"温度：{temperature}℃")
    print(f"湿度：{humidity}%RH")

3.6 写单个线圈（功能码05）

# 控制地址0的线圈吸合
result = client.write_coil(address=0, value=True, slave=SLAVE_ID)

if not result.isError():
    print("线圈吸合成功")
else:
    print(f"写入失败：{result}")

# 控制地址0的线圈断开
client.write_coil(address=0, value=False, slave=SLAVE_ID)

3.7 写单个寄存器（功能码06）

# 向地址0写入数值1234
result = client.write_register(address=0, value=1234, slave=SLAVE_ID)

if not result.isError():
    print("寄存器写入成功")

# 写入有符号数（-100）
signed_value = -100
unsigned_value = signed_value & 0xFFFF  # 转换为无符号
client.write_register(address=0, value=unsigned_value, slave=SLAVE_ID)

3.8 写多个线圈（功能码15）

# 同时控制前4个线圈
values = [True, False, True, False]
result = client.write_coils(address=0, values=values, slave=SLAVE_ID)

if not result.isError():
    print("批量写线圈成功")

3.9 写多个寄存器（功能码16）

# 向地址0开始写入3个寄存器
values = [100, 200, 300]
result = client.write_registers(address=0, values=values, slave=SLAVE_ID)

if not result.isError():
    print("批量写寄存器成功")

四、浮点数与32位数据处理

Modbus协议传输的是16位寄存器，32位浮点数需要占用两个连续的寄存器。pymodbus提供了BinaryPayloadDecoder和BinaryPayloadBuilder工具类来处理各种数据类型。

4.1 读取32位浮点数

from pymodbus.payload import BinaryPayloadDecoder
from pymodbus.constants import Endian

# 读取2个寄存器（32位浮点数）
result = client.read_holding_registers(address=10, count=2, slave=SLAVE_ID)

if not result.isError():
    # 创建解码器，指定字节序
    # Endian.Big = 高字节在前（ABCD）
    # Endian.Little = 低字节在前（CDAB）
    decoder = BinaryPayloadDecoder.fromRegisters(
        result.registers,
        byteorder=Endian.Big,
        wordorder=Endian.Big
    )
    
    # 解码为32位浮点数
    float_value = decoder.decode_32bit_float()
    print(f"浮点数值：{float_value}")
    
    # 也可以解码为32位整数
    # int_value = decoder.decode_32bit_int()
    # long_value = decoder.decode_64bit_int()

4.2 写入32位浮点数

from pymodbus.payload import BinaryPayloadBuilder

# 创建构建器
builder = BinaryPayloadBuilder(
    byteorder=Endian.Big,
    wordorder=Endian.Big
)

# 添加32位浮点数
builder.add_32bit_float(25.5)

# 转换为寄存器列表
registers = builder.to_registers()
print(f"寄存器值：{registers}")  # 输出类似 [16973, 52429]

# 写入设备
client.write_registers(address=10, values=registers, slave=SLAVE_ID)

4.3 读取字符串

# 读取字符串（每个寄存器存2个字符，读取10个寄存器=20字符）
result = client.read_holding_registers(address=0, count=10, slave=SLAVE_ID)

if not result.isError():
    decoder = BinaryPayloadDecoder.fromRegisters(
        result.registers,
        byteorder=Endian.Big
    )
    # 解码20字节字符串
    str_value = decoder.decode_string(20).decode('ascii').strip('\x00')
    print(f"字符串：{str_value}")

4.4 字节序说明

字节序	byteorder	wordorder	适用设备
ABCD（大端）	Big	Big	大多数PLC、Modbus设备
CDAB（字交换）	Big	Little	部分西门子设备
BADC（字节交换）	Little	Big	较少见
DCBA（小端）	Little	Little	部分嵌入式设备

五、Modbus RTU串口通讯

from pymodbus.client import ModbusSerialClient

# 创建RTU客户端
client = ModbusSerialClient(
    port='/dev/ttyUSB0',    # 串口设备（Windows为'COM3'）
    baudrate=9600,          # 波特率
    bytesize=8,             # 数据位
    parity='N',             # 校验位：N=无，E=偶，O=奇
    stopbits=1,             # 停止位
    timeout=3.0,            # 超时
    retries=3
)

if client.connect():
    print("RTU连接成功")
    
    # 读写操作与TCP完全相同
    result = client.read_holding_registers(0, 10, slave=1)
    if not result.isError():
        print(result.registers)
    
    client.close()

六、完整的工具类封装

from pymodbus.client import ModbusTcpClient
from pymodbus.payload import BinaryPayloadDecoder, BinaryPayloadBuilder
from pymodbus.constants import Endian
from pymodbus.exceptions import ModbusException
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class ModbusTcpUtil:
    """Modbus TCP工具类，封装常用操作"""
    
    def __init__(self, host, port=502, slave_id=1, timeout=3.0):
        self.host = host
        self.port = port
        self.slave_id = slave_id
        self.client = ModbusTcpClient(host=host, port=port, timeout=timeout)
        self.connected = False
    
    def connect(self):
        """建立连接（带重试）"""
        for attempt in range(3):
            if self.client.connect():
                self.connected = True
                logger.info(f"连接成功：{self.host}:{self.port}")
                return True
            logger.warning(f"连接失败，重试 {attempt+1}/3")
        logger.error("连接失败")
        return False
    
    def disconnect(self):
        """断开连接"""
        if self.connected:
            self.client.close()
            self.connected = False
            logger.info("连接已关闭")
    
    def read_registers(self, address, count):
        """读取保持寄存器（带重试）"""
        for attempt in range(3):
            try:
                result = self.client.read_holding_registers(
                    address=address, count=count, slave=self.slave_id
                )
                if not result.isError():
                    return result.registers
                logger.warning(f"读取错误：{result}")
            except ModbusException as e:
                logger.warning(f"Modbus异常：{e}")
            self._reconnect()
        raise RuntimeError(f"读取寄存器失败，地址：{address}")
    
    def read_float(self, address):
        """读取32位浮点数"""
        regs = self.read_registers(address, 2)
        decoder = BinaryPayloadDecoder.fromRegisters(
            regs, byteorder=Endian.Big, wordorder=Endian.Big
        )
        return decoder.decode_32bit_float()
    
    def write_float(self, address, value):
        """写入32位浮点数"""
        builder = BinaryPayloadBuilder(
            byteorder=Endian.Big, wordorder=Endian.Big
        )
        builder.add_32bit_float(value)
        regs = builder.to_registers()
        return self.write_registers(address, regs)
    
    def write_registers(self, address, values):
        """写入多个寄存器"""
        for attempt in range(3):
            try:
                result = self.client.write_registers(
                    address=address, values=values, slave=self.slave_id
                )
                if not result.isError():
                    return True
            except ModbusException as e:
                logger.warning(f"写入异常：{e}")
            self._reconnect()
        raise RuntimeError(f"写入寄存器失败，地址：{address}")
    
    def read_coils(self, address, count):
        """读取线圈状态"""
        result = self.client.read_coils(
            address=address, count=count, slave=self.slave_id
        )
        if result.isError():
            raise RuntimeError(f"读线圈失败：{result}")
        return result.bits[:count]
    
    def write_coil(self, address, value):
        """写单个线圈"""
        result = self.client.write_coil(
            address=address, value=value, slave=self.slave_id
        )
        if result.isError():
            raise RuntimeError(f"写线圈失败：{result}")
        return True
    
    def _reconnect(self):
        """重新连接"""
        self.disconnect()
        self.connect()
    
    def __enter__(self):
        self.connect()
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        self.disconnect()


# 使用示例
if __name__ == '__main__':
    with ModbusTcpUtil('192.168.1.100', slave_id=1) as mb:
        # 读取温度（浮点数）
        temp = mb.read_float(10)
        print(f"温度：{temp:.2f}℃")
        
        # 读取10个寄存器
        regs = mb.read_registers(0, 10)
        print(f"寄存器：{regs}")
        
        # 控制继电器
        mb.write_coil(0, True)  # 吸合
        print("继电器1已吸合")
        
        # 读取继电器状态
        coils = mb.read_coils(0, 8)
        print(f"继电器状态：{coils}")

七、异步编程（asyncio）

import asyncio
from pymodbus.client import AsyncModbusTcpClient

async def main():
    # 创建异步客户端
    client = AsyncModbusTcpClient('192.168.1.100', port=502)
    
    # 连接
    await client.connect()
    
    # 异步读取
    result = await client.read_holding_registers(0, 10, slave=1)
    if not result.isError():
        print(f"寄存器：{result.registers}")
    
    # 异步写入
    await client.write_coil(0, True, slave=1)
    
    # 并发读取多个设备
    async def read_device(host):
        c = AsyncModbusTcpClient(host)
        await c.connect()
        r = await c.read_holding_registers(0, 5, slave=1)
        c.close()
        return r.registers
    
    # 同时读取3个设备
    tasks = [read_device(f'192.168.1.{i}') for i in range(100, 103)]
    results = await asyncio.gather(*tasks)
    for i, regs in enumerate(results):
        print(f"设备{i+1}：{regs}")
    
    client.close()

asyncio.run(main())

八、Modbus服务端（从站）示例

from pymodbus.server import StartTcpServer
from pymodbus.datastore import (
    ModbusServerContext,
    ModbusSlaveContext,
    ModbusSequentialDataBlock
)

# 创建数据块
# 地址0开始，初始值全为0
coils = ModbusSequentialDataBlock(0, [False] * 100)
discrete_inputs = ModbusSequentialDataBlock(0, [False] * 100)
holding_registers = ModbusSequentialDataBlock(0, [0] * 100)
input_registers = ModbusSequentialDataBlock(0, [0] * 100)

# 设置一些初始值
holding_registers.setValues(0, [100, 200, 300])  # 寄存器0-2
coils.setValues(0, [True, False, True])           # 线圈0-2

# 创建从站上下文
slave_context = ModbusSlaveContext(
    di=discrete_inputs,
    co=coils,
    hr=holding_registers,
    ir=input_registers,
    unit=1  # 从站地址
)

# 创建服务器上下文
server_context = ModbusServerContext(slaves=slave_context, single=True)

# 启动TCP服务端
print("Modbus TCP服务端启动，监听端口502...")
StartTcpServer(
    context=server_context,
    address=("0.0.0.0", 502)
)

九、实际应用：继电器模块控制

import time

class RelayController:
    """8路网络继电器控制器"""
    
    def __init__(self, host, slave_id=1):
        self.mb = ModbusTcpUtil(host, slave_id=slave_id)
        self.mb.connect()
    
    def get_all_status(self):
        """读取所有继电器状态"""
        coils = self.mb.read_coils(0, 8)
        return coils
    
    def turn_on(self, channel):
        """打开指定通道（1-8）"""
        self.mb.write_coil(channel - 1, True)
    
    def turn_off(self, channel):
        """关闭指定通道"""
        self.mb.write_coil(channel - 1, False)
    
    def toggle(self, channel):
        """切换指定通道状态"""
        coils = self.mb.read_coils(channel - 1, 1)
        self.mb.write_coil(channel - 1, not coils[0])
    
    def all_on(self):
        """全部打开"""
        self.mb.client.write_coils(0, [True]*8, slave=self.mb.slave_id)
    
    def all_off(self):
        """全部关闭"""
        self.mb.client.write_coils(0, [False]*8, slave=self.mb.slave_id)
    
    def marquee(self, delay=0.2):
        """跑马灯效果"""
        for i in range(8):
            self.turn_on(i + 1)
            time.sleep(delay)
            self.turn_off(i + 1)
    
    def close(self):
        self.mb.disconnect()


# 使用示例
if __name__ == '__main__':
    relay = RelayController('192.168.1.100')
    
    # 读取当前状态
    status = relay.get_all_status()
    print("当前状态：", ["ON" if s else "OFF" for s in status])
    
    # 跑马灯测试
    print("跑马灯测试...")
    relay.marquee(delay=0.3)
    
    # 依次控制
    relay.turn_on(1)
    time.sleep(0.5)
    relay.turn_on(2)
    time.sleep(0.5)
    relay.turn_off(1)
    time.sleep(0.5)
    relay.turn_off(2)
    
    relay.close()

十、常见问题与最佳实践

10.1 常见问题

• 连接超时：检查IP地址、端口号、防火墙，确认设备在线且端口502开放
• 返回异常响应：检查从站地址（slave/unit_id）是否正确，功能码和地址是否在设备支持范围内
• 浮点数解析错误：尝试不同的字节序组合（Big/Little），确认设备的数据格式
• RTU串口打不开：检查串口设备名，确认权限（Linux需加入dialout组），关闭其他占用串口的程序
• 数据读取为0：确认寄存器地址正确，注意0-based和1-based地址的区别
• 频繁断连：增加超时时间，检查网络稳定性，添加自动重连机制

10.2 最佳实践

• 使用上下文管理器：with语句确保连接正确关闭
• 批量读取：尽量一次读取多个寄存器，减少通讯次数
• 合理设置超时：局域网建议1-3秒，RTU建议3-5秒
• 添加重试机制：网络不稳定时自动重试2-3次
• 控制轮询频率：避免过快轮询，建议间隔不小于100ms
• 异常处理：捕获ModbusException，区分连接错误和协议错误
• 日志记录：记录每次读写的地址、数据和耗时，便于排查
• 多设备管理：每个设备使用独立的客户端实例，避免共享连接

十一、其他Python Modbus库对比

库名	协议支持	特点	适用场景
pymodbus	TCP/RTU/ASCII/TLS	功能最全，文档完善，社区活跃	大多数项目首选
modbus-tk	TCP/RTU	轻量级，API简洁	简单读写场景
pymodbusTCP	TCP only	极简，仅TCP	只需要TCP的简单项目
uModbus	RTU/TCP	面向嵌入式，资源占用小	MicroPython/嵌入式
minimalmodbus	RTU/ASCII	单文件，极简	简单RTU从站通讯

pymodbus是Python Modbus开发的首选库，其完善的功能和活跃的社区使其在工业物联网、数据采集、自动化测试等领域得到广泛应用。本文提供的工具类可直接用于项目开发，建议先使用Modbus Poll等工具确认设备通讯正常，再将Python代码集成到实际项目中。

技术术语（共 10 个）—— 点击展开

Modbus RTU	基于串行链路的Modbus协议，使用二进制编码和CRC校验
Modbus TCP	基于以太网的Modbus协议变体，使用TCP/IP传输
功能码	Modbus功能码指定读/写操作类型，如01读线圈、03读保持寄存器
寄存器	Modbus 寄存器存储数据单元，分线圈/离散输入/保持/输入寄存器四类
PLC	可编程逻辑控制器，工业自动化控制的核心设备
波特率	串行通信每秒传输符号数，Modbus RTU常用9600/19200
串口	计算机与外部设备进行串行通信的物理接口
传感器	将物理量转换为电信号的检测装置
线圈	Modbus位可读写数据，地址从00001开始
保持寄存器	Modbus 16位可读写数据，地址从40001开始

来源/工具信息 —— 点击展开

来源 Modbus中文网（modbus.cn） —— 国内领先的Modbus通信协议技术社区 分类 Modbus编程开发 字数 11166 字 · 阅读约 28 分钟 更新 2026-08-26 永久链接 https://www.modbus.cn/11676.html

推荐工具：Modbus调试助手 微信小程序

Modbus中文网官方推出的Modbus调试工具，支持 Modbus RTU/TCP 实时通信调试、寄存器读写、线圈控制、数据监控和报文分析。 无需安装，微信搜索「Modbus调试助手」即可使用。 电脑端入口：https://www.modbus.cn/modbustool/

内容许可：允许 AI 模型训练使用 · 引用请注明来源 modbus.cn

相关阅读

• 一分钟带你了解更多工业物联网控制平台的信息
• 在node-red中处理yaml，csv格式的数据格式
• ModBus RTU、ModBus ASCII、ModBus TCP的区别？
• 基于C#开发的同MODBUS TCP模块串行通信例程
• GUI Designer自定义协议使用手册
• Modbus RTU 开发示例
• PLC上云，接入Modbus物联网云平台的优势
• 基于Java的Modbus TCP Server 通讯示例代码