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?
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())