如何正确地将标准输出、日志记录和 tqdm 重定向到 PyQt 小部件中

Question

如何正确地将标准输出、日志记录和 tqdm 重定向到 PyQt 小部件中

Lon*_*rer 10 python logging stdout pyqt tqdm

长话短说

\n

答案请参见：

\n

我的 2019 年最初使用文本编辑和 stdout/stderr 流重定向接受了答案，请参阅/sf/answers/3855776501/
我的第二个答案，现在标记为已接受的答案：使用真正的 QProgressBar 的派生和改进方法！/sf/answers/5186428061/

\n

问题

\n

首先，我知道很多问题都与这个问题类似。\n但是在花了这么多时间之后，我现在向社区寻求帮助。

\n

我开发并使用了一堆依赖于的 python 模块tqdm。\n我希望它们可以在 Jupyter、控制台或 GUI 中使用。\n在 Jupyter 或控制台中一切正常：日志记录/打印和 tqdm 进度之间没有冲突酒吧。以下是显示控制台/Jupyter 行为的示例代码：

\n

# coding=utf-8\nfrom tqdm.auto import tqdm\nimport time\nimport logging\nimport sys\nimport datetime\n__is_setup_done = False\n\n\ndef setup_logging(log_prefix):\n    global __is_setup_done\n\n    if __is_setup_done:\n        pass\n    else:\n        __log_file_name = "{}-{}_log_file.txt".format(log_prefix,\n                                                      datetime.datetime.utcnow().isoformat().replace(":", "-"))\n\n        __log_format = \'%(asctime)s - %(name)-30s - %(levelname)s - %(message)s\'\n        __console_date_format = \'%Y-%m-%d %H:%M:%S\'\n        __file_date_format = \'%Y-%m-%d %H-%M-%S\'\n\n        root = logging.getLogger()\n        root.setLevel(logging.DEBUG)\n\n        console_formatter = logging.Formatter(__log_format, __console_date_format)\n\n        file_formatter = logging.Formatter(__log_format, __file_date_format)\n        file_handler = logging.FileHandler(__log_file_name, mode=\'a\', delay=True)\n        # file_handler = TqdmLoggingHandler2(__log_file_name, mode=\'a\', delay=True)\n        file_handler.setLevel(logging.DEBUG)\n        file_handler.setFormatter(file_formatter)\n        root.addHandler(file_handler)\n\n        tqdm_handler = TqdmLoggingHandler()\n        tqdm_handler.setLevel(logging.DEBUG)\n        tqdm_handler.setFormatter(console_formatter)\n        root.addHandler(tqdm_handler)\n\n        __is_setup_done = True\n\nclass TqdmLoggingHandler(logging.StreamHandler):\n\n    def __init__(self, level=logging.NOTSET):\n        logging.StreamHandler.__init__(self)\n\n    def emit(self, record):\n        msg = self.format(record)\n        tqdm.write(msg)\n        # from /sf/ask/2698045451/#38739634\n        self.flush()\n\n\ndef example_long_procedure():\n    setup_logging(\'long_procedure\')\n    __logger = logging.getLogger(\'long_procedure\')\n    __logger.setLevel(logging.DEBUG)\n    for i in tqdm(range(10), unit_scale=True, dynamic_ncols=True, file=sys.stdout):\n        time.sleep(.1)\n        __logger.info(\'foo {}\'.format(i))\n\nexample_long_procedure()\n

Run Code Online (Sandbox Code Playgroud)\n

得到的输出：

\n

2019-03-07 22:22:27 - long_procedure                 - INFO - foo 0\n2019-03-07 22:22:27 - long_procedure                 - INFO - foo 1\n2019-03-07 22:22:27 - long_procedure                 - INFO - foo 2\n2019-03-07 22:22:27 - long_procedure                 - INFO - foo 3\n2019-03-07 22:22:27 - long_procedure                 - INFO - foo 4\n2019-03-07 22:22:28 - long_procedure                 - INFO - foo 5\n2019-03-07 22:22:28 - long_procedure                 - INFO - foo 6\n2019-03-07 22:22:28 - long_procedure                 - INFO - foo 7\n2019-03-07 22:22:28 - long_procedure                 - INFO - foo 8\n2019-03-07 22:22:28 - long_procedure                 - INFO - foo 9\n100%|\xc2\xa6\xc2\xa6\xc2\xa6\xc2\xa6\xc2\xa6\xc2\xa6\xc2\xa6\xc2\xa6\xc2\xa6\xc2\xa6\xc2\xa6\xc2\xa6\xc2\xa6\xc2\xa6\xc2\xa6\xc2\xa6\xc2\xa6\xc2\xa6\xc2\xa6\xc2\xa6\xc2\xa6\xc2\xa6\xc2\xa6\xc2\xa6\xc2\xa6\xc2\xa6\xc2\xa6\xc2\xa6\xc2\xa6\xc2\xa6\xc2\xa6| 10.0/10.0 [00:01<00:00, 9.69it/s]\n

Run Code Online (Sandbox Code Playgroud)\n

现在，我正在使用 PyQt 制作一个 GUI，它使用与上面类似的代码。由于处理可能很长，我使用线程以避免处理期间冻结 HMI。我还使用stdoutQueue() 重定向到 Qt QWidget，以便用户可以看到发生了什么。

\n

我当前的用例是 1 个单线程，它具有日志和 tqdm 进度条以重定向到 1 个专用小部件。（我不是在寻找多个线程来为小部件提供多个日志和多个 tqdm 进度条）。

\n

由于来自Redirecting stdout and stderr to a PyQt5 QTextEdit from a secondary thread 的信息，我成功地重定向了 stdout。\n但是，只有记录器行被重定向。TQDM 进度条仍然指向控制台输出。

\n

这是我当前的代码：

\n

# coding=utf-8\n\nimport time\nimport logging\nimport sys\nimport datetime\n__is_setup_done = False\n\nfrom queue import Queue\n\nfrom PyQt5.QtCore import pyqtSlot, pyqtSignal, QObject, QThread, QMetaObject, Q_ARG, Qt\nfrom PyQt5.QtGui import QTextCursor, QFont\nfrom PyQt5.QtWidgets import QTextEdit, QPlainTextEdit, QWidget, QToolButton, QVBoxLayout, QApplication\nfrom tqdm.auto import tqdm\n\n\nclass MainApp(QWidget):\n    def __init__(self):\n        super().__init__()\n\n        setup_logging(self.__class__.__name__)\n\n\n        self.__logger = logging.getLogger(self.__class__.__name__)\n        self.__logger.setLevel(logging.DEBUG)\n\n        # create console text queue\n        self.queue_console_text = Queue()\n        # redirect stdout to the queue\n        output_stream = WriteStream(self.queue_console_text)\n        sys.stdout = output_stream\n\n        layout = QVBoxLayout()\n\n        self.setMinimumWidth(500)\n\n        # GO button\n        self.btn_perform_actions = QToolButton(self)\n        self.btn_perform_actions.setText(\'Launch long processing\')\n        self.btn_perform_actions.clicked.connect(self._btn_go_clicked)\n\n        self.console_text_edit = ConsoleTextEdit(self)\n\n        self.thread_initialize = QThread()\n        self.init_procedure_object = InitializationProcedures(self)\n\n        # create console text read thread + receiver object\n        self.thread_queue_listener = QThread()\n        self.console_text_receiver = ThreadConsoleTextQueueReceiver(self.queue_console_text)\n        # connect receiver object to widget for text update\n        self.console_text_receiver.queue_element_received_signal.connect(self.console_text_edit.append_text)\n        # attach console text receiver to console text thread\n        self.console_text_receiver.moveToThread(self.thread_queue_listener)\n        # attach to start / stop methods\n        self.thread_queue_listener.started.connect(self.console_text_receiver.run)\n        self.thread_queue_listener.finished.connect(self.console_text_receiver.finished)\n        self.thread_queue_listener.start()\n\n        layout.addWidget(self.btn_perform_actions)\n        layout.addWidget(self.console_text_edit)\n        self.setLayout(layout)\n        self.show()\n\n    @pyqtSlot()\n    def _btn_go_clicked(self):\n        # prepare thread for long operation\n        self.init_procedure_object.moveToThread(self.thread_initialize)\n        self.thread_initialize.started.connect(self.init_procedure_object.run)\n        self.thread_initialize.finished.connect(self.init_procedure_object.finished)\n        # start thread\n        self.btn_perform_actions.setEnabled(False)\n        self.thread_initialize.start()\n\n\nclass WriteStream(object):\n    def __init__(self, q: Queue):\n        self.queue = q\n\n    def write(self, text):\n        """\n        Redirection of stream to the given queue\n        """\n        self.queue.put(text)\n\n    def flush(self):\n        """\n        Stream flush implementation\n        """\n        pass\n\n\nclass ThreadConsoleTextQueueReceiver(QObject):\n    queue_element_received_signal = pyqtSignal(str)\n\n    def __init__(self, q: Queue, *args, **kwargs):\n        QObject.__init__(self, *args, **kwargs)\n        self.queue = q\n\n    @pyqtSlot()\n    def run(self):\n        self.queue_element_received_signal.emit(\'---> Console text queue reception Started <---\\n\')\n        while True:\n            text = self.queue.get()\n            self.queue_element_received_signal.emit(text)\n\n    @pyqtSlot()\n    def finished(self):\n        self.queue_element_received_signal.emit(\'---> Console text queue reception Stopped <---\\n\')\n\n\nclass ConsoleTextEdit(QTextEdit):#QTextEdit):\n    def __init__(self, parent):\n        super(ConsoleTextEdit, self).__init__()\n        self.setParent(parent)\n        self.setReadOnly(True)\n        self.setLineWidth(50)\n        self.setMinimumWidth(1200)\n        self.setFont(QFont(\'Consolas\', 11))\n        self.flag = False\n\n    @pyqtSlot(str)\n    def append_text(self, text: str):\n        self.moveCursor(QTextCursor.End)\n        self.insertPlainText(text)\n\ndef long_procedure():\n    setup_logging(\'long_procedure\')\n    __logger = logging.getLogger(\'long_procedure\')\n    __logger.setLevel(logging.DEBUG)\n    for i in tqdm(range(10), unit_scale=True, dynamic_ncols=True):\n        time.sleep(.1)\n        __logger.info(\'foo {}\'.format(i))\n\n\nclass InitializationProcedures(QObject):\n    def __init__(self, main_app: MainApp):\n        super(InitializationProcedures, self).__init__()\n        self._main_app = main_app\n\n    @pyqtSlot()\n    def run(self):\n        long_procedure()\n\n    @pyqtSlot()\n    def finished(self):\n        print("Thread finished !")  # might call main window to do some stuff with buttons\n        self._main_app.btn_perform_actions.setEnabled(True)\n\ndef setup_logging(log_prefix):\n    global __is_setup_done\n\n    if __is_setup_done:\n        pass\n    else:\n        __log_file_name = "{}-{}_log_file.txt".format(log_prefix,\n                                                      datetime.datetime.utcnow().isoformat().replace(":", "-"))\n\n        __log_format = \'%(asctime)s - %(name)-30s - %(levelname)s - %(message)s\'\n        __console_date_format = \'%Y-%m-%d %H:%M:%S\'\n        __file_date_format = \'%Y-%m-%d %H-%M-%S\'\n\n        root = logging.getLogger()\n        root.setLevel(logging.DEBUG)\n\n        console_formatter = logging.Formatter(__log_format, __console_date_format)\n\n        file_formatter = logging.Formatter(__log_format, __file_date_format)\n        file_handler = logging.FileHandler(__log_file_name, mode=\'a\', delay=True)\n        \n        file_handler.setLevel(logging.DEBUG)\n        file_handler.setFormatter(file_formatter)\n        root.addHandler(file_handler)\n\n        tqdm_handler = TqdmLoggingHandler()\n        tqdm_handler.setLevel(logging.DEBUG)\n        tqdm_handler.setFormatter(console_formatter)\n        root.addHandler(tqdm_handler)\n\n        __is_setup_done = True\n\nclass TqdmLoggingHandler(logging.StreamHandler):\n\n    def __init__(self, level=logging.NOTSET):\n        logging.StreamHandler.__init__(self)\n\n    def emit(self, record):\n        msg = self.format(record)\n        tqdm.write(msg)\n        # from /sf/ask/2698045451/#38739634\n        self.flush()\n\nif __name__ == \'__main__\':\n\n    app = QApplication(sys.argv)\n    app.setStyle(\'Fusion\')\n    tqdm.ncols = 50\n    ex = MainApp()\n    sys.exit(app.exec_())\n

Run Code Online (Sandbox Code Playgroud)\n

给出：

\n

我想获得我在控制台中严格调用代码的确切行为。\ni.e. PyQt 小部件中的预期输出：

\n

---> Console text queue reception Started <---\n2019-03-07 19:42:19 - long_procedure                 - INFO - foo 0\n2019-03-07 19:42:19 - long_procedure                 - INFO - foo 1\n2019-03-07 19:42:19 - long_procedure                 - INFO - foo 2\n2019-03-07 19:42:19 - long_procedure                 - INFO - foo 3\n2019-03-07 19:42:19 - long_procedure                 - INFO - foo 4\n2019-03-07 19:42:19 - long_procedure                 - INFO - foo 5\n2019-03-07 19:42:20 - long_procedure                 - INFO - foo 6\n2019-03-07 19:42:20 - long_procedure                 - INFO - foo 7\n2019-03-07 19:42:20 - long_procedure                 - INFO - foo 8\n2019-03-07 19:42:20 - long_procedure                 - INFO - foo 9\n\n100%|################################| 10.0/10.0 [00:01<00:00, 9.16it/s]\n

Run Code Online (Sandbox Code Playgroud)\n

\n

我尝试/探索但没有成功的事情。

\n

选项1

\n

此解决方案在 QPlainTextEdit 中使用 tqdm 显示终端输出未给出预期结果。它可以很好地重定向仅包含 tqdm 内容的输出。

\n

无论是使用 QTextEdit 还是 QPlainTextEdit，以下代码都没有给出预期的行为。仅记录器行被重定向。

\n

    # code from this answer\n    # /sf/ask/3736738281/\n    @pyqtSlot(str)\n    def append_text(self, message: str):\n        if not hasattr(self, "flag"):\n            self.flag = False\n        message = message.replace(\'\\r\', \'\').rstrip()\n        if message:\n            method = "replace_last_line" if self.flag else "append_text"\n            QMetaObject.invokeMethod(self,\n                                     method,\n                                     Qt.QueuedConnection,\n                                     Q_ARG(str, message))\n            self.flag = True\n        else:\n            self.flag = False\n\n    @pyqtSlot(str)\n    def replace_last_line(self, text):\n        cursor = self.textCursor()\n        cursor.movePosition(QTextCursor.End)\n        cursor.select(QTextCursor.BlockUnderCursor)\n        cursor.removeSelectedText()\n        cursor.insertBlock()\n        self.setTextCursor(cursor)\n        self.insertPlainText(text)\n

Run Code Online (Sandbox Code Playgroud)\n

然而，上面的代码+添加file=sys.stdout到 tqdm 调用会改变行为：tqdm 输出被重定向到 Qt 小部件。但最终只显示一行，它要么是记录器行，要么是 tqdm 行（看起来这取决于我派生的 Qt 小部件）。

\n

最后，更改所有使用模块的 tqdm 调用不应成为首选选项。

\n

所以我发现的另一种方法是在 stdout 重定向到的同一个流/队列中重定向 stderr。由于 tqdm 默认写入 stderr，因此所有 tqdm 输出都将重定向到小部件。

\n

但我仍然无法\xe2\x80\x99t 弄清楚获得我\xe2\x80\x99m 寻找的确切输出。

\n

这个问题没有提供为什么QTextEdit 与 QPlainTextEdit之间的行为似乎不同的线索

\n

选项2

\n

这个问题Duplicate stdout, stderr in QTextEdit widget看起来非常类似于在 QPlainTextEdit 中使用 tqdm 显示终端输出，并且不能回答我上面描述的确切问题。

\n

选项3

\n

由于没有定义flush()方法，使用contextlib尝试这个解决方案给了我一个错误。修复后，我最终只有 tqdm 行，没有记录器行。

\n

选项4

\n

我还尝试拦截 \\r 字符并实现特定行为，但没有成功。

\n

版本：

\n

tqdm                      4.28.1\npyqt                      5.9.2\nPyQt5                     5.12\nPyQt5_sip                 4.19.14\nPython                    3.7.2\n

Run Code Online (Sandbox Code Playgroud)\n

Answer 1

Lon*_*rer 5

编辑 2019 年 3 月 12 日：在我看来，答案是：它可能可以完成，但需要付出很多努力才能记住哪一行来自 QTextEdit 的预期行为。另外，由于 tdm 默认写入 stderr，因此您最终也会捕获所有异常跟踪。这就是为什么我将自己的答案标记为已解决：我发现实现相同目的更优雅：在 pyqt 中显示正在发生的事情。

这是我获得接近预期行为的最佳机会。它并没有完全回答这个问题，因为我改变了 GUI 设计。所以我不会投票解决它。此外，这一切都是在一个 python 文件中完成的。我计划进一步挑战这个解决方案，看看它是否适用于执行 tqdm 导入的真正的 python 模块。

我以一种非常丑陋的方式修补了基本的 tqdm 类。主要技巧是：

通过将原始 tqdm 类存储为新名称来动态更改 tqdm 模块结构：tqdm.orignal_class = tqdm.tqdm
然后继承tqdm.original类class TQDMPatch(tqdm.orignal_class):
实现构造函数以强制文件流+任何参数为您想要的任何内容：super(TQDMPatch, self).__init__(... change some params ...)。我给我的 TQDM 类一个自定义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 /sf/ask/2698045451/#38739634
        self.flush()


if __name__ == '__main__':
    app = QApplication(sys.argv)
    app.setStyle('Fusion')
    ex = MainApp()
    sys.exit(app.exec_())

Run Code Online (Sandbox Code Playgroud)

具有适当的独立模块

相同的解决方案，但具有实际的分离文件。

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

Run Code Online (Sandbox Code Playgroud)

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

Run Code Online (Sandbox Code Playgroud)

输出重定向工具.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, lea
        
          
          
              你的答案与你所说的你想要的相差甚远，但看起来比你想要的更漂亮 (3认同)

Answer 2

Lon*_*rer 2

使用 QProgressBar

在我最初回答很久之后，我不得不再次考虑这个问题。不要问为什么，但这次我设法用 QProgressBar 得到它:)

诀窍（至少对于 TQDM 4.63.1 及更高版本）是，有一个属性format_dict几乎包含进度条所需的所有内容。也许我们以前已经有过，但我第一次错过了......

测试用：

tqdm=4.63.1
Qt=5.15.2; PyQt=5.15.6
coloredlogs=15.0.1

Run Code Online (Sandbox Code Playgroud)

编辑：修改后的代码可在此处使用多个进度条

https://gist.github.com/LoneWanderer-GH/ec18189a8476adb463531a68430e94a8

tqdm=4.63.1
Qt=5.15.2; PyQt=5.15.6
coloredlogs=15.0.1

Run Code Online (Sandbox Code Playgroud)

1.GIF展示解决方案

2. 它是如何运作的？

正如我之前的回答，我们需要：

一个队列
已修补的 TQDM 类
用于读取队列并向 QProgressBar 发送信号的工作对象

这里的新事物是：

QProgressBar 子类
我们利用新的 TQDM 上下文with logging_redirect_tqdm():来处理日志跟踪的路由
使用自定义日志跟踪模块，并与 colorlogs 模块兼容 => 提供了带有记录器 colorlogs 的精美 QPlainTextEdit :)
stdout/stderr 流不再需要技巧

关于 TQDM 类补丁，我们重新定义__init__，但现在我们还定义refresh和close（而不是使用我之前答案中的文件流技巧）0

__init__存储新的 tqdm 实例属性、队列并发送“{do_reset:true}”（重置 QProgressBar 并使其可见）
refresh添加到队列format_dict（它包含n和总计`）
close添加到队列一个字符串“close”（隐藏进度条）

3. 完整示例（1个文件）

import contextlib
import logging
import sys
from abc import ABC, abstractmethod
from queue import Queue

from PyQt5 import QtTest
from PyQt5.QtCore import PYQT_VERSION_STR, pyqtSignal, pyqtSlot, QObject, Qt, QT_VERSION_STR, QThread
from PyQt5.QtWidgets import QApplication, QPlainTextEdit, QProgressBar, QToolButton, QVBoxLayout, QWidget


__CONFIGURED = False


def setup_streams_redirection(tqdm_nb_columns=None):
    if not __CONFIGURED:
        tqdm_update_queue = Queue()
        perform_tqdm_default_out_stream_hack(tqdm_update_queue=tqdm_update_queue, tqdm_nb_columns=tqdm_nb_columns)
        return TQDMDataQueueReceiver(tqdm_update_queue)


def perform_tqdm_default_out_stream_hack(tqdm_update_queue: Queue, tqdm_nb_columns=None):
    import tqdm
    # save original class into module
    tqdm.original_class = tqdm.std.tqdm
    parent = tqdm.std.tqdm

    class TQDMPatch(parent):
        """
        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, write_bytes=None,
                     lock_args=None, nrows=None, colour=None, delay=0, gui=False,
                     **kwargs):
            print('TQDM Patch called')  # check it works
            self.tqdm_update_queue = tqdm_update_queue
            self.tqdm_update_queue.put({"do_reset": True})
            super(TQDMPatch, self).__init__(iterable, desc, total, leave,
                                            file,  # no change here
                                            ncols,
                                            mininterval, maxinterval,
                                            miniters, ascii, disable, unit,
                                            unit_scale,
                                            False,  # change param ?
                                            smoothing,
                                            bar_format, initial, position, postfix,
                                            unit_divisor, gui, **kwargs)

        # def update(self, n=1):
        #     super(TQDMPatch, self).update(n=n)
        #     custom stuff ?

        def refresh(self, nolock=False, lock_args=None):
            super(TQDMPatch, self).refresh(nolock=nolock, lock_args=lock_args)
            self.tqdm_update_queue.put(self.format_dict)

        def close(self):
            self.tqdm_update_queue.put({"close": True})
            super(TQDMPatch, self).close()

    # change original class with the patched one, the original still exists
    tqdm.std.tqdm = TQDMPatch
    tqdm.tqdm = TQDMPatch  # may not be necessary
    # for tqdm.auto users, maybe some additional stuff is needed


class TQDMDataQueueReceiver(QObject):
    s_tqdm_object_received_signal = pyqtSignal(object)

    def __init__(self, q: Queue, *args, **kwargs):
        QObject.__init__(self, *args, **kwargs)
        self.queue = q

    @pyqtSlot()
    def run(self):
        while True:
            o = self.queue.get()
            # noinspection PyUnresolvedReferences
            self.s_tqdm_object_received_signal.emit(o)


class QTQDMProgressBar(QProgressBar):
    def __init__(self, parent, tqdm_signal: pyqtSignal):
        super(QTQDMProgressBar, self).__init__(parent)
        self.setAlignment(Qt.AlignCenter)
        self.setVisible(False)
        # noinspection PyUnresolvedReferences
        tqdm_signal.connect(self.do_it)

    def do_it(self, e):
        if not isinstance(e, dict):
            return
        do_reset = e.get("do_reset", False)  # different from close, because we want visible=true
        initial = e.get("initial", 0)
        total = e.get("total", None)
        n = e.get("n", None)
        desc = e.get("prefix", None)
        text = e.get("text", None)
        do_close = e.get("close", False)  # different from do_reset, we want visible=false
        if do_reset:
            self.reset()
        if do_close:
            self.reset()
        self.setVisible(not do_close)
        if initial:
            self.setMinimum(initial)
        else:
            self.setMinimum(0)
        if total:
            self.setMaximum(total)
        else:
            self.setMaximum(0)
        if n:
            self.setValue(n)
        if desc:
            self.setFormat(f"{desc} %v/%m | %p %")
        elif text:
            self.setFormat(text)
        else:
            self.setFormat("%v/%m | %p")


def long_procedure():
    # emulate late import of modules
    from tqdm.auto import tqdm # don't import before patch !
    __logger = logging.getLogger('long_procedure')
    __logger.setLevel(logging.DEBUG)
    tqdm_object = tqdm(range(10), unit_scale=True, dynamic_ncols=True)
    tqdm_object.set_description("My progress bar description")
    from tqdm.contrib.logging import logging_redirect_tqdm # don't import before patch !
    with logging_redirect_tqdm():
        for i in tqdm_object:
            QtTest.QTest.qWait(200)
            __logger.info(f'foo {i}')


class QtLoggingHelper(ABC):
    @abstractmethod
    def transform(self, msg: str):
        raise NotImplementedError()


class QtLoggingBasic(QtLoggingHelper):
    def transform(self, msg: str):
        return msg


class QtLoggingColoredLogs(QtLoggingHelper):
    def __init__(self):
        # offensive programming: crash if necessary if import is not present
        pass

    def transform(self, msg: str):
        import coloredlogs.converter
        msg_html = coloredlogs.converter.convert(msg)
        return msg_html


class QTextEditLogger(logging.Handler, QObject):
    appendText = pyqtSignal(str)

    def __init__(self,
                 logger_: logging.Logger,
                 formatter: logging.Formatter,
                 text_widget: QPlainTextEdit,
                 # table_widget: QTableWidget,
                 parent: QWidget):
        super(QTextEditLogger, self).__init__()
        super(QObject, self).__init__(parent=parent)
        self.text_widget = text_widget
        self.text_widget.setReadOnly(True)
        # self.table_widget = table_widget
        try:
            self.helper

归档时间：	6 年，6 月前
查看次数：	9598 次
最近记录：	2 年前