ros_sugar.base_clients#

ROS Service/Action Client Wrapper

Module Contents#

Classes#

ServiceClientConfig

Basic configuration for any ROS service client

ActionClientConfig

Basic configuration for any ROS action client

ServiceClientHandler

General purpose service client class

ActionClientHandler

General purpose action client class

API#

class ros_sugar.base_clients.ServiceClientConfig#

Bases: ros_sugar.config.BaseAttrs

Basic configuration for any ROS service client

asdict(filter: Optional[Callable] = None) → Dict#
to_dict() → Dict#
asdict_explicit() → Dict#
from_dict(dict_obj: Dict) → None#
from_file(file_path: str, nested_root_name: Union[str, None] = None, get_common: bool = False) → bool#
to_json() → Union[str, bytes, bytearray]#
from_json(json_obj: Union[str, bytes, bytearray]) → None#
has_attribute(attr_name: str) → bool#
get_attribute_type(attr_name: str) → Optional[type]#
update_value(attr_name: str, attr_value: Any) → bool#
classmethod get_fields_info(class_object) → Dict[str, Dict[str, Any]]#
class ros_sugar.base_clients.ActionClientConfig#

Bases: ros_sugar.config.BaseAttrs

Basic configuration for any ROS action client

asdict(filter: Optional[Callable] = None) → Dict#
to_dict() → Dict#
asdict_explicit() → Dict#
from_dict(dict_obj: Dict) → None#
from_file(file_path: str, nested_root_name: Union[str, None] = None, get_common: bool = False) → bool#
to_json() → Union[str, bytes, bytearray]#
from_json(json_obj: Union[str, bytes, bytearray]) → None#
has_attribute(attr_name: str) → bool#
get_attribute_type(attr_name: str) → Optional[type]#
update_value(attr_name: str, attr_value: Any) → bool#
classmethod get_fields_info(class_object) → Dict[str, Dict[str, Any]]#
class ros_sugar.base_clients.ServiceClientHandler(client_node: rclpy.node.Node, config: Optional[ros_sugar.base_clients.ServiceClientConfig] = None, srv_name: Optional[str] = None, srv_type: Optional[type] = None, callback_group: Optional[rclpy.callback_groups.CallbackGroup] = None)#

General purpose service client class

send_request_from_dict(request_fields: Dict[str, Any])#

Send a service request using a serialized Dict request data

Parameters:

request_fields (Dict[str, Any]) – Request data [key, value]

Raises:

ValueError – If a field cannot be set from its value, naming both

Returns:

Service result

Return type:

Any

send_request(req_msg, timeout: Optional[float] = None)#

Sends a request to the service returns the response In case of failure, the method attempts sending the request again multiple time according to the given config

Parameters:

  • req_msg (Any) – Service request msg

  • timeout (Optional[float]) – Seconds to wait for the response, instead of the configured timeout. math.inf waits for as long as the node lives, for a caller that has a deadline of its own and whose call may legitimately take longer than any fixed budget

Returns:

Service result

Return type:

Any

class ros_sugar.base_clients.ActionClientHandler(client_node: rclpy.node.Node, config: Optional[ros_sugar.base_clients.ActionClientConfig] = None, action_name: Optional[str] = None, action_type: Optional[type] = None)#

General purpose action client class

reset()#

Reset the client handler

send_request_from_dict(request_fields: Dict[str, Any], wait_until_first_feedback: bool = False, still_wanted: Optional[Callable[[], bool]] = None) → Optional[bool]#

Send an action request using a serialized Dict request data

Parameters:

  • request_fields (Dict[str, Any]) – Request data [key, value]

  • still_wanted – See send_request

Raises:

ValueError – If a field cannot be set from its value, naming both. Raised before anything is sent, so a running goal is left alone

wait_until_idle(timeout: float) → bool#

Wait for the goal this client has on the server to end.

For a caller tearing things down: the cancels it sent are on their way, and a server told to stop is worth waiting for before its node goes.

Parameters:

timeout – Seconds to wait

Returns:

Whether the client ended up with no goal in flight

Return type:

bool

send_request(request_msg: Any, wait_until_first_feedback: bool = False, still_wanted: Optional[Callable[[], bool]] = None) → bool#

Sends a request to an action server

Parameters:

  • request_msg (Action_Type.Goal) – Action request message

  • wait_until_first_feedback (bool, optional) – Wait until the server returns its first feedback, defaults to True

  • still_wanted (Callable[[], bool], optional) – Asked while the request waits for the client and for the server, and once more just before it is sent. Once it says no, nothing is sent: a caller that stopped while it waited would otherwise start a goal that nothing is watching any more

Returns:

If action server is available

Return type:

bool

action_response_callback(future)#

Callback when getting the action server responses

Parameters:

future (Any) – Action result future

action_result_callback(future)#

Treats the path tracker action result

Parameters:

future (Any) – Action result future

add_feedback_listener(listener: Callable[[], None]) → None#

Register a zero-arg callback fired on every feedback message and on terminal state (in the ROS executor thread).

remove_feedback_listener(listener: Callable[[], None]) → None#

Remove a previously registered feedback listener.

action_feedback_callback(feedback_msg: Any)#

Handles feedback messages received during action execution.

Parameters:

feedback_msg (Any) – Action feedback message

got_new_feedback() → bool#

Checks if the client got a new feedback from the server within a specified time limit

Returns:

Feedback updated on time

Return type:

bool

cancel_request(wait: bool = True) → Tuple[bool, str]#

Cancel an active action goal and return result

Parameters:

wait (bool) – Wait for the goal to return before reporting. Without it the cancel request is only sent, which is all that can be done once nothing spins to deliver the server’s answer, e.g. at shutdown

Returns:

If cancellation is successful

Return type:

Tuple[bool, str]

get_ui_elements() → Dict#

Get updated client elements for the UI

Returns:

description

Return type:

Dict