andy/python_tools
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

添加jwt模块程序升级命令0xec

Browse Source
This commit is contained in:
ranchuan
2023-10-08 18:25:31 +08:00
parent b784c66e23
commit 5ee17b6ce2
6 changed files with 272 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
.gitignore vendored
View File

@@ -8,3 +8,4 @@ download/
*.csv
quest_info.txt
scheme_export_info.txt
device_log.txt

5
ReadMe.txt
View File

@@ -67,4 +67,9 @@
赋码工具在模块接触异常时也保存管码
2023.9.28
修改 daemon.py 检测sd卡目录为 /run/media/sda/updata ，解决sd卡未挂载完成造成的无法打开目录的问题
2023.10.8
添加jwt模块程序升级命令0xec
添加赋码仪批检仪联动脚本
发送文件时打印文件的crc32

233
coder/code_with_checker.py Normal file
View File

@@ -0,0 +1,233 @@
import serial
import serial.tools.list_ports
import threading
import time
import socket
import sys
# import checker_save
# 使用批检仪和赋码仪联动，
# 批检仪控制电机，赋码仪注码
def crc16(data:bytearray,offset:int,len:int):
if(len>0):
crc=0xffff
for i in range(len-offset):
crc=(crc^data[i+offset])&0xffff
for j in range(8):
if(crc&1)!=0:
crc=((crc>>1)^0xa001)&0xffff
else:
crc=(crc>>1)&0xffff
return crc&0xff,(crc>>8)&0xff
return 0,0
# 把tcp封装为串口
class utcp:
is_open=False
def __init__(self,port:int)->None:
self.ser = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.ser.bind(("",port))
self.ser.settimeout(10)
self.ser.listen(128)
print("wait for mcu connect.")
self.client,self.client_addr=self.ser.accept()
print("client:",self.client_addr)
self.is_open=True
def read(self,len:int):
return self.client.recv(len)
def write(self,data:bytearray):
return self.client.send(data)
def close(self):
self.client.close()
self.ser.close()
self.is_open=False
class port:
def __init__(self) -> None:
self.cmd=0
self.cmd_no=0
self.str_err=""
self.is_big_data=False
self.num_to_recv=0
self.recv_data=bytearray()
def open(self,name:str,bsp:int):
if(name!="utcp"):
self.ser = serial.Serial(port=name, baudrate=bsp,bytesize=serial.EIGHTBITS,parity=serial.PARITY_NONE,
stopbits=serial.STOPBITS_ONE,timeout=None)
else:
self.ser=utcp(bsp)
# 打开赋码仪端口
def open_code(self,name:str,bsp:int):
if(name!="utcp"):
self.code_ser = serial.Serial(port=name, baudrate=bsp,bytesize=serial.EIGHTBITS,parity=serial.PARITY_NONE,
stopbits=serial.STOPBITS_ONE,timeout=None)
else:
self.code_ser=utcp(bsp)
def start_recv(self):
self.thread_ = threading.Thread(target=self.recv, args=())
self.thread_.start()
def decode(self,data:bytearray):
self.str_err="ok"
if(len(data)<10):
print("recv data len too less.")
self.str_err="recv data len too less."
return bytearray()
if(data[0]!=0x59 or data[1]!=0x6d or data[2]!=0x43):
print("frame head not 0x59 0x6d.")
self.str_err="frame head not 0x59 0x6d."
return bytearray()
length=data[3]|(data[4]<<8)
if(length==65535):
# 重新设置数据长度
length=data[7]|(data[8]<<8)|(data[9]<<16)|(data[10]<<24);
self.is_big_data=True
else:
self.is_big_data=False
if(length+7!=len(data)):
print("recv data have lossed")
self.str_err="recv data have lossed"
return bytearray()
a,b=crc16(data,3,length+5)
if(a!=data[-2] or b!=data[-1]):
print("recv data check error.h_crc=%02x %02x,crc=%02x %02x",a,b,data[-2],data[-1])
self.str_err="recv data check error."
self.cmd_no=data[6]|(data[7]<<8)
self.cmd=data[5]
if(self.is_big_data==False):
return data[8:-2]
else:
return data[12:-2]
def recv(self):
data=bytearray()
while(True):
d=bytes()
try:
d=self.ser.read(1)
except Exception as a:
# print("err:",str(a))
print("port closed")
return
data+=d
if(len(data)==3):
if(data[0]==0x59 and data[1]==0x6d and data[2]==0x43):
self.num_to_recv=5
else:
data=data[1:]
self.num_to_recv=0
elif(len(data)==5):
length=data[3]|(data[4]<<8)
if(length<65535):
self.num_to_recv+=length+2
self.is_big_data=False
else:
self.num_to_recv=12
self.is_big_data=True
elif(len(data)==12):
if(self.is_big_data==True):
length=data[8]|(data[9]<<8)|(data[10]<<16)|(data[11]<<24)
self.num_to_recv=5+length+2
if(self.num_to_recv>0 and self.num_to_recv==len(data)):
# print("recv:",data.hex(" "))
return self.decode(data)
def recv_code(self,num:int):
d=bytearray()
try:
while(num>len(d)):
d+=self.code_ser.read(num-len(d))
except Exception as a:
print("port closed")
return 0
print("recv code:",d.hex(" "))
# return d[0]
def send_code(self,data:bytearray):
self.code_ser.write(data)
print("send code:",data.hex(' '))
def encode(self,data:bytearray):
t=bytearray()
length=len(data)+3
t.append(0x59)
t.append(0x6d)
t.append(length&0xff)
t.append(length>>8)
t.append(self.cmd)
t.append(self.cmd_no&0xff)
t.append(self.cmd_no>>8)
t+=data
a,b=crc16(t,2,length+4)
t.append(a)
t.append(b)
# print("encode:",t.hex(","))
return t
def send(self,cmd:int,data:bytearray):
self.cmd=cmd
self.cmd_no+=1
d=self.encode(data)
# print("send:",d.hex(","))
self.ser.write(d)
# 测试批检仪
def checker_test(self,times:int,scheme:str,script:str):
print("批检仪测试：")
tick=0
# save=checker_save.save(scheme,script)
data=bytearray()
while(tick<times):
tick+=1
print("当前={d1},总共={d2}".format(d1=tick,d2=times))
start = time.perf_counter()
# 接收请求检测命令
data=self.recv()
if(self.cmd==0x37):
self.send(0x40,bytearray([3]))
# 接收指令应答
self.recv()
# 延时2秒等电机下压
time.sleep(3)
# 开始检测
self.send_code(bytearray([0xB0, 0x00, 0x02, 0x00, 0x60]))
# 接收指令应答
self.recv_code(5)
# 接收检测结果
self.recv_code(24)
# 开始注码
# self.send_code(bytearray([0xF0, 0x00, 0x02, 0xE0, 0x41, 0xB1, 0x00, 0x02, 0x00, 0xFB]))
# # 接收指令应答
# self.recv_code(5)
# # 接收注码 结果
# self.recv_code(305)
# # 打标结束，保存数据库
# self.send_code(bytearray([0xF0, 0x00, 0x02, 0xE2, 0x23, 0xB2, 0x00, 0x02, 0x00, 0x67]))
# # 接收指令应答
# self.recv_code(5)
# 电机上升
self.send(0x40,bytearray([2]))
end = time.perf_counter()
print("检测耗时:",end-start)
if __name__ == "__main__":
if(len(sys.argv)<3):
print("argv num too less")
print("请输入批检仪串口号，赋码仪串口号")
checker_port=sys.argv[1]
coder_port=sys.argv[2]
p=port()
# 批检仪测试
p.open(checker_port,57600)
p.open_code(coder_port,115200)
# p.open("utcp",9527)
p.checker_test(5000,"file/EX_Coder_Test_2023-07-4.json","file/judge-xt.lua")

32
coder/coder_test.py
View File

@@ -3,6 +3,7 @@ import serial.tools.list_ports
import threading
import time
import socket
import sys
# 把tcp封装为串口
class utcp:
@@ -109,18 +110,39 @@ class port:
end = time.perf_counter()
print("总耗时:",end-start)
if __name__ == "__main__":
def help():
print("------------|检测赋码仪系统测试工具|------------")
print("请使用如下命令行调用本程序：")
print("./coder_test.exe com8 5000")
print("com8 为使用的串口号 根据需要修改")
print("5000 为次数 根据需要修改")
def main():
p=port()
if(len(sys.argv)<3):
help()
return
com=sys.argv[1]
count=int(sys.argv[2])
print("串口号为：{d1} ,运行次数为：{d2}".format(d1=com,d2=count))
# 检测赋码仪系统测试
# p.open("com8",115200)
# p.ecode_test(50000)
p.open(com,115200)
p.ecode_test(count)
# 赋码仪设备串口测试
p.open("com7",115200)
p.coder_test(1)
# p.open("com7",115200)
# p.coder_test(1)
# p.open("com16",115200)
# p.coder_test(1000)
# 赋码仪设备网口测试
# p.open("utcp",0)
# p.coder_test(5)
if __name__ == "__main__":
main()

1
updata/prottcp.py
View File

@@ -229,6 +229,7 @@ class handle:
with open(name,"rb") as f:
self.data=f.read()
self.name=f.name.split('/')[-1]
print("file:{d1},crc32={d2}".format(d1=self.name,d2=hex(crc32(self.data))))
self.stat=0
self.packet_all=(len(self.data)+199)//200
self.sent_bytes=0

6
updata/updata_uart.py
View File

@@ -268,7 +268,7 @@ class updata_dlg(QObject):
print("com changed")
self.close_port()
def get_cmd(self,file:str):
l=[(".bin",0xee),(".pkt",0xed),(".json",0x32)]
l=[(".bin",0xee),(".pkt",0xed),(".json",0x32),(".jwt",0xec)]
for i in l:
if(file.endswith(i[0])):
return i[1]
@@ -407,7 +407,7 @@ class updata_dlg(QObject):
# 扫描文件
def scan_file(self):
self.file_list.clear()
self.file_list.addItems(self.find_type([".bin",".json",".pkt"]))
self.file_list.addItems(self.find_type([".bin",".json",".pkt",".jwt"]))
# 扫描指定类型的文件
@@ -467,6 +467,8 @@ class updata_dlg(QObject):
pass
def recv_str_slot(self,cmd:int,txt:str,err:str):
print("|-|",txt)
with open("device_log.txt","a+") as f:
f.write(txt+'\n')
def recv_slot(self,cmd:int,data:bytearray,err:str):
# print("recv:",cmd,data)
if(self.cmd!=cmd):