Skip to content

compas_eve.ros ¤

Classes¤

RosTransport ¤

RosTransport(
    host: str = "localhost",
    port: int = 9090,
    connect_timeout: float = 10,
    **client_options: Any
)

ROS transport backed by a rosbridge WebSocket connection.

Messages use native ROS message types and are represented as JSON-compatible dictionaries, matching the behavior of roslibpy.

Parameters:

  • host (str, default: 'localhost' ) –

    Host name of the rosbridge server.

  • port (int, default: 9090 ) –

    WebSocket port of the rosbridge server. Defaults to 9090.

  • connect_timeout (float, default: 10 ) –

    Maximum number of seconds to wait for the initial connection.

  • **client_options (Any, default: {} ) –

    Additional options passed to roslibpy.Ros.

Notes

The topic's message_type must be a ROS message type string, such as "std_msgs/String". ROS topic options can be passed to Topic: compression, latch, throttle_rate, queue_size, queue_length, and reconnect_on_close.

Methods:¤

advertise ¤
advertise(topic: Topic) -> str

Advertise a native ROS topic.

close ¤
close() -> None

Close all topics and terminate the rosbridge client.

on_ready ¤
on_ready(callback: Callable) -> None

Invoke a callback when the rosbridge connection is ready.

publish ¤
publish(topic: Topic, message: Message | dict, **options: Any) -> None

Publish a native ROS message dictionary to a topic.

subscribe ¤
subscribe(topic: Topic, callback: Callable) -> str

Subscribe to a native ROS topic.

unadvertise ¤
unadvertise(topic: Topic) -> None

Stop advertising a native ROS topic.

unsubscribe ¤
unsubscribe(topic: Topic) -> None

Unsubscribe all local callbacks from a topic.

unsubscribe_by_id ¤
unsubscribe_by_id(subscribe_id: str) -> None

Unsubscribe one callback using its subscription identifier.