feat: Hierarquical statemachines with compose and parallel

Author: fgmacedo

docs/transitions.md

@@ -84,7 +84,7 @@ Syntax:
 >>> draft = State("Draft")
 
 >>> draft.to.itself()
-TransitionList([Transition('Draft', 'Draft', event='', internal=False)])
+TransitionList([Transition('Draft', 'Draft', event=[], internal=False)])
 
 ```
 
@@ -101,7 +101,7 @@ Syntax:
 >>> draft = State("Draft")
 
 >>> draft.to.itself(internal=True)
-TransitionList([Transition('Draft', 'Draft', event='', internal=True)])
+TransitionList([Transition('Draft', 'Draft', event=[], internal=True)])
 
 ```
 

statemachine/contrib/diagram.py

@@ -172,10 +172,10 @@ def _transition_as_edge(self, transition):
 
     def get_graph(self):
         graph = self._get_graph(self.machine)
-        self._graph_states(self.machine, graph)
+        self._graph_states(self.machine, graph, is_root=True)
         return graph
 
-    def _graph_states(self, state, graph):
+    def _graph_states(self, state, graph, is_root=False):
         initial_node = self._initial_node(state)
         initial_subgraph = pydot.Subgraph(
             graph_name=f"{initial_node.get_name()}_initial",
@@ -193,8 +193,9 @@ def _graph_states(self, state, graph):
         graph.add_subgraph(initial_subgraph)
         graph.add_subgraph(atomic_states_subgraph)
 
-        initial = next(s for s in state.states if s.initial)
-        graph.add_edge(self._initial_edge(initial_node, initial))
+        if is_root:
+            initial = next(s for s in state.states if s.initial)
+            graph.add_edge(self._initial_edge(initial_node, initial))
 
         for substate in state.states:
             if substate.states:

statemachine/engines/base.py

@@ -241,8 +241,10 @@ def _select_transitions(
         def first_transition_that_matches(state: State, event: Event) -> "Transition | None":
             for s in chain([state], state.ancestors()):
                 for transition in s.transitions:
-                    if predicate(transition, event) and self._conditions_match(
-                        transition, trigger_data
+                    if (
+                        not transition.initial
+                        and predicate(transition, event)
+                        and self._conditions_match(transition, trigger_data)
                     ):
                         return transition
 
@@ -481,31 +483,28 @@ def add_descendant_states_to_enter(
             # Handle compound states
             states_for_default_entry.add(info)
             initial_state = next(s for s in state.states if s.initial)
-            for transition in initial_state.transitions:
-                info_initial = StateTransition(
-                    transition=transition,
-                    target=transition.target,
-                    source=transition.source,
-                )
-                self.add_descendant_states_to_enter(
-                    info_initial,
-                    states_to_enter,
-                    states_for_default_entry,
-                    default_history_content,
-                )
-            for transition in initial_state.transitions:
-                info_initial = StateTransition(
-                    transition=transition,
-                    target=transition.target,
-                    source=transition.source,
-                )
-                self.add_ancestor_states_to_enter(
-                    info_initial,
-                    state,
-                    states_to_enter,
-                    states_for_default_entry,
-                    default_history_content,
-                )
+            transition = next(
+                t for t in state.transitions if t.initial and t.target == initial_state
+            )
+            info_initial = StateTransition(
+                transition=transition,
+                target=transition.target,
+                source=transition.source,
+            )
+            self.add_descendant_states_to_enter(
+                info_initial,
+                states_to_enter,
+                states_for_default_entry,
+                default_history_content,
+            )
+
+            self.add_ancestor_states_to_enter(
+                info_initial,
+                state,
+                states_to_enter,
+                states_for_default_entry,
+                default_history_content,
+            )
         elif state.parallel:
             # Handle parallel states
             for child_state in state.states:

statemachine/events.py

@@ -8,10 +8,13 @@ class Events:
     def __init__(self):
         self._items: list[Event] = []
 
-    def __repr__(self):
+    def __str__(self):
         sep = " " if len(self._items) > 1 else ""
         return sep.join(item for item in self._items)
 
+    def __repr__(self):
+        return f"{self._items!r}"
+
     def __iter__(self):
         return iter(self._items)
 

statemachine/factory.py

@@ -32,6 +32,8 @@ def __init__(
         registry.register(cls)
         cls.name = cls.__name__
         cls.id = cls.name.lower()
+        # TODO: Experiment with the IDEA of a root state
+        # cls.root = State(id=cls.id, name=cls.name)
         cls.states: States = States()
         cls.states_map: Dict[Any, State] = {}
         """Map of ``state.value`` to the corresponding :ref:`state`."""
@@ -50,7 +52,7 @@ def __init__(
         if not cls.states:
             return
 
-        cls._initials_by_document_order(cls.states)
+        cls._initials_by_document_order(cls.states, parent=None)
 
         initials = [s for s in cls.states if s.initial]
         parallels = [s.id for s in cls.states if s.parallel]
@@ -79,16 +81,24 @@ def __init__(
 
         def __getattr__(self, attribute: str) -> Any: ...
 
-    def _initials_by_document_order(cls, states):
+    def _initials_by_document_order(cls, states, parent: "State | None" = None):
         """Set initial state by document order if no explicit initial state is set"""
-        has_initial = False
+        initial: "State | None" = None
         for s in states:
-            cls._initials_by_document_order(s.states)
+            cls._initials_by_document_order(s.states, s)
             if s.initial:
-                has_initial = True
+                initial = s
                 break
-        if not has_initial and states:
-            states[0]._initial = True
+        if not initial and states:
+            initial = states[0]
+            initial._initial = True
+
+        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)
 
     def _unpack_builders_callbacks(cls):
         callbacks = {}

statemachine/graph.py

@@ -11,7 +11,6 @@ def visit_connected_states(state):
             continue
         already_visited.add(state)
         yield state
-        visit.extend(s for s in state.states if s.initial)
         visit.extend(t.target for t in state.transitions)
 
 

statemachine/io/__init__.py

@@ -22,6 +22,7 @@ class TransitionDict(TypedDict, total=False):
     target: str
     event: "str | None"
     internal: bool
+    initial: bool
     validators: bool
     cond: "str | ActionProtocol | Sequence[str] | Sequence[ActionProtocol]"
     unless: "str | ActionProtocol | Sequence[str] | Sequence[ActionProtocol]"
@@ -117,6 +118,7 @@ def create_machine_class_from_definition(
                 transition = source.to(
                     target,
                     event=event_name,
+                    initial=transition_data.get("initial"),
                     cond=transition_data.get("cond"),
                     unless=transition_data.get("unless"),
                     on=transition_data.get("on"),

statemachine/io/scxml/parser.py

@@ -1,5 +1,6 @@
 import re
 import xml.etree.ElementTree as ET
+from typing import Set
 
 from .schema import Action
 from .schema import AssignAction
@@ -32,6 +33,12 @@ def strip_namespaces(tree: ET.Element):
                 attrib[new_name] = attrib.pop(name)
 
 
+def _parse_initial(initial_content: "str | None") -> Set[str]:
+    if initial_content is None:
+        return set()
+    return set(initial_content.split())
+
+
 def parse_scxml(scxml_content: str) -> StateMachineDefinition:
     root = ET.fromstring(scxml_content)
     strip_namespaces(root)
@@ -40,9 +47,9 @@ def parse_scxml(scxml_content: str) -> StateMachineDefinition:
     if scxml is None:
         raise ValueError("No scxml element found in document")
 
-    initial_state = scxml.get("initial")
+    initial_state = _parse_initial(scxml.get("initial"))
 
-    definition = StateMachineDefinition(initial_state=initial_state)
+    definition = StateMachineDefinition(initial_states=initial_state)
 
     # Parse datamodel
     datamodel = parse_datamodel(scxml)
@@ -52,19 +59,19 @@ def parse_scxml(scxml_content: str) -> StateMachineDefinition:
     # Parse states
     for state_elem in scxml:
         if state_elem.tag == "state":
-            state = parse_state(state_elem, definition.initial_state)
+            state = parse_state(state_elem, definition.initial_states)
             definition.states[state.id] = state
         elif state_elem.tag == "final":
-            state = parse_state(state_elem, definition.initial_state, is_final=True)
+            state = parse_state(state_elem, definition.initial_states, is_final=True)
             definition.states[state.id] = state
         elif state_elem.tag == "parallel":
-            state = parse_state(state_elem, definition.initial_state, is_parallel=True)
+            state = parse_state(state_elem, definition.initial_states, is_parallel=True)
             definition.states[state.id] = state
 
     # If no initial state was specified, pick the first state
-    if not definition.initial_state and definition.states:
-        definition.initial_state = next(iter(definition.states.keys()))
-        definition.states[definition.initial_state].initial = True
+    if not definition.initial_states and definition.states:
+        definition.initial_states = next(iter(definition.states.keys()))
+        definition.states[definition.initial_states].initial = True
 
     return definition
 
@@ -95,15 +102,15 @@ def parse_datamodel(root: ET.Element) -> "DataModel | None":
 
 def parse_state(
     state_elem: ET.Element,
-    initial_state: "str | None",
+    initial_states: Set[str],
     is_final: bool = False,
     is_parallel: bool = False,
 ) -> State:
     state_id = state_elem.get("id")
     if not state_id:
         raise ValueError("State must have an 'id' attribute")
 
-    initial = state_id == initial_state
+    initial = state_id in initial_states
     state = State(id=state_id, initial=initial, final=is_final, parallel=is_parallel)
 
     # Parse onentry actions
@@ -122,18 +129,25 @@ def parse_state(
         state.transitions.append(transition)
 
     # Parse child states
-    initial_state = state_elem.get("initial")
+    initial_states |= _parse_initial(state_elem.get("initial"))
+    initial_elem = state_elem.find("initial")
+    if initial_elem is not None:
+        for trans_elem in initial_elem.findall("transition"):
+            transition = parse_transition(trans_elem, initial=True)
+            state.transitions.append(transition)
+            initial_states |= _parse_initial(trans_elem.get("target"))
+
     for child_state_elem in state_elem.findall("state"):
-        child_state = parse_state(child_state_elem, initial_state=initial_state)
+        child_state = parse_state(child_state_elem, initial_states=initial_states)
         state.states[child_state.id] = child_state
     for child_state_elem in state_elem.findall("parallel"):
-        state = parse_state(child_state_elem, initial_state=initial_state, is_parallel=True)
+        state = parse_state(child_state_elem, initial_states=initial_states, is_parallel=True)
         state.states[child_state.id] = child_state
 
     return state
 
 
-def parse_transition(trans_elem: ET.Element) -> Transition:
+def parse_transition(trans_elem: ET.Element, initial: bool = False) -> Transition:
     target = trans_elem.get("target")
     if not target:
         raise ValueError("Transition must have a 'target' attribute")
@@ -143,7 +157,9 @@ def parse_transition(trans_elem: ET.Element) -> Transition:
 
     executable_content = parse_executable_content(trans_elem)
 
-    return Transition(target=target, event=event, cond=cond, on=executable_content)
+    return Transition(
+        target=target, initial=initial, event=event, cond=cond, on=executable_content
+    )
 
 
 def parse_executable_content(element: ET.Element) -> ExecutableContent:

statemachine/io/scxml/processor.py

@@ -88,7 +88,10 @@ def _process_transitions(self, transitions: List[Transition]):
             event = transition.event or None
             if event not in on_dict:
                 on_dict[event] = []
-            transition_dict: TransitionDict = {"target": transition.target}
+            transition_dict: TransitionDict = {
+                "target": transition.target,
+                "initial": transition.initial,
+            }
 
             # Process cond
             if transition.cond:

statemachine/io/scxml/schema.py

@@ -2,6 +2,7 @@
 from dataclasses import field
 from typing import Dict
 from typing import List
+from typing import Set
 
 
 @dataclass
@@ -93,6 +94,7 @@ class ScriptAction(Action):
 @dataclass
 class Transition:
     target: str
+    initial: bool = False
     event: "str | None" = None
     cond: "str | None" = None
     on: "ExecutableContent | None" = None
@@ -127,5 +129,5 @@ class DataModel:
 @dataclass
 class StateMachineDefinition:
     states: Dict[str, State] = field(default_factory=dict)
-    initial_state: "str | None" = None
+    initial_states: Set[str] = field(default_factory=set)
     datamodel: "DataModel | None" = None