Skip to content

Instantly share code, notes, and snippets.

@gentimouton
Created September 8, 2012 18:20
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save gentimouton/3678274 to your computer and use it in GitHub Desktop.
Save gentimouton/3678274 to your computer and use it in GitHub Desktop.
Example of an event manager aware of MVC for games
from collections import defaultdict
from time import sleep
import logging
class TickEvent:
pass
# components subscribed to TickEvent as "input" will be notified first,
# then components with priority as "model", and finally the rest (views).
PRIO_TICK_INPUT = 1
PRIO_TICK_MODEL = 2
class EventManager:
""" Coordinate the communication between components. """
def __init__(self):
""" Make 4 sets of callbacks:
1 for input components(= controllers), they have to be ticked first,
1 for the model components, ticked just after,
and 1 for the rendering/view components, ticked at the end.
Last list = non-tick events.
"""
self._c_callbacks = set() # controller callbacks for tick event
self._tc_callbacks = set() # temporary controller callbacks
self._m_callbacks = set() # model
self._tm_callbacks = set()
self._v_callbacks = set() # view
self._tv_callbacks = set()
self._callbacks = defaultdict(set) # map non-tick events to their callbacks
self._tmp_callbacks = defaultdict(set)
def subscribe(self, ev_class, callback, priority=None):
""" Register a callback for a particular event. """
if ev_class == TickEvent:
if priority == PRIO_TICK_INPUT:
self._tc_callbacks.add(callback)
elif priority == PRIO_TICK_MODEL:
self._tm_callbacks.add(callback)
else: # views/renderers
self._tv_callbacks.add(callback)
else:# non-tick events
self._callbacks[ev_class].add(callback)
def publish(self, event):
""" Publish an event. """
ev_class = event.__class__
if ev_class == TickEvent:
# iterate over the callbacks: new callbacks will be added to _tc_...
for cb in self._c_callbacks:
cb(event)
for cb in self._m_callbacks:
cb(event)
for cb in self._v_callbacks:
cb(event)
# iterate over the new callbacks
while self._tc_callbacks or self._tm_callbacks or self._tv_callbacks:
# incorporate the new callbacks
self._c_callbacks = self._c_callbacks.union(self._tc_callbacks)
self._m_callbacks = self._m_callbacks.union(self._tm_callbacks)
self._v_callbacks = self._v_callbacks.union(self._tv_callbacks)
# copy the new callback sets
c_cbs_cpy = self._tc_callbacks # controller callbacks copy
m_cbs_cpy = self._tm_callbacks
v_cbs_cpy = self._tv_callbacks
# empty the new callback sets
self._tc_callbacks = set()
self._tm_callbacks = set()
self._tv_callbacks = set()
# iterate over the copies
for cb in c_cbs_cpy:
cb(event)# might add new subscriber callbacks to self._t?_callbacks
for cb in m_cbs_cpy:
cb(event)
for cb in v_cbs_cpy:
cb(event)
else: # non-tick event
for cb in self._callbacks[ev_class]:
cb(event)
# TODO: non-tick events may have the same problem as tick events
# So there may need to be the same structure of set-copies.
class Clock:
def __init__(self, em):
self._em = em
self.elapsed_frames = 0
def start(self):
while self.elapsed_frames < 5: # run for 5 frames, then stop
logging.info('---- Clock tick %d' % self.elapsed_frames)
self._em.publish(TickEvent()) # tick all registered components
self.elapsed_frames += 1
sleep(0.1) # in seconds
logging.info('Stopped the clock')
return
###########################################################
class DummyCtrler:
def __init__(self, em):
logging.info('Created controller')
self._em = em
em.subscribe(TickEvent, self.on_tick, PRIO_TICK_INPUT)
def on_tick(self, ev):
logging.info('Ticked controller')
####################################################
class DummyModel:
def __init__(self, em):
logging.info('Created model')
self._em = em
self.boss = None
em.subscribe(TickEvent, self.on_tick, PRIO_TICK_MODEL)
def on_tick(self, ev):
""" Create a boss if I dont have any yet. """
logging.info('Ticked model')
if not self.boss:
self.boss = BossModel(self._em)
class BossModel:
def __init__(self, em):
logging.info('Created boss')
self.minion = None
self._em = em
em.subscribe(TickEvent, self.on_tick, PRIO_TICK_MODEL)
def on_tick(self, ev):
logging.info('Ticked boss')
if not self.minion:
self.minion = MinionModel(self._em)
class MinionModel:
def __init__(self, em):
logging.info('Created minion')
# no need to tick it
##################################################
class DummyView:
def __init__(self, em):
logging.info('Created view')
self._em = em
em.subscribe(TickEvent, self.on_tick)
def on_tick(self, ev):
logging.info('Ticked view')
##################################################
def main():
logging.basicConfig(level=logging.INFO)
em = EventManager()
clock = Clock(em)
ctrler = DummyCtrler(em)
model = DummyModel(em)
view = DummyView(em)
clock.start()
if __name__ == "__main__":
main()
@gentimouton
Copy link
Author

Console output:

INFO:root:Created controller
INFO:root:Created model
INFO:root:Created view
INFO:root:---- Clock tick 0
INFO:root:Ticked controller
INFO:root:Ticked model
INFO:root:Created boss
INFO:root:Ticked view
INFO:root:Ticked boss
INFO:root:Created minion
INFO:root:---- Clock tick 1
INFO:root:Ticked controller
INFO:root:Ticked model
INFO:root:Ticked boss
INFO:root:Ticked view
...

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment