python-action-guide

# MaaNTE Python CustomAction 编写指南 ## 架构定位 Python CustomAction 处理 Pipeline JSON 无法覆盖的复杂逻辑（图像算法、状态机、音频检测、MIDI 播放等）。**流程控制由 Pipeline 负责，Python 只处理难点。** 一句话：**Pipeline 管流程，Python 管难点。** 没有必要的 Python 逻辑会大大增加代码复杂度，造成下一位开发者开发调试极其困难。 所有坐标与图像以 **720p (1280×720)** 为基准。 ## 目录结构 ``` agent/ ├── main.py # 入口：venv、依赖安装、AgentServer 启动 ├── custom/ │ ├── __init__.py # from .action import * │ └── action/ │ ├── __init__.py # 所有 CustomAction 的 import 与 __all__ 注册 │ ├── Common/ # 通用工具（click、alt_click、logger、utils） │ │ ├── __init__.py │ │ ├── click.py │ │ ├── alt_click.py │ │ ├── logger.py │ │ └── utils.py # get_image、click_rect、match_template_in_region │ ├── AutoFish/ # 钓鱼业务（按功能聚合） │ ├── Tetris/ # 俄罗斯方块业务 │ │ ├── feats/ # 功能实现 │ │ └── utils/ # 工具类（棋盘、场景检测） │ ├── rhythm/ # 节奏任务 │ │ ├── feats/ │ │ └── utils/ │ ├── SoundTrigger/ # 音频触发业务 │ ├── Movement/ # 移动控制 │ ├── auto_piano/ # 自动钢琴 │ ├── pinkpaw/ # 粉爪大劫案 │ ├── DatasetCollection/ # 数据集采集 │ ├── auto_make_coffee.py # 单文件简单动作 │ ├── auto_tetris.py # 俄罗斯方块入口 │ ├── auto_f_scroll.py # 自动滚屏 │ ├── furniture_claim.py # 家具收取 │ ├── realtime_task.py # 实时任务调度 │ ├── withdraw_money_choose_item.py # 补货取钱物品选择 │ └── __init__.py # 所有 CustomAction 的导入与注册 └── utils/ ├── __init__.py # 子模块聚合导出 ├── logger.py # 日志系统（loguru/logging） ├── pienv.py # PI 环境变量 ├── maafocus.py # 用户可见消息（Pipeline focus 协议） ├── i18n.py # 多语言/翻译 ├── screen.py # 屏幕缩放/分辨率 ├── time.py # 时间工具 └── win32_process.py # Win32 窗口查找 ``` ## 注册机制 ### 装饰器注册 每个 CustomAction 类使用 `@AgentServer.custom_action("name")` 注册，名称必须与 Pipeline JSON 中 `custom_action` 的值一致： ```python from maa.agent.agent_server import AgentServer from maa.custom_action import CustomAction from maa.context import Context @AgentServer.custom_action("auto_make_coffee") class AutoMakeCoffee(CustomAction): def run( self, context: Context, argv: CustomAction.RunArg ) -> CustomAction.RunResult: ... ``` Pipeline JSON 中对应： ```jsonc { "action": { "type": "Custom", "param": { "custom_action": "auto_make_coffee", "custom_action_param": { "count": 10 } } } } ``` ### `__init__.py` 注册 所有 CustomAction 必须在 `agent/custom/action/__init__.py` 中导入并加入 `__all__`： ```python from .auto_make_coffee import * from .Tetris import * __all__ = [ "AutoMakeCoffee", "AutoTetris", "TetrisResetContext", "TetrisCheckVitalityAction", ] ``` 遗漏导入 = 动作不生效。 ## 命名 - **类名**：PascalCase，语义化。如 `AutoMakeCoffee`、`TetrisGamePlayer`、`SoundDodgeAction`。 - **注册名**：snake_case，与 Pipeline JSON 一致。如 `"auto_make_coffee"`、`"auto_tetris"`。 - **文件名**：snake_case。单文件简单动作直接用 `.py` 文件；复杂模块用目录 + `__init__.py`。 - **内部函数/变量**：snake_case，模块内部使用的加前导下划线 `_`。 ## Common 内置动作 以下 Custom Action 位于 `agent/custom/action/Common/`，可在 Pipeline 中直接使用，无需额外编写 Python 代码。 ### click_override 自定义点击。通过 `custom_action_param` 指定目标 rect，或使用当前识别结果的 box。 - 注册名：`click_override` - 参数 `custom_action_param`：`{ "target": [x, y, w, h] }` - 若未提供 `custom_action_param`，则使用 `argv.box`（识别结果 box） ```jsonc { "action": { "type": "Custom", "param": { "custom_action": "click_override", "custom_action_param": { "target": [100, 200, 50, 50] } } } } ``` ### alt_click Alt + 点击。先按下 Alt 键，再点击识别结果 box 位置，最后松开 Alt。 - 注册名：`alt_click` - 无需额外参数，点击位置由识别结果的 `box` 决定 ```jsonc { "recognition": { "type": "TemplateMatch", "param": { "template": "xxx.png" } }, "action": { "type": "Custom", "param": { "custom_action": "alt_click" } } } ``` ## Common 工具函数 `agent/custom/action/Common/utils.py` 提供常用辅助函数： | 函数 | 说明 | |------|------| | `get_image(controller)` | 截图，返回 numpy array | | `click_rect(controller, rect, delay)` | 点击指定 rect 的中心 | | `match_template_in_region(img, region, template, min_similarity, green_mask)` | 在区域内做模板匹配，返回 `(hit, score, x, y)` | ```python from Common.utils import get_image, click_rect, match_template_in_region img = get_image(controller) hit, score, x, y = match_template_in_region(img, [0, 0, 1280, 720], template, 0.8) if hit: click_rect(controller, [x, y, 50, 50]) ``` ## CustomAction 模板 ### 简单动作 ```python import json from maa.agent.agent_server import AgentServer from maa.custom_action import CustomAction from maa.context import Context from utils.logger import logger from utils.maafocus import PrintT @AgentServer.custom_action("my_action") class MyAction(CustomAction): def run( self, context: Context, argv: CustomAction.RunArg ) -> CustomAction.RunResult: PrintT(context, "my_action.started") params = {} if argv.custom_action_param: try: params = json.loads(argv.custom_action_param) except (json.JSONDecodeError, TypeError): pass controller = context.tasker.controller # 截图 controller.post_screencap().wait() img = controller.cached_image # 检查停止信号 if context.tasker.stopping: return CustomAction.RunResult(success=False) # 业务逻辑 ... logger.debug("识别结果: score=%.2f", score) return CustomAction.RunResult(success=True) ``` ### 带状态管理的动作 对于需要跨多次调用保持状态的场景（如连打计数），使用模块级全局变量，并通过独立的 reset action 重置： ```python _round_count = 0 _task_config = {} @AgentServer.custom_action("my_reset") class MyResetContext(CustomAction): def run(self, context, argv) -> CustomAction.RunResult: global _round_count, _task_config _round_count = 0 _task_config = {} return CustomAction.RunResult(success=True) @AgentServer.custom_action("my_action") class MyAction(CustomAction): def run(self, context, argv) -> CustomAction.RunResult: global _round_count _round_count += 1 ... ``` ## Controller API 通过 `context.tasker.controller` 访问，常用操作： ```python controller = context.tasker.controller # 截图 controller.post_screencap().wait() img = controller.cached_image # numpy array (BGR or BGRA) # 点击（绝对坐标，720p 基准） controller.post_touch_down(x, y).wait() time.sleep(0.005) # 短暂按下 controller.post_touch_up().wait() # 按键（虚拟键码） controller.post_key_down(0x46).wait() # F 键 time.sleep(0.1) controller.post_key_up(0x46).wait() # 组合键：先 Alt 再点 controller.post_key_down(0x12).wait() # VK_MENU (Alt) # ... 执行点击 ... controller.post_key_up(0x12).wait() ``` 常用虚拟键码： | 键 | 码 | |----|----| | Esc | 27 | | F | 70 | | E | 69 | | Q | 81 | | D | 0x44 | | F | 0x46 | | J | 0x4A | | K | 0x4B | ## Pipeline 集成 ### context.run_recognition 调用已定义的 Pipeline 识别节点： ```python # 基本调用 result = context.run_recognition("FurnitureOcrRec", image) # 带参数覆盖（动态 expected 文本、自定义 roi 等） result = context.run_recognition( "FurnitureOcrRec", image, pipeline_override={ "FurnitureOcrRec": { "recognition": {"param": {"expected": "仓鼠球"}} } }, ) if result and result.box: x, y, w, h = result.box.x, result.box.y, result.box.w, result.box.h ``` ### RecognitionDetail 结构 `context.run_recognition()` 返回 `RecognitionDetail`（`maa.define.RecognitionDetail`），结构如下： ```python @dataclass class RecognitionDetail: reco_id: int name: str algorithm: Union[AlgorithmEnum, str] hit: bool # 是否有命中 box: Optional[Rect] # 最佳命中的 box all_results: List[RecognitionResult] # 全部原始结果 filtered_results: List[RecognitionResult] # 过滤后的结果（常用） best_result: Optional[RecognitionResult] # 最佳结果 raw_detail: Dict raw_image: numpy.ndarray draw_images: List[numpy.ndarray] ``` 关键方法/属性： ```python result = context.run_recognition("MyNode", image) # 是否存在命中 if result and result.hit: # 遍历所有命中（正确的多结果方式，不是 .all） for r in result.filtered_results: # r.box — Rect(x, y, w, h) # 根据算法类型，r 还有不同属性： # OCRResult: r.text (str), r.score (float) # ColorMatchResult: r.count (int) # TemplateMatchResult: r.score (float) # result.box — 最佳命中的 Rect（快捷方式） # result.box.x, result.box.y, result.box.w, result.box.h 都是 int ``` ### 多结果模式 **点击所有命中 rect（如 ColorMatch connected 组件、OCR 多匹配）：** ```python def _filtered_boxes(result): """返回 filtered_results 中所有命中的 Rect 列表""" if result is None or not result.hit: return [] return [r.box for r in result.filtered_results if r.box is not None] result = context.run_recognition("SomeColorMatch", image) for box in _filtered_boxes(result): cx, cy = box.x + box.w // 2, box.y + box.h // 2 controller.post_touch_down(cx, cy).wait() time.sleep(0.001) controller.post_touch_up().wait() ``` **遍历 OCR 多结果并提取文本：** ```python result = context.run_recognition("SomeOCRNode", image) for r in result.filtered_results if result else []: text = getattr(r, "text", "") # OCRResult 有 .text，其他类型无 box = r.box # Rect # 处理 text 和 box ... ``` **常见错误：** - ~~`result.all`~~ → 不存在，应该用 `result.filtered_results` - ~~`box.text`~~ → `Rect` 没有 `.text`，应该用 `getattr(r, "text", "")` 在结果对象上取 - ~~`ColorMatchResult.box.text`~~ → ColorMatchResult 只有 `.box`（Rect）和 `.count`（int），无 `.text` **`box` 类型注意**：MaaFramework Python 绑定返回的 `.box` 实际是 `list [x,y,w,h]` 而非 `Rect` 对象（虽然类型标注为 `Rect`），转换时需兼容： ```python def _box_to_rect(box): if isinstance(box, (list, tuple)): return list(box) return [box.x, box.y, box.w, box.h] ``` **真实 `raw_detail` 结构参考（ColorMatch）：** ```json { "reco_id": 400000008, "algorithm": "ColorMatch", "box": [256, 384, 19, 18], "detail": { "all": [ {"box": [256, 384, 19, 18], "count": 185}, {"box": [256, 531, 19, 17], "count": 182} ], "best": {"box": [256, 384, 19, 18], "count": 185}, "filtered": [ {"box": [256, 384, 19, 18], "count": 185}, {"box": [256, 531, 19, 17], "count": 182} ] } } ``` - `result.filtered_results[i].box` → `list [x,y,w,h]` - `result.filtered_results[i].count` → `int`（ColorMatch）/ `result.filtered_results[i].text` → `str`（OCR） - `result.raw_detail` 包含完整 JSON-serializable dict，可作为兜底访问 ### context.run_recognition_direct 在 Python 中直接调用 MaaFramework 识别算法（无需 Pipeline 节点定义）： ```python from maa.pipeline import JRecognitionType, JOCR detail = context.run_recognition_direct( JRecognitionType.OCR, JOCR(roi=[x, y, w, h]), frame