roslaunch.launch

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

238 - def __init__(self, run_id, config, server_uri=None, pmon=None, is_core=False, remote_runner=None, is_child=False, is_rostest=False, num_workers=NUM_WORKERS, timeout=None):

239 """ 240 @param run_id: /run_id for this launch. If the core is not 241 running, this value will be used to initialize /run_id. If 242 the core is already running, this value will be checked 243 against the value stored on the core. L{ROSLaunchRunner} will 244 fail during L{launch()} if they do not match. 245 @type run_id: str 246 @param config: roslauch instance to run 247 @type config: L{ROSLaunchConfig} 248 @param server_uri: XML-RPC URI of roslaunch server. 249 @type server_uri: str 250 @param pmon: optionally override the process 251 monitor the runner uses for starting and tracking processes 252 @type pmon: L{ProcessMonitor} 253 254 @param is_core: if True, this runner is a roscore 255 instance. This affects the error behavior if a master is 256 already running -- aborts if is_core is True and a core is 257 detected. 258 @type is_core: bool 259 @param remote_runner: remote roslaunch process runner 260 @param is_rostest: if True, this runner is a rostest 261 instance. This affects certain validation checks. 262 @type is_rostest: bool 263 @param num_workers: If this is the core, the number of worker-threads to use. 264 @type num_workers: int 265 @param timeout: If this is the core, the socket-timeout to use. 266 @type timeout: Float or None 267 """ 268 if run_id is None: 269 raise RLException("run_id is None") 270 self.run_id = run_id 271 272 # In the future we should can separate the notion of a core 273 # config vs. the config being launched. In that way, we can 274 # start to migrate to a model where a config is a parameter to 275 # a launch() method, rather than a central thing to the 276 # runner. 277 self.config = config 278 self.server_uri = server_uri 279 self.is_child = is_child 280 self.is_core = is_core 281 self.is_rostest = is_rostest 282 self.num_workers = num_workers 283 self.timeout = timeout 284 self.logger = logging.getLogger('roslaunch') 285 self.pm = pmon or start_process_monitor() 286 287 # wire in ProcessMonitor events to our listeners 288 # aggregator. We similarly wire in the remote events when we 289 # create the remote runner. 290 self.listeners = _ROSLaunchListeners() 291 if self.pm is None: 292 raise RLException("unable to initialize roslaunch process monitor") 293 if self.pm.is_shutdown: 294 raise RLException("bad roslaunch process monitor initialization: process monitor is already dead") 295 296 self.pm.add_process_listener(self.listeners) 297 298 self.remote_runner = remote_runner

299

300 - def add_process_listener(self, l):

301 """ 302 Add listener to list of listeners. Not threadsafe. Must be 303 called before processes started. 304 @param l: listener 305 @type l: L{ProcessListener} 306 """ 307 self.listeners.add_process_listener(l)

308

309 - def _load_parameters(self):

310 """ 311 Load parameters onto the parameter server 312 """ 313 self.logger.info("load_parameters starting ...") 314 config = self.config 315 param_server = config.master.get() 316 p = None 317 try: 318 # multi-call style xmlrpc 319 param_server_multi = config.master.get_multi() 320 321 # clear specified parameter namespaces 322 # #2468 unify clear params to prevent error 323 for p in _unify_clear_params(config.clear_params): 324 if param_server.hasParam(_ID, p)[2]: 325 #printlog("deleting parameter [%s]"%p) 326 param_server_multi.deleteParam(_ID, p) 327 r = param_server_multi() 328 for code, msg, _ in r: 329 if code != 1: 330 raise RLException("Failed to clear parameter: %s"%(msg)) 331 332 # multi-call objects are not reusable 333 param_server_multi = config.master.get_multi() 334 for p in config.params.values(): 335 # suppressing this as it causes too much spam 336 #printlog("setting parameter [%s]"%p.key) 337 param_server_multi.setParam(_ID, p.key, p.value) 338 r = param_server_multi() 339 for code, msg, _ in r: 340 if code != 1: 341 raise RLException("Failed to set parameter: %s"%(msg)) 342 except RLException: 343 raise 344 except Exception as e: 345 printerrlog("load_parameters: unable to set parameters (last param was [%s]): %s"%(p,e)) 346 raise #re-raise as this is fatal 347 self.logger.info("... load_parameters complete")

348

349 - def _launch_nodes(self):

350 """ 351 Launch all the declared nodes/master 352 @return: two lists of node names where the first 353 is the nodes that successfully launched and the second is the 354 nodes that failed to launch. 355 @rtype: [[str], [str]] 356 """ 357 config = self.config 358 succeeded = [] 359 failed = [] 360 self.logger.info("launch_nodes: launching local nodes ...") 361 local_nodes = config.nodes 362 363 # don't launch remote nodes 364 local_nodes = [n for n in config.nodes if is_machine_local(n.machine)] 365 366 for node in local_nodes: 367 proc, success = self.launch_node(node) 368 if success: 369 succeeded.append(str(proc)) 370 else: 371 failed.append(str(proc)) 372 373 if self.remote_runner: 374 self.logger.info("launch_nodes: launching remote nodes ...") 375 r_succ, r_fail = self.remote_runner.launch_remote_nodes() 376 succeeded.extend(r_succ) 377 failed.extend(r_fail) 378 379 self.logger.info("... launch_nodes complete") 380 return succeeded, failed

381

382 - def _launch_master(self):

383 """ 384 Launches master if requested. 385 @raise RLException: if master launch fails 386 """ 387 m = self.config.master 388 is_running = m.is_running() 389 390 if self.is_core and is_running: 391 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)) 392 393 if not is_running: 394 validate_master_launch(m, self.is_core, self.is_rostest) 395 396 printlog("auto-starting new master") 397 p = create_master_process(self.run_id, m.type, get_ros_root(), m.get_port(), self.num_workers, self.timeout) 398 self.pm.register_core_proc(p) 399 success = p.start() 400 if not success: 401 raise RLException("ERROR: unable to auto-start master process") 402 timeout_t = time.time() + _TIMEOUT_MASTER_START 403 while not m.is_running() and time.time() < timeout_t: 404 time.sleep(0.1) 405 406 if not m.is_running(): 407 raise RLException("ERROR: could not contact master [%s]"%m.uri) 408 409 printlog_bold("ROS_MASTER_URI=%s"%m.uri) 410 # TODO: although this dependency doesn't cause anything bad, 411 # it's still odd for launch to know about console stuff. This 412 # really should be an event. 413 update_terminal_name(m.uri) 414 415 # Param Server config params 416 param_server = m.get() 417 418 # #773: unique run ID 419 self._check_and_set_run_id(param_server, self.run_id) 420 421 if self.server_uri: 422 # store parent XML-RPC URI on param server 423 # - param name is the /roslaunch/hostname:port so that multiple roslaunches can store at once 424 hostname, port = rosgraph.network.parse_http_host_and_port(self.server_uri) 425 hostname = _hostname_to_rosname(hostname) 426 self.logger.info("setting /roslaunch/uris/%s__%s' to %s"%(hostname, port, self.server_uri)) 427 param_server.setParam(_ID, '/roslaunch/uris/%s__%s'%(hostname, port),self.server_uri)

428

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

430 """ 431 Initialize self.run_id to existing value or setup parameter 432 server with /run_id set to default_run_id 433 @param default_run_id: run_id to use if value is not set 434 @type default_run_id: str 435 @param param_server: parameter server proxy 436 @type param_server: xmlrpclib.ServerProxy 437 """ 438 code, _, val = param_server.hasParam(_ID, '/run_id') 439 if code == 1 and not val: 440 printlog_bold("setting /run_id to %s"%run_id) 441 param_server.setParam('/roslaunch', '/run_id', run_id) 442 else: 443 # verify that the run_id we have been set to matches what's on the parameter server 444 code, _, val = param_server.getParam('/roslaunch', '/run_id') 445 if code != 1: 446 #could only happen in a bad race condition with 447 #someone else restarting core 448 raise RLException("ERROR: unable to retrieve /run_id from parameter server") 449 if run_id != val: 450 raise RLException("run_id on parameter server does not match declared run_id: %s vs %s"%(val, run_id))

451 #self.run_id = val 452 #printlog_bold("/run_id is %s"%self.run_id) 453

454 - def _launch_executable(self, e):

455 """ 456 Launch a single L{Executable} object. Blocks until executable finishes. 457 @param e: Executable 458 @type e: L{Executable} 459 @raise RLException: if exectuable fails. Failure includes non-zero exit code. 460 """ 461 try: 462 #kwc: I'm still debating whether shell=True is proper 463 cmd = e.command 464 cmd = "%s %s"%(cmd, ' '.join(e.args)) 465 print("running %s" % cmd) 466 local_machine = self.config.machines[''] 467 env = setup_env(None, local_machine, self.config.master.uri) 468 retcode = subprocess.call(cmd, shell=True, env=env) 469 if retcode < 0: 470 raise RLException("command [%s] failed with exit code %s"%(cmd, retcode)) 471 except OSError as e: 472 raise RLException("command [%s] failed: %s"%(cmd, e))

473 474 #TODO: _launch_run_executables, _launch_teardown_executables 475 #TODO: define and implement behavior for remote launch

476 - def _launch_setup_executables(self):

477 """ 478 @raise RLException: if exectuable fails. Failure includes non-zero exit code. 479 """ 480 exes = [e for e in self.config.executables if e.phase == PHASE_SETUP] 481 for e in exes: 482 self._launch_executable(e)

483

484 - def _launch_core_nodes(self):

485 """ 486 launch any core services that are not already running. master must 487 be already running 488 @raise RLException: if core launches fail 489 """ 490 config = self.config 491 master = config.master.get() 492 tolaunch = [] 493 for node in config.nodes_core: 494 node_name = rosgraph.names.ns_join(node.namespace, node.name) 495 code, msg, _ = master.lookupNode(_ID, node_name) 496 if code == -1: 497 tolaunch.append(node) 498 elif code == 1: 499 print("core service [%s] found" % node_name) 500 else: 501 print("WARN: master is not behaving well (unexpected return value when looking up node)", file=sys.stderr) 502 self.logger.error("ERROR: master return [%s][%s] on lookupNode call"%(code,msg)) 503 504 for node in tolaunch: 505 node_name = rosgraph.names.ns_join(node.namespace, node.name) 506 name, success = self.launch_node(node, core=True) 507 if success: 508 print("started core service [%s]" % node_name) 509 else: 510 raise RLException("failed to start core service [%s]"%node_name)

511

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

513 """ 514 Launch a single node locally. Remote launching is handled separately by the remote module. 515 If node name is not assigned, one will be created for it. 516 517 @param node Node: node to launch 518 @param core bool: if True, core node 519 @return obj, bool: Process handle, successful launch. If success, return actual Process instance. Otherwise return name. 520 """ 521 self.logger.info("... preparing to launch node of type [%s/%s]", node.package, node.type) 522 523 # TODO: should this always override, per spec?. I added this 524 # so that this api can be called w/o having to go through an 525 # extra assign machines step. 526 if node.machine is None: 527 node.machine = self.config.machines[''] 528 if node.name is None: 529 node.name = rosgraph.names.anonymous_name(node.type) 530 531 master = self.config.master 532 import roslaunch.node_args 533 try: 534 process = create_node_process(self.run_id, node, master.uri) 535 except roslaunch.node_args.NodeParamsException as e: 536 self.logger.error(e) 537 if node.package == 'rosout' and node.type == 'rosout': 538 printerrlog("\n\n\nERROR: rosout is not built. Please run 'rosmake rosout'\n\n\n") 539 else: 540 printerrlog("ERROR: cannot launch node of type [%s/%s]: %s"%(node.package, node.type, str(e))) 541 if node.name: 542 return node.name, False 543 else: 544 return "%s/%s"%(node.package,node.type), False 545 546 self.logger.info("... created process [%s]", process.name) 547 if core: 548 self.pm.register_core_proc(process) 549 else: 550 self.pm.register(process) 551 node.process_name = process.name #store this in the node object for easy reference 552 self.logger.info("... registered process [%s]", process.name) 553 554 # note: this may raise FatalProcessLaunch, which aborts the entire launch 555 success = process.start() 556 if not success: 557 if node.machine.name: 558 printerrlog("launch of %s/%s on %s failed"%(node.package, node.type, node.machine.name)) 559 else: 560 printerrlog("local launch of %s/%s failed"%(node.package, node.type)) 561 else: 562 self.logger.info("... successfully launched [%s]", process.name) 563 return process, success

564

565 - def is_node_running(self, node):

566 """ 567 Check for running node process. 568 @param node Node: node object to check 569 @return bool: True if process associated with node is running (launched && !dead) 570 """ 571 #process_name is not set until node is launched. 572 return node.process_name and self.pm.has_process(node.process_name)

573

574 - def spin_once(self):

575 """ 576 Same as spin() but only does one cycle. must be run from the main thread. 577 """ 578 if not self.pm: 579 return False 580 return self.pm.mainthread_spin_once()

581

582 - def spin(self):

583 """ 584 spin() must be run from the main thread. spin() is very 585 important for roslaunch as it picks up jobs that the process 586 monitor need to be run in the main thread. 587 """ 588 self.logger.info("spin") 589 590 # #556: if we're just setting parameters and aren't launching 591 # any processes, exit. 592 if not self.pm or not self.pm.get_active_names(): 593 printlog_bold("No processes to monitor") 594 self.stop() 595 return # no processes 596 self.pm.mainthread_spin() 597 #self.pm.join() 598 self.logger.info("process monitor is done spinning, initiating full shutdown") 599 self.stop() 600 printlog_bold("done")

601

602 - def stop(self):

603 """ 604 Stop the launch and all associated processes. not thread-safe. 605 """ 606 self.logger.info("runner.stop()") 607 if self.pm is not None: 608 printlog("shutting down processing monitor...") 609 self.logger.info("shutting down processing monitor %s"%self.pm) 610 self.pm.shutdown() 611 self.pm.join() 612 self.pm = None 613 printlog("... shutting down processing monitor complete") 614 else: 615 self.logger.info("... roslaunch runner has already been stopped")

616

617 - def _setup(self):

618 """ 619 Setup the state of the ROS network, including the parameter 620 server state and core services 621 """ 622 # this may have already been done, but do just in case 623 self.config.assign_machines() 624 625 if self.remote_runner: 626 # hook in our listener aggregator 627 self.remote_runner.add_process_listener(self.listeners) 628 629 # start up the core: master + core nodes defined in core.xml 630 self._launch_master() 631 self._launch_core_nodes() 632 633 # run exectuables marked as setup period. this will block 634 # until these executables exit. setup executable have to run 635 # *before* parameters are uploaded so that commands like 636 # rosparam delete can execute. 637 self._launch_setup_executables() 638 639 # no parameters for a child process 640 if not self.is_child: 641 self._load_parameters()

642

643 - def launch(self):

644 """ 645 Run the launch. Depending on usage, caller should call 646 spin_once or spin as appropriate after launch(). 647 @return ([str], [str]): tuple containing list of nodes that 648 successfully launches and list of nodes that failed to launch 649 @rtype: ([str], [str]) 650 @raise RLException: if launch fails (e.g. run_id parameter does 651 not match ID on parameter server) 652 """ 653 try: 654 self._setup() 655 succeeded, failed = self._launch_nodes() 656 return succeeded, failed 657 except KeyboardInterrupt: 658 self.stop() 659 raise

660

661 - def run_test(self, test):

662 """ 663 Run the test node. Blocks until completion or timeout. 664 @param test: test node to run 665 @type test: Test 666 @raise RLTestTimeoutException: if test fails to launch or test times out 667 """ 668 self.logger.info("... preparing to run test [%s] of type [%s/%s]", test.test_name, test.package, test.type) 669 proc_h, success = self.launch_node(test) 670 if not success: 671 raise RLException("test [%s] failed to launch"%test.test_name) 672 673 #poll until test terminates or alloted time exceed 674 timeout_t = time.time() + test.time_limit 675 pm = self.pm 676 while pm.mainthread_spin_once() and self.is_node_running(test): 677 #test fails on timeout 678 if time.time() > timeout_t: 679 raise RLTestTimeoutException( 680 "max time [%ss] allotted for test [%s] of type [%s/%s]" % 681 (test.time_limit, test.test_name, test.package, test.type)) 682 time.sleep(0.1)

Source Code for Module roslaunch.launch