How to capture the output of a long-running program and present it in a GUI in Python? How to capture the output of a long-running program and present it in a GUI in Python? tkinter tkinter

How to capture the output of a long-running program and present it in a GUI in Python?


Under the assumption that the process you're calling is long-running and doesn't produce all its output in one go, it means you cannot use subprocess.Popen.communicate(), as that is designed to read all output up to an end of file.

You will have to use other standard techniques to read from the pipe.

As you want to integrate it with a GUI and the process is long-running, you will need to coordinate reading its output with the GUI's main loop. This complicates things somewhat.

TkInter

Let's first assume you want to use TkInter, as in one of your examples. That confronts us with a couple of Problems:

  • There's no integration of TkInter with the select module.
  • There's even no canonical integration of TkInter with asyncio as of now (also see https://bugs.python.org/issue27546).
  • Hacking together a custom main loop using root.update() is usually recommended against, leaving us solving with threading what should have been an event based approach.
  • TkInter's event_generate() is missing Tk's ability to send user data along with the event, so we can't use TkInter events to pass the received output from one thread to another.

Thus, we will tackle it with threading (even if I'd prefer not to), where the main thread controls the Tk GUI and a helper thread reads the output from the process, and lacking a native way in TkInter to pass data around, we utilize a thread-safe Queue.

#!/usr/bin/env python3from subprocess import Popen, PIPE, STDOUT, TimeoutExpiredfrom threading import Thread, Eventfrom queue import Queue, Emptyfrom tkinter import Tk, Text, ENDclass ProcessOutputReader(Thread):    def __init__(self, queue, cmd, params=(),                 group=None, name=None, daemon=True):        super().__init__(group=group, name=name, daemon=daemon)        self._stop_request = Event()        self.queue = queue        self.process = Popen((cmd,) + tuple(params),                             stdout=PIPE,                             stderr=STDOUT,                             universal_newlines=True)    def run(self):        for line in self.process.stdout:            if self._stop_request.is_set():                # if stopping was requested, terminate the process and bail out                self.process.terminate()                break            self.queue.put(line)  # enqueue the line for further processing        try:            # give process a chance to exit gracefully            self.process.wait(timeout=3)        except TimeoutExpired:            # otherwise try to terminate it forcefully            self.process.kill()    def stop(self):        # request the thread to exit gracefully during its next loop iteration        self._stop_request.set()        # empty the queue, so the thread will be woken up        # if it is blocking on a full queue        while True:            try:                self.queue.get(block=False)            except Empty:                break            self.queue.task_done()  # acknowledge line has been processedclass MyConsole(Text):    def __init__(self, parent, queue, update_interval=50, process_lines=500):        super().__init__(parent)        self.queue = queue        self.update_interval = update_interval        self.process_lines = process_lines        self.after(self.update_interval, self.fetch_lines)    def fetch_lines(self):        something_inserted = False        for _ in range(self.process_lines):            try:                line = self.queue.get(block=False)            except Empty:                break            self.insert(END, line)            self.queue.task_done()  # acknowledge line has been processed            # ensure scrolling the view is at most done once per interval            something_inserted = True        if something_inserted:            self.see(END)        self.after(self.update_interval, self.fetch_lines)# create the root widgetroot = Tk()# create a queue for sending the lines from the process output reader thread# to the TkInter main threadline_queue = Queue(maxsize=1000)# create a process output readerreader = ProcessOutputReader(line_queue, 'python3', params=['-u', 'test.py'])# create a consoleconsole = MyConsole(root, line_queue)reader.start()   # start the processconsole.pack()   # make the console visibleroot.mainloop()  # run the TkInter main loopreader.stop()reader.join(timeout=5)  # give thread a chance to exit gracefullyif reader.is_alive():    raise RuntimeError("process output reader failed to stop")

Due to the aforementioned caveats, the TkInter code ends up a bit on the larger side.

PyQt

Using PyQt instead, we can considerably improve our situation, as that framework already comes with a native way to integrate with a subprocess, in the shape of its QProcess class.

That means we can do away with threads and use Qt's native Signal and Slot mechanism instead.

#!/usr/bin/env python3import sysfrom PyQt5.QtCore import pyqtSignal, pyqtSlot, QProcess, QTextCodecfrom PyQt5.QtGui import QTextCursorfrom PyQt5.QtWidgets import QApplication, QPlainTextEditclass ProcessOutputReader(QProcess):    produce_output = pyqtSignal(str)    def __init__(self, parent=None):        super().__init__(parent=parent)        # merge stderr channel into stdout channel        self.setProcessChannelMode(QProcess.MergedChannels)        # prepare decoding process' output to Unicode        codec = QTextCodec.codecForLocale()        self._decoder_stdout = codec.makeDecoder()        # only necessary when stderr channel isn't merged into stdout:        # self._decoder_stderr = codec.makeDecoder()        self.readyReadStandardOutput.connect(self._ready_read_standard_output)        # only necessary when stderr channel isn't merged into stdout:        # self.readyReadStandardError.connect(self._ready_read_standard_error)    @pyqtSlot()    def _ready_read_standard_output(self):        raw_bytes = self.readAllStandardOutput()        text = self._decoder_stdout.toUnicode(raw_bytes)        self.produce_output.emit(text)    # only necessary when stderr channel isn't merged into stdout:    # @pyqtSlot()    # def _ready_read_standard_error(self):    #     raw_bytes = self.readAllStandardError()    #     text = self._decoder_stderr.toUnicode(raw_bytes)    #     self.produce_output.emit(text)class MyConsole(QPlainTextEdit):    def __init__(self, parent=None):        super().__init__(parent=parent)        self.setReadOnly(True)        self.setMaximumBlockCount(10000)  # limit console to 10000 lines        self._cursor_output = self.textCursor()    @pyqtSlot(str)    def append_output(self, text):        self._cursor_output.insertText(text)        self.scroll_to_last_line()    def scroll_to_last_line(self):        cursor = self.textCursor()        cursor.movePosition(QTextCursor.End)        cursor.movePosition(QTextCursor.Up if cursor.atBlockStart() else                            QTextCursor.StartOfLine)        self.setTextCursor(cursor)# create the application instanceapp = QApplication(sys.argv)# create a process output readerreader = ProcessOutputReader()# create a console and connect the process output reader to itconsole = MyConsole()reader.produce_output.connect(console.append_output)reader.start('python3', ['-u', 'test.py'])  # start the processconsole.show()                              # make the console visibleapp.exec_()                                 # run the PyQt main loop

We end up with a little boilerplate deriving from the Qt classes, but with an overall cleaner approach.

General considerations

Also make sure that the process you're calling is not buffering multiple output lines, as otherwise it will still look as if the console got stuck.

In particular if the callee is a python program, you can either ensure that it's using print(..., flush=True) or call it with python -u callee.py to enforce unbuffered output.