Binding · Python

Python — in 5 Minuten zum ersten Sample

PyO3-Bindings für die DCPS-API + typisierte Topics aus zerodds-idlc generate --lang python + AsyncIO-Wrapper. abi3-Wheels für Python 3.9 und neuer. Der erzeugte Code ruft den XCDR-Codec in zerodds-wire.

Python: 3.9+ (abi3) · erzeugter Code 3.10+ Install: pip install zerodds Spec: Vendor-Spec

5-Minuten-Quickstart

Ein virtualenv, eine Dependency, ~15 Zeilen Python.

1 · zerodds-py installieren

Das Paket zerodds ist auf PyPI und zieht zerodds-wire in derselben Version mit:

python3 -m venv .venv
source .venv/bin/activate    # Windows: .venv\Scripts\activate
pip install zerodds==1.0.0rc7

Aus dem Source baut maturin die Erweiterung; vorher kommt die Wire-Runtime aus runtime/python:

git clone https://github.com/zero-objects/zero-dds.git
cd zero-dds
python3 -m venv .venv
source .venv/bin/activate    # Windows: .venv\Scripts\activate
pip install maturin
pip install ./runtime/python
cd crates/py
maturin develop --release --features extension-module

Wenn du nur die Runtime-Library (libzerodds.so) systemweit haben willst, kannst du das pre-built Paket nehmen:

# Ubuntu/Debian (amd64; arm64: _arm64.deb)
wget https://github.com/zero-objects/zero-dds/releases/download/v1.0.0-rc.7/zerodds-core_1.0.0-rc.7_amd64.deb
sudo dpkg -i zerodds-core_1.0.0-rc.7_amd64.deb

# Fedora/RHEL: die RPMs von rc.7 lassen sich nicht installieren (zerodds-core verlangt libzerodds.so()(64bit), das kein Paket bereitstellt); aus der Quelle bauen, siehe User Guide

# Arch Linux — AUR
yay -S zerodds-bin

Die Pakete liefern libzerodds.so, die zerodds.loader über ctypes lädt. Für macOS und Windows gibt es kein vorgebautes Paket mit libzerodds; libzerodds.dylib und zerodds.dll entstehen mit cargo build --release -p zerodds-c-api. Die PyO3-Erweiterung aus dem PyPI-Paket bringt den DDS-Stack selbst mit.

2 · Pub/Sub-Roundtrip (`hello.py`)

import zerodds

factory = zerodds.DomainParticipantFactory.instance()
participant = factory.create_participant(0)

topic = participant.create_bytes_topic("Chatter")
publisher = participant.create_publisher()
subscriber = participant.create_subscriber()

writer = publisher.create_bytes_writer(topic)
reader = subscriber.create_bytes_reader(topic)

writer.wait_for_matched_subscription(1, timeout_secs=5.0)
reader.wait_for_matched_publication(1, timeout_secs=5.0)

writer.write(b"hello")
reader.wait_for_data(timeout_secs=3.0)
for payload in reader.take():
    print(f"got: {payload!r}")

▶ Runnable example: python-hello-bytes

3 · Laufen

python hello.py
# got: b'hello'

Installation

Aus PyPI

pip install zerodds==1.0.0rc7

Zieht zerodds-wire 1.0.0rc7 mit. Wheels für Linux x86_64/aarch64, macOS arm64 und Windows x86_64; ein abi3-Wheel pro Plattform. Das Paket verlangt Python 3.9 oder neuer.

Aus dem Source (Dev)

git clone https://github.com/zero-objects/zero-dds.git
cd zero-dds/crates/py
python3 -m venv .venv
source .venv/bin/activate
pip install maturin pytest
pip install ../../runtime/python
maturin develop --features extension-module
pytest python/tests/

Voraussetzungen

Python 3.9 oder höher; der Code aus zerodds-idlc generate --lang python braucht 3.10 oder höher.
Bei Build-aus-Source: Rust 1.97+ (siehe Dev-Install) + Xcode-CLT (macOS) bzw. build-essential (Linux).
Aus-Source kompiliert ~30-60s auf einem 8-Core; aus PyPI <5s.

Typisierte Messages — `zerodds-idlc generate --lang python`

Typen kommen aus IDL. zerodds-idlc generate --lang python schreibt eine Dataclass je Typ, ihren Codec über zerodds_wire und ZD_TOPIC_DESCRIPTORS; zerodds.IdlTopic nimmt einen Eintrag daraus. Den Decorator @idl_struct (zerodds.idl) und zerodds.cdr gibt es nicht mehr. Der erzeugte Code braucht Python 3.10 oder neuer.

IDL (`temperature.idl`)

module sensor_msgs {
    module msg {
        struct Temperature {
            long celsius;
            string sensor_id;
            sequence<octet> raw_blob;
        };
    };
};

Codegen + Nutzen

zerodds-idlc generate --lang python --out . temperature.idl
# schreibt ./temperature.py

import zerodds
from temperature import ZD_TOPIC_DESCRIPTORS

factory = zerodds.DomainParticipantFactory.instance()
p = factory.create_participant(0)

topic = zerodds.IdlTopic(p, "Temp", ZD_TOPIC_DESCRIPTORS["sensor_msgs::msg::Temperature"])
Temperature = topic.cls
writer = topic.create_writer(p.create_publisher())
reader = topic.create_reader(p.create_subscriber())

writer.wait_for_matched_subscription(1, 5.0)
reader.wait_for_matched_publication(1, 5.0)

writer.write(Temperature(celsius=23, sensor_id="A7", raw_blob=[]))
reader.wait_for_data(3.0)
for msg in reader.take():
    print(msg)
# sensor_msgs.msg.Temperature(celsius=23, sensor_id='A7', raw_blob=[])

▶ Runnable example: python-idl-struct

Was Python im Typ nicht ausdrücken kann (Breite einer Ganzzahl, Grenze eines Strings oder einer Sequenz), steht als typing.Annotated-Metadaten in der erzeugten Klasse; typing.get_type_hints(cls, include_extras=True) liest es zurück.

AsyncIO

Für async-fokussierte Apps gibt's zerodds.aio-Wrapper. Alle wait_*/write-Calls laufen über asyncio.to_thread; Event-Loop bleibt unblockiert.

import asyncio
import zerodds
from zerodds import aio as zaio

async def main():
    factory = zerodds.DomainParticipantFactory.instance()
    p = factory.create_participant(0)
    topic = p.create_bytes_topic("Chatter")

    writer = zaio.AsyncBytesWriter(p.create_publisher().create_bytes_writer(topic))
    reader = zaio.AsyncBytesReader(p.create_subscriber().create_bytes_reader(topic))

    await writer.wait_for_matched_subscription(1, 5.0)
    await reader.wait_for_matched_publication(1, 5.0)

    await writer.write(b"async hello")
    await reader.wait_for_data(3.0)
    for payload in reader.take():
        print(payload)

asyncio.run(main())

▶ Runnable example: python-aio

Klassen: AsyncBytesWriter, AsyncBytesReader, AsyncShapeWriter, AsyncShapeReader, AsyncWaitSet. Implementation: crates/py/python/zerodds/aio.py

QoS-Builder — alle 22 Policies

Statt Default-QoS kannst du jeden der 22 Policies setzen. Pattern: QosBuilder.set_X(...), dann create_*_writer_with_qos(topic, qos).

import zerodds

factory = zerodds.DomainParticipantFactory.instance()
p = factory.create_participant(0)
topic = p.create_bytes_topic("Setpoint")

# State-Topic mit Late-Joiner-Cache (siehe QoS Cookbook §2)
qos = zerodds.DataWriterQos()
qos.set_reliability("Reliable", 0.1)
qos.set_durability("TransientLocal")
qos.set_history("KeepLast", 1)

writer = p.create_publisher().create_bytes_writer_with_qos(topic, qos)

▶ Runnable example: python-qos-writer

Verfügbare Setter (auswahl): set_reliability, set_durability, set_history, set_deadline, set_lifespan, set_latency_budget, set_liveliness, set_ownership, set_ownership_strength, set_partition, set_presentation, set_destination_order, set_resource_limits, set_transport_priority, set_writer_data_lifecycle, set_user_data, set_topic_data, set_group_data, set_durability_service.

Vollständige Reference: QoS Reference.

Tests & CI

Im Repo unter crates/py/python/tests/ liegen die pytest-Tests. Lauf:

cd crates/py
maturin develop --features extension-module
pytest python/tests/ -v

Conditional-Skips:

test_loader_smoke.py — skippt ohne gebaute libzerodds.so. Aktivieren: cargo build -p zerodds-c-api && ZERODDS_LIB=$PWD/../../target/debug/libzerodds.so pytest …
test_shapesdemo_interop.py — skippt ohne ddsperf im PATH (Cyclone DDS installieren für Cross-Vendor-Smoke).
tests/ros2/ — skippt ohne ROS_DISTRO + RMW_IMPLEMENTATION=rmw_zerodds_shim.

CI-Pattern: GitHub Actions / GitLab CI mit maturin develop-Step plus pytest.

Binding · Python

Python — first sample in 5 minutes

PyO3 bindings for the DCPS API + typed topics from zerodds-idlc generate --lang python + an AsyncIO wrapper. abi3 wheels for Python 3.9 and newer. The generated code calls the XCDR codec in zerodds-wire.

Python: 3.9+ (abi3) · generated code 3.10+ Install: pip install zerodds Spec: vendor spec

5-minute quickstart

A virtualenv, one dependency, ~15 lines of Python.

1 · Install zerodds-py

The zerodds package is on PyPI and pulls in zerodds-wire of the same version:

python3 -m venv .venv
source .venv/bin/activate    # Windows: .venv\Scripts\activate
pip install zerodds==1.0.0rc7

From source, maturin builds the extension; the wire runtime from runtime/python goes in first:

git clone https://github.com/zero-objects/zero-dds.git
cd zero-dds
python3 -m venv .venv
source .venv/bin/activate    # Windows: .venv\Scripts\activate
pip install maturin
pip install ./runtime/python
cd crates/py
maturin develop --release --features extension-module

If you only want the runtime library (libzerodds.so) system-wide, you can take the pre-built package:

# Ubuntu/Debian (amd64; arm64: _arm64.deb)
wget https://github.com/zero-objects/zero-dds/releases/download/v1.0.0-rc.7/zerodds-core_1.0.0-rc.7_amd64.deb
sudo dpkg -i zerodds-core_1.0.0-rc.7_amd64.deb

# Fedora/RHEL: the rc.7 RPMs cannot be installed (zerodds-core requires libzerodds.so()(64bit), which no package provides); build from source, see the User Guide

# Arch Linux — AUR
yay -S zerodds-bin

The packages provide libzerodds.so, which zerodds.loader loads through ctypes. For macOS and Windows there is no pre-built package with libzerodds; libzerodds.dylib and zerodds.dll come from cargo build --release -p zerodds-c-api. The PyO3 extension from the PyPI package carries the DDS stack itself.

2 · Pub/sub roundtrip (`hello.py`)

import zerodds

factory = zerodds.DomainParticipantFactory.instance()
participant = factory.create_participant(0)

topic = participant.create_bytes_topic("Chatter")
publisher = participant.create_publisher()
subscriber = participant.create_subscriber()

writer = publisher.create_bytes_writer(topic)
reader = subscriber.create_bytes_reader(topic)

writer.wait_for_matched_subscription(1, timeout_secs=5.0)
reader.wait_for_matched_publication(1, timeout_secs=5.0)

writer.write(b"hello")
reader.wait_for_data(timeout_secs=3.0)
for payload in reader.take():
    print(f"got: {payload!r}")

▶ Runnable example: python-hello-bytes

3 · Run

python hello.py
# got: b'hello'

Installation

From PyPI

pip install zerodds==1.0.0rc7

Pulls in zerodds-wire 1.0.0rc7. Wheels for Linux x86_64/aarch64, macOS arm64 and Windows x86_64; one abi3 wheel per platform. The package requires Python 3.9 or newer.

From source (dev)

git clone https://github.com/zero-objects/zero-dds.git
cd zero-dds/crates/py
python3 -m venv .venv
source .venv/bin/activate
pip install maturin pytest
pip install ../../runtime/python
maturin develop --features extension-module
pytest python/tests/

Prerequisites

Python 3.9 or higher; the code from zerodds-idlc generate --lang python needs 3.10 or higher.
For a build from source: Rust 1.97+ (see Dev Install) + Xcode CLT (macOS) or build-essential (Linux).
From source it compiles in ~30-60s on an 8-core; from PyPI <5s.

Typed messages — `zerodds-idlc generate --lang python`

Types come from IDL. zerodds-idlc generate --lang python writes a dataclass per type, its codec over zerodds_wire and ZD_TOPIC_DESCRIPTORS; zerodds.IdlTopic takes one entry of it. The @idl_struct decorator (zerodds.idl) and zerodds.cdr are gone. The generated code needs Python 3.10 or newer.

IDL (`temperature.idl`)

module sensor_msgs {
    module msg {
        struct Temperature {
            long celsius;
            string sensor_id;
            sequence<octet> raw_blob;
        };
    };
};

Codegen + use

zerodds-idlc generate --lang python --out . temperature.idl
# writes ./temperature.py

import zerodds
from temperature import ZD_TOPIC_DESCRIPTORS

factory = zerodds.DomainParticipantFactory.instance()
p = factory.create_participant(0)

topic = zerodds.IdlTopic(p, "Temp", ZD_TOPIC_DESCRIPTORS["sensor_msgs::msg::Temperature"])
Temperature = topic.cls
writer = topic.create_writer(p.create_publisher())
reader = topic.create_reader(p.create_subscriber())

writer.wait_for_matched_subscription(1, 5.0)
reader.wait_for_matched_publication(1, 5.0)

writer.write(Temperature(celsius=23, sensor_id="A7", raw_blob=[]))
reader.wait_for_data(3.0)
for msg in reader.take():
    print(msg)
# sensor_msgs.msg.Temperature(celsius=23, sensor_id='A7', raw_blob=[])

▶ Runnable example: python-idl-struct

What Python cannot express in a type (the width of an integer, the bound of a string or a sequence) is written as typing.Annotated metadata on the generated class; typing.get_type_hints(cls, include_extras=True) reads it back.

AsyncIO

For async-focused apps there's a zerodds.aio wrapper. All wait_*/write calls run over asyncio.to_thread; the event loop stays unblocked.

import asyncio
import zerodds
from zerodds import aio as zaio

async def main():
    factory = zerodds.DomainParticipantFactory.instance()
    p = factory.create_participant(0)
    topic = p.create_bytes_topic("Chatter")

    writer = zaio.AsyncBytesWriter(p.create_publisher().create_bytes_writer(topic))
    reader = zaio.AsyncBytesReader(p.create_subscriber().create_bytes_reader(topic))

    await writer.wait_for_matched_subscription(1, 5.0)
    await reader.wait_for_matched_publication(1, 5.0)

    await writer.write(b"async hello")
    await reader.wait_for_data(3.0)
    for payload in reader.take():
        print(payload)

asyncio.run(main())

▶ Runnable example: python-aio

Classes: AsyncBytesWriter, AsyncBytesReader, AsyncShapeWriter, AsyncShapeReader, AsyncWaitSet. Implementation: crates/py/python/zerodds/aio.py

QoS builder — all 22 policies

Instead of the default QoS you can set any of the 22 policies. Pattern: QosBuilder.set_X(...), then create_*_writer_with_qos(topic, qos).

import zerodds

factory = zerodds.DomainParticipantFactory.instance()
p = factory.create_participant(0)
topic = p.create_bytes_topic("Setpoint")

# state topic with a late-joiner cache (see QoS Cookbook §2)
qos = zerodds.DataWriterQos()
qos.set_reliability("Reliable", 0.1)
qos.set_durability("TransientLocal")
qos.set_history("KeepLast", 1)

writer = p.create_publisher().create_bytes_writer_with_qos(topic, qos)

▶ Runnable example: python-qos-writer

Available setters (selection): set_reliability, set_durability, set_history, set_deadline, set_lifespan, set_latency_budget, set_liveliness, set_ownership, set_ownership_strength, set_partition, set_presentation, set_destination_order, set_resource_limits, set_transport_priority, set_writer_data_lifecycle, set_user_data, set_topic_data, set_group_data, set_durability_service.

Full reference: QoS Reference.

Tests & CI

In the repo under crates/py/python/tests/ are the pytest tests. Run:

cd crates/py
maturin develop --features extension-module
pytest python/tests/ -v

Conditional skips:

test_loader_smoke.py — skipped without a built libzerodds.so. Enable: cargo build -p zerodds-c-api && ZERODDS_LIB=$PWD/../../target/debug/libzerodds.so pytest …
test_shapesdemo_interop.py — skipped without ddsperf in PATH (install Cyclone DDS for the cross-vendor smoke).
tests/ros2/ — skipped without ROS_DISTRO + RMW_IMPLEMENTATION=rmw_zerodds_shim.

CI pattern: GitHub Actions / GitLab CI with a maturin develop step plus pytest.

Python — in 5 Minuten zum ersten Sample

5-Minuten-Quickstart

1 · zerodds-py installieren

2 · Pub/Sub-Roundtrip (hello.py)

3 · Laufen

Installation

Aus PyPI

Aus dem Source (Dev)

Voraussetzungen

Typisierte Messages — zerodds-idlc generate --lang python

IDL (temperature.idl)

Codegen + Nutzen

AsyncIO

QoS-Builder — alle 22 Policies

Tests & CI

Verwandt

Python — first sample in 5 minutes

5-minute quickstart

1 · Install zerodds-py

2 · Pub/sub roundtrip (hello.py)

3 · Run

Installation

From PyPI

From source (dev)

Prerequisites

Typed messages — zerodds-idlc generate --lang python

IDL (temperature.idl)

Codegen + use

AsyncIO

QoS builder — all 22 policies

Tests & CI

Related

2 · Pub/Sub-Roundtrip (`hello.py`)

Typisierte Messages — `zerodds-idlc generate --lang python`

IDL (`temperature.idl`)

2 · Pub/sub roundtrip (`hello.py`)

Typed messages — `zerodds-idlc generate --lang python`

IDL (`temperature.idl`)