python-asyncioasynccallbackpytransitions

How to use AsyncState class of pytransitions Asyncio extension explicitly?


I'm pretty new to Python, asyncio and pytransitions, so I apologize if I'm asking stupid questions.

I'm familiar with GoF's state design pattern and I'm wondering how I can implement this pattern with AsyncMachine and AsyncState. The goal is a standardized interface for my class MyBaseAsyncState(AsyncState).

In the first step I want to start from example "Very asynchronous dancing" and try to use the class AsyncState explicitly. I changed the original:

from transitions.extensions.asyncio import AsyncMachine
import asyncio

class Dancer:
    
    states = ['start', 'left_food_left', 'left', 'right_food_right']
    
    def __init__(self, name, beat):
        self.my_name = name
        self.my_beat = beat
        self.moves_done = 0
        
    async def on_enter_start(self):
        self.moves_done += 1
        
    async def wait(self):
        print(f'{self.my_name} stepped {self.state}')
        await asyncio.sleep(self.my_beat)

    async def dance(self):
        while self.moves_done < 5:
            await self.step()
        
dancer1 = Dancer('Tick', 1)
dancer2 = Dancer('Tock', 1.1)

m = AsyncMachine(model=[dancer1, dancer2], states=Dancer.states, initial='start', after_state_change='wait')
m.add_ordered_transitions(trigger='step')

# it starts okay but becomes quite a mess
_ = await asyncio.gather(dancer1.dance(), dancer2.dance()) 

to this:

from transitions.extensions.asyncio import AsyncMachine, AsyncState, AsyncEvent
import asyncio

class StartState(AsyncState):
    def __init__(self, name, on_enter=None, on_exit = None, ignore_invalid_trigger = True, final = False) -> None:
        super().__init__(name, on_enter, on_exit, ignore_invalid_triggers, final)
        self.add_callback('enter', self.on_enter)

    async def on_enter(self, eventdata: AsyncEvent):
        print("On_enter start state")
        eventdata.model.moves_done += 1

class Dancer:
    
    states = [StartState(name='start'), 'left_food_left', 'left', 'right_food_right']
    
    def __init__(self, name, beat):
        self.my_name = name
        self.my_beat = beat
        self.moves_done = 0
        
    # async def on_enter_start(self):
    #     self.moves_done += 1
        
    async def wait(self):
        print(f'{self.my_name} stepped {self.state}')
        await asyncio.sleep(self.my_beat)

    async def dance(self):
        while self.moves_done < 10:
            await self.step()
        
dancer1 = Dancer('Tick', 1)

m = AsyncMachine(model=[dancer1], states=Dancer.states, initial='start', after_state_change='wait')
m.add_ordered_transitions(trigger='step')
 
await asyncio.gather(dancer1.dance())

I would have expected StartState:on_enter() to be called automatically. Because that doesn't happen, I'm trying to add it explicitly in __init__. But I get the typeerror: 'list' object is not callable

What is my misstake?


Solution

  • There is three small things that need to be adjusted.

    First, on_enter and on_exit are 'reserved' in (Async)States. They are considered dynamic methods and get a special treatment. So, when you change async def on_enter to (for instance) on_enter_state the list error should go away.

    Second, if you want 'ordinary' callbacks to receive AsyncEventData you need to pass send_event=True to the machine constructor (as mentioned in the Callbacks section of the documentation). This will, however, require all callbacks to have an appropriate signature (as of transitions 0.9.2). So you need to change async def wait(self) to async def wait(self, event_data: AsyncEventData).

    Third, a machine passes an instance of type (Async)EventData to the callback and not the (Async)Event itself.

    Your code looks somewhat like this with the aforementioned changes:

    from transitions.extensions.asyncio import AsyncMachine, AsyncState, AsyncEventData
    import asyncio
    
    
    class StartState(AsyncState):
        def __init__(self, name, on_enter=None, on_exit=None, ignore_invalid_trigger=True, final=False) -> None:
            super().__init__(name, on_enter, on_exit, ignore_invalid_trigger, final)
            self.add_callback('enter', self.on_enter_state)
    
        async def on_enter_state(self, eventdata: AsyncEventData):
            print("On_enter start state")
            eventdata.model.moves_done += 1
    
    
    class Dancer:
        states = [StartState(name='start'), 'left_food_left', 'left', 'right_food_right']
    
        def __init__(self, name, beat):
            self.my_name = name
            self.my_beat = beat
            self.moves_done = 0
    
        # async def on_enter_start(self):
        #     self.moves_done += 1
    
        async def wait(self, event_data: AsyncEventData):
            print(f'{self.my_name} stepped {self.state}')
            await asyncio.sleep(self.my_beat)
    
        async def dance(self):
            while self.moves_done < 10:
                await self.step()
    
    
    dancer1 = Dancer('Tick', 1)
    
    m = AsyncMachine(model=[dancer1], states=Dancer.states, initial='start', after_state_change='wait', send_event=True)
    m.add_ordered_transitions(trigger='step')
    
    async def main():
        await asyncio.gather(dancer1.dance())
    
    asyncio.run(main())