Source Code for Module smach.util

  1  import smach 
  2   
  3  __all__ = ['is_shutdown', 'set_shutdown_check', 
  4             'cb_interface', 'has_smach_interface', 'CBInterface'] 
  5   
  6   
  7 -def is_shutdown(): 
  8      return False 
  9   
 10   
 11 -def set_shutdown_check(cb): 
 12      smach.is_shutdown = cb 
 13   
 14   
 15 -def has_smach_interface(obj): 
 16      """Returns True if the object has SMACH interface accessors.""" 
 17      return (hasattr(obj, 'get_registered_input_keys') and 
 18              hasattr(obj, 'get_registered_output_keys') and 
 19              hasattr(obj, 'get_registered_outcomes')) 
 20   
 21   
 22  # Callback decorator for describing userdata 
 23 -class cb_interface(object): 
 24 -    def __init__(self, outcomes=None, input_keys=None, output_keys=None): 
 25          self._outcomes = outcomes or [] 
 26          self._input_keys = input_keys or [] 
 27          self._output_keys = output_keys or [] 
 28   
 29 -    def __call__(self, cb): 
 30          return CBInterface(cb, self._outcomes, self._input_keys, self._output_keys) 
 31   
 32   
 33 -class CBInterface(object): 
 34      """Decorator to describe the extension of a state's SMACH userdata and outcome interface. 
 35   
 36      Some SMACH states can be extended with the use of user callbacks. Since 
 37      the SMACH interface and SMACH userdata are strictly controlled, the ways in 
 38      which these callbacks interact with SMACH must be delcared. This decorator 
 39      allows this information to be attached to a given callback function. 
 40   
 41      If a callback adds a potential outcome to a state, suppose 'critical_failure', 
 42      then one could write this when defining the callback: 
 43   
 44      >>> import smach 
 45      >>> @smach.cb_interface(outcomes=['critical_failure']) 
 46      >>> def my_cb(x,y,z): 
 47      >>>     # User code 
 48      >>>     return 'critical_failure' 
 49   
 50      Suppose a state retrieves data that it passes into a callback. If the user 
 51      wants to take that data and put some of all of it into userdata, this 
 52      interface must be declared. In this case, the user could write: 
 53   
 54      >>> import smach 
 55      >>> @smach.cb_interface(output_keys=['processed_res']) 
 56      >>> def my_cb(ud, data): 
 57      >>>     ud.processed_res = data 
 58   
 59      """ 
 60 -    def __init__(self, cb, outcomes=None, input_keys=None, output_keys=None, 
 61                   io_keys=None): 
 62          outcomes = outcomes or [] 
 63          input_keys = input_keys or [] 
 64          output_keys = output_keys or [] 
 65          io_keys = io_keys or [] 
 66          """Describe callback SMACH interface. 
 67   
 68          @type outcomes: array of strings 
 69          @param outcomes: Custom outcomes for this state. 
 70   
 71          @type input_keys: array of strings 
 72          @param input_keys: The userdata keys from which this state might read 
 73          at runtime. 
 74   
 75          @type output_keys: array of strings 
 76          @param output_keys: The userdata keys to which this state might write 
 77          at runtime. 
 78   
 79          @type io_keys: array of strings 
 80          @param io_keys: The userdata keys to which this state might write or 
 81          from which it might read at runtime. 
 82          """ 
 83   
 84          self._input_keys = set(input_keys) 
 85          self._input_keys.union(io_keys) 
 86   
 87          self._output_keys = set(output_keys) 
 88          self._output_keys.union(io_keys) 
 89   
 90          self._outcomes = outcomes 
 91   
 92          self._cb = cb 
 93   
 94 -    def __call__(self, *args, **kwargs): 
 95          return self._cb(*args, **kwargs) 
 96   
 97      # SMACH Interface API 
 98 -    def get_registered_input_keys(self): 
 99          """Get a tuple of registered input keys.""" 
100          return tuple(self._input_keys) 
101   
102 -    def get_registered_output_keys(self): 
103          """Get a tuple of registered output keys.""" 
104          return tuple(self._output_keys) 
105   
106 -    def get_registered_outcomes(self): 
107          """Get a list of registered outcomes. 
108          @rtype: tuple of string 
109          @return: Tuple of registered outcome strings. 
110          """ 
111          return tuple(self._outcomes) 
112