feat: Nested states (compound / parallel)

Author: fgmacedo

statemachine/factory.py

@@ -32,8 +32,26 @@ def __init__(cls, name: str, bases: Tuple[type], attrs: Dict[str, Any]):
         cls.add_inherited(bases)
         cls.add_from_attributes(attrs)
 
+        if not cls.states:
+            return
+        initials = [s for s in cls.states if s.initial]
+        parallels = [s.id for s in cls.states if s.parallel]
+        root_only_has_parallels = len(cls.states) == len(parallels)
+
+        if len(initials) != 1 and not root_only_has_parallels:
+            raise InvalidDefinition(
+                _(
+                    "There should be one and only one initial state. "
+                    "Your currently have these: {0}"
+                ).format(", ".join(s.id for s in initials))
+            )
+
         try:
-            cls.initial_state: State = next(s for s in cls.states if s.initial)
+            if root_only_has_parallels:
+                # TODO: Temp, whe should fix initial, and current state design
+                cls.initial_state: State = next(s for s in cls.states if s.initial)
+            else:
+                cls.initial_state: State = next(s for s in initials if s.initial)
         except StopIteration:
             cls.initial_state = None  # Abstract SM still don't have states
 
@@ -142,15 +160,19 @@ def _add_unbounded_callback(cls, attr_name, func):
 
     def add_state(cls, id, state: State):
         state._set_id(id)
-        cls.states.append(state)
-        cls.states_map[state.value] = state
-        if not hasattr(cls, id):
-            setattr(cls, id, state)
+        if not state.parent:
+            cls.states.append(state)
+            cls.states_map[state.value] = state
+            if not hasattr(cls, id):
+                setattr(cls, id, state)
 
         # also register all events associated directly with transitions
         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,25 @@
     from .statemachine import StateMachine
 
 
+class NestedStateFactory(type):
+    def __new__(  # type: ignore [misc]
+        cls, classname, bases, attrs, name=None, **kwargs
+    ) -> "State":
+
+        if not bases:
+            return super().__new__(cls, classname, bases, attrs)  # type: ignore [return-value]
+
+        substates = []
+        for key, value in attrs.items():
+            if isinstance(value, State):
+                value._set_id(key)
+                substates.append(value)
+            if isinstance(value, TransitionList):
+                value.add_event(key)
+
+        return State(name=name, substates=substates, **kwargs)
+
+
 class State:
     """
     A State in a :ref:`StateMachine` describes a particular behavior of the machine.
@@ -92,23 +111,59 @@ class State:
 
     """
 
+    class Builder(metaclass=NestedStateFactory):
+
+        # Mimic the :ref:`State` public API to help linters discover the result of the Builder
+        # class.
+
+        @classmethod
+        def to(cls, *args: "State", **kwargs) -> "TransitionList":  # pragma: no cover
+            """Create transitions to the given target states.
+
+            .. note: This method is only a type hint for mypy.
+                The actual implementation belongs to the :ref:`State` class.
+            """
+            return TransitionList()
+
+        @classmethod
+        def from_(
+            cls, *args: "State", **kwargs
+        ) -> "TransitionList":  # pragma: no cover
+            """Create transitions from the given target states (reversed).
+
+            .. note: This method is only a type hint for mypy.
+                The actual implementation belongs to the :ref:`State` class.
+            """
+            return TransitionList()
+
     def __init__(
         self,
         name: str = "",
         value: Any = None,
         initial: bool = False,
         final: bool = False,
+        parallel: bool = False,
+        substates: Any = None,
         enter: Any = None,
         exit: Any = None,
     ):
         self.name = name
         self.value = value
+        self.parallel = parallel
+        self.substates = substates or []
         self._initial = initial
         self._final = final
         self._id: str = ""
+        self.parent: "State" = None
         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
+            setattr(self, substate.id, substate)
 
     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="Microwave oven", parallel=True):
+        class engine(State.Builder):
+            off = State("Off", initial=True)
+
+            class on(State.Builder):
+                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):
+            closed = State(initial=True)
+            open = State()
+
+            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/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)
+
+    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__(allow_event_without_transition=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: int = 0):
+        print(f"entering reset_elapsed from {event} with {time}")
+        self.last_turn = time
+
+    @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()
+sm.send("anything")

tests/test_compound.py

@@ -0,0 +1,68 @@
+import pytest
+
+from statemachine import State
+
+
+@pytest.fixture()
+def microwave_cls():
+    from tests.examples.microwave_inheritance_machine import MicroWave
+
+    return MicroWave
+
+
+def assert_state(s, name, initial=False, final=False, parallel=False, substates=None):
+    if substates is None:
+        substates = []
+
+    assert isinstance(s, State)
+    assert s.name == name
+    assert s.initial is initial
+    assert s.final is final
+    assert s.parallel is parallel
+    assert isinstance(s, State)
+    assert set(s.substates) == set(substates)
+
+
+class TestNestedSyntax:
+    def test_capture_constructor_arguments(self, microwave_cls):
+        sm = microwave_cls()
+
+        assert_state(
+            sm.oven,
+            "Microwave oven",
+            parallel=True,
+            substates=[sm.oven.engine, sm.oven.door],
+        )
+        assert_state(
+            sm.oven.engine,
+            "Engine",
+            initial=False,
+            substates=[sm.oven.engine.on, sm.oven.engine.off],
+        )
+        assert_state(sm.oven.engine.off, "Off", initial=True)
+        assert_state(
+            sm.oven.engine.on,
+            "On",
+            substates=[sm.oven.engine.on.idle, sm.oven.engine.on.cooking],
+        )
+        assert_state(
+            sm.oven.door,
+            "Door",
+            initial=False,
+            substates=[sm.oven.door.closed, sm.oven.door.open],
+        )
+        assert_state(sm.oven.door.closed, "Closed", initial=True)
+        assert_state(sm.oven.door.open, "Open")
+
+    def test_list_children_states(self, microwave_cls):
+        sm = microwave_cls()
+        assert [s.id for s in sm.oven.engine.substates] == ["off", "on"]
+
+    def test_list_events(self, microwave_cls):
+        sm = microwave_cls()
+        assert [e.name for e in sm.events] == [
+            "turn_on",
+            "turn_off",
+            "door_open",
+            "door_close",
+        ]

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"