Python 鸭子类型用于 pygame 中的 MVC 事件处理

Question

我和一个朋友一直在玩 pygame，并遇到了这个构建游戏的教程使用 pygame.我们真的很喜欢它如何将游戏分解为模型-视图-控制器系统，并以事件为中介，但代码大量使用了 isinstance 检查事件系统。

示例：

class CPUSpinnerController:
    ...
    def Notify(self, event):
        if isinstance( event, QuitEvent ):
            self.keepGoing = 0

这会产生一些极其不Python的代码。有人对如何改进有任何建议吗？或者实现 MVC 的替代方法？

---

这是我根据 @Mark-Hildreth 答案编写的一些代码（如何链接用户？）还有其他人有什么好的建议吗？在选择解决方案之前，我将把这个问题再保留一天左右。

class EventManager:
    def __init__(self):
        from weakref import WeakKeyDictionary
        self.listeners = WeakKeyDictionary()

    def add(self, listener):
        self.listeners[ listener ] = 1

    def remove(self, listener):
        del self.listeners[ listener ]

    def post(self, event):
        print "post event %s" % event.name
        for listener in self.listeners.keys():
            listener.notify(event)

class Listener:
    def __init__(self, event_mgr=None):
        if event_mgr is not None:
            event_mgr.add(self)

    def notify(self, event):
        event(self)


class Event:
    def __init__(self, name="Generic Event"):
        self.name = name

    def __call__(self, controller):
        pass

class QuitEvent(Event):
    def __init__(self):
        Event.__init__(self, "Quit")

    def __call__(self, listener):
        listener.exit(self)

class RunController(Listener):
    def __init__(self, event_mgr):
        Listener.__init__(self, event_mgr)
        self.running = True
        self.event_mgr = event_mgr

    def exit(self, event):
        print "exit called"
        self.running = False

    def run(self):
        print "run called"
        while self.running:
            event = QuitEvent()
            self.event_mgr.post(event)

em = EventManager()
run = RunController(em)
run.run()

这是使用@Paul 的示例的另一个构建 - 非常简单！

class WeakBoundMethod:
    def __init__(self, meth):
        import weakref
        self._self = weakref.ref(meth.__self__)
        self._func = meth.__func__

    def __call__(self, *args, **kwargs):
        self._func(self._self(), *args, **kwargs)

class EventManager:
    def __init__(self):
        # does this actually do anything?
        self._listeners = { None : [ None ] }

    def add(self, eventClass, listener):
        print "add %s" % eventClass.__name__
        key = eventClass.__name__

        if (hasattr(listener, '__self__') and
            hasattr(listener, '__func__')):
            listener = WeakBoundMethod(listener)

        try:
            self._listeners[key].append(listener)
        except KeyError:
            # why did you not need this in your code?
            self._listeners[key] = [listener]

        print "add count %s" % len(self._listeners[key])

    def remove(self, eventClass, listener):
        key = eventClass.__name__
        self._listeners[key].remove(listener)

    def post(self, event):
        eventClass = event.__class__
        key = eventClass.__name__
        print "post event %s (keys %s)" % (
            key, len(self._listeners[key]))
        for listener in self._listeners[key]:
            listener(event)

class Event:
    pass

class QuitEvent(Event):
    pass

class RunController:
    def __init__(self, event_mgr):
        event_mgr.add(QuitEvent, self.exit)
        self.running = True
        self.event_mgr = event_mgr

    def exit(self, event):
        print "exit called"
        self.running = False

    def run(self):
        print "run called"
        while self.running:
            event = QuitEvent()
            self.event_mgr.post(event)

em = EventManager()
run = RunController(em)
run.run()

Answer 1

处理事件的一种更简洁的方法（而且速度更快，但可能消耗更多内存）是在代码中使用多个事件处理函数。沿着这些思路：

所需的接口

class KeyboardEvent:
    pass

class MouseEvent:
    pass

class NotifyThisClass:
    def __init__(self, event_dispatcher):
        self.ed = event_dispatcher
        self.ed.add(KeyboardEvent, self.on_keyboard_event)
        self.ed.add(MouseEvent, self.on_mouse_event)

    def __del__(self):
        self.ed.remove(KeyboardEvent, self.on_keyboard_event)
        self.ed.remove(MouseEvent, self.on_mouse_event)

    def on_keyboard_event(self, event):
        pass

    def on_mouse_event(self, event):
        pass

这里，__init__ 方法接收一个 EventDispatcher 作为参数。 EventDispatcher.add 函数现在采用您感兴趣的事件类型和侦听器。

这对效率有好处，因为侦听器只会被其感兴趣的事件调用。它还会在 EventDispatcher 本身内部产生更通用的代码：

EventDispatcher

class EventDispatcher:
    def __init__(self):
        # Dict that maps event types to lists of listeners
        self._listeners = dict()

    def add(self, eventcls, listener):
        self._listeners.setdefault(eventcls, list()).append(listener)

    def post(self, event):
        try:
            for listener in self._listeners[event.__class__]:
                listener(event)
        except KeyError:
            pass # No listener interested in this event

实现是这个实现的一个问题。在 NotifyThisClass 中，您执行以下操作：

self.ed.add(KeyboardEvent, self.on_keyboard_event)

问题在于 self.on_keyboard_event：它是一个绑定方法，您将其传递给 EventDispatcher< /代码>。绑定方法保存对 self 的引用；这意味着只要EventDispatcher有绑定方法，self就不会被删除。

WeakBoundMethod

您需要创建一个 WeakBoundMethod 类，该类仅保存对 self 的弱引用（我发现您已经了解弱引用），以便 EventDispatcher code> 不会阻止删除 self。

另一种方法是在删除对象之前调用一个 NotifyThisClass.remove_listeners 函数，但这并不是最干净的解决方案，而且我发现它很容易出错（很容易忘记这样做）。

WeakBoundMethod 的实现看起来像这样：

class WeakBoundMethod:
    def __init__(self, meth):
        self._self = weakref.ref(meth.__self__)
        self._func = meth.__func__

    def __call__(self, *args, **kwargs):
        self._func(self._self(), *args, **kwargs)

这是我发布的一个更强大的实现在 CodeReview 上，这里是如何使用该类的示例：

from weak_bound_method import WeakBoundMethod as Wbm

class NotifyThisClass:
    def __init__(self, event_dispatcher):
        self.ed = event_dispatcher
        self.ed.add(KeyboardEvent, Wbm(self.on_keyboard_event))
        self.ed.add(MouseEvent, Wbm(self.on_mouse_event))

Connection 对象（可选）

从管理器/调度程序中删除侦听器时，而不是使EventDispatcher 无需搜索侦听器，直到找到正确的事件类型，然后搜索列表，直到找到正确的侦听器，您可能会得到如下内容：

class NotifyThisClass:
    def __init__(self, event_dispatcher):
        self.ed = event_dispatcher
        self._connections = [
            self.ed.add(KeyboardEvent, Wbm(self.on_keyboard_event)),
            self.ed.add(MouseEvent, Wbm(self.on_mouse_event))
        ]

这里 EventDispatcher.add 返回一个 Connection 对象，该对象知道它位于 EventDispatcher 列表字典中的位置。当 NotifyThisClass 对象被删除时，self._connections 也会被删除，它将调用 Connection.__del__，这将从 >事件调度程序。

这可以使您的代码更快、更易于使用，因为您只需显式添加函数，它们会自动删除，但是否要执行此操作由您决定。如果您这样做，请注意 EventDispatcher.remove 不应再存在。

Answer 2

我过去偶然发现了 SJ Brown 的游戏制作教程。这是一个很棒的页面，是我读过的最好的页面之一。但是，和您一样，我不喜欢对 isinstance 的调用，也不喜欢所有侦听器接收所有事件的事实。

首先， isinstance 比检查两个字符串是否相等要慢，因此我最终在事件中存储了一个名称并测试该名称而不是类。但是，带有 if 电池的通知函数仍然让我感到痒痒，因为它感觉像是浪费时间。我们可以在这里做两个优化：

1. 大多数侦听器只对几种类型的事件感兴趣。出于性能原因，当发布 QuitEvent 时，只应通知对此感兴趣的侦听器。事件管理器跟踪哪个侦听器想要侦听哪个事件。
2. 然后，为了避免在单个 notify 方法中执行大量 if 语句，我们将为每种类型的事件提供一个方法。

示例：

class GameLoopController(...):
    ...
    def onQuitEvent(self, event):
        # Directly called by the event manager when a QuitEvent is posted.
        # I call this an event handler.
        self._running = False

因为我希望开发人员尽可能少地键入，所以我做了以下事情：

当侦听器注册到事件管理器时，事件管理器会扫描侦听器的所有方法。当一个方法以“on”（或您喜欢的任何前缀）开头时，它会查看其余部分（“QuitEvent”）并将该名称绑定到该方法。稍后，当事件管理器抽取其事件列表时，它会查看事件类名称：“QuitEvent”。它知道该名称，因此可以直接调用所有相应的事件处理程序。开发人员只需添加 onWhateverEvent 方法即可使其正常工作。

它有一些缺点：

1. 如果我在处理程序的名称中输入错误（“onRunPhysicsEvent”
   而不是“onPhysicsRanEvent”，例如“）然后我的处理程序将
   永远不会被叫到，我想知道为什么。但我知道诀窍所以我
   不要奇怪为什么很长。
2. 我无法在侦听器之后添加事件处理程序
   挂号的。我必须取消注册并重新注册。确实，
   仅在注册期间扫描事件处理程序。然后
   再说一次，无论如何我从来没有必要这样做，所以我不会错过它。

尽管有这些缺点，我更喜欢它，而不是让侦听器的构造函数明确地向事件管理器解释它想要密切关注这个、这个、这个和这个事件。无论如何，它的执行速度是相同的。

第二点：

在设计我们的事件管理器时，我们要小心。通常，侦听器将通过创建-注册或取消注册-销毁侦听器来响应事件。这种情况经常发生。如果我们不考虑这一点，那么我们的游戏可能会因RuntimeError：字典在迭代期间更改大小而中断。您建议的代码会迭代字典的副本，以便防止爆炸；但它的后果需要注意：
- 由于事件而注册的侦听器将不会收到该事件。
- 由于事件而未注册的听众仍会收到该消息
事件。
但我从来没有发现这是一个问题。

我自己为我正在开发的游戏实现了这一点。我可以将您链接到我就该主题撰写的两篇半文章：

• http://niriel.wordpress.com/2011/08/06/who-controls-the-controllers/
• http://niriel.wordpress.com/2011/ 08/08/the-event-management-is-in-place/
• http://niriel.wordpress.com/2011/08/ 11/the-first-screenshot-of-infiniworld/

我的github帐户的链接将直接带您到相关部分的源代码。如果您等不及了，请看以下内容：https://github。 com/Niriel/Infiniworld/blob/v0.0.2/src/evtman.py 。在那里，您会看到我的事件类的代码有点大，但每个继承的事件都是在两行中声明的：基事件类使您的生活变得轻松。

因此，这一切都可以使用 python 的自省机制来实现，并且使用方法是像任何其他可以放入字典中的对象一样的对象这一事实。我认为这很Pythony:)。

Answer 3

为每个事件提供一个方法（甚至可能使用 __call__ ），并传入 Controller 对象作为参数。然后，“call”方法应该调用控制器对象。例如...

class QuitEvent:
    ...
    def __call__(self, controller):
        controller.on_quit(self) # or possibly... controller.on_quit(self.val1, self.val2)

class CPUSpinnerController:
    ...
    def on_quit(self, event):
        ...

无论您使用什么代码将事件路由到控制器，都将使用正确的控制器调用 __call__ 方法。

Answer 4

我偶然发现了同样的问题（几乎十年后！），这是我一直在使用的实现，以便 EventManager 仅通知侦听器的子集。
它基于 defaultdict：EventManager 的 _listeners 属性是 WeakKeyDictionary() 的 defaultdict。
事件都是继承自一个空的抽象 Event 类，因此侦听器可以只关注他们想要侦听的某些事件类。
下面是一个简单的代码来了解其背后的想法：

from collections import defaultdict
from weakref import WeakKeyDictionary

class Event:
    def __init__(self):
       pass

class KeyboardEvent(Event): # for instance, a keyboard event class with the key pressed
    def __init__(self, key):
        self._key = key

class EventManager:
    def __init__(self):
        self._listeners = defaultdict(lambda: WeakKeyDictionary())

    def register_listener(self, event_types, listener):
        for event_type in event_types:
            self._listeners[event_type][listener] = 1

    def unregister_listener(self, listener):
        for event_type in self._listeners:
            self._listeners[event_type].pop(listener, None)

    def post_event(self, event):
        for listener in self._listeners[event.__class__]:
            listener.notify(event)

注册时，侦听器告诉事件管理器它想要通知哪种事件类型。
发布事件时，事件管理器只会通知注册为该类型事件通知的侦听器。
当然，这段代码的范围比 @Paul Manta 提出的非常通用（而且非常优雅）的解决方案要小得多，但就我而言，它有助于消除对 isinstance 和其他的一些重复调用检查的同时尽可能保持事情简单。
这样做的缺点之一是所有类型的事件都必须是某个类的对象，但在 OO python 中，这应该是正确的方法。