rb_flow_manager 사용 설명서

rb_flow_manager는 Step 트리를 멀티프로세스로 실행하고, 실행 상태/일시정지/재개/중지 제어를 제공하는 워크플로우 엔진입니다.

1. 핵심 구성요소

Step 계열 (step.py)
- 기본: Step
- 제어 구조: FolderStep, RepeatStep, ConditionStep, BreakStep, JumpToStep
- 고급: SubTaskStep, CallEventStep, SyncStep
실행기: ScriptExecutor (executor.py)
- 다중 프로세스 실행
- 상태 조회/제어 API 제공
컨트롤러 인터페이스: BaseController
- 실행 이벤트 훅(on_start, on_pause, on_error 등) 구현 가능
기본 컨트롤러 구현 예시: Zenoh_Controller

2. 최소 실행 예제

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. Step 작성 방식

Step는 두 방식으로 함수 실행을 지원합니다.

직접 함수 연결
- func=<callable>
문자열 함수명 방식
- func_name="rb_manipulate_sdk.point"
- 런타임에 import/호출

또한 args를 통해 스텝 인자를 전달하고, children으로 트리를 구성합니다.

4. 실행 제어 API

ScriptExecutor 주요 메서드:

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. 상태 모델

RB_Flow_Manager_ProgramState:

IDLE
RUNNING
PAUSED
WAITING
STOPPED
POST_START
COMPLETED
ERROR

각 프로세스별 상세 실행 상태는 ExecutorState TypedDict 구조로 관리됩니다.

6. 컨트롤러 연동

커스텀 컨트롤러를 사용하려면 BaseController를 구현한 뒤 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. 실무 팁

process_id는 서비스 내에서 유일하게 관리하세요.
긴 실행 트리는 min_step_interval을 사용해 과도한 호출을 방지하세요.
다중 트리 병렬 실행 시 start([...]) 형태를 사용하면 일괄 시작 제어가 쉽습니다.

8. 직렬화 주의 (func vs func_name)

event_sub_tree_list(CallEventStep) 와 SubTaskStep 가 로드하는 트리는 실행 시점에 to_dict() → from_dict() 직렬화를 거치거나 별도 프로세스로 전달됩니다. 이때:

func=<callable> 는 보존되지 않습니다 — funcName(문자열)만 직렬화됩니다. 직렬화 경로의 트리는 SDK 함수(func_name=...)나 구조 Step(Folder/Repeat/Condition)으로 구성하세요.
부모 메인 트리/post_tree 는 객체째 워커로 전달되므로 func= 콜러블을 그대로 써도 됩니다.
SubTaskStep 의 서브태스크 파일 내부는 importlib 로드라 func= 를 써도 됩니다.

9. 실행 가능한 예제

examples/ 디렉토리에 실로봇 없이 바로 돌아가는 동작 예제가 있습니다.

파일	주제
`examples/01_multiple_tasks.py`	여러 태스크 동시 실행(배치)
`examples/02_single_task.py`	단일 태스크
`examples/03_post_tree.py`	post_tree 포함
`examples/04_sub_task.py`	서브 태스크(SubTaskStep, INSERT)
`examples/05_event_task.py`	이벤트 태스크(CallEventStep, SYNC/ASYNC)

python examples/02_single_task.py

자세한 설명은 examples/README.md 참고.

Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

rainbow_rb_flow_manager-0.0.76.tar.gz (83.6 kB view details)

Uploaded Aug 10, 2026 Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

The dropdown lists show the available interpreters, ABIs, and platforms. Enable javascript to be able to filter the list of wheel files.

rainbow_rb_flow_manager-0.0.76-py3-none-any.whl (95.0 kB view details)

Uploaded Aug 10, 2026 Python 3

File details

Details for the file rainbow_rb_flow_manager-0.0.76.tar.gz.

File metadata

Download URL: rainbow_rb_flow_manager-0.0.76.tar.gz
Upload date: Aug 10, 2026
Size: 83.6 kB
Tags: Source
Uploaded using Trusted Publishing? No
Uploaded via: twine/7.0.0 CPython/3.12.5

File hashes

Hashes for rainbow_rb_flow_manager-0.0.76.tar.gz
Algorithm	Hash digest
SHA256	`8817ab9825cdaaba710c86d80a34b71f556f633e224863b06f90448302a0d41f`
MD5	`60b8c951057701e133bf14c2565b7f49`
BLAKE2b-256	`b1532871f6ab269d893bc5b4ad760fe013ad9a7612998be493913ad647f2967e`

See more details on using hashes here.

File details

Details for the file rainbow_rb_flow_manager-0.0.76-py3-none-any.whl.

File metadata

Download URL: rainbow_rb_flow_manager-0.0.76-py3-none-any.whl
Upload date: Aug 10, 2026
Size: 95.0 kB
Tags: Python 3
Uploaded using Trusted Publishing? No
Uploaded via: twine/7.0.0 CPython/3.12.5

File hashes

Hashes for rainbow_rb_flow_manager-0.0.76-py3-none-any.whl
Algorithm	Hash digest
SHA256	`4346ff71f7315432b75e6a743a53970cc2797d59f58a445ebbbe99868bf7308d`
MD5	`155bbcca47218fa3fe40e5cc0aa978e2`
BLAKE2b-256	`3b786cc9cc4ffe6cd832f174355979ce4e468b763095d786793f991d0a7e600c`