Zzz Od Dev State Machine

Operation 状态机开发指南

本项目的“状态机”通常指 Operation 节点图：

• 普通流程继承 ZOperation：src/zzz_od/operation/zzz_operation.py
• 完整应用继承 ZApplication：src/zzz_od/application/zzz_application.py
• 底层执行引擎：src/one_dragon/base/operation/operation.py
• 节点声明：operation_node
• 边声明：node_from
• 单轮结果：OperationRoundResult

1. 最小结构

from one_dragon.base.operation.operation_edge import node_from
from one_dragon.base.operation.operation_node import operation_node
from one_dragon.base.operation.operation_round_result import OperationRoundResult
from zzz_od.context.zzz_context import ZContext
from zzz_od.operation.zzz_operation import ZOperation


class MyOperation(ZOperation):

    def __init__(self, ctx: ZContext):
        ZOperation.__init__(self, ctx, op_name='我的流程')

    @operation_node(name='入口', is_start_node=True)
    def start(self) -> OperationRoundResult:
        return self.round_success('下一步')

    @node_from(from_name='入口', status='下一步')
    @operation_node(name='处理')
    def do_work(self) -> OperationRoundResult:
        return self.round_success()

2. 节点与边

@operation_node

常用参数：

@node_from

常用匹配方式：

@node_from(from_name='入口')                      # 上游成功即可进入
@node_from(from_name='入口', success=False)       # 上游失败进入
@node_from(from_name='入口', status='指定状态')    # 上游 status 精确匹配
@node_from(from_name='入口', success=True, status='指定状态')

同一个节点可以声明多个 @node_from，表示多个入口都能流转到这里。

3. 轮次结果怎么选

规则：

• 需要继续停在当前节点时，用 round_retry 或 round_wait。
• 等待外部状态变化且不希望耗尽重试次数时，用 round_wait。
• status 是流转条件的一部分；被 @node_from(status=...) 使用的字符串不要随意改。

4. 截图时机

Operation.execute() 每次循环调用 _execute_one_round()。如果当前节点是 @operation_node 方法，且该节点的 screenshot_before_round=True，框架会在调用节点方法前执行一次 self.screenshot()，更新：

• self.last_screenshot
• self.last_screenshot_time

默认 screenshot_before_round=True。

4.1 返回值和截图关系

结论：返回值本身不会截图；它只决定是否进入下一轮。真正截图发生在下一轮节点方法执行前。

4.2 同一个节点方法内部不会自动重新截图

在一次节点方法调用期间，self.last_screenshot 是同一张图，除非你显式调用：

screen = self.screenshot()

因此同一个节点里连续写多次识别，默认都基于同一张截图：

@operation_node(name='识别并点击', is_start_node=True)
def check_and_click(self) -> OperationRoundResult:
    # 这里使用本轮开始前自动截取的 last_screenshot
    result = self.round_by_find_area(self.last_screenshot, '画面', '按钮')
    if result.is_success:
        self.round_by_click_area('画面', '按钮')

    # 这里仍然是同一轮的 last_screenshot，不会因为刚才点击过就自动刷新
    result = self.round_by_find_area(self.last_screenshot, '画面', '点击后出现的按钮')
    return self.round_retry('等待点击后界面刷新', wait=1)

这种场景应拆成两个节点，或在点击后显式截图：

self.round_by_click_area('画面', '按钮')
self.screenshot()
return self.round_by_find_area(self.last_screenshot, '画面', '点击后出现的按钮')

4.3 辅助方法默认不主动截图

多数 round_by_find_area、round_by_find_and_click_area、round_by_goto_screen、round_by_ocr... 辅助方法在 screen=None 时使用 self.last_screenshot。不要误以为传 None 就会立即截图。

需要新画面时，先调用 self.screenshot()，或让本轮返回 round_retry / round_wait，等下一轮自动截图。

5. 常见状态机模式

5.1 入口识别后分流

@operation_node(name='画面识别', is_start_node=True)
def check_screen(self) -> OperationRoundResult:
    current = self.check_and_update_current_screen(self.last_screenshot)
    if current == '大世界':
        return self.round_success('已在大世界')
    if current == '菜单':
        return self.round_success('需要返回')
    return self.round_retry('未识别当前画面', wait=1)

@node_from(from_name='画面识别', status='需要返回')
@operation_node(name='返回大世界')
def back_to_world(self) -> OperationRoundResult:
    ...

5.2 调用子 Operation

@operation_node(name='返回大世界', is_start_node=True)
def back_to_world(self) -> OperationRoundResult:
    op = BackToNormalWorld(self.ctx)
    return self.round_by_op_result(op.execute())

适合把通用流程拆成可复用 Operation。不要把子 Operation 的内部节点复制到当前类里。

5.3 查找并点击画面区域

@operation_node(name='点击按钮')
def click_button(self) -> OperationRoundResult:
    return self.round_by_find_and_click_area(
        self.last_screenshot,
        '快捷手册',
        '今日最大活跃度',
        success_wait=1,
        retry_wait=1,
    )

区域名来自 assets/game_data/screen_info/。新增或改动区域时，要同步检查 screen YAML。

5.4 高频轮询

@operation_node(name='画面识别', mute=True)
def check_screen(self) -> OperationRoundResult:
    self.ctx.auto_battle_context.check_battle_state(self.last_screenshot)
    return self.round_wait(wait_round_time=self.ctx.battle_assistant_config.screenshot_interval)

高频节点建议设置 mute=True，避免日志淹没真正错误。

5.5 失败兜底

@node_from(from_name='确认', success=False)
@operation_node(name='返回大世界')
def back_after_fail(self) -> OperationRoundResult:
    op = BackToNormalWorld(self.ctx)
    op.execute()
    return self.round_fail('确认失败，已尝试返回')

失败兜底可以清理现场，但不要把业务失败伪装成成功。

6. Application 写法

应用状态机继承 ZApplication，节点写法与 ZOperation 相同。

from zzz_od.application.zzz_application import ZApplication


class MyApp(ZApplication):

    def __init__(self, ctx: ZContext):
        ZApplication.__init__(
            self,
            ctx=ctx,
            app_id=my_app_const.APP_ID,
            op_name=my_app_const.APP_NAME,
        )

应用还需要工厂、常量、可选配置和运行记录。新增应用/插件见 zzz-od-dev-plugin。

7. 编写检查清单

• 节点名稳定，@node_from(from_name=...) 能精确对应。
• 只有一个 is_start_node=True。
• 所有会被边匹配的 status 都是固定字符串或类常量。
• 等待外部变化用 round_wait，失败重试用 round_retry。
• 需要点击后识别新画面时，拆节点或显式调用 self.screenshot()。
• 子流程优先封装为子 Operation，通过 round_by_op_result 接入。
• 高频轮询节点使用 mute=True。
• 失败节点返回 round_fail，不要吞掉关键错误。
• 修改代码后只对改动文件运行 ruff check。