New subsystem API

ppb/engine.py

@@ -36,6 +36,7 @@ def __init__(self, first_scene: Type, *,
         self.event_extensions = defaultdict(dict)
         self.running = False
         self.entered = False
+        self.last_heartbeat = None
 
         # Systems
         self.systems_classes = systems
@@ -80,12 +81,20 @@ def run(self):
 
     def start(self):
         self.running = True
+        self.last_heartbeat = time.monotonic()
         self.activate({"scene_class": self.first_scene,
                        "kwargs": self.scene_kwargs})
 
     def main_loop(self):
         while self.running:
             time.sleep(0)
+            now = time.monotonic()
+            self.signal(events.Heartbeat(now - self.last_heartbeat))
+            self.last_heartbeat = now
+            # Okay, doing this twice is hacky as hell, but fine while I remove
+            # the old system.
+            while self.events:
+                self.publish()
             for system in self.systems:
                 for event in system.activate(self):
                     self.signal(event)

ppb/events.py

@@ -271,6 +271,15 @@ class StopScene:
     scene: Scene = None
 
 
+@dataclass
+class Heartbeat:
+    """
+    An engine plumbing event to pump timing information to subsystems.
+    """
+    time_delta: float
+    scene: Scene = None
+
+
 @dataclass
 class Update:
     """

ppb/testutils.py

@@ -2,24 +2,26 @@
 from typing import Callable
 
 from ppb.engine import GameEngine
+from ppb.events import Heartbeat
 from ppb.events import Quit
 from ppb.systems import System
 
 
 class Failer(System):
 
     def __init__(self, *, fail: Callable[[GameEngine], bool], message: str,
-                 run_time: float=1, **kwargs):
+                 run_time: float=1, engine, **kwargs):
         super().__init__(**kwargs)
         self.fail = fail
         self.message = message
         self.start = time.monotonic()
         self.run_time = run_time
+        self.engine = engine
 
-    def activate(self, engine):
+    def on_heartbeat(self, heartbeat_event: Heartbeat, signal):
         if time.monotonic() - self.start > self.run_time:
             raise AssertionError("Test ran too long.")
-        if self.fail(engine):
+        if self.fail(self.engine):
             raise AssertionError(self.message)
         return ()
 

tests/test_engine.py

@@ -98,8 +98,9 @@ def fail(engine):
         if parent.count > 0 and engine.current_scene != parent:
             return True
 
-    failer = Failer(fail=fail, message="ParentScene should not be counting while a child exists.")
-    engine = GameEngine(ParentScene, systems=[Updater(time_step=0.001), failer])
+    engine = GameEngine(ParentScene,
+                        systems=[Updater(time_step=0.001), Failer], fail=fail,
+                        message="ParentScene should not be counting while a child exists.")
     engine.run()
 
 
@@ -114,8 +115,8 @@ def __init__(self, *args, **kwargs):
         def change(self):
             return super().change()
 
-    failer = Failer(fail=lambda n: False, message="Will only time out.")
-    with GameEngine(Scene, systems=[Updater, failer]) as ge:
+    with GameEngine(Scene, systems=[Updater, Failer], fail=lambda n: False,
+                    message="Will only time out.") as ge:
         ge.run()
 
 
@@ -175,6 +176,7 @@ def on_scene_started(self, event, signal):
     class Tester(System):
         listening = False
 
+# TODO: This is going to break with the new change.
         def activate(self, engine):
             if self.listening:
                 assert isinstance(engine.current_scene, SecondScene)
@@ -209,7 +211,7 @@ def on_scene_started(self, event, signal):
     class TestFailer(Failer):
 
         def __init__(self, engine):
-            super().__init__(fail=self.fail, message="Will not call")
+            super().__init__(fail=self.fail, message="Will not call", engine=engine)
             self.first_scene_ended = False
 
         def on_scene_stopped(self, event, signal):
@@ -240,8 +242,7 @@ def on_scene_stopped(self, event, signal):
             assert event.scene is self
             test_function()
 
-    failer = Failer(fail=lambda x: False, message="Will only time out.")
-    with GameEngine(TestScene, systems=[Updater, failer]) as ge:
+    with GameEngine(TestScene, systems=[Updater, Failer], fail=lambda x: False, message="Will only time out.") as ge:
         ge.run()
 
     test_function.assert_called()
@@ -258,3 +259,18 @@ def test_flush_events():
     ge.flush_events()
 
     assert len(ge.events) == 0
+
+
+def test_heartbeats():
+    """This test confirms that Heartbeats work."""
+    was_called = False
+
+    class TestSystem(System):
+
+        def on_heartbeat(self, event: events.Heartbeat, signal):
+            global was_called
+            was_called = True
+            signal(events.Quit())
+
+    with GameEngine(BaseScene, systems=[TestSystem, Failer], fail=lambda x: False, message="Can only time out.") as ge:
+        ge.run()

tests/test_testutil.py

@@ -4,6 +4,7 @@
 from pytest import raises
 
 import ppb.testutils as testutil
+from ppb.events import Heartbeat
 from ppb.events import Quit
 
 
@@ -18,20 +19,20 @@ def test_quitter(loop_count):
 
 
 def test_failer_immediate():
-    failer = testutil.Failer(fail=lambda e: True, message="Expected failure.")
+    failer = testutil.Failer(fail=lambda e: True, message="Expected failure.", engine=None)
 
     with raises(AssertionError):
-        failer.activate(None)
+        failer.__event__(Heartbeat(0.0), lambda x: None)
 
 
 def test_failer_timed():
-    failer = testutil.Failer(fail=lambda e: False, message="Should time out", run_time=0.1)
+    failer = testutil.Failer(fail=lambda e: False, message="Should time out", run_time=0.1, engine=None)
 
     start_time = monotonic()
 
     while True:
         try:
-            failer.activate(None)
+            failer.__event__(Heartbeat(0.0), lambda x: None)
         except AssertionError as e:
             if e.args[0] == "Test ran too long.":
                 end_time = monotonic()