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

Question

我是 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

Answer 1

线程时，您无法调用

sg.popup

或

extractMREBOM

中的大多数 GUI 调用，请尝试通过

window.write_event_value

向事件循环发送信号以调用

sg.popup

，然后在事件循环中关闭窗口。另一个问题是在

stepNameReplace

中打开的文件未关闭。

一个简单的例子供大家参考

import threading
from time import sleep
import PySimpleGUI as sg

def execute_func(window):
    global stop
    total = 100
    for i in range(total):
        if stop:
            break
        sleep(0.1)
        value = check(i+1)
        if value is None:
            window.write_event_value("POPUP", i+1)
        elif value:
            window.write_event_value("Update", (i+1, total))
        else:
            window.write_event_value("Pass", i+1)
    window.write_event_value("Done", None)

def check(count):
    if count == 50:
        return None
    if count % 7:
        return True
    else:
        return False

sg.theme("DarkBlue")
sg.set_options(font=("Courier New", 16))

layout = [
    [sg.Multiline("", size=(40, 10), key="PASS")],
    [sg.ProgressBar(100, orientation='h', size=(20, 20), expand_x=True, expand_y=True, key='progressbar'),
     sg.Button("Execute")],
]
window = sg.Window("Demo", layout)
running, stop = False, True

while True:

    event, values = window.read()

    if event == sg.WIN_CLOSED:
        stop = True
        sleep(1)
        break

    elif event == "Execute" and not running:
        running, stop = True, False
        threading.Thread(target=execute_func, args=(window,), daemon=True).start()

    elif event == "Update":
        count, total = values[event]
        window["progressbar"].update(current_count=count)

    elif event == "Pass":
        count = values[event]
        window["PASS"].update(f"Number {count} passed !!!\n", append=True, autoscroll=True)

    elif event == "POPUP":
        sg.popup("Pause here for count to 50, threading still running !", title="Message")

    elif event == "Done":
        running = False
        window["PASS"].update(f"~~~ All Done ~~~", append=True, autoscroll=True)

window.close()