roshlaunch.runner

224 """ 225 Runs a roslaunch. The normal sequence of API calls is L{launch()} 226 followed by L{spin()}. An external thread can call L{stop()}; otherwise 227 the runner will block until an exit signal. Another usage is to 228 call L{launch()} followed by repeated calls to L{spin_once()}. This usage 229 allows the main thread to continue to do work while processes are 230 monitored. 231 """ 232

233 - def __init__(self, run_id, server_uri=None, pmon=None, is_core=False, remote_runner=None):

234 """ 235 @param run_id: /run_id for this launch. If the core is not 236 running, this value will be used to initialize /run_id. If 237 the core is already running, this value will be checked 238 against the value stored on the core. L{ROSLaunchRunner} will 239 fail during L{launch()} if they do not match. 240 @type run_id: str 241 @param server_uri: XML-RPC URI of roslaunch server. 242 @type server_uri: str 243 @param pmon: optionally override the process 244 monitor the runner uses for starting and tracking processes 245 @type pmon: L{ProcessMonitor} 246 247 @param is_core: if True, this runner is a roscore 248 instance. This affects the error behavior if a master is 249 already running -- aborts if is_core is True and a core is 250 detected. 251 @type is_core: bool 252 @param remote_runner: remote roslaunch process runner 253 """ 254 if run_id is None: 255 raise RLException("run_id is None") 256 self.run_id = run_id 257 self.server_uri = server_uri 258 259 self.is_core = is_core 260 self.logger = logging.getLogger('roslaunch') 261 self.pm = pmon or start_process_monitor() 262 263 # wire in ProcessMonitor events to our listeners 264 # aggregator. We similarly wire in the remote events when we 265 # create the remote runner. 266 self.listeners = _ROSLaunchListeners() 267 if self.pm is None: 268 raise RLException("unable to initialize roslaunch process monitor") 269 if self.pm.is_shutdown: 270 raise RLException("bad roslaunch process monitor initialization: process monitor is already dead") 271 272 self.pm.add_process_listener(self.listeners) 273 274 self.remote_runner = remote_runner

275

276 - def add_process_listener(self, l):

277 """ 278 Add listener to list of listeners. Not threadsafe. Must be 279 called before processes started. 280 @param l: listener 281 @type l: L{ProcessListener} 282 """ 283 self.listeners.add_process_listener(l)

284

285 - def _load_parameters(self, config):

286 """ 287 Load parameters onto the parameter server 288 """ 289 self.logger.info("load_parameters starting ...") 290 param_server = config.master.get() 291 p = None 292 try: 293 # multi-call style xmlrpc 294 param_server_multi = config.master.get_multi() 295 296 # clear specified parameter namespaces 297 # #2468 unify clear params to prevent error 298 for p in _unify_clear_params(config.clear_params): 299 if param_server.hasParam(_ID, p)[2]: 300 #printlog("deleting parameter [%s]"%p) 301 param_server_multi.deleteParam(_ID, p) 302 r = param_server_multi() 303 for code, msg, _ in r: 304 if code != 1: 305 raise RLException("Failed to clear parameter: %s"%(msg)) 306 307 # multi-call objects are not reusable 308 param_server_multi = config.master.get_multi() 309 for p in config.params.itervalues(): 310 # suppressing this as it causes too much spam 311 #printlog("setting parameter [%s]"%p.key) 312 param_server_multi.setParam(_ID, p.key, p.value) 313 r = param_server_multi() 314 for code, msg, _ in r: 315 if code != 1: 316 raise RLException("Failed to set parameter: %s"%(msg)) 317 except RLException: 318 raise 319 except Exception, e: 320 printerrlog("load_parameters: unable to set parameters (last param was [%s]): %s"%(p,e)) 321 raise #re-raise as this is fatal 322 self.logger.info("... load_parameters complete")

323 324 # ROSHTODO: should return Process handles instead. RemoteRunner should provide a remote killer

325 - def _launch_nodes(self, config):

326 """ 327 Launch all the declared nodes/master 328 @return: two lists of node names where the first 329 is the nodes that successfully launched and the second is the 330 nodes that failed to launch. 331 @rtype: [[str], [str]] 332 """ 333 succeeded = [] 334 failed = [] 335 self.logger.info("launch_nodes: launching local nodes ...") 336 local_nodes = config.nodes 337 338 # don't launch remote nodes 339 local_nodes = [n for n in config.nodes if is_machine_local(n.machine)] 340 341 for node in local_nodes: 342 proc, success = self.launch_node(config, node) 343 if success: 344 succeeded.append(str(proc)) 345 else: 346 failed.append(str(proc)) 347 348 if self.remote_runner: 349 self.logger.info("launch_nodes: launching remote nodes ...") 350 r_succ, r_fail = self.remote_runner.launch_remote_nodes() 351 succeeded.extend(r_succ) 352 failed.extend(r_fail) 353 354 self.logger.info("... launch_nodes complete") 355 return succeeded, failed

356

357 - def _setup_master(self, config):

358 """ 359 Validates master configuration and changes the master URI if 360 necessary. Also shuts down any existing master. 361 @raise RLException: if existing master cannot be killed 362 """ 363 m = config.master 364 self.logger.info("initial ROS_MASTER_URI is %s", m.uri) 365 if m.auto in [m.AUTO_START, m.AUTO_RESTART]: 366 running = m.is_running() #save state as calls are expensive 367 if m.auto == m.AUTO_RESTART and running: 368 print "shutting down existing master" 369 try: 370 m.get().shutdown(_ID, 'roslaunch restart request') 371 except: 372 pass 373 timeout_t = time.time() + _TIMEOUT_MASTER_STOP 374 while m.is_running() and time.time() < timeout_t: 375 time.sleep(0.1) 376 if m.is_running(): 377 raise RLException("ERROR: could not stop existing master") 378 running = False 379 if not running: 380 # force the master URI to be for this machine as we are starting it locally 381 olduri = m.uri 382 m.uri = remap_localhost_uri(m.uri, True) 383 384 # this block does some basic DNS checks so that we can 385 # warn the user in the _very_ common case that their 386 # hostnames are not configured properly 387 hostname, _ = roslib.network.parse_http_host_and_port(m.uri) 388 local_addrs = roslib.network.get_local_addresses() 389 import socket 390 reverse_ip = socket.gethostbyname(hostname) 391 # 127. check is due to #1260 392 if reverse_ip not in local_addrs and not reverse_ip.startswith('127.'): 393 self.logger.warn("IP address %s local hostname '%s' not in local addresses (%s)."%(reverse_ip, hostname, ','.join(local_addrs))) 394 print >> sys.stderr, \ 395 """WARNING: IP address %s for local hostname '%s' does not appear to match 396 any local IP address (%s). Your ROS nodes may fail to communicate. 397 398 Please use ROS_IP to set the correct IP address to use."""%(reverse_ip, hostname, ','.join(local_addrs)) 399 400 if m.uri != olduri: 401 self.logger.info("changing ROS_MASTER_URI to [%s] for starting master locally", m.uri) 402 print "changing ROS_MASTER_URI to [%s] for starting master locally"%m.uri

403

404 - def _launch_master(self, config):

405 """ 406 Launches master if requested. Must be run after L{_setup_master()}. 407 @raise RLException: if master launch fails 408 """ 409 m = config.master 410 is_running = m.is_running() 411 412 if self.is_core and is_running: 413 raise RLException("roscore cannot run as another roscore/master is already running. \nPlease kill other roscore/master processes before relaunching.\nThe ROS_MASTER_URI is %s"%(m.uri)) 414 415 if not is_running: 416 validate_master_launch(m, self.is_core) 417 418 printlog("auto-starting new master") 419 p = create_master_process(self.run_id, m.type, get_ros_root(), m.get_port()) 420 self.pm.register_core_proc(p) 421 success = p.start() 422 if not success: 423 raise RLException("ERROR: unable to auto-start master process") 424 timeout_t = time.time() + _TIMEOUT_MASTER_START 425 while not m.is_running() and time.time() < timeout_t: 426 time.sleep(0.1) 427 428 if not m.is_running(): 429 raise RLException("ERROR: could not contact master [%s]"%m.uri) 430 431 printlog_bold("ROS_MASTER_URI=%s"%m.uri) 432 # TODO: although this dependency doesn't cause anything bad, 433 # it's still odd for launch to know about console stuff. This 434 # really should be an event. 435 update_terminal_name(m.uri) 436 437 # Param Server config params 438 param_server = m.get() 439 440 # #773: unique run ID 441 self._check_and_set_run_id(param_server, self.run_id) 442 443 if self.server_uri: 444 # store parent XML-RPC URI on param server 445 # - param name is the /roslaunch/hostname:port so that multiple roslaunches can store at once 446 hostname, port = roslib.network.parse_http_host_and_port(self.server_uri) 447 hostname = _hostname_to_rosname(hostname) 448 self.logger.info("setting /roslaunch/uris/%s__%s' to %s"%(hostname, port, self.server_uri)) 449 param_server.setParam(_ID, '/roslaunch/uris/%s__%s'%(hostname, port),self.server_uri)

450

451 - def _check_and_set_run_id(self, param_server, run_id):

452 """ 453 Initialize self.run_id to existing value or setup parameter 454 server with /run_id set to default_run_id 455 @param default_run_id: run_id to use if value is not set 456 @type default_run_id: str 457 @param param_server: parameter server proxy 458 @type param_server: xmlrpclib.ServerProxy 459 """ 460 code, _, val = param_server.hasParam(_ID, '/run_id') 461 if code == 1 and not val: 462 printlog_bold("setting /run_id to %s"%run_id) 463 param_server.setParam('/roslaunch', '/run_id', run_id) 464 else: 465 # verify that the run_id we have been set to matches what's on the parameter server 466 code, _, val = param_server.getParam('/roslaunch', '/run_id') 467 if code != 1: 468 #could only happen in a bad race condition with 469 #someone else restarting core 470 raise RLException("ERROR: unable to retrieve /run_id from parameter server") 471 if run_id != val: 472 raise RLException("run_id on parameter server does not match declared run_id: %s vs %s"%(val, run_id))

473 #self.run_id = val 474 #printlog_bold("/run_id is %s"%self.run_id) 475

476 - def _launch_executable(self, config, e):

477 """ 478 Launch a single L{Executable} object. Blocks until executable finishes. 479 @param e: Executable 480 @type e: L{Executable} 481 @raise RLException: if exectuable fails. Failure includes non-zero exit code. 482 """ 483 try: 484 #kwc: I'm still debating whether shell=True is proper 485 cmd = e.command 486 if isinstance(e, RosbinExecutable): 487 cmd = os.path.join(get_ros_root(), 'bin', cmd) 488 cmd = "%s %s"%(cmd, ' '.join(e.args)) 489 print "running %s"%cmd 490 local_machine = config.machines[''] 491 import roslaunch.node_args 492 env = roslaunch.node_args.create_local_process_env(None, local_machine, config.master.uri) 493 import subprocess 494 retcode = subprocess.call(cmd, shell=True, env=env) 495 if retcode < 0: 496 raise RLException("command [%s] failed with exit code %s"%(cmd, retcode)) 497 except OSError, e: 498 raise RLException("command [%s] failed: %s"%(cmd, e))

499 500 #TODO: _launch_run_executables, _launch_teardown_executables 501 #TODO: define and implement behavior for remote launch

502 - def _launch_setup_executables(self, config):

503 """ 504 @raise RLException: if exectuable fails. Failure includes non-zero exit code. 505 """ 506 exes = [e for e in config.executables if e.phase == PHASE_SETUP] 507 for e in exes: 508 self._launch_executable(config, e)

509

510 - def _launch_core_nodes(self, config):

511 """ 512 launch any core services that are not already running. master must 513 be already running 514 @raise RLException: if core launches fail 515 """ 516 import roslib.names 517 518 master = config.master.get() 519 tolaunch = [] 520 for node in config.nodes_core: 521 node_name = roslib.names.ns_join(node.namespace, node.name) 522 code, msg, _ = master.lookupNode(_ID, node_name) 523 if code == -1: 524 tolaunch.append(node) 525 elif code == 1: 526 print "core service [%s] found"%node_name 527 else: 528 print >> sys.stderr, "WARN: master is not behaving well (unexpected return value when looking up node)" 529 self.logger.error("ERROR: master return [%s][%s] on lookupNode call"%(code,msg)) 530 531 for node in tolaunch: 532 node_name = roslib.names.ns_join(node.namespace, node.name) 533 name, success = self.launch_node(node, core=True) 534 if success: 535 print "started core service [%s]"%node_name 536 else: 537 raise RLException("failed to start core service [%s]"%node_name)

538

539 - def launch_node(self, config, node, core=False):

540 """ 541 Launch a single node locally. Remote launching is handled separately by the remote module. 542 If node name is not assigned, one will be created for it. 543 544 @param node Node: node to launch 545 @param core bool: if True, core node 546 @return obj, bool: Process handle, successful launch. If success, return actual Process instance. Otherwise return name. 547 """ 548 self.logger.info("... preparing to launch node of type [%s/%s]", node.package, node.type) 549 550 # TODO: should this always override, per spec?. I added this 551 # so that this api can be called w/o having to go through an 552 # extra assign machines step. 553 if node.machine is None: 554 node.machine = config.machines[''] 555 if node.name is None: 556 node.name = roslib.names.anonymous_name(node.type) 557 558 master = config.master 559 import roslaunch.node_args 560 try: 561 process = create_node_process(self.run_id, node, master.uri) 562 except roslaunch.node_args.NodeParamsException, e: 563 self.logger.error(e) 564 if node.package == 'rosout' and node.type == 'rosout': 565 printerrlog("\n\n\nERROR: rosout is not built. Please run 'rosmake rosout'\n\n\n") 566 else: 567 printerrlog("ERROR: cannot launch node of type [%s/%s]: %s"%(node.package, node.type, str(e))) 568 if node.name: 569 return node.name, False 570 else: 571 return "%s/%s"%(node.package,node.type), False 572 573 self.logger.info("... created process [%s]", process.name) 574 if core: 575 self.pm.register_core_proc(process) 576 else: 577 self.pm.register(process) 578 node.process_name = process.name #store this in the node object for easy reference 579 self.logger.info("... registered process [%s]", process.name) 580 581 # note: this may raise FatalProcessLaunch, which aborts the entire launch 582 success = process.start() 583 if not success: 584 if node.machine.name: 585 printerrlog("launch of %s/%s on %s failed"%(node.package, node.type, node.machine.name)) 586 else: 587 printerrlog("local launch of %s/%s failed"%(node.package, node.type)) 588 else: 589 self.logger.info("... successfully launched [%s]", process.name) 590 return process, success

591

592 - def is_node_running(self, node):

593 """ 594 Check for running node process. 595 @param node Node: node object to check 596 @return bool: True if process associated with node is running (launched && !dead) 597 """ 598 #process_name is not set until node is launched. 599 return node.process_name and self.pm.has_process(node.process_name)

600

601 - def spin_once(self):

602 """ 603 Same as spin() but only does one cycle. must be run from the main thread. 604 """ 605 if not self.pm: 606 return False 607 return self.pm.mainthread_spin_once()

608

609 - def spin(self):

610 """ 611 spin() must be run from the main thread. spin() is very 612 important for roslaunch as it picks up jobs that the process 613 monitor need to be run in the main thread. 614 """ 615 self.logger.info("spin") 616 617 # #556: if we're just setting parameters and aren't launching 618 # any processes, exit. 619 if not self.pm or not self.pm.get_active_names(): 620 printlog_bold("No processes to monitor") 621 self.stop() 622 return # no processes 623 self.pm.mainthread_spin() 624 #self.pm.join() 625 self.logger.info("process monitor is done spinning, initiating full shutdown") 626 self.stop() 627 printlog_bold("done")

628

629 - def stop(self):

630 """ 631 Stop the launch and all associated processes. not thread-safe. 632 """ 633 self.logger.info("runner.stop()") 634 if self.pm is not None: 635 printlog("shutting down processing monitor...") 636 self.logger.info("shutting down processing monitor %s"%self.pm) 637 self.pm.shutdown() 638 self.pm.join() 639 self.pm = None 640 printlog("... shutting down processing monitor complete") 641 else: 642 self.logger.info("... roslaunch runner has already been stopped")

643

644 - def _setup(self, config):

645 """ 646 Setup the state of the ROS network, including the parameter 647 server state and core services 648 """ 649 # this may have already been done, but do just in case 650 config.assign_machines() 651 652 if self.remote_runner: 653 # hook in our listener aggregator 654 self.remote_runner.add_process_listener(self.listeners) 655 656 # start up the core: master + core nodes defined in core.xml 657 self._launch_master(config) 658 self._launch_core_nodes(config) 659 660 # run exectuables marked as setup period. this will block 661 # until these executables exit. setup executable have to run 662 # *before* parameters are uploaded so that commands like 663 # rosparam delete can execute. 664 self._launch_setup_executables(config) 665 666 # no parameters for a child process 667 self._load_parameters(config)

668

669 - def launch(self, config):

670 """ 671 Run the launch. Depending on usage, caller should call 672 spin_once or spin as appropriate after launch(). 673 @return ([str], [str]): tuple containing list of nodes that 674 successfully launches and list of nodes that failed to launch 675 @rtype: ([str], [str]) 676 @raise RLException: if launch fails (e.g. run_id parameter does 677 not match ID on parameter server) 678 """ 679 try: 680 self._setup(config) 681 succeeded, failed = self._launch_nodes(config) 682 return succeeded, failed 683 except KeyboardInterrupt: 684 self.stop() 685 raise

686 687 #ROSHTODO: move into rostest

688 - def run_test(self, config, test):

689 """ 690 Run the test node. Blocks until completion or timeout. 691 @param test: test node to run 692 @type test: Test 693 @raise RLTestTimeoutException: if test fails to launch or test times out 694 """ 695 self.logger.info("... preparing to run test [%s] of type [%s/%s]", test.test_name, test.package, test.type) 696 proc_h, success = self.launch_node(config, test) 697 if not success: 698 raise RLException("test [%s] failed to launch"%test.test_name) 699 700 #poll until test terminates or alloted time exceed 701 timeout_t = time.time() + test.time_limit 702 pm = self.pm 703 while pm.mainthread_spin_once() and self.is_node_running(test): 704 #test fails on timeout 705 if time.time() > timeout_t: 706 raise RLTestTimeoutException("test max time allotted") 707 time.sleep(0.1)

Source Code for Module roshlaunch.runner