Binding · Python

Python — in 5 Minuten zum ersten Sample

PyO3-Bindings für die DCPS-API + @idl_struct-Decorator für Codegen-freien Topic-Pfad + AsyncIO-Wrapper. Wheels für CPython 3.8–3.14 (abi3-py38). Pure-Python cdr.py-Codec ist byte-identisch zum Rust-Pfad.

Python: 3.8–3.14 (abi3-py38) Install: Source (RC1) · pip install zerodds ab RC3 Status: ✓ RC3 · Vendor-Spec

5-Minuten-Quickstart

Ein virtualenv, eine Dependency, ~15 Zeilen Python.

1 · zerodds-py installieren

zerodds-py ist auf PyPI — pip install zerodds. Alternativ aus dem Source: Heute geht der Build aus Source via maturin:

git clone https://github.com/zero-objects/zero-dds.git
cd zero-dds/crates/py
python3 -m venv .venv
source .venv/bin/activate    # Windows: .venv\Scripts\activate
pip install maturin
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
wget https://github.com/zero-objects/zero-dds/releases/download/v1.0.0-rc.3/zerodds-core_1.0.0-rc.3_amd64.deb
sudo dpkg -i zerodds-core_1.0.0-rc.3_amd64.deb

# Fedora/RHEL
sudo dnf install https://github.com/zero-objects/zero-dds/releases/download/v1.0.0-rc.3/zerodds-core-1.0.0-0.rc1.fc40.x86_64.rpm

# macOS — Homebrew Tap
brew tap zero-objects/zerodds && brew install zerodds

# Arch Linux — AUR
yay -S zerodds-bin

# Docker
docker pull fishermen21/zerodds-cli

Das deckt die libzerodds.{so,dylib,dll}-Runtime ab, gegen die zerodds-py via PyO3 linkt. Der Python-Wrapper selbst kommt aktuell nur aus Source.

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 (RC3-Plan, noch nicht live)

# Ab RC3:
pip install zerodds

Wheels für Linux x86_64/aarch64, macOS arm64/x86_64, Windows x86_64. abi3-py38: ein Wheel pro Plattform deckt CPython 3.8–3.14 ab.

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
maturin develop --features extension-module
pytest python/tests/

Voraussetzungen

Python 3.8 oder höher (3.10+ empfohlen).
Bei Build-aus-Source: Rust 1.88+ (siehe Dev-Install) + Xcode-CLT (macOS) bzw. build-essential (Linux).
Aus-Source kompiliert ~30-60s auf einem 8-Core; aus PyPI <5s.

Typisierte Messages — `@idl_struct`

Kein externer IDL-Compiler nötig. Decorator über eine @dataclass, der Encoder/Decoder entsteht zur Decorator-Zeit. Byte-identisch zum Rust-Codegen-Pfad.

from dataclasses import dataclass
import zerodds
from zerodds.idl import idl_struct, Int32, String, Bytes

@idl_struct(typename="sensor_msgs::msg::Temperature")
@dataclass
class Temperature:
    celsius: Int32
    sensor_id: String
    raw_blob: Bytes = b""

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

topic = zerodds.IdlTopic(p, "Temp", Temperature)
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"))
reader.wait_for_data(3.0)
for msg in reader.take():
    print(msg)
# Temperature(celsius=23, sensor_id='A7', raw_blob=b'')

▶ Runnable example: python-idl-struct

Unterstützte Field-Types: Bool, Int8/16/32/64, UInt8/16/32/64, Float32/64, String, Bytes, Sequence[T], Array[T, N], Optional[T], idl_enum, idl_union, nested @idl_struct. Python-Primitives (bool/int/float/str/bytes) werden automatisch auf XCDR2-Defaults gemapped.

Reference: Vendor-Spec §3 IDL-Mapping.

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 102 pytest-Tests (28 Smoke + 74 Pure-Python). 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 + an @idl_struct decorator for a codegen-free topic path + an AsyncIO wrapper. Wheels for CPython 3.8–3.14 (abi3-py38). The pure-Python cdr.py codec is byte-identical to the Rust path.

Python: 3.8–3.14 (abi3-py38) Install: source (RC1) · pip install zerodds from RC3 Status: ✓ RC3 · vendor spec

5-minute quickstart

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

1 · Install zerodds-py

zerodds-py is on PyPI — pip install zerodds. Or build from source: Today the build works from source via maturin:

git clone https://github.com/zero-objects/zero-dds.git
cd zero-dds/crates/py
python3 -m venv .venv
source .venv/bin/activate    # Windows: .venv\Scripts\activate
pip install maturin
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
wget https://github.com/zero-objects/zero-dds/releases/download/v1.0.0-rc.3/zerodds-core_1.0.0-rc.3_amd64.deb
sudo dpkg -i zerodds-core_1.0.0-rc.3_amd64.deb

# Fedora/RHEL
sudo dnf install https://github.com/zero-objects/zero-dds/releases/download/v1.0.0-rc.3/zerodds-core-1.0.0-0.rc1.fc40.x86_64.rpm

# macOS — Homebrew tap
brew tap zero-objects/zerodds && brew install zerodds

# Arch Linux — AUR
yay -S zerodds-bin

# Docker
docker pull fishermen21/zerodds-cli

That covers the libzerodds.{so,dylib,dll} runtime that zerodds-py links against via PyO3. The Python wrapper itself currently comes only from source.

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 (RC3 plan, not live yet)

# from RC3:
pip install zerodds

Wheels for Linux x86_64/aarch64, macOS arm64/x86_64, Windows x86_64. abi3-py38: one wheel per platform covers CPython 3.8–3.14.

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
maturin develop --features extension-module
pytest python/tests/

Prerequisites

Python 3.8 or higher (3.10+ recommended).
For a build from source: Rust 1.88+ (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 — `@idl_struct`

No external IDL compiler needed. A decorator over a @dataclass; the encoder/decoder is built at decorator time. Byte-identical to the Rust codegen path.

from dataclasses import dataclass
import zerodds
from zerodds.idl import idl_struct, Int32, String, Bytes

@idl_struct(typename="sensor_msgs::msg::Temperature")
@dataclass
class Temperature:
    celsius: Int32
    sensor_id: String
    raw_blob: Bytes = b""

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

topic = zerodds.IdlTopic(p, "Temp", Temperature)
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"))
reader.wait_for_data(3.0)
for msg in reader.take():
    print(msg)
# Temperature(celsius=23, sensor_id='A7', raw_blob=b'')

▶ Runnable example: python-idl-struct

Supported field types: Bool, Int8/16/32/64, UInt8/16/32/64, Float32/64, String, Bytes, Sequence[T], Array[T, N], Optional[T], idl_enum, idl_union, nested @idl_struct. Python primitives (bool/int/float/str/bytes) are auto-mapped to XCDR2 defaults.

Reference: vendor spec §3 IDL mapping.

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/ there are 102 pytest tests (28 smoke + 74 pure-Python). 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 (RC3-Plan, noch nicht live)

Aus dem Source (Dev)

Voraussetzungen

Typisierte Messages — @idl_struct

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 (RC3 plan, not live yet)

From source (dev)

Prerequisites

Typed messages — @idl_struct

AsyncIO

QoS builder — all 22 policies

Tests & CI

Related

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

Typisierte Messages — `@idl_struct`

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

Typed messages — `@idl_struct`