Python搭建Modbus模拟环境遇连接错误，寻求技术帮助

你的服务器启动逻辑存在致命错误，同时客户端缺少必要的连接校验和动态读取逻辑，这是导致连接失败及数据异常的核心原因。以下是完整修复方案：

修复后的服务器代码 #

import random
from pymodbus.datastore import ModbusSequentialDataBlock, ModbusSlaveContext, ModbusServerContext
from pymodbus.server.async_io import StartTcpServer

# PyModbus数据块采用0基地址，这里创建100个寄存器，对应Modbus协议的1-100地址
coils = ModbusSequentialDataBlock(0, [False] * 100)
discrete_inputs = ModbusSequentialDataBlock(0, [False] * 100)
holding_registers = ModbusSequentialDataBlock(0, [0] * 100)
input_registers = ModbusSequentialDataBlock(0, [0] * 100)

# 初始化温度和流量数据
temperature_values = [random.uniform(4, 15) for _ in range(7)]
flow_values = [random.uniform(-1, 1) for _ in range(2)]

# 写入到0基地址的0-6（对应Modbus 1-7）和10-11（对应Modbus 11-12）
holding_registers.setValues(0, temperature_values)
holding_registers.setValues(10, flow_values)

# 创建Slave上下文，开启zero_mode统一地址模式
slave_context = ModbusSlaveContext(
    di=discrete_inputs,
    co=coils,
    hr=holding_registers,
    ir=input_registers,
    zero_mode=True
)

server_context = ModbusServerContext(slaves=slave_context, single=True)

if __name__ == "__main__":
    # 正确启动TCP服务器，绑定本地502端口
    StartTcpServer(context=server_context, address=("localhost", 502))

修复后的客户端代码 #

from pymodbus.client.tcp import ModbusTcpClient
from time import sleep

# 初始化客户端
client = ModbusTcpClient('localhost', port=502)

# 先校验连接状态
if not client.connect():
    print("无法连接到Modbus服务器，请检查服务器是否启动")
    exit(1)

try:
    while True:
        # 读取Modbus 1基地址的寄存器（对应服务器0基的0-99）
        coils = client.read_coils(address=1, count=100)
        discrete_inputs = client.read_discrete_inputs(address=1, count=100)
        holding_registers = client.read_holding_registers(address=1, count=100)
        input_registers = client.read_input_registers(address=1, count=100)

        # 校验读取结果，避免异常崩溃
        if not coils.isError():
            print("Coils:", coils.bits)
        else:
            print("读取Coils失败:", coils)
        
        if not discrete_inputs.isError():
            print("Discrete Inputs:", discrete_inputs.bits)
        else:
            print("读取Discrete Inputs失败:", discrete_inputs)
        
        if not holding_registers.isError():
            print("Holding Registers:", holding_registers.registers)
        else:
            print("读取Holding Registers失败:", holding_registers)
        
        if not input_registers.isError():
            print("Input Registers:", input_registers.registers)
        else:
            print("读取Input Registers失败:", input_registers)
        
        print("-"*50)
        sleep(5)
finally:
    # 确保程序退出时关闭连接
    client.close()

核心问题说明 #

• 服务器启动错误：原代码同时实例化ModbusTcpServer和调用无参StartTcpServer()，导致服务器未正确绑定上下文和端口，根本没有启动成功。
• 地址模式混淆：PyModbus的数据块默认是0基，但SlaveContext默认使用1基地址，通过zero_mode=True统一地址规则，避免写入/读取地址偏移。
• 客户端连接缺失：原代码未调用client.connect()就直接执行读取操作，必然会失败。
• 静态读取失效：原客户端仅在循环前读取一次数据，循环内重复打印旧值，修复后每次循环重新读取最新数据。
• 异常处理缺失：添加isError()判断，捕获读取失败的情况，避免程序直接崩溃。

内容的提问来源于stack exchange，提问作者rasmusdamlarsen