Quick Python3 Gtk3 app to execute Python scripts to test long running threads in Gtk3 app. Some things could be better, like the python paths collection and displaying the PID, but I was focused on running threads without freezing the GUI and it just kind of happened. Hope it’s useful to someone.
#!/usr/bin/python3
import sys
if sys.version_info[0] < 3:
raise Exception("Python 3 or a more recent version is required.")
if 'linux' not in sys.platform:
raise Exception("Linux is required.")
try:
import gi
gi.require_version('Gtk', '3.0')
from gi.repository import Gtk, Gio, GLib, GObject
except:
raise Exception("Missing 'gi' [Gtk 3].")
try:
import psutil
except:
raise Exception("Missing 'psutil', try sudo apt install python3-psutil.")
import os
import os.path
import fnmatch
import pickle
import threading
import time
import getpass
from subprocess import PIPE
'''
Find all Python executables on Linux machine.
Allow easy execution of Python scripts from GUI,
because I'm lazy...
'''
GObject.threads_init()
class Worker(threading.Thread):
def __init__(self, cmd, callback):
threading.Thread.__init__(self)
self.cmd = cmd
self.callback = callback
def run(self):
if not self.cmd:
_get_execs() # Get Python execs.
GObject.idle_add(self.callback)
else:
msg,err = _process(self.cmd)
GObject.idle_add(self.callback,[msg,err])
def app_load():
p = '/home/%s/.pyexec' % getpass.getuser()
if not os.path.exists(p):
os.mkdir(p)
p = os.path.join(p,'execs.bin')
with open(p, 'rb') as stream:
data = stream.read()
execs = pickle.loads(data)
return execs
def update_execs(data):
p = '/home/%s/.pyexec' % getpass.getuser()
if not os.path.exists(p):
os.mkdir(p)
p = os.path.join(p,'execs.bin')
with open(p, 'wb') as output:
output.write(data)
def load_user():
user = None
p = '/home/%s/.pyexec/user.bin' % getpass.getuser()
if os.path.exists(p):
with open(p, 'rb') as stream:
data = stream.read()
user = pickle.loads(data)
return user
def save_user(data):
p = '/home/%s/.pyexec' % getpass.getuser()
if not os.path.exists(p):
os.mkdir(p)
p = os.path.join(p,'user.bin')
data = pickle.dumps(data)
with open(p, 'wb') as output:
output.write(data)
def search(where='/', find='python*[!-config]'):
for dirpath, dirnames, filenames in os.walk(where):
if 'flatpak' in dirpath.lower(): continue
if 'timeshift' in dirpath.lower(): continue
if 'bin' not in dirpath.lower(): continue
for filename in filenames:
file_path = os.path.join(dirpath, filename)
if not os.access(file_path, os.X_OK): continue
if not fnmatch.fnmatch(filename, find): continue
if not os.path.islink(file_path): continue
yield file_path
def _get_execs():
python_execs = [item for item in search()]
dumped = pickle.dumps(python_execs)
update_execs(dumped)
def _process(cmd):
proc = psutil.Popen(cmd, stdout=PIPE, stderr=PIPE)
print('NAME: %s PID = %s' % (proc.name(), str(proc.pid).split()[0]))
msg, err = proc.communicate()
return msg, err
class Executor(Gtk.Window):
'''
Main app window class.
'''
def __init__(self):
Gtk.Window.__init__(self, title="Python Executor")
self.set_border_width(10)
self.set_default_size(400, 350)
self.python_list = None
self.proj_path = None
self.script = None
self.selected = None
self.cmd = None
self.msg = ''
self.err = ''
self.pid = ''
self.versions = ['/usr/bin/python'] # placeholder
# If the python list does not exist create, else populate versions.
self.app_load_check()
# Load user settings.
self.proj_path_default = "/home/%s/Documents/Python" % getpass.getuser()
self.script_default = "script.py"
self.cbox_index = 0
user_settings = load_user()
if user_settings:
self.cbox_index, self.proj_path_default, self.script_default = user_settings
# Define header add buttons.
header = Gtk.HeaderBar(title="Python Executor")
header.props.show_close_button = True
self.set_titlebar(header)
# Button to refresh python list.
refresh_button = Gtk.Button()
icon_refresh = Gio.ThemedIcon(name="reload")
image_refresh = Gtk.Image.new_from_gicon(icon_refresh, Gtk.IconSize.BUTTON)
refresh_button.add(image_refresh)
refresh_button.set_tooltip_text("Refresh environments")
refresh_button.connect("clicked", self.on_refresh_clicked)
header.pack_start(refresh_button)
# Kill button.
kill_button = Gtk.Button()
icon_kill = Gio.ThemedIcon(name="process-stop")
image_kill = Gtk.Image.new_from_gicon(icon_kill, Gtk.IconSize.BUTTON)
kill_button.add(image_kill)
kill_button.set_tooltip_text("Kill Process")
kill_button.connect("clicked", self.on_kill_clicked)
header.pack_start(kill_button)
# Execute button.
run_button = Gtk.Button()
icon_run = Gio.ThemedIcon(name="run-build")
image_run = Gtk.Image.new_from_gicon(icon_run, Gtk.IconSize.BUTTON)
run_button.add(image_run)
run_button.set_tooltip_text("Run")
run_button.connect("clicked", self.on_run_clicked)
header.pack_start(run_button)
# Define layout.
self.box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=10)
self.pybox = Gtk.Box(spacing=10)
self.prjbox = Gtk.Box(spacing=10)
self.scrbox = Gtk.Box(spacing=10)
self.add(self.box)
# Python execs.
py_label = Gtk.Label()
py_label.set_text("Python")
py_label.set_justify(Gtk.Justification.LEFT)
py_label.set_hexpand(False)
self.py_store = Gtk.ListStore(str)
for ver in self.versions:
if isinstance(ver, str):
self.py_store.append([ver])
self.python_list = Gtk.ComboBox.new_with_model(self.py_store)
self.python_list.connect("changed", self.on_changed)
self.python_list.set_hexpand(False)
renderer_text = Gtk.CellRendererText()
self.python_list.pack_start(renderer_text, True)
self.python_list.add_attribute(renderer_text, "text", 0)
self.python_list.set_active(self.cbox_index)
self.pybox.pack_start(py_label, False, False, 0)
self.pybox.pack_end(self.python_list, False, True, 0)
self.box.pack_start(self.pybox, False, True, 0)
# Project Path.
prj_label = Gtk.Label()
prj_label.set_text("Project ")
prj_label.set_justify(Gtk.Justification.LEFT)
prj_label.set_hexpand(False)
self.proj_path = Gtk.Entry()
self.proj_path.set_text(self.proj_path_default )
self.proj_path.set_hexpand(True)
self.prjbox.pack_start(prj_label, False, True, 0)
self.prjbox.pack_end(self.proj_path, False, True, 0)
self.box.pack_start(self.prjbox, False, True, 0)
# Script
scr_label = Gtk.Label()
scr_label.set_text("Script ")
scr_label.set_justify(Gtk.Justification.LEFT)
scr_label.set_hexpand(False)
self.script = Gtk.Entry()
self.script.set_text(self.script_default)
self.script.set_hexpand(True)
self.scrbox.pack_start(scr_label, False, True, 0)
self.scrbox.pack_end(self.script, False, True, 0)
self.box.pack_start(self.scrbox, False, True, 0)
# Message area.
scrolledwindow = Gtk.ScrolledWindow()
scrolledwindow.set_hexpand(True)
scrolledwindow.set_vexpand(True)
self.message_view = Gtk.TextView()
self.textbuffer = self.message_view.get_buffer()
self.tag_err = self.textbuffer.create_tag('err', foreground="#FF0000")
self.tag_msg = self.textbuffer.create_tag('msg', foreground="#FFFFFF")
scrolledwindow.add(self.message_view)
self.box.pack_start(scrolledwindow, True, True, 0)
self.show_all()
def app_load_check(self):
'''Fetch Python paths.'''
p = '/home/%s/.pyexec' % getpass.getuser()
if not os.path.exists(p):
os.mkdir(p)
p = os.path.join(p,'execs.bin')
if not os.path.exists(p):
thread = Worker(None, self.end_refresh)
thread.start()
else:
self.versions = app_load()
def on_refresh_clicked(self, widget):
'''Refresh Python paths.'''
thread = Worker(None, self.end_refresh)
thread.start()
def end_refresh(self):
'''Thread callback.'''
self.versions = app_load()
for ver in self.versions:
if ver not in self.py_store:
if isinstance(ver, str):
self.py_store.append([ver])
def on_changed(self, widget):
'''Selected Python path.'''
index = widget.get_active()
self.selected = widget.get_model()[index][0]
def on_kill_clicked(self, widget):
kill_msg = '\n* kill: Nothing to do.'
end_iter = self.textbuffer.get_end_iter()
for process in psutil.process_iter():
if process.cmdline() == self.cmd:
process.terminate()
kill_msg = '\n* %s: %s killed' % (process.pid, process.cmdline())
self.textbuffer.insert(end_iter, kill_msg)
break
def on_run_clicked(self, widget):
'''Execute Python script.'''
ok = True
self.cmd = None
python_path = self.selected
python_path_id = self.python_list.get_active()
if not python_path:
ok = False
self.textbuffer.set_text('Python exec not defined.')
if not os.path.exists(python_path):
ok = False
self.textbuffer.set_text('Python exec not found: %s' % python_path)
script_path = self.proj_path.get_text()
script = self.script.get_text()
full_script_path = os.path.join(script_path,script)
if not os.path.exists(full_script_path):
ok = False
self.textbuffer.set_text('Script not found: %s' % full_script_path)
if ok:
user_set = [python_path_id, script_path, script]
save_user(user_set)
self.textbuffer.set_text(self.pid)
self.pid = 'PID: Unkown. Possibly ended quickly.'
self.cmd = [python_path, full_script_path]
thread = Worker(self.cmd, self.messages)
thread.start()
for process in psutil.process_iter():
if process.cmdline() == self.cmd:
self.pid = 'PID = %s: %s' % (process.pid, process.cmdline())
self.textbuffer.set_text(self.pid)
break
def messages(self,rvals):
r_err = rvals[-1].decode()
r_ok = rvals[0].decode()
self.err = '\nResults:\n\n%s' % r_err
self.msg = '\nResults:\n\n%s' % r_ok
start_iter = self.textbuffer.get_start_iter()
end_iter = self.textbuffer.get_end_iter()
self.textbuffer.remove_all_tags(start_iter,end_iter)
if r_err:
self.textbuffer.insert_with_tags(end_iter, self.err, self.tag_err)
elif r_ok:
if 'exception' in r_ok.lower() or 'error' in r_ok.lower() and 'traceback' in r_ok.lower():
self.textbuffer.insert_with_tags(end_iter, self.msg, self.tag_err)
else:
self.textbuffer.insert_with_tags(end_iter, self.msg, self.tag_msg)
self.message_view.show()
if __name__ == '__main__':
win = Executor()
win.connect("destroy", Gtk.main_quit)
win.show_all()
Gtk.main()