Metadata-Version: 2.4
Name: rainbow-rb-zenoh
Version: 0.10.4
Summary: Rainbow Zenoh Python (C++ core via pybind11)
Author-email: Derek <dfd1123@rainbow-robotics.com>
Requires-Python: <3.14,>=3.11
Description-Content-Type: text/markdown
Requires-Dist: psutil<8.0.0,>=7.0.0
Requires-Dist: flatbuffers<26.0.0,>=25.2.10
Requires-Dist: fastapi>=0.116.1
Requires-Dist: rainbow-rb-utils==0.10.4
Requires-Dist: rainbow-rb-log==0.10.4

# rb_zenoh

`rb_zenoh` is a shared Zenoh-based communication package.
Its main features are `publish/subscribe`, `query/queryable`, and `ZenohRouter` decorator-based routing.

## Installation/Dependencies

- Python: `>=3.12,<3.13`
- Package dependencies: `flatbuffers`, `psutil`, `rb_utils`, `rb_modules`
- The `_core` native extension is not built during package installation. Build the
  `_core*.so`/`_core*.pyd` for the required platforms first, then include it in the package.

```bash
# For the current host
bash scripts/backend/build_cpp_core.sh zenoh

# For Docker/Linux
RB_CPP_CORE_TARGETS="linux/amd64 linux/arm64" bash scripts/backend/build_cpp_core.sh zenoh

```

The distributed `rainbow-rb-zenoh` is not built as a fat wheel; it is split into per-platform wheels.
Each wheel contains only the single `_core` artifact for its platform.

## Core Objects

- `ZenohClient`: low-level communication API
- `ZenohRouter`: decorator-based router
- `SubscribeOptions`: options controlling subscribe behavior

## 1) publish

```python
from rb_zenoh.client import ZenohClient

client = ZenohClient()
client.publish("muscat/sample/topic", payload={"ok": True})
```

Publishing a FlatBuffer request:

```python
from rb_flat_buffers.IPC.Request_MotionPause import Request_MotionPauseT

req = Request_MotionPauseT()
client.publish(
    "C500920/call_pause",
    flatbuffer_req_obj=req,
    flatbuffer_buf_size=64,
)
```

## 2) subscribe

```python
from rb_zenoh.client import ZenohClient
from rb_zenoh.schema import SubscribeOptions

client = ZenohClient()

async def on_msg(*, topic, mv, obj_payload, dict_payload, attachment):
    print(topic, obj_payload, dict_payload, attachment)

client.subscribe(
    "muscat/sample/topic",
    on_msg,
    options=SubscribeOptions(dispatch="immediate"),
)
```

Callback signature:
- `topic`: topic string
- `mv`: raw payload (memoryview)
- `obj_payload`: object parsed via `flatbuffer_obj_t=<FlatBuffer T class>` (`T | None`)
- `dict_payload`: dict conversion result, regardless of whether `flatbuffer_obj_t` is set (`dict | None`)
- `attachment`: `sender`, `sender_id` information

## 3) queryable

```python
def on_query(req=None, params=None):
    return {"status": "ok"}

client.queryable("muscat/sample/query", on_query)
```

Callback arguments:
- `req`: query payload body
  - Injected only when `flatbuffer_req_t`/`flatbuffer_req_T_class` is specified.
  - A FlatBuffer `...T` object parsed internally via `InitFromPackedBuf(...)`.
  - If `flatbuffer_req_t` is specified but the request has no payload, an error is raised.
- `params`: query parameters
  - The internal `q.parameters` value is injected as a `dict[str, str]`.
  - Used to pass filters/options separately from the payload (`req`).

Notes:
- Arguments are auto-injected only when the callback parameters are named exactly `req` and `params`.
- Parameters declared with other names such as `request` or `query_params` are not injected.
- If neither is declared, the callback is invoked with no arguments.

## 4) query_one / query_all

### query_one

```python
res = client.query_one("muscat/program/state", timeout=0.3)
print(res.get("dict_payload"))
```

- Returns only the first reply
- Raises `ZenohNoReply` if there is no reply

### query_all

```python
res_list = client.query_all("*/health", timeout=0.5)
for item in res_list:
    print(item.get("key"), item.get("dict_payload"))
```

- Returns all replies received within the timeout as a list
- Returns an empty list if there are no replies

### Which to use when

- Calling a single target: `query_one`
- Collecting from wildcards/multiple services: `query_all`

## FlatBuffer Parameter Details

### `flatbuffer_req_obj`

- Meaning: the FlatBuffer object to send as the request (an object supporting `Pack()`)
- Used in: `publish`, `query_one`, `query_all`
- Internal behavior: packed with a builder, then sent as a bytes payload

### `flatbuffer_buf_size`

- Meaning: initial buffer size of the FlatBuffer serialization builder
- Used in:
  - Request serialization: `publish`, `query_one`, `query_all`
  - queryable response serialization: `queryable(..., flatbuffer_res_buf_size=...)`
- Serialization may fail if it is too small
- Recommended: start from 64/128/256/512 and increase to fit the payload size

### `flatbuffer_res_T_class`

- Meaning: specifies which FlatBuffer type to parse the response payload as
- Used in: `query_one`, `query_all`
- Type: must be a **FlatBuffer Object API class (T class)**
  - e.g. `Response_FunctionsT`, `State_CoreT`
  - Usually a class whose name ends in `...T`
  - Parsed internally via `InitFromPackedBuf(...)`
- Result:
  - If specified: parsed object in `obj_payload`
  - If not specified: rely mainly on `dict_payload`/raw payload

Example:

```python
from rb_flat_buffers.IPC.Request_MotionPause import Request_MotionPauseT
from rb_flat_buffers.IPC.Response_Functions import Response_FunctionsT

res = client.query_one(
    "C500920/call_pause",
    flatbuffer_req_obj=Request_MotionPauseT(),
    flatbuffer_res_T_class=Response_FunctionsT,
    flatbuffer_buf_size=64,
    timeout=0.3,
)
obj = res.get("obj_payload")
```

## 5) Using ZenohRouter

```python
from rb_zenoh.router import ZenohRouter

router = ZenohRouter(prefix="muscat/common")

@router.subscribe("health")
async def on_health(*, topic, obj_payload, dict_payload, attachment):
    pass

@router.queryable("echo")
async def on_echo(params=None):
    return {"ok": True}
```

Lifecycle:
- On startup: `await router.startup()`
- On shutdown: `await router.shutdown()`

## Exceptions/Notes

- `query_one` no-reply: `ZenohNoReply`
- Transport reconnection errors: `ZenohTransportError`
- `ZenohClient` is a per-process singleton, so avoid unnecessary frequent `close()` calls
