115 lines
2.0 KiB
Python
115 lines
2.0 KiB
Python
|
|
|||
|
|
import os
|
|||
|
|
import sys
|
|||
|
|
import json
|
|||
|
|
|
|||
|
|
|
|||
|
|
|
|||
|
|
|
|||
|
|
# 数据包编解码
|
|||
|
|
|
|||
|
|
|
|||
|
|
|
|||
|
|
|
|||
|
|
|
|||
|
|
|
|||
|
|
|
|||
|
|
# 计算整个数据流的CRC
|
|||
|
|
def _crc16(data:bytearray):
|
|||
|
|
lenth=len(data)
|
|||
|
|
if(lenth>0):
|
|||
|
|
crc=0xffff
|
|||
|
|
for i in range(lenth):
|
|||
|
|
crc=(crc^data[i])&0xffff
|
|||
|
|
for j in range(8):
|
|||
|
|
if(crc&1)!=0:
|
|||
|
|
crc=((crc>>1)^0xa001)&0xffff
|
|||
|
|
else:
|
|||
|
|
crc=(crc>>1)&0xffff
|
|||
|
|
return bytearray([crc&0xff,(crc>>8)&0xff])
|
|||
|
|
return bytearray([0,0])
|
|||
|
|
|
|||
|
|
|
|||
|
|
# 0xff帧起始，0xfe帧结尾
|
|||
|
|
# 0xf0转义标识
|
|||
|
|
# 0xf1命令数据分割
|
|||
|
|
# 0xf2数据校验分割
|
|||
|
|
|
|||
|
|
# 字节数据大于等于0xf0 之后要转义
|
|||
|
|
def _tran(data:bytearray):
|
|||
|
|
ret=bytearray()
|
|||
|
|
for item in data:
|
|||
|
|
if(item>=0xf0):
|
|||
|
|
ret.append(0xf0)
|
|||
|
|
ret.append(item-0xf0)
|
|||
|
|
else:
|
|||
|
|
ret.append(item)
|
|||
|
|
return ret
|
|||
|
|
|
|||
|
|
# 去转义
|
|||
|
|
def _untran(data:bytearray):
|
|||
|
|
ret=bytearray()
|
|||
|
|
i=0
|
|||
|
|
length=len(data)
|
|||
|
|
while i<length:
|
|||
|
|
if(data[i]>=0xf0):
|
|||
|
|
ret.append(data[i]+data[i+1])
|
|||
|
|
i+=1
|
|||
|
|
else:
|
|||
|
|
ret.append(data[i])
|
|||
|
|
i+=1
|
|||
|
|
return ret
|
|||
|
|
|
|||
|
|
|
|||
|
|
# 以命令和数据编码
|
|||
|
|
def encode(cmd:bytearray,data:bytearray):
|
|||
|
|
crc=_crc16(cmd+data)
|
|||
|
|
ret=bytearray()
|
|||
|
|
ret.append(0xff)
|
|||
|
|
ret+=_tran(cmd)
|
|||
|
|
ret.append(0xf1)
|
|||
|
|
ret+=_tran(data)
|
|||
|
|
ret.append(0xf2)
|
|||
|
|
ret+=_tran(crc)
|
|||
|
|
ret.append(0xfe)
|
|||
|
|
return ret
|
|||
|
|
|
|||
|
|
|
|||
|
|
# 解码为命令和数据
|
|||
|
|
def decode(data:bytearray):
|
|||
|
|
# print('prot_decode:',data.hex(' '))
|
|||
|
|
while len(data)>0:
|
|||
|
|
if(data[0]!=0xff):
|
|||
|
|
data=data[1:]
|
|||
|
|
else:
|
|||
|
|
break
|
|||
|
|
if(len(data)==0):
|
|||
|
|
return (None,None)
|
|||
|
|
if(data[0]==0xff and data[-1]==0xfe):
|
|||
|
|
data=data[1:-1]
|
|||
|
|
cmd_index=data.find(0xf1)
|
|||
|
|
data_index=data.find(0xf2)
|
|||
|
|
cmd=_untran(data[:cmd_index])
|
|||
|
|
da=_untran(data[cmd_index+1:data_index])
|
|||
|
|
crc=_untran(data[data_index+1:])
|
|||
|
|
if(crc==_crc16(cmd+da)):
|
|||
|
|
return (cmd,da)
|
|||
|
|
return (None,None)
|
|||
|
|
|
|||
|
|
|
|||
|
|
|
|||
|
|
|
|||
|
|
|
|||
|
|
|
|||
|
|
|
|||
|
|
|
|||
|
|
if __name__ == "__main__":
|
|||
|
|
data=encode(json.dumps({'device':'client','option':'data'}).encode('utf-8'),bytearray([0xff,0xfd,0xff,0xde,0xde]))
|
|||
|
|
print(data.hex(' '))
|
|||
|
|
print(decode(data))
|
|||
|
|
print(b'\xff'.hex(' '))
|
|||
|
|
|
|||
|
|
|
|||
|
|
|
|||
|
|
|