UV Project to test zenoh example in ardupilot

main.py

import threading
import time

import zenoh


def listener(sample):
    print(f"[SUB] {sample.key_expr} => {sample.payload.to_string()}")


def publisher_loop(session):
    publisher = session.declare_publisher("ardupilot/receive")
    while True:
        publisher.put("pong")
        print("[PUB] ardupilot/receive => pong")
        time.sleep(1)


def main():
    config = zenoh.Config()
    with zenoh.open(config) as session:
        sub = session.declare_subscriber("ardupilot/output", listener)

        pub_thread = threading.Thread(target=publisher_loop, args=(session,), daemon=True)
        pub_thread.start()

        print("Running... Press Ctrl+C to stop.")
        try:
            while True:
                time.sleep(1)
        except KeyboardInterrupt:
            print("Shutting down.")
        finally:
            sub.undeclare()


if __name__ == "__main__":
    main()

pyproject.toml

[project]
name = "Zenoh-ArduPino-S2"
version = "0.1.0"
description = "Add your description here"
readme = "README.md"
requires-python = ">=3.11"
dependencies = [
    "eclipse-zenoh==1.7.2",
]