Defining and implementing finite state machines in Python with the transitions module

1   Introduction

The transitions package makes it convenient and relatively easy to define and implement FSMs (finite state machines) in python. For information about transitions, see: https://github.com/tyarkoni/transitions.

You can install transitions with pip:

$ pip install transitions

2   Implementing an FSM with transitions

First, we need a conceptual picture or model of an FSM that will help make clear how to implement an FSM. So:

• An FSM is a set of states and a set of transitions.
• We can represent an FSM as a digraph (a directed graph) in which the states are represented by nodes in the graph and transitions are represented by the edges. Each edge may be annotated with (1) a condition that must be satisfied in order for that transition to be selected and (2) an action to be performed when that transition is performed.
• So, an edge says, in effect: If you are in this state (the from-state or source state) and this condition is true, then perform this action and put the FSM in that state (the to-state or destination state).
• Therefore, we can represent the edges/transitions as a set of rules.

Here is a state-transition table that shows what that set of rules might be like:

Our first task is to learn how to transform a table like the above into a transitions FSM.

So, here is another state-transition table that defines the FSM that we will implement. This FSM reads a sequence of characters and transforms any characters inside double quotes to upper-case, leaving other characters unchanged. I've numbered the rules so that we can identify them in the code that follows:

So, now, we can begin the process of translating our table of rules into an implementation of a transitions FSM:

1. Choose the structure for our FSM class. The transitions documentation talks about a model (a class that we implement and the Machine (a class implemented by transitions. There are several alternative ways in which the model and the machine (an instance of class Machine) can relate or be connected to each other. (see: https://github.com/tyarkoni/transitions#alternative-initialization-patterns). However, we'll pick the organization that is in the "Quickstart" section, where the model class creates an instance of class Machine, informs the machine that it is the model, and holds the instance of class machine in an instance variable.

2. Define the states -- Here is what the first part of our model class might look like:

   class ConvertString(object):
       """FSM which converts strings.  Characters inside double quotes
       are converted to upper case.
       """
   
       states = ['start', 'in_quotes', 'not_in_quotes', 'error', 'end', ]
   

3. Create the machine -- Next we'll write our constructor method (__init__), and create an instance of class transitions.Machine:

   def __init__(self):
       self.machine = Machine(
           model=self,
           states=self.states,
           initial='start')
   

   Notes:

   • When we create the machine, we pass it the model (which is the current instance) and a list of the names of the states and the name of the initial or start state.
   • There are alternative ways of specifying the states. For example, we could have passed a list of transitions.State objects, which would have allowed us to specify callbacks to be called when that state is entered or exited. See the transitions documentation on States.

4. Add the transitions -- Next we add a transition for each rule in our table of rules/transitions. Doing this is pretty straight-forward. We add one call to self.machine.add_transition(...) for each row in our state-transition table above. If you inspect the following code, you'll see that:

   • The "Source" column is passed as parameter source.
   • The "Condition" column is passed as parameter conditions.
   • The "Action" column is passed as parameter after.
   • The "Destination" column is passed as parameter dest.

   Here is the code:

   def __init__(self):
       self.machine = Machine(
           model=self,
           states=self.states,
           initial='start')
       self.machine.add_transition(        # Rule no. 1
           trigger='feed_char',
           source='start',
           dest='in_quotes',
           conditions=[self.is_quote])
       self.machine.add_transition(        # Rule no. 2
           trigger='feed_char',
           source='start',
           dest='not_in_quotes',
           after='print_char',
           conditions=[self.is_not_quote])
       self.machine.add_transition(        # Rule no. 3
           trigger='feed_char',
           source='in_quotes',
           dest='not_in_quotes',
           conditions=[self.is_quote])
       self.machine.add_transition(        # Rule no. 4
           trigger='feed_char',
           source='in_quotes',
           dest='in_quotes',
           after='print_char_upper',
           conditions=[self.is_not_quote])
       self.machine.add_transition(        # Rule no. 5
           trigger='feed_char',
           source='not_in_quotes',
           dest='not_in_quotes',
           after='print_char',
           conditions=[self.is_not_quote])
       self.machine.add_transition(        # Rule no. 6
           trigger='feed_char',
           source='not_in_quotes',
           dest='in_quotes',
           conditions=[self.is_quote])
       self.machine.add_transition(        # Rule no. 7
           trigger='feed_char',
           source='in_quotes',
           dest='error',
           after=[self.print_status, self.exit],
           conditions=[self.is_eof])
       self.machine.add_transition(        # Rule no. 8
           trigger='feed_char',
           source='not_in_quotes',
           dest='end',
           after='print_status',
           conditions=[self.is_eof])
   

   Notes:

   • The rule number comments correspond to the line numbers in our state-transition table above.

5. Add the callback methods that implement the conditions and the actions. For our sample FSM, here they are:

   def is_quote(self, char):
       """Return True if char is not end of input char and is quote char."""
       result = True if char != '$' and char == '"' else False
       dbgprint('(is_quote) char: {}  result: {}'.format(char, result))
       return result
   
   def is_not_quote(self, char):
       """Return True if char is not end of input char and not quote char."""
       result = True if char != '$' and char != '"' else False
       dbgprint('(is_quote) char: {}  result: {}'.format(char, result))
       return result
   
   def is_eof(self, char):
       """Return True if current char is the end of input character."""
       dbgprint('(is_eof) char: {}'.format(char))
       return True if char == '$' else False
   
   def print_char(self, char):
       """Print the current char unchanged."""
       print('char: {}'.format(char))
   
   def print_char_upper(self, char):
       """Convert the char to upper-case and print it."""
       print('char: {}'.format(char.upper()))
   
   def print_status(self, char):
       """Print the current status (state, char) of the FSM."""
       print('status -- state: {}  char: {}'.format(self.state, char))
   
   def exit(self, char):
       sys.exit('finished')
   

You can find the complete source code for our FSM application here: transitions_test04.py

You can run a test with:

$ python transitions_test04.py run "hello \"dave\", how are you?"

3   Exporting an FSM to JSON

It is possible, using inspection on our FSM model class, to export a JSON description of that class. Later, in this post, we'll explore the ability to read that JSON description and generate the Python source code for that model FSM class.

Basically, to create that JSON output, we do the following:

1. Create a dictionary. We'll be adding the description of our FSM to this dictionary.
2. Use various Python object inspection techniques (including the use of the inspect module in the Python standard library) to obtain information about our FSM model class.
3. Insert this information into our dictionary using keys that will enable us to access it, later.

3. Use the dump function in the json module (from the Python standard library) to write a JSON representation of the object we've created to a disk file.

The keys in our dictionary object and the information associated with each key are the following:

• "model-class-name" -- The name of the class of the model object.
• "states" -- The names of the states in the FSM.
• "initial-state" -- The name of the initial state.
• "transitions" -- A list of the transitions. Each item in this list is a dictionary with the following keys:
  • "trigger"
  • "source"
  • "dest"
  • "before"
  • "after"
  • "prepare"
• "callbacs" -- A list of callback function names and their source code. Each item in this list is a dictionary with the following keys:
  • "name" -- The callback method's name.
  • "source" -- The source code for the callback method.

Here is the code that constructs that dictionary, then writes it to a file as JSON:

def to_json(model):
    machine = model.machine
    jobj = {}
    jobj['model-class-name'] = model.__class__.__name__
    jobj['states'] = list(machine.states)
    jobj['initial-state'] = machine.initial
    jtransitions = []
    items = []
    #callbacknames = []
    for key, event in machine.events.items():
        if not key.startswith('to_'):
            items.append((key, event))
    for key, event in items:
        for tr_name, transitions in event.transitions.items():
            for transition in transitions:
                funcs = []
                for condition in transition.conditions:
                    funcs.append(condition.func.__name__)
                jtransition = {
                    'trigger': key,
                    'source': transition.source,
                    'dest': transition.dest,
                    'conditions': funcs,
                }
                if transition.before:
                    names = collect_names(transition.before)
                    jtransition['before'] = names
                    #callbacknames.extend(names)
                if transition.after:
                    names = collect_names(transition.after)
                    jtransition['after'] = names
                    #callbacknames.extend(names)
                if transition.prepare:
                    names = collect_names(transition.prepare)
                    jtransition['prepare'] = names
                    #callbacknames.extend(names)
                jtransitions.append(jtransition)
    jobj['transitions'] = jtransitions
    members = inspect.getmembers(model, inspect.ismethod)
    jcallbacks = []
    for callbackname, callback in members:
        if callbackname == '__init__':
            continue
        # pull out the method name, the parameters, and code body.
        # look at IPython implementation for help.
        source = inspect.getsource(callback)
        jcallback = {
            'name': callbackname,
            'source': source,
        }
        jcallbacks.append(jcallback)
    jobj['callbacks'] = jcallbacks
    json.dump(jobj, sys.stdout)
    print()

def collect_names(callbacks):
    names = []
    for callback in callbacks:
        if isinstance(callback, str):
            names.append(callback)
        else:
            names.append(callback.__name__)
    return names

Notes:

• We write the JSON content to stdout. The user can redirect it to a disk file, as needed.

You can also find the above functions here: transitions_test04.py

4   Generating Python FSM code from JSON

Here is the function that reads a JSON file, converts it to Python objects, and then extracts information from that structured object so that it can write out Python source code that implements that FSM using transitions. The code itself is actually quite simple:

def load(options):
    infilename = getattr(options, 'json-file')
    with open(infilename, 'r') as infile:
        jobj = json.load(infile)
    wrt("""\
#!/usr/bin/env python

from __future__ import print_function
import sys
import argparse
from transitions import Machine
""")
    wrt()
    wrt('class {}(object):'.format(jobj['model-class-name']))
    names = [str(name) for name in jobj['states']]
    wrt('    states = {}'.format(names))
    wrt()
    wrt('    def __init__(self):')
    wrt('        self.machine = Machine(')
    wrt('            model=self,')
    wrt('            states=self.states,')
    wrt("            initial='{}')".format(jobj['initial-state']))
    transitions = jobj['transitions']
    for transition in transitions:
        wrt("        self.machine.add_transitions(")
        wrt("            trigger='{}',".format(str(transition['trigger'])))
        wrt("            source='{}',".format(str(transition['source'])))
        wrt("            dest='{}',".format(str(transition['dest'])))
        item = transition.get('conditions')
        if item is not None:
            names = [str(name) for name in item]
            wrt("            conditions={},".format(names))
        item = transition.get('after')
        if item is not None:
            names = [str(name) for name in item]
            wrt("            after={},".format(names))
        wrt('        )')
    wrt()
    callbacks = jobj['callbacks']
    for callback in callbacks:
        wrt(callback['source'])

5   Injecting an FSM into a class from JSON

In this section we discuss how to read/load the JSON code and "inject" an FSM into a class.

Basically, we do the following:

1. Load the JSON FSM description from a file.
2. Create an instance of an "empty" class. This is our FSM model.
3. Access the various pieces of information about the FSM using the dictionary keys described above.
4. Add attributes to the newly created instance/model. In particular, (a) we add instance.states, and (b) we create an instance of transitions.Machine and save that in instance.machine.
5. Create the needed transitions FSM objects and insert them into the instance of our model class.

Here is the code that does it:

class TestFsm(object):
    pass

def inject(options):
    instance = TestFsm()
    fsm_json_file_name = getattr(options, 'json-file')
    inject_fsm(instance, fsm_json_file_name)

def inject_fsm(instance, fsm_json_file_name):
    with open(fsm_json_file_name, 'r') as infile:
        jobj = json.load(infile)
    names = jobj.get('states', [])
    initial_state = jobj.get('initial-state')
    instance.states = names
    instance.machine = Machine(
        model=instance,
        states=instance.states,
        initial=initial_state,
    )
    transitions = jobj['transitions']
    for transition in transitions:
        conditions = transition.get('conditions', [])
        after = transition.get('after', [])
        instance.machine.add_transition(
            trigger='{}'.format(str(transition['trigger'])),
            source='{}'.format(str(transition['source'])),
            dest='{}'.format(str(transition['dest'])),
            conditions=conditions,
            after=after,
        )
    callbacks = jobj['callbacks']
    inject_fsm_callbacks(instance, callbacks)

def inject_fsm_callbacks(instance, callbacks):
    globals1 = {}
    for callback in callbacks:
        name = callback.get('name')
        source = callback.get('source')
        source1 = '\n'.join((line[4:] for line in source.splitlines()))
        dbgprint('# --------------------------------------')
        dbgprint('name: {}  source1:\n"{}"'.format(name, source1))
        dbgprint('# --------------------------------------')
        code_obj = compile(source1, '<string>', 'exec')
        exec(code_obj, globals1)
        func_obj = globals1.get(name)
        setattr(instance, name, types.MethodType(func_obj, instance))