== State Machine Decorator Module == <> === Overview === This module provides a set of decorators that are useful for implementing state machines of the type described by UML 2.0 state charts. The overhead of these decorators may be too high for them to be useful in parsing applications. The code for the state machine decorator module is given below. Examples are given following the code. === License === {{{ Copyright (C) 2010, 2011 Rodney Drenth All rights reserved. Redistribution and use in source and binary forms, with or without modification, are permitted provided that the following conditions are met: 1. Redistributions of source code must retain the above copyright notice, this list of conditions and the following disclaimer. 2. Redistributions in binary form must reproduce the above copyright notice, this list of conditions and the following disclaimer in the documentation and/or other materials provided with the distribution. 3. Neither the name of the project nor the names of the author may be used to endorse or promote products derived from this software without specific prior written permission. THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDER AND CONTRIBUTERS``AS IS'' AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE AUTHOR OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. }}} === Python Code === {{{ #!/bin/env/python import types import itertools import logging logging.basicConfig( filename="gf_info.txt", format = "%(levelname)-10s %(message)s", level = logging.ERROR ) from functools import wraps def truncated(alist, cmprsn): for x in alist: if x.__name__ == cmprsn: break yield x class ContextBase(object): pass class _StateVariable(object): """ Attribute of a class to maintain state . State Variable objects are instantiated indirectly via calls to the TransitionTable class's initialize method. TransitionTable objects are created at the class level. """ def __init__(self, transTable, context): """Constructor - set to initial state""" self.__current_state = transTable.initialstate self.__next_state = transTable.initialstate self.sTable = transTable self.__statestack=[] self.__ctxClass = context.__class__ def toNextState(self, context): """Transition to next state, if a next_state is differnt. In addition to the actual state transition, it invokes onLeave and onEnter methods as required. """ if self.__next_state is not self.__current_state: cc = context.__class__ tt_name = self.sTable.inst_state_name logging.debug("Transitioning to state %s"%self.__next_state.__name__) def callInState(methName, crnt_state): if (hasattr(crnt_state, methName) or hasattr(context, methName)): nmro = [crnt_state,] nmro.extend(cc.__mro__) psudoClassName = "%s_%s"%(cc.__name__, crnt_state.__name__) stCls = type( psudoClassName, tuple(nmro), {}) context.__class__ = stCls getattr(context, methName)() # call the onEnter or onLeave method here context.__class__ = cc callInState('onLeave', self.__current_state) self.__setState( context ) callInState('onEnter', self.__current_state) def __setState(self, context ): """low level funky called from toNextState""" cc = context.__class__ mro = cc.__mro__ if ( self.__current_state not in mro): self.__current_state = self.__next_state return logging.debug("Current state %s in mro"% self.__current_state.__name__) def f(anc): return self.__next_state if anc == self.__current_state else anc newmro = tuple(f(anc) for anc in cc.__mro__) tt_name = self.sTable.inst_state_name cls_name ="%s_%s"%(self.__ctxClass.__name__, self.__next_state.__name__) context.__class__ = type(cls_name, newmro, {}) def pushState(self, newState, context = None): """PushState - allows going to another state with intent of returning to the current one.""" self.__statestack.append(self._current_state) self.__next_state = newState if context: self.toNextState(context) def popState(self, context = None): """Pop back to the previously pushed state (pushState)""" self.__next_state = self.__statestack.pop() if (context): self.toNextState( context) def name(self): """Return name of current state""" return self.__current_state.__name__ def setXition(self, func): """ Sets the state to transition to upon seeing a transtion event This method should only be called by the decorators impl'd in this module. """ nxState = self.__current_state.nextStates[func.__name__] if nxState is not None: self.__next_state = nxState; def getFunc(self, func, contxt): """Gets the state dependant action method, wrapped in a try-catch block. This method should only be called by the decorators impl'd in this module. """ crnt = self.__current_state svar_name = self.sTable.inst_state_name svCtxt = self.__ctxClass cc = contxt.__class__ pseudoclas = "%s_%s"%(cc.__name__, crnt.__name__) nmro = [crnt] lhead = itertools.takewhile( lambda x: x != svCtxt, crnt.__mro__) if svCtxt in cc.__mro__: ltail = itertools.dropwhile( lambda x: x!= svCtxt, cc.__mro__) else: ltail = cc.__mro__ nmro.extend(ltail) logging.debug("%s - %s - %s - [%s]\n"%(func.__name__, cc.__name__, svar_name, ", ".join( cls.__name__ for cls in truncated(nmro,'TopLevelWindow' )))) stCls = type( pseudoclas, tuple(nmro), {}) contxt.__class__ = stCls try: funky = getattr(contxt, func.__name__) except: funky = None contxt.__class__ = cc # revert... if funky is None: t = "'%s' has no attribute '%s' in state %s" % (self.name(), func.__name__, crnt.__name__) raise NotImplementedError(t) # function with wrapping attribute means we've recursed all the way back # to the context class and need to call the func as a default. if hasattr(funky, "wrapping") and (funky.wrapping == self.sTable.inst_state_name): def funcA(*args, **kwargs): return func(contxt, *args, **kwargs) funky = funcA def wrappd2( self, *args, **kwargs): # wrap in try - except in event that funky() does something funky try: self.__class__ = stCls retn = funky( *args, **kwargs) finally: self.__class__ = cc return retn return wrappd2 # ----------------------------------------------------------------------------- class TransitionTable(object): """Defines a state table for a state machine class A state table for a class is associated with the state variable in the instances of the class. The name of the state variable is given in the constructor to the StateTable object. StateTable objects are attributes of state machine classes, not intances of the state machine class. A state machine class can have more than one StateTable. """ def __init__(self, stateVarblName): """Transition Table constructor - state varblName is name of associated instance state variable. """ self.inst_state_name = stateVarblName self.eventList = [] self.initalState = None nextStates = {} def initialize(self, ctxt): """Create a new state variable in the context. State variable refs this transition table.""" ctxt.__dict__[self.inst_state_name] = _StateVariable(self, ctxt) def _addEventHandler(self, funcName): """Notifies the current object of a metho that handles a transition. This is called by two of the decorators implemented below """ self.eventList.append(funcName) def nextStates(self, subState, nslList): """Sets up transitions from the state specified by substate subState is one of the derived state classes, subclassed from the context state machine class. nslList is a list of states to which the context will transition upon the invocation of one of the transition methods. 'None' may be specified instead of an actual state if the context is to remain in the same state upon invocation of the corresponding method. """ if len(nslList) != len(self.eventList): j = "Expected %s Got %s."%(len(self.eventList), len(nslList)) raise RuntimeError("Wrong number of states in transition list.\n%s"%j) subState.nextStates = dict(zip(self.eventList, nslList)) # ----------------------------------------------------------------------------- def event( state_table): """Decorator for indicating an Event or 'Action' method. The decorator is applied to the methods of the state machine class to indicate that the method will invoke a state dependant behavior. States are implemented as subclasses of the context(state machine) class . """ stVarName = state_table.inst_state_name def wrapper(func): @wraps(func) def objCall(self, *args, **kwargs): state_var = getattr(self, stVarName) rtn = state_var.getFunc(func, self)(self, *args, **kwargs) return rtn objCall.wrapping = stVarName return objCall return wrapper def transition( state_table ): """Decorator used to set up methods which cause transitions between states. The decorator is applied to methods of the context (state machine) class. Invoking the method may cause a transition to another state. To define what the transitions are, the nextStates method of the TransitionTable class is used. """ stVarName = state_table.inst_state_name def wrapper(func): state_table._addEventHandler( func.__name__) @wraps(func) def objCall(self, *args, **kwargs): state_var = getattr(self, stVarName) state_var.setXition(func) rtn = func(self, *args, **kwargs) state_var.toNextState(self) return rtn objCall.wrapping =stVarName return objCall return wrapper def transitionevent( state_table): """A decorator which is essentially the combination of the above two. Can both invoke state dependent method and trigger a state transition. Mostly equivalent to : @Transition(xitionTable) @Event(xitionTable) """ stVarName = state_table.inst_state_name def wrapper(func): state_table._addEventHandler( func.__name__) @wraps(func) def objCall(self, *args, **kwargs): state_var = getattr(self, stVarName) state_var.setXition(func) rtn = state_var.getFunc(func, self)(self, *args, **kwargs) state_var.toNextState(self) return rtn objCall.wrapping = stVarName return objCall return wrapper }}} == Examples of Use == === Simple Example === The example has three states, which rotate to the next state whenever the writeName method is called. In StateA, the text is printed out in lower case. In states StateB and StateC the text is printed out in upper case. {{{ #!/bin/env/python import DecoratorStateMachine as dsm class StateContext( dsm.ContextBase): ttable = dsm.TransitionTable('myState') def __init__(self): self.ttable.initialize(self) @dsm.transitionevent(ttable) def writeName(self, name): pass class StateA(StateContext): def writeName(self, name): print name.lower() class StateB(StateContext): def writeName(self, name): print name.upper() class StateC(StateB): pass # Set up transition table to cause states totoggle StateContext.ttable.nextStates(StateA, (StateB,)) StateContext.ttable.nextStates(StateB, (StateC,)) StateContext.ttable.nextStates(StateC, (StateA,)) StateContext.ttable.initialstate = StateA if __name__=='__main__': days=("Monday","Tuesday","Wednesday","Thursday", "Friday","Saturday","Sunday") ctxt = StateContext() for day in days: ctxt.writeName(day) x = raw_input("done>") }}} ==== Output ==== {{{ monday TUESDAY WEDNESDAY thursday FRIDAY SATURDAY sunday }}} === Miss Grant's Controller === The specification for this controller comes from Martin Fowler. This example uses wxPython as well as the state machine module. {{{ #!/bin/env/python import wx import DecoratorStateMachine as dsm class MyFrame(wx.Frame, dsm.ContextBase): xtable = dsm.TransitionTable('pstate') dtable = dsm.TransitionTable('dstate') def __init__(self): self.xtable.initialize(self) self.dtable.initialize(self) wx.Frame.__init__(self, None, -1, "My Frame", size=(410,250)) family = wx.SWISS style = wx.NORMAL weight = wx.BOLD font = wx.Font(12,family,style,weight, False, "Verdana") self.SetFont(font) panel = wx.Panel(self, -1) self.btnDoor = self.makeButton(panel, "Door", (50,20), self.onToggleDoor) self.btnLight = self.makeButton(panel, "Light", (180,20), self.onLightOn ) self.btnDrawer = self.makeButton(panel, "Drawer", (50,60), self.onOpenDrawer) self.btnPanel = self.makeButton(panel, "Panel", (180,60), self.onClosePanel) self.btnPanel.Disable() self.textArea = wx.StaticText(panel, -1, "Locked", pos=(50,100), size=(100,35)) # onEnter called here would invoke MyFrame.onEnter (below) # call the current state's onEnter method indirectly through onInit() self.onInit() def onEnter(self): print "Shouldn't get here. Should call some state's onEnter functions instead." @dsm.transitionevent(dtable) def onToggleDoor(self, event): pass @dsm.event(dtable) def onInit(self): self.onEnter() # calls onEnter for current dtable/dstate state @dsm.transition(xtable) def onOpenDoor(self): pass @dsm.transition(xtable) def onCloseDoor(self): pass @dsm.transition(xtable) def onLightOn(self, event): pass @dsm.transition(xtable) def onOpenDrawer(self, event): pass @dsm.transition(xtable) def onClosePanel(self, event): pass def makeButton( self, panel, label, positn, handler ): button = wx.Button(panel, -1, label, pos=positn, size=(120,35)) self.Bind(wx.EVT_BUTTON, handler, button) return button class DoorOpen(MyFrame): doorLabel = "Close Door" def onEnter(self): print self.dstate.name() self.btnDoor.SetLabel( self.doorLabel ) def onToggleDoor(self, event): self.onCloseDoor() class DoorClosed(DoorOpen): doorLabel = "Open Door" def onToggleDoor(self, event): self.onOpenDoor() MyFrame.dtable.nextStates(DoorOpen, (DoorClosed,)) MyFrame.dtable.nextStates(DoorClosed, (DoorOpen,)) MyFrame.dtable.initialstate = DoorOpen class Idle(MyFrame): """this is an initial state""" def onEnter(self): print self.pstate.name() class Unlocked(Idle): def onEnter(self): print self.pstate.name() self.btnPanel.Enable() self.btnDoor.Disable() self.textArea.SetLabel("Unlocked") def onLeave(self): self.textArea.SetLabel("Locked") self.btnPanel.Disable() self.btnDoor.Enable() class Active(Idle): pass class LightOn(Idle): pass class DrawerOpen(Idle): pass MyFrame.xtable.nextStates(Idle, (Idle, Active, Idle, Idle, Idle)) MyFrame.xtable.nextStates(Active, (Idle, Active, LightOn, DrawerOpen, None)) MyFrame.xtable.nextStates(LightOn, (Idle, None, None, Unlocked, None )) MyFrame.xtable.nextStates(DrawerOpen, (Idle, None, Unlocked, None, None)) MyFrame.xtable.nextStates(Unlocked, (Idle, None, None, None, Idle)) MyFrame.xtable.initialstate = Idle if __name__=='__main__': app = wx.PySimpleApp() frame = MyFrame() frame.Show(True) app.MainLoop() }}} ==== Explanation ==== There are actually two states in the context. One is for the state of the door, opened or closed. The other is for the main controller. The !DoorOpen and !DoorClosed states simply translate the onToggleDoor event to invoke either onCloseDoor or onOpenDoor. The @transition decorator indicates the method can cause a state transition. The method body will be invoked if one is provided. The parameter on the decorator is the state table that governs the transition. When leaving a state, the state's onLeave method is called, if one is defined. When entering a state the state's onEnter method is called. The @event decorator indicates the method is state dependent. The parameter on the decorator is used to determine which state variable (via the transition table) in the context (there may be multiple) contains the state whose method is to be invoked. The @transitionevent decorator is a combination of the above two. A state dependent method is invoked, and it may cause a transition to a new state. The transition happens after the event method is invoked. Since states are subclasses of the context, or subclasses of other states, rules governing method or attribute resolution apply. For instance !DoorClosed is a subclass of !DoorOpened, so when 'onEnter' of the !DoorClosed state is called, it uses the one for !DoorOpened. Since !DoorClosed has defined a different value for doorLabel, the correct label is set on the door button. === Alternative Miss Grant's Controller Example === {{{ #!/bin/env/python import wx import DecoratorStateMachine as dsm class MyFrame(wx.Frame, dsm.ContextBase): dtable = dsm.TransitionTable('dstate') def __init__(self): self.dtable.initialize(self) wx.Frame.__init__(self, None, -1, "My Frame", size=(410,250)) font = wx.Font(11, wx.SWISS, wx.NORMAL, wx.BOLD, False, "Verdana") self.SetFont(font) panel = wx.Panel(self, -1) self.btnDoor = self.makeButton(panel, "Door", (50,20), self.onToggleDoor) self.btnLight = self.makeButton(panel, "Light", (180,20), self.onLightOn ) self.btnDrawer = self.makeButton(panel, "Drawer", (50,60), self.onOpenDrawer) self.btnPanel = self.makeButton(panel, "Panel", (180,60), self.onClosePanel) self.btnPanel.Disable() self.textArea = wx.StaticText(panel, -1, "Locked", pos=(50,100), size=(100,35)) # onEnter called here would invoke MyFrame.onEnter (below) # call the current state's onEnter method indirectly through onInit() self.onInit() def onEnter(self): print "Shouldn't get here. Should call some state's onEnter function instead." @dsm.transition(dtable) def onToggleDoor(self, event): pass @dsm.event(dtable) def onInit(self): self.onEnter() # calls onEnter for current dtable/dstate state @dsm.event(dtable) def onLightOn(self, event): pass @dsm.event(dtable) def onOpenDrawer(self, event): pass @dsm.event(dtable) def onClosePanel(self, event): pass def makeButton( self, panel, label, positn, handler ): button = wx.Button(panel, -1, label, pos=positn, size=(120,35)) self.Bind(wx.EVT_BUTTON, handler, button) return button class DoorOpen(MyFrame): doorLabel = "Close Door" def onEnter(self): print self.dstate.name() self.btnDoor.SetLabel( self.doorLabel ) class DoorClosed(DoorOpen): doorLabel = "Open Door" xtable = dsm.TransitionTable('pstate') def onEnter(self): # Check self's class and return if it's not DoorClosed. # otherwise if one of the xtable substates hasn't defined 'onEnter', we # could go into infinite recursion. if self.__class__.__name__ != "MyFrame_DoorClosed": return DoorOpen.onEnter(self) self.xtable.initialize(self) self.doEnter() @dsm.event(xtable) def doEnter(self): self.onEnter() @dsm.transition(xtable) def onLightOn(self, event): pass @dsm.transition(xtable) def onOpenDrawer(self, event): pass @dsm.transition(xtable) def onClosePanel(self, event): pass MyFrame.dtable.nextStates(DoorOpen, (DoorClosed,)) MyFrame.dtable.nextStates(DoorClosed, (DoorOpen,)) MyFrame.dtable.initialstate = DoorOpen class Active(DoorClosed): def onEnter(self): print self.pstate.name() class LightOn(Active): pass class DrawerOpen(Active): pass class Idle(Active): pass class Unlocked(Active): def onEnter(self): print self.pstate.name() self.btnPanel.Enable() self.btnDoor.Disable() self.textArea.SetLabel("Unlocked") def onLeave(self): self.textArea.SetLabel("Locked") self.btnPanel.Disable() self.btnDoor.Enable() DoorClosed.xtable.nextStates(Active, (LightOn, DrawerOpen, None)) DoorClosed.xtable.nextStates(LightOn, (None, Unlocked, None )) DoorClosed.xtable.nextStates(DrawerOpen, (Unlocked, None, None)) DoorClosed.xtable.nextStates(Unlocked, (None, None, Idle)) DoorClosed.xtable.nextStates(Idle, (None, None, None)) DoorClosed.xtable.initialstate = Active if __name__=='__main__': app = wx.PySimpleApp() frame = MyFrame() frame.Show(True) app.MainLoop() }}} ==== Explanation ==== There are also two state tables in this example. The difference being that the !DoorClosed state acts as the context for the second set of states. The onEnter method of the !DoorClosed state re-initializes the second state to Active. In the !DoorClosed state, the onLightOn, onOpenDrawer, and onClosePanel can cause transitions on the second state varible. These methods are events on the first state variable(dstate), and when in the !DoorOpen state, the events do not get invoked for the xtable related state.