TQDM 4.28.1+ : Change tqdm file stream for all instanciation (eg: for redirection to PyQt)

[LoneWanderer-GH]

• I have visited the source website, and in particular
  read the known issues
• I have searched through the issue tracker for duplicates ==> issue 482 is related, issue 172 is related
• I have mentioned version numbers, operating system and
  environment, where applicable:

  import tqdm, sys
  print(tqdm.__version__, sys.version, sys.platform)

TQDM 4.28.1 Python 3.7.2 (default, Feb 21 2019, 17:35:59) [MSC v.1915 64 bit (AMD64)] win32

I'm working on PyQt5 project where I invoke python modules that use TQDM.
These modules rely on TQDM and never specify a file stream at tqdm instanciation.
I already had some logging configuration to nicely integrate logging and tqdm in console.

For the PyQt5 GUI, I wanted to catch all console outputs (logging and tqdm) to redirect them into 1 Qt widget. I did not succeed in having both stdout and stderr behave in PyQt the same way it did in console, but I found a workaround.

What I ended up with is :

• intercept stdout stream with a custom stream class instance
• intercept all tqdm instanciations and force tqdm instanciation `file=' paramter with a custom stream class instance
• have 1 Qt widget for all intercepted stdout stream prints
• have 1 Qt widget for all intercepted tqdm stream prints
• (this way, stderr stream is preserved and all error stacks go into console)

To intercept all tqdm instance creations and prints without rewriting all my python modules, I ended up with code described here: https://stackoverflow.com/a/55082521/7237062
I give the code here in case SO goes away (who knows what might happen ...).

The workaround is as follows:

• I moved the original tqdm class into another symbol, and replaced it with my own derived class.

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

• The derived class forces the file parameter to use one of my own. This way, I could redirect all tqdm prints to a Qt widget.

class TQDMPatch(tqdm.orignal_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)

This could pave the way to a full tqdm 4+ features interception for a GUI progress bar (PyQt or other), but I have not yet tried it myself (use of format_dict ? see this commit). However, I don't what developments are currently made, and this could proove obsolete in the next versions (v5 https://github.com/tqdm/tqdm/milestone/8).

So I'm not expecting the following code to be approved as a tqdm feature, or an open ticket. But it might be of help for anyone trying to do the same. I also know that the tqdm class trick is messy, ugly, non-canonical. But it gave me a rather acceptable behavior.

But if there is any better way to do this kind of manipulation, I would be glad to hear it.

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

[casperdcl]

Thanks for this example. Yes, there are probably more canonical ways of doing this especially if I ever get around to v5. Not like I get any funding for this out of control hobby project :)

[WBP20]

Hello, this is very helpfull, I could adapt it to my code, but for me

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

This doesn't work, I can't perform action when Thread is finish. Do you know where problem might come from ?

[LoneWanderer-GH]

@WBP20 this is a code error, it is not a canonical Qt code sample ...

For what it's worth, I managed to make a PyQt QProgressBar working with TQDM.
Needless to say, all I had to do was to correctly use the format_dict property (the naming was confusing to me, since it is a dictionnary that basically tells everything concerning the current stage of progress ... I would favor a more explicit naming like (progress_status_dict ?)

Anyway, here is the complete code example + gif
Link to SO answer: https://stackoverflow.com/a/74091829/7237062

@WBP20 it also contains a fix of the issue o re-enabling the button at the end of the processing)
@casperdcl this should be more canonical :)

---

This does 3 things:

• redirect logger text to a QWidget
• format logger text to colored HTML in case using coloredlogs
• redirect 1 TQDM progress bar to a QProgressBar widget which shows at tqdm creation and hides at tqdm close

TODO:

• determine if a canonical integration into tqdm is possible
• handle multiple nested progress bar : it needs as much QProgressBar widgets, and probably play aournd with the self.pos attribute ?
• check if it answers to tqdm + pandas + pyqt #943 ?

GIF/screen

Code

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

# YOU NEED TO DEFER TQDM IMPORT SO THE PATCH CAN TAKE PLACE **BEFORE** TQDM class instanciation

__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
    __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
    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 = QtLoggingColoredLogs()
            self.appendText.connect(self.text_widget.appendHtml)
            logger_.info("Using QtLoggingColoredLogs")
        except ImportError:
            self.helper = QtLoggingBasic()
            self.appendText.connect(self.text_widget.appendPlainText)
            logger_.warning("Using QtLoggingBasic")
        # logTextBox = QTextEditLogger(self)
        # You can format what is printed to text box
        self.setFormatter(formatter)
        logger_.addHandler(self)
        # You can control the logging level
        self.setLevel(logging.DEBUG)

    def emit(self, record: logging.LogRecord):
        msg = self.format(record)
        display_msg = self.helper.transform(msg=msg)
        self.appendText.emit(display_msg)
        # self.add_row(record)


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

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

        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.thread_initialize = QThread()
        self.init_procedure_object = LongProcedureWorker(self)

        self.thread_tqdm_update_queue_listener = QThread()
        # must be done before any TQDM import
        self.tqdm_update_receiver = setup_streams_redirection()
        self.tqdm_update_receiver.moveToThread(self.thread_tqdm_update_queue_listener)
        self.thread_tqdm_update_queue_listener.started.connect(self.tqdm_update_receiver.run)

        self.pb_tqdm = QTQDMProgressBar(self, tqdm_signal=self.tqdm_update_receiver.s_tqdm_object_received_signal)
        layout.addWidget(self.pb_tqdm)
        self.thread_tqdm_update_queue_listener.start()

        self.plain_text_edit_logger = QPlainTextEdit(self)
        LOG_FMT = "{asctime} | {levelname:10s} | {message}"
        try:
            import coloredlogs
            FORMATTER = coloredlogs.ColoredFormatter(fmt=LOG_FMT, style="{")
        except ImportError:
            FORMATTER = logging.Formatter(fmt=LOG_FMT, style="{")

        self.logging_ = QTextEditLogger(logger_=logging.getLogger(),  # root logger, to intercept every log of app
                                        formatter=FORMATTER,
                                        text_widget=self.plain_text_edit_logger,
                                        parent=self)
        layout.addWidget(self.plain_text_edit_logger)
        layout.addWidget(self.btn_perform_actions)
        self.setLayout(layout)
        import tqdm
        self.__logger.info(f"tqdm {tqdm.__version__}")
        self.__logger.info(f"Qt={QT_VERSION_STR}; PyQt={PYQT_VERSION_STR}")
        with contextlib.suppress(ImportError):
            import coloredlogs
            self.__logger.info(f"coloredlogs {coloredlogs.__version__}")
        # prepare thread for long operation
        self.init_procedure_object.moveToThread(self.thread_initialize)
        self.thread_initialize.started.connect(self.init_procedure_object.run)
        self.init_procedure_object.finished.connect(self._init_procedure_finished)
        self.init_procedure_object.finished.connect(self.thread_initialize.quit)
        self.show()

    @pyqtSlot()
    def _btn_go_clicked(self):
        # start thread
        self.btn_perform_actions.setEnabled(False)
        self.__logger.info("Launch Thread")
        self.thread_initialize.start()

    def _init_procedure_finished(self):
        self.btn_perform_actions.setEnabled(True)


class LongProcedureWorker(QObject):
    finished = pyqtSignal()

    def __init__(self, main_app: MainApp):
        super(LongProcedureWorker, self).__init__()
        self._main_app = main_app

    @pyqtSlot()
    def run(self):
        long_procedure()
        self.finished.emit()


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

[LoneWanderer-GH]

And multiple progress bars with PyQt example:
https://gist.github.com/LoneWanderer-GH/ec18189a8476adb463531a68430e94a8

[LoneWanderer-GH]

maybe related :
#1253
#297
#172