feat: Nested states (compound / parallel)

Author: fgmacedo

statemachine/callbacks.py

@@ -86,7 +86,7 @@ def __str__(self):
         return ", ".join(str(c) for c in self)
 
     def setup(self, resolver):
-        """Validate configuracions"""
+        """Validate configurations"""
         self._resolver = resolver
         self.items = [
             callback for callback in self.items if callback.setup(self._resolver)

statemachine/factory.py

@@ -30,13 +30,13 @@ def __init__(cls, name, bases, attrs):
     def _set_special_states(cls):
         if not cls.states:
             return
-        initials = [s for s in cls.states if s.initial]
+        initials = [s for s in cls.states if s.initial and not s.parent]
         if len(initials) != 1:
             raise InvalidDefinition(
                 _(
                     "There should be one and only one initial state. "
-                    "Your currently have these: {!r}".format(initials)
-                )
+                    "Your currently have these: {0}"
+                ).format(", ".join(s.id for s in initials))
             )
         cls.initial_state = initials[0]
         cls.final_states = [state for state in cls.states if state.final]
@@ -70,8 +70,9 @@ def _check(cls):
         if not cls.states:
             raise InvalidDefinition(_("There are no states."))
 
-        if not cls._events:
-            raise InvalidDefinition(_("There are no events."))
+        # TODO: Validate no events if has nested states
+        # if not cls._events:
+        #     raise InvalidDefinition(_("There are no events."))
 
         cls._check_disconnected_state()
 
@@ -127,6 +128,9 @@ def add_state(cls, id, 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

@@ -7,6 +7,26 @@
 from .utils import ugettext as _
 
 
+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 state machine describes a particular behaviour of the machine.
@@ -73,18 +93,36 @@ class State:
 
     """
 
+    Builder = NestedStateBuilder
+
     def __init__(
-        self, name, value=None, initial=False, final=False, enter=None, exit=None
+        self,
+        name,
+        value=None,
+        initial=False,
+        final=False,
+        parallel=False,
+        substates=None,
+        enter=None,
+        exit=None,
     ):
-        # type: (str, Optional[Any], bool, bool, Optional[Any], Optional[Any]) -> None
+        # type: (str, Optional[Any], bool, bool, bool, Optional[Any], Optional[Any], Optional[Any]) -> None  # noqa
         self.name = name
         self.value = value
+        self.parallel = parallel
+        self.parent: "State" = None
+        self.substates = substates or []
         self._id = None  # type: Optional[str]
         self._initial = initial
         self.transitions = TransitionList()
         self._final = final
         self.enter = Callbacks().add(enter)
         self.exit = Callbacks().add(exit)
+        self._init_substates()
+
+    def _init_substates(self):
+        for substate in self.substates:
+            substate.parent = self
 
     def _setup(self, resolver):
         self.enter.setup(resolver)

statemachine/statemachine.py

@@ -124,7 +124,6 @@ def current_state_value(self, value):
 
     @property
     def current_state(self):
-        # type: () -> Optional[State]
         return self.states_map.get(self.current_state_value, None)
 
     @current_state.setter

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):
+        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"