如何正确地将标准输出、日志记录和 tqdm 重定向到 PyQt 小部件
How to correctly redirect stdout, logging and tqdm into a PyQt widget
首先，我知道很多问题都和这个类似。
但是在花了这么多时间之后，我现在寻求社区的帮助。
我开发并使用了一堆 python 依赖于 tqdm 的模块。
我希望它们可以在 Jupyter、控制台或 GUI 中使用。
在 Jupyter 或控制台中一切正常：logging/prints 和 tqdm 进度条之间没有冲突。这是显示 console/Jupyter 行为的示例代码：
# coding=utf-8
from tqdm.auto import tqdm
import time
import logging
import sys
import datetime
__is_setup_done = False
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 = TqdmLoggingHandler2(__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, level=logging.NOTSET):
logging.StreamHandler.__init__(self)
def emit(self, record):
msg = self.format(record)
tqdm.write(msg)
# from
self.flush()
def example_long_procedure():
setup_logging('long_procedure')
__logger = logging.getLogger('long_procedure')
__logger.setLevel(logging.DEBUG)
for i in tqdm(range(10), unit_scale=True, dynamic_ncols=True, file=sys.stdout):
time.sleep(.1)
__logger.info('foo {}'.format(i))
example_long_procedure()
得到的输出：
2019-03-07 22:22:27 - long_procedure - INFO - foo 0
2019-03-07 22:22:27 - long_procedure - INFO - foo 1
2019-03-07 22:22:27 - long_procedure - INFO - foo 2
2019-03-07 22:22:27 - long_procedure - INFO - foo 3
2019-03-07 22:22:27 - long_procedure - INFO - foo 4
2019-03-07 22:22:28 - long_procedure - INFO - foo 5
2019-03-07 22:22:28 - long_procedure - INFO - foo 6
2019-03-07 22:22:28 - long_procedure - INFO - foo 7
2019-03-07 22:22:28 - long_procedure - INFO - foo 8
2019-03-07 22:22:28 - long_procedure - INFO - foo 9
100%|¦¦¦¦¦¦¦¦¦¦¦¦¦¦¦¦¦¦¦¦¦¦¦¦¦¦¦¦¦¦¦| 10.0/10.0 [00:01<00:00, 9.69it/s]
现在，我正在用 PyQt 制作一个 GUI，它使用与上面类似的代码。由于处理时间可能很长，我使用了线程以避免在处理过程中冻结 HMI。我还使用 stdout 使用 Queue() 重定向到 Qt QWidget，以便用户可以看到发生了什么。
我当前的用例是 1 个单线程，它具有日志和 tqdm 进度条以重定向到 1 个专用小部件。 （我不是在寻找多个线程来为小部件提供多个日志和多个 tqdm 进度条）。
感谢 Redirecting stdout and stderr to a PyQt5 QTextEdit from a secondary thread 提供的信息，我成功地重定向了标准输出。
但是，只有记录器行被重定向。 TQDM 进度条仍然定向到控制台输出。
这是我当前的代码：
# coding=utf-8
import time
import logging
import sys
import datetime
__is_setup_done = False
from queue import Queue
from PyQt5.QtCore import pyqtSlot, pyqtSignal, QObject, QThread, QMetaObject, Q_ARG, Qt
from PyQt5.QtGui import QTextCursor, QFont
from PyQt5.QtWidgets import QTextEdit, QPlainTextEdit, QWidget, QToolButton, QVBoxLayout, QApplication
from tqdm.auto import tqdm
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 console text queue
self.queue_console_text = Queue()
# redirect stdout to the queue
output_stream = WriteStream(self.queue_console_text)
sys.stdout = output_stream
layout = QVBoxLayout()
self.setMinimumWidth(500)
# GO button
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.console_text_edit = ConsoleTextEdit(self)
self.thread_initialize = QThread()
self.init_procedure_object = InitializationProcedures(self)
# create console text read thread + receiver object
self.thread_queue_listener = QThread()
self.console_text_receiver = ThreadConsoleTextQueueReceiver(self.queue_console_text)
# connect receiver object to widget for text update
self.console_text_receiver.queue_element_received_signal.connect(self.console_text_edit.append_text)
# attach console text receiver to console text thread
self.console_text_receiver.moveToThread(self.thread_queue_listener)
# attach to start / stop methods
self.thread_queue_listener.started.connect(self.console_text_receiver.run)
self.thread_queue_listener.finished.connect(self.console_text_receiver.finished)
self.thread_queue_listener.start()
layout.addWidget(self.btn_perform_actions)
layout.addWidget(self.console_text_edit)
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 WriteStream(object):
def __init__(self, q: Queue):
self.queue = q
def write(self, text):
"""
Redirection of stream to the given queue
"""
self.queue.put(text)
def flush(self):
"""
Stream flush implementation
"""
pass
class ThreadConsoleTextQueueReceiver(QObject):
queue_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_element_received_signal.emit('---> Console text queue reception Started <---\n')
while True:
text = self.queue.get()
self.queue_element_received_signal.emit(text)
@pyqtSlot()
def finished(self):
self.queue_element_received_signal.emit('---> Console text queue reception Stopped <---\n')
class ConsoleTextEdit(QTextEdit):#QTextEdit):
def __init__(self, parent):
super(ConsoleTextEdit, self).__init__()
self.setParent(parent)
self.setReadOnly(True)
self.setLineWidth(50)
self.setMinimumWidth(1200)
self.setFont(QFont('Consolas', 11))
self.flag = False
@pyqtSlot(str)
def append_text(self, text: str):
self.moveCursor(QTextCursor.End)
self.insertPlainText(text)
def long_procedure():
setup_logging('long_procedure')
__logger = logging.getLogger('long_procedure')
__logger.setLevel(logging.DEBUG)
for i in tqdm(range(10), unit_scale=True, dynamic_ncols=True):
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, level=logging.NOTSET):
logging.StreamHandler.__init__(self)
def emit(self, record):
msg = self.format(record)
tqdm.write(msg)
# from
self.flush()
if __name__ == '__main__':
app = QApplication(sys.argv)
app.setStyle('Fusion')
tqdm.ncols = 50
ex = MainApp()
sys.exit(app.exec_())
给出：
我想获得在控制台中严格调用代码时的确切行为。
即 PyQt 小部件中的预期输出：
---> Console text queue reception Started <---
2019-03-07 19:42:19 - long_procedure - INFO - foo 0
2019-03-07 19:42:19 - long_procedure - INFO - foo 1
2019-03-07 19:42:19 - long_procedure - INFO - foo 2
2019-03-07 19:42:19 - long_procedure - INFO - foo 3
2019-03-07 19:42:19 - long_procedure - INFO - foo 4
2019-03-07 19:42:19 - long_procedure - INFO - foo 5
2019-03-07 19:42:20 - long_procedure - INFO - foo 6
2019-03-07 19:42:20 - long_procedure - INFO - foo 7
2019-03-07 19:42:20 - long_procedure - INFO - foo 8
2019-03-07 19:42:20 - long_procedure - INFO - foo 9
100%|################################| 10.0/10.0 [00:01<00:00, 9.16it/s]
我尝试/探索的事情没有成功。
选项 1
此解决方案 没有给出预期的结果。重定向仅包含 tqdm 内容的输出效果很好。
无论是 QTextEdit 还是 QPlainTextEdit，下面的代码都没有给出预期的行为。仅重定向记录器行。
# code from this answer
#
@pyqtSlot(str)
def append_text(self, message: str):
if not hasattr(self, "flag"):
self.flag = False
message = message.replace('\r', '').rstrip()
if message:
method = "replace_last_line" if self.flag else "append_text"
QMetaObject.invokeMethod(self,
method,
Qt.QueuedConnection,
Q_ARG(str, message))
self.flag = True
else:
self.flag = False
@pyqtSlot(str)
def replace_last_line(self, text):
cursor = self.textCursor()
cursor.movePosition(QTextCursor.End)
cursor.select(QTextCursor.BlockUnderCursor)
cursor.removeSelectedText()
cursor.insertBlock()
self.setTextCursor(cursor)
self.insertPlainText(text)
但是，上面的代码 + 添加 file=sys.stdout 到 tqdm 调用改变了行为：tqdm 输出被重定向到 Qt 小部件。但是最后只显示了一行，要么是logger行，要么是tqdm行（看样子要看我导出的是哪个Qt widget了）
最后，改变所有使用模块的 tqdm 调用不应该是首选。
所以我发现的另一种方法是在同一个 stream/queue stdout 中重定向 stderr。由于 tqdm 默认写入 stderr，因此所有 tqdm 输出都被重定向到 widget。
但我仍然无法弄清楚获得我正在寻找的确切输出。
这个问题没有提供关于为什么 QTextEdit vs QPlainTextEdit
之间的行为似乎不同的线索
选项 2
这个问题Duplicate stdout, stderr in QTextEdit widget看起来与非常相似并且没有回答我上面描述的确切问题。
选项 3
尝试 使用 contextlib 给了我一个错误，因为没有定义 flush() 方法。修复后，我最终只有 tqdm 行而没有记录器行。
选项 4
我也试过截取 \r 字符并实现特定行为，但没有成功。
版本：
tqdm 4.28.1
pyqt 5.9.2
PyQt5 5.12
PyQt5_sip 4.19.14
Python 3.7.2
编辑 2019 年 3 月 12 日：在我看来，答案是：它可能会完成，但需要付出很多努力才能记住 QTextEdit 的哪一行来自哪里按预期行事。另外，由于默认情况下 tdm 写入 stderr，您最终也会捕获所有异常跟踪。
这就是为什么我将自己的答案标记为已解决：我发现实现相同目的更优雅：在 pyqt 中显示正在发生的事情。
这是我获得接近预期行为的最佳机会。
它并没有完全回答这个问题，因为我改变了 GUI 设计。
所以我不会将其投票为已解决。此外，这一切都在一个 python 文件中完成。我计划进一步挑战这个解决方案，看看它是否适用于真正的 python 模块进行 tqdm 导入。
我以非常丑陋的方式修补了基本的 tqdm class。主要技巧是：
- 通过将原始 tqdm class 存储为新名称来动态更改 tqdm 模块结构：
tqdm.orignal_class = tqdm.tqdm
- 然后继承tqdm.originalclass
class TQDMPatch(tqdm.orignal_class):
- 实现构造函数以将文件流 + 任何参数强制为您想要的任何参数：
super(TQDMPatch, self).__init__(... change some params ...)。我给了我的 TQDM class 一个自定义 WriteStream() 写入 Queue()
- 更改 GUI 策略以拦截自定义 tqdm 流并将其重定向到单独的 Qt 小部件。我的小部件假设所有收到的打印件都包含
\r（TQDM 似乎在做）。
它既可以在单个 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
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 hack 工作。
# 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
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
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))
首先，我知道很多问题都和这个类似。 但是在花了这么多时间之后，我现在寻求社区的帮助。
我开发并使用了一堆 python 依赖于 tqdm 的模块。
我希望它们可以在 Jupyter、控制台或 GUI 中使用。
在 Jupyter 或控制台中一切正常：logging/prints 和 tqdm 进度条之间没有冲突。这是显示 console/Jupyter 行为的示例代码：
# coding=utf-8
from tqdm.auto import tqdm
import time
import logging
import sys
import datetime
__is_setup_done = False
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 = TqdmLoggingHandler2(__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, level=logging.NOTSET):
logging.StreamHandler.__init__(self)
def emit(self, record):
msg = self.format(record)
tqdm.write(msg)
# from
self.flush()
def example_long_procedure():
setup_logging('long_procedure')
__logger = logging.getLogger('long_procedure')
__logger.setLevel(logging.DEBUG)
for i in tqdm(range(10), unit_scale=True, dynamic_ncols=True, file=sys.stdout):
time.sleep(.1)
__logger.info('foo {}'.format(i))
example_long_procedure()
得到的输出：
2019-03-07 22:22:27 - long_procedure - INFO - foo 0
2019-03-07 22:22:27 - long_procedure - INFO - foo 1
2019-03-07 22:22:27 - long_procedure - INFO - foo 2
2019-03-07 22:22:27 - long_procedure - INFO - foo 3
2019-03-07 22:22:27 - long_procedure - INFO - foo 4
2019-03-07 22:22:28 - long_procedure - INFO - foo 5
2019-03-07 22:22:28 - long_procedure - INFO - foo 6
2019-03-07 22:22:28 - long_procedure - INFO - foo 7
2019-03-07 22:22:28 - long_procedure - INFO - foo 8
2019-03-07 22:22:28 - long_procedure - INFO - foo 9
100%|¦¦¦¦¦¦¦¦¦¦¦¦¦¦¦¦¦¦¦¦¦¦¦¦¦¦¦¦¦¦¦| 10.0/10.0 [00:01<00:00, 9.69it/s]
现在，我正在用 PyQt 制作一个 GUI，它使用与上面类似的代码。由于处理时间可能很长，我使用了线程以避免在处理过程中冻结 HMI。我还使用 stdout 使用 Queue() 重定向到 Qt QWidget，以便用户可以看到发生了什么。
我当前的用例是 1 个单线程，它具有日志和 tqdm 进度条以重定向到 1 个专用小部件。 （我不是在寻找多个线程来为小部件提供多个日志和多个 tqdm 进度条）。
感谢 Redirecting stdout and stderr to a PyQt5 QTextEdit from a secondary thread 提供的信息，我成功地重定向了标准输出。 但是，只有记录器行被重定向。 TQDM 进度条仍然定向到控制台输出。
这是我当前的代码：
# coding=utf-8
import time
import logging
import sys
import datetime
__is_setup_done = False
from queue import Queue
from PyQt5.QtCore import pyqtSlot, pyqtSignal, QObject, QThread, QMetaObject, Q_ARG, Qt
from PyQt5.QtGui import QTextCursor, QFont
from PyQt5.QtWidgets import QTextEdit, QPlainTextEdit, QWidget, QToolButton, QVBoxLayout, QApplication
from tqdm.auto import tqdm
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 console text queue
self.queue_console_text = Queue()
# redirect stdout to the queue
output_stream = WriteStream(self.queue_console_text)
sys.stdout = output_stream
layout = QVBoxLayout()
self.setMinimumWidth(500)
# GO button
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.console_text_edit = ConsoleTextEdit(self)
self.thread_initialize = QThread()
self.init_procedure_object = InitializationProcedures(self)
# create console text read thread + receiver object
self.thread_queue_listener = QThread()
self.console_text_receiver = ThreadConsoleTextQueueReceiver(self.queue_console_text)
# connect receiver object to widget for text update
self.console_text_receiver.queue_element_received_signal.connect(self.console_text_edit.append_text)
# attach console text receiver to console text thread
self.console_text_receiver.moveToThread(self.thread_queue_listener)
# attach to start / stop methods
self.thread_queue_listener.started.connect(self.console_text_receiver.run)
self.thread_queue_listener.finished.connect(self.console_text_receiver.finished)
self.thread_queue_listener.start()
layout.addWidget(self.btn_perform_actions)
layout.addWidget(self.console_text_edit)
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 WriteStream(object):
def __init__(self, q: Queue):
self.queue = q
def write(self, text):
"""
Redirection of stream to the given queue
"""
self.queue.put(text)
def flush(self):
"""
Stream flush implementation
"""
pass
class ThreadConsoleTextQueueReceiver(QObject):
queue_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_element_received_signal.emit('---> Console text queue reception Started <---\n')
while True:
text = self.queue.get()
self.queue_element_received_signal.emit(text)
@pyqtSlot()
def finished(self):
self.queue_element_received_signal.emit('---> Console text queue reception Stopped <---\n')
class ConsoleTextEdit(QTextEdit):#QTextEdit):
def __init__(self, parent):
super(ConsoleTextEdit, self).__init__()
self.setParent(parent)
self.setReadOnly(True)
self.setLineWidth(50)
self.setMinimumWidth(1200)
self.setFont(QFont('Consolas', 11))
self.flag = False
@pyqtSlot(str)
def append_text(self, text: str):
self.moveCursor(QTextCursor.End)
self.insertPlainText(text)
def long_procedure():
setup_logging('long_procedure')
__logger = logging.getLogger('long_procedure')
__logger.setLevel(logging.DEBUG)
for i in tqdm(range(10), unit_scale=True, dynamic_ncols=True):
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, level=logging.NOTSET):
logging.StreamHandler.__init__(self)
def emit(self, record):
msg = self.format(record)
tqdm.write(msg)
# from
self.flush()
if __name__ == '__main__':
app = QApplication(sys.argv)
app.setStyle('Fusion')
tqdm.ncols = 50
ex = MainApp()
sys.exit(app.exec_())
给出：
我想获得在控制台中严格调用代码时的确切行为。 即 PyQt 小部件中的预期输出：
---> Console text queue reception Started <---
2019-03-07 19:42:19 - long_procedure - INFO - foo 0
2019-03-07 19:42:19 - long_procedure - INFO - foo 1
2019-03-07 19:42:19 - long_procedure - INFO - foo 2
2019-03-07 19:42:19 - long_procedure - INFO - foo 3
2019-03-07 19:42:19 - long_procedure - INFO - foo 4
2019-03-07 19:42:19 - long_procedure - INFO - foo 5
2019-03-07 19:42:20 - long_procedure - INFO - foo 6
2019-03-07 19:42:20 - long_procedure - INFO - foo 7
2019-03-07 19:42:20 - long_procedure - INFO - foo 8
2019-03-07 19:42:20 - long_procedure - INFO - foo 9
100%|################################| 10.0/10.0 [00:01<00:00, 9.16it/s]
我尝试/探索的事情没有成功。
选项 1
此解决方案
无论是 QTextEdit 还是 QPlainTextEdit，下面的代码都没有给出预期的行为。仅重定向记录器行。
# code from this answer
#
@pyqtSlot(str)
def append_text(self, message: str):
if not hasattr(self, "flag"):
self.flag = False
message = message.replace('\r', '').rstrip()
if message:
method = "replace_last_line" if self.flag else "append_text"
QMetaObject.invokeMethod(self,
method,
Qt.QueuedConnection,
Q_ARG(str, message))
self.flag = True
else:
self.flag = False
@pyqtSlot(str)
def replace_last_line(self, text):
cursor = self.textCursor()
cursor.movePosition(QTextCursor.End)
cursor.select(QTextCursor.BlockUnderCursor)
cursor.removeSelectedText()
cursor.insertBlock()
self.setTextCursor(cursor)
self.insertPlainText(text)
但是，上面的代码 + 添加 file=sys.stdout 到 tqdm 调用改变了行为：tqdm 输出被重定向到 Qt 小部件。但是最后只显示了一行，要么是logger行，要么是tqdm行（看样子要看我导出的是哪个Qt widget了）
最后，改变所有使用模块的 tqdm 调用不应该是首选。
所以我发现的另一种方法是在同一个 stream/queue stdout 中重定向 stderr。由于 tqdm 默认写入 stderr，因此所有 tqdm 输出都被重定向到 widget。
但我仍然无法弄清楚获得我正在寻找的确切输出。
这个问题没有提供关于为什么 QTextEdit vs QPlainTextEdit
之间的行为似乎不同的线索选项 2
这个问题Duplicate stdout, stderr in QTextEdit widget看起来与
选项 3
尝试
选项 4
我也试过截取 \r 字符并实现特定行为，但没有成功。
版本：
tqdm 4.28.1
pyqt 5.9.2
PyQt5 5.12
PyQt5_sip 4.19.14
Python 3.7.2
编辑 2019 年 3 月 12 日：在我看来，答案是：它可能会完成，但需要付出很多努力才能记住 QTextEdit 的哪一行来自哪里按预期行事。另外，由于默认情况下 tdm 写入 stderr，您最终也会捕获所有异常跟踪。 这就是为什么我将自己的答案标记为已解决：我发现实现相同目的更优雅：在 pyqt 中显示正在发生的事情。
这是我获得接近预期行为的最佳机会。 它并没有完全回答这个问题，因为我改变了 GUI 设计。 所以我不会将其投票为已解决。此外，这一切都在一个 python 文件中完成。我计划进一步挑战这个解决方案，看看它是否适用于真正的 python 模块进行 tqdm 导入。
我以非常丑陋的方式修补了基本的 tqdm class。主要技巧是：
- 通过将原始 tqdm class 存储为新名称来动态更改 tqdm 模块结构：
tqdm.orignal_class = tqdm.tqdm - 然后继承tqdm.originalclass
class TQDMPatch(tqdm.orignal_class): - 实现构造函数以将文件流 + 任何参数强制为您想要的任何参数：
super(TQDMPatch, self).__init__(... change some params ...)。我给了我的 TQDM class 一个自定义WriteStream()写入Queue() - 更改 GUI 策略以拦截自定义 tqdm 流并将其重定向到单独的 Qt 小部件。我的小部件假设所有收到的打印件都包含
\r（TQDM 似乎在做）。
它既可以在单个 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
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 hack 工作。
# 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
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
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))