Source Code for Module intera_interface.cuff

  1  # Copyright (c) 2013-2017, Rethink Robotics Inc. 
  2  # 
  3  # Licensed under the Apache License, Version 2.0 (the "License"); 
  4  # you may not use this file except in compliance with the License. 
  5  # You may obtain a copy of the License at 
  6  # 
  7  #     http://www.apache.org/licenses/LICENSE-2.0 
  8  # 
  9  # Unless required by applicable law or agreed to in writing, software 
 10  # distributed under the License is distributed on an "AS IS" BASIS, 
 11  # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 
 12  # See the License for the specific language governing permissions and 
 13  # limitations under the License. 
 14   
 15  import rospy 
 16  import intera_dataflow 
 17  from intera_io import IODeviceInterface 
 18  from intera_core_msgs.msg import IODeviceConfiguration 
 19  from robot_params import RobotParams 
 20   
 21   
 22 -class Cuff(object): 
 23      """ 
 24      Interface class for cuff on the Intera robots. 
 25      """ 
 26   
 27 -    def __init__(self, limb="right"): 
 28          """ 
 29          Constructor. 
 30   
 31          @type limb: str 
 32          @param limb: limb side to interface 
 33          """ 
 34          params = RobotParams() 
 35          limb_names = params.get_limb_names() 
 36          if limb not in limb_names: 
 37              rospy.logerr("Cannot detect Cuff's limb {0} on this robot." 
 38                           " Valid limbs are {1}. Exiting Cuff.init().".format( 
 39                                                              limb, limb_names)) 
 40              return 
 41          self.limb = limb 
 42          self.device = None 
 43          self.name = "cuff" 
 44          self.cuff_config_sub = rospy.Subscriber('/io/robot/cuff/config', IODeviceConfiguration, self._config_callback) 
 45          # Wait for the cuff status to be true 
 46          intera_dataflow.wait_for( 
 47              lambda: not self.device is None, timeout=5.0, 
 48              timeout_msg=("Failed find cuff on limb '{0}'.".format(limb)) 
 49              ) 
 50          self._cuff_io = IODeviceInterface("robot", self.name) 
 51   
 52 -    def _config_callback(self, msg): 
 53          """ 
 54          config topic callback 
 55          """ 
 56          if msg.device != []: 
 57              if str(msg.device.name) == self.name: 
 58                  self.device = msg.device 
 59   
 60 -    def lower_button(self): 
 61          """ 
 62          Returns a boolean describing whether the lower button on cuff is pressed. 
 63   
 64          @rtype: bool 
 65          @return: a variable representing button state: (True: pressed, False: unpressed) 
 66          """ 
 67          return bool(self._cuff_io.get_signal_value('_'.join([self.limb, "button_lower"]))) 
 68   
 69 -    def upper_button(self): 
 70          """ 
 71          Returns a boolean describing whether the upper button on cuff is pressed. 
 72          (True: pressed, False: unpressed) 
 73          @rtype: bool 
 74          @return:  a variable representing button state: (True: pressed, False: unpressed) 
 75          """ 
 76          return bool(self._cuff_io.get_signal_value('_'.join([self.limb, "button_upper"]))) 
 77   
 78 -    def cuff_button(self): 
 79          """ 
 80          Returns a boolean describing whether the cuff button on cuff is pressed. 
 81          (True: pressed, False: unpressed) 
 82          @rtype: bool 
 83          @return:  a variable representing cuff button state: (True: pressed, False: unpressed) 
 84          """ 
 85          return bool(self._cuff_io.get_signal_value('_'.join([self.limb, "cuff"]))) 
 86   
 87 -    def register_callback(self, callback_function, signal_name, poll_rate=10): 
 88          """ 
 89          Registers a supplied callback to a change in state of supplied 
 90          signal_name's value. Spawns a thread that will call the callback with 
 91          the updated value. 
 92   
 93          @type callback_function: function 
 94          @param callback_function: function handle for callback function 
 95          @type signal_name: str 
 96          @param signal_name: the name of the signal to poll for value change 
 97          @type poll_rate: int 
 98          @param poll_rate: the rate at which to poll for a value change (in a separate 
 99                  thread) 
100   
101          @rtype: str 
102          @return: callback_id retuned if the callback was registered, and an 
103                   empty string if the requested signal_name does not exist in the 
104                   Navigator 
105          """ 
106          return self._cuff_io.register_callback( 
107                                   callback_function=callback_function, 
108                                   signal_name=signal_name, 
109                                   poll_rate=poll_rate) 
110   
111 -    def deregister_callback(self, callback_id): 
112          """ 
113          Deregisters a callback based on the supplied callback_id. 
114   
115          @type callback_id: str 
116          @param callback_id: the callback_id string to deregister 
117   
118          @rtype: bool 
119          @return: returns bool True if the callback was successfully 
120                   deregistered, and False otherwise. 
121          """ 
122          return self._cuff_io.deregister_callback(callback_id) 
123