fix: Better support for targetless transitions

Author: fgmacedo

statemachine/contrib/diagram.py

@@ -153,15 +153,20 @@ def _transition_as_edge(self, transition):
             cond = f"\n[{cond}]"
 
         extra_params = {}
-        has_substates = transition.source.states or transition.target.states
+        has_substates = transition.source.states or (
+            transition.target and transition.target.states
+        )
         if transition.source.states:
             extra_params["ltail"] = f"cluster_{transition.source.id}"
-        if transition.target.states:
+        if transition.target and transition.target.states:
             extra_params["lhead"] = f"cluster_{transition.target.id}"
 
+        targetless = transition.target is None
         return pydot.Edge(
             self._state_id(transition.source),
-            self._state_id(transition.target),
+            self._state_id(transition.target)
+            if not targetless
+            else self._state_id(transition.source),
             label=f"{transition.event}{cond}",
             color="blue",
             fontname=self.font_name,

statemachine/engines/base.py

@@ -226,7 +226,7 @@ def find_lcca(states: List[State]) -> "State | None":
 
     def get_effective_target_states(self, transition: Transition) -> OrderedSet[State]:
         # TODO: Handle history states
-        return OrderedSet([transition.target])
+        return OrderedSet([transition.target]) if transition.target else OrderedSet()
 
     def select_eventless_transitions(self, trigger_data: TriggerData):
         """
@@ -484,10 +484,10 @@ def _enter_states(  # noqa: C901
                     ).put(donedata=donedata)
 
                     if grandparent and grandparent.parallel:
-                        if all(child.final for child in grandparent.states):
-                            BoundEvent(f"done.state.{parent.id}", _sm=self.sm, internal=True).put(
-                                donedata=donedata
-                            )
+                        if all(self.is_in_final_state(child) for child in grandparent.states):
+                            BoundEvent(
+                                f"done.state.{grandparent.id}", _sm=self.sm, internal=True
+                            ).put(donedata=donedata)
         return result
 
     def compute_entry_set(
@@ -506,6 +506,8 @@ def compute_entry_set(
         for transition in transitions:
             # Process each target state of the transition
             for target_state in [transition.target]:
+                if target_state is None:
+                    continue
                 info = StateTransition(
                     transition=transition, target=target_state, source=transition.source
                 )
@@ -575,7 +577,6 @@ def add_descendant_states_to_enter(
         else:
             states_to_enter.add(info)
         state = info.target
-        assert state
 
         if state.parallel:
             for child_state in state.states:
@@ -662,3 +663,11 @@ def add_ancestor_states_to_enter(
                             states_for_default_entry,
                             default_history_content,
                         )
+
+    def is_in_final_state(self, state: State) -> bool:
+        if state.is_compound:
+            return any(s.final and s in self.sm.configuration for s in state.states)
+        elif state.parallel:
+            return all(self.is_in_final_state(s) for s in state.states)
+        else:
+            return False

statemachine/graph.py

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

statemachine/io/__init__.py

@@ -112,7 +112,8 @@ def create_machine_class_from_definition(
             for transition_data in transitions_data:
                 source = states_instances[state_id]
 
-                target = states_instances[transition_data["target"]]
+                target_state_id = transition_data["target"]
+                target = states_instances[target_state_id] if target_state_id else None
 
                 # TODO: Join `trantion_data.event` with `event_name`
                 transition = source.to(

statemachine/io/scxml/parser.py

@@ -145,16 +145,16 @@ def parse_state(
         child_state = parse_state(child_state_elem, initial_states=initial_states, is_final=True)
         state.states[child_state.id] = child_state
     for child_state_elem in state_elem.findall("parallel"):
-        state = parse_state(child_state_elem, initial_states=initial_states, is_parallel=True)
+        child_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, initial: bool = False) -> Transition:
     target = trans_elem.get("target")
-    if not target:
-        raise ValueError("Transition must have a 'target' attribute")
 
     event = trans_elem.get("event")
     cond = trans_elem.get("cond")

statemachine/io/scxml/schema.py

@@ -102,7 +102,7 @@ class ScriptAction(Action):
 
 @dataclass
 class Transition:
-    target: str
+    target: "str | None" = None
     internal: bool = False
     initial: bool = False
     event: "str | None" = None

statemachine/state.py

@@ -29,7 +29,7 @@ def __call__(self, *states: "State", **kwargs):
 
 
 class _ToState(_TransitionBuilder):
-    def __call__(self, *states: "State", **kwargs):
+    def __call__(self, *states: "State | None", **kwargs):
         transitions = TransitionList(Transition(self._state, state, **kwargs) for state in states)
         self._state.transitions.add_transitions(transitions)
         return transitions

statemachine/transition.py

@@ -40,7 +40,7 @@ class Transition:
     def __init__(
         self,
         source: "State",
-        target: "State",
+        target: "State | None" = None,
         event=None,
         internal=False,
         initial=False,
@@ -58,7 +58,7 @@ def __init__(
         self.is_self = target is source
         """Is the target state the same as the source state?"""
 
-        if internal and not (self.is_self or target.is_descendant(source)):
+        if internal and not (self.is_self or (target and target.is_descendant(source))):
             raise InvalidDefinition(
                 _(
                     "Not a valid internal transition from source {source!r}, "
@@ -89,7 +89,7 @@ def __init__(
 
     def __repr__(self):
         return (
-            f"{type(self).__name__}({self.source.name!r}, {self.target.name!r}, "
+            f"{type(self).__name__}({self.source.name!r}, {self.target and self.target.name!r}, "
             f"event={self._events!r}, internal={self.internal!r}, initial={self.initial!r})"
         )
 

tests/scxml/test_scxml_cases.py

@@ -45,7 +45,7 @@ def on_transition(self, event: Event, source: State, target: State, event_data):
                 source=f"{source and source.id}",
                 event=f"{event and event.id}",
                 data=f"{event_data.trigger_data.kwargs}",
-                target=f"{target.id}",
+                target=f"{target and target.id}",
             )
         )