feat: Support for History pseudo state

Author: fgmacedo

docs/releases/3.0.0.md

@@ -20,6 +20,10 @@ To verify the standard adoption, now the automated tests suite includes several
 While these are exiting news for the library and our community, it also introduces several backwards incompatible changes. Due to the major version release, the new behaviour is assumed by default, but we put
 a lot of effort to minimize the changes needed in your codebase, and also introduced a few configuration options that you can enable to restore the old behaviour when possible. The following sections navigate to the new features and includes a migration guide.
 
+### History pseudo-states
+
+The **History pseudo-state** is a special state that is used to record the configuration of the state machine when leaving a compound state. When the state machine transitions into a history state, it will automatically transition to the state that was previously recorded. This allows the state machine to remember the configuration of its child states.
+
 
 ### Create state machine class from a dict definition
 

statemachine/contrib/diagram.py

@@ -140,7 +140,10 @@ def _state_as_node(self, state):
             fontsize=self.state_font_size,
             peripheries=2 if state.final else 1,
         )
-        if state == self.machine.current_state:
+        if (
+            isinstance(self.machine, StateMachine)
+            and state.value in self.machine.configuration_values
+        ):
             node.set_penwidth(self.state_active_penwidth)
             node.set_fillcolor(self.state_active_fillcolor)
         else:

statemachine/engines/async_.py

@@ -68,7 +68,9 @@ async def processing_loop(self):
     async def _trigger(self, trigger_data: TriggerData):
         executed = False
         if trigger_data.event == "__initial__":
-            transition = self._initial_transition(trigger_data)
+            transitions = self._initial_transitions(trigger_data)
+            # TODO: Async does not support multiple initial state activation yet
+            transition = transitions[0]
             await self._activate(trigger_data, transition)
             return self._sentinel
 

statemachine/engines/base.py

@@ -35,10 +35,10 @@ def activate_initial_state(self):
         """
         if self.sm.current_state_value is None:
             trigger_data = BoundEvent("__initial__", _sm=self.sm).build_trigger(machine=self.sm)
-            transition = self._initial_transition(trigger_data)
+            transitions = self._initial_transitions(trigger_data)
             self._processing.acquire(blocking=False)
             try:
-                self._enter_states([transition], trigger_data, OrderedSet(), OrderedSet())
+                self._enter_states(transitions, trigger_data, OrderedSet(), OrderedSet())
             finally:
                 self._processing.release()
         return self.processing_loop()
@@ -75,6 +75,8 @@ def processing_loop(self):  # noqa: C901
 
                 # handles eventless transitions and internal events
                 while not macrostep_done:
+                    logger.debug("Macrostep: eventless/internal queue")
+
                     self.clear_cache()
                     internal_event = TriggerData(
                         self.sm, event=None
@@ -85,10 +87,9 @@ def processing_loop(self):  # noqa: C901
                             macrostep_done = True
                         else:
                             internal_event = self.internal_queue.pop()
-
                             enabled_transitions = self.select_transitions(internal_event)
                     if enabled_transitions:
-                        logger.debug("Eventless/internal queue: %s", enabled_transitions)
+                        logger.debug("Enabled transitions: %s", enabled_transitions)
                         took_events = True
                         self.microstep(list(enabled_transitions), internal_event)
 
@@ -106,6 +107,7 @@ def processing_loop(self):  # noqa: C901
                         self.microstep(list(enabled_transitions), internal_event)
 
                 # Process external events
+                logger.debug("Macrostep: external queue")
                 while not self.external_queue.is_empty():
                     self.clear_cache()
                     took_events = True

statemachine/engines/sync.py

@@ -91,34 +91,38 @@ def __init__(
 
         def __getattr__(self, attribute: str) -> Any: ...
 
-    def _initials_by_document_order(
+    def _initials_by_document_order(  # noqa: C901
         cls, states: List[State], parent: "State | None" = None, order: int = 1
     ):
         """Set initial state by document order if no explicit initial state is set"""
-        initial: "State | None" = None
+        initials: List[State] = []
         for s in states:
             s.document_order = order
             order += 1
             if s.states:
                 cls._initials_by_document_order(s.states, s, order)
             if s.initial:
-                initial = s
-        if not initial and states:
+                initials.append(s)
+
+        if not initials and states:
             initial = states[0]
             initial._initial = True
+            initials.append(initial)
+
+        if not parent:
+            return
+
+        for initial in initials:
+            if not any(t for t in parent.transitions if t.initial and t.target == initial):
+                parent.to(initial, initial=True)
+
+        if not parent.parallel:
+            return
 
-        if (
-            parent
-            and initial
-            and not any(t for t in parent.transitions if t.initial and t.target == initial)
-        ):
-            parent.to(initial, initial=True)
-
-        if parent and parent.parallel:
-            for state in states:
-                state._initial = True
-                if not any(t for t in parent.transitions if t.initial and t.target == state):
-                    parent.to(state, initial=True)
+        for state in states:
+            state._initial = True
+            if not any(t for t in parent.transitions if t.initial and t.target == state):
+                parent.to(state, initial=True)
 
     def _unpack_builders_callbacks(cls):
         callbacks = {}

statemachine/factory.py

@@ -31,13 +31,17 @@ def iterate_states_and_transitions(states: Iterable["State"]):
         yield from state.transitions
         if state.states:
             yield from iterate_states_and_transitions(state.states)
+        if state.history:
+            yield from iterate_states_and_transitions(state.history)
 
 
 def iterate_states(states: Iterable["State"]):
     for state in states:
         yield state
         if state.states:
             yield from iterate_states(state.states)
+        if state.history:
+            yield from iterate_states(state.history)
 
 
 def states_without_path_to_final_states(states: Iterable["State"]):

statemachine/graph.py

@@ -9,6 +9,7 @@
 from typing import cast
 
 from ..factory import StateMachineMetaclass
+from ..state import HistoryState
 from ..state import State
 from ..statemachine import StateMachine
 from ..transition_list import TransitionList
@@ -32,6 +33,7 @@ class TransitionDict(TypedDict, total=False):
 
 
 TransitionsDict = Dict["str | None", List[TransitionDict]]
+TransitionsList = List[TransitionDict]
 
 
 class BaseStateKwargs(TypedDict, total=False):
@@ -46,11 +48,46 @@ class BaseStateKwargs(TypedDict, total=False):
 
 class StateKwargs(BaseStateKwargs, total=False):
     states: List[State]
+    history: List[HistoryState]
+
+
+class HistoryKwargs(TypedDict, total=False):
+    name: str
+    value: Any
+    deep: bool
+
+
+class HistoryDefinition(HistoryKwargs, total=False):
+    on: TransitionsDict
+    transitions: TransitionsList
 
 
 class StateDefinition(BaseStateKwargs, total=False):
     states: Dict[str, "StateDefinition"]
+    history: Dict[str, "HistoryDefinition"]
     on: TransitionsDict
+    transitions: TransitionsList
+
+
+def _parse_history(
+    states: Mapping[str, "HistoryKwargs |HistoryDefinition"],
+) -> Tuple[Dict[str, HistoryState], Dict[str, dict]]:
+    states_instances: Dict[str, HistoryState] = {}
+    events_definitions: Dict[str, dict] = {}
+    for state_id, state_definition in states.items():
+        state_definition = cast(HistoryDefinition, state_definition)
+        transition_defs = state_definition.pop("on", {})
+        transition_list = state_definition.pop("transitions", [])
+        if transition_list:
+            transition_defs[None] = transition_list
+
+        if transition_defs:
+            events_definitions[state_id] = transition_defs
+
+        state_definition = cast(HistoryKwargs, state_definition)
+        states_instances[state_id] = HistoryState(**state_definition)
+
+    return (states_instances, events_definitions)
 
 
 def _parse_states(
@@ -59,27 +96,47 @@ def _parse_states(
     states_instances: Dict[str, State] = {}
     events_definitions: Dict[str, dict] = {}
 
-    for state_id, state_kwargs in states.items():
+    for state_id, state_definition in states.items():
         # Process nested states. Replaces `states` as a definition by a list of `State` instances.
-        inner_states_definitions: Dict[str, StateDefinition] = cast(
-            StateDefinition, state_kwargs
-        ).pop("states", {})
-        if inner_states_definitions:
-            inner_states, inner_events = _parse_states(inner_states_definitions)
+        state_definition = cast(StateDefinition, state_definition)
+
+        # pop the nested states, history and transitions definitions
+        inner_states_defs: Dict[str, StateDefinition] = state_definition.pop("states", {})
+        inner_history_defs: Dict[str, HistoryDefinition] = state_definition.pop("history", {})
+        transition_defs = state_definition.pop("on", {})
+        transition_list = state_definition.pop("transitions", [])
+        if transition_list:
+            transition_defs[None] = transition_list
+
+        if inner_states_defs:
+            inner_states, inner_events = _parse_states(inner_states_defs)
 
             top_level_states = [
                 state._set_id(state_id)
                 for state_id, state in inner_states.items()
                 if not state.parent
             ]
-            state_kwargs["states"] = top_level_states  # type: ignore
+            state_definition["states"] = top_level_states  # type: ignore
             states_instances.update(inner_states)
             events_definitions.update(inner_events)
-        transition_definitions = cast(StateDefinition, state_kwargs).pop("on", {})
-        if transition_definitions:
-            events_definitions[state_id] = transition_definitions
 
-        states_instances[state_id] = State(**state_kwargs)
+        if inner_history_defs:
+            inner_history, inner_events = _parse_history(inner_history_defs)
+
+            top_level_history = [
+                state._set_id(state_id)
+                for state_id, state in inner_history.items()
+                if not state.parent
+            ]
+            state_definition["history"] = top_level_history  # type: ignore
+            states_instances.update(inner_history)
+            events_definitions.update(inner_events)
+
+        if transition_defs:
+            events_definitions[state_id] = transition_defs
+
+        state_definition = cast(BaseStateKwargs, state_definition)
+        states_instances[state_id] = State(**state_definition)
 
     return (states_instances, events_definitions)
 
@@ -114,11 +171,13 @@ def create_machine_class_from_definition(
 
                 target_state_id = transition_data["target"]
                 target = states_instances[target_state_id] if target_state_id else None
+                transition_event_name = transition_data.get("event")
+                if event_name is not None:
+                    transition_event_name = f"{event_name} {transition_event_name}".strip()
 
-                # TODO: Join `trantion_data.event` with `event_name`
                 transition = source.to(
                     target,
-                    event=event_name,
+                    event=transition_event_name,
                     internal=transition_data.get("internal"),
                     initial=transition_data.get("initial"),
                     cond=transition_data.get("cond"),

statemachine/io/__init__.py

@@ -472,7 +472,13 @@ def create_datamodel_action_callable(action: DataModel) -> "Callable | None":
     if not data_elements:
         return None
 
+    initialized = False
+
     def datamodel(*args, **kwargs):
+        nonlocal initialized
+        if initialized:
+            return
+        initialized = True
         machine: StateMachine = kwargs["machine"]
         for act in data_elements:
             try: