python - 在 Python 中使用事件监控系统

Question

我正在用 Python 创建一个项目，我想添加一个使用事件和事件处理程序的监控系统。我希望这个系统在整个项目中都可用。我想到了以下操作:

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

如何创建这样的系统？有没有图书馆可以帮助我解决这个问题？我特别想知道如何使这个系统在整个项目中透明可用。

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