在Tkinter GUI中实时输出运行进程

5
我正在尝试在Tkinter Python中创建GUI。我想要将工具的输出显示在我的Tkinter界面上。该工具在命令行中运行良好,但它是一个连续的扫描器。有点像连续的ping(我的意思是Linux中没有选项的ping命令)。
现在的问题是,由于ping的输出永远不会完成,因此我无法在Tkinter中打印输出。这也使我的应用程序冻结。我也不能在几秒钟后停止命令以显示输出。 在PHP中运行具有实时输出的进程 我发现上述链接对php有帮助,但我该如何将此代码转换为Python:

https://dev59.com/4XM_5IYBdhLWcg3wq1CF#6144213

这是一些示例代码,我想在Tkinter框架上显示。
#!/usr....

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

这在命令行上很好用,但在tkinter界面上没有输出。

3个回答

5
首先,我必须承认我对模块 subprocessthreading 不是非常熟悉,但我试图创建一个简单的控制台,允许您编写命令,并将其输出显示在一个 Text 小部件中。
基本思想是拥有一个新的并行运行的 线程,当您点击按钮 Execute 时处理命令。我们不断迭代 stdout 的每一行并将其插入到 Text 小部件中。
这似乎适用于任何命令,但我相信存在一些问题和错误。如果您更熟悉我引用的模块,并发现我的代码有严重问题或有任何改进意见,我肯定会倾听您的建议以改善此示例。
现在,这是代码:
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()

只是一个附录。当使用shell=True时,您希望保留对psutil的引用,因为例如,至少有2个甚至可能有3个进程需要终止。示例代码(在我所知道的所有操作系统上都可以工作)https://dev59.com/E3M_5IYBdhLWcg3wyWWt - user4171906
请原谅我拙劣的解释。我只是想让您知道它的存在。如果您曾经遇到需要关闭父进程和子进程的情况,这将节省大量寻找解决方案和头痛的时间。 - user4171906
不应该从不同的线程调用tkinter方法。这可能会导致tkinter崩溃。 - TheLizzard

1
一种对@nbro答案的改进:

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()

我们答案的唯一区别在于我移除了类中除ScrolledText之外的所有小部件,并确保我以线程安全的方式使用了tkinter。Tkinter的某些部分不是线程安全的,也不应该从不同的线程调用(可能会引发错误)。在最糟糕的情况下,tkinter可能会崩溃而不会给出错误或回溯。

-1
如果您将代码更改为以下内容,您将看到 ping 而不是“print x”显示在控制台上。
import subprocess
x = subprocess.call(["ping", "127.0.0.1"])
print "x is", x   ## or comment out this line

你需要使用管道和定期清空标准输出来实现你想要的效果。请参考Doug Hellmann的Python模块之一中的popen方法http://pymotw.com/2/subprocess/index.html#module-subprocess

网页内容由stack overflow 提供, 点击上面的
可以查看英文原文,
原文链接