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

使用选项卡来选择两种升级方式

Browse Source
This commit is contained in:
andy
2023-10-29 11:32:46 +08:00
parent 2660834312
commit 91e1b64839
4 changed files with 1059 additions and 941 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
ReadMe.txt
View File

@@ -96,3 +96,5 @@
updata.py 后台下载服务器文件
2023.10.29
updata 文件列表添加右键菜单,查看文件详情
2023.10.29
使用选项卡来选择两种升级方式

934
updata/updata.py
View File

@@ -1,937 +1,49 @@
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 sys
import base64
import updata_ssh
import updata_uart
import memory_pic
import ctypes
import select_list
import dhcp.dhcp as dhcp
import mysql
import console_uart
import file_detail
import float_lable
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]以进行升级。"
QUESTION_SSH_TXT="升级文件中包含系统服务。\n\
升级系统服务会导致远程登陆失效，即本程序将无法与设备建立通信。\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])
# 数据库下载文件结束信号
sql_download_end_signal =pyqtSignal([])
class updata_dlg(QWidget):
def __init__(self):
QObject.__init__(self)
self.app = QApplication(sys.argv)
self.widget = QWidget()
self.widget.resize(870, 410)
self.widget.setWindowTitle("批检仪程序升级")
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()
self.slave_list_init()
self.save_but_init()
self.dhcp_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()
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.scan_file()
self.scan_slave()
self.infotext_init()
self.widget.destroyed.connect(self.quit)
self.sql_download_end_signal.connect(self.scan_file)
QWidget.__init__(self)
self.resize(870, 430)
self.tab_widget=QTabWidget(self)
self.setWindowTitle("设备软件升级维护工具")
self.tab_widget.setGeometry(QRect(0,0,870,430))
widget_ssh=updata_ssh.updata_dlg()
widget_uart=updata_uart.updata_dlg()
self.tab_widget.setTabPosition(QTabWidget.TabPosition.North)
self.tab_widget.addTab(widget_ssh,"批检仪/赋码仪主板(SSH)")
self.tab_widget.addTab(widget_uart,"MCU串口升级")
self.tab_widget.setTabShape(QTabWidget.TabShape.Triangular)
self.setWindowFlags(self.windowFlags() &(~ Qt.WindowType.WindowMaximizeButtonHint))
self.setFixedSize(self.width(), self.height())
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("请选择要升级的文件，只有选中的文件会发送到主板中。")
ack=self.file_list.setProperty("contextMenuPolicy",Qt.ContextMenuPolicy.CustomContextMenu)
self.file_list.customContextMenuRequested.connect(self.file_item_menu_slot)
def file_item_menu_slot(self,pos:QPoint):
print("show pop menu")
menu=QMenu(self.file_list)
action=QAction("删除选中条目",self.file_list)
action.triggered.connect(self.file_item_delete)
menu.addAction(action)
action=QAction("查看选中条目详情",self.file_list)
action.triggered.connect(self.file_item_detail)
menu.addAction(action)
action=QAction("发送选中条目到主板",self.file_list)
action.triggered.connect(self.save_but_clicked)
menu.addAction(action)
menu.addAction(action)
action=QAction("刷新文件列表",self.file_list)
action.triggered.connect(self.scan_file)
menu.addAction(action)
menu.exec(self.file_list.mapToGlobal(pos))
def file_item_delete(self):
print("delete slected items.")
items=self.file_list.selectedItems()
for i in items:
os.remove("file\\"+i.text())
self.scan_file()
def file_item_detail(self):
print("show detail with slected items.")
items=self.file_list.selectedItems()
details=[]
for i in items:
details+=file_detail.detail("file/"+i.text())
if(len(details)==0):
return
pos=QCursor.pos()
lab=float_lable.floatBox(self.widget,details,x=pos.x(),y=pos.y())
lab.show()
# 初始化从机列表
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, 150, 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, 150, 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)
# 初始化DHCP服务器按钮
def dhcp_but_init(self):
self.dhcp_but = QPushButton(self.widget)
self.dhcp_but.setObjectName(u"dhcp_but")
self.dhcp_but.setGeometry(QRect(750, 60, 93, 28))
self.dhcp_but.setText("打开DHCP")
self.dhcp_but.clicked.connect(self.dhcp_but_clicked)
self.dhcp_but.setToolTip("如果没有搜索到从机，则打开DHCP服务器。")
# 初始化控制台按钮
def console_but_init(self):
self.console_but = QPushButton(self.widget)
self.console_but.setObjectName(u"console_but")
self.console_but.setGeometry(QRect(750, 100, 93, 28))
self.console_but.setText("串口控制台")
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(750, 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, 150, 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)
self.cmd_but.setObjectName(u"save_but")
self.cmd_but.setGeometry(QRect(590, self.but_y, 150, 28))
self.but_y+=self.but_y_step
self.cmd_but.setText("升级小板程序")
self.cmd_but.clicked.connect(self.cmd_but_clicked)
self.cmd_but.setToolTip("请先将要升级的小板程序发送到主板中，然后点击此按钮开始升级小板程序。")
# 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, 150, 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, 150, 28))
self.but_y+=self.but_y_step
self.sstate_but.setText("小板在线状态")
self.sstate_but.clicked.connect(self.sstate_but_clicked)
self.sstate_but.setToolTip("点击此按钮查看小板在线情况，显示在线小板的地址。")
self.sstate_but.setEnabled(False)
# 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, 150, 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, 150, 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, 150, 28))
self.but_y+=self.but_y_step
self.download_but.setText("从服务器下载文件")
self.download_but.clicked.connect(self.download_but_clicked)
# 提示信息
def infotext_init(self):
self.infotext=QLabel(self.widget)
self.infotext.setGeometry(QRect(25, 380, 800, 15))
self.infotext.setText("欢迎使用云铭公司产线设备维护工具.")
def set_infotext(self,text:str):
try:
self.infotext.setText(text)
except Exception as e:
pass
# 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,types:list):
file_list=[]
items=self.file_list.selectedItems()
for i in items:
sp=i.text().split(".")[-1]
sp='.'+sp
if(sp in types):
file_list.append(i.text())
if(len(file_list)!=1):
self.set_infotext("请选择一个并且只选择一个 "+str(types)+" 文件")
return ""
return file_list[0]
# 获取已选择的文件列表
def get_selected_fils(self):
file_list=[]
items=self.file_list.selectedItems()
if(len(items)==0):
self.set_infotext("请选择至少一个文件")
return []
have_elf=False
have_bin=False
have_lua=False
for i in items:
if(i.text()[-4:]==".elf"):
if(have_elf==True):
self.set_infotext("只可选择一个 .elf 文件")
return []
have_elf=True
if(i.text()[-4:]==".bin"):
if(have_bin==True):
self.set_infotext("只可选择一个 .bin 文件")
return []
have_bin=True
if(i.text()[-4:]==".lua"):
if(have_lua==True):
self.set_infotext("只可选择一个 .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.set_infotext("请选择至少一个主板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):
self.systemd_stop=False
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"):
if(i!="cfg.json"):
dst_list.append("/home/root/config/"+"checker_ye_cfg.json")
else:
dst_list.append("/home/root/config/"+"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")
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
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 console_but_clicked(self):
print("console_but_clicked.")
dlg=console_uart.console_dlg(self.widget,"串口控制台")
dlg.show()
print("Console end.")
def dhcp_but_clicked(self):
print("dhcp_but clicked.")
if(self.dhcp_server==None):
self.dhcp_but.setEnabled(False)
self.dhcp_server=dhcp.creat_dhcp_server()
self.dhcp_server.new_ip_addr_signal.connect(self.add_slave_slot)
self.dhcp_server.server_start_signal.connect(self.dhcp_start_slot)
self.dhcp_server.server_end_signal.connect(self.dhcp_end_slot)
self.dhcp_server.run_in_thread()
else:
self.dhcp_but.setEnabled(False)
self.dhcp_server.new_ip_addr_signal.disconnect(dlg.add_slave_slot)
self.dhcp_server.close()
def dhcp_start_slot(self):
self.dhcp_but.setText("关闭DHCP")
self.dhcp_but.setEnabled(True)
def dhcp_end_slot(self):
self.dhcp_but.setText("打开DHCP")
self.dhcp_but.setEnabled(True)
self.dhcp_server.server_start_signal.disconnect(self.dhcp_start_slot)
self.dhcp_server.server_end_signal.disconnect(self.dhcp_end_slot)
self.dhcp_server=None
# 发送文件按钮按下
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
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))
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",".pkt"])
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("升级方案")
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 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);;启动脚本(*.sh);;
判定脚本(*.lua);;守护进程脚本(*.py);;守护进程服务(*.service);;设备树文件(*.dtb);;任意文件(*)""")
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("|")
self.set_infotext("正在从服务器下载文件,可能需要一些时间,请耐心等待.")
t = threading.Thread(target=self.sql_download, args=(sql,int(item[0]),))
t.start()
# 从数据库下载文件,在后台运行
def sql_download(self,sql:mysql.sql,index:int):
file_name=sql.download(index)
dst="file/"+file_name.split("/")[-1]
shutil.copy(file_name,dst)
self.sql_download_end_signal.emit()
self.set_infotext("已下载文件:"+dst)
# 开始运行
def run(self):
self.widget.show()
a=self.app.exec()
print("run end.")
if(self.dhcp_server!=None):
self.dhcp_server.close()
# 扫描文件
def scan_file(self):
self.file_list.clear()
self.file_list.addItems(self.find_type([".sh",".elf",".bin",".pkt",".lua",".json",".dtb",".axf",".py",".service"]))
# 扫描从机
def scan_slave(self):
self.set_infotext("正在刷新主板IP地址")
u=udp.myudp(1,255)
ip_prefix=self.ip_prefix.text()
u.find(ip_prefix)
list_str=[]
for i in u.dst_ip_list:
list_str.append(i[0]+','+i[1])
if(len(list_str)==0):
self.set_infotext("未找到主板IP地址,请确保主板已正常连接并运行")
self.slave_list.addItems(list_str)
# 添加单个从机地址
def add_slave_slot(self,mac:str,ip:str,name:str):
for i in range(self.slave_list.count()):
item=self.slave_list.item(i).text()
if(item.split(",")[0]==ip):
return
self.slave_list.addItem(ip+",dhcp_find")
# 扫描指定类型的文件
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 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)
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",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,"完成")
# 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)
self.setWindowIcon(icon)
if __name__ == "__main__":
lock=locker()
app = QApplication(sys.argv)
dlg=updata_dlg()
dlg.run()
dlg.show()
app.exec()

940
updata/updata_ssh.py Normal file
View File

@@ -0,0 +1,940 @@
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 select_list
import dhcp.dhcp as dhcp
import mysql
import console_uart
import file_detail
import float_lable
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]以进行升级。"
QUESTION_SSH_TXT="升级文件中包含系统服务。\n\
升级系统服务会导致远程登陆失效，即本程序将无法与设备建立通信。\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(QWidget):
# 进度信号，ip，1~100
rate_signal =pyqtSignal([str,int])
# 结束信号，ip，成败，描述
end_signal = pyqtSignal([str,bool,str])
# 数据库下载文件结束信号
sql_download_end_signal =pyqtSignal([])
def __init__(self):
QWidget.__init__(self)
self.widget = self
self.widget.resize(870, 410)
self.widget.setWindowTitle("批检仪程序升级")
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()
self.slave_list_init()
self.save_but_init()
self.dhcp_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()
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.scan_file()
self.scan_slave()
self.infotext_init()
self.widget.destroyed.connect(self.quit)
self.sql_download_end_signal.connect(self.scan_file)
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("请选择要升级的文件，只有选中的文件会发送到主板中。")
ack=self.file_list.setProperty("contextMenuPolicy",Qt.ContextMenuPolicy.CustomContextMenu)
self.file_list.customContextMenuRequested.connect(self.file_item_menu_slot)
def file_item_menu_slot(self,pos:QPoint):
print("show pop menu")
menu=QMenu(self.file_list)
action=QAction("删除选中条目",self.file_list)
action.triggered.connect(self.file_item_delete)
menu.addAction(action)
action=QAction("查看选中条目详情",self.file_list)
action.triggered.connect(self.file_item_detail)
menu.addAction(action)
action=QAction("发送选中条目到主板",self.file_list)
action.triggered.connect(self.save_but_clicked)
menu.addAction(action)
menu.addAction(action)
action=QAction("刷新文件列表",self.file_list)
action.triggered.connect(self.scan_file)
menu.addAction(action)
menu.exec(self.file_list.mapToGlobal(pos))
def file_item_delete(self):
print("delete slected items.")
items=self.file_list.selectedItems()
for i in items:
os.remove("file\\"+i.text())
self.scan_file()
def file_item_detail(self):
print("show detail with slected items.")
items=self.file_list.selectedItems()
details=[]
for i in items:
details+=file_detail.detail("file/"+i.text())
if(len(details)==0):
return
pos=QCursor.pos()
lab=float_lable.floatBox(self.widget,details,x=pos.x(),y=pos.y())
lab.show()
# 初始化从机列表
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, 150, 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, 150, 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)
# 初始化DHCP服务器按钮
def dhcp_but_init(self):
self.dhcp_but = QPushButton(self.widget)
self.dhcp_but.setObjectName(u"dhcp_but")
self.dhcp_but.setGeometry(QRect(750, 60, 93, 28))
self.dhcp_but.setText("打开DHCP")
self.dhcp_but.clicked.connect(self.dhcp_but_clicked)
self.dhcp_but.setToolTip("如果没有搜索到从机，则打开DHCP服务器。")
# 初始化控制台按钮
def console_but_init(self):
self.console_but = QPushButton(self.widget)
self.console_but.setObjectName(u"console_but")
self.console_but.setGeometry(QRect(750, 100, 93, 28))
self.console_but.setText("串口控制台")
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(750, 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, 150, 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)
self.cmd_but.setObjectName(u"save_but")
self.cmd_but.setGeometry(QRect(590, self.but_y, 150, 28))
self.but_y+=self.but_y_step
self.cmd_but.setText("升级小板程序")
self.cmd_but.clicked.connect(self.cmd_but_clicked)
self.cmd_but.setToolTip("请先将要升级的小板程序发送到主板中，然后点击此按钮开始升级小板程序。")
# 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, 150, 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, 150, 28))
self.but_y+=self.but_y_step
self.sstate_but.setText("小板在线状态")
self.sstate_but.clicked.connect(self.sstate_but_clicked)
self.sstate_but.setToolTip("点击此按钮查看小板在线情况，显示在线小板的地址。")
self.sstate_but.setEnabled(False)
# 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, 150, 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, 150, 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, 150, 28))
self.but_y+=self.but_y_step
self.download_but.setText("从服务器下载文件")
self.download_but.clicked.connect(self.download_but_clicked)
self.download_but.setToolTip("从服务器下载文件到本地。")
# 提示信息
def infotext_init(self):
self.infotext=QLabel(self.widget)
self.infotext.setGeometry(QRect(25, 380, 800, 15))
self.infotext.setText("欢迎使用云铭公司设备维护工具.")
def set_infotext(self,text:str):
try:
self.infotext.setText(text)
except Exception as e:
pass
# 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,types:list):
file_list=[]
items=self.file_list.selectedItems()
for i in items:
sp=i.text().split(".")[-1]
sp='.'+sp
if(sp in types):
file_list.append(i.text())
if(len(file_list)!=1):
self.set_infotext("请选择一个并且只选择一个 "+str(types)+" 文件")
return ""
return file_list[0]
# 获取已选择的文件列表
def get_selected_fils(self):
file_list=[]
items=self.file_list.selectedItems()
if(len(items)==0):
self.set_infotext("请选择至少一个文件")
return []
have_elf=False
have_bin=False
have_lua=False
for i in items:
if(i.text()[-4:]==".elf"):
if(have_elf==True):
self.set_infotext("只可选择一个 .elf 文件")
return []
have_elf=True
if(i.text()[-4:]==".bin"):
if(have_bin==True):
self.set_infotext("只可选择一个 .bin 文件")
return []
have_bin=True
if(i.text()[-4:]==".lua"):
if(have_lua==True):
self.set_infotext("只可选择一个 .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.set_infotext("请选择至少一个主板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):
self.systemd_stop=False
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"):
if(i!="cfg.json"):
dst_list.append("/home/root/config/"+"checker_ye_cfg.json")
else:
dst_list.append("/home/root/config/"+"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")
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
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 console_but_clicked(self):
print("console_but_clicked.")
dlg=console_uart.console_dlg(self.widget,"串口控制台")
dlg.show()
print("Console end.")
def dhcp_but_clicked(self):
print("dhcp_but clicked.")
if(self.dhcp_server==None):
self.dhcp_but.setEnabled(False)
self.dhcp_server=dhcp.creat_dhcp_server()
self.dhcp_server.new_ip_addr_signal.connect(self.add_slave_slot)
self.dhcp_server.server_start_signal.connect(self.dhcp_start_slot)
self.dhcp_server.server_end_signal.connect(self.dhcp_end_slot)
self.dhcp_server.run_in_thread()
else:
self.dhcp_but.setEnabled(False)
self.dhcp_server.new_ip_addr_signal.disconnect(dlg.add_slave_slot)
self.dhcp_server.close()
def dhcp_start_slot(self):
self.dhcp_but.setText("关闭DHCP")
self.dhcp_but.setEnabled(True)
def dhcp_end_slot(self):
self.dhcp_but.setText("打开DHCP")
self.dhcp_but.setEnabled(True)
self.dhcp_server.server_start_signal.disconnect(self.dhcp_start_slot)
self.dhcp_server.server_end_signal.disconnect(self.dhcp_end_slot)
self.dhcp_server=None
# 发送文件按钮按下
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
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))
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",".pkt"])
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("升级方案")
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 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);;启动脚本(*.sh);;
判定脚本(*.lua);;守护进程脚本(*.py);;守护进程服务(*.service);;设备树文件(*.dtb);;任意文件(*)""")
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("|")
self.set_infotext("正在从服务器下载文件,可能需要一些时间,请耐心等待.")
t = threading.Thread(target=self.sql_download, args=(sql,int(item[0]),))
t.start()
# 从数据库下载文件,在后台运行
def sql_download(self,sql:mysql.sql,index:int):
file_name=sql.download(index)
dst="file/"+file_name.split("/")[-1]
shutil.copy(file_name,dst)
self.sql_download_end_signal.emit()
self.set_infotext("已下载文件:"+dst)
# 开始运行
def run(self):
self.widget.show()
def end(self):
print("run end.")
if(self.dhcp_server!=None):
self.dhcp_server.close()
# 扫描文件
def scan_file(self):
self.file_list.clear()
self.file_list.addItems(self.find_type([".sh",".elf",".bin",".pkt",".lua",".json",".dtb",".axf",".py",".service"]))
# 扫描从机
def scan_slave(self):
self.set_infotext("正在刷新主板IP地址")
u=udp.myudp(1,255)
ip_prefix=self.ip_prefix.text()
u.find(ip_prefix)
list_str=[]
for i in u.dst_ip_list:
list_str.append(i[0]+','+i[1])
if(len(list_str)==0):
self.set_infotext("未找到主板IP地址,请确保主板已正常连接并运行")
self.slave_list.addItems(list_str)
# 添加单个从机地址
def add_slave_slot(self,mac:str,ip:str,name:str):
for i in range(self.slave_list.count()):
item=self.slave_list.item(i).text()
if(item.split(",")[0]==ip):
return
self.slave_list.addItem(ip+",dhcp_find")
# 扫描指定类型的文件
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 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)
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",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,"完成")
# 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()
app = QApplication(sys.argv)
dlg=updata_dlg()
dlg.run()
app.exec()
dlg.end()

124
updata/updata_uart.py
View File

@@ -14,6 +14,8 @@ import serial.tools.list_ports
import prottcp
import mysql
import select_list
import file_detail
import float_lable
STR_RED="\033[1;31m"
@@ -65,18 +67,19 @@ class data_box(QObject):
class updata_dlg(QObject):
class updata_dlg(QWidget):
# 进度信号，ip，1~100
rate_signal =pyqtSignal([int])
# 结束信号，ip，成败，描述
end_signal = pyqtSignal([bool,str])
failed_signal = pyqtSignal()
# 数据库下载文件结束信号
sql_download_end_signal =pyqtSignal([])
def __init__(self):
QObject.__init__(self)
self.app = QApplication(sys.argv)
self.widget = QWidget()
QWidget.__init__(self)
self.widget = self
self.widget.setAttribute(Qt.WidgetAttribute.WA_DeleteOnClose)
self.widget.resize(703, 409)
self.widget.setWindowTitle("赋码仪程序升级")
@@ -94,7 +97,20 @@ class updata_dlg(QObject):
self.failed_signal.connect(self.updata_failed)
self.cmd=0
self.port_is_open=False
self.scan_file()
self.infotext_init()
# 提示信息
def infotext_init(self):
self.infotext=QLabel(self.widget)
self.infotext.setGeometry(QRect(25, 380, 800, 15))
self.infotext.setText("欢迎使用云铭公司设备维护工具.")
def set_infotext(self,text:str):
try:
self.infotext.setText(text)
except Exception as e:
pass
def quit(self):
self.close_port()
# 程序退出
@@ -109,6 +125,42 @@ class updata_dlg(QObject):
self.file_list.setVerticalScrollBarPolicy(Qt.ScrollBarPolicy.ScrollBarAsNeeded)
self.file_list.setSelectionMode(QAbstractItemView.SelectionMode.SingleSelection)
self.file_list.itemDoubleClicked.connect(self.item_clicked)
ack=self.file_list.setProperty("contextMenuPolicy",Qt.ContextMenuPolicy.CustomContextMenu)
self.file_list.customContextMenuRequested.connect(self.file_item_menu_slot)
def file_item_menu_slot(self,pos:QPoint):
print("show pop menu")
menu=QMenu(self.file_list)
action=QAction("删除选中条目",self.file_list)
action.triggered.connect(self.file_item_delete)
menu.addAction(action)
action=QAction("查看选中条目详情",self.file_list)
action.triggered.connect(self.file_item_detail)
menu.addAction(action)
action=QAction("发送选中条目到主板",self.file_list)
action.triggered.connect(self.save_but_clicked)
menu.addAction(action)
menu.addAction(action)
action=QAction("刷新文件列表",self.file_list)
action.triggered.connect(self.scan_file)
menu.addAction(action)
menu.exec(self.file_list.mapToGlobal(pos))
def file_item_delete(self):
print("delete slected items.")
items=self.file_list.selectedItems()
for i in items:
os.remove("file\\"+i.text())
self.scan_file()
def file_item_detail(self):
print("show detail with slected items.")
items=self.file_list.selectedItems()
details=[]
for i in items:
details+=file_detail.detail("file/"+i.text())
if(len(details)==0):
return
pos=QCursor.pos()
lab=float_lable.floatBox(self.widget,details,x=pos.x(),y=pos.y())
lab.show()
@@ -116,7 +168,7 @@ class updata_dlg(QObject):
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.setGeometry(QRect(590, 10, 150, 28))
self.com_but.setText("打开端口")
self.com_but.clicked.connect(self.com_but_clicked)
@@ -125,49 +177,54 @@ class updata_dlg(QObject):
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.setGeometry(QRect(590, 60, 150, 28))
self.save_but.setText("发送文件到串口设备")
self.save_but.clicked.connect(self.save_but_clicked)
self.save_but.setToolTip("请选择要升级的文件，将发送选择的文件到串口设备中。")
# 初始化发送命令按钮
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.setGeometry(QRect(590, 100, 150, 28))
self.cmd_but.setText("升级从设备")
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.setGeometry(QRect(590, 100, 150, 28))
self.updatas_but.setText("升级小板")
self.updatas_but.clicked.connect(self.updatas_but_clicked)
self.updatas_but.setToolTip("先将小板程序发送到串口设备中,再使用此按钮升级小板。")
# 初始化在线状态按钮
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.setGeometry(QRect(590, 140, 150, 28))
self.sstate_but.setText("串口设备参数")
self.sstate_but.clicked.connect(self.sstate_but_clicked)
self.sstate_but.setToolTip("显示串口设备的基本参数。")
# 初始化方案状态按钮
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.setGeometry(QRect(590, 180, 150, 28))
self.scheme_but.setText("方案参数")
self.scheme_but.clicked.connect(self.scheme_but_clicked)
self.scheme_but.setToolTip("显示串口设备的方案参数。")
# 初始化下载文件按钮
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.setGeometry(QRect(590, 220, 150, 28))
self.download_but.setText("从服务器下载文件")
self.download_but.clicked.connect(self.download_but_clicked)
self.download_but.setToolTip("从服务器下载文件到本地。")
# com口
@@ -218,7 +275,7 @@ class updata_dlg(QObject):
if(i.text()[-len(type):]==type):
file_list.append(i.text())
if(len(file_list)!=1):
self.show_msg("请选择一个并且只选择一个 "+type+" 文件")
self.set_infotext("请选择一个并且只选择一个 "+type+" 文件")
return ""
return file_list[0]
@@ -227,7 +284,7 @@ class updata_dlg(QObject):
file_list=[]
items=self.file_list.selectedItems()
if(len(items)==0):
self.show_msg("请选择至少一个文件")
self.set_infotext("请选择至少一个文件")
return []
have_elf=False
have_bin=False
@@ -235,17 +292,17 @@ class updata_dlg(QObject):
for i in items:
if(i.text()[-4:]==".elf"):
if(have_elf==True):
self.show_msg("只可选择一个 .elf 文件")
self.set_infotext("只可选择一个 .elf 文件")
return []
have_elf=True
if(i.text()[-4:]==".bin"):
if(have_bin==True):
self.show_msg("只可选择一个 .bin 文件")
self.set_infotext("只可选择一个 .bin 文件")
return []
have_bin=True
if(i.text()[-4:]==".lua"):
if(have_lua==True):
self.show_msg("只可选择一个 .lua 文件")
self.set_infotext("只可选择一个 .lua 文件")
return []
have_lua=True
file_list.append(i.text())
@@ -297,7 +354,7 @@ class updata_dlg(QObject):
self.port.send(self.cmd,d)
except Exception as e:
print("com not open")
self.show_msg("端口未打开")
self.set_infotext("端口未打开")
del self.handle_
w.close()
return
@@ -306,7 +363,7 @@ class updata_dlg(QObject):
# self.failed_signal.connect(w.close)
# w.destroyed.connect(self.close_port)
def updata_failed(self):
self.show_msg("打开端口失败")
self.set_infotext("打开端口失败")
# 创建进度条
def creat_progress(self,father:QDialog):
@@ -352,9 +409,9 @@ class updata_dlg(QObject):
print("updatas_but clicked.")
try:
self.port.send_str("updatas 1,2,3,4,5,6,7,8,9,10")
self.show_msg("已发送升级指令，请留意小板升级情况")
self.set_infotext("已发送升级指令，请留意小板升级情况")
except Exception as e:
self.show_msg("命令发送失败")
self.set_infotext("命令发送失败,是否未打开端口?")
print("com not open")
print(str(e))
@@ -387,15 +444,21 @@ class updata_dlg(QObject):
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()
self.set_infotext("正在从服务器下载文件,可能需要一些时间,请耐心等待.")
t = threading.Thread(target=self.sql_download, args=(sql,int(item[0]),))
t.start()
# 从数据库下载文件,在后台运行
def sql_download(self,sql:mysql.sql,index:int):
file_name=sql.download(index)
dst="file/"+file_name.split("/")[-1]
shutil.copy(file_name,dst)
self.sql_download_end_signal.emit()
self.set_infotext("已下载文件:"+dst)
# 开始运行
def run(self):
self.widget.show()
sys.exit(self.app.exec())
def set_port_state(self,state:bool):
self.port_is_open=state
@@ -549,8 +612,9 @@ class locker():
if __name__ == "__main__":
lock=locker()
app = QApplication(sys.argv)
dlg=updata_dlg()
dlg.scan_file()
dlg.run()
app.exec()