feat: Support for SCXML foreach tag

Author: fgmacedo

statemachine/engines/sync.py

@@ -76,7 +76,7 @@ def processing_loop(self):
             self._processing.release()
         return first_result if first_result is not self._sentinel else None
 
-    def _trigger(self, trigger_data: TriggerData):
+    def _trigger(self, trigger_data: TriggerData):  # noqa: C901
         executed = False
         if trigger_data.event == "__initial__":
             transition = self._initial_transition(trigger_data)
@@ -92,6 +92,8 @@ def _trigger(self, trigger_data: TriggerData):
             if not executed:
                 continue
 
+            if self.sm.current_state.transitions.has_eventless_transition:
+                self.put(TriggerData(self.sm, event=None))
             break
         else:
             if not self.sm.allow_event_without_transition:

statemachine/event_data.py

@@ -14,7 +14,7 @@
 class TriggerData:
     machine: "StateMachine"
 
-    event: "Event"
+    event: "Event | None"
     """The Event that was triggered."""
 
     model: Any = field(init=False)

statemachine/events.py

@@ -32,8 +32,14 @@ def add(self, events):
         return self
 
     def match(self, event: str):
-        return any(e == event for e in self)
+        if event is None and self.is_empty:
+            return True
+        return any(e == event or e == "*" for e in self)
 
     def _replace(self, old, new):
         self._items.remove(old)
         self._items.append(new)
+
+    @property
+    def is_empty(self):
+        return len(self._items) == 0

statemachine/exceptions.py

@@ -32,8 +32,10 @@ class AttrNotFound(InvalidDefinition):
 class TransitionNotAllowed(StateMachineError):
     "Raised when there's no transition that can run from the current :ref:`state`."
 
-    def __init__(self, event: "Event", state: "State"):
+    def __init__(self, event: "Event | None", state: "State"):
         self.event = event
         self.state = state
-        msg = _("Can't {} when in {}.").format(self.event.name, self.state.name)
+        msg = _("Can't {} when in {}.").format(
+            self.event and self.event.name or "transition", self.state.name
+        )
         super().__init__(msg)

statemachine/io/__init__.py

@@ -6,7 +6,7 @@
 from ..transition_list import TransitionList
 
 
-def create_machine_class_from_definition(name: str, definition: dict) -> StateMachine:
+def create_machine_class_from_definition(name: str, definition: dict) -> StateMachine:  # noqa: C901
     """
     Creates a StateMachine class from a dictionary definition, using the StateMachineMetaclass.
 
@@ -16,32 +16,30 @@ def create_machine_class_from_definition(name: str, definition: dict) -> StateMa
     ...     "TrafficLightMachine",
     ...     {
     ...         "states": {
-    ...             "green": {"initial": True},
-    ...             "yellow": {},
-    ...             "red": {},
-    ...         },
-    ...         "events": {
-    ...             "change": [
-    ...                 {"from": "green", "to": "yellow"},
-    ...                 {"from": "yellow", "to": "red"},
-    ...                 {"from": "red", "to": "green"},
-    ...             ]
+    ...             "green": {"initial": True, "on": {"change": {"target": "yellow"}}},
+    ...             "yellow": {"on": {"change": {"target": "red"}}},
+    ...             "red": {"on": {"change": {"target": "green"}}},
     ...         },
     ...     }
     ... )
 
     """
+    states_instances: Dict[str, State] = {}
+    events_definitions: Dict[str, dict] = {}
+
+    for state_id, state_kwargs in definition.pop("states").items():
+        on_events = state_kwargs.pop("on", {})
+        if on_events:
+            events_definitions[state_id] = on_events
 
-    states_instances = {
-        state_id: State(**state_kwargs)
-        for state_id, state_kwargs in definition.pop("states").items()
-    }
+        states_instances[state_id] = State(**state_kwargs)
 
     events: Dict[str, TransitionList] = {}
-    for event_name, transitions in definition.pop("events").items():
-        for transition_data in transitions:
-            source = states_instances[transition_data["from"]]
-            target = states_instances[transition_data["to"]]
+    for state_id, state_events in events_definitions.items():
+        for event_name, transition_data in state_events.items():
+            source = states_instances[state_id]
+
+            target = states_instances[transition_data["target"]]
 
             transition = source.to(
                 target,

statemachine/io/scxml.py

@@ -12,17 +12,17 @@
 from statemachine.statemachine import StateMachine
 
 
-def send_event(machine: StateMachine, event_to_send: str) -> None:
-    machine.send(event_to_send)
-
-
 def parse_onentry(element):
     """Parses the <onentry> XML into a callable."""
     actions = [parse_element(child) for child in element]
 
     def execute_block(*args, **kwargs):
-        for action in actions:
-            action(*args, **kwargs)
+        machine = kwargs["machine"]
+        try:
+            for action in actions:
+                action(*args, **kwargs)
+        except Exception:
+            machine.send("error.execution")
 
     return execute_block
 
@@ -34,6 +34,8 @@ def parse_element(element):
         return parse_raise(element)
     elif tag == "assign":
         return parse_assign(element)
+    elif tag == "foreach":
+        return parse_foreach(element)
     elif tag == "log":
         return parse_log(element)
     elif tag == "if":
@@ -80,6 +82,58 @@ def assign_action(*args, **kwargs):
     return assign_action
 
 
+def parse_foreach(element):  # noqa: C901
+    """
+    Parses the <foreach> element into a callable.
+
+    - `array`: The iterable collection (required).
+    - `item`: The variable name for the current item (required).
+    - `index`: The variable name for the current index (optional).
+    - Child elements are executed for each iteration.
+    """
+    array_expr = element.attrib.get("array")
+    if not array_expr:
+        raise ValueError("<foreach> must have an 'array' attribute")
+
+    item_var = element.attrib.get("item")
+    if not item_var:
+        raise ValueError("<foreach> must have an 'item' attribute")
+
+    index_var = element.attrib.get("index")
+    child_actions = [parse_element(child) for child in element]
+
+    def foreach_action(*args, **kwargs):  # noqa: C901
+        machine = kwargs["machine"]
+        context = {**machine.model.__dict__}  # Shallow copy of the model's attributes
+
+        try:
+            # Evaluate the array expression to get the iterable
+            array = eval(array_expr, {}, context)
+            if not hasattr(array, "__iter__"):
+                raise ValueError(
+                    f"<foreach> 'array' must evaluate to an iterable, got: {type(array).__name__}"
+                )
+        except Exception as e:
+            raise ValueError(f"Error evaluating <foreach> 'array' expression: {e}") from e
+
+        if not item_var.isidentifier():
+            raise ValueError(
+                f"<foreach> 'item' must be a valid Python attribute name, got: {item_var}"
+            )
+        # Iterate over the array
+        for index, item in enumerate(array):
+            # Assign the item and optionally the index
+            setattr(machine.model, item_var, item)
+            if index_var:
+                setattr(machine.model, index_var, index)
+
+            # Execute child actions
+            for action in child_actions:
+                action(*args, **kwargs)
+
+    return foreach_action
+
+
 def _normalize_cond(cond: "str | None") -> "str | None":
     """
     Normalizes a JavaScript-like condition string to be compatible with Python's eval.
@@ -108,6 +162,19 @@ def _normalize_cond(cond: "str | None") -> "str | None":
     return pattern.sub(lambda match: replacements[match.group(0)], cond)
 
 
+def parse_cond(cond):
+    """Parses the <cond> element into a callable."""
+    cond = _normalize_cond(cond)
+    if cond is None:
+        return None
+
+    def cond_action(*args, **kwargs):
+        machine = kwargs["machine"]
+        return eval(cond, {}, {"machine": machine, **machine.model.__dict__})
+
+    return cond_action
+
+
 def parse_if(element):  # noqa: C901
     """Parses the <if> element into a callable."""
     branches = []
@@ -198,6 +265,8 @@ def parse_data(element):
     """
     data_id = element.attrib["id"]
     expr = element.attrib.get("expr")
+    if not expr:
+        expr = element.text and element.text.strip()
 
     def data_initializer(model):
         # Evaluate the expression if provided, or set to None
@@ -230,19 +299,8 @@ def parse_scxml(scxml_content: str) -> Dict[str, Any]:  # noqa: C901
     Parse SCXML content and return a dictionary definition compatible with
     create_machine_class_from_definition.
 
-    The returned dictionary has the format:
-    {
-        "states": {
-            "state_id": {"initial": True},
-            ...
-        },
-        "events": {
-            "event_name": [
-                {"from": "source_state", "to": "target_state"},
-                ...
-            ]
-        }
-    }
+    The returned dictionary has the format compatible with
+    :ref:`create_machine_class_from_definition`.
     """
     # Parse XML content
     root = ET.fromstring(scxml_content)
@@ -258,7 +316,6 @@ def parse_scxml(scxml_content: str) -> Dict[str, Any]:  # noqa: C901
 
     # Build states dictionary
     states = {}
-    events: Dict[str, List[Dict[str, str]]] = {}
 
     def _parse_state(state_elem, final=False):  # noqa: C901
         state_id = state_elem.get("id")
@@ -272,21 +329,21 @@ def _parse_state(state_elem, final=False):  # noqa: C901
         for trans_elem in state_elem.findall("transition"):
             event = trans_elem.get("event") or None
             target = trans_elem.get("target")
+            cond = parse_cond(trans_elem.get("cond"))
 
             if target:
-                if event not in events:
-                    events[event] = []
+                state = states[state_id]
+                if "on" not in state:
+                    state["on"] = {}
+
+                if event not in state["on"]:
+                    state["on"][event] = {"target": target}
+                    if cond:
+                        state["on"][event]["cond"] = cond
 
                 if target not in states:
                     states[target] = {}
 
-                events[event].append(
-                    {
-                        "from": state_id,
-                        "to": target,
-                    }
-                )
-
         for onentry_elem in state_elem.findall("onentry"):
             entry_action = parse_onentry(onentry_elem)
             state = states[state_id]
@@ -316,4 +373,4 @@ def _parse_state(state_elem, final=False):  # noqa: C901
         first_state = next(iter(states))
         states[first_state]["initial"] = True
 
-    return {"states": states, "events": events, **extra_data}
+    return {"states": states, **extra_data}

statemachine/statemachine.py

@@ -82,7 +82,7 @@ def __init__(
         self._callbacks = CallbacksRegistry()
         self._states_for_instance: Dict[State, State] = {}
 
-        self._listeners: Dict[Any, Any] = {}
+        self._listeners: Dict[int, Any] = {}
         """Listeners that provides attributes to be used as callbacks."""
 
         if self._abstract:
@@ -182,7 +182,7 @@ def _add_listener(self, listeners: "Listeners", allowed_references: SpecReferenc
         return self
 
     def _register_callbacks(self, listeners: List[object]):
-        self._listeners.update({listener: None for listener in listeners})
+        self._listeners.update({id(listener): listener for listener in listeners})
         self._add_listener(
             Listeners.from_listeners(
                 (
@@ -223,9 +223,9 @@ def add_listener(self, *listeners):
 
             :ref:`listeners`.
         """
-        self._listeners.update({o: None for o in listeners})
+        self._listeners.update({id(listener): listener for listener in listeners})
         return self._add_listener(
-            Listeners.from_listeners(Listener.from_obj(o) for o in listeners),
+            Listeners.from_listeners(Listener.from_obj(listener) for listener in listeners),
             allowed_references=SPECS_SAFE,
         )
 

statemachine/transition.py

@@ -145,3 +145,7 @@ def _copy_with_args(self, **kwargs):
             new_transition._specs.add(new_spec, new_spec.group)
 
         return new_transition
+
+    @property
+    def is_eventless(self):
+        return self._events.is_empty

statemachine/transition_list.py

@@ -122,3 +122,7 @@ def unique_events(self) -> List["Event"]:
                 tmp_ordered_unique_events_as_keys_on_dict[event] = True
 
         return list(tmp_ordered_unique_events_as_keys_on_dict.keys())
+
+    @property
+    def has_eventless_transition(self):
+        return any(transition.is_eventless for transition in self.transitions)

tests/w3c_tests/test_testcases.py

@@ -1,4 +1,9 @@
+from dataclasses import dataclass
+from dataclasses import field
+
+from statemachine import State
 from statemachine import StateMachine
+from statemachine.event import Event
 
 """
 Test cases as defined by W3C SCXML Test Suite
@@ -11,12 +16,21 @@
 """  # noqa: E501
 
 
+@dataclass(frozen=True, unsafe_hash=True)
+class DebugListener:
+    events: list = field(default_factory=list)
+
+    def on_transition(self, event: Event, source: State, target: State):
+        self.events.append(f"{source and source.id} --({event and event.id})--> {target.id}")
+
+
 def test_usecase(testcase_path, sm_class):
     # from statemachine.contrib.diagram import DotGraphMachine
 
     # DotGraphMachine(sm_class).get_graph().write_png(
     #     testcase_path.parent / f"{testcase_path.stem}.png"
     # )
-    sm = sm_class()
+    debug = DebugListener()
+    sm = sm_class(listeners=[debug])
     assert isinstance(sm, StateMachine)
-    assert sm.current_state.id == "pass"
+    assert sm.current_state.id == "pass", debug