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

添加U盘升级功能

Browse Source
This commit is contained in:
ranchuan
2023-09-22 18:26:14 +08:00
parent 4292478f67
commit d8e41f133b
9 changed files with 304 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
ReadMe.txt
View File

@@ -53,5 +53,11 @@
2023.9.20
串口控制台根据文件后缀自动识别文件类型,其中配置文件为cfg.json,
方案文件后缀为scheme.json
2023.9.21
添加创建小板下载镜像的脚本 creat_slave_boot.py
2023.9.22
实现U盘升级脚本，updata添加升级服务文件功能
串口控制台添加升级服务文件功能
添加同步设备时间的按钮

103
daemon/daemon.py Normal file
View File

@@ -0,0 +1,103 @@
#!/usr/bin/python3
import os
import sys
import shutil
import threading
import re
import time
# 定义守护进程
# 日志文件路径
log_filepath = '/home/root/log/daemon_log.txt'
def write_info(text:str):
fm = '%Y-%m-%d %X'
nowtime = time.strftime(fm, time.localtime())
with open(log_filepath, 'a') as fp:
fp.write(nowtime+ " | "+text+"\n")
print(text)
def _do_cmd(cmd:str):
write_info(cmd)
ret = os.popen(cmd).readlines()
for i in range(len(ret)):
ret[i]=ret[i].strip()
write_info(ret[i])
return ret
class auto_updata(object):
def __init__(self):
self.sd_path="/run/media/sda"
self.file_path=self.sd_path+'/updata'
self.sd_inserd_state = False
self.file_list=[]
def sd_check(self):
ack=os.path.exists(self.sd_path)
if(ack!=self.sd_inserd_state):
if(ack==True):
write_info("sd inserd.")
self.copy_file()
else:
write_info("sd extracted.")
self.sd_inserd_state=ack
def copy_file(self):
self.file_list.clear()
try:
self.file_list=os.listdir(self.file_path)
except Exception as e:
write_info(str(e))
return False
for i in self.file_list:
write_info("|---| "+i)
_do_cmd("systemctl stop atk-qtapp-start.service")
_do_cmd("mkdir /home/root/config")
_do_cmd("cp "+self.find_file_by_type(".elf")+" /usr/local/QDesktop-fb")
_do_cmd("chmod 777 /usr/local/QDesktop-fb")
_do_cmd("cp "+self.find_file_by_type(".bin")+" /home/root/config/checker_slave.bin")
_do_cmd("cp "+self.find_file_by_type("scheme.json")+" /home/root/config/checker_ye_cfg.json")
_do_cmd("cp "+self.file_path+"/cfg.json /home/root/config/cfg.json")
_do_cmd("cp "+self.find_file_by_type(".axf")+" /lib/firmware/checker_m4.axf")
# _do_cmd("cp "+self.find_file_by_type(".dtb")+" /boot/stm32mp157d-atk.dtb")
_do_cmd("sync")
_do_cmd("systemctl restart atk-qtapp-start.service")
write_info("autoupdata end.")
return True
def find_file_by_type(self,type:str):
for i in self.file_list:
if(i[-len(type):]==type):
return self.file_path+'/'+i
return "unknown"
if __name__ == '__main__':
dir_name=os.path.dirname(log_filepath)
if not os.path.exists(dir_name):
os.makedirs(dir_name)
updata=auto_updata()
while True:
updata.sd_check()
time.sleep(5)
# if __name__ == "__main__":
# scan_files("/run/media")

14
daemon/pydeamon.service Normal file
View File

@@ -0,0 +1,14 @@
[Unit]
Description=pydaemon daemon
After=rc-local.service
[Service]
Type=simple
User=root
Group=root
WorkingDirectory=/home/root
ExecStart=/usr/bin/python3 daemon.py
Restart=always
[Install]
WantedBy=multi-user.target

20
updata/console_uart.py
View File

@@ -160,7 +160,7 @@ class console_dlg(QObject):
self._item_append_green_str("已识别到U盘.")
if(index==5):
# 打开目录失败
self._item_append_red_str("打开目录失败,其插入U盘并且在根目录建立 'updata' 文件夹.")
self._item_append_red_str("打开目录失败,请插入U盘并且在根目录建立 'updata' 文件夹.")
self.cmd_list.clear()
if(index==6):
# 找不到相应文件
@@ -203,7 +203,7 @@ class console_dlg(QObject):
self.console_text.moveCursor(QTextCursor.MoveOperation.End)
except Exception as e:
print(str(e))
def file_file_by_type(self,type:str):
def find_file_by_type(self,type:str):
for i in self.file_list:
if(i[-len(type):]==type):
return i
@@ -220,16 +220,22 @@ class console_dlg(QObject):
# 组装命令列表
def _pack_cmd_list(self):
self.cmd_list.append("systemctl stop atk-qtapp-start.service")
self.cmd_list.append("systemctl stop pyeamon.service")
self.cmd_list.append("systemctl disable pyeamon.service")
self.cmd_list.append("mkdir /home/root/config")
self.cmd_list.append("cp "+self.file_file_by_type(".elf")+" /usr/local/QDesktop-fb")
self.cmd_list.append("cp "+self.find_file_by_type(".elf")+" /usr/local/QDesktop-fb")
self.cmd_list.append("chmod 777 /usr/local/QDesktop-fb")
self.cmd_list.append("cp "+self.file_file_by_type(".bin")+" /home/root/config/checker_slave.bin")
self.cmd_list.append("cp "+self.file_file_by_type("scheme.json")+" /home/root/config/checker_ye_cfg.json")
self.cmd_list.append("cp "+self.find_file_by_type(".bin")+" /home/root/config/checker_slave.bin")
self.cmd_list.append("cp "+self.find_file_by_type("scheme.json")+" /home/root/config/checker_ye_cfg.json")
self.cmd_list.append("cp cfg.json /home/root/config/cfg.json")
self.cmd_list.append("cp "+self.file_file_by_type(".axf")+" /lib/firmware/checker_m4.axf")
self.cmd_list.append("cp "+self.file_file_by_type(".dtb")+" /boot/stm32mp157d-atk.dtb")
self.cmd_list.append("cp "+self.find_file_by_type(".axf")+" /lib/firmware/checker_m4.axf")
self.cmd_list.append("cp "+self.find_file_by_type(".dtb")+" /boot/stm32mp157d-atk.dtb")
self.cmd_list.append("cp "+self.find_file_by_type(".py")+ " /home/root/daemon.py")
self.cmd_list.append("cp "+self.find_file_by_type(".service")+ " /lib/systemd/system/pydeamon.service")
self.cmd_list.append("sync")
self.cmd_list.append("systemctl restart atk-qtapp-start.service")
self.cmd_list.append("systemctl enable pyeamon.service")
self.cmd_list.append("systemctl restart pyeamon.service")
def send_cmdlist(self):
if(len(self.cmd_list)>0):

52
updata/creat_slave_boot.py Normal file
View File

@@ -0,0 +1,52 @@
import shutil
import sys
import os
import json
import prottcp
BOOT_PATH ="file/checker_slave_boot_can.bin"
APP_PATH ="file/checker_slave_app_can.bin"
SCHEME_PATH = "file/YM硬件EX工厂注码091902--9.json"
OUT_PATH = "file/checker_slave_boot_can_.bin"
# 创建离线下载器的镜像
# 填充指定个数的byte
def arr_byte_copy(byte:int,num:int):
t=bytearray()
for i in range(num):
t.append(byte)
return t
# int转数组
def arr_from_int(num:int):
return bytearray([num&0xff,(num>>8)&0xff,(num>>16)&0xff,(num>>24)&0xff])
def creat():
boot=BOOT_PATH
app=APP_PATH
scheme=SCHEME_PATH
d=bytearray()
with open(boot,"rb") as f:
d+=f.read()
d+=arr_byte_copy(0xff,0x4000-len(d))
with open(app,"rb") as f:
d+=f.read()
d+=arr_byte_copy(0xff,0x3f000-len(d))
with open(scheme,"rb") as f:
json_obj=json.loads(f.read())
d+=prottcp.scheme_to_byte(json_obj)
d+=arr_byte_copy(0xff,0x3f800-len(d))
d+=arr_from_int(0x669955aa)
with open(OUT_PATH,"wb+") as f:
f.write(d)
print(OUT_PATH+" create boot file success.")
if __name__ == "__main__":
creat()

4
updata/ftp.py
View File

@@ -130,9 +130,9 @@ class myftp(QObject):
src_path=src_list[self.index]
if os.path.exists(src_path):
self.send_cmd("rm "+i)
self.send_cmd("rm "+i,sleep_s=2)
# self.sftp.chmod(i.replace(file_name[-1],""),777)
self.send_cmd("chmod "+"777 "+i.replace(file_name[-1],""))
self.send_cmd("chmod "+"777 "+i.replace(file_name[-1],""),sleep_s=2)
try:
self.sftp.put(localpath=src_path, remotepath=i,callback=self.put_cb)
except Exception as r:

1
updata/mysql.py
View File

@@ -173,6 +173,7 @@ if __name__ == "__main__":
# file=s.download(1)
# print("save file:",file)
s.show_tables()
print(get_time())

24
updata/select_list.py
View File

@@ -10,8 +10,31 @@ from PyQt5.QtWidgets import *
# font: 25 9pt "Microsoft YaHei";
class select_list(QObject):
style_sheet ="""
QListView {
font: 25 9pt;
border: 15px solid white;
border-radius: 10px;
show-decoration-selected: 1;
}
QListView::item {
height: 40px;
}
QListView::item:hover {
background-color: transparent;
padding: 10px;
border-left: 3px solid rgb(130, 130, 130);
}
QListView::item:selected {
background-color: transparent;
color: black;
padding: 10px;
border-left: 3px solid black;
}
"""
def __init__(self,father:QDialog,title:str,str_list:list):
QObject.__init__(self)
self.w=QDialog(father)
@@ -28,6 +51,7 @@ class select_list(QObject):
self.file_list.setVerticalScrollBarPolicy(Qt.ScrollBarPolicy.ScrollBarAsNeeded)
self.file_list.setSelectionMode(QAbstractItemView.SelectionMode.SingleSelection)
self.file_list.itemDoubleClicked.connect(self.item_clicked)
self.file_list.setStyleSheet(self.style_sheet)
self.item_append(str_list)
def item_append(self,items:list):

94
updata/updata.py
View File

@@ -33,6 +33,10 @@ WARN_TXT="警告：\n程序升级会删除主板中以前的程序，为保证
QUESTION_TXT="程序升级会删除主板中以前的程序，非专业人员不要操作。\n\
为保证升级成功，请耐心等待升级完成，不要退出升级界面，期间不要给主板断电、重启。\n\
点击[YES]以进行升级。"
QUESTION_SSH_TXT="升级文件中包含系统服务。\n\
升级系统服务会导致远程登陆失效，即本程序将无法与设备建立通信。\n\
请等待升级完成之后重新启动设备，以保证系统服务正常运行。\n\
点击[YES]以进行升级。"
def warn_creat(father:QDialog,text:str,rect:QRect):
lable=QLabel(father)
@@ -111,6 +115,8 @@ class updata_dlg(QObject):
self.widget.setWindowFlags(Qt.WindowType.WindowStaysOnTopHint)
self.addrs=""
self.dhcp_server=None
# 如果要升级服务，则在升级之后ssh会连接不上
self.systemd_stop=False
self.but_y=60
self.but_y_step=40
self.file_list_init()
@@ -118,7 +124,8 @@ class updata_dlg(QObject):
self.save_but_init()
self.dhcp_but_init()
self.console_but_init()
self.console_but_init()
self.settime_but_init()
# self.restart_but_init()
self.cmd_but_init()
self.refresh_but_init()
self.sstate_but_init()
@@ -218,6 +225,24 @@ class updata_dlg(QObject):
self.console_but.clicked.connect(self.console_but_clicked)
self.console_but.setToolTip("通过设备的串口控制台升级程序，这种方式需要使用到U盘。")
# 初始化同步时间按钮
def settime_but_init(self):
self.settime_but = QPushButton(self.widget)
self.settime_but.setObjectName(u"settime_but")
self.settime_but.setGeometry(QRect(700, 140, 93, 28))
self.settime_but.setText("同步时间")
self.settime_but.clicked.connect(self.settime_but_clicked)
self.settime_but.setToolTip("同步主板时间。")
# 初始化重新启动按钮
def restart_but_init(self):
self.restart_but = QPushButton(self.widget)
self.restart_but.setObjectName(u"restart_but")
self.restart_but.setGeometry(QRect(700, 180, 93, 28))
self.restart_but.setText("重启软件")
self.restart_but.clicked.connect(self.restart_but_clicked)
self.restart_but.setToolTip("如果已知设备ip地址，但刷新不出，可尝试重启设备软件。")
# 初始化发送命令按钮
def cmd_but_init(self):
self.cmd_but = QPushButton(self.widget)
@@ -396,6 +421,7 @@ class updata_dlg(QObject):
# 根据文件列表生成目标列表
def build_dst_list(self,file_list):
self.systemd_stop=False
dst_list=[]
for i in file_list:
if(i[-4:]==".elf"):
@@ -411,6 +437,11 @@ class updata_dlg(QObject):
dst_list.append("/home/root/config/"+"judge.lua")
elif(i[-4:]==".axf"):
dst_list.append("/lib/firmware/"+"checker_m4.axf")
elif(i[-3:]==".py"):
dst_list.append("/home/root/"+"daemon.py")
elif(i[-8:]==".service"):
dst_list.append("/lib/systemd/system/pydeamon.service")
self.systemd_stop=True
else:
dst_list.append("/home/root/config/"+i)
return dst_list
@@ -474,6 +505,9 @@ class updata_dlg(QObject):
return
if(self.show_question("警告",QUESTION_TXT)==False):
return
if(self.systemd_stop==True):
if(self.show_question("警告",QUESTION_SSH_TXT)==False):
return
print("slaves:",slave_list)
w=QDialog(self.widget)
warn_creat(w,WARN_TXT,QRect(0,0,700-100,90))
@@ -519,7 +553,7 @@ class updata_dlg(QObject):
print("slaves:",slave_list)
w=QDialog(self.widget)
w.resize(700-100, len(slave_list)*40+20)
w.setWindowTitle("升级mcu")
w.setWindowTitle("升级方案")
self.scheme_mcu(slave_list)
self.creat_progress(w,0,slave_list)
w.show()
@@ -537,10 +571,29 @@ class updata_dlg(QObject):
self.data_list=[]
self.comm_test(slave_list)
# 时间同步按下
def settime_but_clicked(self):
print("settime_but clicked.")
slave_list=self.get_selected_slave()
if(len(slave_list)==0):
return
self.data_list=[]
self.settime_host(slave_list)
# 重新启动按下
def restart_but_clicked(self):
print("restart_but clicked.")
slave_list=self.get_selected_slave()
if(len(slave_list)==0):
return
self.data_list=[]
self.restart_host(slave_list)
def addfile_but_clicked(self):
print("addfile_but clicked")
fileName,fileType = QFileDialog.getOpenFileNames(None, "选取文件", os.getcwd(),
"主板程序(*.elf);;主板m4程序(*.axf);;小板程序(*.bin);;检测方案(*.json);;判定脚本(*.lua);;任意文件(*)")
"""主板程序(*.elf);;主板m4程序(*.axf);;小板程序(*.bin);;检测方案(*.json);;启动脚本(*.sh);;
判定脚本(*.lua);;守护进程脚本(*.py);;守护进程服务(*.service);;设备树文件(*.dtb);;任意文件(*)""")
print(fileName,fileType)
path=self.getpath()+"file\\"
for i in fileName:
@@ -593,7 +646,7 @@ class updata_dlg(QObject):
# 扫描文件
def scan_file(self):
self.file_list.clear()
self.file_list.addItems(self.find_type([".sh",".elf",".bin",".lua",".json",".dtb",".axf"]))
self.file_list.addItems(self.find_type([".sh",".elf",".bin",".lua",".json",".dtb",".axf",".py",".service"]))
# 扫描从机
def scan_slave(self):
@@ -679,6 +732,30 @@ class updata_dlg(QObject):
t = threading.Thread(target=u.bordcast, args=(cmd_list,))
t.start()
# 同步时间
def settime_host(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="local settime "+mysql.get_time()+' '
cmd_list=[]
cmd_list.append((updata_cmd,1,9))
t = threading.Thread(target=u.bordcast, args=(cmd_list,))
t.start()
# 重新启动
def restart_host(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="systemctl restart atk-qtapp-start.service "
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)
@@ -749,11 +826,18 @@ class updata_dlg(QObject):
f.rate_signal.connect(self.rate_slot)
if(restart==True):
print(ip,"|stop app.")
f.send_cmd("systemctl stop atk-qtapp-start.service")
f.send_cmd("systemctl stop atk-qtapp-start.service",sleep_s=2)
f.send_cmd("systemctl stop pydeamon.service",sleep_s=2)
if(self.systemd_stop==True):
f.send_cmd("systemctl disable pydeamon.service",sleep_s=2)
f.put_file(ip,dst_list,src_list)
if(restart==True):
print(ip,"|start app.")
f.send_cmd("systemctl restart atk-qtapp-start.service")
# 控制enable disable会导致ssh无法连接
if(self.systemd_stop==True):
f.send_cmd("systemctl enable pydeamon.service",sleep_s=2)
f.send_cmd("systemctl start pydeamon.service",sleep_s=2)
f.send_cmd("sync",sleep_s=3)
f.close()
self.end_signal.emit(ip,True,"完成")