feat: Delayed events; Support for SCXML <send> tag

Author: fgmacedo

statemachine/engines/async_.py

@@ -1,3 +1,6 @@
+import asyncio
+import heapq
+from time import time
 from typing import TYPE_CHECKING
 
 from ..event_data import EventData
@@ -61,7 +64,12 @@ async def processing_loop(self):
         try:
             # Execute the triggers in the queue in FIFO order until the queue is empty
             while self._external_queue:
-                trigger_data = self._external_queue.popleft()
+                trigger_data = heapq.heappop(self._external_queue)
+                current_time = time()
+                if trigger_data.execution_time > current_time:
+                    self.put(trigger_data)
+                    await asyncio.sleep(0.001)
+                    continue
                 try:
                     result = await self._trigger(trigger_data)
                     if first_result is self._sentinel:

statemachine/engines/base.py

@@ -1,4 +1,4 @@
-from collections import deque
+import heapq
 from threading import Lock
 from typing import TYPE_CHECKING
 from weakref import proxy
@@ -16,7 +16,7 @@
 class BaseEngine:
     def __init__(self, sm: "StateMachine", rtc: bool = True):
         self.sm: StateMachine = proxy(sm)
-        self._external_queue: deque = deque()
+        self._external_queue: list = []
         self._sentinel = object()
         self._rtc = rtc
         self._processing = Lock()
@@ -26,7 +26,8 @@ def put(self, trigger_data: TriggerData):
         """Put the trigger on the queue without blocking the caller."""
         if not self._running and not self.sm.allow_event_without_transition:
             raise TransitionNotAllowed(trigger_data.event, self.sm.current_state)
-        self._external_queue.append(trigger_data)
+
+        heapq.heappush(self._external_queue, trigger_data)
 
     def start(self):
         if self.sm.current_state_value is not None:

statemachine/engines/sync.py

@@ -1,3 +1,6 @@
+import heapq
+from time import sleep
+from time import time
 from typing import TYPE_CHECKING
 
 from ..event_data import EventData
@@ -47,7 +50,7 @@ def processing_loop(self):
         """
         if not self._rtc:
             # The machine is in "synchronous" mode
-            trigger_data = self._external_queue.popleft()
+            trigger_data = heapq.heappop(self._external_queue)
             return self._trigger(trigger_data)
 
         # We make sure that only the first event enters the processing critical section,
@@ -62,7 +65,12 @@ def processing_loop(self):
         try:
             # Execute the triggers in the queue in FIFO order until the queue is empty
             while self._running and self._external_queue:
-                trigger_data = self._external_queue.popleft()
+                trigger_data = heapq.heappop(self._external_queue)
+                current_time = time()
+                if trigger_data.execution_time > current_time:
+                    self.put(trigger_data)
+                    sleep(0.001)
+                    continue
                 try:
                     result = self._trigger(trigger_data)
                     if first_result is self._sentinel:

statemachine/event.py

@@ -1,4 +1,3 @@
-from inspect import isawaitable
 from typing import TYPE_CHECKING
 from typing import List
 from uuid import uuid4
@@ -8,7 +7,6 @@
 from .exceptions import InvalidDefinition
 from .i18n import _
 from .transition_mixin import AddCallbacksMixin
-from .utils import run_async_from_sync
 
 if TYPE_CHECKING:
     from .statemachine import StateMachine
@@ -44,6 +42,9 @@ class Event(AddCallbacksMixin, str):
     name: str
     """The event name."""
 
+    delay: float = 0
+    """The delay in milliseconds before the event is triggered. Default is 0."""
+
     _sm: "StateMachine | None" = None
     """The state machine instance."""
 
@@ -55,6 +56,7 @@ def __new__(
         transitions: "str | TransitionList | None" = None,
         id: "str | None" = None,
         name: "str | None" = None,
+        delay: float = 0,
         _sm: "StateMachine | None" = None,
     ):
         if isinstance(transitions, str):
@@ -66,6 +68,7 @@ def __new__(
 
         instance = super().__new__(cls, id)
         instance.id = id
+        instance.delay = delay
         if name:
             instance.name = name
         elif _has_real_id:
@@ -106,19 +109,13 @@ def __get__(self, instance, owner):
         """
         if instance is None:
             return self
-        return BoundEvent(id=self.id, name=self.name, _sm=instance)
+        return BoundEvent(id=self.id, name=self.name, delay=self.delay, _sm=instance)
 
-    def __call__(self, *args, **kwargs):
-        """Send this event to the current state machine.
-
-        Triggering an event on a state machine means invoking or sending a signal, initiating the
-        process that may result in executing a transition.
-        """
+    def put(self, *args, machine: "StateMachine", **kwargs):
         # The `__call__` is declared here to help IDEs knowing that an `Event`
         # can be called as a method. But it is not meant to be called without
         # an SM instance. Such SM instance is provided by `__get__` method when
         # used as a property descriptor.
-        machine = self._sm
         if machine is None:
             raise RuntimeError(_("Event {} cannot be called without a SM instance").format(self))
 
@@ -130,10 +127,20 @@ def __call__(self, *args, **kwargs):
             kwargs=kwargs,
         )
         machine._put_nonblocking(trigger_data)
-        result = machine._processing_loop()
-        if not isawaitable(result):
-            return result
-        return run_async_from_sync(result)
+
+    def __call__(self, *args, **kwargs):
+        """Send this event to the current state machine.
+
+        Triggering an event on a state machine means invoking or sending a signal, initiating the
+        process that may result in executing a transition.
+        """
+        # The `__call__` is declared here to help IDEs knowing that an `Event`
+        # can be called as a method. But it is not meant to be called without
+        # an SM instance. Such SM instance is provided by `__get__` method when
+        # used as a property descriptor.
+        machine = self._sm
+        self.put(*args, machine=machine, **kwargs)
+        return machine._processing_loop()
 
     def split(  # type: ignore[override]
         self, sep: "str | None" = None, maxsplit: int = -1

statemachine/event_data.py

@@ -1,5 +1,6 @@
 from dataclasses import dataclass
 from dataclasses import field
+from time import time
 from typing import TYPE_CHECKING
 from typing import Any
 
@@ -11,23 +12,36 @@
 
 
 @dataclass
+class _Data:
+    kwargs: dict
+
+    def __getattr__(self, name):
+        return self.kwargs.get(name, None)
+
+
+@dataclass(order=True)
 class TriggerData:
-    machine: "StateMachine"
+    machine: "StateMachine" = field(compare=False)
 
-    event: "Event | None"
+    event: "Event | None" = field(compare=False)
     """The Event that was triggered."""
 
-    model: Any = field(init=False)
+    execution_time: float = field(default=0.0)
+    """The time at which the :ref:`Event` should run."""
+
+    model: Any = field(init=False, compare=False)
     """A reference to the underlying model that holds the current :ref:`State`."""
 
-    args: tuple = field(default_factory=tuple)
+    args: tuple = field(default_factory=tuple, compare=False)
     """All positional arguments provided on the :ref:`Event`."""
 
-    kwargs: dict = field(default_factory=dict)
+    kwargs: dict = field(default_factory=dict, compare=False)
     """All keyword arguments provided on the :ref:`Event`."""
 
     def __post_init__(self):
         self.model = self.machine.model
+        delay = self.event.delay if self.event and self.event.delay else 0
+        self.execution_time = time() + (delay / 1000)
 
 
 @dataclass
@@ -77,3 +91,13 @@ def extended_kwargs(self):
         kwargs["source"] = self.source
         kwargs["target"] = self.target
         return kwargs
+
+    @property
+    def data(self):
+        "Property used by the SCXML namespace"
+        if self.trigger_data.kwargs:
+            return _Data(self.trigger_data.kwargs)
+        elif self.trigger_data.args and len(self.trigger_data.args) == 1:
+            return self.trigger_data.args[0]
+        else:
+            return self.trigger_data.args