Aborting an async sleep on transition

[amitoren777]

I apologize for posting a question here, but I could not post on Stack-Overflow

I'm implementing an async state machine using python-statemachine, and on entry to a state "A" I'm initiating a timer, by calling asyncio.sleep().
The exit from this state can happen in two ways:

1. The sleep expires and a transition is invoked to move the state machine to state "B"
2. An event arrived, where the sleep has to be aborted, and the machine has to move to state "C"

Is there an example of how to accomplish this ?, I'm unclear if under the RTC execution model, its possible to prematurely exit a state.

Thanks for any help.

[brunolnetto]

Try this. On example, the corner case for setUp timeout is the same as the even occurrence depends on StateMachine precedence of actions.

import nest_asyncio
import asyncio
from statemachine import State, StateMachine

# Apply nest_asyncio to allow nested event loops
nest_asyncio.apply()

class AsyncStateMachine(StateMachine):
    """A state machine to demonstrate async transitions with timers."""

    state_a = State("State A", initial=True)
    state_b = State("State B", final=True)
    state_c = State("State C", final=True)

    to_b = state_a.to(state_b)
    to_c = state_a.to(state_c)

    def __init__(self, timeout_s: int = 3):
        super().__init__()
        self.timer_task = None
        self.timeout_s = timeout_s

    async def enter_state_a(self):
        """On entering State A, start a timer."""
        print("Entering State A. Starting timer...")
        self.timer_task = asyncio.create_task(self.timer_to_b())
        await asyncio.sleep(0)  # Allow the event loop to process other tasks

    async def exit_state_a(self):
        """On exiting State A, cancel the timer."""
        if self.timer_task and not self.timer_task.done():
            print("Exiting State A. Cancelling timer...")
            self.timer_task.cancel()
            try:
                await self.timer_task
            except asyncio.CancelledError:
                pass

    async def timer_to_b(self):
        """Timer to transition to State B after the timeout."""
        try:
            await asyncio.sleep(self.timeout_s)
            print("Timer expired. Transitioning to State B.")
            if self.current_state == self.state_a:
                self.to_b()
        except asyncio.CancelledError:
            print("Timer was cancelled.")


async def event_listener(sm: AsyncStateMachine, when: int):
    """Simulates an external event triggering a transition to State C."""
    await asyncio.sleep(when)  # Simulate waiting for an external event
    print("Event received. Trying to transition to State C.")

    try:
        await sm.exit_state_a()  # Ensure timer task is canceled
        sm.to_c()
    except Exception as e:
        print(f"Failed to transition to C. Current state is {sm.current_state}.")

async def main():
    sm = AsyncStateMachine()

    # Start the state machine
    await sm.enter_state_a()

    # Simulate an external event listener
    ## Final state: C
    # event_at = 2

    # Final state: B
    event_at = 3
    
    ## Final state: B
    # event_at = 4
    await event_listener(sm, event_at)

    print(f"Final State: {sm.current_state}")

# Run the async main loop directly
await main()

Happy New Year! 🎆