feat: Queued mode preview

Author: fgmacedo

statemachine/event.py

@@ -44,7 +44,8 @@ def _process(self, event_data):
                 event_data.executed = True
                 break
         else:
-            raise TransitionNotAllowed(event_data.event, event_data.state)
+            if not event_data.machine.queued:
+                raise TransitionNotAllowed(event_data.event, event_data.state)
 
 
 def trigger_event_factory(event):

statemachine/state.py

@@ -11,7 +11,7 @@
 
 class NestedStateFactory(type):
     def __new__(  # type: ignore [misc]
-        cls, classname, bases, attrs, name=None, initial=False, parallel=False
+        cls, classname, bases, attrs, name=None, **kwargs
     ) -> "State":
 
         if not bases:
@@ -25,7 +25,7 @@ def __new__(  # type: ignore [misc]
             if isinstance(value, TransitionList):
                 value.add_event(key)
 
-        return State(name=name, initial=initial, parallel=parallel, substates=substates)
+        return State(name=name, substates=substates, **kwargs)
 
 
 class NestedStateBuilder(metaclass=NestedStateFactory):

statemachine/statemachine.py

@@ -25,10 +25,11 @@ class StateMachine(metaclass=StateMachineMetaclass):
     states = []  # type: List[State]
     states_map = {}  # type: Dict[Any, State]
 
-    def __init__(self, model=None, state_field="state", start_value=None):
+    def __init__(self, model=None, state_field="state", start_value=None, queued=False):
         self.model = model if model else Model()
         self.state_field = state_field
         self.start_value = start_value
+        self.queued = queued
 
         if self._abstract:
             raise InvalidDefinition(_("There are no states or transitions."))

tests/examples/traffic_light_nested_machine.py

@@ -0,0 +1,64 @@
+"""
+Nested Traffic light machine
+----------------------------
+
+Demonstrates the concept of nested compound states.
+
+From this example on XState: https://xstate.js.org/docs/guides/hierarchical.html#api
+
+"""
+import time
+
+from statemachine import State
+from statemachine import StateMachine
+
+
+class NestedTrafficLightMachine(StateMachine):
+    "A traffic light machine"
+    green = State(initial=True, enter="reset_elapsed")
+    yellow = State(enter="reset_elapsed")
+
+    class red(State.Builder, enter="reset_elapsed"):
+        "Pedestrian states"
+        walk = State(initial=True)
+        wait = State()
+        stop = State()
+        blinking = State()
+
+        ped_countdown = walk.to(wait) | wait.to(stop)
+
+    assert isinstance(red, State)
+    timer = green.to(yellow) | yellow.to(red) | red.to(green)
+    power_outage = red.blinking.from_()
+    power_restored = red.from_()
+
+    def __init__(self, seconds_to_turn_state=5, seconds_running=20):
+        self.seconds_to_turn_state = seconds_to_turn_state
+        self.seconds_running = seconds_running
+        super().__init__(queued=True)
+
+    def on_timer(self, event: str, source: State, target: State):
+        print(f".. Running {event} from {source.id} to {target.id}")
+
+    def reset_elapsed(self, event: str, time):
+        print(f"entering reset_elapsed from {event} with {time}")
+        self.last_turn = time if time else 0
+
+    @timer.cond
+    def time_is_over(self, time):
+        return time - self.last_turn > self.seconds_to_turn_state
+
+    def run_forever(self):
+        self.running = True
+        start_time = time.time()
+        while self.running:
+            print("tick!")
+            time.sleep(1)
+            curr_time = time.time()
+            self.send("timer", time=curr_time)
+
+            if curr_time - start_time > self.seconds_running:
+                self.running = False
+
+
+sm = NestedTrafficLightMachine()

tests/test_compound.py

@@ -23,7 +23,7 @@ def assert_state(s, name, initial=False, final=False, parallel=False, substates=
     assert set(s.substates) == set(substates)
 
 
-class TestNestedDeclarations:
+class TestNestedSyntax:
     def test_capture_constructor_arguments(self, microwave_cls):
         sm = microwave_cls()