如何使用Pymodbus快速查找符合条件的Modbus从设备？

问题背景 #

我有一台通过计算机COM口连接的Modbus主设备，最多可接入40台同类型从设备。需要使用Pymodbus筛选出符合条件的从设备：向寄存器976发起读请求，返回值为60的即为目标设备。

当前采用遍历1-249地址的检测方式，但无设备响应的地址需等待约1秒，整体耗时过长，希望找到更高效的预获取从设备地址的方法。

当前实现代码 #

from pymodbus.client.sync import ModbusSerialClient as ModbusClient

def get_slaves(port):
    client = ModbusClient(method='rtu', port=port, stopbits=2, bytesize=8, parity='N', baudrate=19200)
    client.connect()
    slaves = []
    for i in range(1, 250):
        try:
            response  = client.read_holding_registers(address=976, count=1, unit=i)
            registers_values = response.registers
            print(registers_values)
            # if registers_values[0] == 60:
            slaves.append(i)
            print(slaves)
        except:
            print('ошибка')
            continue
    print(slaves)

优化方案 #

1. 缩短超时时间+缩小遍历范围 #

默认超时时间通常较长，初始化客户端时手动设置timeout参数（比如0.1-0.5秒），让无响应的地址快速跳过；同时结合设备接入上限（最多40台），缩小遍历范围到1-50，直接减少循环次数。

优化后代码：

from pymodbus.client.sync import ModbusSerialClient as ModbusClient

def get_slaves(port):
    # 自定义超时时间为0.2秒，缩小遍历范围到1-50
    client = ModbusClient(method='rtu', port=port, stopbits=2, bytesize=8, parity='N', baudrate=19200, timeout=0.2)
    client.connect()
    slaves = []
    for i in range(1, 51):
        try:
            response = client.read_holding_registers(address=976, count=1, unit=i)
            # 检查响应是否正常
            if response.isError():
                continue
            registers_values = response.registers
            if registers_values[0] == 60:
                slaves.append(i)
                print(f"找到目标设备：地址{i}")
        except Exception:
            continue
    client.close()  # 记得关闭串口连接
    print(f"所有目标设备地址：{slaves}")
    return slaves

2. 异步并行查询 #

使用Pymodbus的异步客户端，同时发起多个设备地址的查询请求，利用异步IO特性大幅减少总耗时，适合批量检测场景。

异步实现代码：

import asyncio
from pymodbus.client import AsyncModbusSerialClient

async def check_single_slave(client, slave_id):
    try:
        response = await client.read_holding_registers(address=976, count=1, unit=slave_id)
        if not response.isError() and response.registers[0] == 60:
            return slave_id
    except Exception:
        pass
    return None

async def get_slaves_async(port):
    client = AsyncModbusSerialClient(method='rtu', port=port, stopbits=2, bytesize=8, parity='N', baudrate=19200, timeout=0.2)
    await client.connect()
    # 生成所有检测任务
    tasks = [check_single_slave(client, i) for i in range(1, 51)]
    # 并发执行所有任务
    results = await asyncio.gather(*tasks)
    # 过滤出有效设备地址
    slaves = [res for res in results if res is not None]
    await client.close()
    print(f"所有目标设备地址：{slaves}")
    return slaves

# 运行异步函数
if __name__ == "__main__":
    asyncio.run(get_slaves_async("COM3"))  # 替换为实际串口

3. 谨慎使用广播查询 #

Modbus RTU支持广播地址0，发送请求到地址0时所有从设备都会接收，但大部分设备不会响应广播读请求——因为多个设备同时回包会导致总线冲突。如果设备手册明确支持广播读响应，可以尝试，但不推荐作为通用方案。

注意事项 #

• 超时时间需根据实际总线环境调整：太短可能漏检延迟设备，太长则失去优化意义。
• 并行查询时避免一次性发起过多请求：建议控制并发数或分批次检测，防止总线拥堵。
• 优先缩小遍历范围：如果知道设备可能的地址区间（比如出厂默认地址段），直接遍历该区间而非全范围。

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