### Run Zenoh Python examples

Source: https://github.com/eclipse-zenoh/zenoh-python/blob/main/README.md

Execute the 'z_info.py' example script to test the Zenoh Python installation. This helps verify that the library is correctly built and functional.

```bash
python examples/z_info.py
```

--------------------------------

### Run Zenoh Python Examples

Source: https://github.com/eclipse-zenoh/zenoh-python/blob/main/README.md

Executes a sample zenoh-python example script. Ensure you are using the same Python interpreter that was used for installation.

```bash
python3 examples/z_info.py
```

--------------------------------

### Test Documentation Examples

Source: https://github.com/eclipse-zenoh/zenoh-python/blob/main/docs/README.md

Run all examples located in the 'docs/examples/' directory using pytest. This ensures that the code examples are functional and do not time out.

```bash
# Test all docs examples (from project root)
python3 -m pytest tests/examples_check.py::test_docs_examples -v
```

--------------------------------

### Install Documentation Dependencies

Source: https://github.com/eclipse-zenoh/zenoh-python/blob/main/docs/README.md

Install the necessary Python packages for building the documentation. Navigate to the 'docs' directory first.

```bash
cd docs
pip install -r requirements.txt
```

--------------------------------

### Install zenoh-python

Source: https://github.com/eclipse-zenoh/zenoh-python/blob/main/docs/README.md

Build and install the zenoh-python package from source. Ensure you are in the project's root directory.

```bash
cd /path/to/zenoh-python
pip install -e .
```

--------------------------------

### Install Documentation Requirements

Source: https://github.com/eclipse-zenoh/zenoh-python/blob/main/README.md

Installs the Python packages required for building the project's documentation, as listed in the docs/requirements.txt file.

```bash
pip install -r docs/requirements.txt
```

--------------------------------

### Install zenoh-python

Source: https://context7.com/eclipse-zenoh/zenoh-python/llms.txt

Install the zenoh-python library from PyPI. To enable shared-memory transport, build from source with specific configuration.

```bash
pip install eclipse-zenoh

# To enable shared-memory transport (requires building from source):
pip install eclipse-zenoh --no-binary :all: --config-settings build-args="--features=zenoh/shared-memory"
```

--------------------------------

### Install Built Wheel

Source: https://github.com/eclipse-zenoh/zenoh-python/blob/main/README.md

Installs the zenoh-python wheel package that was just built. The --break-system-packages flag is used to allow installation in system-wide locations if necessary.

```bash
pip install ./target/wheels/*.whl --break-system-packages
```

--------------------------------

### Build HTML Documentation

Source: https://github.com/eclipse-zenoh/zenoh-python/blob/main/README.md

Navigates into the 'docs' directory and builds the HTML version of the documentation using the 'make html' command. This requires the documentation requirements to be installed.

```bash
cd docs
make html
```

--------------------------------

### Run Zenoh Python Examples

Source: https://github.com/eclipse-zenoh/zenoh-python/blob/main/examples/README.md

Basic command to execute any Zenoh Python example script. Use `-h` for detailed options. Add `-e tcp/localhost:7447` when running within a Docker container.

```bash
python3 <example.py>
```

--------------------------------

### Install eclipse-zenoh with shared-memory feature

Source: https://github.com/eclipse-zenoh/zenoh-python/blob/main/README.md

Install the eclipse-zenoh library from source, enabling the 'shared-memory' feature. This is useful for specific performance optimizations and requires a Rust toolchain to be installed.

```bash
pip install eclipse-zenoh --no-binary :all: --config-settings build-args="--features=zenoh/shared-memory"
```

--------------------------------

### Install development requirements

Source: https://github.com/eclipse-zenoh/zenoh-python/blob/main/README.md

Install the necessary development dependencies for Zenoh Python by referencing the 'requirements-dev.txt' file. This should be done within an activated virtual environment.

```bash
pip install -r requirements-dev.txt
```

--------------------------------

### Install eclipse-zenoh with pip

Source: https://github.com/eclipse-zenoh/zenoh-python/blob/main/README.md

Install the latest version of the eclipse-zenoh library using pip. This command is suitable for most users and assumes binary wheels are available for your platform.

```bash
pip install eclipse-zenoh
```

--------------------------------

### Get Zenoh Session Information

Source: https://github.com/eclipse-zenoh/zenoh-python/blob/main/examples/README.md

Retrieves information about the current Zenoh session. This is a typical usage example.

```bash
python3 z_info.py
```

--------------------------------

### Build and install Zenoh Python in development mode

Source: https://github.com/eclipse-zenoh/zenoh-python/blob/main/README.md

Build and install the Zenoh Python library in development mode using 'maturin develop'. This command compiles the Rust backend and makes the Python package available in the current environment.

```bash
maturin develop --release
```

--------------------------------

### Implement Zenoh Storage

Source: https://github.com/eclipse-zenoh/zenoh-python/blob/main/examples/README.md

A simple in-memory storage example. It creates a subscriber to store data and a queryable to serve stored data. Can be run with default or a specified key expression.

```bash
python3 z_storage.py
```

```bash
python3 z_storage.py -k 'demo/**'
```

--------------------------------

### Install Built Wheel for Specific Python Version

Source: https://github.com/eclipse-zenoh/zenoh-python/blob/main/README.md

Installs the zenoh-python wheel using 'python3 -m pip' to guarantee installation for the correct Python version, especially when multiple Python installations exist.

```bash
python3 -m pip install ./target/wheels/*.whl
```

--------------------------------

### Session.get

Source: https://context7.com/eclipse-zenoh/zenoh-python/llms.txt

Sends a get request to all matching queryables and returns an iterable of `Reply` objects. Each `Reply` has an `ok` field (a `Sample`) on success or an `err` field (a `ReplyError`) on failure.

```APIDOC
## `Session.get` — Query Queryables

Sends a get request to all matching queryables and returns an iterable of `Reply` objects. Each `Reply` has an `ok` field (a `Sample`) on success or an `err` field (a `ReplyError`) on failure.

```python
import zenoh

with zenoh.open(zenoh.Config()) as session:
    # Simple selector string
    for reply in session.get("room/temperature/history?day=2024-01-15"):
        if reply.ok:
            print(f"Temperature: {reply.ok.payload.to_string()}")
        else:
            print(f"Error: {reply.err.payload.to_string()}")

    # With explicit options
    for reply in session.get(
        "demo/example/**",
        target=zenoh.QueryTarget.ALL,
        timeout=5.0,
        payload="optional query payload",
    ):
        if reply.ok:
            print(f"({reply.ok.key_expr}) => {reply.ok.payload.to_string()}")
```
```

--------------------------------

### Get Session and Network Information in Python

Source: https://context7.com/eclipse-zenoh/zenoh-python/llms.txt

Retrieves Zenoh session details like ZID, connected routers, peers, transports, and links. It also demonstrates how to listen for live transport and link events.

```python
import zenoh
import time

with zenoh.open(zenoh.Config()) as session:
    info = session.info

    print(f"My ZID:    {info.zid()}")
    print(f"Routers:   {info.routers_zid()}")
    print(f"Peers:     {info.peers_zid()}")

    for transport in info.transports():
        print(f"Transport: {transport}")
    for link in info.links():
        print(f"Link: {link}")

    # Listen for live transport/link events
    transport_listener = info.declare_transport_events_listener(history=False)
    link_listener = info.declare_link_events_listener(history=False)

    for _ in range(30):  # Poll for 3 seconds
        while (event := transport_listener.try_recv()) is not None:
            print(f"Transport event: {event}")
        while (event := link_listener.try_recv()) is not None:
            print(f"Link event: {event}")
        time.sleep(0.1)

    transport_listener.undeclare()
    link_listener.undeclare()
```

--------------------------------

### Query Queryables with Session.get

Source: https://context7.com/eclipse-zenoh/zenoh-python/llms.txt

Send a get request to queryables and receive `Reply` objects. Each reply has an `ok` field (a `Sample`) on success or an `err` field (a `ReplyError`) on failure.

```python
import zenoh

with zenoh.open(zenoh.Config()) as session:
    # Simple selector string
    for reply in session.get("room/temperature/history?day=2024-01-15"):
        if reply.ok:
            print(f"Temperature: {reply.ok.payload.to_string()}")
        else:
            print(f"Error: {reply.err.payload.to_string()}")

    # With explicit options
    for reply in session.get(
        "demo/example/**",
        target=zenoh.QueryTarget.ALL,
        timeout=5.0,
        payload="optional query payload",
    ):
        if reply.ok:
            print(f"({reply.ok.key_expr}) => {reply.ok.payload.to_string()}")
```

--------------------------------

### Declare Matching Listener for Publisher/Querier Status

Source: https://context7.com/eclipse-zenoh/zenoh-python/llms.txt

Use `declare_matching_listener` to get notified when subscribers or queryables appear or disappear. This helps optimize bandwidth by publishing only when there are active listeners. Check `matching_status()` for the current state.

```python
import zenoh

with zenoh.open(zenoh.Config()) as session:
    # --- Publisher matching ---
    pub = session.declare_publisher("sensor/temperature")
    match_listener = pub.declare_matching_listener()

    # Also check the current status immediately
    status = pub.matching_status()
    print(f"Publisher currently has subscribers: {status.matching}")

    for status in match_listener:
        if status.matching:
            print("Publisher: at least one subscriber online — start publishing")
            pub.put("22.5°C")
        else:
            print("Publisher: no subscribers — pausing to save bandwidth")
        break

    # --- Querier matching ---
    querier = session.declare_querier("service/api")
    q_listener = querier.declare_matching_listener()

    for status in q_listener:
        if status.matching:
            print("Querier: a queryable is available")
        else:
            print("Querier: no queryables available")
        break
```

--------------------------------

### Get Data from Zenoh Queryables

Source: https://github.com/eclipse-zenoh/zenoh-python/blob/main/examples/README.md

Sends a query message for a selector to matching queryables (e.g., z_queryable, z_storage). The query callback receives the replies. Can be run with default or a specified selector.

```bash
python3 z_get.py
```

```bash
python3 z_get.py -s 'demo/**'
```

--------------------------------

### Add Maturin to PATH

Source: https://github.com/eclipse-zenoh/zenoh-python/blob/main/README.md

Ensures that the 'maturin' build tool, installed by the previous step, is accessible in the system's PATH. This is crucial for executing the build command.

```bash
export PATH="$HOME/.local/bin:$PATH"
```

--------------------------------

### Create a Zenoh Queryable

Source: https://github.com/eclipse-zenoh/zenoh-python/blob/main/examples/README.md

Defines a queryable function triggered by matching get operations. It returns a payload to the querier. Can be run with default or specified key expression and payload.

```bash
python3 z_queryable.py
```

```bash
python3 z_queryable.py -k demo/example/queryable -p 'This is the result'
```

--------------------------------

### Continuously Query Zenoh

Source: https://github.com/eclipse-zenoh/zenoh-python/blob/main/examples/README.md

Continuously sends query messages for a selector to matching queryables. The querier's callback receives the replies. Note: The example command provided in the source uses `z_get.py` instead of `z_querier.py`.

```bash
python3 z_querier.py
```

```bash
python3 z_get.py -s 'demo/**'
```

--------------------------------

### Quick Build and Open Documentation

Source: https://github.com/eclipse-zenoh/zenoh-python/blob/main/docs/README.md

Use the provided script to automate the documentation build process and open it in your browser. This script handles stub conversion, HTML generation, and cleanup.

```bash
./open.sh
```

--------------------------------

### Create and activate virtual environment

Source: https://github.com/eclipse-zenoh/zenoh-python/blob/main/README.md

Set up and activate a Python virtual environment to manage dependencies and avoid conflicts. This is a recommended practice for development.

```bash
python3 -m venv .venv
source .venv/bin/activate  # On Windows: .venv\Scripts\activate
```

--------------------------------

### Open Built Documentation

Source: https://github.com/eclipse-zenoh/zenoh-python/blob/main/README.md

Opens the locally built HTML documentation in the default web browser. The command varies depending on the operating system.

```bash
open _build/html/index.html  # macOS
```

```bash
xdg-open _build/html/index.html  # Linux
```

--------------------------------

### Verify Pip and Python Versions

Source: https://github.com/eclipse-zenoh/zenoh-python/blob/main/README.md

Checks that the 'pip' and 'python3' commands correspond to the same Python installation. This is important to ensure the package is installed for the intended Python environment.

```bash
pip --version
```

```bash
python3 --version
```

--------------------------------

### Create Zenoh Session with Context Manager

Source: https://github.com/eclipse-zenoh/zenoh-python/blob/main/docs/concepts.rst

Use a context manager for session creation to ensure proper resource management and avoid potential hangs on exit. This is the recommended approach.

```python
import zenoh

conf = zenoh.Config()
# configure zenoh with conf

with zenoh.open(conf) as session:
    # use session
    pass
# session is closed automatically
```

--------------------------------

### Using ZBytes for Raw Data

Source: https://github.com/eclipse-zenoh/zenoh-python/blob/main/docs/concepts.rst

Demonstrates how to create and use ZBytes for raw data payloads and attachments. Ensure ZBytes are properly initialized before use.

```python
import zenoh

conf = zenoh.Config()

# Using ZBytes
# ~~~~~~~~~~~~

# Create a ZBytes from raw data
raw_data = b'This is raw data'
zbytes_data = zenoh.ZBytes(raw_data)

# Access the raw data
print(f"Raw data: {zbytes_data.pop()}")

# Create a ZBytes from a string
zbytes_string = zenoh.ZBytes("This is a string")

# Access the string
print(f"String data: {zbytes_string.pop()}")

# Create a ZBytes from a list of integers
zbytes_list = zenoh.ZBytes([1, 2, 3])

# Access the list
print(f"List data: {zbytes_list.pop()}")
```

--------------------------------

### Scout Zenoh Network

Source: https://github.com/eclipse-zenoh/zenoh-python/blob/main/examples/README.md

Scouts for available Zenoh peers and routers on the network. This is a typical usage example.

```bash
python3 z_scout.py
```

--------------------------------

### Build HTML Documentation

Source: https://github.com/eclipse-zenoh/zenoh-python/blob/main/docs/README.md

Generate the HTML version of the documentation using the 'make html' command. This command is typically run after converting stubs to sources.

```bash
make html
```

--------------------------------

### Get Liveliness Tokens in Python

Source: https://github.com/eclipse-zenoh/zenoh-python/blob/main/docs/concepts.rst

Retrieves currently active liveliness tokens. Ensure the Zenoh session is established before calling.

```python
import zenoh

session = zenoh.open()

# Get currently present liveliness tokens
print(session.liveliness_get())

session.close()
```

--------------------------------

### Receive Data via Channels in Python

Source: https://github.com/eclipse-zenoh/zenoh-python/blob/main/docs/concepts.rst

Demonstrates receiving sequential data samples from a Zenoh subscriber using a channel. This approach allows iterating over received samples or using explicit receive methods.

```python
import zenoh

session = zenoh.open()

# Declare a subscriber using a FIFO channel
subscriber = session.declare_subscriber("my/topic", handler=zenoh.handlers.FifoChannel())

# Receive data samples sequentially
for sample in subscriber:
    print(f"Received: {sample.payload} on {sample.source}")

session.close()
```

--------------------------------

### Get Key/Value from Zenoh Python

Source: https://github.com/eclipse-zenoh/zenoh-python/blob/main/docs/quickstart.rst

Retrieve a specific key/value pair from Zenoh. This is useful for fetching current state or specific data points.

```python
import zenoh

conf = zenoh.Config()

# You can optionally configure Zenoh here
# For example, to enable the REST API:
# conf.insert("rest.addr", "0.0.0.0:8080")

# Start Zenoh session
keyexpr = zenoh.KeyExpr("example/resource")

with zenoh.open(conf) as session:
    # Get the value for the key
    print(f"Getting key='{keyexpr}'")
    sample = session.get(keyexpr)

    if sample:
        print(f"Received: key='{sample.key}' value='{sample.value}'")
    else:
        print("No value found.")

print("Done.")
```

--------------------------------

### Declare Zenoh Key Expressions

Source: https://github.com/eclipse-zenoh/zenoh-python/blob/main/docs/concepts.rst

Declare key expressions with a Zenoh session to optimize routing and network usage. This is typically done during session setup.

```python
import zenoh

conf = zenoh.Config()
# configure zenoh with conf

with zenoh.open(conf) as session:
    # Declare a publisher for a specific key expression
    publisher = session.declare_publisher("robot/device/+/status")
    # Use the publisher
    pass
# session is closed automatically
```

--------------------------------

### zenoh.open — Open a Session

Source: https://context7.com/eclipse-zenoh/zenoh-python/llms.txt

Creates a Session connected to the Zenoh network. It's recommended to use a `with` statement for automatic session closure.

```APIDOC
## zenoh.open — Open a Session

Creates a `Session` connected to the Zenoh network. Use a `with` statement (recommended) to ensure the session is closed properly; an unclosed session can cause the process to hang on exit.

```python
import zenoh

# Recommended: context manager ensures automatic close
with zenoh.open(zenoh.Config()) as session:
    session.put("demo/example/hello", "Hello World!")
# session is automatically closed here

# Alternative: explicit close with try/finally
session = zenoh.open(zenoh.Config())
try:
    session.put("demo/example/hello", "Hello World!")
finally:
    session.close()
```
```

--------------------------------

### Construct Query Parameters and Selectors with zenoh.Parameters and zenoh.Selector

Source: https://context7.com/eclipse-zenoh/zenoh-python/llms.txt

Use `zenoh.Parameters` to build query parameter sets from dictionaries. Combine a key expression and parameters using `zenoh.Selector` to form a query target. On the queryable side, parameters are accessed via `query.parameters`.

```python
import zenoh

with zenoh.open(zenoh.Config()) as session:
    # Build parameters from dict
    params = zenoh.Parameters({"day": "2024-01-15", "format": "fahrenheit"})

    # Combine key expression and parameters into a selector
    selector = zenoh.Selector("room/temperature/history", params)
    # Equivalent: selector = "room/temperature/history?day=2024-01-15;format=fahrenheit"

    for reply in session.get(selector):
        if reply.ok:
            print(reply.ok.payload.to_string())

    # On the queryable side, access parameters via query.parameters
    with session.declare_queryable("room/temperature/history") as qbl:
        for query in qbl:
            day = query.parameters.get("day", "unknown")
            fmt = query.parameters.get("format", "celsius")
            query.reply("room/temperature/history", f"22.5 on {day} ({fmt})")
            break
```

--------------------------------

### Data Serialization and Deserialization

Source: https://github.com/eclipse-zenoh/zenoh-python/blob/main/docs/concepts.rst

Shows how to serialize and deserialize basic types and structures using zenoh.ext.z_serialize and zenoh.ext.z_deserialize. Ensure the data types are compatible with the serialization functions.

```python
import zenoh

conf = zenoh.Config()

# Data serialization
# ~~~~~~~~~~~~~~~~~~

# Serialize a dictionary
my_dict = {"key": "value", "number": 123}
serialized_dict = zenoh.ext.z_serialize(my_dict)

# Deserialize the dictionary
deserialized_dict = zenoh.ext.z_deserialize(serialized_dict)
print(f"Serialized dict: {deserialized_dict}")

# Serialize a list
my_list = [1, 2, 3, 4, 5]
serialized_list = zenoh.ext.z_serialize(my_list)

# Deserialize the list
deserialized_list = zenoh.ext.z_deserialize(serialized_list)
print(f"Serialized list: {deserialized_list}")

# Serialize a string
my_string = "Hello Zenoh!"
serialized_string = zenoh.ext.z_serialize(my_string)

# Deserialize the string
deserialized_string = zenoh.ext.z_deserialize(serialized_string)
print(f"Serialized string: {deserialized_string}")
```

--------------------------------

### Declare Subscriber with Callback and Cleanup

Source: https://context7.com/eclipse-zenoh/zenoh-python/llms.txt

Illustrates creating a subscriber with a callback function for handling incoming samples and an optional cleanup function executed when the subscriber is undeclared. Note: Avoid calling Zenoh APIs within the callback.

```python
import zenoh

with zenoh.open(zenoh.Config()) as session:
    def on_sample(sample: zenoh.Sample):
        # NOTE: Do NOT call any Zenoh API from within a callback
        print(f">> Received: {sample.payload.to_string()}")

    def on_cleanup():
        print("Subscriber has been undeclared, performing cleanup...")

    handler = zenoh.handlers.Callback(on_sample, drop=on_cleanup)
    sub = session.declare_subscriber("demo/example/**", handler)
    # Subscriber is in background mode; lives as long as the session
    # Explicitly undeclare to trigger cleanup:
    # sub.undeclare()
```

--------------------------------

### Declare Subscribers with Different Channel Types

Source: https://context7.com/eclipse-zenoh/zenoh-python/llms.txt

Shows how to declare subscribers using default, FIFO, and Ring channel types with specified capacities. Blocking and non-blocking receives, as well as iteration, are demonstrated.

```python
import zenoh

with zenoh.open(zenoh.Config()) as session:
    # Default handler (FIFO with default capacity)
    sub_default = session.declare_subscriber("sensor/**")

    # Explicit FIFO with capacity 100 — blocks sender when full
    sub_fifo = session.declare_subscriber(
        "sensor/**", zenoh.handlers.FifoChannel(100)
    )

    # Ring buffer with capacity 10 — drops oldest when full (always wraps)
    sub_ring = session.declare_subscriber(
        "sensor/**", zenoh.handlers.RingChannel(10)
    )

    # Blocking receive
    sample = sub_fifo.recv()
    print(sample.payload.to_string())

    # Non-blocking receive (returns None if empty)
    sample = sub_ring.try_recv()
    if sample:
        print(sample.payload.to_string())

    # Iterate (blocks until undeclared)
    for sample in sub_default:
        print(f"{sample.key_expr}: {sample.payload.to_string()}")
```

--------------------------------

### Logging Initialization

Source: https://context7.com/eclipse-zenoh/zenoh-python/llms.txt

Initializes Zenoh's internal Rust logger. The log level can be controlled via the `RUST_LOG` environment variable or a fallback default.

```APIDOC
## Logging 

### Description
Initializes Zenoh's internal Rust logger. The log level can be controlled via the `RUST_LOG` environment variable or a fallback default.

### Usage
```python
import zenoh

# Use RUST_LOG env var; fall back to "error" if not set
zenoh.init_log_from_env_or("error")

# Use RUST_LOG env var only; no-op if RUST_LOG is not set
zenoh.try_init_log_from_env()

# Common log levels: "trace", "debug", "info", "warn", "error"
# RUST_LOG=debug python my_app.py  — enables debug logging
with zenoh.open(zenoh.Config()) as session:
    session.put("demo/hello", "world")
```
```

--------------------------------

### Using Schema with Zenoh Encoding

Source: https://github.com/eclipse-zenoh/zenoh-python/blob/main/docs/concepts.rst

Demonstrates how to associate a schema with a Zenoh encoding using the with_schema method. This allows for more structured data interpretation.

```python
import zenoh

conf = zenoh.Config()

# Using the schema
# ~~~~~~~~~~~~~~~~

# Create an Encoding with a schema
encoding_with_schema = zenoh.Encoding.APPLICATION_JSON.with_schema("http://example.com/schema.json")
print(f"Encoding with schema: {encoding_with_schema}")

# Accessing the schema
print(f"Schema: {encoding_with_schema.schema}")

# Convert Encoding to string
print(f"Encoding to string: {str(encoding_with_schema)}")
```

--------------------------------

### Declare and Use a Publisher with Session.declare_publisher

Source: https://context7.com/eclipse-zenoh/zenoh-python/llms.txt

Declare a reusable `Publisher` for efficient repeated publishing. Use `publisher.put()` to send data and `publisher.delete()` to signal removal.

```python
import zenoh
import time

with zenoh.open(zenoh.Config()) as session:
    pub = session.declare_publisher(
        "demo/example/sensor",
        encoding=zenoh.Encoding.TEXT_PLAIN,
        congestion_control=zenoh.CongestionControl.BLOCK,
        priority=zenoh.Priority.REAL_TIME,
    )

    # Publish a sequence of values
    for i in range(5):
        value = f"[{i:4d}] Sensor reading: {20.0 + i * 0.5:.1f}°C"
        print(f"Publishing: {value}")
        pub.put(value)
        time.sleep(1.0)

    # Signal that the key no longer has an associated value
    pub.delete()
```

--------------------------------

### Handle Raw Payloads with zenoh.ZBytes

Source: https://context7.com/eclipse-zenoh/zenoh-python/llms.txt

Zenoh uses `ZBytes` as the universal payload type for all data. It wraps raw bytes or strings. Use `.to_bytes()` or `.to_string()` to extract the data. It supports round-trips for JSON and can be used with Protobuf if installed.

```python
import zenoh
import json

# From raw bytes
b = zenoh.ZBytes(b"\x00\x01\x02\x03")
assert b.to_bytes() == b"\x00\x01\x02\x03"

# From string
s = zenoh.ZBytes("Hello Zenoh!")
assert s.to_string() == "Hello Zenoh!"

# JSON payload round-trip
data = {"sensor": "temp", "value": 22.5, "unit": "celsius"}
payload = zenoh.ZBytes(json.dumps(data))
recovered = json.loads(payload.to_string())
assert recovered == data

# Protobuf (if protobuf is installed)
# import entity_pb2
# entity = entity_pb2.Entity(id=1, name="Robot")
# payload = zenoh.ZBytes(entity.SerializeToString())
# recovered = entity_pb2.Entity(); recovered.ParseFromString(payload.to_bytes())
```

--------------------------------

### Simple Callback for Zenoh Data in Python

Source: https://github.com/eclipse-zenoh/zenoh-python/blob/main/docs/concepts.rst

Illustrates handling incoming Zenoh samples using a simple callback function. This method runs the subscriber in background mode, allowing the main program to continue without managing the subscriber object's lifetime.

```python
import zenoh

def on_sample(sample):
    print(f"Received: {sample.payload} on {sample.source}")

session = zenoh.open()

# Declare a subscriber with a simple callback handler
session.declare_subscriber("my/topic", on_sample)

# Keep the session alive to receive notifications
input("Press Enter to stop subscriber...")

session.close()
```

--------------------------------

### Scouting for Zenoh Nodes

Source: https://github.com/eclipse-zenoh/zenoh-python/blob/main/docs/concepts.rst

Shows how to use the zenoh.scout function to discover Zenoh nodes on the network. This returns a Scout object that yields Hello messages for each discovered node.

```python
import zenoh

conf = zenoh.Config()

# Scouting for Zenoh nodes
# ~~~~~~~~~~~~~~~~~~~~~~~~~

# Start scouting
sc = zenoh.scout(conf)
print("Scouting for Zenoh nodes...")

# Print information about discovered nodes
for hello in sc.take_all():
    print(f"- Node: {hello.zid} ({hello.whatami}) at {hello.locators}")

# Stop scouting
sc.stop()
```

--------------------------------

### Use Session Methods Directly for Publishing

Source: https://github.com/eclipse-zenoh/zenoh-python/blob/main/docs/concepts.rst

Demonstrates publishing data directly using session methods like put and delete, bypassing explicit publisher declaration.

```python
import zenoh
import time

def main():
    conf = zenoh.Config()
    # You can optionally configure the Zenoh session here
    # conf.insert_from_file('zenoh_config.json')

    z = zenoh.open(conf)
    print("Publishing data directly from session...")

    # Publish data directly using the session
    for i in range(5):
        z.put("demo/cpu/load", f"CPU Load {i}".encode(   ))
        time.sleep(0.1)

    # Delete the key expression directly using the session
    print("Deleting key expression directly from session...")
    z.delete("demo/cpu/load")

    # Close the Zenoh session
    z.close()

if __name__ == '__in_main__':
    main()

```

--------------------------------

### Build Zenoh Python Wheel

Source: https://github.com/eclipse-zenoh/zenoh-python/blob/main/README.md

Builds a release version of the zenoh-python wheel package using the 'maturin' tool. This command should be run after ensuring 'maturin' is in the PATH.

```bash
maturin build --release
```

--------------------------------

### Publish Resource with Payload

Source: https://github.com/eclipse-zenoh/zenoh-python/blob/main/examples/README.md

Declares a resource with a path and publisher, then sends a payload. Data is received by subscribers like z_sub and z_storage. Can be run with default or specified key/payload.

```bash
python3 z_pub.py
```

```bash
python3 z_pub.py -k demo/example/test -p 'Hello World'
```

--------------------------------

### Zenoh Liveliness API Usage

Source: https://context7.com/eclipse-zenoh/zenoh-python/llms.txt

Demonstrates using the `Session.liveliness` API to announce node presence, query alive nodes, and subscribe to liveness changes. Includes declaring a token, querying, and subscribing to PUT/DELETE events.

```python
import zenoh
import time, threading

with zenoh.open(zenoh.Config()) as session:
    lv = session.liveliness()

    # Announce this node's presence
    token = lv.declare_token("nodes/robot_arm_01")
    # Token is alive as long as `token` object exists and session is open

    # Query which nodes are currently alive
    for reply in lv.get("nodes/**", timeout=2.0):
        if reply.ok:
            print(f"Alive: {reply.ok.key_expr}")

    # Subscribe to liveness changes (history=True: get existing tokens immediately)
    with lv.declare_subscriber("nodes/**", history=True) as sub:
        for sample in sub:
            if sample.kind == zenoh.SampleKind.PUT:
                print(f"Node appeared: {sample.key_expr}")
            elif sample.kind == zenoh.SampleKind.DELETE:
                print(f"Node disappeared: {sample.key_expr}")
            break

    # Explicitly undeclare to trigger DELETE event for subscribers
    token.undeclare()
```

--------------------------------

### zenoh.Parameters / zenoh.Selector

Source: https://context7.com/eclipse-zenoh/zenoh-python/llms.txt

Constructs query parameter sets from dictionaries and combines key expressions with parameters to form selectors, similar to URLs with query strings.

```APIDOC
## `zenoh.Parameters` / `zenoh.Selector` — Query Parameters

`Parameters` constructs query parameter sets from dictionaries. `Selector` combines a key expression with parameters, equivalent to a URL with a query string (e.g., `key/expr?a=1;b=2`).

```python
import zenoh

with zenoh.open(zenoh.Config()) as session:
    # Build parameters from dict
    params = zenoh.Parameters({"day": "2024-01-15", "format": "fahrenheit"})

    # Combine key expression and parameters into a selector
    selector = zenoh.Selector("room/temperature/history", params)
    # Equivalent: selector = "room/temperature/history?day=2024-01-15;format=fahrenheit"

    for reply in session.get(selector):
        if reply.ok:
            print(reply.ok.payload.to_string())

    # On the queryable side, access parameters via query.parameters
    with session.declare_queryable("room/temperature/history") as qbl:
        for query in qbl:
            day = query.parameters.get("day", "unknown")
            fmt = query.parameters.get("format", "celsius")
            query.reply("room/temperature/history", f"22.5 on {day} ({fmt})")
            break
```
```

--------------------------------

### Construct Selector from Dictionary

Source: https://github.com/eclipse-zenoh/zenoh-python/blob/main/docs/concepts.rst

Shows how to construct a Zenoh Selector object from a Python dictionary for parameters, which can then be used in queries.

```python
import zenoh

def main():
    conf = zenoh.Config()
    # You can optionally configure the Zenoh session here
    # conf.insert_from_file('zenoh_config.json')

    z = zenoh.open(conf)

    # Define parameters as a dictionary
    params_dict = {
        "param1": "value1",
        "param2": "value2"
    }

    # Create Zenoh Parameters object
    params = zenoh.Parameters(params_dict)

    # Create a Selector combining key expression and parameters
    selector = zenoh.Selector("demo/cpu/load", params)

    print(f"Constructed Selector: {selector}")

    # You can now use this selector with z.get() or querier.query()
    # For example:
    # replies = z.get(selector)
    # ... process replies ...

    # Close the Zenoh session
    z.close()

if __name__ == '__in_main__':
    main()

```

--------------------------------

### Update Rust toolchain

Source: https://github.com/eclipse-zenoh/zenoh-python/blob/main/README.md

Ensure your Rust toolchain is up-to-date by running the 'rustup update' command. This is a prerequisite for building Zenoh Python from source.

```bash
rustup update
```

--------------------------------

### Subscribe to Keys with Zenoh Python

Source: https://github.com/eclipse-zenoh/zenoh-python/blob/main/docs/quickstart.rst

Subscribe to a set of keys using a Key Expression. This snippet demonstrates how to receive data published to matching keys.

```python
import zenoh

conf = zenoh.Config()

# You can optionally configure Zenoh here
# For example, to enable the REST API:
# conf.insert("rest.addr", "0.0.0.0:8080")

# Start Zenoh session
keyexpr = zenoh.KeyExpr("example/*")

with zenoh.open(conf) as session:
    # Define a callback function for received samples
    def listener(sample):
        print(f"Received: key='{sample.key}' value='{sample.value}'")

    # Subscribe to the key expression
    sub = session.declare_subscriber(keyexpr, listener)

    print(f"Subscribed to '{keyexpr}'. Press Ctrl+C to stop.")
    # Keep the session alive to receive samples
    try:
        while True:
            zenoh.sleep(1000) # sleep for 1 second
    except KeyboardInterrupt:
        print("Unsubscribing...")
        sub.undeclare()

print("Done.")
```

--------------------------------

### Initialize Zenoh Logging in Python

Source: https://context7.com/eclipse-zenoh/zenoh-python/llms.txt

Initializes Zenoh's internal logger using environment variables or a fallback default. The log level can be controlled via the RUST_LOG environment variable.

```python
import zenoh

# Use RUST_LOG env var; fall back to "error" if not set
zenoh.init_log_from_env_or("error")

# Use RUST_LOG env var only; no-op if RUST_LOG is not set
zenoh.try_init_log_from_env()

# Common log levels: "trace", "debug", "info", "warn", "error"
# RUST_LOG=debug python my_app.py  — enables debug logging
with zenoh.open(zenoh.Config()) as session:
    session.put("demo/hello", "world")
```

--------------------------------

### Open a Zenoh Session

Source: https://context7.com/eclipse-zenoh/zenoh-python/llms.txt

Opens a Zenoh session using a `with` statement for automatic closing or explicit `try/finally` blocks. Ensure sessions are closed to prevent process hangs.

```python
import zenoh

# Recommended: context manager ensures automatic close
with zenoh.open(zenoh.Config()) as session:
    session.put("demo/example/hello", "Hello World!")
# session is automatically closed here

# Alternative: explicit close with try/finally
session = zenoh.open(zenoh.Config())
try:
    session.put("demo/example/hello", "Hello World!")
finally:
    session.close()
```

--------------------------------

### zenoh.ext.z_serialize / zenoh.ext.z_deserialize

Source: https://context7.com/eclipse-zenoh/zenoh-python/llms.txt

Provides functions for serializing and deserializing Python values to/from `ZBytes` using the `zenoh.ext` module, supporting various data types and inter-language compatibility.

```APIDOC
## `zenoh.ext.z_serialize` / `zenoh.ext.z_deserialize` — Typed Serialization

Serialize and deserialize Python values to/from `ZBytes` using the `zenoh.ext` module. Supports primitives, lists, dicts, tuples, and numeric types with inter-language compatibility (works with Rust, C++, etc.).

```python
import zenoh
from zenoh.ext import (
    z_serialize, z_deserialize,
    UInt8, UInt16, UInt32, UInt64,
    Int32, Float32, Float64,
)

# Primitive numeric types
payload = z_serialize(UInt32(42))
value = z_deserialize(UInt32, payload)
assert value == UInt32(42)

# Floating point
payload = z_serialize(Float64(3.14159))
assert z_deserialize(Float64, payload) == Float64(3.14159)

# Lists (all items same type)
payload = z_serialize([1.0, 2.5, 3.0])
result = z_deserialize(list[float], payload)
assert result == [1.0, 2.5, 3.0]

# Dicts
payload = z_serialize({"temperature": 22.5, "humidity": 60.0})
result = z_deserialize(dict[str, float], payload)
assert result == {"temperature": 22.5, "humidity": 60.0}

# Tuples
payload = z_serialize((42, "sensor_A", 3.14))
result = z_deserialize(tuple[int, str, float], payload)
assert result == (42, "sensor_A", 3.14)

# Use in pub/sub
with zenoh.open(zenoh.Config()) as session:
    session.put("sensor/reading", z_serialize({"temp": 22.5, "hum": 60.0}))
    for reply in session.get("sensor/reading"):
        if reply.ok:
            data = z_deserialize(dict[str, float], reply.ok.payload)
            print(data)  # {"temp": 22.5, "hum": 60.0}
```
```

--------------------------------

### Advanced Callback Handler with Cleanup in Python

Source: https://github.com/eclipse-zenoh/zenoh-python/blob/main/docs/concepts.rst

Shows how to use zenoh.handlers.Callback for more advanced callback scenarios, including custom cleanup logic. This provides a structured way to manage resources associated with callbacks.

```python
import zenoh

class MyCallbackHandler:
    def __init__(self):
        print("Callback handler initialized")

    def __call__(self, sample):
        print(f"Received: {sample.payload} on {sample.source}")

    def close(self):
        print("Callback handler closed")

session = zenoh.open()

# Declare a subscriber with an advanced callback handler
callback_handler = MyCallbackHandler()
subscriber = session.declare_subscriber("my/topic", handler=zenoh.handlers.Callback(callback_handler, callback_handler.close))

# Keep the subscriber alive to receive notifications
input("Press Enter to stop subscriber...")

session.close()
```

--------------------------------

### Subscribe to Zenoh Data

Source: https://github.com/eclipse-zenoh/zenoh-python/blob/main/examples/README.md

Creates a subscriber for a given key expression. It receives notifications for matching puts. Can be run with default or a specific key expression.

```bash
python3 z_sub.py
```

```bash
python3 z_sub.py -k 'demo/**'
```

--------------------------------

### Publish Key/Value Pair with Zenoh Python

Source: https://github.com/eclipse-zenoh/zenoh-python/blob/main/docs/quickstart.rst

Use this snippet to publish a key/value pair onto the Zenoh network. Ensure Zenoh is running and accessible.

```python
import zenoh

conf = zenoh.Config()

# You can optionally configure Zenoh here
# For example, to enable the REST API:
# conf.insert("rest.addr", "0.0.0.0:8080")

# Start Zenoh session
keyexpr = zenoh.KeyExpr("example/resource")

with zenoh.open(conf) as session:
    # Publish a key/value pair
    print(f"Putting key='{keyexpr}' value='Hello Zenoh!'")
    session.put(keyexpr, "Hello Zenoh!")

print("Done.")
```

--------------------------------

### Declare a Queryable with Session.declare_queryable

Source: https://context7.com/eclipse-zenoh/zenoh-python/llms.txt

Create a `Queryable` to handle incoming `Query` requests. Replies are sent using `query.reply()`, `query.reply_del()`, or `query.reply_err()`.

```python
import zenoh

temperature_db = {
    "2024-01-15": "22.5°C",
    "2024-01-16": "23.1°C",
    "2024-01-17": "21.8°C",
}

with zenoh.open(zenoh.Config()) as session:
    with session.declare_queryable("room/temperature/history") as qbl:
        print("Queryable ready, waiting for queries...")
        for query in qbl:
            print(f"Received query: {query.selector}")
            params = query.selector.parameters
            if "day" not in params:
                query.reply_err("missing 'day' parameter")
            elif params["day"] in temperature_db:
                query.reply(
                    "room/temperature/history",
                    temperature_db[params["day"]],
                )
            else:
                # Use reply_del to signal the key exists but has no value
                query.reply_del("room/temperature/history")
```

--------------------------------

### Declare Matching Listener for Publisher in Python

Source: https://github.com/eclipse-zenoh/zenoh-python/blob/main/docs/concepts.rst

Declares a matching listener for a Zenoh publisher. This allows the publisher to be notified when subscribers express interest in its data, enabling resource optimization by not sending data if no one is listening.

```python
import zenoh

def on_matching_status(status):
    print(f"Matching status changed: {status}")

session = zenoh.open()

# Declare a matching listener for a publisher
publisher = session.declare_publisher("my/topic")
publisher.declare_matching_listener(on_matching_status)

# Keep the publisher alive to receive notifications
input("Press Enter to stop publisher...")

session.close()
```

--------------------------------

### Configure Zenoh Session

Source: https://context7.com/eclipse-zenoh/zenoh-python/llms.txt

Configure a Zenoh session using default settings, from a JSON file, or programmatically. Displays session information upon connection.

```python
import zenoh

# Default config (peer mode, multicast scouting)
config = zenoh.Config()

# Load from a JSON config file
config = zenoh.Config.from_file("zenoh_config.json")

# Open session with the config
with zenoh.open(config) as session:
    info = session.info
    print(f"Session ZID: {info.zid()}")
    print(f"Connected routers: {info.routers_zid()}")
    print(f"Connected peers: {info.peers_zid()}")
```

--------------------------------

### Put Data into Zenoh

Source: https://github.com/eclipse-zenoh/zenoh-python/blob/main/examples/README.md

Publishes a key/payload into Zenoh, making it available to subscribers like z_sub and z_storage. Can be run with default or specified key/payload.

```bash
python3 z_put.py
```

```bash
python3 z_put.py -k demo/example/test -p 'Hello World'
```

--------------------------------

### Session.info

Source: https://context7.com/eclipse-zenoh/zenoh-python/llms.txt

Retrieves SessionInfo object with Zenoh ID, connected routers, peers, transports, and network links. Also provides listeners for transport and link changes.

```APIDOC
## Session.info 

### Description
Returns a `SessionInfo` object exposing the session's Zenoh ID, connected routers, connected peers, current transports, and network links. Also provides event listeners for transport and link changes.

### Usage
```python
import zenoh
import time

with zenoh.open(zenoh.Config()) as session:
    info = session.info

    print(f"My ZID:    {info.zid()}")
    print(f"Routers:   {info.routers_zid()}")
    print(f"Peers:     {info.peers_zid()}")

    for transport in info.transports():
        print(f"Transport: {transport}")
    for link in info.links():
        print(f"Link: {link}")

    # Listen for live transport/link events
    transport_listener = info.declare_transport_events_listener(history=False)
    link_listener = info.declare_link_events_listener(history=False)

    for _ in range(30):  # Poll for 3 seconds
        while (event := transport_listener.try_recv()) is not None:
            print(f"Transport event: {event}")
        while (event := link_listener.try_recv()) is not None:
            print(f"Link event: {event}")
        time.sleep(0.1)

    transport_listener.undeclare()
    link_listener.undeclare()
```
```

--------------------------------

### Scout Zenoh Nodes on the Network

Source: https://context7.com/eclipse-zenoh/zenoh-python/llms.txt

Use `zenoh.scout` to discover Zenoh nodes (peers, routers, clients) on the local network. The scout yields `Hello` messages containing node ID, type, and locators. You can specify the types of nodes to scout for and provide a custom configuration.

```python
import zenoh
import threading

# Scout for peers and routers for 2 seconds
scout = zenoh.scout(what="peer|router")
threading.Timer(2.0, lambda: scout.stop()).start()

for hello in scout:
    print(f"Discovered node:")
    print(f"  ZID:      {hello.zid}")
    print(f"  WhatAmI:  {hello.whatami}")   # "peer", "router", or "client"
    print(f"  Locators: {hello.locators}")  # e.g. ["tcp/192.168.1.5:7447"]

# Scout with a custom config
config = zenoh.Config()
scout2 = zenoh.scout(what="router", config=config)
threading.Timer(1.0, scout2.stop).start()
for hello in scout2:
    print(f"Router found: {hello.zid} at {hello.locators}")
```

--------------------------------

### Declare Advanced Publisher with Caching

Source: https://context7.com/eclipse-zenoh/zenoh-python/llms.txt

Create an advanced publisher using `declare_advanced_publisher` that caches recent publications. This allows late-joining subscribers to retrieve missed samples. Configure cache size and sample miss detection with heartbeats.

```python
import zenoh
from zenoh.ext import (
    declare_advanced_publisher,
    CacheConfig,
    MissDetectionConfig
)

with zenoh.open(zenoh.Config()) as session:
    pub = declare_advanced_publisher(
        session,
        "demo/example/sensor",
        cache=CacheConfig(max_samples=10),        # Cache last 10 publications
        sample_miss_detection=MissDetectionConfig(heartbeat=5),  # Heartbeat every 5s
        publisher_detection=True,                 # Announce publisher presence
    )

    for i in range(20):
        pub.put(f"reading_{i}: {20.0 + i * 0.1:.1f}°C")
```

--------------------------------

### Declare and Use a Subscriber with Session.declare_subscriber

Source: https://context7.com/eclipse-zenoh/zenoh-python/llms.txt

Create a `Subscriber` to receive `Sample` objects matching a key expression. Supports wildcards and can operate in channel-based or callback-based modes.

```python
import zenoh

with zenoh.open(zenoh.Config()) as session:
    # Channel-based: iterate over received samples
    with session.declare_subscriber("demo/example/**") as sub:
        for sample in sub:
            ke = sample.key_expr
            value = sample.payload.to_string()
            kind = sample.kind  # SampleKind.PUT or SampleKind.DELETE
            ts = sample.timestamp   # optional Timestamp
            att = sample.attachment  # optional ZBytes attachment
            print(f"[{kind}] {ke} => {value}")
            if kind == zenoh.SampleKind.DELETE:
                print(f"  Key '{ke}' was deleted")

    # Callback-based: runs in background; subscriber stays alive
    def on_data(sample: zenoh.Sample):
        print(f">> {sample.payload.to_string()}")

    session.declare_subscriber("sensor/temperature", on_data)
    # Subscriber remains active while the session is alive
```

--------------------------------

### Use a Querier to Request Data

Source: https://github.com/eclipse-zenoh/zenoh-python/blob/main/docs/concepts.rst

Demonstrates using a Querier object to request data from queryables. This provides a more object-oriented way to handle queries.

```python
import zenoh
import time

def main():
    conf = zenoh.Config()
    # You can optionally configure the Zenoh session here
    # conf.insert_from_file('zenoh_config.json')

    z = zenoh.open(conf)
    print("Using a Querier to request data...")

    # Create a querier for the 'demo/cpu/load' key expression
    querier = z.declare_querier("demo/cpu/load")

    # Send a query and get replies
    replies = querier.query()

    print("Received replies:")
    for reply in replies:
        if isinstance(reply, zenoh.Sample):
            print(f"  Sample: {reply.kind} {reply.data}")
        elif isinstance(reply, zenoh.ReplyError):
            print(f"  Error: {reply.code} {reply.message}")

    # Close the Zenoh session
    z.close()

if __name__ == '__in_main__':
    main()

```

--------------------------------

### Upgrade pip

Source: https://github.com/eclipse-zenoh/zenoh-python/blob/main/README.md

If necessary, upgrade your pip version to at least 19.3.1 to ensure full support for PEP 517, which is required for building Zenoh Python from source.

```bash
sudo pip install --upgrade pip
```