Как правильно перенаправить стандартный вывод, ведение журнала и tqdm в виджет PyQt - PullRequest
1 голос
/ 07 марта 2019

Прежде всего, я знаю, что многие вопросы похожи на этот. Но потратив на это столько времени, я теперь ищу помощи у сообщества.

Я разработал и использую несколько модулей Python, которые полагаются на tqdm. Я хочу, чтобы их можно было использовать внутри Jupyter, в консоли или с графическим интерфейсом. Все хорошо работает в 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]

Теперь я делаю GUI с PyQt, который использует код, аналогичный приведенному выше. Поскольку обработка может быть длительной, я использовал многопоточность, чтобы избежать зависания HMI во время обработки. Я также использовал перенаправление stdout с использованием Queue () к Qt QWidget, чтобы пользователь мог видеть, что происходит.

Мой текущий вариант использования - 1 отдельный поток с журналами и индикаторами выполнения tqdm для перенаправления на 1 выделенный виджет. (Я не ищу несколько потоков, чтобы кормить виджет несколькими журналами и несколькими индикаторами выполнения tqdm).

Мне удалось перенаправить stdout благодаря информации из Перенаправление stdout и stderr в PyQt5 QTextEdit из вторичного потока . Однако перенаправляются только строки журнала. Индикатор выполнения 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 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_())

Дает: Logging correctly redirected but not TQDM progress bar output

Я хотел бы получить точное поведение, которое у меня было бы при строгом вызове кода в консоли. то есть ожидаемый результат в виджете 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 в QPlainTextEdit не дает ожидаемых результатов. Он хорошо работает для перенаправления выводов, содержащих только tqdm.

Следующий код не дает предполагаемого поведения, является ли это с QTextEdit или 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)

Однако приведенный выше код + добавление file=sys.stdout к вызову tqdm меняет поведение: вывод tqdm перенаправляется в виджет Qt. Но, в конце концов, отображается только одна строка, и это либо строка логгера, либо строка tqdm (похоже, это зависит от того, какой виджет Qt я получил).

В конце концов, изменение всех используемых вызовов tqdm не должно быть предпочтительным вариантом.

Таким образом, другой подход, который я нашел, заключается в перенаправлении stderr в том же потоке / очереди, куда перенаправляется stdout. Поскольку tqdm записывает в stderr по умолчанию, все выходы tqdm перенаправляются в виджет.

Но я до сих пор не могу понять, какой именно результат я ищу.

Этот вопрос не дает представление о том, почему поведение, кажется, отличается между QTextEdit против QPlainTextEdit

Вариант 2

Этот вопрос Duplicate stdout, stderr в виджете QTextEdit выглядит очень похоже на Отображение вывода терминала с tqdm в QPlainTextEdit и не отвечает на мою точную проблему, описанную выше.

Вариант 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

1 Ответ

0 голосов
/ 10 марта 2019

РЕДАКТИРОВАТЬ 2019-mar-12: мне кажется, что ответ таков: это, вероятно, можно сделать, но требует много усилий, чтобы запомнить, откуда берется строка для QTextEdit, чтобы вести себя как намеренно,Кроме того, поскольку tdm записывает в stderr по умолчанию, вам также придется перехватывать все трассировки исключений.Вот почему я отмечу свой собственный ответ как решенный: я считаю более элегантным достижение той же цели: покажи в pyqt, что происходит.


Вот мой лучший способ получитьчто-то близкое к предполагаемому поведению.Это не совсем отвечает на вопрос, потому что я изменил дизайн GUI.Поэтому я не буду голосовать, как решено.Более того, все это делается в одном файле Python.Я планирую в дальнейшем оспорить это решение, чтобы увидеть, работает ли оно с реальными модулями Python, выполняющими импорт tqdm.

Я исправил базовый класс 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()
  • изменение стратегии графического интерфейса для перехвата и перенаправления вашего пользовательского потока tqdm в отдельный виджет Qt.Мой виджет предполагает, что все полученные отпечатки содержат \r (что, похоже, делает TQDM).

Он работает как в одном файле Python, так и с несколькими отдельными модулями.В последнем случае порядок импорта при запуске является критическим.

Снимки экрана:

Перед запуском обработки

Before launching processing

Во время обработки

During processing

В конце обработки

At end of procesing


Вот код

Все-в-одном файле

# coding=utf-8

import datetime
import logging
import sys
import time
from queue import Queue

from PyQt5.QtCore import pyqtSlot, pyqtSignal, QObject, QThread, Qt
from PyQt5.QtGui import QTextCursor, QFont
from PyQt5.QtWidgets import QTextEdit, QWidget, QToolButton, QVBoxLayout, QApplication, QLineEdit


# DEFINITION NEEDED FIRST ...
class WriteStream(object):
    def __init__(self, q: Queue):
        self.queue = q

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

    def flush(self):
        pass


# prepare queue and streams
queue_tqdm = Queue()
write_stream_tqdm = WriteStream(queue_tqdm)

################## START TQDM patch procedure ##################
import tqdm

# save original class into module
tqdm.orignal_class = tqdm.tqdm


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

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

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

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


# I mainly used tqdm.auto in my modules, so use that for patch
# unsure if this will work with all possible tqdm import methods
# might not work for tqdm_gui !
import tqdm.auto as AUTO

# change original class with the patched one, the original still exists
AUTO.tqdm = TQDMPatch
################## END of TQDM patch ##################

# normal MCVE code
__is_setup_done = False


class MainApp(QWidget):
    def __init__(self):
        super().__init__()

        setup_logging(self.__class__.__name__)

        self.__logger = logging.getLogger(self.__class__.__name__)
        self.__logger.setLevel(logging.DEBUG)

        # create stdout text queue
        self.queue_std_out = Queue()
        sys.stdout = WriteStream(self.queue_std_out)

        layout = QVBoxLayout()

        self.setMinimumWidth(500)

        self.btn_perform_actions = QToolButton(self)
        self.btn_perform_actions.setText('Launch long processing')
        self.btn_perform_actions.clicked.connect(self._btn_go_clicked)

        self.text_edit_std_out = StdOutTextEdit(self)
        self.text_edit_tqdm = StdTQDMTextEdit(self)

        self.thread_initialize = QThread()
        self.init_procedure_object = InitializationProcedures(self)

        # std out stream management
        # create console text read thread + receiver object
        self.thread_std_out_queue_listener = QThread()
        self.std_out_text_receiver = ThreadStdOutStreamTextQueueReceiver(self.queue_std_out)
        # connect receiver object to widget for text update
        self.std_out_text_receiver.queue_std_out_element_received_signal.connect(self.text_edit_std_out.append_text)
        # attach console text receiver to console text thread
        self.std_out_text_receiver.moveToThread(self.thread_std_out_queue_listener)
        # attach to start / stop methods
        self.thread_std_out_queue_listener.started.connect(self.std_out_text_receiver.run)
        self.thread_std_out_queue_listener.start()

        # NEW: TQDM stream management
        self.thread_tqdm_queue_listener = QThread()
        self.tqdm_text_receiver = ThreadTQDMStreamTextQueueReceiver(queue_tqdm)
        # connect receiver object to widget for text update
        self.tqdm_text_receiver.queue_tqdm_element_received_signal.connect(self.text_edit_tqdm.set_tqdm_text)
        # attach console text receiver to console text thread
        self.tqdm_text_receiver.moveToThread(self.thread_tqdm_queue_listener)
        # attach to start / stop methods
        self.thread_tqdm_queue_listener.started.connect(self.tqdm_text_receiver.run)
        self.thread_tqdm_queue_listener.start()

        layout.addWidget(self.btn_perform_actions)
        layout.addWidget(self.text_edit_std_out)
        layout.addWidget(self.text_edit_tqdm)
        self.setLayout(layout)
        self.show()

    @pyqtSlot()
    def _btn_go_clicked(self):
        # prepare thread for long operation
        self.init_procedure_object.moveToThread(self.thread_initialize)
        self.thread_initialize.started.connect(self.init_procedure_object.run)
        self.thread_initialize.finished.connect(self.init_procedure_object.finished)
        # start thread
        self.btn_perform_actions.setEnabled(False)
        self.thread_initialize.start()


class ThreadStdOutStreamTextQueueReceiver(QObject):
    queue_std_out_element_received_signal = pyqtSignal(str)

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

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


# NEW: dedicated receiving object for TQDM
class ThreadTQDMStreamTextQueueReceiver(QObject):
    queue_tqdm_element_received_signal = pyqtSignal(str)

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

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


class StdOutTextEdit(QTextEdit):  # QTextEdit):
    def __init__(self, parent):
        super(StdOutTextEdit, self).__init__()
        self.setParent(parent)
        self.setReadOnly(True)
        self.setLineWidth(50)
        self.setMinimumWidth(500)
        self.setFont(QFont('Consolas', 11))

    @pyqtSlot(str)
    def append_text(self, text: str):
        self.moveCursor(QTextCursor.End)
        self.insertPlainText(text)


class StdTQDMTextEdit(QLineEdit):
    def __init__(self, parent):
        super(StdTQDMTextEdit, self).__init__()
        self.setParent(parent)
        self.setReadOnly(True)
        self.setEnabled(True)
        self.setMinimumWidth(500)
        self.setAlignment(Qt.AlignLeft | Qt.AlignVCenter)
        self.setClearButtonEnabled(True)
        self.setFont(QFont('Consolas', 11))

    @pyqtSlot(str)
    def set_tqdm_text(self, text: str):
        new_text = text
        if new_text.find('\r') >= 0:
            new_text = new_text.replace('\r', '').rstrip()
            if new_text:
                self.setText(new_text)
        else:
            # we suppose that all TQDM prints have \r
            # so drop the rest
            pass


def long_procedure():
    # emulate import of modules
    from tqdm.auto import tqdm

    setup_logging('long_procedure')
    __logger = logging.getLogger('long_procedure')
    __logger.setLevel(logging.DEBUG)
    tqdm_obect = tqdm(range(10), unit_scale=True, dynamic_ncols=True)
    tqdm_obect.set_description("My progress bar description")
    for i in tqdm_obect:
        time.sleep(.1)
        __logger.info('foo {}'.format(i))


class InitializationProcedures(QObject):
    def __init__(self, main_app: MainApp):
        super(InitializationProcedures, self).__init__()
        self._main_app = main_app

    @pyqtSlot()
    def run(self):
        long_procedure()

    @pyqtSlot()
    def finished(self):
        print("Thread finished !")  # might call main window to do some stuff with buttons
        self._main_app.btn_perform_actions.setEnabled(True)


def setup_logging(log_prefix):
    global __is_setup_done

    if __is_setup_done:
        pass
    else:
        __log_file_name = "{}-{}_log_file.txt".format(log_prefix,
                                                      datetime.datetime.utcnow().isoformat().replace(":", "-"))

        __log_format = '%(asctime)s - %(name)-30s - %(levelname)s - %(message)s'
        __console_date_format = '%Y-%m-%d %H:%M:%S'
        __file_date_format = '%Y-%m-%d %H-%M-%S'

        root = logging.getLogger()
        root.setLevel(logging.DEBUG)

        console_formatter = logging.Formatter(__log_format, __console_date_format)

        file_formatter = logging.Formatter(__log_format, __file_date_format)
        file_handler = logging.FileHandler(__log_file_name, mode='a', delay=True)

        file_handler.setLevel(logging.DEBUG)
        file_handler.setFormatter(file_formatter)
        root.addHandler(file_handler)

        tqdm_handler = TqdmLoggingHandler()
        tqdm_handler.setLevel(logging.DEBUG)
        tqdm_handler.setFormatter(console_formatter)
        root.addHandler(tqdm_handler)

        __is_setup_done = True


class TqdmLoggingHandler(logging.StreamHandler):
    def __init__(self):
        logging.StreamHandler.__init__(self)

    def emit(self, record):
        msg = self.format(record)
        tqdm.tqdm.write(msg)
        # from https://stackoverflow.com/questions/38543506/change-logging-print-function-to-tqdm-write-so-logging-doesnt-interfere-wit/38739634#38739634
        self.flush()


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

с правильными отдельными модулями

То же решение, но с фактически разделенными файлами.

  • MyPyQtGUI.py, точка входа в программу
  • output_redirection_tools.py самый первый импорт, который должен быть выполнен во время выполнения.Содержит все волшебство.
  • config.py, модуль конфигурации, содержащий элементы конфигурации
  • my_logging.py, настраиваемая конфигурация регистрации
  • third_party_module_not_to_change.py, пример версии некоторого кода Iиспользуйте, но не хотите менять.

MyPyQtGUI.py

Важно отметить, что самый первый импорт проекта должен быть import output_redirection_tools поскольку он выполняет всю работу по взлому tqdm.

# looks like an unused import, but it actually does the TQDM class trick to intercept prints
import output_redirection_tools # KEEP ME !!!

import logging
import sys

from PyQt5.QtCore import pyqtSlot, QObject, QThread, Qt
from PyQt5.QtGui import QTextCursor, QFont
from PyQt5.QtWidgets import QTextEdit, QWidget, QToolButton, QVBoxLayout, QApplication, QLineEdit

from config import config_dict, STDOUT_WRITE_STREAM_CONFIG, TQDM_WRITE_STREAM_CONFIG, STREAM_CONFIG_KEY_QUEUE, \
    STREAM_CONFIG_KEY_QT_QUEUE_RECEIVER
from my_logging import setup_logging

import third_party_module_not_to_change

class MainApp(QWidget):
    def __init__(self):
        super().__init__()

        setup_logging(self.__class__.__name__)

        self.__logger = logging.getLogger(self.__class__.__name__)
        self.__logger.setLevel(logging.DEBUG)

        self.queue_std_out = config_dict[STDOUT_WRITE_STREAM_CONFIG][STREAM_CONFIG_KEY_QUEUE]
        self.queue_tqdm = config_dict[TQDM_WRITE_STREAM_CONFIG][STREAM_CONFIG_KEY_QUEUE]

        layout = QVBoxLayout()

        self.setMinimumWidth(500)

        self.btn_perform_actions = QToolButton(self)
        self.btn_perform_actions.setText('Launch long processing')
        self.btn_perform_actions.clicked.connect(self._btn_go_clicked)

        self.text_edit_std_out = StdOutTextEdit(self)
        self.text_edit_tqdm = StdTQDMTextEdit(self)

        self.thread_initialize = QThread()
        self.init_procedure_object = LongProcedureWrapper(self)

        # std out stream management
        # create console text read thread + receiver object
        self.thread_std_out_queue_listener = QThread()
        self.std_out_text_receiver = config_dict[STDOUT_WRITE_STREAM_CONFIG][STREAM_CONFIG_KEY_QT_QUEUE_RECEIVER]
        # connect receiver object to widget for text update
        self.std_out_text_receiver.queue_std_out_element_received_signal.connect(self.text_edit_std_out.append_text)
        # attach console text receiver to console text thread
        self.std_out_text_receiver.moveToThread(self.thread_std_out_queue_listener)
        # attach to start / stop methods
        self.thread_std_out_queue_listener.started.connect(self.std_out_text_receiver.run)
        self.thread_std_out_queue_listener.start()

        # NEW: TQDM stream management
        self.thread_tqdm_queue_listener = QThread()
        self.tqdm_text_receiver = config_dict[TQDM_WRITE_STREAM_CONFIG][STREAM_CONFIG_KEY_QT_QUEUE_RECEIVER]
        # connect receiver object to widget for text update
        self.tqdm_text_receiver.queue_tqdm_element_received_signal.connect(self.text_edit_tqdm.set_tqdm_text)
        # attach console text receiver to console text thread
        self.tqdm_text_receiver.moveToThread(self.thread_tqdm_queue_listener)
        # attach to start / stop methods
        self.thread_tqdm_queue_listener.started.connect(self.tqdm_text_receiver.run)
        self.thread_tqdm_queue_listener.start()

        layout.addWidget(self.btn_perform_actions)
        layout.addWidget(self.text_edit_std_out)
        layout.addWidget(self.text_edit_tqdm)
        self.setLayout(layout)
        self.show()

    @pyqtSlot()
    def _btn_go_clicked(self):
        # prepare thread for long operation
        self.init_procedure_object.moveToThread(self.thread_initialize)
        self.thread_initialize.started.connect(self.init_procedure_object.run)
        # start thread
        self.btn_perform_actions.setEnabled(False)
        self.thread_initialize.start()


class StdOutTextEdit(QTextEdit):
    def __init__(self, parent):
        super(StdOutTextEdit, self).__init__()
        self.setParent(parent)
        self.setReadOnly(True)
        self.setLineWidth(50)
        self.setMinimumWidth(500)
        self.setFont(QFont('Consolas', 11))

    @pyqtSlot(str)
    def append_text(self, text: str):
        self.moveCursor(QTextCursor.End)
        self.insertPlainText(text)


class StdTQDMTextEdit(QLineEdit):
    def __init__(self, parent):
        super(StdTQDMTextEdit, self).__init__()
        self.setParent(parent)
        self.setReadOnly(True)
        self.setEnabled(True)
        self.setMinimumWidth(500)
        self.setAlignment(Qt.AlignLeft | Qt.AlignVCenter)
        self.setClearButtonEnabled(True)
        self.setFont(QFont('Consolas', 11))

    @pyqtSlot(str)
    def set_tqdm_text(self, text: str):
        new_text = text
        if new_text.find('\r') >= 0:
            new_text = new_text.replace('\r', '').rstrip()
            if new_text:
                self.setText(new_text)
        else:
            # we suppose that all TQDM prints have \r, so drop the rest
            pass


class LongProcedureWrapper(QObject):
    def __init__(self, main_app: MainApp):
        super(LongProcedureWrapper, self).__init__()
        self._main_app = main_app

    @pyqtSlot()
    def run(self):
        third_party_module_not_to_change.long_procedure()


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

my_logging.py

import logging
import datetime
import tqdm

from config import config_dict, IS_SETUP_DONE


def setup_logging(log_prefix, force_debug_level=logging.DEBUG):

    root = logging.getLogger()
    root.setLevel(force_debug_level)

    if config_dict[IS_SETUP_DONE]:
        pass
    else:
        __log_file_name = "{}-{}_log_file.txt".format(log_prefix,
                                                      datetime.datetime.utcnow().isoformat().replace(":", "-"))

        __log_format = '%(asctime)s - %(name)-30s - %(levelname)s - %(message)s'
        __console_date_format = '%Y-%m-%d %H:%M:%S'
        __file_date_format = '%Y-%m-%d %H-%M-%S'

        console_formatter = logging.Formatter(__log_format, __console_date_format)

        file_formatter = logging.Formatter(__log_format, __file_date_format)
        file_handler = logging.FileHandler(__log_file_name, mode='a', delay=True)

        file_handler.setLevel(logging.DEBUG)
        file_handler.setFormatter(file_formatter)
        root.addHandler(file_handler)

        tqdm_handler = TqdmLoggingHandler()
        tqdm_handler.setLevel(logging.DEBUG)
        tqdm_handler.setFormatter(console_formatter)
        root.addHandler(tqdm_handler)

        config_dict[IS_SETUP_DONE] = True


class TqdmLoggingHandler(logging.StreamHandler):
    def __init__(self):
        logging.StreamHandler.__init__(self)

    def emit(self, record):
        msg = self.format(record)
        tqdm.tqdm.write(msg)
        self.flush()

output_redirection_tools.py

import sys
from queue import Queue

from PyQt5.QtCore import pyqtSlot, pyqtSignal, QObject

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


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

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

    def flush(self):
        pass


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

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

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

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

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

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


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


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


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


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


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

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

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


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

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

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


setup_streams_redirection()

config.py

IS_SETUP_DONE = 'is_setup_done'

TQDM_WRITE_STREAM_CONFIG = 'TQDM_WRITE_STREAM_CONFIG'
STDOUT_WRITE_STREAM_CONFIG = 'STDOUT_WRITE_STREAM_CONFIG'
IS_STREAMS_REDIRECTION_SETUP_DONE = 'IS_STREAMS_REDIRECTION_SETUP_DONE'

STREAM_CONFIG_KEY_QUEUE = 'queue'
STREAM_CONFIG_KEY_STREAM = 'write_stream'
STREAM_CONFIG_KEY_QT_QUEUE_RECEIVER = 'qt_queue_receiver'

default_config_dict = {
    IS_SETUP_DONE: False,
    IS_STREAMS_REDIRECTION_SETUP_DONE: False,
    TQDM_WRITE_STREAM_CONFIG: None,
    STDOUT_WRITE_STREAM_CONFIG: None,
}

config_dict = default_config_dict

third_part_module_not_to_change.py

представляет тип кода, который я использую и не хочуна / не может измениться.

from tqdm.auto import tqdm
import logging
from my_logging import setup_logging
import time


def long_procedure():

    setup_logging('long_procedure')
    __logger = logging.getLogger('long_procedure')
    __logger.setLevel(logging.DEBUG)
    tqdm_obect = tqdm(range(10), unit_scale=True, dynamic_ncols=True)
    tqdm_obect.set_description("My progress bar description")
    for i in tqdm_obect:
        time.sleep(.1)
        __logger.info('foo {}'.format(i))
...