rainbow-rb-flow-manager 0.0.9.dev37

Rainbow Python Flow Manager

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 참고.