DORA Python API 开发实战指南Node、Operator 与 DataflowBuilder 全解析【免费下载链接】doraDORA (Dataflow-Oriented Robotic Architecture) is middleware designed to streamline and simplify the creation of AI-based robotic applications. It offers low latency, composable, and distributed dataflow capabilities. Applications are modeled as directed graphs, also referred to as pipelines.项目地址: https://gitcode.com/GitHub_Trending/do/doraDORADataflow-Oriented Robotic Architecture是一套面向 AI 机器人应用的低延迟、可组合、分布式数据流中间件应用被建模为有向图pipeline。本文基于仓库 docs/api-python.md 编写系统讲解 DORA 的 Python 编程接口如何用dora.Node编写自定义节点、如何在 dora runtime 进程内运行Operator、如何用DataflowBuilder在 Python 中程序化生成数据流 YAML并延伸到 CUDA 张量零拷贝传输、服务/动作/流式通信模式与结构化日志。读完本文你可以从零编写一个可运行的 Python 节点并将其接入 dora daemon 管理的数据流图。安装与运行环境要求DORA 的 Python 客户端以dora-rs包发布安装命令pip install dora-rs解释器兼容性CPython 3.11 及以上。官方 wheel 以abi3-py311构建单个 wheel 即可在所有后续 CPython 版本上运行无需等待 dora 发版——3.11 是 dora 1.x 的下限floor并非锁定版本。不支持的构建free-threaded 构建python3.13t、python3.14t不在 abi3 覆盖范围内也没有为其发布 wheel。完整保证请参阅 Python 版本策略。开发调试本仓库内的 Python 绑定位于 apis/python/node时README 给出了本地构建方式uv venv --seed -p 3.11 uv pip install -e .类型提示需额外生成 stub 文件python generate_stubs.py dora dora/__init__.pyi maturin developNode API自定义节点的核心接口Node是编写自定义节点的首要接口它连接到一个正在运行的数据流dataflow接收输入事件并发送输出from dora import Node从源码结构看Node类由 Rust 通过 pyo3 导出见 apis/python/node/src/lib.rs内部持有DoraNode与一个Events事件流并在构造时自动安装日志桥接见下文“日志”小节。__init__(node_idNone)创建一个新节点并连接正在运行的数据流# 标准用法节点 ID 从 daemon 设置的环境变量读取 node Node() # 动态用法通过显式节点 ID 连接到正在运行的数据流 node Node(node_idmy-dynamic-node)参数node_idstr可选——动态节点的显式 ID。省略时节点从 dora daemon 设置的环境变量读取身份。异常节点无法连接数据流时抛出RuntimeError。从类型定义dora/init.pyi可以看到构造签名还支持daemon_portNode(my-node, daemon_port6789)可用于连接自定义端口上的 daemon。Rust 实现中lib.rs显式传入node_id走DoraNode::init_flexible(node_id)带端口时使用DoraNode::builder()省略时走DoraNode::init_from_env()。接收事件next/drain/try_recv/recv_async/is_emptynext(timeoutNone)—— 从事件流取下一个事件阻塞直到有事件可用或超时event node.next() # 无限期阻塞 event node.next(timeout2.0) # 最多阻塞 2 秒timeoutfloat可选——最大等待秒数。返回事件字典当所有发送方都已关闭或超时到期时返回None。实现上lib.rstimeout会被转换为DurationRust 侧对 NaN、负数或无穷大的超时输入返回干净的ValueError而非 panic见timeout_to_durationlib.rs。drain()—— 不阻塞地取出所有缓冲事件events node.drain() for event in events: print(event[type])返回list[dict]无缓冲事件时返回空列表。try_recv()—— 非阻塞接收返回下一个缓冲事件若有event node.try_recv() if event is not None: print(event[type])返回dict | None。recv_async(timeoutNone)—— 异步接收配合asyncio使用event await node.recv_async() event await node.recv_async(timeout5.0)超时到达时返回错误所有发送方关闭时返回None。注意此方法为实验特性pyo3 的 asyncRust-Python FFI集成仍在开发中。is_empty()—— 检查事件流中是否还有缓冲事件if not node.is_empty(): event node.try_recv()返回bool。迭代支持Node实现了__iter__与__next__lib.rs可以直接用for循环消费事件流for event in node: match event[type]: case INPUT: process(event[value]) case STOP: break迭代器每次迭代调用无超时的next()事件流关闭时产出None从而终止循环。发送输出send_output(output_id, data, metadataNone)在输出通道上发送数据import pyarrow as pa # 发送原始字节 node.send_output(status, bOK) # 发送 Apache Arrow 数组可零拷贝 node.send_output(values, pa.array([1, 2, 3])) # 携带元数据发送 node.send_output(image, pa.array(pixels), {camera_id: front})参数output_idstr——数据流 YAML 中声明的输出名。databytes | pyarrow.Array——负载。简单数据用bytes需要零拷贝共享内存传输时用pyarrow.Array。metadatadict可选——附加到消息的键值对。支持的值类型bool、int、float、str、list[int]、list[float]、list[str]、datetime.datetime。异常data既不是bytes也不是pyarrow.Array时抛出RuntimeError。Rust 侧的分发逻辑send_payloadlib.rs验证PyBytes走send_output_bytes原始字节路径pyarrow 数组走 Arrow 数组路径其余类型直接报错。零拷贝补充send_output_rawPython ≥ 3.11。源码中还提供了可写缓冲区的零拷贝发送路径lib.rs预分配指定字节数的共享内存/对齐堆存储通过 Python buffer 协议memoryview、numpy.frombuffer、struct等填充后再发送with node.send_output_raw(rgb_image, 1920 * 1080 * 3, {width: 1920}) as buf: numpy.frombuffer(buf, numpy.uint8).reshape(1080, 1920, 3)[:] frame # 退出 with 块时自动发送在 Python 3.11 上该方法抛出NotImplementedError并给出升级提示lib.rs。服务、动作与流式通信模式Python 节点与 Rust 使用相同的元数据键约定完整指南见 通信模式。参数是键为字符串的普通 dict。预定义元数据键键说明request_id服务请求/响应关联UUID v7goal_id动作目标标识UUID v7goal_status动作结果状态succeeded、aborted或canceledsession_id流式会话标识segment_id会话内的流式分段整数seq流式分块序号整数fin流式分段最后一个分块boolflush丢弃输入上更旧的排队消息bool服务客户端示例import uuid # 发送带唯一 request_id 的请求 request_id str(uuid.uuid7()) # Python 3.13旧版本用 uuid_utils 或 uuid.uuid4() node.send_output(request, data, {request_id: request_id})服务端示例# 原样透传请求中的元数据含 request_id node.send_output(response, result, event[metadata])动作客户端示例goal_id str(uuid.uuid7()) node.send_output(goal, data, {goal_id: goal_id})流式示例用户中断时刷新下游队列# metadata 是参数名 - 值的扁平 dict没有外层 parameters 键 node.send_output( text, data, metadata{ session_id: session_id, segment_id: 1, seq: 0, fin: False, flush: True, }, )源码还提供send_service_request/send_service_response便捷包装lib.rs前者自动注入 UUID v7request_id并返回该 ID后者校验透传的元数据中包含request_id。日志logging自动桥接与显式节点 APIPython 节点既可以使用内置logging模块推荐也可以使用显式节点 API。Pythonlogging模块自动桥接创建Node()时会自动安装一个 handler把 Pythonlogging模块的路由指向 dora daemon无需任何配置import logging from dora import Node node Node() # 安装日志桥接 logging.info(Sensor initialized) # - 结构化 info 日志 logging.warning(High temperature) # - 结构化 warn 日志 logging.debug(Raw bytes: %s, data) # - 结构化 debug 日志这些日志条目携带完整元数据级别、消息、文件路径、行号并支持 daemon 的min_log_level过滤、send_logs_as路由和dora/logs订阅者。注意不要在创建Node()之前调用logging.basicConfig()。构造函数会设置桥接先调用basicConfig()可能安装冲突的 handler。实现细节lib.rssetup_logging将logging.basicConfig包装为默认使用HostHandler后者调用 Rust 函数host_log把logging.LogRecord转码为tracing::Event按 levelno 映射到 ERROR/WARN/INFO/DEBUG/TRACE从而让 Python 日志进入 dora 的统一结构化日志管道。显式节点 APIlog(level, message, targetNone, fieldsNone)发出带可选 target 和键值字段的结构化日志node.log(info, Processing frame, targetvision) node.log(error, Sensor timeout, fields{sensor: lidar, retry: 3})levelstr——日志级别error、warn、info、debug或trace。messagestr——日志消息。targetstr可选——目标模块或子系统名。fieldsdict[str, str]可选——结构化键值上下文字段。log_error(message)、log_warn(message)、log_info(message)、log_debug(message)、log_trace(message)常用级别的便捷方法各自等价于node.log(level, message)node.log_error(Connection failed) node.log_warn(Temperature elevated) node.log_info(Sensor initialized) node.log_debug(Raw bytes received) node.log_trace(Entering loop iteration)如何选择方法结构化字段适用场景logging.info()是否通用日志node.log(info, msg, fields{...})是是结构化上下文sensor_id 等node.log_info(msg)是否快速单行print()否否遗留代码、快速调试数据流内省dataflow_descriptor/node_config/dataflow_iddataflow_descriptor()—— 返回完整数据流描述符解析后的数据流 YAML为 Python 字典descriptor node.dataflow_descriptor() print(descriptor[nodes])node_config()—— 返回数据流描述符中该节点的配置块config node.node_config() model_path config.get(model, default.pt)dataflow_id()—— 返回运行中数据流的唯一标识print(node.dataflow_id()) # 例如 a1b2c3d4-...Rust 实现中这三个方法分别对应dataflow_descriptor()通过pythonize转换、node_config()与dataflow_idlib.rs。故障恢复is_restart()与restart_count()配合 dora 的故障容忍机制见 docs/fault-tolerance.md节点可判断自己是否是被重启的实例is_restart()—— 节点是否在之前退出或失败后被重启。用于决定恢复已保存状态还是全新启动if node.is_restart(): restore_checkpoint()restart_count()—— 返回节点被重启的次数首次运行为0第一次重启后为1依此类推print(fRestart #{node.restart_count()})另外timestamp()返回节点混合逻辑时钟HLC的当前 UTC 时间可与 INPUT 事件的event[metadata][timestamp]对算单事件处理延迟lib.rs。ROS2 集成merge_external_events(subscription)将 ROS2 订阅流合并进节点主事件循环。调用后ROS2 消息会以kind为external的事件到达from dora import Node, Ros2Context, Ros2Node, Ros2NodeOptions, Ros2Topic node Node() ros2_context Ros2Context() ros2_node ros2_context.new_node(listener, Ros2NodeOptions()) topic Ros2Topic(/chatter, std_msgs/String, ros2_node) subscription ros2_node.create_subscription(topic) node.merge_external_events(subscription) for event in node: if event[kind] external: print(ROS2:, event[value]) elif event[type] INPUT: print(Dora:, event[id])参数subscriptiondora.Ros2Subscription——通过 dora ROS2 桥创建的 ROS2 订阅。实现上lib.rsmerge_external_events将订阅流包装为 futures 流并在节点内部把 dora 事件流与外部队列用merge_external_send合并合并后的流上try_recv/drain不可用返回空is_empty保守返回False。完整的 ROS2 桥接说明见 docs/ros2-bridge.md。事件字典结构事件以普通 Python 字典返回结构取决于事件类型。INPUT—— 来自其他节点的输入消息{ type: INPUT, id: camera_image, # 数据流 YAML 中声明的输入 ID kind: dora, # dora 表示数据流事件external 表示 ROS2 value: pyarrow.Array, # 负载Apache Arrow 数组 metadata: { timestamp: datetime, # UTC 时区 datetime.datetime open_telemetry_context: ..., # 追踪上下文若启用 ... # 任意用户元数据 }, }读取数据values event[value].to_pylist() # 转为 Python 列表 array event[value].to_numpy() # 转为 NumPy 数组INPUT_CLOSED—— 输入通道关闭上游节点结束{ type: INPUT_CLOSED, id: camera_image, kind: dora, }STOP—— 数据流正在关闭{ type: STOP, id: MANUAL | ALL_INPUTS_CLOSED, # 停止原因 kind: dora, }ERROR—— runtime 发生错误{ type: ERROR, error: description of the error, kind: dora, }ExternalROS2—— 使用merge_external_events时ROS2 消息到达为{ kind: external, value: pyarrow.Array, # ROS2 消息转成的 Arrow 数组 }事件字典在 Rust 侧由PyEvent::to_py_dict构建apis/python/operator/src/lib.rs它同时给出了事件类型的完整集合INPUT、INPUT_CLOSED、INPUT_RECOVERED、NODE_RESTARTED、RELOAD、PARAM_UPDATE、PARAM_DELETED、NODE_FAILED、ERROR、STOP未来新增变体将以UNKNOWN兜底。DoraStatus 枚举作为 operator 的on_event方法返回值控制事件循环from dora import DoraStatus值含义DoraStatus.CONTINUE继续处理事件值0DoraStatus.STOP停止当前 operator值1DoraStatus.STOP_ALL停止整个数据流值2该枚举定义在 dora/init.py是 PythonEnum的子类。Operator API在 runtime 进程内运行的算子Operator 运行在 dora runtime 进程内部无独立 OS 进程。它们被定义为名为Operator的 Python 类包含on_event方法。Operator 类用户定义创建包含Operator类的 Python 文件from dora import DoraStatus class Operator: def __init__(self): # 在此初始化状态 self.count 0 def on_event(self, dora_event, send_output) - DoraStatus: if dora_event[type] INPUT: self.count 1 # 处理输入可选地发送输出 send_output(result, bprocessed, dora_event[metadata]) return DoraStatus.CONTINUE方法__init__(self)—— 加载 operator 时调用一次。在此初始化状态或模型。on_event(self, dora_event, send_output) - DoraStatus—— 每个到达事件都会调用。必须返回DoraStatus值。on_event的参数dora_eventdict——事件字典。send_output可调用对象——发送输出数据的回调见下。runtime 还会在 operator 实例上设置self.dataflow_descriptor值为解析后的数据流 YAML 字典。send_output 回调send_output回调随on_event传入用于从 operator 发送数据send_output(output_id, data, metadataNone)参数output_idstr——数据流 YAML 中声明的输出名。databytes | pyarrow.Array——负载。metadatadict可选——附加的元数据。传入dora_event[metadata]可传播追踪上下文。示例import pyarrow as pa from dora import DoraStatus class Operator: def on_event(self, dora_event, send_output) - DoraStatus: if dora_event[type] INPUT: result pa.array([42], typepa.int64()) send_output(output, result, dora_event[metadata]) return DoraStatus.CONTINUEDataflowBuilder用 Python 程序化构建数据流from dora.builder import DataflowBuilder, Node, Operator, Output在 Python 中程序化生成数据流 YAML。完整实现见 apis/python/node/dora/builder.py配套测试见 apis/python/node/tests/test_builder.py。DataflowBuilder 类__init__(namedora-dataflow)创建新的数据流构建器flow DataflowBuilder(my-robot)namestr可选——数据流名称默认为dora-dataflow。add_node(id, **kwargs) - Node向数据流添加节点返回用于进一步配置的Node对象sender flow.add_node(sender)idstr——唯一节点标识。**kwargs——透传到 YAML 的额外节点配置。to_yaml(pathNone) - str | None生成数据流的 YAML 表示。给定path时写入文件并返回None否则返回 YAML 字符串# 写入文件 flow.to_yaml(dataflow.yml) # 获取字符串 yaml_str flow.to_yaml()生成的 YAML 结构为{nodes: [node.to_dict() ...]}与手写数据流文件如 examples/python-dataflow/dataflow.yml一致。上下文管理器DataflowBuilder支持with语句with DataflowBuilder(my-flow) as flow: flow.add_node(sender).path(sender.py) flow.to_yaml(dataflow.yml)Node 类构建器由DataflowBuilder.add_node()返回。所有 setter 方法返回self以支持链式调用。path(path) - Node—— 设置节点可执行文件或脚本路径node.path(my_node.py)args(args) - Node—— 设置节点命令行参数node.args(--verbose --port 8080)env(env) - Node—— 设置节点环境变量node.env({MODEL_PATH: /models/yolo.pt})build(command) - Node—— 设置构建命令启动前运行node.build(pip install -r requirements.txt)git(url, branchNone, tagNone, revNone) - Node—— 以 Git 仓库作为节点源码node.git(https://github.com/org/repo.git, branchmain)add_operator(operator) - Node—— 为该节点附加一个Operatorop Operator(detector, pythonobject_detection.py) node.add_operator(op)add_output(output_id) - Output—— 声明节点输出并返回Output引用作为其他节点的输入源output sender.add_output(data)add_input(input_id, source, queue_sizeNone, queue_policyNone) - Node—— 订阅其他节点的输出# 使用 Output 对象 output sender.add_output(data) receiver.add_input(data, output) # 使用字符串引用 receiver.add_input(tick, dora/timer/millis/100) # 自定义队列大小 receiver.add_input(images, camera_output, queue_size2) # 无损输入队列满时阻塞发送方 receiver.add_input(commands, cmd_output, queue_size100, queue_policybackpressure)参数input_idstr——该节点上的输入名。sourcestr | Output——字符串node_id/output_id或Output对象。queue_sizeint可选——该输入缓冲的最大消息数。queue_policystr可选——drop_oldest默认或backpressure缓冲至queue_size的 10 倍后才丢弃。从 builder.py 可以看到queue_policy非drop_oldest/backpressure时会抛出ValueErrorqueue_size 1同样报错测试 test_builder.py 对此有专门覆盖。to_dict() - dict—— 返回节点的字典表示供 YAML 序列化。Output 类构建器由Node.add_output()返回表示节点输出的引用作为add_input()的源output sender.add_output(data) receiver.add_input(sensor_data, output) str(output) # sender/dataOutput.__str__实现为f{node.id}/{output_id}builder.py。Operator 类构建器定义嵌入节点 YAML 配置的 operator。__init__(id, nameNone, descriptionNone, buildNone, pythonNone, shared_libraryNone, send_stdout_asNone)op Operator( iddetector, pythonobject_detection.py, send_stdout_asdetection_text, )参数idstr——唯一 operator 标识。namestr可选——显示名。descriptionstr可选——人类可读描述。buildstr可选——加载前运行的构建命令。pythonstr可选——Python operator 文件路径。shared_librarystr可选——共享库 operator 路径。send_stdout_asstr可选——把 operator 的 stdout 作为该 ID 的输出路由。to_dict() - dict返回字典表示供 YAML 序列化。CUDA 模块GPU 张量零拷贝共享可选扩展不属于 1.0 API。这些辅助函数在 1.0 前从dora包移入了dora_tensor_pool扩展——它们服务于 tensor-pool 传输位于扩展接缝之后不受 dora 1.0 兼容性保证覆盖。从libraries/extensions/tensor-pool/python安装。from dora_tensor_pool import torch_to_ipc_buffer, ipc_buffer_to_ipc_handle, open_ipc_handle这些工具用于节点间通过 CUDA IPC 进行零拷贝 GPU 张量共享。需要带 CUDA 支持的 PyTorch 和带 CUDA 支持的 Numba。torch_to_ipc_buffer(tensor) - tuple[pyarrow.Array, dict]把 PyTorch CUDA 张量转换为包含 CUDA IPC handle 的 Arrow 数组外加元数据字典。二者一起通过数据流发送即可共享 GPU 内存而不拷贝import torch import pyarrow as pa from dora import Node from dora_tensor_pool import torch_to_ipc_buffer node Node() tensor torch.randn(1024, 768, devicecuda) ipc_buffer, metadata torch_to_ipc_buffer(tensor) node.send_output(gpu_data, ipc_buffer, metadata)tensortorch.Tensor——CUDA 张量。返回IPC handle 的 int8 Arrow 数组以及包含 shape、strides、dtype、size、offset 和 source 信息的元数据。ipc_buffer_to_ipc_handle(handle_buffer, metadata) - IpcHandle从收到的 Arrow 缓冲和元数据重建 CUDA IPC handlefrom dora_tensor_pool import ipc_buffer_to_ipc_handle event node.next() ipc_handle ipc_buffer_to_ipc_handle(event[value], event[metadata])handle_bufferpyarrow.Array——来自event[value]的 Arrow 数组。metadatadict——来自event[metadata]的元数据。返回dora_tensor_pool.IpcHandlecudaIpcMemHandle_t的轻量包装调用.open()把 handle 映射进当前进程并获得设备指针.close()释放。open_ipc_handle(ipc_handle, metadata) - ContextManager[torch.Tensor]打开 CUDA IPC handle 并产出 PyTorch 张量。以上下文管理器方式使用以保证正确清理from dora_tensor_pool import ipc_buffer_to_ipc_handle, open_ipc_handle event node.next() ipc_handle ipc_buffer_to_ipc_handle(event[value], event[metadata]) with open_ipc_handle(ipc_handle, event[metadata]) as tensor: result tensor * 2 # 直接在 GPU 上使用张量ipc_handleIpcHandle——ipc_buffer_to_ipc_handle返回的 handle。metadatadict——含 shape、strides、dtype 信息的元数据。返回产出 CUDAtorch.Tensor的上下文管理器。注意Node上的register_tensor_pool/write_tensor_pool/read_tensor_pool/free_tensor_pool也属于该可选扩展lib.rs仅在以tensor-poolfeature 构建的 wheel 中存在。快速开始完整节点示例一个接收图像、处理并发送结果的完整节点#!/usr/bin/env python3 示例节点接收消息、转换并发送输出。 import logging import pyarrow as pa from dora import Node def main(): node Node() for event in node: if event[type] INPUT: input_id event[id] if input_id message: values event[value].to_pylist() number values[0] # 创建含多个字段的 struct 数组 result pa.StructArray.from_arrays( [ pa.array([number * 2]), pa.array([fMessage #{number}]), ], names[doubled, description], ) node.send_output(transformed, result) logging.info(Transformed message %d, number) elif event[type] STOP: logging.info(Node stopping) break if __name__ __main__: main()运行dora run dataflow.yml配套的数据流 YAML 参考 examples/python-dataflow/dataflow.yml其中sender声明输出messagetransformer订阅sender/message并输出transformedreceiver同时订阅两个输入。更完整的 Python 数据流示例含run.rs与动态数据流变体见 examples/python-dataflow。DataflowBuilder 完整示例用代码构建数据流替代手写 YAML#!/usr/bin/env python3 构建一个简单的 sender - receiver 数据流。 from dora.builder import DataflowBuilder, Operator flow DataflowBuilder(example-flow) # 添加定时驱动的发送节点 sender flow.add_node(sender) sender.path(sender.py) tick_output sender.add_output(message) # 添加订阅发送方的接收节点 receiver flow.add_node(receiver) receiver.path(receiver.py) receiver.add_input(message, tick_output) # 添加带定时输入的节点 timed_node flow.add_node(periodic) timed_node.path(periodic.py) timed_node.add_input(tick, dora/timer/millis/100) # 添加带 operator 的节点 runtime_node flow.add_node(runtime-node) op Operator(detector, pythonobject_detection.py) runtime_node.add_operator(op) runtime_node.add_input(image, camera/image) # 写入或打印 YAML flow.to_yaml(dataflow.yml) print(flow.to_yaml())该示例覆盖了本文介绍的全部构建器 APIpath、add_output、add_inputOutput对象与dora/timer/millis/100字符串两种 source 形态、add_operator以及to_yaml的两种输出方式。运行数据流依旧使用dora run dataflow.yml与手写 YAML 完全等效——这也是 DataflowBuilder 的核心价值将数据流定义纳入 Python 工程化流程变量、循环、类型检查同时保持与 YAML 格式的 100% 兼容。【免费下载链接】doraDORA (Dataflow-Oriented Robotic Architecture) is middleware designed to streamline and simplify the creation of AI-based robotic applications. It offers low latency, composable, and distributed dataflow capabilities. Applications are modeled as directed graphs, also referred to as pipelines.项目地址: https://gitcode.com/GitHub_Trending/do/dora创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
DORA Python API 开发实战指南Node、Operator 与 DataflowBuilder 全解析【免费下载链接】doraDORA (Dataflow-Oriented Robotic Architecture) is middleware designed to streamline and simplify the creation of AI-based robotic applications. It offers low latency, composable, and distributed dataflow capabilities. Applications are modeled as directed graphs, also referred to as pipelines.项目地址: https://gitcode.com/GitHub_Trending/do/doraDORADataflow-Oriented Robotic Architecture是一套面向 AI 机器人应用的低延迟、可组合、分布式数据流中间件应用被建模为有向图pipeline。本文基于仓库 docs/api-python.md 编写系统讲解 DORA 的 Python 编程接口如何用dora.Node编写自定义节点、如何在 dora runtime 进程内运行Operator、如何用DataflowBuilder在 Python 中程序化生成数据流 YAML并延伸到 CUDA 张量零拷贝传输、服务/动作/流式通信模式与结构化日志。读完本文你可以从零编写一个可运行的 Python 节点并将其接入 dora daemon 管理的数据流图。安装与运行环境要求DORA 的 Python 客户端以dora-rs包发布安装命令pip install dora-rs解释器兼容性CPython 3.11 及以上。官方 wheel 以abi3-py311构建单个 wheel 即可在所有后续 CPython 版本上运行无需等待 dora 发版——3.11 是 dora 1.x 的下限floor并非锁定版本。不支持的构建free-threaded 构建python3.13t、python3.14t不在 abi3 覆盖范围内也没有为其发布 wheel。完整保证请参阅 Python 版本策略。开发调试本仓库内的 Python 绑定位于 apis/python/node时README 给出了本地构建方式uv venv --seed -p 3.11 uv pip install -e .类型提示需额外生成 stub 文件python generate_stubs.py dora dora/__init__.pyi maturin developNode API自定义节点的核心接口Node是编写自定义节点的首要接口它连接到一个正在运行的数据流dataflow接收输入事件并发送输出from dora import Node从源码结构看Node类由 Rust 通过 pyo3 导出见 apis/python/node/src/lib.rs内部持有DoraNode与一个Events事件流并在构造时自动安装日志桥接见下文“日志”小节。__init__(node_idNone)创建一个新节点并连接正在运行的数据流# 标准用法节点 ID 从 daemon 设置的环境变量读取 node Node() # 动态用法通过显式节点 ID 连接到正在运行的数据流 node Node(node_idmy-dynamic-node)参数node_idstr可选——动态节点的显式 ID。省略时节点从 dora daemon 设置的环境变量读取身份。异常节点无法连接数据流时抛出RuntimeError。从类型定义dora/init.pyi可以看到构造签名还支持daemon_portNode(my-node, daemon_port6789)可用于连接自定义端口上的 daemon。Rust 实现中lib.rs显式传入node_id走DoraNode::init_flexible(node_id)带端口时使用DoraNode::builder()省略时走DoraNode::init_from_env()。接收事件next/drain/try_recv/recv_async/is_emptynext(timeoutNone)—— 从事件流取下一个事件阻塞直到有事件可用或超时event node.next() # 无限期阻塞 event node.next(timeout2.0) # 最多阻塞 2 秒timeoutfloat可选——最大等待秒数。返回事件字典当所有发送方都已关闭或超时到期时返回None。实现上lib.rstimeout会被转换为DurationRust 侧对 NaN、负数或无穷大的超时输入返回干净的ValueError而非 panic见timeout_to_durationlib.rs。drain()—— 不阻塞地取出所有缓冲事件events node.drain() for event in events: print(event[type])返回list[dict]无缓冲事件时返回空列表。try_recv()—— 非阻塞接收返回下一个缓冲事件若有event node.try_recv() if event is not None: print(event[type])返回dict | None。recv_async(timeoutNone)—— 异步接收配合asyncio使用event await node.recv_async() event await node.recv_async(timeout5.0)超时到达时返回错误所有发送方关闭时返回None。注意此方法为实验特性pyo3 的 asyncRust-Python FFI集成仍在开发中。is_empty()—— 检查事件流中是否还有缓冲事件if not node.is_empty(): event node.try_recv()返回bool。迭代支持Node实现了__iter__与__next__lib.rs可以直接用for循环消费事件流for event in node: match event[type]: case INPUT: process(event[value]) case STOP: break迭代器每次迭代调用无超时的next()事件流关闭时产出None从而终止循环。发送输出send_output(output_id, data, metadataNone)在输出通道上发送数据import pyarrow as pa # 发送原始字节 node.send_output(status, bOK) # 发送 Apache Arrow 数组可零拷贝 node.send_output(values, pa.array([1, 2, 3])) # 携带元数据发送 node.send_output(image, pa.array(pixels), {camera_id: front})参数output_idstr——数据流 YAML 中声明的输出名。databytes | pyarrow.Array——负载。简单数据用bytes需要零拷贝共享内存传输时用pyarrow.Array。metadatadict可选——附加到消息的键值对。支持的值类型bool、int、float、str、list[int]、list[float]、list[str]、datetime.datetime。异常data既不是bytes也不是pyarrow.Array时抛出RuntimeError。Rust 侧的分发逻辑send_payloadlib.rs验证PyBytes走send_output_bytes原始字节路径pyarrow 数组走 Arrow 数组路径其余类型直接报错。零拷贝补充send_output_rawPython ≥ 3.11。源码中还提供了可写缓冲区的零拷贝发送路径lib.rs预分配指定字节数的共享内存/对齐堆存储通过 Python buffer 协议memoryview、numpy.frombuffer、struct等填充后再发送with node.send_output_raw(rgb_image, 1920 * 1080 * 3, {width: 1920}) as buf: numpy.frombuffer(buf, numpy.uint8).reshape(1080, 1920, 3)[:] frame # 退出 with 块时自动发送在 Python 3.11 上该方法抛出NotImplementedError并给出升级提示lib.rs。服务、动作与流式通信模式Python 节点与 Rust 使用相同的元数据键约定完整指南见 通信模式。参数是键为字符串的普通 dict。预定义元数据键键说明request_id服务请求/响应关联UUID v7goal_id动作目标标识UUID v7goal_status动作结果状态succeeded、aborted或canceledsession_id流式会话标识segment_id会话内的流式分段整数seq流式分块序号整数fin流式分段最后一个分块boolflush丢弃输入上更旧的排队消息bool服务客户端示例import uuid # 发送带唯一 request_id 的请求 request_id str(uuid.uuid7()) # Python 3.13旧版本用 uuid_utils 或 uuid.uuid4() node.send_output(request, data, {request_id: request_id})服务端示例# 原样透传请求中的元数据含 request_id node.send_output(response, result, event[metadata])动作客户端示例goal_id str(uuid.uuid7()) node.send_output(goal, data, {goal_id: goal_id})流式示例用户中断时刷新下游队列# metadata 是参数名 - 值的扁平 dict没有外层 parameters 键 node.send_output( text, data, metadata{ session_id: session_id, segment_id: 1, seq: 0, fin: False, flush: True, }, )源码还提供send_service_request/send_service_response便捷包装lib.rs前者自动注入 UUID v7request_id并返回该 ID后者校验透传的元数据中包含request_id。日志logging自动桥接与显式节点 APIPython 节点既可以使用内置logging模块推荐也可以使用显式节点 API。Pythonlogging模块自动桥接创建Node()时会自动安装一个 handler把 Pythonlogging模块的路由指向 dora daemon无需任何配置import logging from dora import Node node Node() # 安装日志桥接 logging.info(Sensor initialized) # - 结构化 info 日志 logging.warning(High temperature) # - 结构化 warn 日志 logging.debug(Raw bytes: %s, data) # - 结构化 debug 日志这些日志条目携带完整元数据级别、消息、文件路径、行号并支持 daemon 的min_log_level过滤、send_logs_as路由和dora/logs订阅者。注意不要在创建Node()之前调用logging.basicConfig()。构造函数会设置桥接先调用basicConfig()可能安装冲突的 handler。实现细节lib.rssetup_logging将logging.basicConfig包装为默认使用HostHandler后者调用 Rust 函数host_log把logging.LogRecord转码为tracing::Event按 levelno 映射到 ERROR/WARN/INFO/DEBUG/TRACE从而让 Python 日志进入 dora 的统一结构化日志管道。显式节点 APIlog(level, message, targetNone, fieldsNone)发出带可选 target 和键值字段的结构化日志node.log(info, Processing frame, targetvision) node.log(error, Sensor timeout, fields{sensor: lidar, retry: 3})levelstr——日志级别error、warn、info、debug或trace。messagestr——日志消息。targetstr可选——目标模块或子系统名。fieldsdict[str, str]可选——结构化键值上下文字段。log_error(message)、log_warn(message)、log_info(message)、log_debug(message)、log_trace(message)常用级别的便捷方法各自等价于node.log(level, message)node.log_error(Connection failed) node.log_warn(Temperature elevated) node.log_info(Sensor initialized) node.log_debug(Raw bytes received) node.log_trace(Entering loop iteration)如何选择方法结构化字段适用场景logging.info()是否通用日志node.log(info, msg, fields{...})是是结构化上下文sensor_id 等node.log_info(msg)是否快速单行print()否否遗留代码、快速调试数据流内省dataflow_descriptor/node_config/dataflow_iddataflow_descriptor()—— 返回完整数据流描述符解析后的数据流 YAML为 Python 字典descriptor node.dataflow_descriptor() print(descriptor[nodes])node_config()—— 返回数据流描述符中该节点的配置块config node.node_config() model_path config.get(model, default.pt)dataflow_id()—— 返回运行中数据流的唯一标识print(node.dataflow_id()) # 例如 a1b2c3d4-...Rust 实现中这三个方法分别对应dataflow_descriptor()通过pythonize转换、node_config()与dataflow_idlib.rs。故障恢复is_restart()与restart_count()配合 dora 的故障容忍机制见 docs/fault-tolerance.md节点可判断自己是否是被重启的实例is_restart()—— 节点是否在之前退出或失败后被重启。用于决定恢复已保存状态还是全新启动if node.is_restart(): restore_checkpoint()restart_count()—— 返回节点被重启的次数首次运行为0第一次重启后为1依此类推print(fRestart #{node.restart_count()})另外timestamp()返回节点混合逻辑时钟HLC的当前 UTC 时间可与 INPUT 事件的event[metadata][timestamp]对算单事件处理延迟lib.rs。ROS2 集成merge_external_events(subscription)将 ROS2 订阅流合并进节点主事件循环。调用后ROS2 消息会以kind为external的事件到达from dora import Node, Ros2Context, Ros2Node, Ros2NodeOptions, Ros2Topic node Node() ros2_context Ros2Context() ros2_node ros2_context.new_node(listener, Ros2NodeOptions()) topic Ros2Topic(/chatter, std_msgs/String, ros2_node) subscription ros2_node.create_subscription(topic) node.merge_external_events(subscription) for event in node: if event[kind] external: print(ROS2:, event[value]) elif event[type] INPUT: print(Dora:, event[id])参数subscriptiondora.Ros2Subscription——通过 dora ROS2 桥创建的 ROS2 订阅。实现上lib.rsmerge_external_events将订阅流包装为 futures 流并在节点内部把 dora 事件流与外部队列用merge_external_send合并合并后的流上try_recv/drain不可用返回空is_empty保守返回False。完整的 ROS2 桥接说明见 docs/ros2-bridge.md。事件字典结构事件以普通 Python 字典返回结构取决于事件类型。INPUT—— 来自其他节点的输入消息{ type: INPUT, id: camera_image, # 数据流 YAML 中声明的输入 ID kind: dora, # dora 表示数据流事件external 表示 ROS2 value: pyarrow.Array, # 负载Apache Arrow 数组 metadata: { timestamp: datetime, # UTC 时区 datetime.datetime open_telemetry_context: ..., # 追踪上下文若启用 ... # 任意用户元数据 }, }读取数据values event[value].to_pylist() # 转为 Python 列表 array event[value].to_numpy() # 转为 NumPy 数组INPUT_CLOSED—— 输入通道关闭上游节点结束{ type: INPUT_CLOSED, id: camera_image, kind: dora, }STOP—— 数据流正在关闭{ type: STOP, id: MANUAL | ALL_INPUTS_CLOSED, # 停止原因 kind: dora, }ERROR—— runtime 发生错误{ type: ERROR, error: description of the error, kind: dora, }ExternalROS2—— 使用merge_external_events时ROS2 消息到达为{ kind: external, value: pyarrow.Array, # ROS2 消息转成的 Arrow 数组 }事件字典在 Rust 侧由PyEvent::to_py_dict构建apis/python/operator/src/lib.rs它同时给出了事件类型的完整集合INPUT、INPUT_CLOSED、INPUT_RECOVERED、NODE_RESTARTED、RELOAD、PARAM_UPDATE、PARAM_DELETED、NODE_FAILED、ERROR、STOP未来新增变体将以UNKNOWN兜底。DoraStatus 枚举作为 operator 的on_event方法返回值控制事件循环from dora import DoraStatus值含义DoraStatus.CONTINUE继续处理事件值0DoraStatus.STOP停止当前 operator值1DoraStatus.STOP_ALL停止整个数据流值2该枚举定义在 dora/init.py是 PythonEnum的子类。Operator API在 runtime 进程内运行的算子Operator 运行在 dora runtime 进程内部无独立 OS 进程。它们被定义为名为Operator的 Python 类包含on_event方法。Operator 类用户定义创建包含Operator类的 Python 文件from dora import DoraStatus class Operator: def __init__(self): # 在此初始化状态 self.count 0 def on_event(self, dora_event, send_output) - DoraStatus: if dora_event[type] INPUT: self.count 1 # 处理输入可选地发送输出 send_output(result, bprocessed, dora_event[metadata]) return DoraStatus.CONTINUE方法__init__(self)—— 加载 operator 时调用一次。在此初始化状态或模型。on_event(self, dora_event, send_output) - DoraStatus—— 每个到达事件都会调用。必须返回DoraStatus值。on_event的参数dora_eventdict——事件字典。send_output可调用对象——发送输出数据的回调见下。runtime 还会在 operator 实例上设置self.dataflow_descriptor值为解析后的数据流 YAML 字典。send_output 回调send_output回调随on_event传入用于从 operator 发送数据send_output(output_id, data, metadataNone)参数output_idstr——数据流 YAML 中声明的输出名。databytes | pyarrow.Array——负载。metadatadict可选——附加的元数据。传入dora_event[metadata]可传播追踪上下文。示例import pyarrow as pa from dora import DoraStatus class Operator: def on_event(self, dora_event, send_output) - DoraStatus: if dora_event[type] INPUT: result pa.array([42], typepa.int64()) send_output(output, result, dora_event[metadata]) return DoraStatus.CONTINUEDataflowBuilder用 Python 程序化构建数据流from dora.builder import DataflowBuilder, Node, Operator, Output在 Python 中程序化生成数据流 YAML。完整实现见 apis/python/node/dora/builder.py配套测试见 apis/python/node/tests/test_builder.py。DataflowBuilder 类__init__(namedora-dataflow)创建新的数据流构建器flow DataflowBuilder(my-robot)namestr可选——数据流名称默认为dora-dataflow。add_node(id, **kwargs) - Node向数据流添加节点返回用于进一步配置的Node对象sender flow.add_node(sender)idstr——唯一节点标识。**kwargs——透传到 YAML 的额外节点配置。to_yaml(pathNone) - str | None生成数据流的 YAML 表示。给定path时写入文件并返回None否则返回 YAML 字符串# 写入文件 flow.to_yaml(dataflow.yml) # 获取字符串 yaml_str flow.to_yaml()生成的 YAML 结构为{nodes: [node.to_dict() ...]}与手写数据流文件如 examples/python-dataflow/dataflow.yml一致。上下文管理器DataflowBuilder支持with语句with DataflowBuilder(my-flow) as flow: flow.add_node(sender).path(sender.py) flow.to_yaml(dataflow.yml)Node 类构建器由DataflowBuilder.add_node()返回。所有 setter 方法返回self以支持链式调用。path(path) - Node—— 设置节点可执行文件或脚本路径node.path(my_node.py)args(args) - Node—— 设置节点命令行参数node.args(--verbose --port 8080)env(env) - Node—— 设置节点环境变量node.env({MODEL_PATH: /models/yolo.pt})build(command) - Node—— 设置构建命令启动前运行node.build(pip install -r requirements.txt)git(url, branchNone, tagNone, revNone) - Node—— 以 Git 仓库作为节点源码node.git(https://github.com/org/repo.git, branchmain)add_operator(operator) - Node—— 为该节点附加一个Operatorop Operator(detector, pythonobject_detection.py) node.add_operator(op)add_output(output_id) - Output—— 声明节点输出并返回Output引用作为其他节点的输入源output sender.add_output(data)add_input(input_id, source, queue_sizeNone, queue_policyNone) - Node—— 订阅其他节点的输出# 使用 Output 对象 output sender.add_output(data) receiver.add_input(data, output) # 使用字符串引用 receiver.add_input(tick, dora/timer/millis/100) # 自定义队列大小 receiver.add_input(images, camera_output, queue_size2) # 无损输入队列满时阻塞发送方 receiver.add_input(commands, cmd_output, queue_size100, queue_policybackpressure)参数input_idstr——该节点上的输入名。sourcestr | Output——字符串node_id/output_id或Output对象。queue_sizeint可选——该输入缓冲的最大消息数。queue_policystr可选——drop_oldest默认或backpressure缓冲至queue_size的 10 倍后才丢弃。从 builder.py 可以看到queue_policy非drop_oldest/backpressure时会抛出ValueErrorqueue_size 1同样报错测试 test_builder.py 对此有专门覆盖。to_dict() - dict—— 返回节点的字典表示供 YAML 序列化。Output 类构建器由Node.add_output()返回表示节点输出的引用作为add_input()的源output sender.add_output(data) receiver.add_input(sensor_data, output) str(output) # sender/dataOutput.__str__实现为f{node.id}/{output_id}builder.py。Operator 类构建器定义嵌入节点 YAML 配置的 operator。__init__(id, nameNone, descriptionNone, buildNone, pythonNone, shared_libraryNone, send_stdout_asNone)op Operator( iddetector, pythonobject_detection.py, send_stdout_asdetection_text, )参数idstr——唯一 operator 标识。namestr可选——显示名。descriptionstr可选——人类可读描述。buildstr可选——加载前运行的构建命令。pythonstr可选——Python operator 文件路径。shared_librarystr可选——共享库 operator 路径。send_stdout_asstr可选——把 operator 的 stdout 作为该 ID 的输出路由。to_dict() - dict返回字典表示供 YAML 序列化。CUDA 模块GPU 张量零拷贝共享可选扩展不属于 1.0 API。这些辅助函数在 1.0 前从dora包移入了dora_tensor_pool扩展——它们服务于 tensor-pool 传输位于扩展接缝之后不受 dora 1.0 兼容性保证覆盖。从libraries/extensions/tensor-pool/python安装。from dora_tensor_pool import torch_to_ipc_buffer, ipc_buffer_to_ipc_handle, open_ipc_handle这些工具用于节点间通过 CUDA IPC 进行零拷贝 GPU 张量共享。需要带 CUDA 支持的 PyTorch 和带 CUDA 支持的 Numba。torch_to_ipc_buffer(tensor) - tuple[pyarrow.Array, dict]把 PyTorch CUDA 张量转换为包含 CUDA IPC handle 的 Arrow 数组外加元数据字典。二者一起通过数据流发送即可共享 GPU 内存而不拷贝import torch import pyarrow as pa from dora import Node from dora_tensor_pool import torch_to_ipc_buffer node Node() tensor torch.randn(1024, 768, devicecuda) ipc_buffer, metadata torch_to_ipc_buffer(tensor) node.send_output(gpu_data, ipc_buffer, metadata)tensortorch.Tensor——CUDA 张量。返回IPC handle 的 int8 Arrow 数组以及包含 shape、strides、dtype、size、offset 和 source 信息的元数据。ipc_buffer_to_ipc_handle(handle_buffer, metadata) - IpcHandle从收到的 Arrow 缓冲和元数据重建 CUDA IPC handlefrom dora_tensor_pool import ipc_buffer_to_ipc_handle event node.next() ipc_handle ipc_buffer_to_ipc_handle(event[value], event[metadata])handle_bufferpyarrow.Array——来自event[value]的 Arrow 数组。metadatadict——来自event[metadata]的元数据。返回dora_tensor_pool.IpcHandlecudaIpcMemHandle_t的轻量包装调用.open()把 handle 映射进当前进程并获得设备指针.close()释放。open_ipc_handle(ipc_handle, metadata) - ContextManager[torch.Tensor]打开 CUDA IPC handle 并产出 PyTorch 张量。以上下文管理器方式使用以保证正确清理from dora_tensor_pool import ipc_buffer_to_ipc_handle, open_ipc_handle event node.next() ipc_handle ipc_buffer_to_ipc_handle(event[value], event[metadata]) with open_ipc_handle(ipc_handle, event[metadata]) as tensor: result tensor * 2 # 直接在 GPU 上使用张量ipc_handleIpcHandle——ipc_buffer_to_ipc_handle返回的 handle。metadatadict——含 shape、strides、dtype 信息的元数据。返回产出 CUDAtorch.Tensor的上下文管理器。注意Node上的register_tensor_pool/write_tensor_pool/read_tensor_pool/free_tensor_pool也属于该可选扩展lib.rs仅在以tensor-poolfeature 构建的 wheel 中存在。快速开始完整节点示例一个接收图像、处理并发送结果的完整节点#!/usr/bin/env python3 示例节点接收消息、转换并发送输出。 import logging import pyarrow as pa from dora import Node def main(): node Node() for event in node: if event[type] INPUT: input_id event[id] if input_id message: values event[value].to_pylist() number values[0] # 创建含多个字段的 struct 数组 result pa.StructArray.from_arrays( [ pa.array([number * 2]), pa.array([fMessage #{number}]), ], names[doubled, description], ) node.send_output(transformed, result) logging.info(Transformed message %d, number) elif event[type] STOP: logging.info(Node stopping) break if __name__ __main__: main()运行dora run dataflow.yml配套的数据流 YAML 参考 examples/python-dataflow/dataflow.yml其中sender声明输出messagetransformer订阅sender/message并输出transformedreceiver同时订阅两个输入。更完整的 Python 数据流示例含run.rs与动态数据流变体见 examples/python-dataflow。DataflowBuilder 完整示例用代码构建数据流替代手写 YAML#!/usr/bin/env python3 构建一个简单的 sender - receiver 数据流。 from dora.builder import DataflowBuilder, Operator flow DataflowBuilder(example-flow) # 添加定时驱动的发送节点 sender flow.add_node(sender) sender.path(sender.py) tick_output sender.add_output(message) # 添加订阅发送方的接收节点 receiver flow.add_node(receiver) receiver.path(receiver.py) receiver.add_input(message, tick_output) # 添加带定时输入的节点 timed_node flow.add_node(periodic) timed_node.path(periodic.py) timed_node.add_input(tick, dora/timer/millis/100) # 添加带 operator 的节点 runtime_node flow.add_node(runtime-node) op Operator(detector, pythonobject_detection.py) runtime_node.add_operator(op) runtime_node.add_input(image, camera/image) # 写入或打印 YAML flow.to_yaml(dataflow.yml) print(flow.to_yaml())该示例覆盖了本文介绍的全部构建器 APIpath、add_output、add_inputOutput对象与dora/timer/millis/100字符串两种 source 形态、add_operator以及to_yaml的两种输出方式。运行数据流依旧使用dora run dataflow.yml与手写 YAML 完全等效——这也是 DataflowBuilder 的核心价值将数据流定义纳入 Python 工程化流程变量、循环、类型检查同时保持与 YAML 格式的 100% 兼容。【免费下载链接】doraDORA (Dataflow-Oriented Robotic Architecture) is middleware designed to streamline and simplify the creation of AI-based robotic applications. It offers low latency, composable, and distributed dataflow capabilities. Applications are modeled as directed graphs, also referred to as pipelines.项目地址: https://gitcode.com/GitHub_Trending/do/dora创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
