在python中写个自定义数据包协议的打包和解包测试

目录

• 写个自定义数据包协议的打包和解包测试
• 打包代码
• 解包代码
• 总结

写个自定义数据包协议的打包和解包测试

打包代码

'''
帧数据结构创建
'''
import struct
class FrameData():
    def __init__(self,pixStyle=0x14,width=1,height=1):
        self.pixStyle = pixStyle
        self.frameWidth = width
        self.frameHeight = height
        self.pkgLen =self.frameWidth *self.frameHeight *(pixStyle & 0x0f)+6
        self.buf = [0 for i in range(self.pkgLen)]
        #数据头尾定义
        self.buf[0]=0xff
        self.buf[1] = self.pixStyle
        self.buf[2] = self.frameWidth
        #补丁
        if(width>255):
            self.buf[1]=0x40+(pixStyle & 0x0f)
            self.buf[2] = 128
        self.buf[3] = self.frameHeight
        self.buf[self.pkgLen - 2] = (self.pixStyle+self.frameWidth+self.frameHeight)&0xff
        self.buf[self.pkgLen - 1]=0xfe
    def setCrc(self):
        r=0
        for i in range(1, self.pkgLen-2):
            r+=self.buf[i]
        self.buf[self.pkgLen - 2] = r& 0xff
    def setDataToArray(self,arr):
        alen= len(arr)
        dlen= self.pkgLen - 6
        rlen=alen if(alen<dlen) else dlen
        for i in range(4, rlen+4):
            #print(i,arr[i-4])
            self.buf[i]=arr[i-4]
        self.setCrc()
    def setDataToOff(self):
        for i in range(4, self.pkgLen - 2):
            self.buf[i]=0
        self.buf[self.pkgLen - 2]=(self.pixStyle + self.frameWidth + self.frameHeight) & 0xff
        #self.setCrc()
    def setDataToOn(self):
        for i in range(4, self.pkgLen - 2):
            self.buf[i]=255
        self.setCrc()
    def setDataToRGBW(self,r=0,g=0,b=0,w=0):
        sty=self.pixStyle & 0x0f
        if (sty == 1):
            for i in range(0, int((self.pkgLen - 6) / sty)):
                self.buf[i * sty + 4] = r
        if (sty == 2):
            for i in range(0, int((self.pkgLen - 6) / sty)):
                self.buf[i * sty + 4] = r
                self.buf[i * sty + 5] = g
        if(sty==3):
            for i in range(0, int((self.pkgLen - 6)/sty)):
                self.buf[i * sty + 4] = r
                self.buf[i * sty + 5] = g
                self.buf[i * sty + 6] = b
        if (sty == 4):
            for i in range(0, int((self.pkgLen - 6) / sty)):
                self.buf[i * sty + 4] = r
                self.buf[i * sty + 5] = g
                self.buf[i * sty + 6] = b
                self.buf[i * sty + 7] = w
        self.setCrc()
    def packBytes(self):
        packstyle = str( self.pkgLen) + 'B'  # B 0-255     b 0-127
        req = struct.pack(packstyle, *self.buf)
        return req
if __name__ == '__main__':
    d=FrameData(0x14,10,2)
    print("init",d.buf)
    d.setDataToOn()
    print("on",d.buf)
    d.setDataToRGBW(255)
    print("r",d.buf)
    d.setDataToRGBW(0,255)
    print("g",d.buf)
    d.setDataToRGBW(0,0,255)
    print("b",d.buf)
    d.setDataToRGBW(0,0,0,255)
    print("w",d.buf)
    d.setDataToRGBW(255,255, 255, 255)
    print("rgbw", d.buf)
    d.setDataToArray([0,0,0])
    print("000", d.buf)
    import time
    while True:
        time.sleep(1)

输出

init [255, 20, 10, 2, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 编程客栈0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 32, 254]

on [255, 20, 10, 2, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 208, 254]

r [255, 20, 10, 2, 255, 0, 0, 0, 255, 0, 0, 0, 255, 0, 0, 0, 255, 0, 0, 0, 255, 0, 0, 0, 255, 0, 0, 0, 255, 0, 0, 0, 255, 0, 0, 0, 255, 0, 0, 0, 255, 0, 0, 0, 255, 0, 0, 0, 255, 0, 0, 0, 255, 0, 0, 0, 255, 0, 0, 0, 255, 0, 0, 0, 255, 0, 0, 0, 255, 0, 0, 0, 255, 0, 0, 0, 255, 0, 0, 0, 255, 0, 0, 0, 12, 254]

g [255, 20, 10, 2, 0, 255, 0, 0, 0, 255, 0, 0, 0, 255, 0, 0, 0, 255, 0, 0, 0, 255, 0, 0, 0, 255, 0, 0, 0, 255, 0, 0, 0, 255, 0, 0, 0, 255, 0, 0, 0, 255, 0, 0, 0, 255, 0, 0, 0, 255, 0, 0, 0, 255, 0, 0, 0, 255, 0, 0, 0, 255, 0, 0, 0, 255, 0, 0, 0, 255, 0, 0, 0, 255, 0, 0, 0, 255, 0, 0, 0, 255, 0, 0, 12, 254]

b [255, 20, 10, 2, 0, 0, 255, 0, 0, 0, 255, 0, 0, 0, 255, 0, 0, 0, 255, 0, 0, 0, 255, 0, 0, 0, 255, 0, 0, 0, 255, 0, 0, 0, 255, 0, 0, 0, 255, 0, 0, 0, 255, 0, 0, 0, 255, 0, 0, 0, 255, 0, 0, 0, 255, 0, 0, 0, 255, 0, 0, 0, 255, 0, 0, 0, 255, 0, 0, 0, 255, 0, 0, 0, 255, 0, 0, 0, 255, 0, 0, 0, 255, 0, 12, 254]

w [255, 20, 10, 2, 0, 0, 0, 255, 0, 0, 0, 255, 0, 0, 0, 255, 0, 0, 0, 255, 0, 0, 0, 255, 0, 0, 0, 255, 0, 0, 0, 255, 0, 0, 0, 255, 0, 0, 0, 255, 0, 0, 0, 255, 0, 0, 0, 255, 0, 0, 0, 255, 0, 0, 0, 255, 0, 0, 0, 255, 0, 0, 0, 255, 0, 0, 0, 255, 0, 0, 0, 255, 0, 0, 0, 255, 0, 0, 0, 255, 0, 0, 0, 255, 12, 254]

rgbw [255, 20, 10, 2, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 208, 254]

000 [255, 20, 10, 2, 0, 0, 0, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 25android5, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 211, 254]

解包代码

'''
帧数据结构解析
'''
from event import *
class FramePkg(EventDispatcher):
    def __init__(self,pixStyle=0x14,width=1,height=1):
        EventDispatcher.__init__(self)
        self.pixStyle = pixStyle
        self.frameWidth = width
        self.frameHeight = height
        self.pkgLen =self.frameWidth *self.frameHeight *(pixStyle & 0x0f)+6
        self.bufLen = self.pkgLen * 3
        self.buf = [0 for i in range(self.bufLen)]
        self.dataLen = self.pkgLen - 6
        self.linedata = [0 for i in range(self.dataLen)]
        self.startIdx = 0
        self.endIdx = 0
    #获取数据长度 int
    def getDatLen(self):
        if(self.endIdx>=self.startIdx):
            return self.endIdx-self.startIdx
        return self.bufLen+self.endIdx-self.startIdx
    #判断数据长度已经到达一个包长bool
    def pkgIsOk(self):
        return self.getDatLen()>=self.pkgLen
    #检测包头为0xff并且有一个包长的数据bool
    def chkStart(self):
        while(self.buf[self.startIdx]!=0xff and self.pkgIsOk()):
            self.nextByte()
        return self.pkgIsOk()
    #检测索引超出buf末尾，从头开始
    def chkMaxIdx(self,idx):
        if (idx > self.bufLen - 1):
            idx %= self.bufLen  # ??
        return idx
    #获取包尾索引int
    def getPkgEndIdx(self):
        return self.chkMaxIdx(self.startIdx+self.pkgLen-1)
    #检查包尾为0xfe bool
    def chkEnd(self):
        return  self.buf[self.getPkgEndIdx()]==0xfe
    #相对起始索引包数据的每个字节索引的映射int
    def getIdxFromStart(self,i):
        return self.chkMaxIdx(self.startIdx+i)
    #检测校验和bool
    def chkCrc(self):
        crcidx=self.getPkgEndIdx()-1#包尾之前的一个字节
        if(crcidx<0):
            crcidx= self.bufLen+crcidx#包尾索引在缓冲区前面，包头在后面
        d= self.buf[crcidx]
        #print(d)
        r=0
        #计算和
        for i in range(1, self.pkgLen-2):
            #print(self.getIdxFromStart(i),":",self.buf[self.getIdxFromStart(i)])
            r+=self.buf[self.getIdxFromStart(i)]
        r=r&0xff
        #print(r)
        return d==r
    #写数据到显示区
    def writeData(self):
        r=0
        for i in range(4,self.pkgLen-2):
            r=self.buf[self.getIdxFromStart(i)]
            #print(r)
            self.linedata[i-4]=r
    def nextPkg(self):
        self.startIdx = self.chkMaxIdx(self.getPkgEndIdx()+1)
    def nextByte(self):
        self.startIdx=self.chkMaxIdx(self.startIdx+1)
    #解析包
    def parsePKG(self):
        while(self.pkgIsOk()):
            if(self.chkStart() and self.chkEnd() and self.chkCrc()):
                self.writeData()
                self.dispatch_event(PKGEvent(PKGEvent.PKG_DATA_OK,self.linedata ))
                self.nextPkg()
            else:
                self.nextByte()
    def writeToBuf(self,byt):
        self.buf[self.endIdx]=byt
        self.androidendIdx=self.chkMaxIdx(self.endIdx+1)
    def writeArrayToBuf(self,arr):
        for i in range(len(arr)):
            self.writeToBuf(arr[i])
        self.parsePKG()
    #'''
if __name__ == '__main__':
    import time
    bufjs=[0 for i in range(30)]
    buf[0]=0xff
    buf[1]=0x14
    buf[2]=0x01
    buf[3]=0x01
    buf[4]=0xff
    buf[5]=0xff
    buf[6]=0xff
    buf[7]=0xff
    buf[8]=0x11
    buf[9]=0xfe
    buf[10]=0xff
    buf[11]=0x14
    buf[12]=0x01
    buf[13]=0x01
    buf[14]=0xff
    buf[15]=0xff
    buf[16]=0xff
    buf[17]=0xff
    buf[18]=0x12
    buf[19]=0xfe
    buf[22] = 0xff
    buf[23] = 0x14
    buf[24] = 0x01
    buf[25] = 0x01
    buf[26] = 0xff
    buf[27] = 0xff
    buf[28] = 0xff
    buf[29] = 0xff
    buf[0] = 0x12
    buf[1] = 0xfe
    #startIdx=5
    #endIdx=2
    #dataLen=pkgLen-6
    #linedata=[0 for i in range(dataLen)]
    def on_data_ok(e):
        print("event ",e.data)
    pkg= FramePkg()
    pkg.add_event_listener(PKGEvent.PKG_DATA_OK,on_data_ok)
    c=0
    while True:
        #print(1)
        #print(pkg.buf)
        #print(pkg.linedata )
        '''print("getDatLen",pkg.getDatLen())
        print("pkgIsOk",pkg.pkgIsOk())
        print("chkStart",pkg.chkStart())
        print("chkEnd",pkg.chkEnd())
        print("chkCrc",pkg.chkCrc())
        print("getIdxFromStart",pkg.getIdxFromStart(30))
        pkg.writeData ()
        print("getPkgEndIdx-------------",pkg.getPkgEndIdx())
        pkg.nextPkg() '''
        #print("startIdx",pkg.startIdx)
        pkg.parsePKG()
        if (c < 30):
            pkg.writeToBuf(buf[c])
        else:
            pkg.endIdx=pkg.chkMaxIdx(pkg.endIdx+1)
        c += 1
        if(c%100==0):
            print(c)
        time.sleep(0.01)

参考前辈的event类

 
class Event(object):
    '''
      事件初始化的一个方式
    '''
    def __init__(self,event_type,data=None):
        self._type = event_type
        self._data = data
    @property
    def type(self):
        return self._type
    @property
    def data(self):
        re编程客栈turn self._data
class EventDispatcher(object):
     """
    event分发类 监听和分发event事件
    """
     def __init__(self):
         #初始化事件
         self._events = dict()
     def __del__(self):
         self._events = None
     def has_listener(self,event_type,listener):
        if event_type in self._events.keys():
            return listener in self._events[event_type]
        else:
            return False
     def dispatch_event(self,event):
          """
        Dispatch an instance of Event class
        """
        # 分发event到所有关联的listener
          if event.type in self._events.keys():
              listeners = self._events[event.type]
              for listener in listeners:
                  listener(event)
     def add_event_listener(self,event_type,listener):
         #给某种事件类型添加listner
         if not self.has_listener(event_type,listener):
             listeners = self._events.get(event_type,[])
             listeners.append(listener)
             self._events[event_type] = listeners
     def remove_event_listener(self,event_type,listener):
         if self.has_listener(event_type,listener):
             listeners = self._events[event_type]
             if len(listeners) == 1:
                 del self._events[event_type]
             else:
                 listeners.remove(listener)
                 self._events[event_type] = listeners
class PKGEvent(Event):
    PKG_DATA_OK = "PKG_DATA_OK"
if __name__ == '__main__':
    class Test(EventDispatcher):
        def __init__(self, a=1, b=1):
            EventDispatcher.__init__(self)
            self.a = a
            self.b = b
        def testDispEvent(self):
            self.dispatch_event(PKGEvent(PKGEvent.PKG_DATA_OK, 124))
    t=Test()
    def on_ok(e):
        print(e.data)
    t.add_event_listener(PKGEvent.PKG_DATA_OK,on_ok)
    import time
    while True:
        print(1)
        t.testDispEvent()
        time.sleep(1)

总结

以上为个人经验，希望能给大家一个参考，也希望大家多多支持编程客栈(www.devze.com)。