diff --git a/ReadMe.txt b/ReadMe.txt index c3d3f43..35892d7 100644 --- a/ReadMe.txt +++ b/ReadMe.txt @@ -96,3 +96,5 @@ updata.py 后台下载服务器文件 2023.10.29 updata 文件列表添加右键菜单,查看文件详情 +2023.10.29 + 使用选项卡来选择两种升级方式 diff --git a/updata/updata.py b/updata/updata.py index ca427cc..ca8063d 100644 --- a/updata/updata.py +++ b/updata/updata.py @@ -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() diff --git a/updata/updata_ssh.py b/updata/updata_ssh.py new file mode 100644 index 0000000..be87ccd --- /dev/null +++ b/updata/updata_ssh.py @@ -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() diff --git a/updata/updata_uart.py b/updata/updata_uart.py index 78241ac..fdd561b 100644 --- a/updata/updata_uart.py +++ b/updata/updata_uart.py @@ -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()