Components API¶

class lifecore_ros2.components.topic_component.TopicComponent(*args, **kwargs)¶

Bases: LifecycleComponent, ABC, Generic

Intermediate base class for lifecycle-aware topic publisher and subscriber components.

Owns:

The topic name, message type, and QoS profile shared by publisher and subscriber subclasses.

Does not own:

The ROS publisher or subscription objects — those belong to concrete subclasses.
Any lifecycle hook logic — concrete subclasses provide _on_configure, _on_cleanup, and _release_resources implementations.
The callback group — it is borrowed from the application and forwarded to the core.

Override points:

Not intended to be subclassed directly outside the framework.
Subclass LifecyclePublisherComponent or LifecycleSubscriberComponent instead.

Parameters:

name (str)
topic_name (str)
msg_type (type[MsgT] | None)
qos_profile (QoSProfile | int)
callback_group (CallbackGroup | None)
dependencies (Sequence[str])
priority (int)

property topic_name: str¶

property msg_type: type[MsgT]¶

property qos_profile: QoSProfile | int¶

class lifecore_ros2.components.lifecycle_publisher_component.LifecyclePublisherComponent(*args, **kwargs)¶

Bases: TopicComponent, Generic

Publisher component that creates and gates a ROS publisher through the lifecycle.

Owns:

The ROS Publisher instance (created on configure, kept across deactivate, released automatically during cleanup, shutdown, or error).
publish: the activation-gated publication method.

Does not own:

The topic name, message type, or QoS profile (inherited from TopicComponent).
The callback group — it is borrowed from the application; lifetime is owned by the caller.
The node or lifecycle state transitions.
Activation state management (handled by the framework).

Override points:

This class is usable directly without subclassing.
Override _on_configure for additional setup; call super()._on_configure(state) first.
Do not override publish.

Parameters:

name (str)
topic_name (str)
msg_type (type[MsgT] | None)
qos_profile (QoSProfile | int)
callback_group (CallbackGroup | None)
dependencies (Sequence[str])
priority (int)

_on_configure(state)¶

Extension point. Calls super if overridden; creates the ROS publisher.

Override in subclasses for additional setup. Call super()._on_configure(state) first.

Parameters:: state (rclpy.lifecycle.node.LifecycleState)
Return type:: rclpy.lifecycle.node.TransitionCallbackReturn

publish(msg)¶

Publish a message. Raises RuntimeError if not active.

Parameters:: msg (MsgT)
Return type:: None

_release_resources()¶

Extension point. Override to release additional resources; call super()._release_resources() last.

Return type:: None

class lifecore_ros2.components.lifecycle_subscriber_component.LifecycleSubscriberComponent(*args, **kwargs)¶

Bases: TopicComponent, Generic

Subscriber component that creates a ROS subscription and gates message delivery through the lifecycle.

The subscription is created on configure, kept across deactivate, and destroyed automatically during cleanup, shutdown, or error. Incoming messages are silently dropped while the component is inactive, and routed to on_message when active.

Owns:

The ROS Subscription instance (created on configure, kept across deactivate, released automatically during cleanup, shutdown, or error).
_on_message_wrapper: the @when_active-gated internal callback.

Does not own:

The topic name, message type, or QoS profile (inherited from TopicComponent).
The callback group — it is borrowed from the application; lifetime is owned by the caller.
The node or lifecycle state transitions.
Activation state management (handled by the framework).

Override points:

on_message: implement to handle incoming messages. Called only while active.
Override _on_configure only for additional setup; call super()._on_configure(state) first.
Do not override _on_message_wrapper.

Parameters:

name (str)
topic_name (str)
msg_type (type[MsgT] | None)
qos_profile (QoSProfile | int)
callback_group (CallbackGroup | None)
dependencies (Sequence[str])
priority (int)

_on_configure(state)¶

Extension point. Creates the ROS subscription.

Override in subclasses for additional setup. Call super()._on_configure(state) first.

Parameters:: state (rclpy.lifecycle.node.LifecycleState)
Return type:: rclpy.lifecycle.node.TransitionCallbackReturn

abstractmethod on_message(msg)¶

Extension point. Implement to handle incoming messages while active.

This is the subscriber callback contract. Unlike the _on_* lifecycle hooks, on_message is intentionally public because it defines application behavior, not framework behavior. It is only called while the component is active.

Parameters:: msg (MsgT)
Return type:: None

_release_resources()¶

Extension point. Override to release additional resources; call super()._release_resources() last.

Return type:: None

class lifecore_ros2.components.lifecycle_timer_component.LifecycleTimerComponent(*args, **kwargs)¶

Bases: LifecycleComponent

Timer component that creates a periodic ROS timer and gates ticks through the lifecycle.

The timer is created on configure, kept across deactivate, and destroyed automatically during cleanup, shutdown, or error. Ticks are silently dropped while the component is inactive, and routed to on_tick when active.

Owns:

The ROS Timer instance (created on configure, kept across deactivate, released automatically during cleanup, shutdown, or error).
_on_timer_wrapper: the @when_active-gated internal callback.
The autostart flag and the explicit start / stop / reset controls.

Does not own:

The clock — uses the node default clock.
The callback group — it is borrowed from the application; lifetime is owned by the caller.
The node or lifecycle state transitions.
Activation state management (handled by the framework).

Override points:

on_tick: implement to handle each timer tick. Called only while active.
Override _on_configure only for additional setup; call super()._on_configure(state) first.
Do not override _on_timer_wrapper.

Parameters:

name (str)
period (float | Duration)
autostart (bool)
callback_group (CallbackGroup | None)
dependencies (Sequence[str])
priority (int)

property period_sec: float¶: The timer period in seconds.

property autostart: bool¶: Whether the timer starts firing automatically when created on configure.

property is_running: bool¶: True iff the underlying timer exists and is currently firing (not canceled).

_on_configure(state)¶

Extension point. Creates the ROS timer.

Override in subclasses for additional setup. Call super()._on_configure(state) first.

Parameters:: state (rclpy.lifecycle.node.LifecycleState)
Return type:: rclpy.lifecycle.node.TransitionCallbackReturn

abstractmethod on_tick()¶

Extension point. Implement to handle each timer tick while active.

This is the timer callback contract. Unlike the _on_* lifecycle hooks, on_tick is intentionally public because it defines application behavior, not framework behavior. It is only called while the component is active.

Return type:: None

start()¶

Resume firing the underlying timer. No-op if already running.

Raises:: ComponentNotConfiguredError – If called before configure or after cleanup.
Return type:: None

stop()¶

Stop the underlying timer. No-op if already stopped.

Raises:: ComponentNotConfiguredError – If called before configure or after cleanup.
Return type:: None

reset()¶

Restart the timer period from now. Resumes firing if currently stopped.

Raises:: ComponentNotConfiguredError – If called before configure or after cleanup.
Return type:: None

_release_resources()¶

Extension point. Override to release additional resources; call super()._release_resources() last.

Return type:: None

class lifecore_ros2.components.service_component.ServiceComponent(*args, **kwargs)¶

Bases: LifecycleComponent, ABC, Generic

Intermediate base class for lifecycle-aware service server and client components.

Owns:

The service name, service type, and QoS profile shared by server and client subclasses.

Does not own:

The ROS Service or Client objects — those belong to concrete subclasses.
Any lifecycle hook logic — concrete subclasses provide _on_configure, _on_cleanup, and _release_resources implementations.
The callback group — it is borrowed from the application and forwarded to the core.

Override points:

Not intended to be subclassed directly outside the framework.
Subclass LifecycleServiceServerComponent or LifecycleServiceClientComponent instead.

Parameters:

name (str)
service_name (str)
srv_type (type[SrvT] | None)
qos_profile (QoSProfile | None)
callback_group (CallbackGroup | None)
dependencies (Sequence[str])
priority (int)

property service_name: str¶

property srv_type: type[SrvT]¶

property qos_profile: QoSProfile | None¶

class lifecore_ros2.components.lifecycle_service_server_component.LifecycleServiceServerComponent(*args, **kwargs)¶

Bases: ServiceComponent, Generic

Service server component that creates a ROS service and gates request handling through the lifecycle.

The service is created on configure, kept across deactivate, and destroyed automatically during cleanup, shutdown, or error. Incoming requests while the component is inactive are not silently dropped: a warning is logged and a default-constructed response is returned, optionally annotated with diagnostic fields (success=False, message="component inactive").

Owns:

The ROS Service instance (created on configure, kept across deactivate, released automatically during cleanup, shutdown, or error).
_on_request_wrapper: the framework-internal callback registered with rclpy.

Does not own:

The service name, service type, or QoS profile (inherited from ServiceComponent).
The callback group — it is borrowed from the application; lifetime is owned by the caller.
The node or lifecycle state transitions.
Activation state management (handled by the framework).

Override points:

on_service_request: implement to handle incoming requests while active.
Override _on_configure only for additional setup; call super()._on_configure(state) first.
Do not override _on_request_wrapper.

Parameters:

name (str)
service_name (str)
srv_type (type[SrvT] | None)
qos_profile (QoSProfile | None)
callback_group (CallbackGroup | None)
dependencies (Sequence[str])
priority (int)

_on_configure(state)¶

Extension point. Calls super if overridden; creates the ROS service.

Override in subclasses for additional setup. Call super()._on_configure(state) first.

Parameters:: state (rclpy.lifecycle.node.LifecycleState)
Return type:: rclpy.lifecycle.node.TransitionCallbackReturn

abstractmethod on_service_request(request, response)¶

Extension point. Implement to handle incoming service requests while active.

Follows the rclpy service callback convention: fill in response fields and return it. Called only while the component is active.

Parameters:

request (Any) – The incoming service request object.
response (Any) – A default-constructed response object to fill in.

Returns:

The filled-in response object.

Return type:

Any

_release_resources()¶

Extension point. Override to release additional resources; call super()._release_resources() last.

Return type:: None

class lifecore_ros2.components.lifecycle_service_client_component.LifecycleServiceClientComponent(*args, **kwargs)¶

Bases: ServiceComponent, Generic

Service client component that creates a ROS client and gates calls through the lifecycle.

The client is created on configure, kept across deactivate, and destroyed automatically during cleanup, shutdown, or error. Outbound calls (call, call_async, wait_for_service) raise RuntimeError while the component is inactive.

Note

Futures from call_async are not cancelled on deactivate; the application owns them.

Owns:

The ROS Client instance (created on configure, kept across deactivate, released automatically during cleanup, shutdown, or error).

Does not own:

The service name, service type, or QoS profile (inherited from ServiceComponent).
The callback group — it is borrowed from the application; lifetime is owned by the caller.
The node or lifecycle state transitions.
Activation state management (handled by the framework).

Override points:

This class is usable directly without subclassing.
Override _on_configure for additional setup; call super()._on_configure(state) first.
Do not override call, call_async, or wait_for_service.

Parameters:

name (str)
service_name (str)
srv_type (type[SrvT] | None)
qos_profile (QoSProfile | None)
callback_group (CallbackGroup | None)
dependencies (Sequence[str])
priority (int)

_on_configure(state)¶

Extension point. Calls super if overridden; creates the ROS client.

Override in subclasses for additional setup. Call super()._on_configure(state) first.

Parameters:: state (rclpy.lifecycle.node.LifecycleState)
Return type:: rclpy.lifecycle.node.TransitionCallbackReturn

call(request, timeout_service=None, timeout_call=None)¶

Call the service synchronously. Raises RuntimeError if not active.

Parameters:

request (Any) – The service request object.
timeout_service (float | None) – Optional timeout in seconds to wait for the service to become available before sending the request. Raises TimeoutError if the service does not become available in time.
timeout_call (float | None) – Optional timeout in seconds for the synchronous call itself. None waits indefinitely for the response.

Returns:

The service response object.

Raises:

RuntimeError – if the component is not active.
ComponentNotConfiguredError – if the component has not been configured.
TimeoutError – if timeout_service is specified and the service is not available.

Return type:

Any

call_async(request, timeout_service=None)¶

Call the service asynchronously. Raises RuntimeError if not active.

Note

Futures are not cancelled on deactivate; the application owns them.

Parameters:

request (Any) – The service request object.
timeout_service (float | None) – Optional timeout in seconds to wait for the service to become available before submitting the request. Raises TimeoutError if the service does not become available in time.

Returns:

A Future that resolves to the service response.

Raises:

RuntimeError – if the component is not active.
ComponentNotConfiguredError – if the component has not been configured.
TimeoutError – if timeout_service is specified and the service is not available.

Return type:

rclpy.task.Future

wait_for_service(timeout=None)¶

Wait for the service to become available. Raises RuntimeError if not active.

Parameters:

timeout (float | None) – Optional timeout in seconds. None waits indefinitely.

Returns:

True if the service is available; False if the timeout elapsed.

Raises:

RuntimeError – if the component is not active.
ComponentNotConfiguredError – if the component has not been configured.

Return type:

bool

_release_resources()¶

Extension point. Override to release additional resources; call super()._release_resources() last.

Return type:: None