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.