rb_flat_buffers

rb_flat_buffers is a package of Python FlatBuffer code generated from schemas/*.fbs.

Summary

• Source schemas: /schemas/**/*.fbs
• Generated output 1: backend/packages/rb_flat_buffers/src/rb_flat_buffers/** (FlatBuffer Python generated code)
• Generated output 2: backend/packages/rb_schemas/src/rb_schemas/fbs_models/** (Pydantic models)

In other words, running make backend.flatc:

1. Generates the FlatBuffer Python code into rb_flat_buffers, and
2. Generates Pydantic models based on the same schemas into rb_schemas.

How to Generate

From the project root:

make backend.flatc

The backend.flatc target in backend.mk internally runs the following in order.

1. scripts/backend/generate_flatbuffers.py

• Generates code using flatc --python --gen-object-api --gen-all
• Output: backend/packages/rb_flat_buffers/src/rb_flat_buffers

2. backend/packages/rb_flat_buffers/scripts/patch_imports.py

• Fixes generated import paths and adds missing imports

3. scripts/backend/fbs_to_pydantic.py

• Converts .fbs files into Pydantic models
• Output: backend/packages/rb_schemas/src/rb_schemas/fbs_models

When to Run It

You must re-run make backend.flatc in the following cases.

• schemas/*.fbs is modified
• New .fbs files are added or deleted
• FlatBuffer types/fields change (table/struct/enum/union)

rb_flat_buffers Usage Example

Example: sending a request object + receiving a FlatBuffer response with populated values (returnValue)

# server/queryable side
from rb_flat_buffers.IPC.Request_Move_SmoothJogStop import Request_Move_SmoothJogStopT
from rb_flat_buffers.IPC.Response_Functions import Response_FunctionsT
from rb_zenoh.router import ZenohRouter

router = ZenohRouter()

@router.queryable(
    "C500920/call_smoothjog_stop",
    flatbuffer_req_t=Request_Move_SmoothJogStopT,
    flatbuffer_res_buf_size=64,
)
def on_call_pause(req, params=None):
    print(req.stoptime)  # values passed in the request are available
    res = Response_FunctionsT()
    res.returnValue = 123  # contains an actual value, not an empty payload
    return res

# client/query side
from rb_flat_buffers.IPC.Request_Move_SmoothJogStop import Request_Move_SmoothJogStopT
from rb_flat_buffers.IPC.Response_Functions import Response_FunctionsT
from rb_zenoh.client import ZenohClient

client = ZenohClient()
req = Request_Move_SmoothJogStopT()
req.stoptime = 0.7

res = client.query_one(
    "C500920/call_smoothjog_stop",
    flatbuffer_req_obj=req,
    flatbuffer_res_T_class=Response_FunctionsT,
    flatbuffer_buf_size=64,
)

obj = res.get("obj_payload")
print(obj.returnValue)  # 123

rb_schemas (Pydantic) Usage Example

After validating the obj_payload of a query_one response with Pydantic, you can choose the desired key case in model_dump.

from rb_flat_buffers.IPC.Response_Functions import Response_FunctionsT
from rb_schemas.fbs_models.nexus.v1.deploy_models import Response_Deploy_UploadPD
from rb_utils.parser import t_to_dict
from rb_zenoh.client import ZenohClient

client = ZenohClient()

res = client.query_one(
    "deploy/call_upload",
    flatbuffer_res_T_class=Response_FunctionsT,
    timeout=0.3,
)

obj_payload = res.get("obj_payload")

# 1) Convert FlatBuffer T object -> dict
raw_dict = t_to_dict(obj_payload) if obj_payload is not None else {}

# 2) Pydantic validate
pd_obj = Response_Deploy_UploadPD.model_validate(raw_dict)

# 3) Choose key case
snake_case = pd_obj.model_dump(by_alias=False)  # ex) return_value
camel_case = pd_obj.model_dump(by_alias=True)   # ex) returnValue

Notes

• The package name is rb_flat_buffers (not rb_flat_buffer).
• Manually editing generated files is not recommended.
• make backend.flatc fails if flatc is not on the PATH.