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(QObject): # 进度信号，ip，1~100 rate_signal =pyqtSignal([str,int]) # 结束信号，ip，成败，描述 end_signal = pyqtSignal([str,bool,str]) # 数据库下载文件结束信号 sql_download_end_signal =pyqtSignal([]) 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) 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) if __name__ == "__main__": lock=locker() dlg=updata_dlg() dlg.run()