hiroz 0.2.0

Native Rust ROS 2 implementation using Zenoh
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94

#!/usr/bin/env python3
"""
External Zenoh subscriber demonstrating interoperability with hiroz.

This Python script subscribes to hiroz topics and decodes the encoding metadata,
showing how non-ROS applications can interact with hiroz using Zenoh.

Requirements:
    pip install zenoh

Usage:
    # Terminal 1: Start hiroz publisher
    cargo run --example protobuf_interop --features protobuf -- pub

    # Terminal 2: Run this script
    python3 examples/zenoh_subscriber.py
"""

import zenoh
import time
import sys


def main():
    print("=== External Zenoh Subscriber ===\n")
    print("Demonstrating interoperability with hiroz Protobuf encoding\n")

    # Open Zenoh session
    conf = zenoh.Config()
    session = zenoh.open(conf)

    print("Zenoh session opened")
    print("\nSubscribing to:")
    print("  - /interop/protobuf (expects Protobuf encoding)")
    print("  - /interop/cdr (expects CDR encoding)")
    print("\nWaiting for messages...\n")

    def protobuf_callback(sample):
        """Handle Protobuf-encoded messages"""
        encoding = sample.encoding
        payload = sample.payload.to_bytes()

        print("[PROTO] Received message")
        print(f"        Encoding: {encoding}")
        print(f"        Payload size: {len(payload)} bytes")
        print(f"        Raw data: {payload[:50]}...")  # Show first 50 bytes

        # In a real application, you would decode the Protobuf here
        # using the schema from encoding.schema
        if "schema=" in str(encoding):
            schema = str(encoding).split("schema=")[1]
            print(f"        Schema: {schema}")
        print()

    def cdr_callback(sample):
        """Handle CDR-encoded messages"""
        encoding = sample.encoding
        payload = sample.payload.to_bytes()

        print("[CDR]   Received message")
        print(f"        Encoding: {encoding}")
        print(f"        Payload size: {len(payload)} bytes")
        print(f"        Raw data: {payload[:50]}...")  # Show first 50 bytes
        print()

    # Subscribe to both topics
    sub_proto = session.declare_subscriber("/interop/protobuf", protobuf_callback)

    sub_cdr = session.declare_subscriber("/interop/cdr", cdr_callback)

    print("✓ Subscriptions active\n")
    print("Press Ctrl+C to exit\n")

    try:
        # Keep alive
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        print("\n\nShutting down...")

    # Cleanup
    sub_proto.undeclare()
    sub_cdr.undeclare()
    session.close()

    print("=== Complete ===")


if __name__ == "__main__":
    try:
        main()
    except Exception as e:
        print(f"Error: {e}", file=sys.stderr)
        sys.exit(1)