python封装modbus-tcp协议，实现03,06，功能码

 
import socket
import struct
 
class ModbusTcp:
    def __init__(self,ip:str="192.168.0.20",port:int = 502):
        self.ip = ip
        self.port = port
        self.socket = socket.socket(socket.AF_INET,socket.SOCK_STREAM)
        self.socket.settimeout(0.5)
        self.isSucess = None
 
    def connect(self):
        try:
            self.socket.connect((self.ip,self.port))
        except:
            self.isSucess = False
            raise Exception("连接失败")
        else:
            print("连接成功")
            self.isSucess = True
    def read03FunCode(self,addr:int,numbers:int=1)->bytearray:
 
        _read_byarry = bytearray([
            0x00, 0x01,
            0x00, 0x00,
            0x00, 0x06,
            0x01,
            0x03,
            addr>>8 & 0xff,
            addr&0xff,
            numbers>>8 & 0xff,
            numbers & 0xff
        ])
        if self.isSucess:
            try:
                self.socket.send(_read_byarry)
            except:
                raise Exception("读取失败")
            else:
                bytes=self.socket.recv(1024)
                return struct.unpack(f">{numbers}H",bytearray(bytes[9:]))
 
    def write06FunCode(self,addr, value:int):
        bytes = bytearray([
            0x00, 0x01,
            0x00, 0x00,
            0x00, 0x06,  # 0x06代表后面有6 个字节
            0x01,  # MbPA
 
            0x06,  # 代表06功能码
            addr>>8 & 0xff, #起始地址高字节
            addr & 0xff,  # 起始地址低字节
            value>>8 & 0xdd,  #写入数据高字节
            value & 0xff  # 写入数据低字节
 
        ])
 
        self.socket.send(bytes)
 
        recv_data = self.socket.recv(12)
        return struct.unpack(">H",recv_data[10:12])[0]
 
    def write16Funcode(self,start_addr,values:list):
        length = len(values)*2+7
        #报文头
        MBAP = bytearray([
            0x00,0x01,
            0x00,0x00,
            length>>8 & 0xff, #写入的字节数量高位
            length & 0xff,    #写入的字节数量低位
            0x01             #单元标识
        ])
 
        PDU = bytearray(
            [
                0x10,  #功能码
                start_addr>>8 & 0xff,  #起始地址高位字节
                start_addr & 0xff,     #起始地址低位字节
                len(values)>>8 & 0xff, #写入的数量高字节
                len(values) & 0xff,    #写入的数量低字节
                len(values)*2          #字节数
            ]
        )
        # print(pdu)
        for value in values:
            PDU.extend([(value>>8)&0xff,value & 0xff])
 
        request = MBAP+PDU
        if self.isSucess:
            self.socket.send(request)
            data_recv= self.socket.recv(1024)
 
            return data_recv                      #返回报文
 
        else:
            raise Exception("没有连接成功")
 
 
 
    def disConnect(self):
        self.socket.close()
 
 
 
 
 

测试正常。在写modbus协议时要注意每个功能码的格式要求是不一样的，返回的格式意义也不太一样。要注意。每个功能码发送报文和返回报文参考如下，可以按照格式自行实现功能。