Skip to content

@zed /show-subprocess-output-io-watch.py secret
Last active

Embed URL

HTTPS clone URL

Subversion checkout URL

You can clone with
or
.
Download ZIP
#!/usr/bin/python
from subprocess import Popen, PIPE
from gi.repository import GObject, Gtk
# create GUI to show subprocess output
win = Gtk.Window()
win.set_default_size(640, 480)
label = Gtk.Label('process output')
win.add(label)
# start dummy subprocess to generate some output
proc = Popen(["python", "-u", "-c", """import itertools, sys, time
for i in itertools.count():
print(i)
sys.stderr.write(".") # show that we are alive in terminal
time.sleep(0.5)
"""], stdout=PIPE)
# read from subprocess
def read_data(source, condition):
line = source.readline() # might block
if not line:
source.close()
return False # stop reading
# update text
label.set_text('Subprocess output: %r' % (line.strip(),))
return True # continue reading
io_id = GObject.io_add_watch(proc.stdout, GObject.IO_IN, read_data)
# exit subprocess if GUI is closed
def quit(win, event, proc):
GObject.source_remove(io_id) # stop watching
proc.terminate()
# kill subprocess if it hasn't exited in timeout seconds
timeout = 5
def poll(proc, countdown):
if proc.poll() is None: # subprocess hasn't exited yet
countdown[0] -= 1
if countdown[0] <= 0:
proc.kill()
proc.wait()
Gtk.main_quit()
else:
return True # continue polling
else: # subprocess exited
Gtk.main_quit()
# call poll() once a second
GObject.timeout_add(1000, poll, proc, [timeout])
win.connect('delete-event', quit, proc)
win.show_all()
Gtk.main()
#!/usr/bin/python
from contextlib import closing
from threading import Thread
from subprocess import Popen, PIPE
from gi.repository import GObject, Gtk
GObject.threads_init() # all Gtk is in the main thread;
# only GObject.idle_add() is in the background thread
# create GUI to show subprocess output
win = Gtk.Window()
win.set_default_size(640, 480)
label = Gtk.Label('process output')
win.add(label)
# start dummy subprocess to generate some output
proc = Popen(["python", "-u", "-c", """import itertools, sys, time
for i in itertools.count():
print(i)
sys.stderr.write(".") # show that we are alive in terminal
time.sleep(0.5)
"""], stdout=PIPE)
# read from subprocess in a separate thread
def reader_thread(proc, update_text):
with closing(proc.stdout) as file:
for line in iter(file.readline, b''):
# execute update_text() in GUI thread
GObject.idle_add(update_text, 'Subprocess output: %r' % (
line.strip(),))
t = Thread(target=reader_thread, args=[proc, label.set_text])
t.daemon = True # exit with the program
t.start()
# exit subprocess if GUI is closed
def quit(win, event, proc):
proc.terminate()
# kill subprocess if it hasn't exited in timeout seconds
timeout = 5
def poll(proc, countdown):
if proc.poll() is None: # subprocess hasn't exited yet
countdown[0] -= 1
if countdown[0] <= 0:
proc.kill()
proc.wait()
Gtk.main_quit()
else:
return True # continue polling
else: # subprocess exited
Gtk.main_quit()
# call poll() once a second
GObject.timeout_add(1000, poll, proc, [timeout])
win.connect('delete-event', quit, proc)
win.show_all()
Gtk.main()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Something went wrong with that request. Please try again.