Python Worker线程类偶发异常及Tk askopenfilename主线程报错求助
线程Worker脚本偶发异常及Tk弹窗调用错误问题
我编写了一个名为worker的Python脚本,包含Worker和CreateThread两个类,用于处理意外异常并通过线程运行耗时函数以避免GUI冻结。但该脚本存在偶发异常:有时调用Worker类运行函数一切正常,有时却出现奇怪错误甚至GUI崩溃。我尝试调整线程处理顺序、适配错误,但仍存在隐藏问题未解决。
以下是我的worker代码:
# Create a worker class # class Worker(QObject): my_started = pyqtSignal() my_finished = pyqtSignal() my_error = pyqtSignal(str, str) my_success = pyqtSignal() def __init__(self): super(Worker, self).__init__() def run(self, func): try: self.my_started.emit() time.sleep(0.1) x = func() if x is None: self.my_success.emit() self.my_finished.emit() return if x[0] == -1: function_error = x[1] trace_back = x[2] print(f'\nFunction error: {function_error}') print(trace_back) self.my_error.emit(str(function_error), trace_back, str(func.__name__)) self.my_finished.emit() return except Exception as e: self.trace_back = str(traceback.format_exc()) print(f'\nFunction error: {e}') print(self.trace_back) self.my_error.emit(str(e), self.trace_back) self.my_finished.emit() # Creating thread manger class # class CreateThread(Worker, QObject): def __init__(self, atp_name): super(CreateThread, self).__init__() self.stop_flag = False self.atp_name = atp_name self.thread = QThread() self.worker = Worker() def start_thread(self, func, **kwargs): self.kwargs = kwargs self.func_name = func.__name__ if self.stop_flag: return if self.func_name == '<lambda>': self.func_name = self.kwargs['func_name'] # Thread and Worker init # self.thread = QThread() self.worker = Worker() act = partial(self.worker.run, func) self.worker.moveToThread(self.thread) self.thread.started.connect(act) # Signal to Function connection # self.worker.my_started.connect(self.started_signal_connection) self.worker.my_finished.connect(self.finished_signal_connection) self.worker.my_error.connect(self.error_signal_connection) self.worker.my_success.connect(self.success_signal_connection) # Handling the worker and the thread deletion # self.worker.my_finished.connect(self.thread.quit) self.worker.my_finished.connect(self.worker.deleteLater) self.thread.finished.connect(self.thread.deleteLater) self.thread.start() # Connect signals from worker # def started_signal_connection(self): self.stop_flag = True print(f"\nFunction {self.func_name} started\n") def finished_signal_connection(self): self.stop_flag = False print("\nWorker finished\n") def error_signal_connection(self, error_name, trace_back): self.today = date.today() self.now = datetime.now() self.current_date = str( self.now.strftime("%d/%m/%Y %H:%M:%S")).replace(' ', '_').replace( '/', '_').replace(':', '_') self.header_name = f'{self.atp_name}_{self.func_name}_Date_{self.current_date}' self.target_folder = 'target_folder' if os.path.isfile(self.header_name) == False: file_name = f'{self.target_folder}{self.header_name}.txt' try: with open(file_name, 'w', encoding='utf-8') as f: f.write(f'''ATP: {self.atp_name}\nFunction name: {self.func_name}\nError: {error_name}\nDate: {self.current_date}\n\n{trace_back}''') except Exception as e: print(e) def success_signal_connection(self): self.stop_flag = False print("\nFunction success")
另外,我发现当Worker调用包含Tk askopenfilename()的函数时总会出现问题,以下是报错回溯示例:
line 57, in run **x = func()** line 64, in open_main_terminal **self.path1, self.file_name = upload_item()** line 62, in upload_item **filename = askopenfilename()** line 375, in askopenfilename **return Open(**options).show()** line 39, in show **w = Frame(self.master)** line 2741, in __init__ **Widget.__init__(self, master, 'frame', cnf, {}, extra)** line 2296, in __init__ (widgetName, self._w) + extra + self._options(cnf)) **RuntimeError: main thread is not in main loop**
内容的提问来源于stack exchange,提问作者Zohar Hazoom
相关产品推荐
相关产品推荐

