如何在Kivy中实现EventDispatcher？

Question

朋友们，美好的一天！

Kivy 文档对 EventDispatcher 的用法有点不清楚，我无法实现它。

简单描述： 当进入屏幕时，在 on_enter() 方法中，创建一个对象 Task() ，它将执行一个线程函数。该函数将发出（发布）一条消息及其结果。一些 UI 元素（按钮、Labals、自定义小部件）订阅该消息，并将调用接受发布结果的成员函数并执行一些操作。

class Reciever(Button):
    def __init__(self, text, **kwargs):
        super().__init__(**kwargs)
        self.text = text

    def on_process_task_result(self, result):
        print(f"{self} recieved result: {result}")

r1 = Reciever("r1")
r1.register_event_type("on_process_task_result")

 class Task(EventDispatcher):
    def __init__(self, **kwargs):
        super(Task, self).__init__(**kwargs)

    def execute(self, sender):
        # Simulating task execution and generating a result
        result = random.choice(["ok", "fail", "repeat"])
        print("Task result:", result)
        self.dispatch('on_process_task_result', result=result)

总的来说，我认为自定义观察者实现是多余的。当前轮询的实现是一团糟。这就是返工的原因。

有人可以提供一个如何在 Kivy 中实现它的示例吗？

Answer 1

我选择了另一种乍一看似乎不直观的方法。基本思想：在自定义发送者类中使用 Kivy StringProperty，添加接收者，每次 StringProperty 发生变化时触发接收者方法。

class MyScreenClass(Screen):
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.app = App.get_running_app()
        self.task = Task("check_presence")        
        # Trigger task execution manually for demonstration purpose
        self.ids.trigger_btn.bind(on_press=self.task.execute_task)
        
        r1 = RecieverBtn("r1")
        r2 = RecieverBtn("r2")
        l1 = RecieverLbl("l1")
        
        # What is the best approach to pass in subscribers?
        # - pass in widged reference?
        # - pass in widged id?
        # - Task() object gather all UI elements with a specific property?
        self.task.add_subscriber(r1)
        self.task.add_subscriber(r2)
        self.task.add_subscriber(l1)
        
        self.ids.scroll_view_content.add_widget(r1)
        self.ids.scroll_view_content.add_widget(r2)
        self.ids.scroll_view_content.add_widget(l1)

class RecieverBtn(Button):
    # Custom button that represents all my UI elements,
    # that should be notified about the Task() result 
    def __init__(self, text, **kwargs):
        super(RecieverBtn, self).__init__(**kwargs)
        self.text = text
        self.id = text

    def notify_about_task_results(self, sender, value):
        if value != "None":
            print(f"Reciever {self.id} notified by {sender.task_id} that value changed to {value}")

class RecieverLbl(Label):
    def __init__(self, text, **kwargs):
        super(RecieverLbl, self).__init__(**kwargs)
        self.text = text
        self.id = text

    def notify_about_task_results(self, sender, value):
        if value != "None":
            print(f"Reciever {self.id} notified by {sender.task_id} that value changed to {value}")

class Task(EventDispatcher):
    # Using a kivy Property to notify recievers
    # when this value changes. 
    # Do not declare Properties in __init__(), KeyError
    result = StringProperty("None")
    # Keep track of all subscribers
    subscribers = set()
    def __init__(self, task_id, **kwargs):
        super(Task, self).__init__(**kwargs)
        self.task_id = task_id
        
    def execute_task(self, sender):
        # Simulating task execution and generating a result
        self.result = random.choice(["ok", "fail", "repeat"])
        print("Task result:", self.result)
        self.result = "None"

    def add_subscriber(self, ui_widget):
        # When result StringProperty changes, trigger callback function
        if ui_widget not in self.subscribers:
            self.subscribers.add(ui_widget)
            self.bind(result=ui_widget.notify_about_task_results)