使用Python中的事件监视系统

Question

我正在用Python创建一个项目,我想添加一个利用事件和事件处理程序的监视系统.我希望这个系统在整个项目中都可用.我有以下操作:

• 定义一个事件.该事件可以将一些数据作为参数.
• 定义监视器.监视器注册某个事件.多个监视器可以注册同一事件.我想创建不同类型的监视器,例如一个用于打印数据,一个用于创建包含数据的图表等.因此,监视器应该是一个类,能够保存它收集的所有数据,直到某些方法(例如打印,创建 - log,...)被调用.
• 为监视器 - 事件对定义事件处理程序.这定义了给定监视器如何响应给定事件.此操作主要是:将此数据添加到某个监视器类的实例的数据列表中.
• 一个通知函数,可以在事件发生时通知.这将触发为该事件注册的所有监视器的事件处理程序.理想情况下,notify函数应该可以从项目的任何位置调用.

我该如何创建这样的系统？有没有可以帮助我的图书馆？我特别想知道如何以这样的方式制作这个系统,使其在整个项目中透明化.

Answer 1

您可以使用四十行 Python 代码完成大部分您想做的事情。这是我自己的设计，我一直在使用。选择函数名称是为了使其成为 Qt 的“信号”和“槽”的直接替代品。

使用起来很简单。您创建一个PSignal. 您可以通过调用该connect方法来注册处理程序。处理程序可以是任何可调用的。当事件发生时，您可以通过调用函数发出信号（即通知事件）emit。每个注册的可调用对象都会在此时运行。调用的对象emit不知道，也不关心是否有人在听，或者如果有人在听会发生什么。

您还可以断开处理程序。

有很多调试代码，因为我发现否则某些错误可能很难追踪。

在您的问题中，您希望每个处理程序都是一个监视器，而在我的设计中，处理程序只是函数。但在我看来，您的“监视器”概念独立于事件/处理程序机制。您将必须编写函数来使您的应用程序运行，并且让这些函数调用您的监视器应该非常容易。

该代码使用 Python 3.3 进行了广泛的测试。

#! python3
import traceback

class PSignal:
    def __init__(self, debug=False):
        self.debug = debug
        self.__handlers = []

    def clear(self):
        """Deletes all the handlers."""
        self.__handlers.clear()

    def connect(self, f):
        """f is a python function."""
        if not callable(f):
            raise ValueError("Object {!r} is not callable".format(f))
        self.__handlers.append(f)
        if self.debug:
            print("PSIGNAL: Connecting", f, self.__handlers)

    def disconnect(self, f):
        for f1 in self.__handlers:
            if f == f1:
                self.__handlers.remove(f)
                return

    def emit(self, *x, **y):
        self._emit(*x, **y)

    def check_debug(self):
        if self.debug and self.__handlers:
            print("PSIGNAL: Signal emitted")
            traceback.print_stack()

    def _emit(self, *x, **y):
        self.check_debug()
        for f in self.__handlers:
            try:
                if self.debug:
                    print("PSIGNAL: emit", f, len(x), x, y)
                f(*x, **y)
            except Exception:
                print("PSIGNAL: Error in signal", f)
                traceback.print_exc()