Dark Artistry

#!/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()