pythontransitionfsmpytransitions

How to pass event parameters in condition function in pytransition


I am trying to pass a parameter to conditions. But it was giving the following error

code:

from transitions import Machine

class evenodd(object):
    def get_no(self, event):
        self.no = event.kwargs.get('no', 0)
    def calc_mod(self, event):
        self.mod = event.kwargs.get('mod', self.no%2)
    
    def is_even(self, event):
        if self.mod == 0:
            return True
        
obj = evenodd()
machine = Machine(obj, ['init', 'getno', 'even', 'odd'], send_event=True, initial='init', ignore_invalid_triggers=True, auto_transitions=False)
machine.add_transition('enterno', 'init', 'getno', before='get_no')
machine.add_transition('isevenodd', 'getno', 'even', before='calc_mod', conditions=['is_even'])
machine.add_transition('isevenodd', 'getno', 'odd', before='calc_mod', conditions=['is_odd']) 


s_state = obj.state
print("state --> "+s_state)
trigger = machine.get_triggers(s_state)[0]
print("transition --> "+trigger)
obj.enterno(2)
s_state = obj.state
print("state --> "+s_state)
trigger = machine.get_triggers(s_state)[0]
print("transition --> "+trigger)
obj.isevenodd()
s_state = obj.state
print("state --> "+s_state)

Error:

state --> init
transition --> enterno
state --> getno
transition --> isevenodd
---------------------------------------------------------------------------
AttributeError                            Traceback (most recent call last)
<ipython-input-75-d8ede3775dc5> in <module>
      8 trigger = machine.get_triggers(s_state)[0]
      9 print("transition --> "+trigger)
---> 10 obj.isevenodd()
     11 s_state = obj.state
     12 print("state --> "+s_state)

~\Anaconda3\envs\ml_nlp_cpu\lib\site-packages\transitions\core.py in trigger(self, model, *args, **kwargs)
    388         # Machine._process should not be called somewhere else. That's why it should not be exposed
    389         # to Machine users.
--> 390         return self.machine._process(func)
    391 
    392     def _trigger(self, model, *args, **kwargs):

~\Anaconda3\envs\ml_nlp_cpu\lib\site-packages\transitions\core.py in _process(self, trigger)
   1112             if not self._transition_queue:
   1113                 # if trigger raises an Error, it has to be handled by the Machine.process caller
-> 1114                 return trigger()
   1115             else:
   1116                 raise MachineError("Attempt to process events synchronously while transition queue is not empty!")

~\Anaconda3\envs\ml_nlp_cpu\lib\site-packages\transitions\core.py in _trigger(self, model, *args, **kwargs)
    406                 raise MachineError(msg)
    407         event_data = EventData(state, self, self.machine, model, args=args, kwargs=kwargs)
--> 408         return self._process(event_data)
    409 
    410     def _process(self, event_data):

~\Anaconda3\envs\ml_nlp_cpu\lib\site-packages\transitions\core.py in _process(self, event_data)
    415             for trans in self.transitions[event_data.state.name]:
    416                 event_data.transition = trans
--> 417                 if trans.execute(event_data):
    418                     event_data.result = True
    419                     break

~\Anaconda3\envs\ml_nlp_cpu\lib\site-packages\transitions\core.py in execute(self, event_data)
    260         _LOGGER.debug("%sExecuted callbacks before conditions.", event_data.machine.name)
    261 
--> 262         if not self._eval_conditions(event_data):
    263             return False
    264 

~\Anaconda3\envs\ml_nlp_cpu\lib\site-packages\transitions\core.py in _eval_conditions(self, event_data)
    241     def _eval_conditions(self, event_data):
    242         for cond in self.conditions:
--> 243             if not cond.check(event_data):
    244                 _LOGGER.debug("%sTransition condition failed: %s() does not return %s. Transition halted.",
    245                               event_data.machine.name, cond.func, cond.target)

~\Anaconda3\envs\ml_nlp_cpu\lib\site-packages\transitions\core.py in check(self, event_data)
    179         predicate = event_data.machine.resolve_callable(self.func, event_data)
    180         if event_data.machine.send_event:
--> 181             return predicate(event_data) == self.target
    182         return predicate(*event_data.args, **event_data.kwargs) == self.target
    183 

<ipython-input-74-5dc86072aabd> in is_even(self, event)
      7     def is_even(self, event):
      8         """ Basically a coin toss. """
----> 9         if self.mod == 0:
     10             return True
     11 

AttributeError: 'evenodd' object has no attribute 'mod'

How to share the variables no, and mod with each of these callbacks. I tried to use the event. In this simple example, I tried to create a state machine to reach the state based on whether the given input is even or odd.


Solution

  • You can share variables as attributes of the model. You need to consider callback execution order though. According to transitions documentation callbacks are executed in that order:

    You can see that before is triggered after conditions have been evaluated and the transition will definitely happen. Thus, self.mod is not available in is_even since calc_mod has not been called yet. The list also shows how to deal with this: Execute calc_mod in prepare instead of before:

    machine.add_transition('isevenodd', 'getno', 'even', prepare='calc_mod', conditions=['is_even'])
    machine.add_transition('isevenodd', 'getno', 'odd', prepare='calc_mod', conditions=['is_odd'])
    

    prepare has been introduced exactly for that use case where condition checks need some setup. If 'tear down' is also required, you can use machine.finalize_event which will always be called regardless whether a transition took place or not. Speaking of callback resolution order: You can pass condition checks to unless which will halt a transition if they evaluate to true. You could replace conditions='is_odd' with unless='is_even'.

    If you don't mind some implicit logic you could discard the 'is_odd' check entirely. A set of valid triggers is always evaluated in the order they were added. This means that 'getno' -> 'odd' will only be considered when 'getno' -> 'even' has been found invalid. Design-wise, a transition without conditions which is added last will act as an 'else' clause which will be executed when no previous set of conditions could be met.

    How to pass variables to callbacks

    One way is to process the event object which is passed to callback when send_event=True on the machine. You can also pass variables as trigger parameters:

    from transitions import Machine
    
    
    class evenodd(object):
    
        def get_no(self, no=0, **kwargs):
            self.no = no
    
        def calc_mod(self, mod=None, **kwargs):
            self.mod = mod if mod else self.no % 2
    
        def is_even(self):
            return self.mod == 0
    
    
    obj = evenodd()
    machine = Machine(obj, ['init', 'getno', 'even', 'odd'], initial='init', 
                      ignore_invalid_triggers=True, auto_transitions=False)
    machine.add_transition('enterno', 'init', 'getno', before='get_no')
    machine.add_transition('isevenodd', 'getno', 'even', prepare='calc_mod', conditions=['is_even'])
    machine.add_transition('isevenodd', 'getno', 'odd', prepare='calc_mod', conditions=['is_odd'])
    
    obj.enterno(no=2)
    obj.isevenodd()
    print("state --> " + obj.state)
    

    If you do that you must make sure that every callback can consume all parameters thrown at them. Using kwargs makes it quite easy though to 'discard' not required arguments. I'd also suggest always passing parameters as keyword arguments even though positional arguments are supported, too. In my experience, this makes code more comprehensible (e.g. easier to read method definitions) and reduces a source of error (e.g. wrong parameter mapping).