Error handling and threading.Thread() issues

Question

I am newer to Python, and I originally had error handling working as I liked. I wanted to add a progress bar using PySimpleGUI and after some research discovered that I needed to use threading. Once I implemented threading, the error handling no longer works as I'd like, and I cannot seem to figure out how to fix it. I will provide the code before the progress bar and after, but the after code does not have any error handling.

Old Code

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()

New Code

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

• Stas Simonov: And what's wrong with try ... except inside execute_func?
• Jason Yang: You cannot call sg.popup or most of GUI calls in extractMREBOM when threading, try to send a signal by window.write_event_value to your event loop to call popup, then close your window in your event loop. Another one issue is the file opened in stepNameReplace not closed.
• Dominick Cole: @StasSimonov, when I do this, should it be a signal via window.write_event_value to a loop or can I call a popup within the function?
• Stas Simonov: @DominickCole I strongly suggest you don't call any GUI functions inside a non-GUI thread execute_func. Instead, send a signal (event) to main (GUI) thread.
• Dominick Cole: @JasonYang, thank you for pointing out the file close issue. Since extractMREBOM is inside the functions_new.py and not the gui.py file, how would I call window.write_event_value if window is undefined in functions_new.py?

Answer 1

You cannot call sg.popup or most of GUI calls in extractMREBOM when threading, try to send a signal by window.write_event_value to your event loop to call sg.popup, then close your window in your event loop. Another one issue is the file opened in stepNameReplace not closed.

A simple example here for your reference

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()

enter image description here

• Mike from PSG: Not critical, but did want to point out that PySimpleGUI added a simplified threading interface that includes ability to generate an event when the thread exits. This is an example from a Demo Program that uses threading with progress bars. I find the lambda easier to understand as it looks exactly like the call that will be made to start the thread (the args are not a separate parm)... window.start_thread(lambda: the_thread(window), (THREAD_KEY, DL_THREAD_EXITNG)) `