checker_gen1/python/updata.py

import sys
import os
from PyQt5.QtCore import *
from PyQt5.QtGui import *
from PyQt5.QtWidgets import *
import threading
import time
import portalocker
import serial
import serial.tools.list_ports
import prottcp



STR_RED="\033[1;31m"
STR_BLUE="\033[1;34m"
STR_END="\033[0m"





class progress_box(QObject):


    def __init__(self) -> None:
        QObject.__init__(self)
    
    def init(self,father:QDialog,text:str,x:int):
        # print("text=",text)
        self.lable = QLabel(father)
        self.lable.setObjectName(u"label")
        self.lable.setGeometry(QRect(40, x, 120, 15))
        self.lable.setText(text)
        self.p=QProgressBar(father)
        self.p.setValue(0)
        self.p.setGeometry(QRect(40,x+20,450,20))
    def set_rate(self,rate:int):
            # print(text)
            self.p.setValue(rate)



class data_box(QObject):
    def __init__(self) -> None:
        QObject.__init__(self)
    
    def init(self,father:QDialog,title:str,text:str,x:int):
        # print("text=",text)
        self.lable = QLabel(father)
        self.lable.setObjectName(u"label")
        self.lable.setGeometry(QRect(40, x, 120, 20))
        self.lable.setText(title)
        self.text = QLabel(father)
        self.text.setObjectName(u"text")
        self.text.setGeometry(QRect(40,x+20,450,30))
        self.text.setText(text)







class updata_dlg(QObject):
    # 进度信号，ip，1~100
    rate_signal =pyqtSignal([int])
    # 结束信号，ip，成败，描述
    end_signal = pyqtSignal([bool,str])

    failed_signal = pyqtSignal()

    def __init__(self):
        QObject.__init__(self)
        self.app = QApplication(sys.argv)
        self.widget = QWidget()
        self.widget.setAttribute(Qt.WidgetAttribute.WA_DeleteOnClose)
        self.widget.resize(703, 409)
        self.widget.setWindowTitle("赋码仪程序升级")
        self.file_list_init()
        self.com_but_init()
        self.save_but_init()
        self.updatas_but_init()
        # self.cmd_but_init()
        self.sstate_but_init()
        self.scheme_but_init()
        self.com_init()
        self.combsp_init()
        self.widget.destroyed.connect(self.quit)
        self.failed_signal.connect(self.updata_failed)
        self.cmd=0
        self.port_is_open=False
        
    def quit(self):
        self.close_port()
        # 程序退出
        qApp.exit(1)
        # 初始化文件列表
    def file_list_init(self):
        self.file_list = QListWidget(self.widget)
        self.file_list.setObjectName(u"file_list")
        self.file_list.setGeometry(QRect(25, 60, 531, 310))
        self.file_list.setFrameShape(QFrame.Shape.Box)
        self.file_list.setMidLineWidth(1)
        self.file_list.setVerticalScrollBarPolicy(Qt.ScrollBarPolicy.ScrollBarAsNeeded)
        self.file_list.setSelectionMode(QAbstractItemView.SelectionMode.SingleSelection)
        self.file_list.itemDoubleClicked.connect(self.item_clicked)



        # 初始化打开端口按钮
    def com_but_init(self):
        self.com_but = QPushButton(self.widget)
        self.com_but.setObjectName(u"com_but")
        self.com_but.setGeometry(QRect(590, 10, 93, 28))
        self.com_but.setText("打开端口")
        self.com_but.clicked.connect(self.com_but_clicked)


        # 初始化发送文件按钮
    def save_but_init(self):
        self.save_but = QPushButton(self.widget)
        self.save_but.setObjectName(u"save_but")
        self.save_but.setGeometry(QRect(590, 60, 93, 28))
        self.save_but.setText("发送文件")
        self.save_but.clicked.connect(self.save_but_clicked)

        # 初始化发送命令按钮
    def cmd_but_init(self):
        self.cmd_but = QPushButton(self.widget)
        self.cmd_but.setObjectName(u"save_but")
        self.cmd_but.setGeometry(QRect(590, 100, 93, 28))
        self.cmd_but.setText("升级MCU")
        self.cmd_but.clicked.connect(self.cmd_but_clicked)

        # 初始化升级小板按钮
    def updatas_but_init(self):
        self.updatas_but = QPushButton(self.widget)
        self.updatas_but.setObjectName(u"updatas_but")
        self.updatas_but.setGeometry(QRect(590, 100, 93, 28))
        self.updatas_but.setText("升级小板")
        self.updatas_but.clicked.connect(self.updatas_but_clicked)

        # 初始化在线状态按钮
    def sstate_but_init(self):
        self.sstate_but = QPushButton(self.widget)
        self.sstate_but.setObjectName(u"sstate_but")
        self.sstate_but.setGeometry(QRect(590, 140, 93, 28))
        self.sstate_but.setText("主板参数")
        self.sstate_but.clicked.connect(self.sstate_but_clicked)

        # 初始化方案状态按钮
    def scheme_but_init(self):
        self.sstate_but = QPushButton(self.widget)
        self.sstate_but.setObjectName(u"sstate_but")
        self.sstate_but.setGeometry(QRect(590, 180, 93, 28))
        self.sstate_but.setText("方案参数")
        self.sstate_but.clicked.connect(self.scheme_but_clicked)

        # com口
    def com_init(self):
        self.com = QComboBox(self.widget)
        self.com.setObjectName(u"com")
        self.com.setGeometry(QRect(85, 10, 300, 25))
        self.com.setEditable(True)
        self.com.currentIndexChanged.connect(self.com_changed)
        self.com.addItem("utcp:7777")
        ports_list = list(serial.tools.list_ports.comports())
        for comport in ports_list:
            # print(comport.name,comport.description)
            self.com.addItem(comport.name+":"+comport.description)
        self.com_label = QLabel(self.widget)
        self.com_label.setObjectName(u"label")
        self.com_label.setGeometry(QRect(30, 16, 72, 15))
        self.com_label.setText("COM口:")

        # 选择波特率
    def combsp_init(self):
        self.combsp = QComboBox(self.widget)
        self.combsp.setObjectName(u"combsp")
        self.combsp.setGeometry(QRect(470, 10, 80, 25))
        self.combsp.setEditable(True)
        self.combsp.addItem("115200")
        self.combsp.addItem("9600")
        self.combsp_label = QLabel(self.widget)
        self.combsp_label.setObjectName(u"label")
        self.combsp_label.setGeometry(QRect(410, 16, 72, 15))
        self.combsp_label.setText("波特率:")

        # 显示消息框
    def show_msg(self,msg:str):
        m=QMessageBox(self.widget)
        m.setAttribute(Qt.WidgetAttribute.WA_DeleteOnClose)
        m.setText(msg)
        m.setWindowTitle("提示")
        m.show()

        # 找到已选择的文件
    def get_selected_file_by_type(self,type:str):
        file_list=[]    
        items=self.file_list.selectedItems()
        for i in items:
            if(i.text()[-len(type):]==type):
                file_list.append(i.text())
        if(len(file_list)!=1):
            self.show_msg("请选择一个并且只选择一个 "+type+" 文件")
            return ""
        return file_list[0]

        # 获取已选择的文件列表
    def get_selected_fils(self):
        file_list=[]    
        items=self.file_list.selectedItems()
        if(len(items)==0):
            self.show_msg("请选择至少一个文件")
            return []
        have_elf=False
        have_bin=False
        have_lua=False
        for i in items:
            if(i.text()[-4:]==".elf"):
                if(have_elf==True):
                    self.show_msg("只可选择一个 .elf 文件")
                    return []
                have_elf=True
            if(i.text()[-4:]==".bin"):
                if(have_bin==True):
                    self.show_msg("只可选择一个 .bin 文件")
                    return []
                have_bin=True
            if(i.text()[-4:]==".lua"):
                if(have_lua==True):
                    self.show_msg("只可选择一个 .lua 文件")
                    return []
                have_lua=True
            file_list.append(i.text())
        return file_list



    def item_clicked(self,item:QListWidgetItem ):
        print("item clicked.")
        print("slected item is",item.text())

    def com_but_clicked(self):
        print("com but clicked")
        if(self.port_is_open==False):
            self.open_port()
        else:
            self.close_port()

    def com_changed(self,index:int):
        print("com changed")
        self.close_port()
    def get_cmd(self,file:str):
        l=[(".bin",0xee),(".pkt",0xed),(".json",0x32)]
        for i in l:
            if(file.endswith(i[0])):
                return i[1]
        return 0

        # 发送文件按钮按下
    def save_but_clicked(self):
        print("save_but_clicked.")
        path = self.getpath()+"file\\"
        file_list=self.get_selected_fils()
        if(len(file_list)==0):
            return
        self.cmd=self.get_cmd(file_list[0])
        print("cmd=",self.cmd)
        w=QDialog(self.widget)
        w.resize(703-150, 1*40+20)
        w.setWindowTitle("上传文件")
        w.setAttribute(Qt.WidgetAttribute.WA_DeleteOnClose)
        self.handle_=prottcp.handle()
        if(file_list[0].endswith(".json")):
            self.handle_.set_json("file/"+file_list[0])
        else:
            self.handle_.set_file("file/"+file_list[0])
        d=self.handle_.get_data()
        try:
            self.port.send(self.cmd,d)
        except Exception as e:
            print("com not open")
            self.show_msg("端口未打开")
            del self.handle_
            w.close()
            return
        self.creat_progress(w)
        w.show()
        # self.failed_signal.connect(w.close)
        # w.destroyed.connect(self.close_port)
    def updata_failed(self):
        self.show_msg("打开端口失败")

        # 创建进度条
    def creat_progress(self,father:QDialog):
        self.probar_list=[]
        for i in range(1):
            p=progress_box()
            p.init(father,"发送文件",+i*40)
            self.rate_signal.connect(p.set_rate)
            self.probar_list.append(p)

        # 发送命令按钮按下
    def cmd_but_clicked(self):
        print("cmd_but clicked.")
        file=self.get_selected_file_by_type(".bin")
        if(len(file)==0):
            return
        print("file:",file)
        w=QDialog(self.widget)
        w.resize(703-150, 1*40+20)
        w.setWindowTitle("升级mcu")
        self.updata_mcu(file)
        self.creat_progress(w)
        w.show()


    def sstate_but_clicked(self):
        print("sstate_but clicked.")
        try:
            self.port.send_str("sysinfo")
        except Exception as e:
            print("com not open")
            print(str(e))

    def scheme_but_clicked(self):
        print("scheme_but clicked.")
        try:
            self.port.send_str("scheme")
        except Exception as e:
            print("com not open")
            print(str(e))

    def updatas_but_clicked(self):
        print("updatas_but clicked.")
        try:
            self.port.send_str("updatas 1,2,3,4,5,6,7,8,9,10")
            self.show_msg("已发送升级指令，请留意小板升级情况")
        except Exception as e:
            self.show_msg("命令发送失败")
            print("com not open")
            print(str(e))
            

        # 开始运行
    def run(self):
        self.widget.show()
        sys.exit(self.app.exec())

    def set_port_state(self,state:bool):
        self.port_is_open=state
        if(state==True):
            self.com_but.setText("关闭端口")
        else:
            self.com_but.setText("打开端口")

        # 扫描文件
    def scan_file(self):
        self.file_list.addItems(self.find_type([".bin",".json",".pkt"]))


        # 扫描指定类型的文件
    def find_type(self,types:list):
        path = self.getpath()+"file\\"
        if not os.path.exists(path):
            os.makedirs(path)
        list=os.listdir(path)
        file_list=[]
        for t in types:
            for i in list:
                if(len(t)>0):
                    if(i[-len(t):]==t):
                        file_list.append(i)
                else:
                    file_list.append(i)
        return file_list
    
        # 获得文件绝对路径
    def getpath(self):
        path=os.path.abspath(sys.argv[0])
        list_str=path.split("\\")
        return path.replace(list_str[-1],"")
    
        # 调用此函数打开通信
    def open_port(self):
        t = threading.Thread(target=self.com_thread, args=())
        t.start()
    def com_thread(self):
        self.port=prottcp.protu()
        item=self.com.itemText(self.com.currentIndex())
        bsp=self.combsp.itemText(self.combsp.currentIndex())
        com=item.split(":")[0]
        if(com!="utcp"):
            item=com+":"+bsp
        print("item text:",item)
        if(self.port.init(item)==False):
            print("init port failed.")
            self.failed_signal.emit()
            self.set_port_state(False)
            return
        else:
            print("init port success.")
            self.set_port_state(True)
            self.port.recv_signal.connect(self.recv_slot)
            self.port.recv_str_signal.connect(self.recv_str_slot)
            self.port.start_recv()
            self.port.wait()
    def close_port(self):
        print("close port")
        self.set_port_state(False)
        try:
            self.port.close()
            self.port.recv_signal.disconnect(self.recv_slot)
            self.port.recv_str_signal.disconnect(self.recv_str_slot)
        except Exception as e:
            pass
    def recv_str_slot(self,cmd:int,txt:str,err:str):
        print("|-|",txt)
    def recv_slot(self,cmd:int,data:bytearray,err:str):
        # print("recv:",cmd,data)
        if(self.cmd!=cmd):
            return
        try:
            data=self.handle_.get_data()
            if(len(data)>0):
                self.port.send(cmd,data)
                rate=self.handle_.get_rate()
                self.rate_signal.emit(rate)
            else:
                del self.handle_
        except Exception as e:
            print(str(e))
        # 开始升级mcu
    def updata_mcu(self,file):
        pass

        # 小板通信测试
    def comm_test(self):
        pass



    def end_slot(self,ip,ack,err):
        # print(ip,ack,err)
        if(ack==False):
            self.show_msg(ip+":"+err)

    def rate_slot(self,rate):
        # print("rate signal:",ip,rate)
        self.rate_signal.emit(rate)

    def data_slot(self,ip,text):
        pass

        # 创建数据显示
    def creat_databoxs(self,father:QDialog,data_list:list):
        self.datab_list=[]
        for i in range(len(data_list)):
            p=data_box()
            p.init(father,data_list[i][0],data_list[i][1],+i*50)
            self.datab_list.append(p)






class locker():
    def _get_lock(self):
        file_name = os.path.basename(__file__)
        # linux等平台依然使用标准的/var/run，其他nt等平台使用当前目录
        if os.name == "posix":
            lock_file_name = f"/var/run/{file_name}.pid"
        else:
            lock_file_name = f"{file_name}.pid"
        self.fd = open(lock_file_name, "w")
        try:
            portalocker.lock(self.fd, portalocker.LOCK_EX | portalocker.LOCK_NB)
            # 将当前进程号写入文件
            # 如果获取不到锁上一步就已经异常了，所以不用担心覆盖
            self.fd.writelines(str(os.getpid()))
            # 写入的数据太少，默认会先被放在缓冲区，我们强制同步写入到文件
            self.fd.flush()
        except:
            print(f"{file_name} have another instance running.")
            exit(1)
 
    def __init__(self):
        self._get_lock()
 
    # 和fcntl有点区别，portalocker释放锁直接有unlock()方法
    # 还是一样，其实并不需要在最后自己主动释放锁
    def __del__(self):
        portalocker.unlock(self.fd)


if __name__ == "__main__":
    lock=locker()
    dlg=updata_dlg()
    dlg.scan_file()
    dlg.run()