147 lines
5.3 KiB
Python
147 lines
5.3 KiB
Python
import serial
|
|
import serial.tools.list_ports
|
|
import threading
|
|
import time
|
|
import socket
|
|
import sys
|
|
|
|
# 把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:
|
|
pass
|
|
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(9527)
|
|
def start_recv(self):
|
|
self.thread_ = threading.Thread(target=self.recv, args=())
|
|
self.thread_.start()
|
|
def recv(self,num:int):
|
|
d=bytearray()
|
|
try:
|
|
while(num>len(d)):
|
|
d+=self.ser.read(num-len(d))
|
|
# print(d.hex(" "))
|
|
except Exception as a:
|
|
print("port closed")
|
|
return 0
|
|
print("recv code:",d.hex(" "))
|
|
def send(self,data:bytearray):
|
|
self.ser.write(data)
|
|
# 测试上位机
|
|
def ecode_test(self,times:int):
|
|
print("检测赋码系统耗时测试：")
|
|
tick=0
|
|
while(tick<times):
|
|
tick+=1
|
|
print("当前={d1},总共={d2}".format(d1=tick,d2=times))
|
|
# 等待数据库写入
|
|
time.sleep(3)
|
|
start = time.perf_counter()
|
|
# 开始检测
|
|
self.send(bytearray([0xB0, 0x00, 0x02, 0x00, 0x60]))
|
|
# 接收指令应答
|
|
self.recv(5)
|
|
# 接收检测结果
|
|
self.recv(24)
|
|
end = time.perf_counter()
|
|
print("检测耗时:",end-start)
|
|
start2=time.perf_counter()
|
|
# 开始注码
|
|
self.send(bytearray([0xF0, 0x00, 0x02, 0xE0, 0x41, 0xB1, 0x00, 0x02, 0x00, 0xFB]))
|
|
# 接收指令应答
|
|
self.recv(5)
|
|
# 接收注码 结果
|
|
self.recv(305)
|
|
end = time.perf_counter()
|
|
print("注码耗时:",end-start2)
|
|
# 打标结束，保存数据库
|
|
self.send(bytearray([0xF0, 0x00, 0x02, 0xE2, 0x23, 0xB2, 0x00, 0x02, 0x00, 0x67]))
|
|
# 接收指令应答
|
|
self.recv(5)
|
|
end = time.perf_counter()
|
|
print("总耗时:",end-start)
|
|
# 测试注码仪
|
|
def coder_test(self,times:int):
|
|
print("赋码仪耗时测试：")
|
|
tick=0
|
|
while(tick<times):
|
|
tick+=1
|
|
start = time.perf_counter()
|
|
print("当前={d1},总共={d2}".format(d1=tick,d2=times))
|
|
# # 开始检测
|
|
# self.send(bytearray([0x59, 0x6D, 0x05, 0x00, 0x01, 0x66, 0x04, 0x05, 0x00, 0x2F, 0xD9]))
|
|
# # 接收指令应答
|
|
# self.recv(13)
|
|
# # 接收检测结果
|
|
# self.recv(42)
|
|
# end = time.perf_counter()
|
|
# print("检测耗时:",end-start)
|
|
start2=time.perf_counter()
|
|
# # 开始注码
|
|
self.send(bytearray([0x59,0x6D,0x94,0x00,0x02,0xBE,0x01,0x0A,0x32,0x30,0x32,0x33,0x00,0x35,0x38,0x33,0x30,0x36,0x31,0x39,0x41,0x30,0x34,0x30,0x30,0x39,0x01,0x35,0x38,0x33,0x30,0x36,0x31,0x39,0x41,0x30,0x34,0x30,0x30,0x38,0x02,0x35,0x38,0x33,0x30,0x36,0x31,0x39,0x41,0x30,0x34,0x30,0x30,0x37,0x03,0x35,0x38,0x33,0x30,0x36,0x31,0x39,0x41,0x30,0x34,0x30,0x30,0x36,0x04,0x35,0x38,0x33,0x30,0x36,0x31,0x39,0x41,0x30,0x34,0x30,0x30,0x35,0x05,0x35,0x38,0x33,0x30,0x36,0x31,0x39,0x41,0x30,0x34,0x30,0x30,0x34,0x06,0x35,0x38,0x33,0x30,0x36,0x31,0x39,0x41,0x30,0x34,0x30,0x30,0x33,0x07,0x35,0x38,0x33,0x30,0x36,0x31,0x39,0x41,0x30,0x34,0x30,0x30,0x32,0x08,0x35,0x38,0x33,0x30,0x36,0x31,0x39,0x41,0x30,0x34,0x30,0x30,0x31,0x09,0x35,0x38,0x33,0x30,0x36,0x31,0x39,0x41,0x30,0x34,0x30,0x30,0x30,0x44,0xC1]))
|
|
# 接收指令应答
|
|
self.recv(12)
|
|
# 接收注码 结果
|
|
self.recv(401)
|
|
end = time.perf_counter()
|
|
print("注码耗时:",end-start2)
|
|
end = time.perf_counter()
|
|
print("总耗时:",end-start)
|
|
|
|
|
|
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(com,115200)
|
|
p.ecode_test(count)
|
|
|
|
# 赋码仪设备串口测试
|
|
# p.open(com,115200)
|
|
# p.coder_test(count)
|
|
|
|
# 赋码仪设备网口测试
|
|
# p.open("utcp",0)
|
|
# p.coder_test(count)
|
|
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main() |