当前位置：首页 > news >正文

Python - Modbus测试

news 2025/1/17 1:02:47

python -Modbus测试

前言

本实例中，python版本是python 3.10, 使用的pymodbus安装包进行实现，如没有按照此包，需要先执行如下指令：

pip install pymodbus

Modbus客户端的python实现

Modbus Client的客户端程序，通过如下代码实现：

from pymodbus.client import ModbusTcpClient# 创建客户端实例
client = ModbusTcpClient('localhost', port=5020)# 连接到服务器
client.connect()# 读取线圈的状态
result = client.read_coils(0, 10)
if not result.isError():print(result.bits)# 读取离散输入
result = client.read_discrete_inputs(0, 10)
if not result.isError():print(result.bits)# 读取保持寄存器
result = client.read_holding_registers(0, 10)
if not result.isError():print(result.registers)# 读取输入寄存器
result = client.read_input_registers(0, 10)
if not result.isError():print(result.registers)# 断开连接
client.close()

Modbus服务端的python实现

Modbus Server 的代码实现如下

from pymodbus.server import StartTcpServer
from pymodbus.device import ModbusDeviceIdentification
from pymodbus.datastore import ModbusSequentialDataBlock, ModbusSlaveContext, ModbusServerContext
from pymodbus.transaction import ModbusRtuFramer# 创建一个自动递增的寄存器值
def create_auto_increment_block(start_value, size):return ModbusSequentialDataBlock(0, [start_value + i for i in range(size)])# 初始化数据存储
store = ModbusSlaveContext(di=ModbusSequentialDataBlock(0, [0]*100),  # 离散输入co=ModbusSequentialDataBlock(0, [1]*100),  # 线圈#hr=ModbusSequentialDataBlock(0, [17]*100),  # 持续寄存器hr = create_auto_increment_block(0,100),ir=ModbusSequentialDataBlock(0, [5]*100)   # 输入寄存器
)
context = ModbusServerContext(slaves=store, single=True)# 设置身份信息
identity = ModbusDeviceIdentification()
identity.VendorName = 'Pymodbus'
identity.ProductCode = 'PM'
identity.VendorUrl = 'http://github.com/riptideio/pymodbus/'
identity.ProductName = 'Pymodbus Server'
identity.ModelName = 'Pymodbus Server'
identity.MajorMinorRevision = '3.5.0'#设置一个回调函数来处理写操作
def update_hr_write_request(address,value):print(f"持续寄存器{address}接收到写入请求，值为: {value}")def update_ir_write_request(value):print(f"输入寄存器接收到写入请求，值为: {value}")# 获取保持寄存器的数据块
holding_registers = store.getValues(3,0,count=1)
# 由于 getValues 返回的是一个列表，我们需要将回调绑定到数据块本身
print("Initial holding register values:", holding_registers)# 启动服务器
StartTcpServer(context=context, identity=identity, address=("localhost", 5020))
server.serve_forever()