python如何用pymodbus库进行modbus tcp通信

pip install -u pymodbus

'''
 * @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 version

datablock = modbussequentialdatablock.create()
context = modbusslavecontext(
    di=datablock,
    co=datablock,
    hr=datablock,
    ir=datablock,
    )
single = true

# build data storage
store = 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
	)

from pymodbus.client.sync import modbustcpclient
from pymodbus.bit_read_message import readcoilsresponse
from pymodbus.register_read_message import readinputregistersresponse
from pymodbus.exceptions import connectionexception      # 连接失败，用于异常处理

host = '127.0.0.1'
port = 503
client = 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()

# 复杂数据类型

from collections import ordereddict
import logging

from pymodbus.client.sync import modbustcpclient as modbusclient

from pymodbus.constants import endian
from pymodbus.payload import binarypayloadbuilder, binarypayloaddecoder



order_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)