zerodds-py 1.0.0-rc.1

PyO3 bindings for the ZeroDDS DCPS API
Documentation 
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

"""ShapeType-Cross-Vendor-Interop.

Pendant zu OMG ShapesDemo (Cyclone/Fast-DDS/RTI). Start im einem
Terminal::

    python examples/02_shape_pubsub.py publisher

und in einem anderen — egal ob zerodds, cyclonedds-python oder
rticonnextdds::

    python examples/02_shape_pubsub.py subscriber
"""

from __future__ import annotations

import argparse
import time

import zerodds


def run_publisher(
    domain_id: int,
    topic_name: str,
    color: str,
    count: int,
) -> None:
    factory = zerodds.DomainParticipantFactory.instance()
    participant = factory.create_participant(domain_id)
    topic = participant.create_shape_topic(topic_name)
    publisher = participant.create_publisher()
    writer = publisher.create_shape_writer(topic)

    print(f"[pub] '{topic_name}' warte auf Subscriber ...")
    writer.wait_for_matched_subscription(1, timeout_secs=30.0)
    print("[pub] matched, sende")

    x = 50
    y = 50
    direction = 1
    for i in range(count):
        shape = zerodds.Shape(color=color, x=x, y=y, shapesize=30)
        writer.write(shape)
        print(f"[pub] {i:03d}: {shape}")
        x += 5 * direction
        if x >= 240 or x <= 10:
            direction *= -1
        time.sleep(0.1)


def run_subscriber(domain_id: int, topic_name: str, count: int) -> None:
    factory = zerodds.DomainParticipantFactory.instance()
    participant = factory.create_participant(domain_id)
    topic = participant.create_shape_topic(topic_name)
    subscriber = participant.create_subscriber()
    reader = subscriber.create_shape_reader(topic)

    print(f"[sub] '{topic_name}' warte auf Publisher ...")
    reader.wait_for_matched_publication(1, timeout_secs=30.0)
    print("[sub] matched, empfange")

    received = 0
    while received < count:
        reader.wait_for_data(timeout_secs=10.0)
        for shape in reader.take():
            print(f"[sub] {received:03d}: {shape}")
            received += 1
            if received >= count:
                break


def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("role", choices=["publisher", "subscriber"])
    parser.add_argument("--domain", type=int, default=0)
    parser.add_argument(
        "--topic",
        type=str,
        default="Square",
        choices=["Square", "Circle", "Triangle"],
    )
    parser.add_argument("--color", type=str, default="BLUE")
    parser.add_argument("--count", type=int, default=50)
    args = parser.parse_args()

    if args.role == "publisher":
        run_publisher(args.domain, args.topic, args.color, args.count)
    else:
        run_subscriber(args.domain, args.topic, args.count)


if __name__ == "__main__":
    main()