Source Code for Module roslaunch.rlutil

  1  # Software License Agreement (BSD License) 
  2  # 
  3  # Copyright (c) 2009, Willow Garage, Inc. 
  4  # All rights reserved. 
  5  # 
  6  # Redistribution and use in source and binary forms, with or without 
  7  # modification, are permitted provided that the following conditions 
  8  # are met: 
  9  # 
 10  #  * Redistributions of source code must retain the above copyright 
 11  #    notice, this list of conditions and the following disclaimer. 
 12  #  * Redistributions in binary form must reproduce the above 
 13  #    copyright notice, this list of conditions and the following 
 14  #    disclaimer in the documentation and/or other materials provided 
 15  #    with the distribution. 
 16  #  * Neither the name of Willow Garage, Inc. nor the names of its 
 17  #    contributors may be used to endorse or promote products derived 
 18  #    from this software without specific prior written permission. 
 19  # 
 20  # THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS 
 21  # "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT 
 22  # LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS 
 23  # FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE 
 24  # COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, 
 25  # INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, 
 26  # BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; 
 27  # LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER 
 28  # CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT 
 29  # LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN 
 30  # ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE 
 31  # POSSIBILITY OF SUCH DAMAGE. 
 32  # 
 33  # Revision $Id: rlutil.py 11756 2010-10-25 09:51:08Z kwc $ 
 34   
 35  """ 
 36  Uncategorized utility routines for roslaunch. 
 37   
 38  This API should not be considered stable. 
 39  """ 
 40   
 41  import os 
 42  import time 
 43   
 44  from roslib.names import SEP 
 45   
 46  import roslaunch.core 
 47  from rosmaster import DEFAULT_MASTER_PORT 
 48   
 49 -def check_log_disk_usage(): 
 50      """ 
 51      Check size of log directory. If high, print warning to user 
 52      """ 
 53      try: 
 54          import rosclean 
 55          import roslib.rosenv 
 56          d = roslib.rosenv.get_log_dir() 
 57          roslaunch.core.printlog("Checking log directory for disk usage. This may take awhile.\nPress Ctrl-C to interrupt")  
 58          disk_usage = rosclean.get_disk_usage(d) 
 59          # warn if over a gig 
 60          if disk_usage > 1073741824: 
 61              roslaunch.core.printerrlog("WARNING: disk usage in log directory [%s] is over 1GB.\nIt's recommended that you use the 'rosclean' command."%d) 
 62          else: 
 63              roslaunch.core.printlog("Done checking log file disk usage. Usage is <1GB.")             
 64      except: 
 65          pass 
 66   
 67 -def resolve_launch_arguments(args): 
 68      """ 
 69      Resolve command-line args to roslaunch filenames. 
 70      @return: resolved filenames 
 71      @rtype: [str] 
 72      """ 
 73      import roslib.packages 
 74      import roslib.scriptutil 
 75   
 76      # strip remapping args for processing 
 77      args = roslib.scriptutil.myargv(args) 
 78       
 79      # user can either specify: 
 80      #  - filename + launch args 
 81      #  - package + relative-filename + launch args 
 82      if not args: 
 83          return args 
 84      resolved_args = None 
 85      top = args[0] 
 86      if os.path.isfile(top): 
 87          resolved_args = [top] + args[1:] 
 88      elif len(args) == 1: 
 89          raise roslaunch.core.RLException("[%s] does not exist. please specify a package and launch file"%(top)) 
 90      else: 
 91          try: 
 92              resolved = roslib.packages.find_resource(top, args[1]) 
 93              if len(resolved) == 1: 
 94                  resolved = resolved[0] 
 95              elif len(resolved) > 1: 
 96                  raise roslaunch.core.RLException("multiple files named [%s] in package [%s].\nPlease specify full path instead"%(args[1], top)) 
 97          except roslib.packages.InvalidROSPkgException, e: 
 98              raise roslaunch.core.RLException("[%s] is not a package or launch file name"%top) 
 99          if not resolved: 
100              raise roslaunch.core.RLException("cannot locate [%s] in package [%s]"%(args[1], top)) 
101          else: 
102              resolved_args = [resolved] + args[2:] 
103      return resolved_args 
104   
105 -def _wait_for_master(): 
106      """ 
107      Block until ROS Master is online 
108       
109      @raise RuntimeError: if unexpected error occurs 
110      """ 
111      m = roslaunch.core.Master() # get a handle to the default master 
112      is_running = m.is_running() 
113      if not is_running: 
114          roslaunch.core.printlog("roscore/master is not yet running, will wait for it to start") 
115      while not is_running: 
116          time.sleep(0.1) 
117          is_running = m.is_running() 
118      if is_running: 
119          roslaunch.core.printlog("master has started, initiating launch") 
120      else: 
121          raise RuntimeError("unknown error waiting for master to start") 
122   
123  _terminal_name = None 
124   
125 -def _set_terminal(s): 
126      import platform 
127      if platform.system() in ['FreeBSD', 'Linux', 'Darwin', 'Unix']: 
128          try: 
129              print '\033]2;%s\007'%(s) 
130          except: 
131              pass 
132       
133 -def update_terminal_name(ros_master_uri): 
134      """ 
135      append master URI to the terminal name 
136      """ 
137      if _terminal_name: 
138          _set_terminal(_terminal_name + ' ' + ros_master_uri) 
139   
140 -def change_terminal_name(args, is_core): 
141      """ 
142      use echo (where available) to change the name of the terminal window 
143      """ 
144      global _terminal_name 
145      _terminal_name = 'roscore' if is_core else ','.join(args) 
146      _set_terminal(_terminal_name) 
147   
148 -def get_or_generate_uuid(options_runid, options_wait_for_master): 
149      """ 
150      @param options_runid: run_id value from command-line or None 
151      @type  options_runid: str 
152      @param options_wait_for_master: the wait_for_master command 
153        option. If this is True, it means that we must retrieve the 
154        value from the parameter server and need to avoid any race 
155        conditions with the roscore being initialized. 
156      @type  options_wait_for_master: bool 
157      """ 
158      import roslib.scriptutil 
159      # Three possible sources of the run_id: 
160      # 
161      #  - if we're a child process, we get it from options_runid 
162      #  - if there's already a roscore running, read from the param server 
163      #  - generate one if we're running the roscore 
164      if options_runid: 
165          return options_runid 
166   
167      # #773: Generate a run_id to use if we launch a master 
168      # process.  If a master is already running, we'll get the 
169      # run_id from it instead 
170      param_server = roslib.scriptutil.get_param_server() 
171      val = None 
172      while val is None: 
173          try: 
174              code, msg, val = param_server.getParam('/roslaunch', '/run_id') 
175              if code == 1: 
176                  return val 
177              else: 
178                  val = None 
179                  raise RuntimeError("unknown error communicating with Parameter Server: %s"%msg) 
180          except: 
181              if not options_wait_for_master: 
182                  val = roslaunch.core.generate_run_id() 
183      return val 
184       
185 -def check_roslaunch(f): 
186      """ 
187      Check roslaunch file for errors, returning error message if check fails. This routine 
188      is mainly to support rostest's roslaunch_check. 
189   
190      @param f: roslaunch file name 
191      @type  f: str 
192      @return: error message or None 
193      @rtype: str 
194      """ 
195      try: 
196          import roslaunch.config 
197          config = roslaunch.config.load_config_default([f], DEFAULT_MASTER_PORT, verbose=False) 
198      except roslaunch.RLException, e: 
199          return str(e) 
200       
201      errors = [] 
202      # check for missing deps 
203      import roslaunch.depends 
204      base_pkg, file_deps, missing = roslaunch.depends.roslaunch_deps([f]) 
205      for pkg, miss in missing.iteritems(): 
206          if miss: 
207              errors.append("Missing manifest dependencies: %s/manifest.xml: %s"%(pkg, ', '.join(miss))) 
208       
209      # load all node defs 
210      nodes = [] 
211      for filename, rldeps in file_deps.iteritems(): 
212          nodes.extend(rldeps.nodes) 
213   
214      # check for missing packages 
215      import roslib.packages 
216      for pkg, node_type in nodes: 
217          try: 
218              roslib.packages.get_pkg_dir(pkg, required=True) 
219          except: 
220              errors.append("cannot find package [%s] for node [%s]"%(pkg, node_type)) 
221   
222      # check for missing nodes 
223      for pkg, node_type in nodes: 
224          try: 
225              if not roslib.packages.find_node(pkg, node_type): 
226                  errors.append("cannot find node [%s] in package [%s]"%(node_type, pkg)) 
227          except Exception, e: 
228              errors.append("unable to find node [%s/%s]: %s"%(pkg, node_type, str(e))) 
229                   
230      # Check for configuration errors, #2889 
231      for err in config.config_errors: 
232          errors.append('ROSLaunch config error: %s' % err) 
233   
234      if errors: 
235          return '\n'.join(errors) 
236                             
237   
238 -def namespaces_of(name): 
239      if name is None:  
240          raise ValueError('name') 
241      if not isinstance(name, basestring): 
242          raise TypeError('name') 
243      if not name: 
244          return [SEP] 
245   
246      splits = [x for x in name.split(SEP) if x] 
247      return ['/'] + ['/'+SEP.join(splits[:i]) for i in xrange(1, len(splits))] 
248