错误处理和 threading.Thread() 问题

标签：python error-handling python-multithreading pysimplegui

我对 Python 比较陌生，最初我按照我喜欢的方式进行错误处理。我想使用 PySimpleGUI 添加进度条，经过一些研究发现我需要使用线程。一旦我实现了线程，错误处理就不再像我想要的那样工作，而且我似乎不知道如何修复它。我将提供进度条之前和之后的代码，但是之后的代码没有任何错误处理。

旧代码

import PySimpleGUI as sg
import functions_new as fn
import sys
import os

sg.theme("Default1")
sg.set_options(font=("Segoe UI Variable", 10))

input1 = sg.Input()
archive_choose = sg.FolderBrowse("Select Source Directory",
                                 key="source",
                                 pad=(5, 5))

input2 = sg.Input()
BOM_choose = sg.FileBrowse("Select BOM",
                           key="excel",
                           pad=(5, 5))

input3 = sg.Input()
save_dir = sg.FolderBrowse("Select Save Location",
                           key="save",
                           pad=(5, 5))

exec_button = sg.Button("Execute",
                        pad=(5, 5),
                        button_color=('White', 'NavyBlue'))



window = sg.Window(title="Step Name Replace",
                   layout=[[input1, archive_choose],
                           [input2, BOM_choose],
                           [input3, save_dir],
                           [exec_button]])

while True:
    event, values = window.read()

    if event == sg.WIN_CLOSED or event == "Exit":
        break

    if event == "Execute":

        # calls func to search directory and store stepfiles in list
        filepaths = fn.stpFinder(values["source"])
        filepaths = [item.replace("\\", "/") for item in filepaths]

        try:
            print(f"Source Directory: {values['source']}")
            print(f"Source Directory: {values['excel']}")
            print(f"Source Directory: {values['save']}")

            # calls func to search directory and store stepfiles in list
            filepaths = fn.stpFinder(values["source"])
            filepaths = [item.replace("\\", "/") for item in filepaths]
            print(f"Found {len(filepaths)} .stp files")

            #gets df from excel
            df = fn.extractMREBOM(values["excel"])

            #runs name replace function
            for filepath in filepaths:
                new_name = os.path.join(values["save"], os.path.basename(filepath).replace(".stp", "_rename.stp"))

                print(f"Processing file: {filepath} -> {new_name}")
                fn.stepNameReplace(filepath, df, new_name)

            sg.popup("Process completed successfully!",
                     font=("Segoe UI Variable", 10))

        except PermissionError:
            sg.popup("Ensure that BOM and folder locations are closed before executing\n\n"
                     "Click OK to terminate",
                     font=("Segoe UI Variable", 10))
            sys.exit()

        except Exception as e:
            sg.popup(f"An error occurred: {e}",
                     font=("Segoe UI Variable", 10))

window.close()

新代码

import PySimpleGUI as sg
import functions_new as fn
import sys
import os
import threading


def execute_func(window, values):
    # Create new directory within source directory
    output_dir = os.path.join(values["source"], "SNR_Output")
    os.makedirs(output_dir, exist_ok=True)

    # Calls func to search directory and store stepfiles in list
    files = fn.stpFinder(values["source"])
    files = [item.replace("\\", "/") for item in files]
    print(f"Found {len(files)} .stp files")

    # Extracts df from excel
    df = fn.extractMREBOM(values["excel"])

    total_files = len(files)

    for index, file in enumerate(files):

        # Creates name for output step file
        new_name = os.path.join(output_dir, os.path.basename(file))

        # "Update" assigned to be called by elif to update status bar through each iteration
        window.write_event_value("Update", (index + 1, total_files))

        # Processes the file
        print(f"Processing file: {file}")
        fn.stepNameReplace(file, df, new_name)

    # Once loop completes, "Done" value assigned to be called by elif
    window.write_event_value("Done", None)

sg.theme("Default1")
sg.set_options(font=("Segoe UI Variable", 10))

layout = [
    [sg.Input(), sg.FolderBrowse("Select Source Directory", key="source", pad=(5, 5))],
    [sg.Input(), sg.FileBrowse("Select BOM", key="excel", pad=(5, 5))],
    [sg.Button("Execute", button_color=("White", "Grey")),
     sg.ProgressBar(100, orientation='h', size=(20, 20), key='progressbar', bar_color=("Salmon", "LightGrey")),
     sg.Text("", key="info")]
]

window = sg.Window(title="Step Name Replace (SNR)", layout=layout)


while True:
    event, values = window.read(timeout=100)

    if event == sg.WIN_CLOSED:
        break

    elif event == "Execute":
        print(f"Source Directory: {values['source']}")
        print(f"Source Directory: {values['excel']}")

        threading.Thread(target=execute_func, args=(window, values), daemon=True).start()

    elif event == "Update":
        current_file, total_files = values[event]
        window["progressbar"].update(current_file * 100 / total_files)
        window["info"].update(f"Working file {current_file} of {total_files}.")

    elif event == "Done":
        sg.popup("Process completed successfully!",
                 font=("Segoe UI Variable", 10))
        sys.exit()

window.close()

functions_new.py

from pandas import read_excel
import os
import sys
import PySimpleGUI as sg


# searches selected directory and extracts stepfiles into list
def stpFinder(source_dir):
    stp_files = []
    extensions = (".stp", ".step", ".STP", ".STEP")
    for root, dirs, files in os.walk(source_dir):
        for file in files:
            if file.endswith(extensions):
                stp_files.append(os.path.join(root, file))
    return stp_files


# Reads BOM excel and returns dataframe
def extractMREBOM(path):
    df = read_excel(path, sheet_name="BOM", dtype="str")

    des_count = df["DESCRIPTION"].count()
    num_count = df["DOC NUMBER"].count()
    type_count = df["DOC TYPE"].count()
    part_count = df["DOC PART"].count()

    # Checks if BOM is formatted correctly before returning df
    if des_count == num_count and type_count == part_count:
        return df

    else:
        sg.popup("BOM formatted incorrectly (Inconsistent number of rows)\n \n"
                 "Click OK to Terminate",
                 font=("Segoe UI Variable", 10))
        sys.exit()


# Uses returned dataframe to replace string names with BOM Description, then rename
def stepNameReplace(filepath_arg, df_arg, newName):

    output = open(newName, 'w+')
    file = open(filepath_arg, 'rb')

    for row in file:
        tmp_str = row.decode("utf-8").replace("\r\n", "\r")

        # Perform all replacements for the current line
        for i in range(len(df_arg["DESCRIPTION"])):
            tmp_str = tmp_str.replace(
                f"{df_arg['DOC NUMBER'][i]}_{df_arg['DOC TYPE'][i].upper()}_{df_arg['DOC PART'][i]}",
                df_arg['DESCRIPTION'][i])  # case Upper

            tmp_str = tmp_str.replace(
                f"{df_arg['DOC NUMBER'][i]}_{df_arg['DOC TYPE'][i].lower()}_{df_arg['DOC PART'][i]}",
                df_arg['DESCRIPTION'][i])  # case Lower

        # After all replacements, write the modified line to the output file
        output.write(tmp_str)
    output.close()
    return

好的，让我们来解决的错误处理和线程问题。的代码的主要问题在于，正试图从一个在后台运行的线程更新 PySimpleGUI 用户界面，而 PySimpleGUI 并不是线程安全的。这意味着直接从线程更新界面元素（如进度条）可能会导致意外行为。

以下是关于如何安全有效地实现它的方法：

使用 window.write_event_value() 进行线程间通信 ：
PySimpleGUI 提供了 write_event_value() 方法，允许线程安全地与主事件循环通信。
工作线程在执行任务时应生成事件（例如，“更新”、“完成”、“错误”）以及任何必要的数据。
主事件循环将监听这些事件并相应地更新 GUI。
在主线程中处理错误 ：
将 try...except 块移到主事件循环中。这使能够捕获工作线程中发生的任何异常，并在 GUI 中显示适当的错误消息。
使用 queue 模块进行更强大的通信（可选）：
对于更复杂的情况或大量数据传输，请考虑使用 queue 模块在工作线程和主线程之间创建线程安全的队列。

以下是的代码的改进版本，其中包含这些更改：

import PySimpleGUI as sg
import functions_new as fn
import sys
import os
import threading

def execute_func(window, values):
    try:
        # 创建源目录下的新目录
        output_dir = os.path.join(values["source"], "SNR_Output")
        os.makedirs(output_dir, exist_ok=True)

        # 调用函数以搜索目录并将 step 文件存储在列表中
        files = fn.stpFinder(values["source"])
        files = [item.replace("\\", "/") for item in files]
        window.write_event_value("FileCount", len(files))  # 更新文件总数

        # 从 Excel 中提取 df
        df = fn.extractMREBOM(values["excel"])

        total_files = len(files)
        for index, file in enumerate(files):
            # 创建输出 step 文件的名称
            new_name = os.path.join(output_dir, os.path.basename(file))

            # 更新进度条
            window.write_event_value("Update", (index + 1, total_files))

            # 处理文件
            fn.stepNameReplace(file, df, new_name)

        # 任务完成后发送 "Done" 事件
        window.write_event_value("Done", None)

    except Exception as e:
        # 发生错误时发送 "Error" 事件
        window.write_event_value("Error", str(e))

sg.theme("Default1")
sg.set_options(font=("Segoe UI Variable", 10))

layout = [
    [sg.Input(), sg.FolderBrowse("选择源目录", key="source", pad=(5, 5))],
    [sg.Input(), sg.FileBrowse("选择 BOM", key="excel", pad=(5, 5))],
    [sg.Button("执行", button_color=("White", "Grey"))],
    [sg.ProgressBar(100, orientation='h', size=(40, 20), key='progressbar', bar_color=("Salmon", "LightGrey"))],
    [sg.Text("", size=(40, 1), key="info")],
]

window = sg.Window(title="Step Name Replace (SNR)", layout=layout)

while True:
    event, values = window.read(timeout=100)

    if event == sg.WIN_CLOSED:
        break

    elif event == "执行":
        print(f"源目录: {values['source']}")
        print(f"源目录: {values['excel']}")

        # 禁用执行按钮以防止多次点击
        window["执行"].update(disabled=True)
        # 重置进度条和信息文本
        window["progressbar"].update(0)
        window["info"].update("")

        threading.Thread(target=execute_func, args=(window, values), daemon=True).start()

    elif event == "FileCount":
        # 初始化进度条的最大值
        window["progressbar"].update(0, values[event])

    elif event == "Update":
        current_file, total_files = values[event]
        window["progressbar"].update(current_file)
        window["info"].update(f"正在处理文件 {current_file}/{total_files}")

    elif event == "Done":
        sg.popup("处理完成！", font=("Segoe UI Variable", 10))
        window["执行"].update(disabled=False)  # 重新启用执行按钮

    elif event == "Error":
        sg.popup(f"发生错误: {values[event]}", font=("Segoe UI Variable", 10))
        window["执行"].update(disabled=False)  # 重新启用执行按钮

window.close()

说明：

工作线程现在使用 window.write_event_value() 发送不同类型的事件（"Update"、"Done"、"Error"）。
主事件循环监听这些事件并相应地更新 GUI。
try...except 块位于主事件循环中，以捕获工作线程中发生的任何异常。
错误消息现在显示在 PySimpleGUI 弹出窗口中。
添加了对进度条最大值的初始化，以及在处理完成后重新启用 "执行" 按钮。

这些更改将使的错误处理更加可靠，并确保 GUI 更新与的线程同步。

标签：python,error-handling,python-multithreading,pysimplegui
From： 78844332

错误处理和 threading.Thread() 问题

相关文章

赞助商

阅读排行