如何通过 Modbus TCP 使用“struct.pack”类型发送？

Question

我想通过 Modbus TCP 发送数据包。我想使用：但是我不能以这种方式发送我如何发送这个数据包？（我不知道会发生什么）req = struct.pack(   'Something', transaction, identifier, length, unitid, func_code, reg_addr) 这些是我的变量：transaction=0x01identifier=0x00length=[0x00,0x06]unitid=0x01func_code=0x03reg_addr=[0x13,0x14,0x15]

Answer 1

• 首先，您可以使用具有很多功能的pymodbus库。

• 另外struct.pack()不支持列表作为参数。

• 0001 0000 0006 11 03 006B 0003 是 Modbus-TCP 数据包的标准示例，其中包含：

0001: Transaction Identifier

0000: Protocol Identifier

0006: Message Length (6 bytes to follow)

11: The Unit Identifier  (17 = 11 hex)

03: The Function Code (read Analog Output Holding Registers)

006B: The Data Address of the first register requested. (40108-40001 = 107 =6B hex)

0003: The total number of registers requested. (read 3 registers 40108 to 40110)

_参考

因此，您可以使用上述示例创建Modbus-TCP数据包：

import struct

transaction = 0x0001

identifier = 0x0000

length = 0x0006

unitid = 0x11

fcode = 0x03  # Holding register fcode.

reg_addr = 0x006B  # Register address.

count = 0x0003  # Read three register.

total_pack_string = '0x{:04x}{:04x}{:04x}{:02x}{:02x}{:04x}{:04x}'.format(

    transaction, identifier, length, unitid, fcode, reg_addr, count

)

total_pack_hex = hex(int(total_pack_string, 16))

'''Or with using pack method.'''

pack_ = struct.pack(

   '>HHHBBHH', transaction, identifier, length, unitid, fcode, reg_addr, count

)

# Then send the pack_ or total_pack_hex using a TCP-Socket.

[注意]：

• transaction 是 2Byte == 短 == H

• identifier 是 2Byte == 短 == H

• length 是 2Byte == 短 == H

• unitid 是 1Byte == B

• fcode 是 1Byte == B

• reg_addr 是 2Byte == 短 == H

• count 是 2Byte == 短 == H

• B 是 unsigned byte

• H 是 unsigned short

因此，格式将是这样的 >HHHBBHH

---

使用pymodbus等效项：

from pymodbus.client.sync import ModbusTcpClient

unitid = 0x11

fcode = 0x03  # Holding register fcode.

reg_addr = 0x006B  # Register address.

count = 0x0003  # Read three register.

cli = ModbusTcpClient('127.0.0.1', port=502)

if cli.connect():

    res = cli.read_holding_registers(reg_addr, count=count, unit=unitid)

    

    if not res.isError():

        print(res.registers)

    

    else:

        print('There is an error.')

    

    cli.close()

else:

    print('Error in connection.')