如何将进度条连接到函数?

8 投票
3 回答
25953 浏览
提问于 2025-04-17 18:35

我正在尝试把一个进度条连接到我项目中的一个功能上。

这是我目前的进展,但我很确定它没有任何作用:

def main():
    pgBar.start()
    function1()
    function2()
    function3()
    function4()
    pgBar.stop()

这是我创建进度条的代码,希望能帮到你:

pgBar = ttk.Progressbar(window, orient = HORIZONTAL, length=300, mode = "determinate")
pgBar.place(x=45, y=130)

我做了一些研究,了解到当运行一个函数时,tkinter窗口会冻结。有没有办法在主函数内部调用的每个函数结束时“解冻”窗口呢?

3 个回答

2

你需要使用: self.pgBar.step(x) 这里的 'x' 是你想让进度条增加的数量。为了让这个进度条在你的界面上更新,你需要在每次使用 self.pgBar.step(x) 之后加上 self.window.update_idletasks()

17

要理解“冻结”这个问题,首先得明白 mainloop() 是什么。调用这个方法会启动 tkinter 的事件循环。这个循环是由主线程负责的。所以,当你在主线程中运行一个耗时的函数时,它会干扰这个事件循环。为了避免这种情况,你可以使用一个额外的 Thread 来运行你的函数。建议不要让这个额外的线程访问 tkinter 的对象。Allen B.Taylor,mtTkinter 的作者,提到:

问题的根源在于 _tkinter 模块试图通过轮询技术控制主线程,以处理来自其他线程的调用。如果成功了,那就没问题。如果失败了(比如超时),应用程序就会收到一个异常,提示:“RuntimeError: main thread is not in main loop”。

你可以让额外的线程把信息放进一个 Queue 中。然后在主循环中,使用 after() 方法,每隔 x 毫秒检查一次这个 Queue

首先,决定你想要的 Progressbar最大值 是多少。
这个最大值表示 Progressbar 需要多少单位才能填满。例如,你可以设置 maximum=4,然后在每个函数执行完后把相应的值放入 Queue。主线程可以从 Queue 中取出这些值,利用 tkinter.IntVar() 来更新进度。(注意,如果你使用 progbar.step(),Progressbar 在结束时会重置为 0,而不是达到 4。)

下面是如何使用 tkinter.IntVar() 和 Progressbar 的简单示例:

int_var = tkinter.IntVar()
pb_instance = ttk.Progressbar(root, maximum=4)
pb_instance['variable'] = int_var
pb_instance.pack()
# completely fill the Progressbar
int_var.set(4)
# get the progress value
x = int_var.get()

这是一个基于你自己的例子(把“main”函数重命名为“arbitrary”):

import time
import threading

try: import tkinter
except ImportError:
    import Tkinter as tkinter
    import ttk
    import Queue as queue
else:
    from tkinter import ttk
    import queue

class GUI_Core(object):

    def __init__(self):
        self.root = tkinter.Tk()

        self.int_var = tkinter.IntVar()
        progbar = ttk.Progressbar(self.root, maximum=4)
        # associate self.int_var with the progress value
        progbar['variable'] = self.int_var
        progbar.pack()

        self.label = ttk.Label(self.root, text='0/4')
        self.label.pack()

        self.b_start = ttk.Button(self.root, text='Start')
        self.b_start['command'] = self.start_thread
        self.b_start.pack()

    def start_thread(self):
        self.b_start['state'] = 'disable'
        self.int_var.set(0) # empty the Progressbar
        self.label['text'] = '0/4'
        # create then start a secondary thread to run arbitrary()
        self.secondary_thread = threading.Thread(target=arbitrary)
        self.secondary_thread.start()
        # check the Queue in 50ms
        self.root.after(50, self.check_que)

    def check_que(self):
        while True:
            try: x = que.get_nowait()
            except queue.Empty:
                self.root.after(25, self.check_que)
                break
            else: # continue from the try suite
                self.label['text'] = '{}/4'.format(x)
                self.int_var.set(x)
                if x == 4:
                    self.b_start['state'] = 'normal'
                    break


def func_a():
    time.sleep(1) # simulate some work

def func_b():
    time.sleep(0.3)

def func_c():
    time.sleep(0.9)

def func_d():
    time.sleep(0.6)

def arbitrary():
    func_a()
    que.put(1)
    func_b()
    que.put(2)
    func_c()
    que.put(3)
    func_d()
    que.put(4)

que = queue.Queue()
gui = GUI_Core() # see GUI_Core's __init__ method
gui.root.mainloop()

如果你只想让用户知道有活动发生,
你可以把 Progressbar 的 模式 设置为 'indeterminate'
在这个模式下,指示器会来回跳动(速度与最大值有关)。

然后你可以在启动额外线程之前直接调用 Progressbar 的 start() 方法;
secondary_thread.is_alive() 返回 False 后,再调用 stop()

下面是一个示例:

import time
import threading

try: import tkinter
except ImportError:
    import Tkinter as tkinter
    import ttk
else: from tkinter import ttk

class GUI_Core(object):

    def __init__(self):
        self.root = tkinter.Tk()

        self.progbar = ttk.Progressbar(self.root)
        self.progbar.config(maximum=4, mode='indeterminate')
        self.progbar.pack()

        self.b_start = ttk.Button(self.root, text='Start')
        self.b_start['command'] = self.start_thread
        self.b_start.pack()

    def start_thread(self):
        self.b_start['state'] = 'disable'
        self.progbar.start()
        self.secondary_thread = threading.Thread(target=arbitrary)
        self.secondary_thread.start()
        self.root.after(50, self.check_thread)

    def check_thread(self):
        if self.secondary_thread.is_alive():
            self.root.after(50, self.check_thread)
        else:
            self.progbar.stop()
            self.b_start['state'] = 'normal'


def func_a():
    time.sleep(1) # simulate some work

def func_b():
    time.sleep(0.3)

def func_c():
    time.sleep(0.9)

def func_d():
    time.sleep(0.6)

def arbitrary():
    func_a()
    func_b()
    func_c()
    func_d()

gui = GUI_Core()
gui.root.mainloop()

Progressbar 参考

19

因为 tkinter 是 单线程 的,所以你需要另开一个线程来执行你的 main 函数,这样才能避免界面卡住。一个常见的方法是,工作线程把消息放到一个同步对象里(比如一个 Queue),然后 GUI 部分读取这些消息,更新进度条。

下面的代码是基于 ActiveState 上一个详细的 例子

import tkinter as tk
from tkinter import ttk
import threading
import queue
import time


class App(tk.Tk):

    def __init__(self):
        tk.Tk.__init__(self)
        self.queue = queue.Queue()
        self.listbox = tk.Listbox(self, width=20, height=5)
        self.progressbar = ttk.Progressbar(self, orient='horizontal',
                                           length=300, mode='determinate')
        self.button = tk.Button(self, text="Start", command=self.spawnthread)
        self.listbox.pack(padx=10, pady=10)
        self.progressbar.pack(padx=10, pady=10)
        self.button.pack(padx=10, pady=10)

    def spawnthread(self):
        self.button.config(state="disabled")
        self.thread = ThreadedClient(self.queue)
        self.thread.start()
        self.periodiccall()

    def periodiccall(self):
        self.checkqueue()
        if self.thread.is_alive():
            self.after(100, self.periodiccall)
        else:
            self.button.config(state="active")

    def checkqueue(self):
        while self.queue.qsize():
            try:
                msg = self.queue.get(0)
                self.listbox.insert('end', msg)
                self.progressbar.step(25)
            except Queue.Empty:
                pass


class ThreadedClient(threading.Thread):

    def __init__(self, queue):
        threading.Thread.__init__(self)
        self.queue = queue

    def run(self):
        for x in range(1, 5):
            time.sleep(2)
            msg = "Function %s finished..." % x
            self.queue.put(msg)


if __name__ == "__main__":
    app = App()
    app.mainloop()

因为我觉得 ActiveState 上的 原始例子有点乱(ThreadedClientGuiPart 之间的耦合度很高,而且从 GUI 控制线程启动的时机也没有那么简单),所以我对它进行了重构,并且添加了一个按钮来启动新的线程。

撰写回答