在 Python 中的线程期间更新 Tkinter 标签

Question

我目前正在尝试编写由 Raspberry Pi 控制的机械臂。

到目前为止一切正常，除了一件事，我已经用谷歌搜索并尝试了很多小时，但找不到有效的解决方案。

对于机器人手臂的运动，必须使用螺纹“同时”运行所有电机（工作正常）。

我遇到的问题是，我需要更新一个标签，该标签在轴（电机）完成运动后立即显示当前角度，但其他电机仍在运行（线程）。

经过大量研究，我认为我通过使用队列和 Tkinter 后置方法找到了解决方案。但它仍然不起作用，因为标签文本只有在所有线程终止后才会更新。

我编写了一个示例代码，我想在其中获取电机“一”的标签更新，这将在电机“二”（500 次迭代）之前完成其 for 循环（100 次迭代）。我预计一旦电机 1 达到目标，而电机 2 仍在运行，标签就会更新。

但是虽然我使用了 after-method，但它仍然要等到电机 2 完成后才更新标签。

希望你能帮助我！

from tkinter import *
import threading
import time
from queue import * 


class StepperMotors:

    def __init__(self, root):
        self.root = root

        self.start_btn = Button(root, text="Start", command=lambda:self.start_movement())
        self.start_btn.config(width = 10)
        self.start_btn.grid(row=1,column=1)

        self.label_one = Label(root, text='')
        self.label_one.config(width = 10)
        self.label_one.grid(row=2, column=1)

        self.label_two = Label(root, text='')
        self.label_two.config(width = 10)
        self.label_two.grid(row=3, column=1)

    def start_movement(self):
        self.thread_queue = Queue()
        self.root.after(100, self.wait_for_finish) 

        thread_one = threading.Thread(target=self.motor_actuation, args=(1,100))
        thread_two = threading.Thread(target=self.motor_actuation, args=(2,500))

        thread_one.start()
        thread_two.start()

        thread_one.join()
        thread_two.join()

    def motor_actuation(self, motor, iterations):  
        for i in range(iterations):
            i = i+1  
            update_text = str(motor) + " " + str(i) + "\n"
            print(update_text)
            time.sleep(0.01)

        self.thread_queue.put(update_text)

    def wait_for_finish(self):
        try:      
            self.text = self.thread_queue.get()  
            self.label_one.config(text=self.text)  

        except self.thread_queue.empty():
            self.root.after(100, self.wait_for_finish)

if __name__ == "__main__":
    root = Tk()
    root.title("test")
    stepper = StepperMotors(root)
    root.mainloop()

Answer 1

最好使用非阻塞的守护线程。

另外，最好有一个separation of concerns：一个机器人（或机械臂）可以是一个有自己生命周期的对象：一个守护线程。同上，您可以定义一个“LabelUpdater”来读取机器人的状态并更新标签。

让我们定义一个机器人：

• 它在应用程序初始化时创建，并在用户单击“开始”按钮时运行，
• 机器人在应用级多线程队列中移动并报告其角度，

class Robot(threading.Thread):

    def __init__(self, name: str, label_queue: queue.Queue, end_pos: int):
        super().__init__(name=name)
        self.daemon = True
        self.label_queue = label_queue
        self.end_pos = end_pos

    def run(self) -> None:
        for angle in range(self.end_pos):
            self.label_queue.put(angle)
            time.sleep(0.01)

让我们定义一个 LabelUpdater：

• 它在应用程序初始化时创建并永久运行（即使机器人没有运行，它也可以观察它）。
• 它读取机器人队列（例如每秒以避免闪烁）并更新标签

class LabelUpdater(threading.Thread):
    def __init__(self, name: str, label_queue: queue.Queue, root_app: tkinter.Tk, variable: tkinter.Variable):
        super().__init__(name=name)
        self.daemon = True
        self.label_queue = label_queue
        self.root_app = root_app
        self.variable = variable

    def run(self) -> None:
        # run forever
        while True:
            # wait a second please
            time.sleep(1)
            # consume all the queue and keep only the last message
            last_msg = None
            while True:
                try:
                    msg = self.label_queue.get(block=False)
                except queue.Empty:
                    break
                last_msg = msg
                self.label_queue.task_done()
            if last_msg:
                self.variable.set(last_msg)

然后，主应用程序应该定义：

• 2 个多线程队列：每个标签一个，
• 2 tkinter.StringVar 将被更新的变量，
• 机器人和标签更新器，
• 两个更新程序已启动，并将永远运行。

class StepperMotors:
    def __init__(self, root):
        self.root = root
        self.label_one_queue = queue.Queue()
        self.label_two_queue = queue.Queue()

        self.start_btn = tkinter.Button(root, text="Start", command=lambda: self.start_movement())
        self.start_btn.config(width=10)
        self.start_btn.grid(row=1, column=1)

        self.text_one = tkinter.StringVar()
        self.text_one.set("one")
        self.label_one = tkinter.Label(root, textvariable=self.text_one)
        self.label_one.config(width=10)
        self.label_one.grid(row=2, column=1)

        self.text_two = tkinter.StringVar()
        self.text_two.set("two")
        self.label_two = tkinter.Label(root, textvariable=self.text_two)
        self.label_two.config(width=10)
        self.label_two.grid(row=3, column=1)

        self.robot_one = Robot("robot_one", self.label_one_queue, 100)
        self.robot_two = Robot("robot_two", self.label_two_queue, 500)

        self.updater_one = LabelUpdater("updater_one", self.label_one_queue, self.root, self.text_one)
        self.updater_two = LabelUpdater("updater_two", self.label_two_queue, self.root, self.text_two)
        self.updater_one.start()
        self.updater_two.start()

    def start_movement(self):
        self.robot_one.start()
        self.robot_two.start()

当然，您需要一个标志或其他东西来检查每个机器人是否尚未运行。