文章详情

短信预约-IT技能 免费直播动态提醒

请输入下面的图形验证码

提交验证

短信预约提醒成功

python使用pymodbus库进行modbus tcp通信

2023-09-04 09:38

关注

使用python解决工业通信问题是一个非常好的选择,python具有丰富的生态,可以轻松解决工业通信的各种问题。本篇主要介绍使用pymodbus库进行modbus tcp仿真,实现pc端读取plc或工业设备modbus变量。

安装pymodbus:

pip install -U pymodbus

1 创建modbus tcp server

这里我们先创建一个虚拟的modbus设备,如果你手里有一个plc或者工业设备,可以直接跳过本节。

modbus_server.py

''' * @Author: liuzhao  * @Last Modified time: 2022-10-05 09:56:13 '''from pymodbus.server.sync import (    StartTcpServer,)from pymodbus.datastore import (    ModbusSequentialDataBlock,    ModbusServerContext,    ModbusSlaveContext,)from pymodbus.version import versiondatablock = ModbusSequentialDataBlock.create()context = ModbusSlaveContext(    di=datablock,    co=datablock,    hr=datablock,    ir=datablock,    )single = True# Build data storagestore = ModbusServerContext(slaves=context, single=single)if __name__ == '__main__':address = ("0.0.0.0", 503)StartTcpServer(    context=store,  # Data storage    address=address,  # listen address  allow_reuse_address=True,  # allow the reuse of an address)

直接运行该脚本,就可以在本机的503端口创建一台modbus设备了,具体实现暂不深追,我们学习的重点是客户端对modbus变量的读写。

2 读写modbus变量

modbus变量类型以及地址

Object typeAccessSizeAddress
CoilRead-write1 bit00001 – 09999
Discrete inputRead-only1 bit10001 – 19999
Input registerRead-only16 bits30001 – 39999
Holding registerRead-write16 bits40001 – 49999

coil是线圈,Discrete input是数字量输入,Input register是模拟量输入,Holding register是保持寄存器。一般地址范围是0-65535 [wiki]

2.1 读取常规变量

读写线圈 | 读取输入变量 | 读写保持寄存器

from pymodbus.client.sync import ModbusTcpClientfrom pymodbus.bit_read_message import ReadCoilsResponsefrom pymodbus.register_read_message import ReadInputRegistersResponsefrom pymodbus.exceptions import ConnectionException      # 连接失败,用于异常处理host = '127.0.0.1'port = 503client = ModbusTcpClient(host,port)# 写入线圈client.write_coil(1, True)client.write_coil(2, False)client.write_coil(3, True)# 读取线圈    注意对于离散量的读取,第二个参数cout是有坑的,必须为8的倍数个result:ReadCoilsResponse = client.read_coils(address=1,cout=8)     # 从地址1开始读,读取8个线圈,一次读8的倍数个线圈,不设置为8的倍数可能会出现问题print(result.isError())# 不建议使用print(result.getBit(7))            # 这里的参数address不是plc里的地址,而是python列表的address,print('read_coils ')# 建议使用print(result.bits)        # 打印读取结果,一共8位# 读取其中的位print(                       result.bits[0],    result.bits[1],    result.bits[2]    )         # 相当于result.getBit(0)# 读取数字输入result = client.read_discrete_inputs(address=10001,count=8)    # 从10001开始读,读取8位print(result.bits)# 读取模拟输入寄存器input_register_result:ReadInputRegistersResponse = client.read_input_registers(1,count=8)# print(f'is_error:{input_register_result.isError()}')print('read_input_registers ')print(input_register_result.registers)   print(input_register_result.getRegister(0))   # 读写保持寄存器client.write_register(address=40001,value=100)result:ReadInputRegistersResponse = client.read_holding_registers(address=40001,count=1)print('read_holding_registers ')print(result.registers)# 关闭连接client.close()

2.2 读取复杂变量

字符串、浮点数、负数等
这里需要注意modbus设备的存储结构是低位低字节还是低位高字节,也就是设备内存的字节、字的排列顺序。

  根据不同的设备,对照下表调整正确的组合方式。
Word OrderByte orderWord1Word2
BigBig0x12340x5678
BigLittle0x34120x7856
LittleBig0x56780x1234
LittleLittle0x78560x3412
# 复杂数据类型from collections import OrderedDictimport loggingfrom pymodbus.client.sync import ModbusTcpClient as ModbusClientfrom pymodbus.constants import Endianfrom pymodbus.payload import BinaryPayloadBuilder, BinaryPayloadDecoderORDER_DICT = {"<": "LITTLE", ">": "BIG"}def run_binary_payload_client(host:str,port:int):      for word_endian, byte_endian in (        (Endian.Big, Endian.Big),        (Endian.Big, Endian.Little),        (Endian.Little, Endian.Big),        (Endian.Little, Endian.Little),    ):        print("-" * 60)        print(f"Word Order: {ORDER_DICT[word_endian]}")        print(f"Byte Order: {ORDER_DICT[byte_endian]}")        print()            builder = BinaryPayloadBuilder(            wordorder=word_endian,            byteorder=byte_endian,        )# 写入的变量        my_string = "abcd-efgh123345765432"        builder.add_string(my_string)        builder.add_bits([0, 1, 0, 1, 1, 0, 1, 0])        builder.add_8bit_int(-0x12)        builder.add_8bit_uint(0x12)        builder.add_16bit_int(-0x5678)        builder.add_16bit_uint(0x1234)        builder.add_32bit_int(-0x1234)        builder.add_32bit_uint(0x12345678)        builder.add_16bit_float(12.34)        builder.add_16bit_float(-12.34)        builder.add_32bit_float(22.34)        builder.add_32bit_float(-22.34)        builder.add_64bit_int(-0xDEADBEEF)        builder.add_64bit_uint(0x12345678DEADBEEF)        builder.add_64bit_uint(0x12345678DEADBEEF)        builder.add_64bit_float(123.45)        builder.add_64bit_float(-123.45)        registers = builder.to_registers()        print("Writing Registers:")        print(registers)        print("\n")        payload = builder.build()        address = 40001          # 从40001开始写入        # We can write registers        client.write_registers(address, registers, unit=1)    # 写入          # 读取复杂变量        print("Reading Registers:")        address = 40001        count = len(payload)        print(f"payload_len {count}")        result = client.read_holding_registers(address, count, slave=1)        print(result.registers)        print("\n")        decoder = BinaryPayloadDecoder.fromRegisters(            result.registers, byteorder=byte_endian, wordorder=word_endian        )        # Make sure word/byte order is consistent between BinaryPayloadBuilder and BinaryPayloadDecoder        assert (            decoder._byteorder == builder._byteorder  # pylint: disable=protected-access        )  # nosec        assert (            decoder._wordorder == builder._wordorder  # pylint: disable=protected-access        )  # nosec        decoded = OrderedDict(            [                ("string", decoder.decode_string(len(my_string))),                ("bits", decoder.decode_bits()),                ("8int", decoder.decode_8bit_int()),                ("8uint", decoder.decode_8bit_uint()),                ("16int", decoder.decode_16bit_int()),                ("16uint", decoder.decode_16bit_uint()),                ("32int", decoder.decode_32bit_int()),                ("32uint", decoder.decode_32bit_uint()),                ("16float", decoder.decode_16bit_float()),                ("16float2", decoder.decode_16bit_float()),                ("32float", decoder.decode_32bit_float()),                ("32float2", decoder.decode_32bit_float()),                ("64int", decoder.decode_64bit_int()),                ("64uint", decoder.decode_64bit_uint()),                ("ignore", decoder.skip_bytes(8)),                ("64float", decoder.decode_64bit_float()),                ("64float2", decoder.decode_64bit_float()),            ]        )        print("Decoded Data")        for name, value in iter(decoded.items()):            print(                "%s\t" % name,  # pylint: disable=consider-using-f-string                hex(value) if isinstance(value, int) else value,            )        print("\n")# 关闭连接    client.close()if __name__ == "__main__":    run_binary_payload_client("127.0.0.1", 503)    

相关资料

modbus tcp 通信协议
2、modbus tcp 通信协议2
3、pymodbus官方文档
4、pymodbus仓库

欢迎转载

来源地址:https://blog.csdn.net/brandon_l/article/details/127169641

阅读原文内容投诉

免责声明:

① 本站未注明“稿件来源”的信息均来自网络整理。其文字、图片和音视频稿件的所属权归原作者所有。本站收集整理出于非商业性的教育和科研之目的,并不意味着本站赞同其观点或证实其内容的真实性。仅作为临时的测试数据,供内部测试之用。本站并未授权任何人以任何方式主动获取本站任何信息。

② 本站未注明“稿件来源”的临时测试数据将在测试完成后最终做删除处理。有问题或投稿请发送至: 邮箱/279061341@qq.com QQ/279061341

软考中级精品资料免费领

  • 历年真题答案解析
  • 备考技巧名师总结
  • 高频考点精准押题
  • 2024年上半年信息系统项目管理师第二批次真题及答案解析(完整版)

    难度     813人已做
    查看
  • 【考后总结】2024年5月26日信息系统项目管理师第2批次考情分析

    难度     354人已做
    查看
  • 【考后总结】2024年5月25日信息系统项目管理师第1批次考情分析

    难度     318人已做
    查看
  • 2024年上半年软考高项第一、二批次真题考点汇总(完整版)

    难度     435人已做
    查看
  • 2024年上半年系统架构设计师考试综合知识真题

    难度     224人已做
    查看

相关文章

发现更多好内容

猜你喜欢

AI推送时光机
位置:首页-资讯-后端开发
咦!没有更多了?去看看其它编程学习网 内容吧
首页课程
资料下载
问答资讯