ros_sugar.launch.launcher#

Launcher

Module Contents#

Classes#

Launcher

Launcher is a pythonic front-end for bringing up a stack of ROS2 components.

API#

class ros_sugar.launch.launcher.Launcher(namespace: str = '', config_file: Optional[str] = None, activation_timeout: Optional[float] = None, robot_plugin: Optional[ros_sugar.robot.RobotPlugin] = None)#

Launcher is a pythonic front-end for bringing up a stack of ROS2 components.

A Launcher groups one or more components into a launch description, manages their lifecycle, wires up an internal :class:Monitor node to coordinate their activation and to route events and actions, and can optionally serve a web UI for them.

What it does

  • Starts components as ROS2 nodes, either in separate processes (multiprocessing=True on :meth:add_pkg) or in threads within the launcher’s own process (threaded default).

  • Applies a shared ROS2 namespace and optional config file to all components.

  • Manages lifecycle transitions so every lifecycle component reaches the active state once the ROS graph confirms it is discoverable.

  • Dispatches events to actions. Every event/action pair registered via

    meth:

    add_pkg or the component’s own on_fail hook flows through the internal Monitor: the Monitor detects triggers and either invokes the action directly (component actions) or emits an internal event back to the Launcher which executes the corresponding launch action.

  • Process-level crash recovery via :meth:on_process_fail: when enabled, multi-process components that exit unexpectedly are respawned and driven back to the active state, up to a configurable retry cap. Clean exits, shutdown, and user signals (Ctrl+C, SIGTERM) are not treated as crashes and do not trigger respawns.

  • Optional web UI via :meth:enable_ui.

Events and actions

Use this project’s richer event/action system instead of the low-level ROS2 launch event primitives. See :class:~ros_sugar.core.event.Event,

Class:

~ros_sugar.core.action.Action, and the events_actions parameter on :meth:add_pkg: events can be built from topic conditions, compositional boolean expressions, internal signals, or arbitrary callables, and they can drive component methods, lifecycle transitions, or any ROS2 launch action.

add_pkg(components: List[ros_sugar.core.component.BaseComponent], package_name: Optional[str] = None, executable_entry_point: Optional[str] = 'executable', events_actions: Optional[Mapping[ros_sugar.core.event.Event, Union[ros_sugar.core.action.Action, launch.action.Action, List[Union[ros_sugar.core.action.Action, launch.action.Action]]]]] = None, multiprocessing: bool = False, activate_all_components_on_start: bool = True, components_to_activate_on_start: Optional[List[ros_sugar.core.component.BaseComponent]] = None, ros_log_level: Optional[str] = None, rclpy_log_level: Optional[str] = None)#

Add component or a set of components to the launcher from one ROS2 package based on ros_sugar

Parameters:

  • components (List[BaseComponent]) – Component to launch and manage

  • package_name (str, optional) – Components ROS2 package name. Required for multi-process run, defaults to None

  • executable_entry_point (str, optional) – Components ROS2 entry point name. Required for multi-process run, defaults to “executable”

  • events_actions (Dict[ Event, Union[Action, ROSLaunchAction, List[Union[Action, ROSLaunchAction]]] ] | None, optional) – Events/Actions to monitor, defaults to None

  • multiprocessing (bool, optional) – Run the components in multi-processes, otherwise runs in multi-threading, defaults to False

  • activate_all_components_on_start (bool, optional) – To activate all the ROS2 lifecycle nodes on bringup, defaults to False

  • components_to_activate_on_start (Optional[List[BaseComponent]], optional) – Set of components to activate on bringup, defaults to None

  • ros_log_level (str, optional) – Selected logging level for the package components. If provided, it overrides the components ‘log_level’ config parameter, defaults to None

  • rclpy_log_level (str, optional) – Selected ROS internal (RCLPY and RMW) logging level for the package components, defaults to None

on(event: ros_sugar.core.event.Event, action: Union[ros_sugar.core.action.Action, launch.action.Action, List[Union[ros_sugar.core.action.Action, launch.action.Action]]]) → None#

Register an event/action mapping on the launcher.

Convenience sugar equivalent to passing events_actions={event: action} to add_pkg; especially handy with robot-plugin-provided events and action factories::

launcher.on(plugin.events.fall_detected(), plugin.actions.stand_up())
Parameters:

  • event (Event) – The event to monitor.

  • action (Union[Action, ROSLaunchAction, List[Union[Action, ROSLaunchAction]]]) – The action (or list of actions) to run when the event fires.

enable_ui(inputs: Optional[List[Union[ros_sugar.io.Topic, ros_sugar.base_clients.ServiceClientConfig, ros_sugar.base_clients.ActionClientConfig]]] = None, outputs: Optional[List[ros_sugar.io.Topic]] = None, port: int = 5001, ssl_keyfile_path: str = 'key.pem', ssl_certificate_path: str = 'cert.pem', hide_settings_panel: bool = False, serve_browser: bool = True, api_stream_default_rate: float = 10.0, api_max_stream_rate: float = 30.0)#

Enables the user interface (UI) subsystem for recipes, initializing all UI extensions to automatically generate front-end controls and data visualizations .

This method collects and serializes UI input and output elements from all registered UI extensions in :data:UI_EXTENSIONS, and prepares them for runtime interaction. It also configures SSL/TLS settings for the UI server and sets the topics through which the UI communicates with the rest of the system.

Parameters:

  • inputs (Optional[List[Topic]]) – A list of topics that serve as UI input sources. These topics are monitored and reflected in the UI. If None, no input topics are bound.

  • outputs (Optional[List[Topic]]) – A list of topics that serve as UI output sinks. If None, no output topics are bound.

  • port (int) – The TCP port on which the UI server will listen for connections. Defaults to 5001.

  • ssl_keyfile_path (str) – Path to the private key file used for SSL/TLS encryption. Defaults to "key.pem".

  • ssl_certificate_path (str) – Path to the SSL/TLS certificate file used to authenticate the UI server. Defaults to "cert.pem".

  • hide_settings_panel (bool, default False) – Disable the components settings panel in the UI.

  • serve_browser (bool, default True) – Serve the browser front-end alongside the JSON/WebSocket API. Set to False to serve the API only, which requires just starlette and uvicorn instead of the browser dependencies (FastHTML/MonsterUI).

  • api_stream_default_rate (float, default 10.0) – Rate (Hz) at which the JSON API streams rate-sampled output topics to a connected client that does not request a rate explicitly.

  • api_max_stream_rate (float, default 30.0) – Hard upper bound (Hz) for a client-requested API stream rate.

property robot: Dict[str, Any]#

Getter of robot config for all components

Returns:

Robot configuration

Return type:

RobotConfig

add_plugin(plugin: ros_sugar.robot.Plugin, mount: Optional[ros_sugar.robot.Mount] = None) → None#

Attach a plugin to the recipe.

Every component in the recipe is given every attached plugin.

Parameters:

  • plugin (Plugin) – The plugin to attach. Its id – set with id= at construction, or derived from its name – is how a topic addresses it and how its channels are namespaced.

  • mount (Optional[Mount]) – Where this sensor sits, when nothing else publishes its frame into TF. Omit it if a URDF, a robot_state_publisher or the sensor’s own driver already does. A robot plugin’s own sensor placements come from its mounts list instead.

Raises:

ValueError – If a second robot plugin is attached, or if the id is already taken by another plugin.

property frames: Dict[str, Any]#

Getter of robot frames for all components

Returns:

Robot frames configuration

Return type:

RobotFrames

property robot_frame: Dict[str, str]#

Getter of the robot body frame of all components

Returns:

Robot body frame per component

Return type:

Dict[str, str]

property world_frame: Dict[str, str]#

Getter of the world frame of all components

Returns:

World frame per component

Return type:

Dict[str, str]

inputs(**kwargs)#

Update input in all components if exists

outputs(**kwargs)#

Update output in all components if exists

start(node_name: str, **_) → ros_sugar.utils.SomeEntitiesType#

Action to start a node: configure + activate

Parameters:

node_name (str) – description

Returns:

Launch actions

Return type:

List[SomeEntitiesType]

stop(node_name: str, **_) → ros_sugar.utils.SomeEntitiesType#

Action to stop a node: deactivate

Parameters:

node_name (str) – description

Returns:

Launch actions

Return type:

List[SomeEntitiesType]

restart(node_name: str, **_) → ros_sugar.utils.SomeEntitiesType#

Action to restart a node: deactivate + activate

Parameters:

node_name (str) – description

Returns:

Launch actions

Return type:

List[SomeEntitiesType]

property fallback_rate: Dict#

fallback_rate.

Return type:

Dict

on_process_fail(max_retries: int = 3) → None#

Enable process-level crash recovery for all multi-process components.

When a component process exits unexpectedly (non-zero return code, not during launcher shutdown, and not via user signal), the launcher will respawn it up to max_retries times. After the limit is reached, the component is left down and a terminal error is logged.

Parameters:

max_retries (int) – Maximum number of respawn attempts per component. Must be a positive integer. Defaults to 3.

Raises:

ValueError – if max_retries is not a positive integer.

configure(config_file: str, component_name: str | None = None)#

Configure components managed by the Launcher

Parameters:

  • config_file (str) – Path to configuration file (yaml, json ot toml)

  • component_name (str | None, optional) – Configure one component with given name, defaults to None

add_py_executable(path_to_executable: str, name: str = 'python3')#

Adds a python executable to the launcher as a separate process

Parameters:

  • path_to_executable (str) – description

  • name (str, optional) – description, defaults to ‘python3’

add_ros_node(package: str, executable: str, name: Optional[str] = None, parameters: Optional[List] = None, remappings: Optional[List[Tuple[str, str]]] = None, arguments: Optional[List[str]] = None, output: str = 'screen', **launch_node_kwargs) → launch_ros.actions.Node#

Adds an external ROS2 node to the launcher, to be launched alongside the components (e.g. a MoveIt move_group node, a camera driver).

The node always runs in its own process and inherits the Launcher namespace (if one is set). The monitor does not track this node, so to restart the node automatically if it dies, pass the launch_ros keyword arguments respawn=True and optionally respawn_delay=<seconds>.

Parameters:

  • package (str) – Name of the ROS2 package containing the node executable

  • executable (str) – Name of the node executable

  • name (Optional[str]) – Node name, defaults to the executable’s default name

  • parameters (Optional[List]) – Node parameters (list of dicts and/or yaml file paths)

  • remappings (Optional[List[Tuple[str, str]]]) – Topic/service remapping pairs

  • arguments (Optional[List[str]]) – Extra command line arguments for the node

  • output (str) – Output configuration, defaults to ‘screen’

  • launch_node_kwargs – Additional keyword arguments for launch_ros Node

Raises:

  • PackageNotFoundError – If the package is not installed

  • FileNotFoundError – If the package has no such executable

Returns:

The created launch action

Return type:

launch_ros.actions.Node

include_launch_file(package: Optional[str], launch_file: str, launch_args: Optional[Dict[str, Any]] = None) → launch.actions.IncludeLaunchDescription#

Includes an external launch file in the launcher, to be brought up alongside the components (e.g. a robot’s MoveIt config demo launch).

Python, XML and YAML launch files are supported. The included description inherits the Launcher namespace (if one is set).

Parameters:

  • package (Optional[str]) – Name of the ROS2 package containing the launch file. The file is resolved against the package share directory (directly or under ‘launch/’). If None, launch_file is used as a filesystem path

  • launch_file (str) – Launch file name (or path when package is None)

  • launch_args (Optional[Dict[str, Any]]) – Launch arguments passed to the included launch file

Raises:

FileNotFoundError – If the launch file cannot be resolved

Returns:

The created launch action

Return type:

launch.actions.IncludeLaunchDescription

add_method(method: Callable | Awaitable, args: Iterable | None = None, kwargs: Dict | None = None)#

Adds a method action to launch

Parameters:

  • method (Callable | Awaitable) – description

  • args (Iterable | None, optional) – description, defaults to None

  • kwargs (Dict | None, optional) – description, defaults to None

bringup(config_file: str | None = None, introspect: bool = False, launch_debug: bool = False)#

Bring up the Launcher