Add optional ROS transport

.github/workflows/build.yml

@@ -49,6 +49,7 @@ jobs:
         run: |
           docker run -d --name nanomq -p 1883:1883 -p 8083:8083 -p 8883:8883 emqx/nanomq:latest
           docker run -d --name zenoh --init -p 7447:7447/tcp -p 8000:8000/tcp eclipse/zenoh
+          docker run -d --name rosbridge -p 9090:9090 husarion/rosbridge-server:humble ros2 launch rosbridge_server rosbridge_websocket_launch.xml
           docker ps -a
 
       - uses: compas-dev/compas-actions/ci@v1
@@ -65,7 +66,7 @@ jobs:
 
       - name: Stop test services
         if: always()
-        run: docker rm -f nanomq zenoh || true
+        run: docker rm -f nanomq zenoh rosbridge || true
 
   components:
     needs: release-change

.github/workflows/release.yml

@@ -48,6 +48,7 @@ jobs:
         run: |
           docker run -d --name nanomq -p 1883:1883 -p 8083:8083 -p 8883:8883 emqx/nanomq:latest
           docker run -d --name zenoh --init -p 7447:7447/tcp -p 8000:8000/tcp eclipse/zenoh
+          docker run -d --name rosbridge -p 9090:9090 husarion/rosbridge-server:humble ros2 launch rosbridge_server rosbridge_websocket_launch.xml
           docker ps -a
 
       - uses: compas-dev/compas-actions/ci@v1
@@ -64,7 +65,7 @@ jobs:
 
       - name: Stop test services
         if: always()
-        run: docker rm -f nanomq zenoh || true
+        run: docker rm -f nanomq zenoh rosbridge || true
 
   components:
     needs: release

CHANGELOG.md

@@ -9,6 +9,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+* Added an optional ROS transport backed by `roslibpy` and rosbridge.
+
 ### Changed
 
 * Migrated CI, documentation, Grasshopper component generation, and releases to `compas-actions@v1` with trusted publishing and release pull requests.

README.md

@@ -44,6 +44,7 @@ Or using `conda`:
 * Publisher/subscriber communication model (N-to-N communication)
 * In-process events
 * MQTT support
+* ROS support through rosbridge
 * Zenoh support
 * Extensible codec system for message serialization (JSON, Protocol Buffers)
 
@@ -88,6 +89,34 @@ for i in range(10):
 This example shows how to send and receive from a single script, but
 running publishers and subscribers on different scripts, different processes, or even different computers will work the exact same way.
 
+### ROS
+
+The optional ROS backend uses native ROS message types through a rosbridge server:
+
+```bash
+pip install compas_eve[ros]
+```
+
+```python
+import time
+
+import compas_eve as eve
+from compas_eve.ros import RosTransport
+
+tx = RosTransport("localhost", 9090)
+topic = eve.Topic("/chatter", "std_msgs/String", queue_size=10)
+
+sub = eve.EchoSubscriber(topic, transport=tx)
+sub.subscribe()
+eve.Publisher(topic, transport=tx).publish({"data": "Hello ROS"})
+time.sleep(1)
+tx.close()
+```
+
+ROS messages are passed as JSON-compatible dictionaries. Topic options supported
+by `roslibpy`—including `compression`, `latch`, `throttle_rate`, `queue_size`,
+`queue_length`, and `reconnect_on_close`—can be set on `eve.Topic`.
+
 ### Zenoh
 
 Apache Zenoh is a pub/sub/query protocol. In many ways, it is similar to MQTT but with some additional features and optimizations. COMPAS EVE also supports Zenoh as a transport protocol with an identical API to MQTT:

docs/api/compas_eve.ros.md

@@ -0,0 +1 @@
+# ::: compas_eve.ros

docs/examples.md

@@ -119,3 +119,26 @@ Next, we create the matching subscriber:
 ```python
 --8<-- "docs/examples/05_zenoh_distributed_world_sub.py"
 ```
+
+## Communication with ROS
+
+The ROS transport connects to a rosbridge server and works with native ROS
+message types. ROS messages are represented as JSON-compatible dictionaries.
+
+```python
+--8<-- "docs/examples/06_ros_pubsub.py"
+```
+
+The message dictionary is mapped directly to the fields of the declared ROS
+message type. It is not serialized into a string field. For example, the next
+example publishes a native `sensor_msgs/JointState` containing a nested ROS
+header and arrays for three joints. The subscriber receives the same structured
+fields as a dictionary:
+
+```python
+--8<-- "docs/examples/07_ros_joint_state.py"
+```
+
+Because rosbridge handles this as `sensor_msgs/JointState`, regular ROS nodes can
+subscribe to `/compas_eve/joint_states` and access `name`, `position`,
+`velocity`, and `effort` as normal ROS message fields.

docs/examples/06_ros_pubsub.py

@@ -0,0 +1,18 @@
+import time
+
+from compas_eve import EchoSubscriber
+from compas_eve import Publisher
+from compas_eve import Topic
+from compas_eve.ros import RosTransport
+
+transport = RosTransport("localhost", 9090)
+topic = Topic("/chatter", "std_msgs/String", queue_size=10)
+
+subscriber = EchoSubscriber(topic, transport=transport)
+subscriber.subscribe()
+
+publisher = Publisher(topic, transport=transport)
+publisher.publish({"data": "Hello ROS"})
+
+time.sleep(1)
+transport.close()

docs/examples/07_ros_joint_state.py

@@ -0,0 +1,44 @@
+import time
+from threading import Event
+
+from compas_eve import Publisher
+from compas_eve import Subscriber
+from compas_eve import Topic
+from compas_eve.ros import RosTransport
+
+transport = RosTransport("localhost", 9090)
+topic = Topic("/compas_eve/joint_states", "sensor_msgs/JointState", queue_size=10)
+message_received = Event()
+
+
+def print_joint_state(message):
+    print("Frame:", message["header"]["frame_id"])
+    for name, position, velocity in zip(message["name"], message["position"], message["velocity"]):
+        print("{}: position={}, velocity={}".format(name, position, velocity))
+    message_received.set()
+
+
+subscriber = Subscriber(topic, print_joint_state, transport=transport)
+subscriber.subscribe()
+time.sleep(0.5)  # Allow rosbridge to register the subscription.
+
+publisher = Publisher(topic, transport=transport)
+publisher.publish(
+    {
+        "header": {
+            "stamp": {"sec": 0, "nanosec": 0},
+            "frame_id": "robot_base",
+        },
+        "name": ["shoulder_joint", "elbow_joint", "wrist_joint"],
+        "position": [0.25, -0.5, 1.2],
+        "velocity": [0.1, 0.0, -0.1],
+        "effort": [4.2, 2.8, 0.7],
+    }
+)
+
+if not message_received.wait(timeout=5):
+    raise RuntimeError("No joint state received")
+
+subscriber.unsubscribe()
+publisher.unadvertise()
+transport.close()

docs/installation.md

@@ -95,3 +95,14 @@ uv pip install compas_eve[zenoh]
 ```
 
 For more details about Zenoh, refer to the [Eclipse Zenoh](https://zenoh.io/) website.
+
+### ROS Transport
+
+The ROS transport requires `roslibpy` 2.x and a running rosbridge server. Install
+the optional dependency with:
+
+```bash
+uv pip install compas_eve[ros]
+```
+
+The transport connects to rosbridge over WebSockets on port `9090` by default.

mkdocs.yml

@@ -140,6 +140,7 @@ nav:
       - compas_eve.codecs: api/compas_eve.codecs.md
       - compas_eve.memory: api/compas_eve.memory.md
       - compas_eve.mqtt: api/compas_eve.mqtt.md
+      - compas_eve.ros: api/compas_eve.ros.md
       - compas_eve.zenoh: api/compas_eve.zenoh.md
       - compas_eve.ghpython: api/compas_eve.ghpython.md
-  - License: license.md
+  - License: license.md

pyproject.toml

@@ -37,7 +37,7 @@ classifiers = [
 [tool.setuptools.dynamic]
 version = { attr = "compas_eve.__version__" }
 dependencies = { file = "requirements.txt" }
-optional-dependencies = { dev = { file = "requirements-dev.txt" }, zenoh = { file = "requirements-zenoh.txt" } }
+optional-dependencies = { dev = { file = "requirements-dev.txt" }, ros = { file = "requirements-ros.txt" }, zenoh = { file = "requirements-zenoh.txt" } }
 
 [project.entry-points.'compas_pb.plugins']
 serializers = 'compas_eve.codecs.conversions'

requirements-ros.txt

@@ -0,0 +1 @@
+roslibpy >=2, <3

src/compas_eve/core.py

@@ -125,16 +125,16 @@ class Topic(object):
     name
         Name of the topic.
     message_type
-        Class defining the message structure. Use [Message][] for
-        a generic, non-typed checked message implementation.
-        Defaults to [Message][].
+        Class defining the message structure, or a backend-native message type
+        identifier. Use [Message][] for a generic, non-type-checked message
+        implementation. Defaults to [Message][].
     options
         A dictionary of options.
     """
 
     # TODO: Add documentation/examples of possible options
 
-    def __init__(self, name: str, message_type: Optional[Type[Message]] = None, **options: Any) -> None:
+    def __init__(self, name: str, message_type: Optional[Union[Type[Message], str]] = None, **options: Any) -> None:
         self.name = name
         self.message_type = message_type or Message
         self.options = options

src/compas_eve/ros/__init__.py

@@ -0,0 +1,3 @@
+from .ros_transport import RosTransport
+
+__all__ = ["RosTransport"]