feat: Nested states (compound / parallel)

Author: fgmacedo

statemachine/factory.py

@@ -60,8 +60,9 @@ def _check(cls):
         if not has_states:
             raise InvalidDefinition(_("There are no states."))
 
-        if not has_events:
-            raise InvalidDefinition(_("There are no events."))
+        # TODO: Validate no events if has nested states
+        # if not has_events:
+        #     raise InvalidDefinition(_("There are no events."))
 
         cls._check_initial_state()
         cls._check_final_states()
@@ -151,6 +152,9 @@ def add_state(cls, id, state: State):
         for event in state.transitions.unique_events:
             cls.add_event(event)
 
+        for substate in state.substates:
+            cls.add_state(substate.id, substate)
+
     def add_event(cls, event, transitions=None):
         if transitions is not None:
             transitions.add_event(event)

statemachine/state.py

@@ -13,6 +13,26 @@
     from .statemachine import StateMachine
 
 
+class NestedStateFactory(type):
+    def __new__(cls, classname, bases, attrs, name=None, initial=False, parallel=False):
+
+        if not bases:
+            return super().__new__(cls, classname, bases, attrs)
+
+        substates = []
+        for key, value in attrs.items():
+            if not isinstance(value, State):
+                continue
+            value._set_id(key)
+            substates.append(value)
+
+        return State(name, initial=initial, parallel=parallel, substates=substates)
+
+
+class NestedStateBuilder(metaclass=NestedStateFactory):
+    pass
+
+
 class State:
     """
     A State in a :ref:`StateMachine` describes a particular behavior of the machine.
@@ -92,23 +112,35 @@ class State:
 
     """
 
+    Builder = NestedStateBuilder
+
     def __init__(
         self,
         name: str = "",
         value: Any = None,
         initial: bool = False,
         final: bool = False,
+        parallel=False,
+        substates=None,
         enter: Any = None,
         exit: Any = None,
     ):
         self.name = name
         self.value = value
+        self.parallel = parallel
+        self.parent: "State" = None
+        self.substates = substates or []
         self._initial = initial
         self._final = final
         self._id: str = ""
         self.transitions = TransitionList()
         self.enter = CallbackMetaList().add(enter)
         self.exit = CallbackMetaList().add(exit)
+        self._init_substates()
+
+    def _init_substates(self):
+        for substate in self.substates:
+            substate.parent = self
 
     def __eq__(self, other):
         return (

tests/examples/microwave_inheritance_machine.py

@@ -0,0 +1,68 @@
+"""
+Microwave machine
+=================
+
+Example that exercises the Compound and Parallel states.
+
+Compound
+--------
+
+If there are more than one substates, one of them is usually designated as the initial state of
+that compound state.
+
+When a compound state is active, its substates behave as though they were an active state machine:
+ Exactly one child state must also be active.  This means that:
+
+When a compound state is entered, it must also enter exactly one of its substates, usually its
+initial state.
+When an event happens, the substates have priority when it comes to selecting which transition to
+follow. If a substate happens to handles an event, the event is consumed, it isn’t passed to the
+parent compound state.
+When a substate transitions to another substate, both “inside” the compound state, the compound
+state does not exit or enter; it remains active.
+When a compound state exits, its substate is simultaneously exited too. (Technically, the substate
+exits first, then its parent.)
+Compound states may be nested, or include parallel states.
+
+The opposite of a compound state is an atomic state, which is a state with no substates.
+
+A compound state is allowed to define transitions to its child states. Normally, when a transition
+leads from a state, it causes that state to be exited.  For transitions from a compound state to
+one of its descendants, it is possible to define a transition that avoids exiting and entering
+the compound state itself, such transitions are called local transitions.
+
+
+"""
+from statemachine import State
+from statemachine import StateMachine
+
+
+class MicroWave(StateMachine):
+    class oven(State.Builder, name="Oven", initial=True, parallel=True):
+        class engine(State.Builder, name="Engine"):
+            off = State("Off", initial=True)
+
+            class on(State.Builder, name="On"):
+                idle = State("Idle", initial=True)
+                cooking = State("Cooking")
+
+                idle.to(cooking, cond="closed.is_active")
+                cooking.to(idle, cond="open.is_active")
+                cooking.to.itself(internal=True, on="increment_timer")
+
+            turn_off = on.to(off)
+            turn_on = off.to(on)
+            on.to(off, cond="cook_time_is_over")  # eventless transition
+
+        class door(State.Builder, name="Door"):
+            closed = State("Closed", initial=True)
+            open = State("Open")
+
+            door_open = closed.to(open)
+            door_close = open.to(closed)
+
+    def __init__(self):
+        self.cook_time = 5
+        self.door_closed = True
+        self.timer = 0
+        super().__init__()

tests/test_compound.py

@@ -0,0 +1,33 @@
+import pytest
+
+from statemachine import State
+from statemachine import StateMachine
+
+
+@pytest.fixture()
+def compound_engine_cls():
+    class TestMachine(StateMachine):
+        class engine(State.Builder, name="Engine", initial=True):
+            off = State("Off", initial=True)
+            on = State("On")
+
+            turn_off = on.to(off)
+            turn_on = off.to(on)
+
+    return TestMachine
+
+
+class TestNestedDeclarations:
+    def test_capture_constructor_arguments(self, compound_engine_cls):
+        sm = compound_engine_cls()
+        assert isinstance(sm.engine, State)
+        assert sm.engine.name == "Engine"
+        assert sm.engine.initial is True
+
+    def test_list_children_states(self, compound_engine_cls):
+        sm = compound_engine_cls()
+        assert [s.id for s in sm.engine.children] == ["off", "on"]
+
+    def test_list_events(self, compound_engine_cls):
+        sm = compound_engine_cls()
+        assert [e.name for e in sm.events] == ["turn_off", "turn_on"]

tests/test_nested.py

@@ -0,0 +1,5 @@
+def test_nested_sm():
+    from tests.examples.microwave_inheritance_machine import MicroWave
+
+    sm = MicroWave()
+    assert sm.current_state.id == "oven"