开发者

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

开发者 https://www.devze.com 2023-11-19 11:25 出处:网络 作者: qq_278667286
目录写个自定义数据包协议的打包和解包测试打包代码解包代码总结写个自定义数据包协议的打包和解包测试
目录
  • 写个自定义数据包协议的打包和解包测试
  • 打包代码
  • 解包代码
  • 总结

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

在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)。

0

精彩评论

暂无评论...
验证码 换一张
取 消

关注公众号