Redirecting Stdout and Stderr to a Pyqt4 Qtextedit from a Secondary Thread

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

Firstly, +1 for realising how thread-unsafe many of the examples on stack overflow are!

The solution is to use a thread-safe object (like a Python Queue.Queue) to mediate the transfer of information. I've attached some sample code below which redirects stdout to a Python Queue. This Queue is read by a QThread, which emits the contents to the main thread through Qt's signal/slot mechanism (emitting signals is thread-safe). The main thread then writes the text to a text edit.

Hope that is clear, feel free to ask questions if it is not!

EDIT: Note that the code example provided doesn't clean up QThreads nicely, so you'll get warnings printed when you quit. I'll leave it to you to extend to your use case and clean up the thread(s)

import sys
from Queue import Queue
from PyQt4.QtCore import *
from PyQt4.QtGui import *

# The new Stream Object which replaces the default stream associated with sys.stdout
# This object just puts data in a queue!
class WriteStream(object):
    def __init__(self,queue):
        self.queue = queue

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

# A QObject (to be run in a QThread) which sits waiting for data to come through a Queue.Queue().
# It blocks until data is available, and one it has got something from the queue, it sends
# it to the "MainThread" by emitting a Qt Signal 
class MyReceiver(QObject):
    mysignal = pyqtSignal(str)

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

    @pyqtSlot()
    def run(self):
        while True:
            text = self.queue.get()
            self.mysignal.emit(text)

# An example QObject (to be run in a QThread) which outputs information with print
class LongRunningThing(QObject):
    @pyqtSlot()
    def run(self):
        for i in range(1000):
            print i

# An Example application QWidget containing the textedit to redirect stdout to
class MyApp(QWidget):
    def __init__(self,*args,**kwargs):
        QWidget.__init__(self,*args,**kwargs)

        self.layout = QVBoxLayout(self)
        self.textedit = QTextEdit()
        self.button = QPushButton('start long running thread')
        self.button.clicked.connect(self.start_thread)
        self.layout.addWidget(self.textedit)
        self.layout.addWidget(self.button)

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

    @pyqtSlot()
    def start_thread(self):
        self.thread = QThread()
        self.long_running_thing = LongRunningThing()
        self.long_running_thing.moveToThread(self.thread)
        self.thread.started.connect(self.long_running_thing.run)
        self.thread.start()

# Create Queue and redirect sys.stdout to this queue
queue = Queue()
sys.stdout = WriteStream(queue)

# Create QApplication and QWidget
qapp = QApplication(sys.argv)  
app = MyApp()
app.show()

# Create thread that will listen on the other end of the queue, and send the text to the textedit in our application
thread = QThread()
my_receiver = MyReceiver(queue)
my_receiver.mysignal.connect(app.append_text)
my_receiver.moveToThread(thread)
thread.started.connect(my_receiver.run)
thread.start()

qapp.exec_()

Redirecting stdout from a secondary thread (multithreading with a function instead of class?)

For this case you can use the native threading of python:

import sys

import threading
import time

from PyQt5.QtWidgets import QApplication, QMainWindow
from PyQt5.QtCore import QObject, pyqtSignal
from PyQt5.QtGui import QTextCursor

from ui_form import Ui_Form

class EmittingStream(QObject):  # test
    textWritten = pyqtSignal(str)

    def write(self, text):
        self.textWritten.emit(str(text))

class Form(QMainWindow):
    finished = pyqtSignal()
    def __init__(self, parent=None):
        super(Form, self).__init__(parent)

        # Install the custom output stream
        sys.stdout = EmittingStream(textWritten=self.normalOutputWritten)  # test

        self.ui = Ui_Form()
        self.ui.setupUi(self)

        self.ui.pushButton_text.clicked.connect(self.start_task)
        self.finished.connect(lambda: self.ui.pushButton_text.setEnabled(True))

    def start_task(self):
        var = self.ui.lineEdit.text()
        self.thread = threading.Thread(target=self.test_write, args=(args, ))
        self.thread.start()
        self.ui.pushButton_text.setEnabled(False)

    def __del__(self):  # test
        # Restore sys.stdout
        sys.stdout = sys.__stdout__

    def normalOutputWritten(self, text):  # test
        """Append text to the QTextEdit."""
        cursor = self.ui.textEdit.textCursor()
        cursor.movePosition(QTextCursor.End)
        cursor.insertText(text)
        self.ui.textEdit.setTextCursor(cursor)
        self.ui.textEdit.ensureCursorVisible()

    def test_write(self, *args):
        var1 = args[0]
        print("something written")
        time.sleep(5) # simulate expensive task
        print("something written ----")
        self.finished.emit()

def main():
    app = QApplication(sys.argv)
    form = Form()
    form.show()
    sys.exit(app.exec_())

if __name__ == "__main__":
    main()

Duplicate stdout, stderr in QTextEdit widget

It seems that all you need to do is override sys.stderr and sys.stdout with a wrapper object that emits a signal whenever output is written.

Below is a demo script that should do more or less what you want. Note that the wrapper class does not restore sys.stdout/sys.stderr from sys.__stdout__/sys.__stderr__, because the latter objects may not be same as the ones that were orignally replaced.

PyQt5:

import sys
from PyQt5 import QtWidgets, QtGui, QtCore

class OutputWrapper(QtCore.QObject):
    outputWritten = QtCore.pyqtSignal(object, object)

    def __init__(self, parent, stdout=True):
        super().__init__(parent)
        if stdout:
            self._stream = sys.stdout
            sys.stdout = self
        else:
            self._stream = sys.stderr
            sys.stderr = self
        self._stdout = stdout

    def write(self, text):
        self._stream.write(text)
        self.outputWritten.emit(text, self._stdout)

    def __getattr__(self, name):
        return getattr(self._stream, name)

    def __del__(self):
        try:
            if self._stdout:
                sys.stdout = self._stream
            else:
                sys.stderr = self._stream
        except AttributeError:
            pass

class Window(QtWidgets.QMainWindow):
    def __init__(self):
        super().__init__(   )
        widget = QtWidgets.QWidget(self)
        layout = QtWidgets.QVBoxLayout(widget)
        self.setCentralWidget(widget)
        self.terminal = QtWidgets.QTextBrowser(self)
        self._err_color = QtCore.Qt.red
        self.button = QtWidgets.QPushButton('Test', self)
        self.button.clicked.connect(self.handleButton)
        layout.addWidget(self.terminal)
        layout.addWidget(self.button)
        stdout = OutputWrapper(self, True)
        stdout.outputWritten.connect(self.handleOutput)
        stderr = OutputWrapper(self, False)
        stderr.outputWritten.connect(self.handleOutput)

    def handleOutput(self, text, stdout):
        color = self.terminal.textColor()
        self.terminal.moveCursor(QtGui.QTextCursor.End)
        self.terminal.setTextColor(color if stdout else self._err_color)
        self.terminal.insertPlainText(text)
        self.terminal.setTextColor(color)

    def handleButton(self):
        if QtCore.QTime.currentTime().second() % 2:
            print('Printing to stdout...')
        else:
            print('Printing to stderr...', file=sys.stderr)

if __name__ == '__main__':

    app = QtWidgets.QApplication(sys.argv)
    window = Window()
    window.setGeometry(500, 300, 300, 200)
    window.show()
    sys.exit(app.exec_())

PyQt4:

import sys
from PyQt4 import QtGui, QtCore

class OutputWrapper(QtCore.QObject):
    outputWritten = QtCore.pyqtSignal(object, object)

    def __init__(self, parent, stdout=True):
        QtCore.QObject.__init__(self, parent)
        if stdout:
            self._stream = sys.stdout
            sys.stdout = self
        else:
            self._stream = sys.stderr
            sys.stderr = self
        self._stdout = stdout

    def write(self, text):
        self._stream.write(text)
        self.outputWritten.emit(text, self._stdout)

    def __getattr__(self, name):
        return getattr(self._stream, name)

    def __del__(self):
        try:
            if self._stdout:
                sys.stdout = self._stream
            else:
                sys.stderr = self._stream
        except AttributeError:
            pass

class Window(QtGui.QMainWindow):
    def __init__(self):
        QtGui.QMainWindow.__init__(self)
        widget = QtGui.QWidget(self)
        layout = QtGui.QVBoxLayout(widget)
        self.setCentralWidget(widget)
        self.terminal = QtGui.QTextBrowser(self)
        self._err_color = QtCore.Qt.red
        self.button = QtGui.QPushButton('Test', self)
        self.button.clicked.connect(self.handleButton)
        layout.addWidget(self.terminal)
        layout.addWidget(self.button)
        stdout = OutputWrapper(self, True)
        stdout.outputWritten.connect(self.handleOutput)
        stderr = OutputWrapper(self, False)
        stderr.outputWritten.connect(self.handleOutput)

    def handleOutput(self, text, stdout):
        color = self.terminal.textColor()
        self.terminal.moveCursor(QtGui.QTextCursor.End)
        self.terminal.setTextColor(color if stdout else self._err_color)
        self.terminal.insertPlainText(text)
        self.terminal.setTextColor(color)

    def handleButton(self):
        if QtCore.QTime.currentTime().second() % 2:
            print('Printing to stdout...')
        else:
            sys.stderr.write('Printing to stderr...\n')

if __name__ == '__main__':

    app = QtGui.QApplication(sys.argv)
    window = Window()
    window.setGeometry(500, 300, 300, 200)
    window.show()
    sys.exit(app.exec_())

NB:

Instances of the OutputWrapper should be created as early as possible, so as to ensure that other modules that need sys.stdout/sys.stderr (such as the logging module) use the wrapped versions wherever necessary.

PySide: redirect stdout to a dialog

It seems like this can be done easily by subclassing QThread:

class Main(QtGui.QWindow):
  def __init__(self):
    super(Main, self).__init__()
    self.run_button = QtGui.QPushButton("Run")

    mainLayout = QtGui.QVBoxLayout()
    mainLayout.addWidget(self.run_button)
    self.setLayout(mainLayout)

    self.run_button.clicked.connect(self.run_event)

  def run_event(self):
    # Create the dialog to display output
    self.viewer = OutputDialog()

    # Create the worker thread and run it
    self.thread = WorkerThread()
    self.thread.message.connect(self.viewer.write)
    self.thread.start()

class WorkerThread(QtCore.QThread):
  message = QtCore.Signal(str)   

  def run(self):    
    sys.stdout = self

    for i in range(10):
      print "Im doing stuff"
      time.sleep(1)    

  def write(self, text):
    self.message.emit(text)

However, most documentation on the web seems to recommend that you do not subclass QThread and instead use the moveToThread function for processing tasks.

Also, I don't know how you would distinguish between stdout and stderr in the above method (assuming that you redirect both to the workerthread)

How to correctly redirect stdout, logging and tqdm into a PyQt widget

Using QProgressBar

Long after my inital anwser, I had to think about this again.
Don't ask why, but this time I managed to get it with a QProgressBar :)

The trick (at least with TQDM 4.63.1 and higher), is that there is a property format_dict with almost everything necessary for a progress bar. Maybe we already did have that before, but I missed it the first time ...

Tested with:

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

1. GIF showing the solution

2. How does it work?

As in my previous answer, we need:

• a queue
• a patched TQDM class
• a worker object to read queue and send signal to QProgressBar

New thing here are:

• a QProgressBar subclass
• we take advantage of the new TQDM context with logging_redirect_tqdm(): which handles routing of logging traces
• use of a custom logging traces module, with compatibility with the coloredlogs module => provides a fancy QPlainTextEdit with logger coloredlogs :)
• no more tricks with stdout/stderr streams

Concerning the TQDM class patch, we redefine __init__, but now we also define refresh and close (instead of using the file stream trick from my previous answer)0

• __init__ stores a new tqdm instance attribute, the queue and sends a "{do_reset:true}" (to reset the QProgressBar and make it visible)
• refresh adds to queue format_dict (it contains n and total`)
• close adds to queue a string "close" (to hide the progress bar)

3. Full example (1 file)

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

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

__CONFIGURED = False

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

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

    class TQDMPatch(parent):
        """
        Derive from original class
        """

        def __init__(self, iterable=None, desc=None, total=None, leave=True, file=None,
                     ncols=None, mininterval=0.1, maxinterval=10.0, miniters=None,
                     ascii=None, disable=False, unit='it', unit_scale=False,
                     dynamic_ncols=False, smoothing=0.3, bar_format=None, initial=0,
                     position=None, postfix=None, unit_divisor=1000, write_bytes=None,
                     lock_args=None, nrows=None, colour=None, delay=0, gui=False,
                     **kwargs):
            print('TQDM Patch called')  # check it works
            self.tqdm_update_queue = tqdm_update_queue
            self.tqdm_update_queue.put({"do_reset": True})
            super(TQDMPatch, self).__init__(iterable, desc, total, leave,
                                            file,  # no change here
                                            ncols,
                                            mininterval, maxinterval,
                                            miniters, ascii, disable, unit,
                                            unit_scale,
                                            False,  # change param ?
                                            smoothing,
                                            bar_format, initial, position, postfix,
                                            unit_divisor, gui, **kwargs)

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

    def __init__(self,
                 logger_: logging.Logger,
                 formatter: logging.Formatter,
                 text_widget: QPlainTextEdit,
                 # table_widget: QTableWidget,
                 parent: QWidget):
        super(QTextEditLogger, self).__init__()
        super(QObject, self).__init__(parent=parent)
        self.text_widget = text_widget
        self.text_widget.setReadOnly(True)
        # self.table_widget = table_widget
        try:
            self.helper = 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_())

Can't Update QTextEdit While in a Multiprocess

Qt does not support multiprocessing so it is dangerous to update the GUI from another process, the GUI can only and should be updated from the thread of the process where it was created.

On the other hand in this case it is not necessary to use multiprocessing since you can use QProcess:

import sys

from PyQt4 import QtCore, QtGui

class MainWindow(QtGui.QMainWindow):
    def __init__(self, parent=None):
        super(MainWindow, self).__init__(parent)
        self.process = QtCore.QProcess(self)
        self.process.setProcessChannelMode(QtCore.QProcess.MergedChannels)
        self.process.readyReadStandardOutput.connect(self.on_readyReadStandardOutput)
        self.textedit = QtGui.QTextEdit()

        self.setCentralWidget(self.textedit)

    def tail(self, filename):
        self.process.kill()
        self.process.start("tail", ["-F", filename])

    @QtCore.pyqtSlot()
    def on_readyReadStandardOutput(self):
        msg = self.process.readAllStandardOutput().data().encode()
        self.textedit.append(msg)

if __name__ == "__main__":
    app = QtGui.QApplication(sys.argv)
    w = MainWindow()
    w.tail("<FILENAME>")
    w.show()
    sys.exit(app.exec_())

---

Related Topics