Source Code for Module smach.state

  1   
  2  import threading 
  3  import traceback 
  4   
  5  import smach 
  6   
  7  __all__ = ['State','CBState'] 
  8   
  9 -class State(object): 
 10      """Base class for SMACH states. 
 11   
 12      A SMACH state interacts with SMACH containers in two ways. The first is its 
 13      outcome identifier, and the second is the set of userdata variables which 
 14      it reads from and writes to at runtime. Both of these interactions are 
 15      declared before the state goes active (when its C{execute()} method is 
 16      called) and are checked during construction. 
 17      """ 
 18 -    def __init__(self, outcomes=[], input_keys=[], output_keys=[], io_keys=[]): 
 19          """State constructor 
 20          @type outcomes: list of str 
 21          @param outcomes: Custom outcomes for this state. 
 22   
 23          @type input_keys: list of str 
 24          @param input_keys: The userdata keys from which this state might read 
 25          at runtime.  
 26   
 27          @type output_keys: list of str 
 28          @param output_keys: The userdata keys to which this state might write 
 29          at runtime. 
 30   
 31          @type io_keys: list of str 
 32          @param io_keys: The userdata keys to which this state might write or 
 33          from which it might read at runtime. 
 34          """ 
 35          # Store outcomes 
 36          self._outcomes = set(outcomes) 
 37   
 38          # Store userdata interface description 
 39          self._input_keys = set(input_keys + io_keys) 
 40          self._output_keys = set(output_keys + io_keys) 
 41   
 42          # Declare preempt flag 
 43          self._preempt_requested = False 
 44   
 45      ### Meat 
 46 -    def execute(self, ud): 
 47          """Called when executing a state. 
 48          In the base class this raises a NotImplementedError. 
 49   
 50          @type ud: L{UserData} structure 
 51          @param ud: Userdata for the scope in which this state is executing 
 52          """ 
 53          raise NotImplementedError() 
 54       
 55      ### SMACH Interface API 
 56 -    def register_outcomes(self, new_outcomes): 
 57          """Add outcomes to the outcome set.""" 
 58          self._outcomes = self._outcomes.union(new_outcomes) 
 59   
 60 -    def get_registered_outcomes(self): 
 61          """Get a list of registered outcomes. 
 62          @rtype: tuple of str 
 63          @return: Tuple of registered outcome strings. 
 64          """ 
 65          return tuple(self._outcomes) 
 66   
 67      ### Userdata API 
 68 -    def register_io_keys(self, keys): 
 69          """Add keys to the set of keys from which this state may read and write. 
 70          @type keys: list of str 
 71          @param keys: List of keys which may be read from and written to when this 
 72          state is active. 
 73          """ 
 74          self._input_keys = self._input_keys.union(keys) 
 75          self._output_keys = self._output_keys.union(keys) 
 76   
 77 -    def register_input_keys(self, keys): 
 78          """Add keys to the set of keys from which this state may read. 
 79          @type keys: list of str 
 80          @param keys: List of keys which may be read from when this state is 
 81          active. 
 82          """ 
 83          self._input_keys = self._input_keys.union(keys) 
 84   
 85 -    def get_registered_input_keys(self): 
 86          """Get a tuple of registered input keys.""" 
 87          return tuple(self._input_keys) 
 88   
 89 -    def register_output_keys(self, keys): 
 90          """Add keys to the set of keys to which this state may write. 
 91          @type keys: list of str 
 92          @param keys: List of keys which may be written to when this state is 
 93          active. 
 94          """ 
 95          self._output_keys = self._output_keys.union(keys) 
 96   
 97 -    def get_registered_output_keys(self): 
 98          """Get a tuple of registered output keys.""" 
 99          return tuple(self._output_keys) 
100   
101      ### Preemption interface 
102 -    def request_preempt(self): 
103          """Sets preempt_requested to True""" 
104          self._preempt_requested = True 
105   
106 -    def service_preempt(self): 
107          """Sets preempt_requested to False""" 
108          self._preempt_requested = False 
109   
110 -    def recall_preempt(self): 
111          """Sets preempt_requested to False""" 
112          self._preempt_requested = False 
113   
114 -    def preempt_requested(self): 
115          """True if a preempt has been requested.""" 
116          return self._preempt_requested 
117   
118 -class CBState(State): 
119 -    def __init__(self, cb, cb_args=[], cb_kwargs={}, outcomes=[], input_keys=[], output_keys=[], io_keys=[]): 
120          """Create s state from a single function. 
121   
122          @type outcomes: list of str 
123          @param outcomes: Custom outcomes for this state. 
124   
125          @type input_keys: list of str 
126          @param input_keys: The userdata keys from which this state might read 
127          at runtime.  
128   
129          @type output_keys: list of str 
130          @param output_keys: The userdata keys to which this state might write 
131          at runtime. 
132   
133          @type io_keys: list of str 
134          @param io_keys: The userdata keys to which this state might write or 
135          from which it might read at runtime. 
136          """ 
137          State.__init__(self, outcomes, input_keys, output_keys, io_keys) 
138          self._cb = cb 
139          self._cb_args = cb_args 
140          self._cb_kwargs = cb_kwargs 
141   
142          if smach.util.has_smach_interface(cb): 
143              self._cb_input_keys = cb.get_registered_input_keys() 
144              self._cb_output_keys = cb.get_registered_output_keys() 
145              self._cb_outcomes = cb.get_registered_outcomes() 
146   
147              self.register_input_keys(self._cb_input_keys) 
148              self.register_output_keys(self._cb_output_keys) 
149              self.register_outcomes(self._cb_outcomes) 
150   
151 -    def execute(self, ud): 
152          return self._cb(ud, *self._cb_args, **self._cb_kwargs) 
153