Source code for direct.fsm.ClassicFSM

"""Finite State Machine module: contains the ClassicFSM class.

Note:
    This module and class exist only for backward compatibility with
    existing code.  New code should use the :mod:`.FSM` module instead.
"""

__all__ = ['ClassicFSM']

from direct.directnotify.DirectNotifyGlobal import directNotify
from direct.showbase.DirectObject import DirectObject
import weakref

if __debug__:
    _debugFsms = {}

[docs] def printDebugFsmList(): global _debugFsms for k in sorted(_debugFsms.keys()): print("%s %s" % (k, _debugFsms[k]()))
__builtins__['debugFsmList'] = printDebugFsmList
[docs]class ClassicFSM(DirectObject): """ Finite State Machine class. This module and class exist only for backward compatibility with existing code. New code should use the FSM class instead. """ # create ClassicFSM DirectNotify category notify = directNotify.newCategory("ClassicFSM") # special methods # these are flags that tell the ClassicFSM what to do when an # undefined transition is requested: ALLOW = 0 # print a warning, and do the transition DISALLOW = 1 # silently ignore the request (don't do the transition) DISALLOW_VERBOSE = 2 # print a warning, and don't do the transition ERROR = 3 # print an error message and raise an exception
[docs] def __init__(self, name, states=[], initialStateName=None, finalStateName=None, onUndefTransition=DISALLOW_VERBOSE): """__init__(self, string, State[], string, string, int) ClassicFSM constructor: takes name, list of states, initial state and final state as:: fsm = ClassicFSM.ClassicFSM('stopLight', [State.State('red', enterRed, exitRed, ['green']), State.State('yellow', enterYellow, exitYellow, ['red']), State.State('green', enterGreen, exitGreen, ['yellow'])], 'red', 'red') each state's last argument, a list of allowed state transitions, is optional; if left out (or explicitly specified to be State.State.Any) then any transition from the state is 'defined' and allowed 'onUndefTransition' flag determines behavior when undefined transition is requested; see flag definitions above """ self.setName(name) self.setStates(states) self.setInitialState(initialStateName) self.setFinalState(finalStateName) self.onUndefTransition = onUndefTransition # Flag to see if we are inspecting self.inspecting = 0 # We do not enter the initial state to separate # construction from activation self.__currentState = None # We set this while we are modifying the state. No one else # should recursively attempt to modify the state while we are # doing this. self.__internalStateInFlux = 0 if __debug__: global _debugFsms _debugFsms[name]=weakref.ref(self)
# I know this isn't how __repr__ is supposed to be used, but it # is nice and convenient. def __repr__(self): return self.__str__() def __str__(self): """ Print out something useful about the fsm """ currentState = self.getCurrentState() if currentState: str = ("ClassicFSM " + self.getName() + ' in state "' + currentState.getName() + '"') else: str = ("ClassicFSM " + self.getName() + ' not in any state') return str
[docs] def enterInitialState(self, argList=[]): assert not self.__internalStateInFlux if self.__currentState == self.__initialState: return assert self.__currentState == None self.__internalStateInFlux = 1 self.__enter(self.__initialState, argList) assert not self.__internalStateInFlux
# setters and getters
[docs] def getName(self): return(self.__name)
[docs] def setName(self, name): self.__name = name
[docs] def getStates(self): return list(self.__states.values())
[docs] def setStates(self, states): """setStates(self, State[])""" # Make a dictionary from stateName -> state self.__states = {} for state in states: self.__states[state.getName()] = state
[docs] def addState(self, state): self.__states[state.getName()] = state
[docs] def getInitialState(self): return(self.__initialState)
[docs] def setInitialState(self, initialStateName): self.__initialState = self.getStateNamed(initialStateName)
[docs] def getFinalState(self): return(self.__finalState)
[docs] def setFinalState(self, finalStateName): self.__finalState = self.getStateNamed(finalStateName)
[docs] def requestFinalState(self): self.request(self.getFinalState().getName())
[docs] def getCurrentState(self): return(self.__currentState)
# lookup funcs
[docs] def getStateNamed(self, stateName): """ Return the state with given name if found, issue warning otherwise """ state = self.__states.get(stateName) if state: return state else: ClassicFSM.notify.warning("[%s]: getStateNamed: %s, no such state" % (self.__name, stateName))
[docs] def hasStateNamed(self, stateName): """ Return True if stateName is a valid state, False otherwise. """ result = False state = self.__states.get(stateName) if state: result = True return result
# basic ClassicFSM functionality def __exitCurrent(self, argList): """ Exit the current state """ assert self.__internalStateInFlux assert ClassicFSM.notify.debug("[%s]: exiting %s" % (self.__name, self.__currentState.getName())) self.__currentState.exit(argList) # Only send the state change event if we are inspecting it # If this event turns out to be generally useful, we can # turn it on all the time, but for now nobody else is using it if self.inspecting: messenger.send(self.getName() + '_' + self.__currentState.getName() + '_exited') self.__currentState = None def __enter(self, aState, argList=[]): """ Enter a given state, if it exists """ assert self.__internalStateInFlux stateName = aState.getName() if (stateName in self.__states): assert ClassicFSM.notify.debug("[%s]: entering %s" % (self.__name, stateName)) self.__currentState = aState # Only send the state change event if we are inspecting it # If this event turns out to be generally useful, we can # turn it on all the time, but for now nobody else is using it if self.inspecting: messenger.send(self.getName() + '_' + stateName + '_entered') # Once we begin entering the new state, we're allow to # recursively request a transition to another state. # Indicate this by marking our internal state no longer in # flux. self.__internalStateInFlux = 0 aState.enter(argList) else: # notify.error is going to raise an exception; reset the # flux flag first self.__internalStateInFlux = 0 ClassicFSM.notify.error("[%s]: enter: no such state" % (self.__name)) def __transition(self, aState, enterArgList=[], exitArgList=[]): """ Exit currentState and enter given one """ assert not self.__internalStateInFlux self.__internalStateInFlux = 1 self.__exitCurrent(exitArgList) self.__enter(aState, enterArgList) assert not self.__internalStateInFlux
[docs] def request(self, aStateName, enterArgList=[], exitArgList=[], force=0): """ Attempt transition from currentState to given one. Return true is transition exists to given state, false otherwise. """ # If you trigger this assertion failure, you must have # recursively requested a state transition from within the # exitState() function for the previous state. This is not # supported because we're not fully transitioned into the new # state yet. assert not self.__internalStateInFlux if not self.__currentState: # Make this a warning for now ClassicFSM.notify.warning("[%s]: request: never entered initial state" % (self.__name)) self.__currentState = self.__initialState if isinstance(aStateName, str): aState = self.getStateNamed(aStateName) else: # Allow the caller to pass in a state in itself, not just # the name of a state. aState = aStateName aStateName = aState.getName() if aState == None: ClassicFSM.notify.error("[%s]: request: %s, no such state" % (self.__name, aStateName)) # is the transition defined? if it isn't, should we allow it? transitionDefined = self.__currentState.isTransitionDefined(aStateName) transitionAllowed = transitionDefined if self.onUndefTransition == ClassicFSM.ALLOW: transitionAllowed = 1 if not transitionDefined: # the transition is not defined, but we're going to do it # anyway. print a warning. ClassicFSM.notify.warning( "[%s]: performing undefined transition from %s to %s" % (self.__name, self.__currentState.getName(), aStateName)) if transitionAllowed or force: self.__transition(aState, enterArgList, exitArgList) return 1 # We can implicitly always transition to our final state. elif (aStateName == self.__finalState.getName()): if (self.__currentState == self.__finalState): # Do not do the transition if we are already in the # final state assert ClassicFSM.notify.debug( "[%s]: already in final state: %s" % (self.__name, aStateName)) return 1 else: # Force a transition to allow for cleanup assert ClassicFSM.notify.debug( "[%s]: implicit transition to final state: %s" % (self.__name, aStateName)) self.__transition(aState, enterArgList, exitArgList) return 1 # are we already in this state? elif (aStateName == self.__currentState.getName()): assert ClassicFSM.notify.debug( "[%s]: already in state %s and no self transition" % (self.__name, aStateName)) return 0 else: msg = ("[%s]: no transition exists from %s to %s" % (self.__name, self.__currentState.getName(), aStateName)) if self.onUndefTransition == ClassicFSM.ERROR: ClassicFSM.notify.error(msg) elif self.onUndefTransition == ClassicFSM.DISALLOW_VERBOSE: ClassicFSM.notify.warning(msg) return 0
[docs] def forceTransition(self, aStateName, enterArgList=[], exitArgList=[]): """ force a transition -- for debugging ONLY """ self.request(aStateName, enterArgList, exitArgList, force=1)
[docs] def conditional_request(self, aStateName, enterArgList=[], exitArgList=[]): """ 'if this transition is defined, do it' Attempt transition from currentState to given one, if it exists. Return true if transition exists to given state, false otherwise. It is NOT an error/warning to attempt a cond_request if the transition doesn't exist. This lets people be sloppy about ClassicFSM transitions, letting the same fn be used for different states that may not have the same out transitions. """ assert not self.__internalStateInFlux if not self.__currentState: # Make this a warning for now ClassicFSM.notify.warning("[%s]: request: never entered initial state" % (self.__name)) self.__currentState = self.__initialState if isinstance(aStateName, str): aState = self.getStateNamed(aStateName) else: # Allow the caller to pass in a state in itself, not just # the name of a state. aState = aStateName aStateName = aState.getName() if aState == None: ClassicFSM.notify.error("[%s]: request: %s, no such state" % (self.__name, aStateName)) transitionDefined = ( self.__currentState.isTransitionDefined(aStateName) or aStateName in [self.__currentState.getName(), self.__finalState.getName()] ) if transitionDefined: return self.request(aStateName, enterArgList, exitArgList) else: assert ClassicFSM.notify.debug( "[%s]: condition_request: %s, transition doesnt exist" % (self.__name, aStateName)) return 0
[docs] def view(self): # Don't use a regular import, to prevent ModuleFinder from picking # it up as a dependency when building a .p3d package. import importlib FSMInspector = importlib.import_module('direct.tkpanels.FSMInspector') FSMInspector.FSMInspector(self)
[docs] def isInternalStateInFlux(self): return self.__internalStateInFlux