depth-first-searchpytransitions

pytransitions is there a simple way to get the history of triggered events


class Matter(object):
    def __init__(self, states, transitions):
        self.states = states
        self.transitions = transitions
        self.machine = Machine(model=self, states=self.states, transitions=transitions, initial='liquid')
    def get_triggered_events(self, source, dest):
        self.machine.set_state(source)
        eval("self.to_{}()".format(dest))
        return
states=['solid', 'liquid', 'gas', 'plasma']
transitions = [
{ 'trigger': 'melt', 'source': 'solid', 'dest': 'liquid' },
{ 'trigger': 'evaporate', 'source': 'liquid', 'dest': 'gas' },
{ 'trigger': 'sublimate', 'source': 'solid', 'dest': 'gas' },
{ 'trigger': 'ionize', 'source': 'gas', 'dest': 'plasma' }

]

matter=Matter(states,transitions)
matter.get_triggered_events("solid","plasma")

I want to get the history of triggered events from source to destination in get_triggered_events method. E.g. runing matter.get_triggered_events("solid","plasma") will get [["melt","evaporate","ionize"],["sublimate","ionize"]]. Is there a simple way to achieve it?


Solution

  • I guess what you intend to do is to get all possible paths from one state to another. You are interested in the names of the events that must be emitted/triggered to reach your target.

    A solution is to traverse through all events that can be triggered in a state. Since one event can result in multiple transitions we need to loop over a list of transitions as well. With Machine.get_triggers(<state_name>) we'll get a list of all events that can be triggered from <state_name>. With Machine.get_transitions(<event_name>, source=<state_name>) we will get a list of all transitions that are associated with <event_name> and can be triggered from <state_name>.

    We basically feed the result from Machine.get_triggers to Machine.get_transitions and loop over the list of transitions while keeping track of the events that have been processed so far. Furthermore, to prevent cyclic traversal, we also keep track of the transition entities we have already visited:

    from transitions import Machine
    
    
    states = ['solid', 'liquid', 'gas', 'plasma']
    transitions = [
        {'trigger': 'melt', 'source': 'solid', 'dest': 'liquid'},
        {'trigger': 'evaporate', 'source': 'liquid', 'dest': 'gas'},
        {'trigger': 'sublimate', 'source': 'solid', 'dest': 'gas'},
        {'trigger': 'ionize', 'source': 'gas', 'dest': 'plasma'}
    ]
    
    
    class TraverseMachine(Machine):
        def traverse(self, current_state, target_state, seen=None, events=None):
            seen = seen or []
            events = events or []
            # we reached our destination and return the list of events that brought us here
            if current_state == target_state:
                return events
            paths = [self.traverse(t.dest, target_state, seen + [t], events + [e])
                     for e in self.get_triggers(current_state)
                     for t in self.get_transitions(e, source=current_state)
                     if t not in seen]
            # the len check is meant to prevent deep nested return values
            # if you just return path the result will be:
            # [[[['melt', 'evaporate', 'ionize']]], [['sublimate', 'ionize']]]
            return paths[0] if len(paths) == 1 and isinstance(paths[0], list) else paths
    
    # disable auto transitions!
    # otherwise virtually every path ending with `to_<target_state>` is a valid solution
    m = TraverseMachine(states=states, transitions=transitions, initial='solid', auto_transitions=False)
    print(m.traverse("solid", "plasma"))
    # [['melt', 'evaporate', 'ionize'], ['sublimate', 'ionize']]
    

    You might want to have a look at the discussion in transitions issue tracker about automatic traversal as it contains some insights about conditional traversal.