Paste: FSM class and sample from SoS


class FSM(object):
    """A simple Finite State Machine class that uses returned methods to
    determine the next state to run."""

    def __init__(self):
        self.reset()

    def reset(self):
        """Restarts the FSM by calling the START() state."""
        self.state = self.START()

    def START(self):
        """You implement this method as your FSM init method, it is called
        when the object is created."""
        return self.END

    def ERROR(self, *args):
        """Transition to this when you've hit an error, and FSM will do
        that when there's an exception or some similar problem.  If you
        want to handle errors yourself, then implement this state."""
        return self.END

    def END(self, *args):
        """Final state in the FSM.  No more processing will happen until
        reset is called again."""
        print "END: ", self.state.func_name
        return self.END

    def to(self, other, *args):
        """Does an internal transition to the given state, passing on the
        arguments.  This is how you change to another state 
        without waiting for a new event."""
        return other(*args)

    def event(self, *args):
        """Given the args, run the FSM until a state is returned.
        If there's an error, this method will report it and transition to the
        ERROR state.  The default ERROR state is to then END, but you can
        override that."""

        print ">>> EVENT(", self.state.func_name, " '", self.state.__doc__, "'): ARGS: ", args
        assert self.state, "FSM not initialized correctly."

        try:
            nextstate = self.state(*args)
            assert nextstate, ("State: %s returned None for next state." %
                               repr(self.state))
        except RuntimeError, exc:
            print "!!! ERROR: "
            nextstate = self.ERROR(exc)

        if nextstate != self.state:
            print "^^^ TRANS: ", nextstate.func_name
            self.state = nextstate
        else:
            print "<<< STATE: ", self.state.func_name

    def is_finished(self):
        """Tells you if the FSM is done processing (in the END state)."""
        return self.state == self.END

    def __getstate__(self):
        """The pickle module can't handle instance methods, which we use as
        states.  Therefore, we change the state over to a string temporarily."""
        state = self.__dict__.copy()
        state["state"] = self.state.func_name
        return state

    def __setstate__(self, dict):
        """After you unpickle an FSM, this changes the state back from a string
        into an instance method."""
        self.__dict__ = dict
        self.state = self.__getattribute__(self.state)


### example of a simple test list authenticator
from __future__ import with_statement
from sos import fsm
import os

class ListAuthenticate(fsm.FSM):

    def reply_with_state(self, message, state):
        sender = "zedshaw-%s@zedshaw.com" % state
        reply = self.relay.render(message, "state.msg", sender=sender, state=state)
        self.relay.reply(reply)

    def START(self):
        """Sets up for processing."""
        self.reply_count = 0
        return self.AWAITING

    def AWAITING(self, db, message, args):
        """Awaiting the first message."""
        self.reply_with_state(message, "AWAITING")
        return self.PENDING

    def PENDING(self, db, message, args):
        """A reply is pending."""
        self.reply_with_state(message, "PENDING")

        if self.reply_count == 2:
            return self.to(self.AUTHENTICATED, message, args)
        else:
            self.reply_count += 1
            return self.PENDING

    def AUTHENTICATED(self, db, message, args):
        """They're good, let them through."""
        self.reply_with_state(message, "AUTHENTICATED")
        return self.END

    def FAILED(self, db, message, args):
        """The auth failed, reject."""
        self.reply_with_state(message, "FAILED")
        return self.END