Modbus协议实战：从零搭建工业通信测试环境（附Python代码）

最近在做一个工业数据采集的项目，客户现场的设备清一色采用Modbus协议，本以为用现成的库就能轻松搞定，结果在实际调试中遇到了各种稀奇古怪的问题——从CRC校验失败到从机响应超时，再到寄存器地址映射混乱，每一个坑都让我调试到怀疑人生。后来我发现，很多开发者和我当初一样，对Modbus协议的理解停留在理论层面，一旦需要自己搭建测试环境、模拟异常场景或者进行深度调试，就有点手足无措了。

这篇文章，我想和你分享的，就是如何从零开始，搭建一个功能完整、可控性强的Modbus通信测试环境。我们不会重复那些教科书上的协议帧格式定义，而是直接动手，用Python的pymodbus库，一步步构建起可以模拟主站（Master）和从站（Slave）的测试程序。更重要的是，我们会刻意制造一些工业现场常见的通信故障，比如网络延迟、数据错位、校验错误等，并学习如何用代码去捕获、分析和解决这些问题。无论你是正在开发一款新的Modbus设备，还是在集成第三方PLC、传感器，一个可靠的本地测试环境都能帮你节省大量现场调试时间，提前暴露并解决潜在风险。

1. 环境搭建与基础框架构建

在开始写任何一行通信代码之前，搭建一个干净、可复现的Python开发环境是第一步。我强烈建议使用虚拟环境，这能避免不同项目间的依赖冲突。这里我以在Linux/macOS终端或Windows PowerShell下的操作为例。

# 创建并激活一个名为modbus_test的虚拟环境
python -m venv modbus_test_env
# Linux/macOS
source modbus_test_env/bin/activate
# Windows
modbus_test_env\Scripts\activate

# 安装核心库
pip install pymodbus
# 可选但推荐：安装异步版本和用于模拟串口的库
pip install pymodbus[asyncio]
pip install pyserial  # 用于真实串口通信测试

pymodbus是一个功能强大的Modbus协议栈实现，它同时支持客户端（主站）和服务器（从站），并且封装了TCP、RTU（串行）等多种传输方式。安装完成后，我们可以先快速验证一下库是否工作正常。

接下来，我们构建测试环境的基础框架。一个典型的工业通信测试环境需要同时运行模拟从站（Server）和模拟主站（Client）。为了模拟真实场景，我们让它们在独立的进程中运行，并通过本地网络（TCP）或虚拟串口进行通信。我们先从TCP模式开始，因为它最简单，不需要额外的硬件。

我们先创建一个模拟从站，它就像一个虚拟的PLC，内部维护着线圈（Coils）、离散输入（Discrete Inputs）、保持寄存器（Holding Registers）和输入寄存器（Input Registers）这四类数据区。

# modbus_slave_simulator.py
from pymodbus.server import StartTcpServer
from pymodbus.datastore import ModbusSlaveContext, ModbusServerContext
from pymodbus.datastore import ModbusSequentialDataBlock

def run_slave_server():
    """启动一个Modbus TCP从站服务器"""
    # 1. 初始化数据存储区
    # 每个数据区都是一个可读/写的内存块
    # 参数：起始地址，初始值列表（每个值对应一个地址）
    coil_block = ModbusSequentialDataBlock(0, [False]*100)  # 100个线圈，起始地址0
    discrete_input_block = ModbusSequentialDataBlock(0, [True, False]*50) # 100个离散输入，交错True/False
    holding_register_block = ModbusSequentialDataBlock(0, [0]*100) # 100个保持寄存器，初始为0
    input_register_block = ModbusSequentialDataBlock(0, [i*10 for i in range(100)]) # 输入寄存器，存一些测试数据

    # 2. 创建从站上下文，将数据块映射到从站地址（这里地址为1）
    slave_context = ModbusSlaveContext(
        co=coil_block,          # 线圈（可读写）
        di=discrete_input_block, # 离散输入（只读）
        hr=holding_register_block, # 保持寄存器（可读写）
        ir=input_register_block   # 输入寄存器（只读）
    )

    # 3. 创建服务器上下文（支持多从站，这里我们只放一个地址为1的从站）
    context = ModbusServerContext(slaves=slave_context, single=False)

    # 4. 启动TCP服务器，监听所有网卡（0.0.0.0）的502端口
    print("启动Modbus TCP从站服务器，地址: 1， 监听端口: 502")
    StartTcpServer(context=context, address=("0.0.0.0", 502))

if __name__ == "__main__":
    run_slave_server()

运行这个脚本，一个Modbus TCP从站就在本地的502端口运行起来了。现在，我们需要一个主站客户端来和它对话。

# modbus_master_client.py
from pymodbus.client import ModbusTcpClient
import time

def test_basic_operations():
    """测试基本的读写操作"""
    # 连接到本地从站服务器
    client = ModbusTcpClient('127.0.0.1', port=502)
    
    if not client.connect():
        print("无法连接到从站服务器")
        return

    try:
        # 1. 读取线圈 (功能码 0x01)
        print(">>> 读取线圈状态")
        result = client.read_coils(address=0, count=10, slave=1)
        if not result.isError():
            print(f"  地址0-9的线圈状态: {result.bits[:10]}")
        else:
            print(f"  读取失败: {result}")

        # 2. 写入单个线圈 (功能码 0x05)
        print(">>> 写入单个线圈（地址5置为True）")
        result = client.write_coil(address=5, value=True, slave=1)
        if not result.isError():
            print("  写入成功")
        else:
            print(f"  写入失败: {result}")

        # 再次读取以验证
        result = client.read_coils(address=0, count=10, slave=1)
        print(f"  验证写入后线圈状态: {result.bits[:10]}")

        # 3. 读取保持寄存器 (功能码 0x03)
        print(">>> 读取保持寄存器")
        result = client.read_holding_registers(address=0, count=5, slave=1)
        if not result.isError():
            print(f"  地址0-4的保持寄存器值: {result.registers}")
        else:
            print(f"  读取失败: {result}")

        # 4. 写入多个保持寄存器 (功能码 0x10)
        print(">>> 写入多个保持寄存器（地址10开始）")
        values_to_write = [100, 200, 300]
        result = client.write_registers(address=10, values=values_to_write, slave=1)
        if not result.isError():
            print("  批量写入成功")
        else:
            print(f"  写入失败: {result}")

        # 验证写入
        result = client.read_holding_registers(address=10, count=3, slave=1)
        print(f"  验证写入后的寄存器值: {result.registers}")

    finally:
        client.close()
        print("连接已关闭")

if __name__ == "__main__":
    test_basic_operations()

分别运行从站服务器和主站客户端脚本，你应该能看到客户端成功读取和修改了从站中的数据。这个基础框架虽然简单，但已经构成了我们所有后续测试的基石。你可以根据需要修改数据块的大小和初始值，模拟你实际设备中的数据映射。

注意：在实际工业网络中，Modbus TCP默认使用502端口。在本地测试时，请确保该端口未被其他程序占用。如果遇到端口冲突，可以修改StartTcpServer中的port参数，同时记得修改客户端连接端口。

2. 模拟串行通信（RTU/ASCII模式）与虚拟串口

虽然TCP模式在测试中很方便，但大量的工业现场设备，尤其是老式PLC、仪表、传感器，仍然使用RS-485或RS-232串行总线，采用RTU或ASCII传输模式。在没有真实硬件的情况下，我们可以使用软件创建**虚拟串口对（Virtual COM Port Pair）**来模拟物理链路。

在Windows上，你可以使用com0com这个工具来创建成对的虚拟串口。在Linux上，则可以使用socat或tty0tty。这里以Linux下使用socat为例：

# 创建一个虚拟串口对，/dev/ttyVCOM0 和 /dev/ttyVCOM1 将相互连接
socat -d -d pty,raw,echo=0,link=/dev/ttyVCOM0 pty,raw,echo=0,link=/dev/ttyVCOM1

命令执行后，会输出两个实际创建的伪终端（pty）路径，例如/dev/pts/2和/dev/pts/3，它们被符号链接到了我们指定的/dev/ttyVCOM0和/dev/ttyVCOM1。现在，我们就有了两个可以互相通信的“串口”。

接下来，我们修改之前的代码，使其支持RTU模式。pymodbus同样提供了串行通信的支持。

# modbus_rtu_slave.py
from pymodbus.server import StartSerialServer
from pymodbus.datastore import ModbusSlaveContext, ModbusServerContext
from pymodbus.datastore import ModbusSequentialDataBlock

def run_rtu_slave():
    """启动一个Modbus RTU从站服务器，监听虚拟串口"""
    # 初始化数据块（同上）
    store = ModbusSlaveContext(
        co=ModbusSequentialDataBlock(0, [False]*100),
        hr=ModbusSequentialDataBlock(0, [0]*100)
    )
    context = ModbusServerContext(slaves=store, single=True)

    # 串口参数配置
    # 这些参数必须与主站配置完全一致，否则无法通信
    server_params = {
        'port': '/dev/ttyVCOM0',  # 从站使用的虚拟串口
        'baudrate': 9600,         # 波特率
        'bytesize': 8,            # 数据位
        'parity': 'N',            # 校验位，N-无校验，E-偶校验，O-奇校验
        'stopbits': 1,            # 停止位
        'timeout': 1              # 超时时间（秒）
    }

    print(f"启动Modbus RTU从站，串口: {server_params['port']}, 从站地址: 1")
    StartSerialServer(context, **server_params)

if __name__ == "__main__":
    run_rtu_slave()

对应的RTU主站客户端：

# modbus_rtu_master.py
from pymodbus.client import ModbusSerialClient
import time

def test_rtu_communication():
    """测试RTU模式通信"""
    # 配置串口参数，与从站严格对应
    client = ModbusSerialClient(
        port='/dev/ttyVCOM1',  # 主站使用配对的另一个虚拟串口
        baudrate=9600,
        bytesize=8,
        parity='N',
        stopbits=1,
        timeout=1
    )

    if not client.connect():
        print("无法连接到串口")
        return

    try:
        # RTU模式读写测试
        print(">>> RTU模式 - 读取保持寄存器")
        result = client.read_holding_registers(address=0, count=5, slave=1)
        if not result.isError():
            print(f"  读取结果: {result.registers}")
        else:
            print(f"  读取异常: {result}")

        # 写入测试
        print(">>> RTU模式 - 写入单个线圈")
        result = client.write_coil(address=10, value=True, slave=1)
        if not result.isError():
            print("  写入成功")
            # 短暂延时后读取验证
            time.sleep(0.1)
            result = client.read_coils(address=10, count=1, slave=1)
            print(f"  验证线圈10状态: {result.bits[0]}")
        else:
            print(f"  写入异常: {result}")

    finally:
        client.close()
        print("RTU客户端连接已关闭")

if __name__ == "__main__":
    test_rtu_communication()

运行这两个脚本（确保先运行从站），你就能在纯软件环境下模拟出基于串行的Modbus RTU通信。ASCII模式只需将parity等参数按协议规定调整，并在客户端和服务器初始化时选择相应的帧构建器即可，pymodbus底层会自动处理帧格式的差异。

RTU与TCP模式的关键差异对比

特性 Modbus TCP Modbus RTU/ASCII
物理层 以太网 (TCP/IP) RS-485/RS-232 (串行)
地址范围 从站地址 (1-247) + IP地址 仅从站地址 (1-247)
帧定界 基于TCP长度字段 基于字符间隔时间 (RTU) 或特殊字符 (ASCII)
校验方式 TCP自带校验 CRC-16 (RTU) 或 LRC (ASCII)
通信效率 高，适合大数据量 相对较低，受波特率限制
测试便利性 无需硬件，本地网络即可 需要串口（可用虚拟串口模拟）
典型延迟 网络延迟 串行传输延迟 + 可能的线路干扰

通过虚拟串口进行测试，你可以提前验证你的主站代码在不同波特率（如9600, 19200, 115200）、不同校验方式（无校验、偶校验、奇校验）下的兼容性，这是确保代码能在各种现场设备上稳定运行的关键一步。

3. 异常场景模拟与故障注入测试

真实的工业环境远非理想状态。电压波动、线路老化、电磁干扰都可能导致通信错误。一个健壮的Modbus驱动或设备，必须能妥善处理这些异常，而不是直接崩溃。我们的测试环境应该具备**故障注入（Fault Injection）**的能力，主动制造异常，观察系统行为。

pymodbus库的服务器端提供了回调函数（slave_context）和自定义Framer的能力，我们可以利用这些机制来模拟各种异常。

3.1 模拟从站响应异常码

Modbus协议定义了标准的异常码，用于从站告知主站请求为何失败。例如，0x01表示非法功能码，0x02表示非法数据地址，0x03表示非法数据值。我们可以在从站代码中，有选择地对特定请求返回异常响应。

# modbus_slave_with_exceptions.py
from pymodbus.server import StartTcpServer
from pymodbus.datastore import ModbusSlaveContext, ModbusServerContext
from pymodbus.datastore import ModbusSequentialDataBlock
from pymodbus.pdu import ExceptionResponse
from pymodbus.exceptions import ModbusException

class FaultySlaveContext(ModbusSlaveContext):
    """自定义从站上下文，用于注入异常"""
    def validate(self, fc, address, count=1):
        """
        重写验证方法，在特定条件下返回异常
        fc: 功能码
        address: 请求的起始地址
        count: 请求的数据量
        """
        # 模拟场景1：访问不存在的寄存器地址（地址大于100）
        if address >= 100:
            print(f"[模拟异常] 请求地址 {address} 超出范围")
            return False  # 这将导致库返回非法地址异常(0x02)

        # 模拟场景2：对只读的输入寄存器(功能码04)执行写操作（通过功能码06或16请求）
        if self.decode(fc) == 'hr' and fc in [0x06, 0x10]: # 写保持寄存器请求
            # 但我们故意将地址50-60的区域标记为“只读”，模拟配置错误
            if 50 <= address <= 60:
                print(f"[模拟异常] 试图写入只读的保持寄存器区域 {address}")
                return False # 触发非法地址或非法数据值异常

        # 模拟场景3：非法的功能码（例如，使用0x08这种不常用或保留的功能码）
        if fc == 0x08: # 假设0x08是我们不支持的
            print(f"[模拟异常] 不支持的Modbus功能码: 0x{fc:02x}")
            # 这里我们可以选择直接返回一个ExceptionResponse
            # 但为了演示，我们通过返回False让底层处理
            return False

        # 其他情况正常处理
        return super().validate(fc, address, count)

def run_faulty_slave():
    """运行一个会随机返回异常的从站服务器"""
    store = FaultySlaveContext(
        di=ModbusSequentialDataBlock(0, [True]*100),
        co=ModbusSequentialDataBlock(0, [False]*100),
        hr=ModbusSequentialDataBlock(0, [i for i in range(100)]),
        ir=ModbusSequentialDataBlock(0, [i*2 for i in range(100)])
    )
    context = ModbusServerContext(slaves=store, single=True)

    print("启动带异常模拟的Modbus TCP从站 (地址: 1)")
    print("异常模拟策略：")
    print("  - 访问地址>=100的寄存器将返回非法地址异常")
    print("  - 写入地址50-60的保持寄存器将返回异常")
    print("  - 使用功能码0x08将返回非法功能码异常")
    StartTcpServer(context=context, address=("0.0.0.0", 5020)) # 使用不同端口避免冲突

if __name__ == "__main__":
    run_faulty_slave()

然后，我们编写一个主站测试脚本，专门去触发这些异常，并展示如何正确处理它们。

# test_exception_handling.py
from pymodbus.client import ModbusTcpClient
from pymodbus.exceptions import ModbusException

def test_exception_responses():
    client = ModbusTcpClient('127.0.0.1', port=5020)
    client.connect()

    test_cases = [
        ("正常读取", lambda: client.read_holding_registers(10, 5, slave=1)),
        ("访问非法地址(>100)", lambda: client.read_holding_registers(150, 2, slave=1)),
        ("写入只读区域(地址55)", lambda: client.write_register(55, 999, slave=1)),
        ("使用不支持的功能码(0x08)", lambda: client.execute(
            type('CustomRequest', (), {'function_code': 0x08, 'address': 0, 'count': 1})()
        )) # 这是一个简化的模拟，实际需构造完整请求
    ]

    for desc, req_func in test_cases:
        print(f"\n测试: {desc}")
        try:
            result = req_func()
            if hasattr(result, 'isError'):
                if result.isError():
                    # pymodbus会将异常响应封装为ModbusException
                    print(f"  收到从站异常响应")
                    # 可以通过result.exception_code获取具体异常码
                    if hasattr(result, 'exception_code'):
                        print(f"  异常码: 0x{result.exception_code:02x}")
                else:
                    print(f"  请求成功: {result.registers if hasattr(result, 'registers') else result}")
            else:
                print(f"  响应对象: {result}")
        except ModbusException as e:
            print(f"  捕获到ModbusException: {e}")
        except Exception as e:
            print(f"  捕获到其他异常: {type(e).__name__}: {e}")

    client.close()

if __name__ == "__main__":
    test_exception_responses()

运行这两个脚本，你会看到主站如何接收到从站返回的异常，并在代码中对其进行判断和处理。在实际项目中，你必须对每一次Modbus请求的响应进行isError()检查，并根据exception_code进行相应的日志记录、重试或告警，这是保证系统鲁棒性的基本要求。

3.2 模拟网络延迟与超时

工业网络可能不稳定。我们可以通过简单的time.sleep()在从站响应前插入延迟，来模拟网络拥堵或从站设备处理缓慢。

# 在自定义的SlaveContext中，添加处理钩子
class SlowSlaveContext(ModbusSlaveContext):
    def getValues(self, fc, address, count=1):
        # 模拟特定操作的延迟
        if fc == 0x03: # 读保持寄存器
            import time
            time.sleep(2.5) # 延迟2.5秒，超过常见超时时间
            print(f"[模拟延迟] 读保持寄存器延迟2.5秒")
        return super().getValues(fc, address, count)

在主站侧，你需要合理设置timeout参数，并实现重试机制。

from pymodbus.client import ModbusTcpClient
import time

def read_with_retry(client, address, count, retries=3, timeout=2):
    """带重试的读取函数"""
    for attempt in range(retries):
        try:
            result = client.read_holding_registers(address, count, slave=1)
            if not result.isError():
                return result
            else:
                print(f"  第{attempt+1}次尝试：从站返回错误")
        except Exception as e:
            print(f"  第{attempt+1}次尝试异常: {e}")

        if attempt < retries - 1:
            print(f"  等待{timeout}秒后重试...")
            time.sleep(timeout)
    print("  所有重试均失败")
    return None

3.3 模拟数据错乱与CRC校验错误

对于RTU模式，校验和（CRC）错误是常见的。我们可以在数据发送前，手动破坏报文中的某个字节，来模拟传输过程中的比特错误。

# 这是一个概念性示例，实际需要操作底层字节流
def corrupt_rtu_frame(original_frame):
    """
    模拟传输错误：随机修改RTU帧中的一个字节
    original_frame: bytes类型的原始RTU帧
    """
    import random
    if len(original_frame) < 3: # 地址+功能码+数据+CRC至少3字节
        return original_frame
    # 不破坏地址和功能码（前两字节），随机选一个数据字节进行修改
    corrupt_index = random.randint(2, len(original_frame)-3) # 避免改到CRC字节
    corrupted_byte = original_frame[corrupt_index] ^ 0x01 # 翻转最低位
    corrupted_frame = original_frame[:corrupt_index] + bytes([corrupted_byte]) + original_frame[corrupt_index+1:]
    print(f"[模拟错误] 将帧索引{corrupt_index}的字节从0x{original_frame[corrupt_index]:02x}改为0x{corrupted_byte:02x}")
    return corrupted_frame

在主站接收端，pymodbus会在解析RTU帧时自动计算CRC并与报文中的CRC字段比较，如果不匹配，会抛出ModbusIOException或返回一个错误响应。你的代码需要捕获这类异常。

4. 高级调试技巧与实战问题排查

搭建好测试环境并模拟了各种异常后，我们还需要一套方法来定位真实问题。下面分享几个我在项目中用到的调试技巧。

4.1 启用详细日志

pymodbus和Python的logging模块集成得很好。启用调试日志可以让你看到每一帧收发的原始字节，这对于排查协议层面的问题至关重要。

import logging
# 在主站或从站代码开头添加
logging.basicConfig()
log = logging.getLogger()
log.setLevel(logging.DEBUG) # 设置为DEBUG级别以查看帧数据

# 对于pymodbus客户端，还可以在初始化时传入调试参数
client = ModbusTcpClient('127.0.0.1', port=502, debug=True)

启用日志后，控制台会打印出类似下面的信息，你可以清晰地看到请求和响应的原始十六进制数据，方便与协议文档对照。

DEBUG:pymodbus.transaction:Current transaction state - IDLE
DEBUG:pymodbus.transaction:Running transaction 1
DEBUG:pymodbus.transaction:Send: 0x00 0x01 0x00 0x00 0x00 0x06 0x01 0x03 0x00 0x00 0x00 0x0A
DEBUG:pymodbus.transaction:Recv: 0x00 0x01 0x00 0x00 0x00 0x17 0x01 0x03 0x14 0x00 0x00 0x00 0x00 ...

4.2 使用网络抓包工具（TCP模式）

对于Modbus TCP，Wireshark是终极利器。它内置了Modbus/TCP协议解析器。

  • 过滤条件：在Wireshark中使用过滤器 tcp.port == 502 或 modbus。
  • 分析要点：
    • 事务标识符（Transaction Identifier）：检查请求和响应是否匹配。
    • 协议标识符（Protocol Identifier）：应为0（Modbus协议）。
    • 长度字段（Length）：指示后续字节数。
    • 单元标识符（Unit Identifier）：即从站地址。
    • 功能码与数据：核对功能码是否正确，请求与响应的数据长度、格式是否符合预期。

通过Wireshark，你可以确认问题出在发送端（请求帧格式错误）、网络传输（丢包、乱序）还是接收端（响应错误或无响应）。

4.3 串口监听与调试（RTU模式）

对于串行通信，你需要一个串口监听工具。如果使用虚拟串口，socat本身就可以作为中间人。更专业一点，可以使用Serial Port Monitor（Windows）或CuteCom、screen命令（Linux）来监听数据流。

一个简单的Python脚本也可以充当监听器：

# serial_sniffer.py
import serial
import time

def sniff_serial(port, baudrate=9600):
    """监听指定串口的数据流"""
    ser = serial.Serial(port, baudrate, timeout=1)
    print(f"开始监听串口 {port} ...")
    try:
        while True:
            if ser.in_waiting:
                data = ser.read(ser.in_waiting)
                print(f"[{time.strftime('%H:%M:%S')}] RX ({len(data)} bytes): {data.hex(' ').upper()}")
                # 尝试ASCII解码打印
                try:
                    ascii_repr = data.decode('ascii', errors='ignore')
                    if any(c.isprintable() for c in ascii_repr):
                        print(f"       ASCII: {ascii_repr}")
                except:
                    pass
            time.sleep(0.01)
    except KeyboardInterrupt:
        print("\n停止监听")
    finally:
        ser.close()

if __name__ == "__main__":
    # 监听虚拟串口的一端
    sniff_serial('/dev/ttyVCOM0', 9600)

运行这个监听器，然后启动你的RTU主从站程序，你就能看到线上流动的每一个字节，判断帧间隔（RTU的3.5字符时间）、起始/结束是否符合要求，以及CRC计算是否正确。

4.4 常见问题排查清单

当通信失败时，可以按照以下清单逐一排查：

  1. 物理连接/网络：

    • TCP：IP地址、端口是否正确？防火墙是否阻止了502端口？
    • RTU：串口号是否正确？波特率、数据位、停止位、校验位是否与从站设备完全一致？线缆是否完好？

  2. 从站地址：主站请求中的slave参数（或RTU帧中的地址字节）是否与从站配置的地址匹配？地址0通常是广播地址，从站不应响应。

  3. 功能码与数据地址：

    • 你使用的功能码（如0x03读保持寄存器）是否被从站支持？
    • 请求的寄存器/线圈地址是否在从站定义的地址范围内？注意Modbus地址有0起始和1起始两种惯例，容易混淆。pymodbus默认使用0起始地址。

  4. 数据格式（字节序/字序）：

    • Modbus协议规定寄存器中每个字（Word）为2字节，但多字数据（如32位整数、浮点数）的字节顺序（Endianness）和字顺序（Word Order） 没有统一规定。常见的有：
      • CDAB (Big-Endian Byte Swap): 也称为“Modbus”顺序。
      • ABCD (Big-Endian): 也称为“标准”网络字节序。
      • BADC (Little-Endian Byte Swap)。
      • DCBA (Little-Endian)。
    • 这是最常见的坑之一！务必查阅设备手册，确认其使用的格式。pymodbus的payload模块提供了BinaryPayloadDecoder和BinaryPayloadBuilder来处理不同顺序。
    from pymodbus.payload import BinaryPayloadDecoder
from pymodbus.constants import Endian

# 假设从寄存器地址0读取了两个寄存器（4字节），代表一个32位整数
result = client.read_holding_registers(0, 2, slave=1)
registers = result.registers # 例如 [0x1234, 0x5678]

# 解码，需要根据设备手册选择正确的字节序和字序
# 常见组合：Endian.Big, Endian.Little, wordorder=Endian.Big, wordorder=Endian.Little
decoder = BinaryPayloadDecoder.fromRegisters(registers, byteorder=Endian.Big, wordorder=Endian.Big)
value_as_int = decoder.decode_32bit_int() # 以32位有符号整数解码
print(f"解码后的32位整数 (Big-Endian, Big Wordorder): {value_as_int}")

  5. 超时与重试：是否设置了合理的超时时间？是否实现了重试逻辑？对于不稳定的网络，指数退避重试策略会更有效。

  6. 并发与资源管理：确保正确关闭客户端连接（client.close()），避免资源泄漏。在高并发场景下，考虑使用连接池或异步客户端（AsyncModbusTcpClient）。

搭建这样一个涵盖TCP/RTU、支持异常注入、并配备高级调试工具的本地测试环境，初期会花一些时间，但它带来的回报是巨大的。它让你能在舒适的开发环境中，以可控、可重复的方式验证代码的健壮性，模拟出各种现场可能遇到的极端情况。当你的代码通过了这个“魔鬼测试营”的考验，再部署到真实的工业现场时，你会更有信心，调试效率也会成倍提升。毕竟，在电脑前喝着咖啡解决bug，远比在嘈杂的车间里抱着笔记本排查要舒服得多。

Logo
AI Agent技术社区

Agent 垂直技术社区，欢迎活跃、内容共建。

更多推荐

cover

从「能说」到「能做」：一份理清 AI Agent 的学习笔记

avatar AI Agent技术社区
cover

DeepSeek Harness 深度解析：一切皆插件的开源 AI Agent 框架如何运行？

avatar AI Agent技术社区
cover

2026年，集团企业想要AI落地，选RPA还是Agent？

avatar AI Agent技术社区