rb_flow_manager User Guide

rb_flow_manager is a workflow engine that runs Step trees across multiple processes and provides execution state, pause, resume, and stop control.

1. Core Components

• Step family (step.py)
  • Basic: Step
  • Control structures: FolderStep, RepeatStep, ConditionStep, BreakStep, JumpToStep
  • Advanced: SubTaskStep, CallEventStep, SyncStep
• Executor: ScriptExecutor (executor.py)
  • Multi-process execution
  • Provides state query/control APIs
• Controller interface: BaseController
  • Lets you implement execution event hooks (on_start, on_pause, on_error, etc.)
• Example default controller implementation: Zenoh_Controller

2. Minimal Example

from rb_flow_manager.step import Step
from rb_flow_manager.executor import ScriptExecutor
from rb_flow_manager.schema import MakeProcessArgs

def hello_step():
    print("hello step")

root = Step(
    step_id="root",
    name="Root",
    children=[
        Step(step_id="s1", name="Hello", func=hello_step),
    ],
)

executor = ScriptExecutor()

args: MakeProcessArgs = {
    "process_id": "script_1",
    "step": root,
    "repeat_count": 1,
    "robot_model": "C500920",
    "category": "manipulate",
    "step_mode": False,
    "min_step_interval": 0.0,
    "is_ui_execution": False,
    "post_tree": None,
    "parent_process_id": None,
    "event_sub_tree_list": [],
}

executor.start(args)
executor.wait_all()
print(executor.get_all_states())

3. Writing Steps

Step supports two ways of executing a function.

• Direct function binding
  • func=<callable>
• Function name as a string
  • func_name="rb_manipulate_sdk.point"
  • Imported and called at runtime

Step arguments are passed via args, and the tree is built with children.

4. Execution Control API

Main ScriptExecutor methods:

• start(args: MakeProcessArgs | list[MakeProcessArgs]) -> bool
• pause(process_id: str) -> bool
• resume(process_id: str) -> bool
• stop(process_id: str) -> bool
• wait_all(timeout: float | None = None) -> bool
• get_all_states() -> dict[str, dict]

5. State Model

RB_Flow_Manager_ProgramState:

• IDLE
• RUNNING
• PAUSED
• WAITING
• STOPPED
• POST_START
• COMPLETED
• ERROR

The detailed execution state of each process is managed in the ExecutorState TypedDict structure.

6. Controller Integration

To use a custom controller, implement BaseController and pass it via ScriptExecutor(controller=...).

from rb_flow_manager.controller.base_controller import BaseController

class MyController(BaseController):
    def on_init(self, state_dicts): ...
    def on_start(self, task_id): ...
    def on_stop(self, task_id, step_id): ...
    def on_pause(self, task_id, step_id): ...
    def on_wait(self, task_id, step_id): ...
    def on_resume(self, task_id, step_id): ...
    def on_next(self, task_id, step_id): ...
    def on_sub_task_start(self, task_id, sub_task_id, sub_task_type): ...
    def on_sub_task_done(self, task_id, sub_task_id, sub_task_type): ...
    def on_done(self, task_id, step_id): ...
    def on_error(self, task_id, step_id, error): ...
    def on_post_start(self, task_id): ...
    def on_complete(self, task_id): ...
    def on_close(self): ...
    def on_all_complete(self): ...
    def on_all_stop(self): ...
    def on_all_pause(self): ...
    def on_all_wait(self): ...
    def on_all_resume(self): ...

7. Practical Tips

• Keep each process_id unique within a service.
• For long-running trees, use min_step_interval to prevent excessive calls.
• When running multiple trees in parallel, the start([...]) form makes batch start control easier.

8. Serialization Caveat (func vs func_name)

Trees loaded by event_sub_tree_list (CallEventStep) and SubTaskStep go through to_dict() → from_dict() serialization at execution time or are handed off to a separate process. In that case:

• func=<callable> is not preserved — only funcName (a string) is serialized. Build trees on serialized paths from SDK functions (func_name=...) or structural Steps (Folder/Repeat/Condition).
• The parent main tree and post_tree are passed to the worker as objects, so func= callables work as-is.
• Inside a SubTaskStep sub-task file, loading is done via importlib, so func= is fine.

9. Runnable Examples

The examples/ directory contains working examples that run without a real robot.

python examples/02_single_task.py

See examples/README.md for details.