Create Example

This skill guides you through creating new publish/subscribe examples for the zenoh-ros2-sdk.

When to Use

• User wants to add a new example for a message type
• User wants to create publisher/subscriber for a new ROS2 message
• Adding support for trajectory, sensor, geometry, or other message types

Project Structure

Examples are in examples/ with numbered filenames:

• Odd numbers: publishers (e.g., 11_publish_*.py)
• Even numbers: subscribers (e.g., 12_subscribe_*.py)

1. Determine Next Example Numbers

Check existing examples in examples/ directory. Use the next available odd number for publisher, next even for subscriber.

2. Look Up Message Definition

Find the ROS2 message definition. Key resources:

• https://docs.ros2.org/foxy/api/{package}/msg/{MessageName}.html
• Common packages: std_msgs, geometry_msgs, sensor_msgs, trajectory_msgs, nav_msgs

3. Publisher Template

Create examples/{NN}_publish_{msg_name}.py:

python
1#!/usr/bin/env python3
2"""
3{NN} - Publish {MessageName} Messages
4
5Demonstrates how to publish {package}/msg/{MessageName} messages to a ROS2 topic.
6{Brief description of what this message type is used for.}
7"""
8import time
9import numpy as np
10
11from zenoh_ros2_sdk import ROS2Publisher, get_message_class
12
13
14def main():
15    print("{NN} - Publish {MessageName} Messages")
16    print("Publishing to /{topic_name} topic...\n")
17
18    # Get message classes for easy object creation
19    Header = get_message_class("std_msgs/msg/Header")
20    Time = get_message_class("builtin_interfaces/msg/Time")
21    {MessageName} = get_message_class("{package}/msg/{MessageName}")
22    # Add nested message classes as needed
23
24    if not all([Header, Time, {MessageName}]):
25        print("Error: Failed to get message classes")
26        return
27
28    # Create publisher
29    pub = ROS2Publisher(
30        topic="/{topic_name}",
31        msg_type="{package}/msg/{MessageName}"
32    )
33
34    try:
35        print("Publishing {MessageName} messages...")
36        print("Press Ctrl+C to stop\n")
37
38        counter = 0
39        while True:
40            # Get current time
41            now = time.time()
42            sec = int(now)
43            nanosec = int((now - sec) * 1e9)
44
45            # Create header with timestamp
46            header = Header(
47                stamp=Time(sec=sec, nanosec=nanosec),
48                frame_id="base_link"
49            )
50
51            # Create and populate message fields
52            # Note: Arrays need to be numpy arrays for proper serialization
53
54            # Publish message - pass fields as kwargs matching message structure
55            pub.publish(
56                header=header,
57                # ... other fields
58            )
59
60            print(f"Published {MessageName} {counter}")
61            counter += 1
62            time.sleep(0.1)  # 10 Hz
63
64    except KeyboardInterrupt:
65        print("\nInterrupted by user")
66    finally:
67        pub.close()
68        print("Publisher closed")
69
70
71if __name__ == "__main__":
72    main()

4. Subscriber Template

Create examples/{NN+1}_subscribe_{msg_name}.py:

python
1#!/usr/bin/env python3
2"""
3{NN+1} - Subscribe to {MessageName} Messages
4
5Demonstrates how to subscribe to a ROS2 topic and receive {package}/msg/{MessageName} messages.
6{Brief description of what this message type is used for.}
7"""
8import time
9
10from zenoh_ros2_sdk import ROS2Subscriber
11
12
13def main():
14    print("{NN+1} - Subscribe to {MessageName} Messages")
15    print("Subscribing to /{topic_name} topic...\n")
16
17    message_count = [0]  # Use list to allow modification in nested function
18
19    def on_message(msg):
20        """Callback function called when a {MessageName} message is received."""
21        message_count[0] += 1
22
23        # Extract information from the message
24        timestamp = msg.header.stamp
25        frame_id = msg.header.frame_id
26        # ... extract other fields
27
28        # Display received data
29        print(f"\n--- {MessageName} Message #{message_count[0]} ---")
30        print(f"Timestamp: {timestamp.sec}.{timestamp.nanosec:09d}")
31        print(f"Frame ID: {frame_id}")
32        # ... print other fields
33
34    # Create subscriber
35    sub = ROS2Subscriber(
36        topic="/{topic_name}",
37        msg_type="{package}/msg/{MessageName}",
38        callback=on_message
39    )
40
41    try:
42        print("Waiting for {MessageName} messages... Press Ctrl+C to stop")
43        while True:
44            time.sleep(0.1)
45    except KeyboardInterrupt:
46        print("\nInterrupted by user")
47    finally:
48        sub.close()
49        print("Subscriber closed")
50
51
52if __name__ == "__main__":
53    main()

5. Update README

Add entries to examples/README.md:

markdown
1### [`{NN}_publish_{msg_name}.py`]({NN}_publish_{msg_name}.py)
2Demonstrates how to publish `{package}/msg/{MessageName}` messages. {Description}.
3
4**Usage:**
5\```bash
6python3 examples/{NN}_publish_{msg_name}.py
7\```
8
9### [`{NN+1}_subscribe_{msg_name}.py`]({NN+1}_subscribe_{msg_name}.py)
10Demonstrates how to subscribe to `{package}/msg/{MessageName}` messages. {Description}.
11
12**Usage:**
13\```bash
14python3 examples/{NN+1}_subscribe_{msg_name}.py
15\```

6. Key Patterns

Arrays

Use numpy arrays for proper CDR serialization:

python
1positions = np.array([0.0, 0.1, 0.2], dtype=np.float64)

Nested Messages

Get nested message classes and construct them:

python
1Point = get_message_class("trajectory_msgs/msg/JointTrajectoryPoint")
2Duration = get_message_class("builtin_interfaces/msg/Duration")
3
4point = Point(
5    positions=np.array([0.0, 0.1], dtype=np.float64),
6    time_from_start=Duration(sec=1, nanosec=0)
7)

Common Topic Names

• /joint_states - JointState
• /cmd_vel - Twist
• /joint_trajectory - JointTrajectory
• /odom - Odometry
• /tf - TransformStamped