如何正确地将stdout，logging和tqdm重定向到PyQt小部件中

Question

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

我开发并使用了一堆依赖tqdm的python模块。我希望它们可以在Jupyter，控制台或GUI中使用。在Jupyter或控制台中一切正常：日志/打印和tqdm进度条之间没有碰撞。以下是显示控制台/ 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 https://stackoverflow.com/questions/38543506/change-logging-print-function-to-tqdm-write-so-logging-doesnt-interfere-wit/38739634#38739634
        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。我还使用Qazxswpoi重定向使用Queue（）向Qt QWidget使用，这样用户就可以看到发生了什么。

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

由于来自stdout的信息，我设法重定向stdout。但是，只有记录器行被重定向。 TQDM进度条仍然指向控制台输出。

这是我目前的代码：

Redirecting stdout and stderr to a PyQt5 QTextEdit from a secondary thread

得到：

# 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 https://stackoverflow.com/questions/38543506/change-logging-print-function-to-tqdm-write-so-logging-doesnt-interfere-wit/38739634#38739634
        self.flush()

if __name__ == '__main__':

    app = QApplication(sys.argv)
    app.setStyle('Fusion')
    tqdm.ncols = 50
    ex = MainApp()
    sys.exit(app.exec_())

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

---

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

选项1

这个解决方案

---> 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]

没有给出预期的结果。它适用于重定向仅包含tqdm内容的输出。

下面的代码没有给出预期的行为，不管是QTextEdit还是QPlainTextEdit。仅重定向记录器行。

Display terminal output with tqdm in QPlainTextEdit

但是，上面的代码+将

    # code from this answer
    # https://stackoverflow.com/questions/53381975/display-terminal-output-with-tqdm-in-qplaintextedit
    @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)

添加到tqdm调用会改变行为：tqdm输出被重定向到Qt小部件。但最后，只显示一行，它是记录器行或tqdm行（看起来它取决于我派生的Qt小部件）。

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

所以我找到的另一种方法是将stderr重定向到同一个流/队列中，stdout被重定向到。由于tqdm默认写入stderr，因此所有tqdm输出都会重定向到widget。

但我仍然无法弄清楚我正在寻找的确切输出。

这个问题没有提供为什么file=sys.stdout之间的行为似乎有所不同的线索

选项2

这个问题QTextEdit vs QPlainTextEdit看起来非常类似于Duplicate stdout, stderr in QTextEdit widget，并没有回答我上面描述的确切问题。

选项3

使用contextlib尝试Display terminal output with tqdm in QPlainTextEdit给了我一个错误，因为没有定义flush（）方法。修复后，我最终只有tqdm行，没有记录器行。

选项4

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

版本：

this solution

Answer 1

编辑2019-mar-12：在我看来，答案是：它可能已经完成，但需要付出很多努力才能记住QTextEdit来自哪个行的行为与插图有关。另外，由于tdm默认写入stderr，你最终也会捕获所有异常跟踪。这就是我将自己的答案标记为已解决的原因：我发现实现相同目的更为优雅：在pyqt中显示正在发生的事情。

---

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

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

• 通过将原始tqdm类存储为新名称来动态更改tqdm模块结构：

  tqdm                      4.28.1
  pyqt                      5.9.2
  PyQt5                     5.12
  PyQt5_sip                 4.19.14
  Python                    3.7.2

• 然后继承tqdm.original class tqdm.orignal_class = tqdm.tqdm
• 实现构造函数，以强制文件流+任何参数到你想要的任何：class TQDMPatch(tqdm.orignal_class):。我给了我的TQDM类一个自定义的super(TQDMPatch, self).__init__(... change some params ...)写入WriteStream()
• 更改GUI策略以拦截自定义tqdm流并将其重定向到单独的Qt小部件。我的小部件假设所有收到的打印件都包含Queue()（TQDM似乎正在做）。

它既可以在单个python文件中使用，也可以使用多个分离的模块。在后一种情况下，启动时的进口订单至关重要。

截图：

在启动处理之前

\r

在处理过程中

在处理结束时

---

这是代码

一体化文件

---

使用合适的分离模块

相同的解决方案，但与实际分离的文

• # coding=utf-8
  
  import datetime
  import logging
  import sys
  import time
  from queue import Queue
  
  from PyQt5.QtCore import pyqtSlot, pyqtSignal, QObject, QThread, Qt
  from PyQt5.QtGui import QTextCursor, QFont
  from PyQt5.QtWidgets import QTextEdit, QWidget, QToolButton, QVBoxLayout, QApplication, QLineEdit
  
  
  # DEFINITION NEEDED FIRST ...
  class WriteStream(object):
      def __init__(self, q: Queue):
          self.queue = q
  
      def write(self, text):
          self.queue.put(text)
  
      def flush(self):
          pass
  
  
  # prepare queue and streams
  queue_tqdm = Queue()
  write_stream_tqdm = WriteStream(queue_tqdm)
  
  ################## START TQDM patch procedure ##################
  import tqdm
  
  # save original class into module
  tqdm.orignal_class = tqdm.tqdm
  
  
  class TQDMPatch(tqdm.orignal_class):
      """
      Derive from original class
      """
  
      def __init__(self, iterable=None, desc=None, total=None, leave=True,
                   file=None, ncols=None, mininterval=0.1, maxinterval=10.0,
                   miniters=None, ascii=None, disable=False, unit='it',
                   unit_scale=False, dynamic_ncols=False, smoothing=0.3,
                   bar_format=None, initial=0, position=None, postfix=None,
                   unit_divisor=1000, gui=False, **kwargs):
          super(TQDMPatch, self).__init__(iterable, desc, total, leave,
                                          write_stream_tqdm,  # change any chosen file stream with our's
                                          80,  # change nb of columns (gui choice),
                                          mininterval, maxinterval,
                                          miniters, ascii, disable, unit,
                                          unit_scale, False, smoothing,
                                          bar_format, initial, position, postfix,
                                          unit_divisor, gui, **kwargs)
          print('TQDM Patch called') # check it works
  
      @classmethod
      def write(cls, s, file=None, end="\n", nolock=False):
          super(TQDMPatch, cls).write(s=s, file=file, end=end, nolock=nolock)
  
      # all other tqdm.orignal_class @classmethod methods may need to be redefined !
  
  
  # I mainly used tqdm.auto in my modules, so use that for patch
  # unsure if this will work with all possible tqdm import methods
  # might not work for tqdm_gui !
  import tqdm.auto as AUTO
  
  # change original class with the patched one, the original still exists
  AUTO.tqdm = TQDMPatch
  ################## END of TQDM patch ##################
  
  # normal MCVE code
  __is_setup_done = False
  
  
  class MainApp(QWidget):
      def __init__(self):
          super().__init__()
  
          setup_logging(self.__class__.__name__)
  
          self.__logger = logging.getLogger(self.__class__.__name__)
          self.__logger.setLevel(logging.DEBUG)
  
          # create stdout text queue
          self.queue_std_out = Queue()
          sys.stdout = WriteStream(self.queue_std_out)
  
          layout = QVBoxLayout()
  
          self.setMinimumWidth(500)
  
          self.btn_perform_actions = QToolButton(self)
          self.btn_perform_actions.setText('Launch long processing')
          self.btn_perform_actions.clicked.connect(self._btn_go_clicked)
  
          self.text_edit_std_out = StdOutTextEdit(self)
          self.text_edit_tqdm = StdTQDMTextEdit(self)
  
          self.thread_initialize = QThread()
          self.init_procedure_object = InitializationProcedures(self)
  
          # std out stream management
          # create console text read thread + receiver object
          self.thread_std_out_queue_listener = QThread()
          self.std_out_text_receiver = ThreadStdOutStreamTextQueueReceiver(self.queue_std_out)
          # connect receiver object to widget for text update
          self.std_out_text_receiver.queue_std_out_element_received_signal.connect(self.text_edit_std_out.append_text)
          # attach console text receiver to console text thread
          self.std_out_text_receiver.moveToThread(self.thread_std_out_queue_listener)
          # attach to start / stop methods
          self.thread_std_out_queue_listener.started.connect(self.std_out_text_receiver.run)
          self.thread_std_out_queue_listener.start()
  
          # NEW: TQDM stream management
          self.thread_tqdm_queue_listener = QThread()
          self.tqdm_text_receiver = ThreadTQDMStreamTextQueueReceiver(queue_tqdm)
          # connect receiver object to widget for text update
          self.tqdm_text_receiver.queue_tqdm_element_received_signal.connect(self.text_edit_tqdm.set_tqdm_text)
          # attach console text receiver to console text thread
          self.tqdm_text_receiver.moveToThread(self.thread_tqdm_queue_listener)
          # attach to start / stop methods
          self.thread_tqdm_queue_listener.started.connect(self.tqdm_text_receiver.run)
          self.thread_tqdm_queue_listener.start()
  
          layout.addWidget(self.btn_perform_actions)
          layout.addWidget(self.text_edit_std_out)
          layout.addWidget(self.text_edit_tqdm)
          self.setLayout(layout)
          self.show()
  
      @pyqtSlot()
      def _btn_go_clicked(self):
          # prepare thread for long operation
          self.init_procedure_object.moveToThread(self.thread_initialize)
          self.thread_initialize.started.connect(self.init_procedure_object.run)
          self.thread_initialize.finished.connect(self.init_procedure_object.finished)
          # start thread
          self.btn_perform_actions.setEnabled(False)
          self.thread_initialize.start()
  
  
  class ThreadStdOutStreamTextQueueReceiver(QObject):
      queue_std_out_element_received_signal = pyqtSignal(str)
  
      def __init__(self, q: Queue, *args, **kwargs):
          QObject.__init__(self, *args, **kwargs)
          self.queue = q
  
      @pyqtSlot()
      def run(self):
          self.queue_std_out_element_received_signal.emit('---> STD OUT Queue reception Started <---\n')
          while True:
              text = self.queue.get()
              self.queue_std_out_element_received_signal.emit(text)
  
  
  # NEW: dedicated receiving object for TQDM
  class ThreadTQDMStreamTextQueueReceiver(QObject):
      queue_tqdm_element_received_signal = pyqtSignal(str)
  
      def __init__(self, q: Queue, *args, **kwargs):
          QObject.__init__(self, *args, **kwargs)
          self.queue = q
  
      @pyqtSlot()
      def run(self):
          self.queue_tqdm_element_received_signal.emit('\r---> TQDM Queue reception Started <---\n')
          while True:
              text = self.queue.get()
              self.queue_tqdm_element_received_signal.emit(text)
  
  
  class StdOutTextEdit(QTextEdit):  # QTextEdit):
      def __init__(self, parent):
          super(StdOutTextEdit, self).__init__()
          self.setParent(parent)
          self.setReadOnly(True)
          self.setLineWidth(50)
          self.setMinimumWidth(500)
          self.setFont(QFont('Consolas', 11))
  
      @pyqtSlot(str)
      def append_text(self, text: str):
          self.moveCursor(QTextCursor.End)
          self.insertPlainText(text)
  
  
  class StdTQDMTextEdit(QLineEdit):
      def __init__(self, parent):
          super(StdTQDMTextEdit, self).__init__()
          self.setParent(parent)
          self.setReadOnly(True)
          self.setEnabled(True)
          self.setMinimumWidth(500)
          self.setAlignment(Qt.AlignLeft | Qt.AlignVCenter)
          self.setClearButtonEnabled(True)
          self.setFont(QFont('Consolas', 11))
  
      @pyqtSlot(str)
      def set_tqdm_text(self, text: str):
          new_text = text
          if new_text.find('\r') >= 0:
              new_text = new_text.replace('\r', '').rstrip()
              if new_text:
                  self.setText(new_text)
          else:
              # we suppose that all TQDM prints have \r
              # so drop the rest
              pass
  
  
  def long_procedure():
      # emulate import of modules
      from tqdm.auto import tqdm
  
      setup_logging('long_procedure')
      __logger = logging.getLogger('long_procedure')
      __logger.setLevel(logging.DEBUG)
      tqdm_obect = tqdm(range(10), unit_scale=True, dynamic_ncols=True)
      tqdm_obect.set_description("My progress bar description")
      for i in tqdm_obect:
          time.sleep(.1)
          __logger.info('foo {}'.format(i))
  
  
  class InitializationProcedures(QObject):
      def __init__(self, main_app: MainApp):
          super(InitializationProcedures, self).__init__()
          self._main_app = main_app
  
      @pyqtSlot()
      def run(self):
          long_procedure()
  
      @pyqtSlot()
      def finished(self):
          print("Thread finished !")  # might call main window to do some stuff with buttons
          self._main_app.btn_perform_actions.setEnabled(True)
  
  
  def setup_logging(log_prefix):
      global __is_setup_done
  
      if __is_setup_done:
          pass
      else:
          __log_file_name = "{}-{}_log_file.txt".format(log_prefix,
                                                        datetime.datetime.utcnow().isoformat().replace(":", "-"))
  
          __log_format = '%(asctime)s - %(name)-30s - %(levelname)s - %(message)s'
          __console_date_format = '%Y-%m-%d %H:%M:%S'
          __file_date_format = '%Y-%m-%d %H-%M-%S'
  
          root = logging.getLogger()
          root.setLevel(logging.DEBUG)
  
          console_formatter = logging.Formatter(__log_format, __console_date_format)
  
          file_formatter = logging.Formatter(__log_format, __file_date_format)
          file_handler = logging.FileHandler(__log_file_name, mode='a', delay=True)
  
          file_handler.setLevel(logging.DEBUG)
          file_handler.setFormatter(file_formatter)
          root.addHandler(file_handler)
  
          tqdm_handler = TqdmLoggingHandler()
          tqdm_handler.setLevel(logging.DEBUG)
          tqdm_handler.setFormatter(console_formatter)
          root.addHandler(tqdm_handler)
  
          __is_setup_done = True
  
  
  class TqdmLoggingHandler(logging.StreamHandler):
      def __init__(self):
          logging.StreamHandler.__init__(self)
  
      def emit(self, record):
          msg = self.format(record)
          tqdm.tqdm.write(msg)
          # from https://stackoverflow.com/questions/38543506/change-logging-print-function-to-tqdm-write-so-logging-doesnt-interfere-wit/38739634#38739634
          self.flush()
  
  
  if __name__ == '__main__':
      app = QApplication(sys.argv)
      app.setStyle('Fusion')
      ex = MainApp()
      sys.exit(app.exec_())

  ，节目切入点
• MyPyQtGUI.py是执行流程中应该完成的第一次导入。主持所有的魔力。
• output_redirection_tools.py，一个托管配置元素的配置模块
• config.py，自定义日志记录配置
• my_logging.py，我使用但不想改变的一些代码的示例版本。

my py QT GUI.朋友

值得注意的是，项目的第一次导入应该是third_party_module_not_to_change.py，因为它完成了所有tqdm黑客工作。

import output_redirection_tools

没有_logging.朋友

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

output_redirection_tools.朋友

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

config.朋友

import sys
from queue import Queue

from PyQt5.QtCore import pyqtSlot, pyqtSignal, QObject

from config import config_dict, IS_STREAMS_REDIRECTION_SETUP_DONE, TQDM_WRITE_STREAM_CONFIG, STDOUT_WRITE_STREAM_CONFIG, \
    STREAM_CONFIG_KEY_QUEUE, STREAM_CONFIG_KEY_STREAM, STREAM_CONFIG_KEY_QT_QUEUE_RECEIVER


class QueueWriteStream(object):
    def __init__(self, q: Queue):
        self.queue = q

    def write(self, text):
        self.queue.put(text)

    def flush(self):
        pass


def perform_tqdm_default_out_stream_hack(tqdm_file_stream, tqdm_nb_columns=None):
    import tqdm
    # save original class into module
    tqdm.orignal_class = tqdm.tqdm

    class TQDMPatch(tqdm.orignal_class):
        """
        Derive from original class
        """

        def __init__(self, iterable=None, desc=None, total=None, leave=True,
                     file=None, ncols=None, mininterval=0.1, maxinterval=10.0,
                     miniters=None, ascii=None, disable=False, unit='it',
                     unit_scale=False, dynamic_ncols=False, smoothing=0.3,
                     bar_format=None, initial=0, position=None, postfix=None,
                     unit_divisor=1000, gui=False, **kwargs):
            super(TQDMPatch, self).__init__(iterable, desc, total, leave,
                                            tqdm_file_stream,  # change any chosen file stream with our's
                                            tqdm_nb_columns,  # change nb of columns (gui choice),
                                            mininterval, maxinterval,
                                            miniters, ascii, disable, unit,
                                            unit_scale,
                                            False,  # change param
                                            smoothing,
                                            bar_format, initial, position, postfix,
                                            unit_divisor, gui, **kwargs)
            print('TQDM Patch called')  # check it works

        @classmethod
        def write(cls, s, file=None, end="\n", nolock=False):
            super(TQDMPatch, cls).write(s=s, file=file, end=end, nolock=nolock)
            #tqdm.orignal_class.write(s=s, file=file, end=end, nolock=nolock)

        # all other tqdm.orignal_class @classmethod methods may need to be redefined !

    # # I mainly used tqdm.auto in my modules, so use that for patch
    # # unsure if this will work with all possible tqdm import methods
    # # might not work for tqdm_gui !
    import tqdm.auto as AUTO
    #
    # # change original class with the patched one, the original still exists
    AUTO.tqdm = TQDMPatch
    #tqdm.tqdm = TQDMPatch


def setup_streams_redirection(tqdm_nb_columns=None):
    if config_dict[IS_STREAMS_REDIRECTION_SETUP_DONE]:
        pass
    else:
        configure_tqdm_redirection(tqdm_nb_columns)
        configure_std_out_redirection()
        config_dict[IS_STREAMS_REDIRECTION_SETUP_DONE] = True


def configure_std_out_redirection():
    queue_std_out = Queue()
    config_dict[STDOUT_WRITE_STREAM_CONFIG] = {
        STREAM_CONFIG_KEY_QUEUE: queue_std_out,
        STREAM_CONFIG_KEY_STREAM: QueueWriteStream(queue_std_out),
        STREAM_CONFIG_KEY_QT_QUEUE_RECEIVER: StdOutTextQueueReceiver(q=queue_std_out)
    }
    perform_std_out_hack()


def perform_std_out_hack():
    sys.stdout = config_dict[STDOUT_WRITE_STREAM_CONFIG][STREAM_CONFIG_KEY_STREAM]


def configure_tqdm_redirection(tqdm_nb_columns=None):
    queue_tqdm = Queue()
    config_dict[TQDM_WRITE_STREAM_CONFIG] = {
        STREAM_CONFIG_KEY_QUEUE: queue_tqdm,
        STREAM_CONFIG_KEY_STREAM: QueueWriteStream(queue_tqdm),
        STREAM_CONFIG_KEY_QT_QUEUE_RECEIVER: TQDMTextQueueReceiver(q=queue_tqdm)
    }
    perform_tqdm_default_out_stream_hack(
        tqdm_file_stream=config_dict[TQDM_WRITE_STREAM_CONFIG][STREAM_CONFIG_KEY_STREAM],
        tqdm_nb_columns=tqdm_nb_columns)


class StdOutTextQueueReceiver(QObject):
    # we are forced to define 1 signal per class
    # see https://stackoverflow.com/questions/50294652/how-to-create-pyqtsignals-dynamically
    queue_std_out_element_received_signal = pyqtSignal(str)

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

    @pyqtSlot()
    def run(self):
        self.queue_std_out_element_received_signal.emit('---> STD OUT Queue reception Started <---\n')
        while True:
            text = self.queue.get()
            self.queue_std_out_element_received_signal.emit(text)


class TQDMTextQueueReceiver(QObject):
    # we are forced to define 1 signal per class
    # see https://stackoverflow.com/questions/50294652/how-to-create-pyqtsignals-dynamically
    queue_tqdm_element_received_signal = pyqtSignal(str)

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

    @pyqtSlot()
    def run(self):
        # we assume that all TQDM outputs start with \r, so use that to show stream reception is started
        self.queue_tqdm_element_received_signal.emit('\r---> TQDM Queue reception Started <---\n')
        while True:
            text = self.queue.get()
            self.queue_tqdm_element_received_signal.emit(text)


setup_streams_redirection()

third_part_module_not_to_嫦娥.朋友

表示我使用的代码类型，不希望/不能更改。

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