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

整合python目录的代码

Browse Source
This commit is contained in:
ranchuan
2023-09-15 10:53:44 +08:00
parent 58b751d519
commit 8a14e74500
25 changed files with 4746 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

307
updata/ftp.py Normal file
View File

@@ -0,0 +1,307 @@
import paramiko
from ftplib import FTP
import os
import threading
import udp
import sys
import time
import stat
from PyQt5.QtCore import *
STR_RED="\033[1;31m"
STR_BLUE="\033[1;34m"
STR_END="\033[0m"
# 获得文件绝对路径
def getpath():
path=os.path.abspath(sys.argv[0])
list_str=path.split("\\")
return path.replace(list_str[-1],"")
class myftp(QObject):
local_path=getpath()+'file'
# print(local_path)
remot_path='/home/root'
# 进度信号，ip，1~100
rate_signal =pyqtSignal([str,int])
# 结束信号，ip，成败，描述
end_signal = pyqtSignal([str,bool,str])
def __init__(self,ip,user,password):
QObject.__init__(self)
self.ip=ip
self.ssh = paramiko.SSHClient()
# 加入受信主机列表
self.ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
self.ssh.connect(ip, 22, user, password,timeout=10)
# 实例化一个 sftp对象,指定连接的通道
self.sftp=self.ssh.open_sftp()
self.invoke = self.ssh.invoke_shell()
def send_cmd(self,cmd,inputs=[],sleep_s=0.5):
# print("send cmd:",cmd)
console=""
input_list=[]
input_list=inputs.copy()
self.invoke.send(cmd+'\n') # \n很重要，相当于回车
time.sleep(sleep_s)
while True:
out=self.invoke.recv(9999).decode("utf-8") # 提取数据然后解码
if(len(out)>0):
# print(out)
console+=out
else:
break
if(len(input_list)>0):
send_str=input_list.pop(0)
# print(self.ip,"|"+send_str)
self.invoke.send(send_str+'\n')
time.sleep(0.5)
else:
break
console='\n'+console
console=console.replace('\n','\n\t|')
print(self.ip,console)
# print("send cmd end")
def get_file(self,path,src_list):
lo_path=self.local_path+"\\"
if(len(path)>0):
lo_path=lo_path+path+"\\"
if not os.path.exists(lo_path):
os.makedirs(lo_path)
for i in src_list:
file_name=i.split('/')
try:
# print(file_name[-1])
if(len(file_name[-1])>0):
self.sftp.get(remotepath=i, localpath=lo_path+file_name[-1])
else:
file_list=self.scan_dir(i)
# self.sftp.get(remotepath=i, localpath=lo_path[0:-1])
self.get_file(path,file_list)
except Exception as r:
print(lo_path,"|"+str(r))
print(STR_RED,lo_path,"|get ",i,"failed.",STR_END)
return
print(STR_BLUE,lo_path,"|get ",i,"success.",STR_END)
def scan_dir(self,remotdir):
# list=self.sftp.listdir(remotdir)
attrs=self.sftp.listdir_attr(remotdir)
file_list=[]
for attr in attrs:
if not stat.S_ISDIR(attr.st_mode):
file_list.append(remotdir+attr.filename)
# print(file_list)
return file_list
# 关闭连接
def close(self):
self.ssh.close()
def put_cb(self,sent:int,all:int):
# (sended*100/data.size()+((index-1)*100))/get_addrs().size()
self.put_rate=sent*100//all
rate=(self.put_rate+(self.index)*100)//self.len_files
self.rate_signal.emit(self.ip,rate)
# print("sent={d1},all={d2}".format(d1=sent,d2=all))
# print("rate=",rate)
# 添加进度和结束信号，供qt显示
def put_file(self,prefix,dst_list,src_list):
if not os.path.exists(self.local_path):
self.end_signal.emit(self.ip,False,"local path is not exists.")
return
self.len_files=len(dst_list)
self.index=0
for i in dst_list:
self.put_rate=0
file_name=i.split('/')
temp_path=self.remot_path+'/'+file_name[-1]
src_path=""
if(len(src_list)==0):
src_path=self.local_path+'\\'+file_name[-1]
else:
src_path=src_list[self.index]
if os.path.exists(src_path):
self.send_cmd("rm "+i)
# self.sftp.chmod(i.replace(file_name[-1],""),777)
self.send_cmd("chmod "+"777 "+i.replace(file_name[-1],""))
try:
self.sftp.put(localpath=src_path, remotepath=i,callback=self.put_cb)
except Exception as r:
print(prefix,"|"+str(r))
print(STR_RED,prefix,"|put ",i,"failed.",STR_END)
self.end_signal.emit(self.ip,False,str(r)+"|"+i)
return
self.index+=1
print(STR_BLUE,prefix,"|put ",i,"success.",STR_END)
# self.sftp.chmod(i,777)
self.send_cmd("chmod "+"777 "+i)
# self.rate_signal.emit(self.ip,self.index*100//self.len_files)
else:
print(STR_RED,prefix,"|file:"+src_path +" not exist.",STR_END)
self.end_signal.emit(self.ip,False,"file:"+src_path +" not exist.")
return
self.end_signal.emit(self.ip,True,"success")
# 从虚拟机下载文件
def load_file(ip,src_list):
print("load file from:",ip)
# ftp=myftp(ip,"andy","123456")
ftp=myftp(ip,"bitman","123456")
ftp.get_file("",src_list)
ftp.close()
print("load file end.")
# 保存文件到开发板
def save_file(ip,restart:bool,dst_list,src_list):
try:
ftp=myftp(ip,"root","")
if(restart==True):
print(ip,"|stop app.")
ftp.send_cmd("systemctl stop atk-qtapp-start.service")
ftp.put_file(ip,dst_list,src_list)
if(restart==True):
print(ip,"|start app.")
ftp.send_cmd("systemctl restart atk-qtapp-start.service")
ftp.send_cmd("sync")
ftp.close()
# print(ip,"|save file end.")
except Exception as r:
print(STR_RED,ip,"|"+str(r),STR_END)
# 从开发板加载文件
def get_file(ip,path,get_list):
# print("load file from:",ip)
ftp=myftp(ip,"root","")
ftp.get_file(path,get_list)
ftp.close()
# print("load file end.")
def send_cmd(ip,cmd_list):
ftp=myftp(ip,"root","")
for i in cmd_list:
ftp.send_cmd(i[0],i[1])
ftp.close()
# 定义要从虚拟机下载的文件
src_list=[]
# src_list.append("/home/bitman/linux/nfs/stm32mp157d-atk.dtb")
# src_list.append("/home/bitman/linux/nfs/QDesktop-fb")
# src_list.append("/home/bitman/QtWorkSpace/tcp_tr_can/checker_ye_cfg08.json")
src_list.append("/home/bitman/linux/nfs/")
# 定义要保存到开发板的文件
dst_list=[]
# 第一次拷贝设备树会因为权限问题不能运行
# dst_list.append("/run/media/mmcblk2p2/stm32mp157d-atk.dtb")
# dst_list.append("/usr/local/QDesktop-fb")
# dst_list.append("/home/root/config/checker_app_pre.sh")
# dst_list.append("/home/root/config/cfg.json")
# dst_list.append("/home/root/config/checker_ye_cfg11.json")
dst_list.append("/home/root/config/judge.lua")
# dst_list.append("/home/root/config/JQChecker_0008.bin")
# 定义要从开发板获取的文件
get_list=[]
# get_list.append("/home/root/comm_log/comm_log.csv")
# get_list.append("/home/root/log/debug_log.txt")
# get_list.append("/home/root/config/checker_app_pre.sh")
# get_list.append("/home/root/config/cfg.json")
# get_list.append("/home/root/config/judge.lua")
get_list.append("/home/root/comm_log/")
# 定义命令
cmd_list=[]
# cmd_list.append(("/usr/sbin/useradd -d / -g wheel andy",[]))
# cmd_list.append(("passwd andy",["123456","123456"]))
cmd_list.append(("rm ~/config/checker_app_pre.sh",[]))
# cmd_list.append(("ls ~/config/",[]))
# cmd_list.append(("ls /usr/local/",[]))
# 自动扫描从机地址，然后升级文件
# python .\ftp.py get 从开发板下载文件
# python .\ftp.py save 保存文件到开发板
# python .\ftp.py cmd 发送命令列表
def run():
if(len(sys.argv)<2):
print("argv too less")
return
print(sys.argv[0])
cmd=sys.argv[1]
ip_list=[]
u=udp.myudp(1,255)
u.find("192.168.80")
ip_list=u.dst_ip_list
if(len(ip_list)==0):
# ip_list.append(("192.168.80.120","local_id_1"))
ip_list.append(("192.168.80.81","local_id_1"))
ip_list.append(("192.168.80.82","local_id_2"))
ip_list.append(("192.168.80.83","local_id_3"))
ip_list.append(("192.168.80.84","local_id_4"))
ip_list.append(("192.168.80.85","local_id_5"))
ip_list.append(("192.168.80.86","local_id_6"))
ip_list.append(("192.168.80.87","local_id_7"))
ip_list.append(("192.168.80.88","local_id_8"))
print("find ip_list as:")
print(ip_list)
threads = []
if(cmd=="save"):
# load_file("192.168.80.188",src_list)
print('put file to ip_list.')
for i in ip_list:
t = threading.Thread(target=save_file, args=(i[0],True,dst_list ))
threads.append(t)
t.start()
elif(cmd=="get"):
print("get file from ip_list.")
for i in ip_list:
t = threading.Thread(target=get_file, args=(i[0],i[1],get_list ))
threads.append(t)
t.start()
elif(cmd=="cmd"):
print("send cmd to ip_list")
for i in ip_list:
t = threading.Thread(target=send_cmd, args=(i[0],cmd_list ))
threads.append(t)
t.start()
elif(cmd=="load"):
load_file("192.168.80.188",src_list)
#等待所有线程任务结束。
for t in threads:
t.join()
print("all task were end.")
if __name__ == "__main__":
run()

BIN
updata/icon.ico Normal file
View File

Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 66 KiB

1
updata/memory_pic.py Normal file
View File

File diff suppressed because one or more lines are too long

179
updata/mysql.py Normal file
View File

@@ -0,0 +1,179 @@
import pymysql as mysql
import datetime
from datetime import datetime, timedelta
import hashlib
import os
def connect():
try:
db=mysql.connect(host='124.70.178.159',user='admin',passwd='Rc5345750.',port=3306)
print("connect mysql success.")
return db
except Exception as e:
print("can not connect service.")
return None
# 获取北京时间
def get_date():
now_time = datetime.utcnow()
utc_time = now_time + timedelta(hours=8) # UTC只是比北京时间提前了8个小时
utc_time = utc_time.strftime("%Y%m%d")
return utc_time
# 获取北京时间
def get_time():
now_time = datetime.utcnow()
utc_time = now_time + timedelta(hours=8) # UTC只是比北京时间提前了8个小时
utc_time = utc_time.strftime("%Y-%m-%d %H:%M:%S")
return utc_time
# 获取数据md5
def md5(data:bytearray):
m=hashlib.md5(data).hexdigest()
# print(m)
return m
# 获得主板sn号
def get_computer_sn():
sn = os.popen("wmic bios get serialnumber").readlines()
return sn[2].strip()
class sql:
def __init__(self) -> None:
self.download_path="download/"
if not os.path.exists(self.download_path):
os.makedirs(self.download_path)
# 初始化返回True成功
def init(self,table_name:str):
self.db=connect()
self.table_name=table_name
if(self.db!=None):
self.cur=self.db.cursor()
self.cur.execute("use andy_data")
self.cur.execute("select version()")
a=self.cur.fetchone()
print(a)
# self.create_table(self.table_name)
return True
return False
# 创建表
def create_table(self,table_name:str):
cmd="""CREATE TABLE IF NOT EXISTS `{d1}`(
`id` INT UNSIGNED AUTO_INCREMENT,
`time` VARCHAR(30) NOT NULL,
`name` VARCHAR(256) NOT NULL,
`md5` VARCHAR(33) NOT NULL,
`data` MEDIUMBLOB NOT NULL,
`info` VARCHAR(512),
PRIMARY KEY ( `id` )
)ENGINE=InnoDB DEFAULT CHARSET=utf8;""".format(d1=table_name)
self.cur.execute(cmd)
# 插入数据
def insert(self,file_name:str):
s=file_name.split('.')
if(len(s)<2):
print("file name without type suffix,will not insert.")
return
with open(file_name,"rb") as f:
d=f.read()
md=md5(d)
lis=self.show()
if(len(lis)>0):
if(lis[-1][3]==md):
print("the same file was saved,will not insert.")
return
try:
cmd="INSERT INTO {d1} (time,name,md5,data) VALUES (%s,%s,%s,%s);".format(d1=self.table_name)
self.db.begin()
self.cur.execute(cmd,([get_time(),file_name,md,d]))
self.db.commit()
print("insert file success.")
except Exception as e:
self.db.rollback()
print(str(e))
# 查看数据
def items(self):
cmd= "select id,time,name,md5,info from {d1};".format(d1=self.table_name)
self.cur.execute(cmd)
a=self.cur.fetchall()
# for i in a:
# print(i[0],i[1],i[2],i[3])
return a
def show_tables(self):
cmd= "show tables"
self.cur.execute(cmd)
a=self.cur.fetchall()
# for i in a:
# print(i)
return a
# 下载指定文件,返回文件路径
def download(self,id:int):
ack,name=self.exists(id)
if(ack==True):
print("the same file is exists,will not download.")
return name
cmd="select name,data from {d1} WHERE id={d2};".format(d1=self.table_name,d2=id)
self.cur.execute(cmd)
a=self.cur.fetchall()
for i in a:
ss=i[0].replace('\\','/')
ss=ss.split('/')[-1].split('.')
name_prefix='.'.join(ss[:-1])
name=self.download_path+name_prefix+'--'+str(id)+'.'+ss[-1]
with open(name,'+bw') as f:
f.write(i[1])
return name
print("can not find the file with id:",id)
return ""
# 获取md5
def get_md5(self,id:int):
cmd="select md5 from {d1} WHERE id={d2};".format(d1=self.table_name,d2=id)
self.cur.execute(cmd)
a=self.cur.fetchall()[0]
return a[0]
# 扫描文件
def scan_files(self):
path = self.download_path
if not os.path.exists(path):
os.makedirs(path)
list=os.listdir(path)
return list
# 判断文件是否存在
def exists(self,id:int):
for i in self.scan_files():
s=i.split('.')[-2].split('-')[-1]
if(int(s)==id):
name=self.download_path+i
with open(name,"rb") as f:
md=md5(f.read())
if(md==self.get_md5(id)):
return True,name
return False,""
if __name__ == "__main__":
s=sql()
if(s.init("test_data")==True):
# s.insert("file\\check_result.csv")
# s.show()
# file=s.download(1)
# print("save file:",file)
s.show_tables()

33
updata/pic_2py.py Normal file
View File

@@ -0,0 +1,33 @@
import base64
def pictopy(picture_names, py_name):
"""
将图像文件转换为py文件
:param picture_name:
:return:
"""
write_data = []
for picture_name in picture_names:
filename = picture_name.replace('.', '_')
open_pic = open("%s" % picture_name, 'rb')
b64str = base64.b64encode(open_pic.read())
open_pic.close()
# 注意这边b64str一定要加上.decode()
write_data.append('%s = "%s"\n' % (filename, b64str.decode()))
f = open('%s.py' % py_name, 'w+')
for data in write_data:
f.write(data)
f.close()
# pics = ["logo_2.png", "logo.ico"] 中的图片存放在PicToPy.py同一目录中，运行完成后，
# 会在当前路径看到一个memory_pic.py文件
# pics = ["logo_2.png", "subway.ico"]
pics = ["icon.ico"]
pictopy(pics, 'memory_pic') # 将pics里面的图片写到 memory_pic.py 中
print("ok")

536
updata/prottcp.py Normal file
View File

@@ -0,0 +1,536 @@
import serial
import serial.tools.list_ports
import threading
import time
import json
import socket
from PyQt5.QtCore import *
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
def crc32(data:bytearray):
temp=0
crc=0xffffffff
i=0
if(len(data)%4!=0):
return 0
while(i<len(data)):
temp=data[i]|(data[i+1]<<8)|(data[i+2]<<16)|(data[i+3]<<24)
i+=4
for j in range(32):
if((crc^temp)&0x80000000)!=0:
crc=0x04c11db7^(crc<<1)
else:
crc<<=1
temp<<=1
crc&=0xffffffff
return crc
# 把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.ser.recv(len)
return self.client.recv(len)
def write(self,data:bytearray):
# return self.ser.send(data)
return self.client.send(data)
def close(self):
self.client.close()
self.ser.close()
self.is_open=False
# 生成一个任务的参数
def scheme_task_to_byte(j:json):
data=bytearray()
data.append(j["TaskID"])
data.append(j["TaskIndex"])
data.append(j["RetryCount"])
data.append(j["ErrJumpTo"])
data.append((j["ParamCount"]&0x0f)|(j["ReturnCount"]<<4))
for i in j["ParamVal"]:
data.append(i&0xff)
data.append(i>>8)
return data
# 生成任务参数序列
def scheme_tasks_to_byte(j:json):
data=bytearray()
for i in j["TaskArray"]:
data+=scheme_task_to_byte(i)
length=len(data)
if(length<(0x700-4)):
for i in range(0x700-4-length):
data.append(0xff)
return data
# 生成任务id序列
def scheme_taskids_to_byte(j:json):
t=bytearray()
t.append(j["PlanID"]&0xff)
t.append((j["PlanID"]>>8)&0xff)
t.append((j["PlanID"]>>16)&0xff)
t.append((j["PlanID"]>>24)&0xff)
for i in j["TaskArray"]:
t.append(i["TaskID"])
length=len(t)
if(length<(0x100)):
for i in range(0x100-length):
t.append(0xff)
return t
# 根据方案生成小板用的字节数据
def scheme_to_byte(j:json):
t=bytearray()
t+=scheme_taskids_to_byte(j)
t+=scheme_tasks_to_byte(j)
crc=crc32(t)
t.append(crc&0xff)
t.append((crc>>8)&0xff)
t.append((crc>>16)&0xff)
t.append((crc>>24)&0xff)
return t
# int转数组
def arr_from_int(num:int):
return bytearray([num&0xff,(num>>8)&0xff,(num>>16)&0xff,(num>>24)&0xff])
# 提取方案中的范围数据， 先max后min
def scheme_get_task_range(j:json):
t=bytearray()
t+=arr_from_int(j["TaskID"])
t+=arr_from_int(j["TaskIndex"])
t+=arr_from_int(j["ReturnCount"])
t+=arr_from_int(j["ExecuteErrCode"])
index=0
for i in j["TestStandard"]:
t+=arr_from_int(i["Max"])
t+=arr_from_int(i["Min"])
if (index<len(j["ResultErrCode"])):
t+=arr_from_int(j["ResultErrCode"][index])
else:
t+=arr_from_int(0)
index+=1
# 不足16的部分填充为16
n=16-len(j["TestStandard"])
for i in range(n):
t+=arr_from_int(65535)
t+=arr_from_int(0)
t+=arr_from_int(0)
return t
# 根据方案生成主板用的字节数据
def scheme_to_host(j:json):
t=bytearray()
t+=arr_from_int(j["PlanID"])
t+=arr_from_int(j["TimeOutM"])
t+=arr_from_int(len(j["TaskArray"]))
for i in j["TaskArray"]:
t+=scheme_get_task_range(i)
return t
# 发送文件类
class handle:
stat=0
name=""
def __init__(self) -> None:
pass
def get_rate(self):
return self.packet_now*100//self.packet_all
def get_data(self):
if(self.stat==0):
length=len(self.data)
# 1是写，0是读
t=bytearray([1,length&0xff,(length>>8)&0xff,(length>>16)&0xff,(length>>24)&0xff])
t+=self.name.encode()
self.stat=1
return t
elif(self.stat==1):
packet_bytes=200
if(len(self.data)<self.sent_bytes):
packet_bytes=len(self.data)-self.sent_bytes
if(packet_bytes>0):
self.packet_now+=1
t=bytearray()
t.append(2)
t.append(self.packet_now&0xff)
t.append((self.packet_now>>8)&0xff)
t.append(self.packet_all&0xff)
t.append((self.packet_all>>8)&0xff)
t+=self.data[self.sent_bytes:self.sent_bytes+packet_bytes]
self.sent_bytes+=packet_bytes
return t
else:
print("send done.")
return bytearray()
# 设置要发送的文件
def set_file(self,name:str):
self.data=bytearray()
with open(name,"rb") as f:
self.data=f.read()
self.name=f.name.split('/')[-1]
self.stat=0
self.packet_all=(len(self.data)+199)//200
self.sent_bytes=0
self.packet_now=0
self.packet_bytes=200
# 设置要发送的方案
def set_json(self,name:str):
self.data=bytearray()
with open(name,"rb") as f:
json_obj=json.loads(f.read())
d=scheme_to_byte(json_obj)
# print("len=",len(d),d.hex(","))
d+=scheme_to_host(json_obj)
# print("len=",len(d),d.hex(","))
self.data=d
self.name=f.name.split('/')[-1]
self.stat=0
self.packet_all=(len(self.data)+199)//200
self.sent_bytes=0
self.packet_now=0
self.packet_bytes=200
class protu(QObject):
# 进度信号，ip，1~100
rate_signal =pyqtSignal([int])
# 结束信号，ip，成败，描述
end_signal = pyqtSignal([bool,str])
# 接收到数据信号
recv_signal =pyqtSignal([int,bytearray,str])
# 接收到字符串信号
recv_str_signal =pyqtSignal([int,str,str])
hand=handle()
def __init__(self) -> None:
QObject.__init__(self)
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 init(self,com:str):
s=com.split(":")
try:
if(s[0]=="utcp"):
self.ser = utcp(int(s[1]))
else:
bsp=int(s[1])
self.ser = serial.Serial(port=s[0], baudrate=bsp,bytesize=serial.EIGHTBITS,parity=serial.PARITY_NONE,
stopbits=serial.STOPBITS_ONE,timeout=None)
except Exception as e:
print(str(e))
return False
return True
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 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 bytearray(data[8:-2])
else:
return bytearray(data[12:-2])
# 不带校验的接收函数
def recv_(self):
while(self.ser.is_open):
try:
data=self.ser.read(500)
except Exception as a:
# print("err:",str(a))
print("port closed")
return
if(len(data)>0):
t=self.decode(data)
if(self.str_err=="ok"):
self.recv_data+=t
# print("recv",t.hex(","))
# print(type(self.cmd),type(t),type(self.str_err))
self.recv_signal.emit(self.cmd,t,self.str_err)
# self.send_file_next(self.cmd,t,self.str_err)
# print("sent signal---")
else:
print(data.decode("utf-8"))
# 带帧校验的接收函数
def recv(self):
# self.recv_signal.connect(self.send_file_next)
data=bytearray()
data_str=bytearray()
while(self.ser.is_open):
d=bytes()
try:
d=self.ser.read(1)
except Exception as a:
# print("err:",str(a))
print("port closed")
return
data+=d
data_str+=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)):
t=self.decode(data)
self.recv_data+=t
# print("recv",t.hex(","))
self.recv_signal.emit(self.cmd,t,self.str_err)
# self.send_file_next(self.cmd,t,self.str_err)
# print("sent signal---")
data.clear()
data_str.clear()
elif(len(data_str)>=2):
if((data_str[-2]==0x0d) and (data_str[-1]==0x0a)):
data_str=data_str[:-2]
recv_str=""
try:
recv_str=data_str.decode("utf-8")
self.recv_str_signal.emit(0,recv_str,"ok")
data_str.clear()
except Exception as e:
print(str(e))
# print(recv_str)
# else:
# print("len(data)={d1},num_ro_recv={d2}".format(d1=len(data),d2=self.num_to_recv))
def send(self,cmd:int,data:bytearray):
self.cmd=cmd
self.cmd_no+=1
# print("send:",data.hex(","))
self.ser.write(self.encode(data))
def send_str(self,txt:str):
self.ser.write(txt.encode("utf-8"))
def start_recv(self):
self.thread_ = threading.Thread(target=self.recv, args=())
self.thread_.start()
def wait(self):
self.thread_.join()
def close(self):
try:
if(self.ser.is_open):
self.ser.close()
except Exception as e:
print(str(e))
def send_file_next(self,cmd:int,data:bytearray,err:str):
print("send_file_next")
data=self.hand.get_data()
if(len(data)>0):
self.send(cmd,data)
else:
self.close()
def send_file(self,cmd:int,name:str):
self.start_recv()
self.hand.set_file(name)
# self.send_file_next(cmd,bytearray([0]),"ok")
self.recv_signal.emit(cmd,bytearray([0]),"ok")
self.wait()
# 0x30 开始检测
# 0x31 检测上报
# 0x32 读写方案
# 0x33 连接维护
# 0x34 自检数据
# 0x35 自检上报
# 0x36 升级脚本
# 0xed 主板升级
# 0xee 小板升级
# 0x40 电机矫正
# 0x41 小板电阻矫正
# 0x42 小板电阻测量
# 0x43 小板硬件版本
# with open("file/judge.lua","rb") as f:
# send_file("COM5",0x36,f.name.split('/')[-1],f.read())
# with open("file/JQ_JCXB_V54.bin","rb") as f:
# send_file("COM5",0xee,f.name.split('/')[-1],f.read())
# with open("file/checker_ye_cfg.json","rb") as f:
# json_obj=json.loads(f.read())
# d=scheme_to_byte(json_obj)
# print("len=",len(d),d.hex(","))
# d=scheme_to_host(json_obj)
# print("len=",len(d),d.hex(","))
# send_file("COM5",0x32,f.name.split('/')[-1],f.read())
def int2str(num:int):
s=str(num//100)
s=s+str(num%100//10)
s=s+str(num%10)
return s
if __name__ == "__main__":
u=protu()
# u.init("utcp:7777")
# u.send_file(0xee,"file/JQ_JCXB_V54.bin")
# u.send_file(0xed,"../Objects/checker_gen1_app_20230602.bin")
# 设置电阻 矫正值
u.cmd=0x41
data=bytearray([1,0,0x00,2,0,0x00,3,0,0x00,4,0,0x00,5,0,0x00,6,0,0x00,7,0,0x00,8,0,0x00,9,0,0x00,10,0,0x00,11,0,0x00,12,0,0x00,13,0,0x00,14,0,0x00,15,0,0x00,16,0,0x00,17,0,0x00,18,0,0x00,19,0,0x00,20,0,0x00])
# 测量电阻
# u.cmd=0x42
# data=bytearray([0])
# 设置硬件版本号
# u.cmd=0x43
# data=bytearray([1,50,0x00,2,51,0x00,3,52,0x00,4,53,0x00,5,54,0x00,6,55,0x00,7,56,0x00,8,57,0x00,9,58,0x00,10,59,0x00,11,60,0x00,12,61,0x00,13,62,0x00,14,63,0x00,15,64,0x00,16,65,0x00,17,66,0x00,18,67,0x00,19,68,0x00,20,69,0x00])
# 设置电机校正值
# u.cmd=0x40
# data=bytearray([0x01,100,0])
# data=bytearray([0x02]) # 上升
# data=bytearray([0x03]) # 下降
print(u.encode(data).hex(' '))
# with open("file/EX_Coder_Test_2023-07-6.json","rb") as f:
# json_obj=json.loads(f.read())
# d=scheme_to_byte(json_obj)
# print("len=",len(d),d.hex(","))
# d+=scheme_to_host(json_obj)
# print(int2str(20))
# s="{d:03d}".format(d=2)
# print(s)
# with open("file/7-15.json","rb") as f:
# u.cmd=0x22
# p=u.encode(f.read())
# print(p.hex(' '))
# 开始检测
# 59 6d 03 00 30 00 00 60 0f
# 结束应答
# 59 6D 03 00 31 00 00 31 CF

52
updata/select_list.py Normal file
View File

@@ -0,0 +1,52 @@
from PyQt5.QtCore import *
from PyQt5.QtGui import *
from PyQt5.QtWidgets import *
class select_list(QObject):
def __init__(self,father:QDialog,title:str,str_list:list):
QObject.__init__(self)
self.w=QDialog(father)
self.w.resize(800,400)
self.w.setWindowTitle(title)
self.w.setAttribute(Qt.WidgetAttribute.WA_DeleteOnClose)
self.w.setWindowModality(Qt.WindowModality.ApplicationModal)
self.file_list = QListWidget(self.w)
self.file_list.setObjectName(u"str_list")
self.file_list.setGeometry(QRect(20, 20, 760, 360))
self.file_list.setFrameShape(QFrame.Shape.Box)
self.file_list.setMidLineWidth(1)
self.file_list.setVerticalScrollBarPolicy(Qt.ScrollBarPolicy.ScrollBarAsNeeded)
self.file_list.setSelectionMode(QAbstractItemView.SelectionMode.SingleSelection)
self.file_list.itemDoubleClicked.connect(self.item_clicked)
self.item_append(str_list)
def item_append(self,items:list):
for i in items:
# print("add item",i[0])
self.file_list.addItem(i[0])
def item_clicked(self,item:QListWidgetItem ):
self.select_item=item.text()
self.w.done(QDialog.DialogCode.Accepted)
self.w.close()
def show(self):
# self.w.show()
if(self.w.exec()==QDialog.DialogCode.Accepted):
# print(self.select_item)
return self.select_item
return ""

170
updata/udp.py Normal file
View File

@@ -0,0 +1,170 @@
import socket
import threading
from PyQt5.QtCore import *
STR_RED="\033[1;31m"
STR_BLUE="\033[1;34m"
STR_YELLOW="\033[1;33m"
STR_END="\033[0m"
class myudp(QObject):
dst_ip_list=[]
# 进度信号，ip，1~100
rate_signal =pyqtSignal([str,int])
# 结束信号，ip，成败，描述
end_signal = pyqtSignal([str,bool,str])
# 数据信号，有的命令有返回数据
data_signal = pyqtSignal([str,str])
def __init__(self,start,end):
QObject.__init__(self)
self.start=start
self.end=end
def test_comm(self,ip):
# 1.创建socket套接字
udp = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) # AF_INET表示使用ipv4,默认不变，SOCK_DGRAM表示使用UDP通信协议
# 2.绑定端口port
local_addr = ("", 0) # 默认本机任何ip ，指定端口号7878
udp.bind(local_addr) # 绑定端口
udp.settimeout(1.5)
# 3.收发数据
send_data = "whos\n"
udp.sendto(send_data.encode("utf-8"), (ip, 7777)) # 编码成全球统一数据格式，用元组表示接收方ip和port
try:
recv_data = udp.recvfrom(1024) # 定义单次最大接收字节数
recv_msg = recv_data[0] # 接收的元组形式的数据有两个元素，第一个为发送信息
send_addr = recv_data[1] # 元组第二个元素为发信息方的ip以及port
# print ("收到的信息为：", recv_msg.decode("utf-8")) # 默认从windows发出的数据解码要用”gbk”，保证中文不乱码
# print ("发送方地址为：", str(send_addr)) # 强转为字符串输出地址，保证不乱码
str=recv_msg.decode("utf-8").replace("\n","|")
list=str.split(",")
str=list[1].replace(":","_")
self.dst_ip_list.append((send_addr[0],str))
except Exception as reason:
pass
# print(str(reason))
# 5.关闭套接字
udp.close()
# 查找从机
def find(self,ip_prefix):
self.dst_ip_list.clear()
threads = []
print("find ip {} to {}".format(self.start,self.end))
for i in range(self.start,self.end):
ip=ip_prefix+".{}".format(i)
# print(ip)
# 需要注意的一点是，当创建的元组中只有一个字符串类型的元素时，该元素后面必须要加一个逗号,，否则 Python 解释器会将它视为字符串
t = threading.Thread(target=self.test_comm, args=(ip,))
threads.append(t)
t.start()
#等待所有线程任务结束。
for t in threads:
t.join()
# 去重
self.dst_ip_list=list(set(self.dst_ip_list))
# 发送命令，recv_num返回次数
def send(self,ip,cmd,recv_num,timeout):
resoult=True
udp = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
local_addr = ("", 0)
udp.bind(local_addr)
udp.settimeout(timeout)
udp.sendto(cmd.encode("utf-8"), (ip, 7777))
try:
index=0
while index<recv_num:
recv_data = udp.recvfrom(1024)
recv_msg = recv_data[0]
send_addr = recv_data[1]
out=recv_msg.decode("utf-8")
out_list=out.split(",")
ret_list=out_list[0].split(":")
if(ret_list[0]=="ack"):
index+=1
if(ret_list[1]=="1"):
out=("\n"+out).replace("\n",STR_YELLOW+"|"+STR_BLUE)
out=out+STR_END
elif(ret_list[1]=="0"):
out=("\n"+out).replace("\n",STR_YELLOW+"|"+STR_RED)
out=out+STR_END
resoult=False
elif(ret_list[0]=="data"):
index+=1
self.data_signal.emit(ip,out[5:])
elif(ret_list[0]=="rate"):
rate=int(ret_list[1])
self.rate_signal.emit(ip,rate)
else:
out=("\n"+out).replace("\n",STR_YELLOW+"|"+STR_END)
if(ret_list[0]!="rate"):
print(send_addr[0],out)
except Exception as reason:
print(STR_RED,ip,"|"+str(reason),STR_END)
resoult=False
udp.close()
return resoult
# 发送命令列表
def send_list(self,ip,cmd_list):
for i in range(len(cmd_list)):
resoult=self.send(ip,cmd_list[i][0],cmd_list[i][1],cmd_list[i][2])
if(resoult!=True):
err_str="cmd="+cmd_list[i][0]+" filed."
print(STR_RED,ip,"|"+err_str,STR_END)
self.end_signal.emit(ip,False,err_str)
return
# self.rate_signal.emit(ip,(i*100)//len(cmd_list))
self.end_signal.emit(ip,True,"success")
# 广播命令，在find之后调用,str 要发送的数据，recv_num,返回次数
def bordcast(self,cmd_list):
threads = []
print("bordcast:",cmd_list)
print("ip_list:",self.dst_ip_list)
for i in self.dst_ip_list:
ip=i[0]
t = threading.Thread(target=self.send_list, args=(ip,cmd_list))
threads.append(t)
t.start()
for t in threads:
t.join()
# 定义自动发送的命令列表,命令，返回次数，超时
cmd_list=[]
# cmd_list.append(("local setlog filefilter tcp_client",1,1))
# cmd_list.append(("mcu scheme 1,2,3,4,5,6,7,8,9,10",1,30))
# cmd_list.append(("mcu bootinfo 1,2,3,4,5,6,7,8,9,10",1,5))
# cmd_list.append(("mcu start 10",1,37))
# cmd_list.append(("mcu updata 1,2,3,4,5,6,7,8,9,10",1,900))
# cmd_list.append(("whos",1,2))
# cmd_list.append(("mcu updata 10 /home/root/config/JQChecker_0008.bin",1,900))
# cmd_list.append(("local setcfg hostip 192.168.60.202",1,2))
# cmd_list.append(("local setcfg usetcp true",1,2))
# cmd_list.append(("local setcfg uselua true",1,2))
# cmd_list.append(("local setcfg save",1,2))
cmd_list.append(("mcu start 1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16,17,18,19,20",2,50))
# 扫描从机ip地址
# 返回元组列表，ip，local_id
def main():
udp=myudp(1,255)
udp.find("192.168.60")
print(udp.dst_ip_list)
udp.bordcast(cmd_list)
if __name__ == "__main__":
main()

736
updata/updata.py Normal file
View File

@@ -0,0 +1,736 @@
import udp
import ftp
import sys
import os
import shutil
from PyQt5.QtCore import *
from PyQt5.QtGui import *
from PyQt5.QtWidgets import *
import threading
import time
import portalocker
import encodings.idna
import base64
import memory_pic
import ctypes
import mysql
import select_list
ctypes.windll.shell32.SetCurrentProcessExplicitAppUserModelID("myappid")
STR_RED="\033[1;31m"
STR_BLUE="\033[1;34m"
STR_END="\033[0m"
WARN_TXT="警告：\n程序升级会删除主板中以前的程序，为保证升级成功，请耐心等待升级完成，不要退出本界面，不要给主板断电、重启。"
QUESTION_TXT="程序升级会删除主板中以前的程序，非专业人员不要操作。\n\
为保证升级成功，请耐心等待升级完成，不要退出升级界面，期间不要给主板断电、重启。\n\
点击[YES]以进行升级。"
def warn_creat(father:QDialog,text:str,rect:QRect):
lable=QLabel(father)
lable.setWordWrap(True)
lable.setText(text)
lable.setGeometry(rect)
lable.setMargin(10)
lable.setAlignment(Qt.AlignmentFlag.AlignCenter)
lable.setStyleSheet("QLabel{font:20px;color:red;background-color:rgb(62, 213, 255);}")
return lable
class progress_box(QObject):
rate_signal =pyqtSignal([int])
def __init__(self) -> None:
QObject.__init__(self)
def init(self,father:QDialog,text:str,x:int):
# print("text=",text)
self.lable = QLabel(father)
self.lable.setObjectName(u"label")
self.lable.setGeometry(QRect(40, x, 120, 15))
self.lable.setText(text)
self.stat = QLabel(father)
self.stat.setObjectName(u"label")
self.stat.setGeometry(QRect(500, x+20, 120, 15))
self.stat.setText("进行中...")
self.p=QProgressBar(father)
self.p.setValue(0)
self.p.setGeometry(QRect(40,x+20,450,20))
self.rate_signal.connect(self.p.setValue)
def set_rate(self,text:str,rate:int):
if(text==self.lable.text()):
# print(text)
self.rate_signal.emit(rate)
def set_txt(self,text:str,ack:bool,stat:int):
if(text==self.lable.text()):
self.stat.setText(stat)
class data_box(QObject):
def __init__(self) -> None:
QObject.__init__(self)
def init(self,father:QDialog,title:str,text:str,x:int):
# print("text=",text)
self.lable = QLabel(father)
self.lable.setObjectName(u"label")
self.lable.setGeometry(QRect(40, x, 120, 20))
self.lable.setText(title)
self.text = QLabel(father)
self.text.setObjectName(u"text")
self.text.setGeometry(QRect(40,x+20,450,30))
self.text.setText(text)
class updata_dlg(QObject):
# 进度信号，ip，1~100
rate_signal =pyqtSignal([str,int])
# 结束信号，ip，成败，描述
end_signal = pyqtSignal([str,bool,str])
def __init__(self):
QObject.__init__(self)
self.app = QApplication(sys.argv)
self.widget = QWidget()
self.widget.resize(703, 409)
self.widget.setWindowTitle("批检仪程序升级")
self.widget.setWindowFlags(Qt.WindowType.WindowStaysOnTopHint)
self.addrs=""
self.but_y=60
self.but_y_step=40
self.file_list_init()
self.slave_list_init()
self.save_but_init()
self.cmd_but_init()
self.refresh_but_init()
self.sstate_but_init()
self.hand_but_init()
self.addfile_but_init()
self.scheme_but_init()
self.download_but_init()
self.ip_prefix_init()
self.ip_hand_init()
self.channel_init()
self.widget.destroyed.connect(self.quit)
self.scan_file()
self.scan_slave()
Logo = QPixmap()
Logo.loadFromData(base64.b64decode(memory_pic.icon_ico))
icon = QIcon()
icon.addPixmap(Logo, QIcon.Normal, QIcon.Off)
self.widget.setWindowIcon(icon)
def quit(self):
# 程序退出
qApp.exit(1)
# 初始化文件列表
def file_list_init(self):
self.file_list = QListWidget(self.widget)
self.file_list.setObjectName(u"file_list")
self.file_list.setGeometry(QRect(25, 230, 531, 141))
self.file_list.setFrameShape(QFrame.Shape.Box)
self.file_list.setMidLineWidth(1)
self.file_list.setVerticalScrollBarPolicy(Qt.ScrollBarPolicy.ScrollBarAsNeeded)
self.file_list.setSelectionMode(QAbstractItemView.SelectionMode.MultiSelection)
self.file_list_label = QLabel(self.widget)
self.file_list_label.setObjectName(u"label")
self.file_list_label.setGeometry(QRect(30, 210, 531, 15))
self.file_list_label.setText("默认文件列表（选中以发送）：")
self.file_list_label.setToolTip("请选择要升级的文件，只有选中的文件会发送到主板中。")
self.file_list.setToolTip("请选择要升级的文件，只有选中的文件会发送到主板中。")
# self.file_list_label.setToolTipDuration(1)
# 初始化从机列表
def slave_list_init(self):
self.slave_list = QListWidget(self.widget)
self.slave_list.setObjectName(u"slave_list")
self.slave_list.setGeometry(QRect(25, 60, 531, 141))
self.slave_list.setFrameShape(QFrame.Shape.Box)
self.slave_list.setMidLineWidth(1)
self.slave_list.setVerticalScrollBarPolicy(Qt.ScrollBarPolicy.ScrollBarAsNeeded)
self.slave_list.setSelectionMode(QAbstractItemView.SelectionMode.MultiSelection)
self.slave_list.doubleClicked.connect(self.slave_list.editItem)
self.slave_list_label = QLabel(self.widget)
self.slave_list_label.setObjectName(u"label")
self.slave_list_label.setGeometry(QRect(30, 40, 200, 15))
self.slave_list_label.setText("在线主板列表（选中以发送）：")
self.slave_list_label.setToolTip("请选择要升级的主板设备，只有选中的主板会升级。")
self.slave_list.setToolTip("请选择要升级的主板设备，只有选中的主板会升级。")
# self.slave_list_label.setToolTipDuration(1)
# 初始化手动添加IP按钮
def hand_but_init(self):
self.hand_but = QPushButton(self.widget)
self.hand_but.setObjectName(u"hand_but")
self.hand_but.setGeometry(QRect(590, self.but_y, 93, 28))
self.but_y+=self.but_y_step
self.hand_but.setText("手动添加IP")
self.hand_but.clicked.connect(self.hand_but_clicked)
self.hand_but.setToolTip("请先在[手动添加IP地址]输入框中输入要添加的地址，然后点击此按钮添加到设备列表中。")
# self.hand_but.setToolTipDuration(1)
# 初始化发送文件按钮
def save_but_init(self):
self.save_but = QPushButton(self.widget)
self.save_but.setObjectName(u"save_but")
self.save_but.setGeometry(QRect(590, self.but_y, 93, 28))
self.but_y+=self.but_y_step
self.save_but.setText("发送文件")
self.save_but.clicked.connect(self.save_but_clicked)
self.save_but.setToolTip("请先选中要升级的主板和文件，然后点击此按钮发送到设备中。")
# self.save_but.setToolTipDuration(1)
# 初始化发送命令按钮
def cmd_but_init(self):
self.cmd_but = QPushButton(self.widget)
self.cmd_but.setObjectName(u"save_but")
self.cmd_but.setGeometry(QRect(590, self.but_y, 93, 28))
self.but_y+=self.but_y_step
self.cmd_but.setText("升级MCU")
self.cmd_but.clicked.connect(self.cmd_but_clicked)
self.cmd_but.setToolTip("请先将要升级的MCU程序发送到主板中，然后点击此按钮开始升级MCU程序。")
# self.cmd_but.setToolTipDuration(1)
# 初始化刷新按钮
def refresh_but_init(self):
self.refresh_but = QPushButton(self.widget)
self.refresh_but.setObjectName(u"save_but")
self.refresh_but.setGeometry(QRect(590, self.but_y, 93, 28))
self.but_y+=self.but_y_step
self.refresh_but.setText("刷新IP地址")
self.refresh_but.clicked.connect(self.refresh_but_clicked)
self.refresh_but.setToolTip("点击此按钮刷新主板列表。")
# self.refresh_but.setToolTipDuration(1)
# 初始化在线状态按钮
def sstate_but_init(self):
self.sstate_but = QPushButton(self.widget)
self.sstate_but.setObjectName(u"sstate_but")
self.sstate_but.setGeometry(QRect(590, self.but_y, 93, 28))
self.but_y+=self.but_y_step
self.sstate_but.setText("MCU在线状态")
self.sstate_but.clicked.connect(self.sstate_but_clicked)
self.sstate_but.setToolTip("点击此按钮查看小板在线情况，显示在线小板的地址。")
# self.sstate_but.setToolTipDuration(1)
# 初始化添加文件按钮
def addfile_but_init(self):
self.addfile_but = QPushButton(self.widget)
self.addfile_but.setObjectName(u"addfile_but")
self.addfile_but.setGeometry(QRect(590, self.but_y, 93, 28))
self.but_y+=self.but_y_step
self.addfile_but.setText("添加文件")
self.addfile_but.clicked.connect(self.addfile_but_clicked)
self.addfile_but.setToolTip("如果文件列表中不存在要升级的文件，点击此按钮从外部添加。")
# self.addfile_but.setToolTipDuration(1)
# 初始化升级方案按钮
def scheme_but_init(self):
self.scheme_but = QPushButton(self.widget)
self.scheme_but.setObjectName(u"scheme_but")
self.scheme_but.setGeometry(QRect(590, self.but_y, 93, 28))
self.but_y+=self.but_y_step
self.scheme_but.setText("升级方案")
self.scheme_but.clicked.connect(self.scheme_but_clicked)
self.scheme_but.setToolTip("请先将要升级的方案文件发送到主板中，然后点击此按钮开始升级方案到MCU。")
# self.addfile_but.setToolTipDuration(1)
# 初始化下载文件按钮
def download_but_init(self):
self.download_but = QPushButton(self.widget)
self.download_but.setObjectName(u"download_but")
self.download_but.setGeometry(QRect(590, self.but_y, 93, 28))
self.but_y+=self.but_y_step
self.download_but.setText("下载文件")
self.download_but.clicked.connect(self.download_but_clicked)
# ip前缀
def ip_prefix_init(self):
self.ip_prefix = QLineEdit(self.widget)
self.ip_prefix.setObjectName(u"ip_prefix")
self.ip_prefix.setGeometry(QRect(110, 10, 113, 21))
self.ip_prefix.setText("192.168.80")
self.ip_prefix_label = QLabel(self.widget)
self.ip_prefix_label.setObjectName(u"label")
self.ip_prefix_label.setGeometry(QRect(30, 10, 72, 21))
self.ip_prefix_label.setText("IP前缀：")
self.ip_prefix_label.setAlignment(Qt.AlignmentFlag.AlignCenter)
self.ip_prefix_label.setToolTip("如果扫描不到设备，请确保局域网网段与此相符，将此IP前缀改为和局域网相同。")
self.ip_prefix.setToolTip("如果扫描不到设备，请确保局域网网段与此相符，将此IP前缀改为和局域网相同。")
# self.ip_prefix_label.setToolTipDuration(1)
# 手动添加ip
def ip_hand_init(self):
self.ip_hand = QLineEdit(self.widget)
self.ip_hand.setObjectName(u"ip_hand")
self.ip_hand.setGeometry(QRect(360, 10, 120, 21))
ip=self.ip_prefix.text()
self.ip_hand.setText(ip+".81")
self.ip_hand_label = QLabel(self.widget)
self.ip_hand_label.setObjectName(u"ip_hand_label")
self.ip_hand_label.setGeometry(QRect(230, 10, 120, 21))
self.ip_hand_label.setText("手动添加IP地址：")
self.ip_hand_label.setAlignment(Qt.AlignmentFlag.AlignCenter)
self.ip_hand_label.setToolTip("如果扫描不到设备，可以尝试手动添加设备的IP地址，然后升级。")
self.ip_hand.setToolTip("如果扫描不到设备，可以尝试手动添加设备的IP地址，然后升级。")
# self.ip_hand_label.setToolTipDuration(1)
# 选择通道数
def channel_init(self):
self.channel = QComboBox(self.widget)
self.channel.setObjectName(u"channel")
self.channel.setGeometry(QRect(560, 10, 100, 25))
self.channel.setEditable(True)
self.channel.addItem("10通道")
self.channel.addItem("20通道")
self.channel.currentIndexChanged.connect(self.channel_changed)
self.channel_changed(0)
# self.channel_label = QLabel(self.widget)
# self.channel_label.setObjectName(u"label")
# self.channel_label.setGeometry(QRect(410, 16, 72, 15))
# self.channel_label.setText("波特率:")
# 显示消息框
def show_msg(self,msg:str):
m=QMessageBox(self.widget)
m.setText(msg)
m.setWindowTitle("提示")
m.show()
def show_question(self,title:str,msg:str):
ret=QMessageBox.question(self.widget,title,msg,QMessageBox.StandardButton.Yes|QMessageBox.StandardButton.No,
QMessageBox.StandardButton.No)
if(ret==QMessageBox.StandardButton.Yes):
return True
else:
return False
# 找到已选择的文件
def get_selected_file_by_type(self,type:str):
file_list=[]
items=self.file_list.selectedItems()
for i in items:
if(i.text()[-len(type):]==type):
file_list.append(i.text())
if(len(file_list)!=1):
self.show_msg("请选择一个并且只选择一个 "+type+" 文件")
return ""
return file_list[0]
# 获取已选择的文件列表
def get_selected_fils(self):
file_list=[]
items=self.file_list.selectedItems()
if(len(items)==0):
self.show_msg("请选择至少一个文件")
return []
have_elf=False
have_bin=False
have_lua=False
for i in items:
if(i.text()[-4:]==".elf"):
if(have_elf==True):
self.show_msg("只可选择一个 .elf 文件")
return []
have_elf=True
if(i.text()[-4:]==".bin"):
if(have_bin==True):
self.show_msg("只可选择一个 .bin 文件")
return []
have_bin=True
if(i.text()[-4:]==".lua"):
if(have_lua==True):
self.show_msg("只可选择一个 .lua 文件")
return []
have_lua=True
file_list.append(i.text())
return file_list
def get_selected_slave(self):
slave_list=[]
items=self.slave_list.selectedItems()
if(len(items)==0):
self.show_msg("请选择至少一个ip地址")
return []
for i in items:
str_list=i.text().split(",")
slave_list.append((str_list[0],str_list[1]))
return slave_list
# 根据文件列表生成目标列表
def build_dst_list(self,file_list):
dst_list=[]
for i in file_list:
if(i[-4:]==".elf"):
dst_list.append("/usr/local/QDesktop-fb")
elif(i[-4:]==".dtb"):
dst_list.append("/run/media/mmcblk2p2/"+i)
elif(i[-5:]==".json"):
dst_list.append("/home/root/config/"+"checker_ye_cfg.json")
elif(i[-4:]==".lua"):
dst_list.append("/home/root/config/"+"judge.lua")
elif(i[-4:]==".axf"):
dst_list.append("/lib/firmware/"+"checker_m4.axf")
else:
dst_list.append("/home/root/config/"+i)
return dst_list
def hand_but_clicked(self):
print("hand_but clicked.")
ip=self.ip_hand.text()
self.slave_list.addItem(ip+",local_id_0")
def channel_changed(self,index:int):
s=self.channel.itemText(index)
print("channel_changed,str=",s)
if(s=="10通道"):
self.addrs="1,2,3,4,5,6,7,8,9,10 "
elif(s=="20通道"):
self.addrs="1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16,17,18,19,20 "
print("addrs:",self.addrs)
# 发送文件按钮按下
def save_but_clicked(self):
print("save_but_clicked.")
path = self.getpath()+"file\\"
file_list=self.get_selected_fils()
if(len(file_list)==0):
return
dst_list=self.build_dst_list(file_list)
for i in range(len(file_list)):
file_list[i]=path+file_list[i]
for i in zip(file_list,dst_list):
print("file:",i[0],i[1])
slave_list=self.get_selected_slave()
if(len(slave_list)==0):
return
if(self.show_question("警告",QUESTION_TXT)==False):
return
print("slaves:",slave_list)
w=QDialog(self.widget)
warn_creat(w,WARN_TXT,QRect(0,0,700-100,90))
w.resize(700-100, len(slave_list)*40+20+100)
w.setWindowTitle("上传文件")
self.updata(slave_list,dst_list,file_list)
self.creat_progress(w,100,slave_list)
w.show()
# 创建进度条
def creat_progress(self,father:QDialog,y_off:int,ip_list:list):
self.probar_list=[]
for i in range(len(ip_list)):
p=progress_box()
p.init(father,ip_list[i][0],+i*40+y_off)
self.rate_signal.connect(p.set_rate)
self.end_signal.connect(p.set_txt)
self.probar_list.append(p)
# 发送命令按钮按下
def cmd_but_clicked(self):
print("cmd_but clicked.")
slave_list=self.get_selected_slave()
if(len(slave_list)==0):
return
print("slaves:",slave_list)
file=self.get_selected_file_by_type(".bin")
if(len(file)==0):
return
print("file:",file)
w=QDialog(self.widget)
w.resize(700-100, len(slave_list)*40+20)
w.setWindowTitle("升级mcu")
self.updata_mcu(slave_list,file)
self.creat_progress(w,0,slave_list)
w.show()
# 升级方案按钮按下
def scheme_but_clicked(self):
print("scheme_but clicked.")
slave_list=self.get_selected_slave()
if(len(slave_list)==0):
return
print("slaves:",slave_list)
w=QDialog(self.widget)
w.resize(700-100, len(slave_list)*40+20)
w.setWindowTitle("升级mcu")
self.scheme_mcu(slave_list)
self.creat_progress(w,0,slave_list)
w.show()
def refresh_but_clicked(self):
print("refresh_but clicked.")
self.slave_list.clear()
self.scan_slave()
def sstate_but_clicked(self):
print("sstate_but clicked.")
slave_list=self.get_selected_slave()
if(len(slave_list)==0):
return
self.data_list=[]
self.comm_test(slave_list)
def addfile_but_clicked(self):
print("addfile_but clicked")
fileName,fileType = QFileDialog.getOpenFileNames(None, "选取文件", os.getcwd(),
"主板程序(*.elf);;主板m4程序(*.axf);;小板程序(*.bin);;检测方案(*.json);;判定脚本(*.lua);;任意文件(*)")
print(fileName,fileType)
path=self.getpath()+"file\\"
for i in fileName:
shutil.copy(i,path+i.split("/")[-1])
self.scan_file()
def download_but_clicked(self):
sql=mysql.sql()
if(sql.init("")):
str_list=sql.show_tables()
sel=select_list.select_list(self.widget,"选择文件目录",str_list)
path=sel.show()
# print("path:",path)
if(len(path)<=0):
print("not select any item,break.")
return
sql.table_name=path
items=sql.items()
str_list=[]
for i in items:
item_name=i[2].replace("\\","/")
item_name=item_name.split("/")[-1]
if(i[4]!=None):
s=str(i[0])+"|"+i[1]+"|"+item_name+"|"+i[4]
else:
s=str(i[0])+"|"+i[1]+"|"+item_name
# print(s)
str_list.append((s,))
sel=select_list.select_list(self.widget,"选择文件",str_list)
item=sel.show()
# print("item:",item)
if(len(item)<=0):
print("not select any item,break.")
return
item=item.split("|")
file_name=sql.download(int(item[0]))
dst="file/"+file_name.split("/")[-1]
shutil.copy(file_name,dst)
self.scan_file()
# 开始运行
def run(self):
self.widget.show()
sys.exit(self.app.exec())
# 扫描文件
def scan_file(self):
self.file_list.clear()
self.file_list.addItems(self.find_type([".sh",".elf",".bin",".lua",".json",".dtb",".axf"]))
# 扫描从机
def scan_slave(self):
u=udp.myudp(1,255)
ip_prefix=self.ip_prefix.text()
u.find(ip_prefix)
ip_list=u.dst_ip_list
# if(len(ip_list)==0):
# ret=self.show_question("提示","未扫描到从机，是否自动添加？")
# if(ret==True):
# ip_list.append((ip_prefix+".81","local_id_1"))
# ip_list.append((ip_prefix+".82","local_id_2"))
# ip_list.append((ip_prefix+".83","local_id_3"))
# ip_list.append((ip_prefix+".84","local_id_4"))
# ip_list.append((ip_prefix+".85","local_id_5"))
# ip_list.append((ip_prefix+".86","local_id_6"))
# ip_list.append((ip_prefix+".87","local_id_7"))
# ip_list.append((ip_prefix+".88","local_id_8"))
list_str=[]
for i in u.dst_ip_list:
list_str.append(i[0]+','+i[1])
self.slave_list.addItems(list_str)
# 扫描指定类型的文件
def find_type(self,types:list):
path = self.getpath()+"file\\"
if not os.path.exists(path):
os.makedirs(path)
list=os.listdir(path)
file_list=[]
for t in types:
for i in list:
if(len(t)>0):
if(i[-len(t):]==t):
file_list.append(i)
else:
file_list.append(i)
return file_list
# 获得文件绝对路径
def getpath(self):
path=os.path.abspath(sys.argv[0])
list_str=path.split("\\")
return path.replace(list_str[-1],"")
# 调用此函数开始升级
def updata(self,ip_list,dst_list,src_list):
t = threading.Thread(target=self.updata_thread, args=(ip_list,dst_list,src_list))
# t = threading.Thread(target=self.thread_test, args=(ip_list,dst_list,src_list))
t.start()
# 开始升级mcu
def updata_mcu(self,ip_list,file):
u=udp.myudp(1,255)
u.dst_ip_list=ip_list
u.rate_signal.connect(self.rate_slot)
u.end_signal.connect(self.end_slot)
print("addrs:",self.addrs)
updata_cmd="mcu updata "+self.addrs+' '
updata_cmd+="/home/root/config/"+file
cmd_list=[]
cmd_list.append((updata_cmd,1,900))
t = threading.Thread(target=u.bordcast, args=(cmd_list,))
t.start()
# 开始升级方案
def scheme_mcu(self,ip_list):
u=udp.myudp(1,255)
u.dst_ip_list=ip_list
u.rate_signal.connect(self.rate_slot)
u.end_signal.connect(self.end_slot)
updata_cmd="mcu scheme "+self.addrs+' '
cmd_list=[]
cmd_list.append((updata_cmd,1,9))
t = threading.Thread(target=u.bordcast, args=(cmd_list,))
t.start()
# 小板通信测试
def comm_test(self,ip_list):
u=udp.myudp(1,255)
u.dst_ip_list=ip_list
# u.rate_signal.connect(self.rate_slot)
# u.end_signal.connect(self.end_slot)
u.data_signal.connect(self.data_slot)
updata_cmd="mcu comm_test"
cmd_list=[]
cmd_list.append((updata_cmd,2,5))# 两次返回，五秒超时
t = threading.Thread(target=u.bordcast, args=(cmd_list,))
t.start()
# 测试线程
def thread_test(self,ip_list,dst_list,src_list):
for i in range(100):
for ip in ip_list:
self.rate_slot(ip[0],i+1)
time.sleep(0.02)
def updata_thread(self,ip_list,dst_list,src_list):
threads=[]
for i in ip_list:
t = threading.Thread(target=self.save_file, args=(i[0],True,dst_list ,src_list))
threads.append(t)
t.start()
#等待所有线程任务结束。
for t in threads:
t.join()
def end_slot(self,ip,ack,err):
# print(ip,ack,err)
if(ack==False):
self.show_msg(ip+":"+err)
else:
self.end_signal.emit(ip,True,"完成")
def rate_slot(self,ip,rate):
# print("rate signal:",ip,rate)
self.rate_signal.emit(ip,rate)
def data_slot(self,ip,text):
self.data_list.append((ip,text))
# 全部返回完时显示数据
slave_len=len(self.get_selected_slave())
if(len(self.data_list)>=slave_len):
w=QDialog(self.widget)
w.resize(703-150, slave_len*50+20)
w.setWindowTitle("返回数据展示")
self.creat_databoxs(w,self.data_list)
w.show()
# 创建数据显示
def creat_databoxs(self,father:QDialog,data_list:list):
self.datab_list=[]
for i in range(len(data_list)):
p=data_box()
p.init(father,data_list[i][0],data_list[i][1],+i*50)
self.datab_list.append(p)
def save_file(self,ip,restart:bool,dst_list,src_list):
try:
f=ftp.myftp(ip,"root","")
# f.end_signal.connect(self.end_slot)
f.rate_signal.connect(self.rate_slot)
if(restart==True):
print(ip,"|stop app.")
f.send_cmd("systemctl stop atk-qtapp-start.service")
f.put_file(ip,dst_list,src_list)
if(restart==True):
print(ip,"|start app.")
f.send_cmd("systemctl restart atk-qtapp-start.service")
f.send_cmd("sync",sleep_s=3)
f.close()
self.end_signal.emit(ip,True,"完成")
# print(ip,"|save file end.")
except Exception as r:
err_str=str(r)
print(STR_RED,ip,"|"+"try exception "+err_str,STR_END)
# self.end_slot(ip,False,err_str)
class locker():
def _get_lock(self):
file_name = os.path.basename(__file__)
# linux等平台依然使用标准的/var/run，其他nt等平台使用当前目录
if os.name == "posix":
lock_file_name = f"/var/run/{file_name}.pid"
else:
lock_file_name = f"{file_name}.pid"
self.fd = open(lock_file_name, "w")
try:
portalocker.lock(self.fd, portalocker.LOCK_EX | portalocker.LOCK_NB)
# 将当前进程号写入文件
# 如果获取不到锁上一步就已经异常了，所以不用担心覆盖
self.fd.writelines(str(os.getpid()))
# 写入的数据太少，默认会先被放在缓冲区，我们强制同步写入到文件
self.fd.flush()
except:
print(f"{file_name} have another instance running.")
exit(1)
def __init__(self):
self._get_lock()
# 和fcntl有点区别，portalocker释放锁直接有unlock()方法
# 还是一样，其实并不需要在最后自己主动释放锁
def __del__(self):
portalocker.unlock(self.fd)
if __name__ == "__main__":
lock=locker()
dlg=updata_dlg()
dlg.run()

554
updata/updata_uart.py Normal file
View File

@@ -0,0 +1,554 @@
import sys
import os
import shutil
from PyQt5.QtCore import *
from PyQt5.QtGui import *
from PyQt5.QtWidgets import *
import threading
import time
import portalocker
import serial
import serial.tools.list_ports
import prottcp
import mysql
import select_list
STR_RED="\033[1;31m"
STR_BLUE="\033[1;34m"
STR_END="\033[0m"
class progress_box(QObject):
def __init__(self) -> None:
QObject.__init__(self)
def init(self,father:QDialog,text:str,x:int):
# print("text=",text)
self.lable = QLabel(father)
self.lable.setObjectName(u"label")
self.lable.setGeometry(QRect(40, x, 120, 15))
self.lable.setText(text)
self.p=QProgressBar(father)
self.p.setValue(0)
self.p.setGeometry(QRect(40,x+20,450,20))
def set_rate(self,rate:int):
# print(text)
self.p.setValue(rate)
class data_box(QObject):
def __init__(self) -> None:
QObject.__init__(self)
self.select_item=""
def init(self,father:QDialog,title:str,text:str,x:int):
# print("text=",text)
self.lable = QLabel(father)
self.lable.setObjectName(u"label")
self.lable.setGeometry(QRect(40, x, 120, 20))
self.lable.setText(title)
self.text = QLabel(father)
self.text.setObjectName(u"text")
self.text.setGeometry(QRect(40,x+20,450,30))
self.text.setText(text)
class updata_dlg(QObject):
# 进度信号，ip，1~100
rate_signal =pyqtSignal([int])
# 结束信号，ip，成败，描述
end_signal = pyqtSignal([bool,str])
failed_signal = pyqtSignal()
def __init__(self):
QObject.__init__(self)
self.app = QApplication(sys.argv)
self.widget = QWidget()
self.widget.setAttribute(Qt.WidgetAttribute.WA_DeleteOnClose)
self.widget.resize(703, 409)
self.widget.setWindowTitle("赋码仪程序升级")
self.file_list_init()
self.com_but_init()
self.save_but_init()
self.updatas_but_init()
# self.cmd_but_init()
self.sstate_but_init()
self.scheme_but_init()
self.com_init()
self.combsp_init()
self.download_but_init()
self.widget.destroyed.connect(self.quit)
self.failed_signal.connect(self.updata_failed)
self.cmd=0
self.port_is_open=False
def quit(self):
self.close_port()
# 程序退出
qApp.exit(1)
# 初始化文件列表
def file_list_init(self):
self.file_list = QListWidget(self.widget)
self.file_list.setObjectName(u"file_list")
self.file_list.setGeometry(QRect(25, 60, 531, 310))
self.file_list.setFrameShape(QFrame.Shape.Box)
self.file_list.setMidLineWidth(1)
self.file_list.setVerticalScrollBarPolicy(Qt.ScrollBarPolicy.ScrollBarAsNeeded)
self.file_list.setSelectionMode(QAbstractItemView.SelectionMode.SingleSelection)
self.file_list.itemDoubleClicked.connect(self.item_clicked)
# 初始化打开端口按钮
def com_but_init(self):
self.com_but = QPushButton(self.widget)
self.com_but.setObjectName(u"com_but")
self.com_but.setGeometry(QRect(590, 10, 93, 28))
self.com_but.setText("打开端口")
self.com_but.clicked.connect(self.com_but_clicked)
# 初始化发送文件按钮
def save_but_init(self):
self.save_but = QPushButton(self.widget)
self.save_but.setObjectName(u"save_but")
self.save_but.setGeometry(QRect(590, 60, 93, 28))
self.save_but.setText("发送文件")
self.save_but.clicked.connect(self.save_but_clicked)
# 初始化发送命令按钮
def cmd_but_init(self):
self.cmd_but = QPushButton(self.widget)
self.cmd_but.setObjectName(u"save_but")
self.cmd_but.setGeometry(QRect(590, 100, 93, 28))
self.cmd_but.setText("升级MCU")
self.cmd_but.clicked.connect(self.cmd_but_clicked)
# 初始化升级小板按钮
def updatas_but_init(self):
self.updatas_but = QPushButton(self.widget)
self.updatas_but.setObjectName(u"updatas_but")
self.updatas_but.setGeometry(QRect(590, 100, 93, 28))
self.updatas_but.setText("升级小板")
self.updatas_but.clicked.connect(self.updatas_but_clicked)
# 初始化在线状态按钮
def sstate_but_init(self):
self.sstate_but = QPushButton(self.widget)
self.sstate_but.setObjectName(u"sstate_but")
self.sstate_but.setGeometry(QRect(590, 140, 93, 28))
self.sstate_but.setText("主板参数")
self.sstate_but.clicked.connect(self.sstate_but_clicked)
# 初始化方案状态按钮
def scheme_but_init(self):
self.scheme_but = QPushButton(self.widget)
self.scheme_but.setObjectName(u"scheme_but")
self.scheme_but.setGeometry(QRect(590, 180, 93, 28))
self.scheme_but.setText("方案参数")
self.scheme_but.clicked.connect(self.scheme_but_clicked)
# 初始化下载文件按钮
def download_but_init(self):
self.download_but = QPushButton(self.widget)
self.download_but.setObjectName(u"download_but")
self.download_but.setGeometry(QRect(590, 220, 93, 28))
self.download_but.setText("下载文件")
self.download_but.clicked.connect(self.download_but_clicked)
# com口
def com_init(self):
self.com = QComboBox(self.widget)
self.com.setObjectName(u"com")
self.com.setGeometry(QRect(85, 10, 300, 25))
self.com.setEditable(True)
self.com.currentIndexChanged.connect(self.com_changed)
self.com.addItem("utcp:7777")
ports_list = list(serial.tools.list_ports.comports())
for comport in ports_list:
# print(comport.name,comport.description)
self.com.addItem(comport.name+":"+comport.description)
self.com_label = QLabel(self.widget)
self.com_label.setObjectName(u"label")
self.com_label.setGeometry(QRect(30, 16, 72, 15))
self.com_label.setText("COM口:")
# 选择波特率
def combsp_init(self):
self.combsp = QComboBox(self.widget)
self.combsp.setObjectName(u"combsp")
self.combsp.setGeometry(QRect(470, 10, 80, 25))
self.combsp.setEditable(True)
self.combsp.addItem("115200")
self.combsp.addItem("57600")
self.combsp.addItem("38400")
self.combsp.addItem("9600")
self.combsp_label = QLabel(self.widget)
self.combsp_label.setObjectName(u"label")
self.combsp_label.setGeometry(QRect(410, 16, 72, 15))
self.combsp_label.setText("波特率:")
# 显示消息框
def show_msg(self,msg:str):
m=QMessageBox(self.widget)
m.setAttribute(Qt.WidgetAttribute.WA_DeleteOnClose)
m.setText(msg)
m.setWindowTitle("提示")
m.show()
# 找到已选择的文件
def get_selected_file_by_type(self,type:str):
file_list=[]
items=self.file_list.selectedItems()
for i in items:
if(i.text()[-len(type):]==type):
file_list.append(i.text())
if(len(file_list)!=1):
self.show_msg("请选择一个并且只选择一个 "+type+" 文件")
return ""
return file_list[0]
# 获取已选择的文件列表
def get_selected_fils(self):
file_list=[]
items=self.file_list.selectedItems()
if(len(items)==0):
self.show_msg("请选择至少一个文件")
return []
have_elf=False
have_bin=False
have_lua=False
for i in items:
if(i.text()[-4:]==".elf"):
if(have_elf==True):
self.show_msg("只可选择一个 .elf 文件")
return []
have_elf=True
if(i.text()[-4:]==".bin"):
if(have_bin==True):
self.show_msg("只可选择一个 .bin 文件")
return []
have_bin=True
if(i.text()[-4:]==".lua"):
if(have_lua==True):
self.show_msg("只可选择一个 .lua 文件")
return []
have_lua=True
file_list.append(i.text())
return file_list
def item_clicked(self,item:QListWidgetItem ):
print("item clicked.")
print("slected item is",item.text())
def com_but_clicked(self):
print("com but clicked")
if(self.port_is_open==False):
self.open_port()
else:
self.close_port()
def com_changed(self,index:int):
print("com changed")
self.close_port()
def get_cmd(self,file:str):
l=[(".bin",0xee),(".pkt",0xed),(".json",0x32)]
for i in l:
if(file.endswith(i[0])):
return i[1]
return 0
# 发送文件按钮按下
def save_but_clicked(self):
print("save_but_clicked.")
path = self.getpath()+"file\\"
file_list=self.get_selected_fils()
if(len(file_list)==0):
return
self.cmd=self.get_cmd(file_list[0])
print("cmd=",self.cmd)
w=QDialog(self.widget)
w.resize(703-150, 1*40+20)
w.setWindowTitle("上传文件")
w.setAttribute(Qt.WidgetAttribute.WA_DeleteOnClose)
self.handle_=prottcp.handle()
if(file_list[0].endswith(".json")):
self.handle_.set_json("file/"+file_list[0])
else:
self.handle_.set_file("file/"+file_list[0])
d=self.handle_.get_data()
try:
self.port.send(self.cmd,d)
except Exception as e:
print("com not open")
self.show_msg("端口未打开")
del self.handle_
w.close()
return
self.creat_progress(w)
w.show()
# self.failed_signal.connect(w.close)
# w.destroyed.connect(self.close_port)
def updata_failed(self):
self.show_msg("打开端口失败")
# 创建进度条
def creat_progress(self,father:QDialog):
self.probar_list=[]
for i in range(1):
p=progress_box()
p.init(father,"发送文件",+i*40)
self.rate_signal.connect(p.set_rate)
self.probar_list.append(p)
# 发送命令按钮按下
def cmd_but_clicked(self):
print("cmd_but clicked.")
file=self.get_selected_file_by_type(".bin")
if(len(file)==0):
return
print("file:",file)
w=QDialog(self.widget)
w.resize(703-150, 1*40+20)
w.setWindowTitle("升级mcu")
self.updata_mcu(file)
self.creat_progress(w)
w.show()
def sstate_but_clicked(self):
print("sstate_but clicked.")
try:
self.port.send_str("sysinfo")
except Exception as e:
print("com not open")
print(str(e))
def scheme_but_clicked(self):
print("scheme_but clicked.")
try:
self.port.send_str("scheme")
except Exception as e:
print("com not open")
print(str(e))
def updatas_but_clicked(self):
print("updatas_but clicked.")
try:
self.port.send_str("updatas 1,2,3,4,5,6,7,8,9,10")
self.show_msg("已发送升级指令，请留意小板升级情况")
except Exception as e:
self.show_msg("命令发送失败")
print("com not open")
print(str(e))
def download_but_clicked(self):
sql=mysql.sql()
if(sql.init("")):
str_list=sql.show_tables()
sel=select_list.select_list(self.widget,"选择文件目录",str_list)
path=sel.show()
# print("path:",path)
if(len(path)<=0):
print("not select any item,break.")
return
sql.table_name=path
items=sql.items()
str_list=[]
for i in items:
item_name=i[2].replace("\\","/")
item_name=item_name.split("/")[-1]
if(i[4]!=None):
s=str(i[0])+"|"+i[1]+"|"+item_name+"|"+i[4]
else:
s=str(i[0])+"|"+i[1]+"|"+item_name
# print(s)
str_list.append((s,))
sel=select_list.select_list(self.widget,"选择文件",str_list)
item=sel.show()
# print("item:",item)
if(len(item)<=0):
print("not select any item,break.")
return
item=item.split("|")
file_name=sql.download(int(item[0]))
dst="file/"+file_name.split("/")[-1]
shutil.copy(file_name,dst)
self.scan_file()
# 开始运行
def run(self):
self.widget.show()
sys.exit(self.app.exec())
def set_port_state(self,state:bool):
self.port_is_open=state
if(state==True):
self.com_but.setText("关闭端口")
else:
self.com_but.setText("打开端口")
# 扫描文件
def scan_file(self):
self.file_list.clear()
self.file_list.addItems(self.find_type([".bin",".json",".pkt"]))
# 扫描指定类型的文件
def find_type(self,types:list):
path = self.getpath()+"file\\"
if not os.path.exists(path):
os.makedirs(path)
list=os.listdir(path)
file_list=[]
for t in types:
for i in list:
if(len(t)>0):
if(i[-len(t):]==t):
file_list.append(i)
else:
file_list.append(i)
return file_list
# 获得文件绝对路径
def getpath(self):
path=os.path.abspath(sys.argv[0])
list_str=path.split("\\")
return path.replace(list_str[-1],"")
# 调用此函数打开通信
def open_port(self):
t = threading.Thread(target=self.com_thread, args=())
t.start()
def com_thread(self):
self.port=prottcp.protu()
item=self.com.itemText(self.com.currentIndex())
bsp=self.combsp.itemText(self.combsp.currentIndex())
com=item.split(":")[0]
if(com!="utcp"):
item=com+":"+bsp
print("item text:",item)
if(self.port.init(item)==False):
print("init port failed.")
self.failed_signal.emit()
self.set_port_state(False)
return
else:
print("init port success.")
self.set_port_state(True)
self.port.recv_signal.connect(self.recv_slot)
self.port.recv_str_signal.connect(self.recv_str_slot)
self.port.start_recv()
self.port.wait()
def close_port(self):
print("close port")
self.set_port_state(False)
try:
self.port.close()
self.port.recv_signal.disconnect(self.recv_slot)
self.port.recv_str_signal.disconnect(self.recv_str_slot)
except Exception as e:
pass
def recv_str_slot(self,cmd:int,txt:str,err:str):
print("|-|",txt)
def recv_slot(self,cmd:int,data:bytearray,err:str):
# print("recv:",cmd,data)
if(self.cmd!=cmd):
return
try:
data=self.handle_.get_data()
if(len(data)>0):
self.port.send(cmd,data)
rate=self.handle_.get_rate()
self.rate_signal.emit(rate)
else:
del self.handle_
except Exception as e:
print(str(e))
# 开始升级mcu
def updata_mcu(self,file):
pass
# 小板通信测试
def comm_test(self):
pass
def end_slot(self,ip,ack,err):
# print(ip,ack,err)
if(ack==False):
self.show_msg(ip+":"+err)
def rate_slot(self,rate):
# print("rate signal:",ip,rate)
self.rate_signal.emit(rate)
def data_slot(self,ip,text):
pass
# 创建数据显示
def creat_databoxs(self,father:QDialog,data_list:list):
self.datab_list=[]
for i in range(len(data_list)):
p=data_box()
p.init(father,data_list[i][0],data_list[i][1],+i*50)
self.datab_list.append(p)
class locker():
def _get_lock(self):
file_name = os.path.basename(__file__)
# linux等平台依然使用标准的/var/run，其他nt等平台使用当前目录
if os.name == "posix":
lock_file_name = f"/var/run/{file_name}.pid"
else:
lock_file_name = f"{file_name}.pid"
self.fd = open(lock_file_name, "w")
try:
portalocker.lock(self.fd, portalocker.LOCK_EX | portalocker.LOCK_NB)
# 将当前进程号写入文件
# 如果获取不到锁上一步就已经异常了，所以不用担心覆盖
self.fd.writelines(str(os.getpid()))
# 写入的数据太少，默认会先被放在缓冲区，我们强制同步写入到文件
self.fd.flush()
except:
print(f"{file_name} have another instance running.")
exit(1)
def __init__(self):
self._get_lock()
# 和fcntl有点区别，portalocker释放锁直接有unlock()方法
# 还是一样，其实并不需要在最后自己主动释放锁
def __del__(self):
portalocker.unlock(self.fd)
if __name__ == "__main__":
lock=locker()
dlg=updata_dlg()
dlg.scan_file()
dlg.run()