I recently spotted a lightweight, object-oriented state machine implementation in Python called transitions (https://github.com/pytransitions/transitions). So I'm trying to play a bit with these state machines and especially with the HierarchicalGraphMachine. One the the nice features that I'd like to have is to store the history of the visited states even when the machine does not move (stays in the same state).
And from what I see from the examples, we actually could not do it in a easy way because the before_state_change and after_state_change was NOT called when the machine's state is unchanged. So we cannot extend our history in this case. In order to fix that, I end up by creating a trigger_wrapper function:
def trigger_wrapper(self, trigger_name):
previous_state = self.state
result = None
try:
result = self.trigger(trigger_name)
except AttributeError as attribute_err:
print('Invalid trigger name: {}'.format(attribute_err))
except MachineError as machine_err:
print('Valid trigger name but not reachable: {}'.format(machine_err))
except Exception as err:
print('Cannot make transition with unknown error: {}'.format(err))
if result is False:
print('Trigger name reachable but condition(s) was not fulfilled')
....
current_state = self.state
# update history
.....
return result
Then, we call trigger_wrapper instead of trigger:
before: machine.trigger('drink')
now: machine.trigger_wrapper('drink').
In addition to that, by setting ignore_invalid_triggers = False when initialize the Machine and use this trigger_wrapper function, we could now know the reason why the machine cannot make a move by caching the exceptions.
Is there any better solution for keep tracking the visited state? An another approach I think is to overwrite trigger function but it seems complicated due to the NestedState.
Edit 1 (follow to the suggestion by @aleneum)
Thank you for your response together with an interesting example !!!
Follow the example that using the finalize_event. It goes well, but this callback function seems do not enough to catch the following cases (I added 2 extra lines in the code):
... same setup as before
m.go()
m.internal()
m.reflexive()
m.condition()
m.go() # MachineError: "Can't trigger event go from state B!"
m.goo() # AttributeError: Do not know event named 'goo'.
>>> Expected: ['go', 'internal', 'reflexive', 'condition', 'go', 'goo']
>>> Expected: ['B', 'B', 'B', 'B', 'B', 'B']
In other words, is there another callback that we could catch the exceptions caused by calling invalid trigger (goo in the example) or caused by valid trigger but not reachable from the current state (call go() from state B) ?
Thanks again for your help.
As you already mentioned, before_state_change and after_state_change are only called when a transition takes place. This does not necessarily mean a state change though as internal and reflexive transitions also trigger these callbacks:
from transitions import Machine
def test():
print("triggered")
m = Machine(states=['A', 'B'], transitions=[
['go', 'A', 'B'],
dict(trigger='internal', source='B', dest=None),
dict(trigger='reflexive', source='B', dest='='),
dict(trigger='condition', source='B', dest='A', conditions=lambda: False)
], after_state_change=test, initial='A')
m.go() # >>> triggered
m.internal() # >>> triggered
m.reflexive() # >>> triggered
m.condition() # no output
The only event that does NOT trigger the after_state_change here is m.condition since the transition was halted by the (unfulfilled) condition.
So, when your goal is to track actually conducted transitions, after_state_change is the right spot. If you want to log every trigger/event, you can do this via finalize_event:
'machine.finalize_event' - callbacks will be executed even if no transition took place or an exception has been raised
from transitions import Machine
event_log = []
state_log = []
def log_trigger(event_data):
event_log.append(event_data.event.name)
state_log.append(event_data.state)
m = Machine(states=['A', 'B'], transitions=[
['go', 'A', 'B'],
dict(trigger='internal', source='B', dest=None),
dict(trigger='reflexive', source='B', dest='='),
dict(trigger='condition', source='B', dest='A', conditions=lambda event_data: False)
], finalize_event=log_trigger, initial='A', send_event=True)
m.go()
m.internal()
m.reflexive()
m.condition()
print(event_log) # >>> ['go', 'internal', 'reflexive', 'condition']
print([state.name for state in state_log]) # >>> ['B', 'B', 'B', 'B']
Callbacks passed to finalize_event will always be called, even if the transition raised an exception. By setting send_event=True, all callbacks will receive an EvenData object which contains event, state and transition information as well as an error object if something went wrong. This is way I have to change the condition lambda expression. When send_event=True, ALL callbacks need to be able to process the EventData object.
More information about finalize_event and the callback execution order can be found in this section of the documentation.
finalize_event is only called for valid events which means the event must exist and must also be valid on the current source state. If ALL events should be processed, Machine needs to be extended:
from transitions import Machine
log = []
class LogMachine(Machine):
def _get_trigger(self, model, trigger_name, *args, **kwargs):
res = super(LogMachine, self)._get_trigger(model, trigger_name, *args, **kwargs)
log.append((trigger_name, model.state))
return res
# ...
m = LogMachine(states=..., ignore_invalid_triggers=True)
assert m.trigger("go") # valid
assert not m.trigger("go") # invalid
assert not m.trigger("gooo") # invalid
print(log) # >>> [('go', 'B'), ('go', 'B'), ('gooo', 'B')]
Every model is decorated with a trigger method which is a partial of Machine._get_trigger with assigned model parameter. Model.trigger can be used to trigger events by name and also to process non-existent events. You also need to pass ignore_invalid_triggers=True to not raise MachineError when an event is invalid.
However, if all events should be logged, it is probably more feasible/maintainable to split logging from Machine and handle logging where events are processed, e.g.:
m = Machine(..., ignore_invalid_triggers=True)
# ...
def on_event(event_name):
logging.debug(f"Received event {event_name}") # or log_event.append(event_name)
m.trigger(event_name)
logging.debug(f"Machine state is {m.state}") # or log_state.append(m.state)