DOCSEARCHbyAlgolia
Submit Docs

Zenoh Python API

repository·main·Indexed 41 days ago

https://github.com/eclipse-zenoh/zenoh-python

Python bindings for the Zenoh high-performance protocol, providing a unified interface for data in motion (pub/sub), data at rest (storage/query), and computation. Version 1.10.1 includes core session management, the zenoh.handlers module for message processing, zenoh.ext for extended functionality, and zenoh.shm for high-performance shared memory communication.

Tokens
14.6K
Snippets
57
Records
67
Agent score
51%

What's inside Zenoh

  1. Overview of Zenoh Python API

    main

    Zenoh is a unified stack for data in motion, data at rest, and computations. It combines traditional pub/sub with geo-distributed storage, queries, and computations. The Zenoh protocol enables nodes to form various network topologies (mesh, star, or clique) using Zenoh routers to maintain connectivity and route messages.

    The zenoh-python library provides Python bindings over the core Zenoh implementation written in Rust. For deep technical details on the underlying protocol, the Zenoh Rust API documentation is a useful supplementary resource.

  2. How stub conversion works for zenoh-python documentation

    main

    Since zenoh-python uses Rust (via PyO3), its Python type information is stored in .pyi stub files. Sphinx's autodoc cannot read these directly.

    The stubs_to_sources.py script bridges this gap by:

    • Creating .py versions of the .pyi files containing the documentation.
    • Backing up original .py files to a _stubs_backup/ directory.
    • Leaving the original .pyi files unchanged (they are ignored at runtime).
    • Providing a --recover flag to restore the original state.

  3. Use Key Expressions for addressing data

    main

    Zenoh uses slash-separated paths (e.g., robot/sensor/temp) as an address space. You can use zenoh.KeyExpr to validate and manipulate these expressions.

    Wildcards:

    • *: Matches any single chunk (e.g., robot/sensor/* matches robot/sensor/temp).
    • **: Matches any number of chunks, including zero (e.g., robot/** matches robot/sensor/temp and robot/status).

    Operations:

    • autocanonize(): Converts a key expression to its canonical form.
    • intersects(other): Checks if two expressions overlap.
    • includes(other): Checks if one expression is a subset of another.

    Invalid syntax (like spaces or empty chunks //) will raise a zenoh.ZError.

    import zenoh

# Validation
expr = zenoh.KeyExpr("robot/sensor/*")

# Canonicalization
expr = zenoh.KeyExpr("robot/sensor/**/*").autocanonize()

# Operations
expr1 = zenoh.KeyExpr("robot/sensor/*")
expr2 = zenoh.KeyExpr("robot/sensor/temp")
print(expr1.includes(expr2))  # True

  4. Represent and encode data

    main

    Data is handled as zenoh.ZBytes (raw bytes).

    Data Structures

    • zenoh.Sample: Contains payload (ZBytes), timestamp, encoding, kind, and optional attachment (ZBytes).

    Serialization

    Use zenoh.ext.z_serialize and zenoh.ext.z_deserialize for basic types.

    Encoding

    zenoh.Encoding describes how to interpret data (MIME-like, e.g., application/json).

    • Use predefined attributes like zenoh.Encoding.APPLICATION_JSON for efficiency.
    • You can associate a schema using encoding.with_schema(schema_string).
    import zenoh
from zenoh.ext import z_serialize, z_deserialize

# Serialization
data = {"temp": 25.5}
serialized = z_serialize(data)

# Using Encodings
encoding = zenoh.Encoding.APPLICATION_JSON
sample = zenoh.Sample(payload=serialized, encoding=encoding)

  5. Use Query/Reply for on-demand data

    main

    The Query/Reply paradigm allows requesting data from a provider.

    • Queryable: A provider that responds to requests. Declare it using session.declare_queryable(key_expr). In the callback, use query.reply(sample) for data, query.reply_del() for deletions, or query.reply_err(error) for errors.
    • Querier: A requester that uses session.get(selector) or a zenoh.Querier object.
    • Selector: A zenoh.Selector combines a KeyExpr with optional parameters (e.g., key/expr?param=val).
    import zenoh

# Queryable (Provider)
def queryable_handler(query):
    query.reply(zenoh.Sample(zenoh.ZBytes("data")))

with zenoh.open() as session:
    session.declare_queryable("data/query", queryable_handler)

    # Querier (Requester)
    reply = session.get("data/query")
    for r in reply:
        print(r.sample.payload.decode())

  6. Handle data via Channels or Callbacks

    main

    Zenoh provides two ways to receive sequential data (Samples or Replies):

    Channels (Iterative)

    By providing a channel type (like zenoh.handlers.FifoChannel or zenoh.handlers.RingChannel) to the handler parameter, the returned object becomes an iterable.

    • Use for item in subscriber: to receive data.
    • Use subscriber.recv() for blocking receive or subscriber.try_recv() for non-blocking.
    • The subscriber is automatically undeclared when the object goes out of scope.

    Callbacks (Background Mode)

    Pass a callable to the handler parameter. The subscriber/queryable runs in background mode and remains active even if the returned object goes out of scope.

    CAUTION: Do not call Zenoh API functions or perform blocking operations inside a callback. This is unsafe and can lead to deadlocks or crashes.

    import zenoh
from zenoh.handlers import FifoChannel

# Channel approach
with zenoh.open() as session:
    sub = session.declare_subscriber("key", handler=FifoChannel())
    for sample in sub:
        print(sample.payload)

# Callback approach
def my_callback(sample):
    print(f"Received {sample.payload}")

with zenoh.open() as session:
    # This stays active in the background
    session.declare_subscriber("key", handler=my_callback)

  7. Publish and Subscribe to data

    main

    Zenoh supports two communication paradigms:

    Publish/Subscribe

    • Publishing: Use zenoh.Publisher (via session.declare_publisher(key_expr)) to send data using .put(payload) or .delete() to remove an association. You can also publish directly via session.put() and session.delete().
    • Subscribing: Use zenoh.Subscriber (via session.declare_subscriber(key_expr)) to receive zenoh.Sample objects.
    • Sample Kinds: Subscribers can distinguish between zenoh.SampleKind.PUT (data update) and zenoh.SampleKind.DELETE (data removal) using the sample.kind attribute.
    import zenoh

with zenoh.open() as session:
    # Publisher
    publisher = session.declare_publisher("robot/temp")
    publisher.put("25.5")
    publisher.delete()

    # Subscriber
    subscriber = session.declare_subscriber("robot/temp")
    for sample in subscriber: 
        print(f"Received: {sample.payload.decode()} kind: {sample.kind}")

  8. Run Zenoh Python Examples

    main

    You can run the provided examples using python3 <example.py>. Each example supports the -h or --help flag to display available arguments and their default values.

    Important for Docker users: If you are running tests against a Zenoh router inside a Docker container, you must add the -e tcp/localhost:7447 option to your commands. This is necessary because Docker does not support UDP multicast, which is required for Zenoh's scouting and discovery mechanisms to function.

    # Standard execution
python3 <example.py>

# Execution when using a Zenoh router in Docker
python3 <example.py> -e tcp/localhost:7447

  9. Install eclipse-zenoh via pip

    main

    The eclipse-zenoh library is available on PyPI. It is recommended to install it within a virtual environment.

    Prerequisites for source distributions: If a binary wheel is not available for your platform, pip will attempt to build from source. You must have:

    • pip version 19.3.1 or higher.
    • A Rust toolchain installed (via rustup.rs).
    pip install eclipse-zenoh

  10. Build eclipse-zenoh from source (Alternative: Without Virtual Environment)

    main

    If you cannot use a virtual environment, follow these steps. Note that you may need to use --break-system-packages when installing the built wheel on some systems.

    Steps:

    1. Install development requirements.
    2. Ensure maturin is in your PATH.
    3. Build the wheel using maturin build.
    4. Install the resulting wheel using pip.

    Warning: Ensure the pip you use corresponds to the python3 version you intend to use. If they differ, use python3 -m pip install ... to target the correct installation.

    # 1. Install development requirements
pip install -r requirements-dev.txt

# 2. Ensure maturin is in PATH (example for Ubuntu)
export PATH="$HOME/.local/bin:$PATH"

# 3. Build the wheel
maturin build --release

# 4. Install the built wheel
pip install ./target/wheels/*.whl --break-system-packages

# 5. Run examples
python3 examples/z_info.py

  11. Build the documentation locally

    main

    You can build the HTML documentation locally using make.

    Steps:

    1. Ensure eclipse-zenoh is installed.
    2. Install documentation-specific requirements from docs/requirements.txt.
    3. Run make html inside the docs directory.
    4. Open the generated _build/html/index.html in your browser.
    pip install -r docs/requirements.txt
cd docs
make html
# Then open _build/html/index.html in your browser

  12. Create and manage a Zenoh Session

    main

    A zenoh.Session is the primary entity for performing communication (Publish/Subscribe or Query/Reply). Sessions are created using zenoh.open(config).

    Important: Always use a context manager (with statement) to manage the session lifecycle. If a session is not explicitly closed or managed via a context manager, the script may hang during exit because object finalizers might be called after the library thread has been killed. Alternatively, call zenoh.Session.close() explicitly before exiting.

    import zenoh

# Recommended: Using a context manager
with zenoh.open() as session:
    # Use the session here
    pass

# Alternative: Explicit close
session = zenoh.open()
try:
    # Use the session here
    pass
finally:
    session.close()