Last active
January 4, 2021 11:46
-
-
Save zed/323e6ed224e39107224fd18fa5734804 to your computer and use it in GitHub Desktop.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
#!/usr/bin/env python3 | |
""" | |
- read subprocess output without threads using a socket pair | |
- show the output in a tkinter GUI (while the process is still running) | |
- stop subprocess on a button press | |
""" | |
import logging | |
import os | |
import socket | |
import sys | |
import tkinter as tk | |
from subprocess import Popen, STDOUT | |
logger = logging.getLogger(__name__) | |
# define dummy command to generate some output | |
cmd = [ | |
sys.executable or "python", | |
"-u", | |
"-c", | |
""" | |
import itertools, time | |
for i in itertools.count(): | |
print(i) | |
time.sleep(0.1) | |
""", | |
] | |
class ShowProcessOutputDemo: | |
def __init__(self, root): | |
"""Start subprocess, make GUI widgets.""" | |
self.root = root | |
# start subprocess | |
self.read_sock, write_sock = socket.socketpair() | |
write_sock.set_inheritable(True) | |
self.proc = Popen(cmd, stdout=write_sock, stderr=STDOUT) | |
logger.info("started") | |
# show subprocess' stdout in GUI | |
self.read_sock.setblocking(False) | |
self.root.createfilehandler(self.read_sock, tk.READABLE, self.read_output) | |
self._var = tk.StringVar() # put subprocess output here | |
tk.Label(root, textvariable=self._var).pack() | |
# stop subprocess using a button | |
tk.Button(root, text="Stop subprocess", command=self.stop).pack() | |
def read_output(self, pipe, mask, bufsize=8192): | |
"""Read subprocess' output, pass it to the GUI.""" | |
data = os.read(pipe.fileno(), bufsize) | |
if not data: # clean up | |
logger.info("eof") | |
self.root.deletefilehandler(self.read_sock) | |
self.root.after(5000, self.stop) # stop in 5 seconds | |
return | |
logger.debug("got: %r", data) | |
self._var.set(data.strip(b"\n").decode()) | |
def stop(self, stopping=[]): | |
"""Stop subprocess and quit GUI.""" | |
logger.info("stopping") | |
self.proc.terminate() | |
self.proc.wait() # wait for the subprocess' exit | |
self.root.destroy() # exit GUI | |
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(message)s") | |
root = tk.Tk() | |
app = ShowProcessOutputDemo(root) | |
root.protocol("WM_DELETE_WINDOW", app.stop) # exit subprocess if GUI is closed | |
root.mainloop() | |
logger.info("exited") |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment