编辑2019-mar-12:在我看来,答案是:它可能已经完成,但需要付出很多努力才能记住QTextEdit来自哪个行的行为与插图有关。另外,由于tdm默认写入stderr,你最终也会捕获所有异常跟踪。 这就是我将自己的答案标记为已解决的原因:我发现实现相同目的更为优雅:在pyqt中显示正在发生的事情。
这是我获得接近预期行为的最好机会。 它并没有完全回应这个问题,因为我改变了GUI设计。 所以我不会把它投票解决。而且,这是在一个单独的python文件中完成的。我计划进一步挑战这个解决方案,看看它是否适用于执行tqdm导入的真实python模块。
我以一种非常丑陋的方式修补了基本的tqdm类。主要技巧是:
tqdm.orignal_class = tqdm.tqdm
class TQDMPatch(tqdm.orignal_class):
super(TQDMPatch, self).__init__(... change some params ...)
WriteStream()
Queue()
\r
它既可以在单个python文件中使用,也可以使用多个分离的模块。在后一种情况下,启动时的进口订单至关重要。
截图:
在启动处理之前
在处理过程中
在处理结束时
这是代码
# coding=utf-8 import datetime import logging import sys import time from queue import Queue from PyQt5.QtCore import pyqtSlot, pyqtSignal, QObject, QThread, Qt from PyQt5.QtGui import QTextCursor, QFont from PyQt5.QtWidgets import QTextEdit, QWidget, QToolButton, QVBoxLayout, QApplication, QLineEdit # DEFINITION NEEDED FIRST ... class WriteStream(object): def __init__(self, q: Queue): self.queue = q def write(self, text): self.queue.put(text) def flush(self): pass # prepare queue and streams queue_tqdm = Queue() write_stream_tqdm = WriteStream(queue_tqdm) ################## START TQDM patch procedure ################## import tqdm # save original class into module tqdm.orignal_class = tqdm.tqdm class TQDMPatch(tqdm.orignal_class): """ Derive from original class """ def __init__(self, iterable=None, desc=None, total=None, leave=True, file=None, ncols=None, mininterval=0.1, maxinterval=10.0, miniters=None, ascii=None, disable=False, unit='it', unit_scale=False, dynamic_ncols=False, smoothing=0.3, bar_format=None, initial=0, position=None, postfix=None, unit_divisor=1000, gui=False, **kwargs): super(TQDMPatch, self).__init__(iterable, desc, total, leave, write_stream_tqdm, # change any chosen file stream with our's 80, # change nb of columns (gui choice), mininterval, maxinterval, miniters, ascii, disable, unit, unit_scale, False, smoothing, bar_format, initial, position, postfix, unit_divisor, gui, **kwargs) print('TQDM Patch called') # check it works @classmethod def write(cls, s, file=None, end="\n", nolock=False): super(TQDMPatch, cls).write(s=s, file=file, end=end, nolock=nolock) # all other tqdm.orignal_class @classmethod methods may need to be redefined ! # I mainly used tqdm.auto in my modules, so use that for patch # unsure if this will work with all possible tqdm import methods # might not work for tqdm_gui ! import tqdm.auto as AUTO # change original class with the patched one, the original still exists AUTO.tqdm = TQDMPatch ################## END of TQDM patch ################## # normal MCVE code __is_setup_done = False class MainApp(QWidget): def __init__(self): super().__init__() setup_logging(self.__class__.__name__) self.__logger = logging.getLogger(self.__class__.__name__) self.__logger.setLevel(logging.DEBUG) # create stdout text queue self.queue_std_out = Queue() sys.stdout = WriteStream(self.queue_std_out) layout = QVBoxLayout() self.setMinimumWidth(500) self.btn_perform_actions = QToolButton(self) self.btn_perform_actions.setText('Launch long processing') self.btn_perform_actions.clicked.connect(self._btn_go_clicked) self.text_edit_std_out = StdOutTextEdit(self) self.text_edit_tqdm = StdTQDMTextEdit(self) self.thread_initialize = QThread() self.init_procedure_object = InitializationProcedures(self) # std out stream management # create console text read thread + receiver object self.thread_std_out_queue_listener = QThread() self.std_out_text_receiver = ThreadStdOutStreamTextQueueReceiver(self.queue_std_out) # connect receiver object to widget for text update self.std_out_text_receiver.queue_std_out_element_received_signal.connect(self.text_edit_std_out.append_text) # attach console text receiver to console text thread self.std_out_text_receiver.moveToThread(self.thread_std_out_queue_listener) # attach to start / stop methods self.thread_std_out_queue_listener.started.connect(self.std_out_text_receiver.run) self.thread_std_out_queue_listener.start() # NEW: TQDM stream management self.thread_tqdm_queue_listener = QThread() self.tqdm_text_receiver = ThreadTQDMStreamTextQueueReceiver(queue_tqdm) # connect receiver object to widget for text update self.tqdm_text_receiver.queue_tqdm_element_received_signal.connect(self.text_edit_tqdm.set_tqdm_text) # attach console text receiver to console text thread self.tqdm_text_receiver.moveToThread(self.thread_tqdm_queue_listener) # attach to start / stop methods self.thread_tqdm_queue_listener.started.connect(self.tqdm_text_receiver.run) self.thread_tqdm_queue_listener.start() layout.addWidget(self.btn_perform_actions) layout.addWidget(self.text_edit_std_out) layout.addWidget(self.text_edit_tqdm) self.setLayout(layout) self.show() @pyqtSlot() def _btn_go_clicked(self): # prepare thread for long operation self.init_procedure_object.moveToThread(self.thread_initialize) self.thread_initialize.started.connect(self.init_procedure_object.run) self.thread_initialize.finished.connect(self.init_procedure_object.finished) # start thread self.btn_perform_actions.setEnabled(False) self.thread_initialize.start() class ThreadStdOutStreamTextQueueReceiver(QObject): queue_std_out_element_received_signal = pyqtSignal(str) def __init__(self, q: Queue, *args, **kwargs): QObject.__init__(self, *args, **kwargs) self.queue = q @pyqtSlot() def run(self): self.queue_std_out_element_received_signal.emit('---> STD OUT Queue reception Started <---\n') while True: text = self.queue.get() self.queue_std_out_element_received_signal.emit(text) # NEW: dedicated receiving object for TQDM class ThreadTQDMStreamTextQueueReceiver(QObject): queue_tqdm_element_received_signal = pyqtSignal(str) def __init__(self, q: Queue, *args, **kwargs): QObject.__init__(self, *args, **kwargs) self.queue = q @pyqtSlot() def run(self): self.queue_tqdm_element_received_signal.emit('\r---> TQDM Queue reception Started <---\n') while True: text = self.queue.get() self.queue_tqdm_element_received_signal.emit(text) class StdOutTextEdit(QTextEdit): # QTextEdit): def __init__(self, parent): super(StdOutTextEdit, self).__init__() self.setParent(parent) self.setReadOnly(True) self.setLineWidth(50) self.setMinimumWidth(500) self.setFont(QFont('Consolas', 11)) @pyqtSlot(str) def append_text(self, text: str): self.moveCursor(QTextCursor.End) self.insertPlainText(text) class StdTQDMTextEdit(QLineEdit): def __init__(self, parent): super(StdTQDMTextEdit, self).__init__() self.setParent(parent) self.setReadOnly(True) self.setEnabled(True) self.setMinimumWidth(500) self.setAlignment(Qt.AlignLeft | Qt.AlignVCenter) self.setClearButtonEnabled(True) self.setFont(QFont('Consolas', 11)) @pyqtSlot(str) def set_tqdm_text(self, text: str): new_text = text if new_text.find('\r') >= 0: new_text = new_text.replace('\r', '').rstrip() if new_text: self.setText(new_text) else: # we suppose that all TQDM prints have \r # so drop the rest pass def long_procedure(): # emulate import of modules from tqdm.auto import tqdm setup_logging('long_procedure') __logger = logging.getLogger('long_procedure') __logger.setLevel(logging.DEBUG) tqdm_obect = tqdm(range(10), unit_scale=True, dynamic_ncols=True) tqdm_obect.set_description("My progress bar description") for i in tqdm_obect: time.sleep(.1) __logger.info('foo {}'.format(i)) class InitializationProcedures(QObject): def __init__(self, main_app: MainApp): super(InitializationProcedures, self).__init__() self._main_app = main_app @pyqtSlot() def run(self): long_procedure() @pyqtSlot() def finished(self): print("Thread finished !") # might call main window to do some stuff with buttons self._main_app.btn_perform_actions.setEnabled(True) def setup_logging(log_prefix): global __is_setup_done if __is_setup_done: pass else: __log_file_name = "{}-{}_log_file.txt".format(log_prefix, datetime.datetime.utcnow().isoformat().replace(":", "-")) __log_format = '%(asctime)s - %(name)-30s - %(levelname)s - %(message)s' __console_date_format = '%Y-%m-%d %H:%M:%S' __file_date_format = '%Y-%m-%d %H-%M-%S' root = logging.getLogger() root.setLevel(logging.DEBUG) console_formatter = logging.Formatter(__log_format, __console_date_format) file_formatter = logging.Formatter(__log_format, __file_date_format) file_handler = logging.FileHandler(__log_file_name, mode='a', delay=True) file_handler.setLevel(logging.DEBUG) file_handler.setFormatter(file_formatter) root.addHandler(file_handler) tqdm_handler = TqdmLoggingHandler() tqdm_handler.setLevel(logging.DEBUG) tqdm_handler.setFormatter(console_formatter) root.addHandler(tqdm_handler) __is_setup_done = True class TqdmLoggingHandler(logging.StreamHandler): def __init__(self): logging.StreamHandler.__init__(self) def emit(self, record): msg = self.format(record) tqdm.tqdm.write(msg) # from https://stackoverflow.com/questions/38543506/change-logging-print-function-to-tqdm-write-so-logging-doesnt-interfere-wit/38739634#38739634 self.flush() if __name__ == '__main__': app = QApplication(sys.argv) app.setStyle('Fusion') ex = MainApp() sys.exit(app.exec_())
相同的解决方案,但与实际分离的文
MyPyQtGUI.py
output_redirection_tools.py
config.py
my_logging.py
third_party_module_not_to_change.py
的 MyPyQtGUI.py 强>
值得注意的是,项目的第一次进口应该是 import output_redirection_tools 因为它完成了所有tqdm黑客工作。
import output_redirection_tools
# looks like an unused import, but it actually does the TQDM class trick to intercept prints import output_redirection_tools # KEEP ME !!! import logging import sys from PyQt5.QtCore import pyqtSlot, QObject, QThread, Qt from PyQt5.QtGui import QTextCursor, QFont from PyQt5.QtWidgets import QTextEdit, QWidget, QToolButton, QVBoxLayout, QApplication, QLineEdit from config import config_dict, STDOUT_WRITE_STREAM_CONFIG, TQDM_WRITE_STREAM_CONFIG, STREAM_CONFIG_KEY_QUEUE, \ STREAM_CONFIG_KEY_QT_QUEUE_RECEIVER from my_logging import setup_logging import third_party_module_not_to_change class MainApp(QWidget): def __init__(self): super().__init__() setup_logging(self.__class__.__name__) self.__logger = logging.getLogger(self.__class__.__name__) self.__logger.setLevel(logging.DEBUG) self.queue_std_out = config_dict[STDOUT_WRITE_STREAM_CONFIG][STREAM_CONFIG_KEY_QUEUE] self.queue_tqdm = config_dict[TQDM_WRITE_STREAM_CONFIG][STREAM_CONFIG_KEY_QUEUE] layout = QVBoxLayout() self.setMinimumWidth(500) self.btn_perform_actions = QToolButton(self) self.btn_perform_actions.setText('Launch long processing') self.btn_perform_actions.clicked.connect(self._btn_go_clicked) self.text_edit_std_out = StdOutTextEdit(self) self.text_edit_tqdm = StdTQDMTextEdit(self) self.thread_initialize = QThread() self.init_procedure_object = LongProcedureWrapper(self) # std out stream management # create console text read thread + receiver object self.thread_std_out_queue_listener = QThread() self.std_out_text_receiver = config_dict[STDOUT_WRITE_STREAM_CONFIG][STREAM_CONFIG_KEY_QT_QUEUE_RECEIVER] # connect receiver object to widget for text update self.std_out_text_receiver.queue_std_out_element_received_signal.connect(self.text_edit_std_out.append_text) # attach console text receiver to console text thread self.std_out_text_receiver.moveToThread(self.thread_std_out_queue_listener) # attach to start / stop methods self.thread_std_out_queue_listener.started.connect(self.std_out_text_receiver.run) self.thread_std_out_queue_listener.start() # NEW: TQDM stream management self.thread_tqdm_queue_listener = QThread() self.tqdm_text_receiver = config_dict[TQDM_WRITE_STREAM_CONFIG][STREAM_CONFIG_KEY_QT_QUEUE_RECEIVER] # connect receiver object to widget for text update self.tqdm_text_receiver.queue_tqdm_element_received_signal.connect(self.text_edit_tqdm.set_tqdm_text) # attach console text receiver to console text thread self.tqdm_text_receiver.moveToThread(self.thread_tqdm_queue_listener) # attach to start / stop methods self.thread_tqdm_queue_listener.started.connect(self.tqdm_text_receiver.run) self.thread_tqdm_queue_listener.start() layout.addWidget(self.btn_perform_actions) layout.addWidget(self.text_edit_std_out) layout.addWidget(self.text_edit_tqdm) self.setLayout(layout) self.show() @pyqtSlot() def _btn_go_clicked(self): # prepare thread for long operation self.init_procedure_object.moveToThread(self.thread_initialize) self.thread_initialize.started.connect(self.init_procedure_object.run) # start thread self.btn_perform_actions.setEnabled(False) self.thread_initialize.start() class StdOutTextEdit(QTextEdit): def __init__(self, parent): super(StdOutTextEdit, self).__init__() self.setParent(parent) self.setReadOnly(True) self.setLineWidth(50) self.setMinimumWidth(500) self.setFont(QFont('Consolas', 11)) @pyqtSlot(str) def append_text(self, text: str): self.moveCursor(QTextCursor.End) self.insertPlainText(text) class StdTQDMTextEdit(QLineEdit): def __init__(self, parent): super(StdTQDMTextEdit, self).__init__() self.setParent(parent) self.setReadOnly(True) self.setEnabled(True) self.setMinimumWidth(500) self.setAlignment(Qt.AlignLeft | Qt.AlignVCenter) self.setClearButtonEnabled(True) self.setFont(QFont('Consolas', 11)) @pyqtSlot(str) def set_tqdm_text(self, text: str): new_text = text if new_text.find('\r') >= 0: new_text = new_text.replace('\r', '').rstrip() if new_text: self.setText(new_text) else: # we suppose that all TQDM prints have \r, so drop the rest pass class LongProcedureWrapper(QObject): def __init__(self, main_app: MainApp): super(LongProcedureWrapper, self).__init__() self._main_app = main_app @pyqtSlot() def run(self): third_party_module_not_to_change.long_procedure() if __name__ == '__main__': app = QApplication(sys.argv) app.setStyle('Fusion') ex = MainApp() sys.exit(app.exec_())
的 my_logging.py 强>
import logging import datetime import tqdm from config import config_dict, IS_SETUP_DONE def setup_logging(log_prefix, force_debug_level=logging.DEBUG): root = logging.getLogger() root.setLevel(force_debug_level) if config_dict[IS_SETUP_DONE]: pass else: __log_file_name = "{}-{}_log_file.txt".format(log_prefix, datetime.datetime.utcnow().isoformat().replace(":", "-")) __log_format = '%(asctime)s - %(name)-30s - %(levelname)s - %(message)s' __console_date_format = '%Y-%m-%d %H:%M:%S' __file_date_format = '%Y-%m-%d %H-%M-%S' console_formatter = logging.Formatter(__log_format, __console_date_format) file_formatter = logging.Formatter(__log_format, __file_date_format) file_handler = logging.FileHandler(__log_file_name, mode='a', delay=True) file_handler.setLevel(logging.DEBUG) file_handler.setFormatter(file_formatter) root.addHandler(file_handler) tqdm_handler = TqdmLoggingHandler() tqdm_handler.setLevel(logging.DEBUG) tqdm_handler.setFormatter(console_formatter) root.addHandler(tqdm_handler) config_dict[IS_SETUP_DONE] = True class TqdmLoggingHandler(logging.StreamHandler): def __init__(self): logging.StreamHandler.__init__(self) def emit(self, record): msg = self.format(record) tqdm.tqdm.write(msg) self.flush()
的 output_redirection_tools.py 强>
import sys from queue import Queue from PyQt5.QtCore import pyqtSlot, pyqtSignal, QObject from config import config_dict, IS_STREAMS_REDIRECTION_SETUP_DONE, TQDM_WRITE_STREAM_CONFIG, STDOUT_WRITE_STREAM_CONFIG, \ STREAM_CONFIG_KEY_QUEUE, STREAM_CONFIG_KEY_STREAM, STREAM_CONFIG_KEY_QT_QUEUE_RECEIVER class QueueWriteStream(object): def __init__(self, q: Queue): self.queue = q def write(self, text): self.queue.put(text) def flush(self): pass def perform_tqdm_default_out_stream_hack(tqdm_file_stream, tqdm_nb_columns=None): import tqdm # save original class into module tqdm.orignal_class = tqdm.tqdm class TQDMPatch(tqdm.orignal_class): """ Derive from original class """ def __init__(self, iterable=None, desc=None, total=None, leave=True, file=None, ncols=None, mininterval=0.1, maxinterval=10.0, miniters=None, ascii=None, disable=False, unit='it', unit_scale=False, dynamic_ncols=False, smoothing=0.3, bar_format=None, initial=0, position=None, postfix=None, unit_divisor=1000, gui=False, **kwargs): super(TQDMPatch, self).__init__(iterable, desc, total, leave, tqdm_file_stream, # change any chosen file stream with our's tqdm_nb_columns, # change nb of columns (gui choice), mininterval, maxinterval, miniters, ascii, disable, unit, unit_scale, False, # change param smoothing, bar_format, initial, position, postfix, unit_divisor, gui, **kwargs) print('TQDM Patch called') # check it works @classmethod def write(cls, s, file=None, end="\n", nolock=False): super(TQDMPatch, cls).write(s=s, file=file, end=end, nolock=nolock) #tqdm.orignal_class.write(s=s, file=file, end=end, nolock=nolock) # all other tqdm.orignal_class @classmethod methods may need to be redefined ! # # I mainly used tqdm.auto in my modules, so use that for patch # # unsure if this will work with all possible tqdm import methods # # might not work for tqdm_gui ! import tqdm.auto as AUTO # # # change original class with the patched one, the original still exists AUTO.tqdm = TQDMPatch #tqdm.tqdm = TQDMPatch def setup_streams_redirection(tqdm_nb_columns=None): if config_dict[IS_STREAMS_REDIRECTION_SETUP_DONE]: pass else: configure_tqdm_redirection(tqdm_nb_columns) configure_std_out_redirection() config_dict[IS_STREAMS_REDIRECTION_SETUP_DONE] = True def configure_std_out_redirection(): queue_std_out = Queue() config_dict[STDOUT_WRITE_STREAM_CONFIG] = { STREAM_CONFIG_KEY_QUEUE: queue_std_out, STREAM_CONFIG_KEY_STREAM: QueueWriteStream(queue_std_out), STREAM_CONFIG_KEY_QT_QUEUE_RECEIVER: StdOutTextQueueReceiver(q=queue_std_out) } perform_std_out_hack() def perform_std_out_hack(): sys.stdout = config_dict[STDOUT_WRITE_STREAM_CONFIG][STREAM_CONFIG_KEY_STREAM] def configure_tqdm_redirection(tqdm_nb_columns=None): queue_tqdm = Queue() config_dict[TQDM_WRITE_STREAM_CONFIG] = { STREAM_CONFIG_KEY_QUEUE: queue_tqdm, STREAM_CONFIG_KEY_STREAM: QueueWriteStream(queue_tqdm), STREAM_CONFIG_KEY_QT_QUEUE_RECEIVER: TQDMTextQueueReceiver(q=queue_tqdm) } perform_tqdm_default_out_stream_hack( tqdm_file_stream=config_dict[TQDM_WRITE_STREAM_CONFIG][STREAM_CONFIG_KEY_STREAM], tqdm_nb_columns=tqdm_nb_columns) class StdOutTextQueueReceiver(QObject): # we are forced to define 1 signal per class # see https://stackoverflow.com/questions/50294652/how-to-create-pyqtsignals-dynamically queue_std_out_element_received_signal = pyqtSignal(str) def __init__(self, q: Queue, *args, **kwargs): QObject.__init__(self, *args, **kwargs) self.queue = q @pyqtSlot() def run(self): self.queue_std_out_element_received_signal.emit('---> STD OUT Queue reception Started <---\n') while True: text = self.queue.get() self.queue_std_out_element_received_signal.emit(text) class TQDMTextQueueReceiver(QObject): # we are forced to define 1 signal per class # see https://stackoverflow.com/questions/50294652/how-to-create-pyqtsignals-dynamically queue_tqdm_element_received_signal = pyqtSignal(str) def __init__(self, q: Queue, *args, **kwargs): QObject.__init__(self, *args, **kwargs) self.queue = q @pyqtSlot() def run(self): # we assume that all TQDM outputs start with \r, so use that to show stream reception is started self.queue_tqdm_element_received_signal.emit('\r---> TQDM Queue reception Started <---\n') while True: text = self.queue.get() self.queue_tqdm_element_received_signal.emit(text) setup_streams_redirection()
的 config.py 强>
IS_SETUP_DONE = 'is_setup_done' TQDM_WRITE_STREAM_CONFIG = 'TQDM_WRITE_STREAM_CONFIG' STDOUT_WRITE_STREAM_CONFIG = 'STDOUT_WRITE_STREAM_CONFIG' IS_STREAMS_REDIRECTION_SETUP_DONE = 'IS_STREAMS_REDIRECTION_SETUP_DONE' STREAM_CONFIG_KEY_QUEUE = 'queue' STREAM_CONFIG_KEY_STREAM = 'write_stream' STREAM_CONFIG_KEY_QT_QUEUE_RECEIVER = 'qt_queue_receiver' default_config_dict = { IS_SETUP_DONE: False, IS_STREAMS_REDIRECTION_SETUP_DONE: False, TQDM_WRITE_STREAM_CONFIG: None, STDOUT_WRITE_STREAM_CONFIG: None, } config_dict = default_config_dict
的 third_part_module_not_to_change.py 强>
表示我使用的代码类型,不希望/不能更改。
from tqdm.auto import tqdm import logging from my_logging import setup_logging import time def long_procedure(): setup_logging('long_procedure') __logger = logging.getLogger('long_procedure') __logger.setLevel(logging.DEBUG) tqdm_obect = tqdm(range(10), unit_scale=True, dynamic_ncols=True) tqdm_obect.set_description("My progress bar description") for i in tqdm_obect: time.sleep(.1) __logger.info('foo {}'.format(i))