feat: Add syntax for compound and parallel states (only parser)

Author: fgmacedo

docs/diagram.md

@@ -42,7 +42,7 @@ Graphviz. For example, on Debian-based systems (such as Ubuntu), you can use the
 >>> dot = graph()
 
 >>> dot.to_string()  # doctest: +ELLIPSIS
-'digraph list {...
+'digraph OrderControl {...
 
 ```
 

docs/images/order_control_machine_initial.png

@@ -18,62 +18,79 @@ class DotGraphMachine:
     font_name = "Arial"
     """Graph font face name"""
 
-    state_font_size = "10"
-    """State font size in points"""
+    state_font_size = "10pt"
+    """State font size"""
 
     state_active_penwidth = 2
     """Active state external line width"""
 
     state_active_fillcolor = "turquoise"
 
-    transition_font_size = "9"
-    """Transition font size in points"""
+    transition_font_size = "9pt"
+    """Transition font size"""
 
-    def __init__(self, machine: StateMachine):
+    def __init__(self, machine):
         self.machine = machine
 
-    def _get_graph(self):
-        machine = self.machine
+    def _get_graph(self, machine):
         return pydot.Dot(
-            "list",
+            machine.name,
             graph_type="digraph",
             label=machine.name,
             fontname=self.font_name,
             fontsize=self.state_font_size,
             rankdir=self.graph_rankdir,
+            compound="true",
         )
 
-    def _initial_node(self):
+    def _get_subgraph(self, state):
+        style = ", solid"
+        if state.parent and state.parent.parallel:
+            style = ", dashed"
+        subgraph = pydot.Subgraph(
+            label=f"{state.name}",
+            graph_name=f"cluster_{state.id}",
+            style=f"rounded{style}",
+            cluster="true",
+        )
+        return subgraph
+
+    def _initial_node(self, state):
         node = pydot.Node(
-            "i",
-            shape="circle",
+            self._state_id(state),
+            label="",
+            shape="point",
             style="filled",
-            fontsize="1",
+            fontsize="1pt",
             fixedsize="true",
             width=0.2,
             height=0.2,
         )
         node.set_fillcolor("black")
         return node
 
-    def _initial_edge(self):
+    def _initial_edge(self, initial_node, state):
+        extra_params = {}
+        if state.states:
+            extra_params["lhead"] = f"cluster_{state.id}"
         return pydot.Edge(
-            "i",
-            self.machine.initial_state.id,
+            initial_node.get_name(),
+            self._state_id(state),
             label="",
             color="blue",
             fontname=self.font_name,
             fontsize=self.transition_font_size,
+            **extra_params,
         )
 
     def _actions_getter(self):
         if isinstance(self.machine, StateMachine):
 
-            def getter(grouper) -> str:
+            def getter(grouper):
                 return self.machine._callbacks.str(grouper.key)
         else:
 
-            def getter(grouper) -> str:
+            def getter(grouper):
                 all_names = set(dir(self.machine))
                 return ", ".join(
                     str(c) for c in grouper if not c.is_convention or c.func in all_names
@@ -104,11 +121,18 @@ def _state_actions(self, state):
 
         return actions
 
+    @staticmethod
+    def _state_id(state):
+        if state.states:
+            return f"{state.id}_anchor"
+        else:
+            return state.id
+
     def _state_as_node(self, state):
         actions = self._state_actions(state)
 
         node = pydot.Node(
-            state.id,
+            self._state_id(state),
             label=f"{state.name}{actions}",
             shape="rectangle",
             style="rounded, filled",
@@ -127,29 +151,64 @@ def _transition_as_edge(self, transition):
         cond = ", ".join([str(cond) for cond in transition.cond])
         if cond:
             cond = f"\n[{cond}]"
+
+        extra_params = {}
+        has_substates = transition.source.states or transition.target.states
+        if transition.source.states:
+            extra_params["ltail"] = f"cluster_{transition.source.id}"
+        if transition.target.states:
+            extra_params["lhead"] = f"cluster_{transition.target.id}"
+
         return pydot.Edge(
-            transition.source.id,
-            transition.target.id,
+            self._state_id(transition.source),
+            self._state_id(transition.target),
             label=f"{transition.event}{cond}",
             color="blue",
             fontname=self.font_name,
             fontsize=self.transition_font_size,
+            minlen=2 if has_substates else 1,
+            **extra_params,
         )
 
     def get_graph(self):
-        graph = self._get_graph()
-        graph.add_node(self._initial_node())
-        graph.add_edge(self._initial_edge())
+        graph = self._get_graph(self.machine)
+        self._graph_states(self.machine, graph)
+        return graph
 
-        for state in self.machine.states:
-            graph.add_node(self._state_as_node(state))
-            for transition in state.transitions:
+    def _graph_states(self, state, graph):
+        initial_node = self._initial_node(state)
+        initial_subgraph = pydot.Subgraph(
+            graph_name=f"{initial_node.get_name()}_initial",
+            label="",
+            peripheries=0,
+            margin=0,
+        )
+        atomic_states_subgraph = pydot.Subgraph(
+            graph_name=f"cluster_{initial_node.get_name()}_atomic",
+            label="",
+            peripheries=0,
+            cluster="true",
+        )
+        initial_subgraph.add_node(initial_node)
+        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))
+
+        for substate in state.states:
+            if substate.states:
+                subgraph = self._get_subgraph(substate)
+                self._graph_states(substate, subgraph)
+                graph.add_subgraph(subgraph)
+            else:
+                atomic_states_subgraph.add_node(self._state_as_node(substate))
+
+            for transition in substate.transitions:
                 if transition.internal:
                     continue
                 graph.add_edge(self._transition_as_edge(transition))
 
-        return graph
-
     def __call__(self):
         return self.get_graph()
 
@@ -165,7 +224,11 @@ def quickchart_write_svg(sm: StateMachine, path: str):
     >>> from tests.examples.order_control_machine import OrderControl
     >>> sm = OrderControl()
     >>> print(sm._graph().to_string())
-    digraph list {
+    digraph OrderControl {
+    label=OrderControl;
+    fontname=Arial;
+    fontsize="10pt";
+    rankdir=LR;
     ...
 
     To give you an example, we included this method that will serialize the dot, request the graph

docs/images/order_control_machine_processing.png

@@ -8,6 +8,7 @@
 from . import registry
 from .event import Event
 from .exceptions import InvalidDefinition
+from .graph import iterate_states
 from .graph import iterate_states_and_transitions
 from .graph import visit_connected_states
 from .i18n import _
@@ -30,6 +31,7 @@ def __init__(
         super().__init__(name, bases, attrs)
         registry.register(cls)
         cls.name = cls.__name__
+        cls.id = cls.name.lower()
         cls.states: States = States()
         cls.states_map: Dict[Any, State] = {}
         """Map of ``state.value`` to the corresponding :ref:`state`."""
@@ -42,12 +44,30 @@ def __init__(
 
         cls.add_inherited(bases)
         cls.add_from_attributes(attrs)
+        cls._unpack_builders_callbacks()
         cls._update_event_references()
 
-        try:
-            cls.initial_state: State = next(s for s in cls.states if s.initial)
-        except StopIteration:
-            cls.initial_state = None  # Abstract SM still don't have states
+        if not cls.states:
+            return
+
+        cls._initials_by_document_order(cls.states)
+
+        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))
+            )
+
+        if initials:
+            cls.initial_state = initials[0]
+        else:
+            cls.initial_state = None  # TODO: Check if still enter here for abstract SM
 
         cls.final_states: List[State] = [state for state in cls.states if state.final]
 
@@ -59,6 +79,26 @@ def __init__(
 
         def __getattr__(self, attribute: str) -> Any: ...
 
+    def _initials_by_document_order(cls, states):
+        """Set initial state by document order if no explicit initial state is set"""
+        has_initial = False
+        for s in states:
+            cls._initials_by_document_order(s.states)
+            if s.initial:
+                has_initial = True
+                break
+        if not has_initial and states:
+            states[0]._initial = True
+
+    def _unpack_builders_callbacks(cls):
+        callbacks = {}
+        for state in iterate_states(cls.states):
+            if state._callbacks:
+                callbacks.update(state._callbacks)
+                del state._callbacks
+        for key, value in callbacks.items():
+            setattr(cls, key, value)
+
     def _check(cls):
         has_states = bool(cls.states)
         cls._abstract = not has_states
@@ -82,8 +122,9 @@ def _check_initial_state(cls):
                     "You currently have these: {!r}"
                 ).format([s.id for s in initials])
             )
-        if not initials[0].transitions.transitions:
-            raise InvalidDefinition(_("There are no transitions."))
+        # TODO: Check if this is still needed
+        # if not initials[0].transitions.transitions:
+        #     raise InvalidDefinition(_("There are no transitions."))
 
     def _check_final_states(cls):
         final_state_with_invalid_transitions = [
@@ -205,15 +246,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.states:
+            cls.add_state(substate.id, substate)
+
     def add_event(
         cls,
         event: Event,

docs/images/readme_trafficlightmachine.png

@@ -11,10 +11,20 @@ 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)
 
 
 def iterate_states_and_transitions(states):
     for state in states:
         yield state
         yield from state.transitions
+        if state.states:
+            yield from iterate_states_and_transitions(state.states)
+
+
+def iterate_states(states):
+    for state in states:
+        yield state
+        if state.states:
+            yield from iterate_states(state.states)