Python state-machine design

Question

Related to this Stack Overflow question (C state-machine design), could you Stack Overflow folks share your Python state-machine design techniques with me (and the community)?

At the moment, I am going for an engine based on the following:

class TrackInfoHandler(object):
    def __init__(self):
        self._state="begin"
        self._acc=""

    ## ================================== Event callbacks

    def startElement(self, name, attrs):
        self._dispatch(("startElement", name, attrs))

    def characters(self, ch):
        self._acc+=ch

    def endElement(self, name):
        self._dispatch(("endElement", self._acc))
        self._acc=""

    ## ===================================
    def _missingState(self, _event):
        raise HandlerException("missing state(%s)" % self._state)

    def _dispatch(self, event):
        methodName="st_"+self._state
        getattr(self, methodName, self._missingState)(event)

    ## =================================== State related callbacks

But I am sure there are tons of ways of going at it while leveraging Python's dynamic nature (e.g. dynamic dispatching).

I am after design techniques for the "engine" that receives the "events" and "dispatches" against those based on the "state" of the machine.

Proserpina

To paraphrase Adam's point, I think some more concrete information about what you're trying to accomplish would help. – Iq

@Jason Orendorff: fair enough. I have updated the question accordingly. – Proserpina

Answer 1

I don't really get the question. The State Design pattern is pretty clear. See the Design Patterns book.

class SuperState( object ):
    def someStatefulMethod( self ):
        raise NotImplementedError()
    def transitionRule( self, input ):
        raise NotImplementedError()

class SomeState( SuperState ):
    def someStatefulMethod( self ):
        actually do something()
    def transitionRule( self, input ):
        return NextState()

That's pretty common boilerplate, used in Java, C++, Python (and I'm sure other languages, also).

If your state transition rules happen to be trivial, there are some optimizations to push the transition rule itself into the superclass.

Note that we need to have forward references, so we refer to classes by name, and use eval to translate a class name to an actual class. The alternative is to make the transition rules instance variables instead of class variables and then create the instances after all the classes are defined.

class State( object ):
    def transitionRule( self, input ):
        return eval(self.map[input])()

class S1( State ): 
    map = { "input": "S2", "other": "S3" }
    pass # Overrides to state-specific methods

class S2( State ):
    map = { "foo": "S1", "bar": "S2" }

class S3( State ):
    map = { "quux": "S1" }

In some cases, your event isn't as simple as testing objects for equality, so a more general transition rule is to use a proper list of function-object pairs.

class State( object ):
    def transitionRule( self, input ):
        next_states = [ s for f,s in self.map if f(input)  ]
        assert len(next_states) >= 1, "faulty transition rule"
        return eval(next_states[0])()

class S1( State ):
    map = [ (lambda x: x == "input", "S2"), (lambda x: x == "other", "S3" ) ]

class S2( State ):
    map = [ (lambda x: "bar" <= x <= "foo", "S3"), (lambda x: True, "S1") ]

Since the rules are evaluated sequentially, this allows a "default" rule.

Cocotte

@jldupont: Right out of the book. Didn't invent it, just copied it from the GoF Design Patterns book. That's why I still don't get your question -- they cover this completely. – Cocotte

@scott: I do not have this book. By asking the question, I am hoping to tap the SO community collective wisdom. You can of course choose to ignore the question or contribute: that's your choice. – Proserpina

@scott: ... and I thank you for your contribution. I do not believe I have "complained" but of course the definition of "complaining" falls in the "qualitative territory". Have a good day! – Proserpina

@scott: come to think of it, my previous "complaint" might have been triggered by your insistence in stating that you don't get my question. No hard feelings on my side. – Proserpina

Although you copied it out the book, it seems "nicer" to use something like globals()[self.map[input]]() (or a dict like {'S1': S1}) instead of eval? To further nitpick, it's redefining the built-in next` – Bonar

@jldupont. I don't get your question. It's vague and confusing. It seems like you're trying to use a SAX parser for something, but I can't tell what. A context stack is generally all that's needed to construct a DOM object that represents the original XML. – Cocotte

@scott: the question isn't about XML parsing per-se... and yes it is vague on purpose: that's why I asked suggestion on "design techniques" (in general) and not specifically on XML SAX based parsing. Sorry for the confusion. – Proserpina

Love the code but hate the naming, i.e. wouldn't AbstractState not be a better name than SuperState? And you didn't copy the state transition part. The book actually leaves the transition open and mentions different implementation variations (Context method with big if-else, your implementation, or state transtion dict) – Laudable

Here is a completely different way of doing this. github.com/kashifrazzaqui/themstates – Bodily

Answer 2

In the April, 2009 issue of Python Magazine, I wrote an article on embedding a State DSL within Python, using pyparsing and imputil. This code would allow you to write the module trafficLight.pystate:

# trafficLight.pystate

# define state machine
statemachine TrafficLight:
    Red -> Green
    Green -> Yellow
    Yellow -> Red

# define some class level constants
Red.carsCanGo = False
Yellow.carsCanGo = True
Green.carsCanGo = True

Red.delay = wait(20)
Yellow.delay = wait(3)
Green.delay = wait(15)

and the DSL compiler would create all the necessary TrafficLight, Red, Yellow, and Green classes, and the proper state transition methods. Code could call these classes using something like this:

import statemachine
import trafficLight

tl = trafficLight.Red()
for i in range(6):
    print tl, "GO" if tl.carsCanGo else "STOP"
    tl.delay()
    tl = tl.next_state()

(Unfortunately, imputil has been dropped in Python 3.)

Overturf

Matt Anderson submitted this version of my statemachine code to the pyparsing wiki (pyparsing.wikispaces.com/message/view/Publications/18439845) which is compatible with Py2.7 and up. – Overturf

wikispaces is no longer online. This code is now at the pyparsing GitHub repo: github.com/pyparsing/pyparsing/tree/master/examples/… – Overturf

Answer 3

There is this design pattern for using decorators to implement state machines. From the description on the page:

Decorators are used to specify which methods are the event handlers for the class.

There is example code on the page as well (it is quite long so I won't paste it here).

Lycaonia

Great mention. Always wondered how I would implement exactly that! – Laudable

Answer 4

I also was not happy with the current options for state_machines so I wrote the state_machine library.

You can install it by pip install state_machine and use it like so:

@acts_as_state_machine
class Person():
    name = 'Billy'

    sleeping = State(initial=True)
    running = State()
    cleaning = State()

    run = Event(from_states=sleeping, to_state=running)
    cleanup = Event(from_states=running, to_state=cleaning)
    sleep = Event(from_states=(running, cleaning), to_state=sleeping)

    @before('sleep')
    def do_one_thing(self):
        print "{} is sleepy".format(self.name)

    @before('sleep')
    def do_another_thing(self):
        print "{} is REALLY sleepy".format(self.name)

    @after('sleep')
    def snore(self):
        print "Zzzzzzzzzzzz"

    @after('sleep')
    def big_snore(self):
        print "Zzzzzzzzzzzzzzzzzzzzzz"

person = Person()
print person.current_state == person.sleeping       # True
print person.is_sleeping                            # True
print person.is_running                             # False
person.run()
print person.is_running                             # True
person.sleep()

# Billy is sleepy
# Billy is REALLY sleepy
# Zzzzzzzzzzzz
# Zzzzzzzzzzzzzzzzzzzzzz

print person.is_sleeping                            # True

Griner

Ha, I wasn't happy either - so I wrote this instead, github.com/kashifrazzaqui/themstates – Bodily

Answer 5

I think S. Lott's answer is a much better way to implement a state machine, but if you still want to continue with your approach, using (state,event) as the key for your dict is better. Modifying your code:

class HandlerFsm(object):

  _fsm = {
    ("state_a","event"): "next_state",
    #...
  }

Soothe

+1: thanks, I was contemplating this possibility too. I haven't settled on an implementation just yet. – Proserpina

Answer 6

It probably depends on how complex your state machine is. For simple state machines, a dict of dicts (of event-keys to state-keys for DFAs, or event-keys to lists/sets/tuples of state-keys for NFAs) will probably be the simplest thing to write and understand.

For more complex state machines, I've heard good things about SMC, which can compile declarative state machine descriptions to code in a wide variety of languages, including Python.

Morgenthaler

Answer 7

The following code is a really simple solution. The only interesting part is:

   def next_state(self,cls):
      self.__class__ = cls

All the logic for each state is contained in a separate class. The 'state' is changed by replacing the '__class__' of the running instance.

#!/usr/bin/env python

class State(object):
   call = 0 # shared state variable
   def next_state(self,cls):
      print '-> %s' % (cls.__name__,),
      self.__class__ = cls

   def show_state(self,i):
      print '%2d:%2d:%s' % (self.call,i,self.__class__.__name__),

class State1(State):
   __call = 0  # state variable
   def __call__(self,ok):
      self.show_state(self.__call)
      self.call += 1
      self.__call += 1
      # transition
      if ok: self.next_state(State2)
      print '' # force new line

class State2(State):
   __call = 0
   def __call__(self,ok):
      self.show_state(self.__call)
      self.call += 1
      self.__call += 1
      # transition
      if ok: self.next_state(State3)
      else: self.next_state(State1)
      print '' # force new line

class State3(State):
   __call = 0
   def __call__(self,ok):
      self.show_state(self.__call)
      self.call += 1
      self.__call += 1
      # transition
      if not ok: self.next_state(State2)
      print '' # force new line

if __name__ == '__main__':
   sm = State1()
   for v in [1,1,1,0,0,0,1,1,0,1,1,0,0,1,0,0,1,0,0]:
      sm(v)
   print '---------'
   print vars(sm

Result:

 0: 0:State1 -> State2 
 1: 0:State2 -> State3 
 2: 0:State3 
 3: 1:State3 -> State2 
 4: 1:State2 -> State1 
 5: 1:State1 
 6: 2:State1 -> State2 
 7: 2:State2 -> State3 
 8: 2:State3 -> State2 
 9: 3:State2 -> State3 
10: 3:State3 
11: 4:State3 -> State2 
12: 4:State2 -> State1 
13: 3:State1 -> State2 
14: 5:State2 -> State1 
15: 4:State1 
16: 5:State1 -> State2 
17: 6:State2 -> State1 
18: 6:State1 
---------
{'_State1__call': 7, 'call': 19, '_State3__call': 5, '_State2__call': 7}

Plexor

I've written a few python state machines using this design, and I have to say I think it's the cleanest one out of all the options here. It doesn't require a external library, the internal mechanics are quite simple and concise, and it works very well. – Cynde

Answer 8

I think that the tool PySCXML needs a closer look too.

This project uses the W3C definition: State Chart XML (SCXML): State Machine Notation for Control Abstraction

SCXML provides a generic state-machine based execution environment based on CCXML and Harel State Tables

Currently, SCXML is a working draft; but chances are quite high that it is getting a W3C recommendation soon (it is the 9th draft).

Another interesting point to highlight is that there is an Apache Commons project aimed at creating and maintaining a Java SCXML engine capable of executing a state machine defined using a SCXML document, while abstracting out the environment interfaces...

And for certain other tools, supporting this technology will emerge in the future when SCXML is leaving its draft-status...

Ginelle

Answer 9

I wouldn't think to reach for a finite state machine for handling XML. The usual way to do this, I think, is to use a stack:

class TrackInfoHandler(object):
    def __init__(self):
        self._stack=[]

    ## ================================== Event callbacks

    def startElement(self, name, attrs):
        cls = self.elementClasses[name]
        self._stack.append(cls(**attrs))

    def characters(self, ch):
        self._stack[-1].addCharacters(ch)

    def endElement(self, name):
        e = self._stack.pop()
        e.close()
        if self._stack:
            self._stack[-1].addElement(e)

For each kind of element, you just need a class that supports the addCharacters, addElement, and close methods.

EDIT: To clarify, yes I do mean to argue that finite state machines are usually the wrong answer, that as a general-purpose programming technique they're rubbish and you should stay away.

There are a few really well-understood, cleanly-delineated problems for which FSMs are a nice solution. lex, for example, is good stuff.

That said, FSMs typically don't cope well with change. Suppose someday you want to add a bit of state, perhaps a "have we seen element X yet?" flag. In the code above, you add a boolean attribute to the appropriate element class and you're done. In a finite state machine, you double the number of states and transitions.

Problems that require finite state at first very often evolve to require even more state, like maybe a number, at which point either your FSM scheme is toast, or worse, you evolve it into some kind of generalized state machine, and at that point you're really in trouble. The further you go, the more your rules start to act like code—but code in a slow interpreted language you invented that nobody else knows, for which there's no debugger and no tools.

Iq

@Jason: as far as I am concerned, any (but the trivial) piece of software acts as a "state-machine" in some form or another. The statement that an FSM is of limited use falls in bottomless pit on my territory. Of course, if you are referring to a particular FSM pattern, that might be the case but I tend to be flexible when it comes to applying patterns. – Proserpina

Just about any algorithm can be implemented as a state machine. Very few of them should be. – Pain

@jldupont: The ranty part of my answer refers to a specific technique where you actually design part of your program as a finite state machine in the ordinary sense of the term, states and transitions are explicit in the code, and some logic is encoded in a state transition table. Like your example and all the other answers. :) – Iq

Answer 10

I would definitely not recommend implementing such a well known pattern yourself. Just go for an open source implementation like transitions and wrap another class around it if you need custom features. In this post I explain why I prefer this particular implementation and its features.

Nardi

the link to the post is broken – Centenarian

Answer 11

Other related projects:

• http://fsme.sourceforge.net/

• https://code.google.com/p/visio2python/

You can paint state-machine and then use it in your code.

Margiemargin

Answer 12

Here is a solution for "statefull objects" I've come up with, but it is rather inefficient for your intended purpose because state changes are relatively expensive. However, it may work well for objects which change state infrequently or undergo only a bounded number of state changes. The advantage is that once the state is changed, there is no redundant indirection.

class T:
    """
    Descendant of `object` that rectifies `__new__` overriding.

    This class is intended to be listed as the last base class (just
    before the implicit `object`).  It is a part of a workaround for

      * https://bugs.python.org/issue36827
    """

    @staticmethod
    def __new__(cls, *_args, **_kwargs):
        return object.__new__(cls)

class Stateful:
    """
    Abstract base class (or mixin) for "stateful" classes.

    Subclasses must implement `InitState` mixin.
    """

    @staticmethod
    def __new__(cls, *args, **kwargs):
        # XXX: see https://stackoverflow.com/a/9639512
        class CurrentStateProxy(cls.InitState):
            @staticmethod
            def _set_state(state_cls=cls.InitState):
                __class__.__bases__ = (state_cls,)

        class Eigenclass(CurrentStateProxy, cls):
            __new__ = None  # just in case

        return super(__class__, cls).__new__(Eigenclass, *args, **kwargs)

# XXX: see https://bugs.python.org/issue36827 for the reason for `T`.
class StatefulThing(Stateful, T):
    class StateA:
        """First state mixin."""

        def say_hello(self):
            self._say("Hello!")
            self.hello_count += 1
            self._set_state(self.StateB)
            return True

        def say_goodbye(self):
            self._say("Another goodbye?")
            return False

    class StateB:
        """Second state mixin."""

        def say_hello(self):
            self._say("Another hello?")
            return False

        def say_goodbye(self):
            self._say("Goodbye!")
            self.goodbye_count += 1
            self._set_state(self.StateA)
            return True

    # This one is required by `Stateful`.
    class InitState(StateA):
        """Third state mixin -- the initial state."""

        def say_goodbye(self):
            self._say("Why?")
            return False

    def __init__(self, name):
        self.name = name
        self.hello_count = self.goodbye_count = 0

    def _say(self, message):
        print("{}: {}".format(self.name, message))

    def say_hello_followed_by_goodbye(self):
        self.say_hello() and self.say_goodbye()

# ----------
# ## Demo ##
# ----------
if __name__ == "__main__":
    t1 = StatefulThing("t1")
    t2 = StatefulThing("t2")
    print("> t1, say hello.")
    t1.say_hello()
    print("> t2, say goodbye.")
    t2.say_goodbye()
    print("> t2, say hello.")
    t2.say_hello()
    print("> t1, say hello.")
    t1.say_hello()
    print("> t1, say hello followed by goodbye.")
    t1.say_hello_followed_by_goodbye()
    print("> t2, say goodbye.")
    t2.say_goodbye()
    print("> t2, say hello followed by goodbye.")
    t2.say_hello_followed_by_goodbye()
    print("> t1, say goodbye.")
    t1.say_goodbye()
    print("> t2, say hello.")
    t2.say_hello()
    print("---")
    print( "t1 said {} hellos and {} goodbyes."
           .format(t1.hello_count, t1.goodbye_count) )
    print( "t2 said {} hellos and {} goodbyes."
           .format(t2.hello_count, t2.goodbye_count) )

    # Expected output:
    #
    #     > t1, say hello.
    #     t1: Hello!
    #     > t2, say goodbye.
    #     t2: Why?
    #     > t2, say hello.
    #     t2: Hello!
    #     > t1, say hello.
    #     t1: Another hello?
    #     > t1, say hello followed by goodbye.
    #     t1: Another hello?
    #     > t2, say goodbye.
    #     t2: Goodbye!
    #     > t2, say hello followed by goodbye.
    #     t2: Hello!
    #     t2: Goodbye!
    #     > t1, say goodbye.
    #     t1: Goodbye!
    #     > t2, say hello.
    #     t2: Hello!
    #     ---
    #     t1 said 1 hellos and 1 goodbyes.
    #     t2 said 3 hellos and 2 goodbyes.

I've posted a "request for remarks" here.

Furrow