Run process with realtime output to a Tkinter GUI

Question

I was trying to create a GUI in Tkinter python. I want to display the output of a tool to my Tkinter interface. The tool works great in command line but it is a continuous scanner. Somewhat like a continuous ping (I mean by ping command in Linux with no options).

Now the problem is since the output of ping is never complete, therefore I cannot print the output in Tkinter. It also makes my application go freeze. I also cannot stop the command after few seconds to display output. Run process with realtime output in PHP I found this above link helpful for php, but How can I convert this code in python:

https://stackoverflow.com/a/6144213/4931414

Here is some sample code that I want to display on tkinter frame

#!/usr....

import subprocess
x = subprocess.call(["ping", "127.0.0.1"])
print x

This works great on command line but I am not getting output on tkinter interface.

Answer 1

First of all, I must admit that I am not so familiar with the module subprocess and threading, but I have tried to create a simple console that accepts you to write a command, whose output will be shown in a Text widget.

The basic idea is to have a new running parallel thread that processes a command when you click the button Execute. We keep iterating through the lines of stdout and inserting them into the Text widget.

It seems to work for any command, but I am pretty sure that there are some problems and mistakes. If you guys more familiar with the modules I cited above see any serious problem with my code, or have any suggestions to improve it, I would definitely listen to you in order to improve this example.

Now, this is the code:

import tkinter as tk
from tkinter.scrolledtext import ScrolledText
import threading
from subprocess import Popen, PIPE


class Console(tk.Frame):

    """Simple console that can execute bash commands"""

    def __init__(self, master, *args, **kwargs):
        tk.Frame.__init__(self, master, *args, **kwargs)

        self.text_options = {"state": "disabled",
                             "bg": "black",
                             "fg": "#08c614",
                             "insertbackground": "#08c614",
                             "selectbackground": "#f01c1c"}

        self.text = ScrolledText(self, **self.text_options)

        # It seems not to work when Text is disabled...
        # self.text.bind("<<Modified>>", lambda: self.text.frame.see(tk.END))

        self.text.pack(expand=True, fill="both")

        # bash command, for example 'ping localhost' or 'pwd'
        # that will be executed when "Execute" is pressed
        self.command = ""  
        self.popen = None     # will hold a reference to a Popen object
        self.running = False  # True if the process is running

        self.bottom = tk.Frame(self)

        self.prompt = tk.Label(self.bottom, text="Enter the command: ")
        self.prompt.pack(side="left", fill="x")
        self.entry = tk.Entry(self.bottom)
        self.entry.bind("<Return>", self.start_thread)
        self.entry.bind("<Command-a>", lambda e: self.entry.select_range(0, "end"))
        self.entry.bind("<Command-c>", self.clear)
        self.entry.focus()
        self.entry.pack(side="left", fill="x", expand=True)

        self.executer = tk.Button(self.bottom, text="Execute", command=self.start_thread)
        self.executer.pack(side="left", padx=5, pady=2)
        self.clearer = tk.Button(self.bottom, text="Clear", command=self.clear)
        self.clearer.pack(side="left", padx=5, pady=2)
        self.stopper = tk.Button(self.bottom, text="Stop", command=self.stop)
        self.stopper.pack(side="left", padx=5, pady=2)

        self.bottom.pack(side="bottom", fill="both")

    def clear_text(self):
        """Clears the Text widget"""
        self.text.config(state="normal")
        self.text.delete(1.0, "end-1c")
        self.text.config(state="disabled")

    def clear_entry(self):
        """Clears the Entry command widget"""
        self.entry.delete(0, "end")

    def clear(self, event=None):
        """Does not stop an eventual running process,
        but just clears the Text and Entry widgets."""
        self.clear_entry()
        self.clear_text()

    def show(self, message):
        """Inserts message into the Text wiget"""
        self.text.config(state="normal")
        self.text.insert("end", message)
        self.text.see("end")
        self.text.config(state="disabled")

    def start_thread(self, event=None):
        """Starts a new thread and calls process"""
        self.stop()
        self.running = True
        self.command = self.entry.get()
        # self.process is called by the Thread's run method
        threading.Thread(target=self.process).start()

    def process(self):
        """Runs in an infinite loop until self.running is False""" 
        while self.running:
            self.execute()

    def stop(self):
        """Stops an eventual running process"""
        if self.popen:
            try:
                self.popen.kill()
            except ProcessLookupError:
                pass 
        self.running = False

    def execute(self):
        """Keeps inserting line by line into self.text
        the output of the execution of self.command"""
        try:
            # self.popen is a Popen object
            self.popen = Popen(self.command.split(), stdout=PIPE, bufsize=1)
            lines_iterator = iter(self.popen.stdout.readline, b"")

            # poll() return None if the process has not terminated
            # otherwise poll() returns the process's exit code
            while self.popen.poll() is None:
                for line in lines_iterator:
                    self.show(line.decode("utf-8"))
            self.show("Process " + self.command  + " terminated.\n\n")

        except FileNotFoundError:
            self.show("Unknown command: " + self.command + "\n\n")                               
        except IndexError:
            self.show("No command entered\n\n")

        self.stop()


if __name__ == "__main__":
    root = tk.Tk()
    root.title("Console")
    Console(root).pack(expand=True, fill="both")
    root.mainloop()

Answer 2

An improvement on @nbro's answer:

from tkinter.scrolledtext import ScrolledText
from subprocess import Popen, PIPE
from threading import Thread, Lock
import tkinter as tk


class Console(ScrolledText):
    """
    Simple console that can execute commands
    """

    def __init__(self, master, **kwargs):
        # The default options:
        text_options = {"state": "disabled",
                        "bg": "black",
                        "fg": "#08c614",
                        "selectbackground": "orange"}
        # Take in to account the caller's specified options:
        text_options.update(kwargs)
        super().__init__(master, **text_options)

        self.proc = None # The process
        self.text_to_show = "" # The new text that we need to display on the screen
        self.text_to_show_lock = Lock() # A lock to make sure that it's thread safe

        self.show_text_loop()

    def clear(self) -> None:
        """
        Clears the Text widget
        """
        super().config(state="normal")
        super().delete("0.0", "end")
        super().config(state="disabled")

    def show_text_loop(self) -> None:
        """
        Inserts the new text into the `ScrolledText` wiget
        """
        new_text = ""
        # Get the new text that needs to be displayed
        with self.text_to_show_lock:
            new_text = self.text_to_show.replace("\r", "")
            self.text_to_show = ""

        if len(new_text) > 0:
            # Display the new text:
            super().config(state="normal")
            super().insert("end", new_text)
            super().see("end")
            super().config(state="disabled")

        # After 100ms call `show_text_loop` again
        super().after(100, self.show_text_loop)

    def run(self, command:str) -> None:
        """
        Runs the command specified
        """
        self.stop()
        thread = Thread(target=self._run, daemon=True, args=(command, ))
        thread.start()

    def _run(self, command:str) -> None:
        """
        Runs the command using subprocess and appends the output
        to `self.text_to_show`
        """
        self.proc = Popen(command, shell=True, stdout=PIPE)

        try:
            while self.proc.poll() is None:
                text = self.proc.stdout.read(1).decode()
                with self.text_to_show_lock:
                    self.text_to_show += text

            self.proc = None
        except AttributeError:
            # The process ended prematurely
            pass

    def stop(self, event:tk.Event=None) -> None:
        """
        Stops the process.
        """
        try:
            self.proc.kill()
            self.proc = None
        except AttributeError:
            # No process was running
            pass

    def destroy(self) -> None:
        # Stop the process if the text widget is to be destroyed:
        self.stop()
        super().destroy()


if __name__ == "__main__":
    def run_command_in_entry(event:tk.Event=None):
        console.run(entry.get())
        entry.delete("0", "end")
        return "break"

    root = tk.Tk()
    root.title("Console")

    console = Console(root)
    console.pack(expand=True, fill="both")

    entry = tk.Entry(root, bg="black", fg="white",
                     insertbackground="white")
    entry.insert("end", "ping 8.8.8.8 -n 4")
    entry.bind("<Return>", run_command_in_entry)
    entry.pack(fill="x")

    root.mainloop()

The only difference between our answers is that I removed all of the widgets that were in the class except the ScrolledText and I made sure that I used tkinter in a thread safe way. Parts of tkinter aren't thread safe and aren't meant to be called from different threads (can raise errors). In the worse case scenario tkinter can crash without giving an error or a traceback.