Skip to content

Instantly share code, notes, and snippets.

@zed
Last active January 4, 2021 11:46
Show Gist options
  • Star 1 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save zed/323e6ed224e39107224fd18fa5734804 to your computer and use it in GitHub Desktop.
Save zed/323e6ed224e39107224fd18fa5734804 to your computer and use it in GitHub Desktop.
#!/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