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.

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_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