Metadata-Version: 2.4
Name: rainbow-rb-flow-manager
Version: 0.10.4
Summary: Rainbow Python Flow Manager
Author-email: Derek <dfd1123@rainbow-robotics.com>
Requires-Python: <3.14,>=3.11
Description-Content-Type: text/markdown
Requires-Dist: rainbow-rb-utils==0.10.4
Requires-Dist: rainbow-rb-zenoh==0.10.4
Requires-Dist: rainbow-rb-flat-buffers==0.10.4
Requires-Dist: rainbow-rb-schemas==0.10.4
Requires-Dist: rainbow-rb-sdk==0.10.4
Requires-Dist: rainbow-rb-log==0.10.4

# 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

```python
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=...)`.

```python
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.

| File | Topic |
|------|------|
| `examples/01_multiple_tasks.py` | Running multiple tasks concurrently (batch) |
| `examples/02_single_task.py` | Single task |
| `examples/03_post_tree.py` | With post_tree |
| `examples/04_sub_task.py` | Sub task (SubTaskStep, INSERT) |
| `examples/05_event_task.py` | Event task (CallEventStep, SYNC/ASYNC) |

```bash
python examples/02_single_task.py
```

See `examples/README.md` for details.
