From e0076601dd5db92461e9fb6a26bc91d08d398d95 Mon Sep 17 00:00:00 2001 From: overflow65537 Date: Sun, 28 Jun 2026 22:55:13 +0800 Subject: [PATCH] =?UTF-8?q?feat(customs):=20=E6=96=B0=E5=A2=9E=20Expressio?= =?UTF-8?q?nRecognition=20=E8=A1=A8=E8=BE=BE=E5=BC=8F=E8=AF=86=E5=88=AB=20?= =?UTF-8?q?custom=20=E6=95=99=E7=A8=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: Cursor --- .../ExpressionRecognition.py | 449 ++++++++++++++++++ .../expression-recognition/README.md | 91 ++++ .../expression-recognition/maahub_meta.json | 20 + .../expression-recognition/pipeline.json | 67 +++ 4 files changed, 627 insertions(+) create mode 100644 Storage/customs/overflow65537/expression-recognition/ExpressionRecognition.py create mode 100644 Storage/customs/overflow65537/expression-recognition/README.md create mode 100644 Storage/customs/overflow65537/expression-recognition/maahub_meta.json create mode 100644 Storage/customs/overflow65537/expression-recognition/pipeline.json diff --git a/Storage/customs/overflow65537/expression-recognition/ExpressionRecognition.py b/Storage/customs/overflow65537/expression-recognition/ExpressionRecognition.py new file mode 100644 index 0000000..1532410 --- /dev/null +++ b/Storage/customs/overflow65537/expression-recognition/ExpressionRecognition.py @@ -0,0 +1,449 @@ +import ast +from collections.abc import Iterable +import json +import re +from typing import Any + +from maa.custom_recognition import CustomRecognition +from maa.define import OCRResult + + +PLACEHOLDER_PATTERN = re.compile(r"\{([^{}]+)\}") +NUMBER_PATTERN = re.compile(r"[-+]?\d+(?:\.\d+)?") + + +class NodeResolutionError(ValueError): + + def __init__(self, message: str, payload: dict[str, Any]): + super().__init__(message) + self.payload = payload + + +class ExpressionRecognition(CustomRecognition): + + def analyze( + self, + context, + argv: CustomRecognition.AnalyzeArg, + ) -> CustomRecognition.AnalyzeResult | None: + params = self._parse_params(argv.custom_recognition_param) + expression = params.get("expression") + if not isinstance(expression, str) or not expression.strip(): + return CustomRecognition.AnalyzeResult(box=None, detail={"status": "invalid expression"}) + + image = argv.image + values_cache: dict[str, int | float] = {} + placeholder_mapping: dict[str, str] = {} + node_results: dict[str, Any] = {} + + def replace_placeholder(match: re.Match[str]) -> str: + node_name = match.group(1).strip() + if not node_name: + raise ValueError("empty placeholder") + + variable_name = placeholder_mapping.get(node_name) + if variable_name is None: + variable_name = f"_value_{len(placeholder_mapping)}" + placeholder_mapping[node_name] = variable_name + try: + value, node_result = self._resolve_node_value(context, image, node_name) + except NodeResolutionError as exc: + node_results[node_name] = exc.payload + raise + + values_cache[variable_name] = value + node_results[node_name] = node_result + return variable_name + + try: + python_expression = PLACEHOLDER_PATTERN.sub(replace_placeholder, expression) + python_expression = self._normalize_expression(python_expression) + parsed = ast.parse(python_expression, mode="eval") + self._validate_ast(parsed, set(values_cache.keys())) + result = eval( + compile(parsed, "", "eval"), + {"__builtins__": {}}, + values_cache, + ) + except (ValueError, SyntaxError, TypeError, ZeroDivisionError) as exc: + detail = { + "status": "invalid expression", + "reason": str(exc), + "expression": expression, + "resolved_expression": locals().get("python_expression", expression), + "resolved_values": values_cache, + "node_results": node_results, + } + detail["summary"] = self._format_summary(detail) + return CustomRecognition.AnalyzeResult( + box=None, + detail=detail, + ) + + if type(result) is not bool: + detail = { + "status": "expression did not evaluate to boolean", + "expression": expression, + "resolved_expression": python_expression, + "resolved_values": values_cache, + "node_results": node_results, + } + detail["summary"] = self._format_summary(detail) + return CustomRecognition.AnalyzeResult( + box=None, + detail=detail, + ) + + if not result: + detail = { + "status": "expression evaluated to false", + "expression": expression, + "resolved_expression": python_expression, + "resolved_values": values_cache, + "node_results": node_results, + } + detail["summary"] = self._format_summary(detail) + return CustomRecognition.AnalyzeResult( + box=None, + detail=detail, + ) + + detail = { + "status": "success", + "expression": expression, + "resolved_expression": python_expression, + "resolved_values": values_cache, + "node_results": node_results, + } + detail["summary"] = self._format_summary(detail) + return CustomRecognition.AnalyzeResult( + box=(0, 0, 100, 100), + detail=detail, + ) + + def _parse_params(self, raw_params: Any) -> dict[str, Any]: + if isinstance(raw_params, dict): + return raw_params + if isinstance(raw_params, str): + try: + parsed = json.loads(raw_params) + except json.JSONDecodeError: + return {} + if isinstance(parsed, dict): + return parsed + return {} + + def _resolve_node_value(self, context, image, node_name: str) -> tuple[int | float, dict[str, Any]]: + node_data = context.get_node_data(node_name) or {} + box_index = self._get_box_index(node_data) + recognition = context.run_recognition(node_name, image) + payload = { + "node_data": self._to_jsonable(node_data), + "box_index": box_index, + "recognition": self._summarize_recognition(recognition), + } + if not (recognition and recognition.hit): + raise NodeResolutionError(f"node {node_name} has no OCR result", payload) + + text = self._extract_text(recognition, box_index=box_index) + payload["extracted_text"] = text + if text is None: + raise NodeResolutionError(f"node {node_name} has no OCR text", payload) + + match = NUMBER_PATTERN.search(text) + if match is None: + raise NodeResolutionError(f"node {node_name} has no numeric OCR text", payload) + + number_text = match.group(0) + if "." in number_text: + value = float(number_text) + else: + value = int(number_text) + + payload["value"] = value + return value, payload + + def _get_box_index(self, node_data: Any) -> int | None: + if not isinstance(node_data, dict): + return None + + recognition = node_data.get("recognition") + if not isinstance(recognition, dict): + return None + + param = recognition.get("param") + if not isinstance(param, dict): + return None + + box_index = param.get("box_index") + if isinstance(box_index, int) and box_index >= 0: + return box_index + return None + + def _extract_text(self, result: Any, box_index: int | None = None) -> str | None: + if box_index is not None: + indexed_text = self._extract_text_by_index(result, box_index) + if indexed_text is not None: + return indexed_text + + if isinstance(result, dict): + direct_text = result.get("text") + if isinstance(direct_text, str): + return direct_text + + for key in ("best", "best_result", "detail", "filtered", "filtered_results", "all"): + if key in result: + nested_text = self._extract_text(result[key]) + if nested_text is not None: + return nested_text + return None + + if isinstance(result, Iterable) and not isinstance(result, (str, bytes, bytearray)): + for item in result: + item_text = self._extract_text(item) + if item_text is not None: + return item_text + return None + + if isinstance(result, OCRResult): + return result.text + + text = getattr(result, "text", None) + if isinstance(text, str): + return text + + detail = getattr(result, "detail", None) + if detail is not None and detail is not result: + detail_text = self._extract_text(detail) + if detail_text is not None: + return detail_text + + best_result = getattr(result, "best_result", None) + if best_result is not None and best_result is not result: + best_text = self._extract_text(best_result) + if best_text is not None: + return best_text + + filtered_results = getattr(result, "filtered_results", None) + if filtered_results is not None and filtered_results is not result: + for item in filtered_results: + item_text = self._extract_text(item) + if item_text is not None: + return item_text + + return None + + def _extract_text_by_index(self, result: Any, box_index: int) -> str | None: + children = self._extract_children(result) + if children is None or not 0 <= box_index < len(children): + return None + + return self._extract_text(children[box_index]) + + def _extract_children(self, result: Any) -> list[Any] | None: + if isinstance(result, dict): + for key in ("sub_results", "detail", "all", "filtered", "filtered_results"): + value = self._coerce_sequence(result.get(key)) + if value is not None: + return value + + raw_detail = result.get("raw_detail") + if raw_detail is not None: + raw_children = self._extract_children(raw_detail) + if raw_children is not None: + return raw_children + return None + + sub_results = self._coerce_sequence(getattr(result, "sub_results", None)) + if sub_results is not None: + return sub_results + + detail = self._coerce_sequence(getattr(result, "detail", None)) + if detail is not None: + return detail + + raw_detail = getattr(result, "raw_detail", None) + if raw_detail is not None: + raw_children = self._extract_children(raw_detail) + if raw_children is not None: + return raw_children + + all_results = self._coerce_sequence(getattr(result, "all", None)) + if all_results is not None: + return all_results + + filtered_results = self._coerce_sequence(getattr(result, "filtered_results", None)) + if filtered_results is not None: + if len(filtered_results) == 1: + nested_children = self._extract_children(filtered_results[0]) + if nested_children is not None: + return nested_children + return filtered_results + + best_result = getattr(result, "best_result", None) + if best_result is not None: + best_children = self._extract_children(best_result) + if best_children is not None: + return best_children + + return None + + def _coerce_sequence(self, value: Any) -> list[Any] | None: + if value is None or isinstance(value, (str, bytes, bytearray, dict)): + return None + + if isinstance(value, list): + return value + + try: + return list(value) + except TypeError: + return None + + def _summarize_recognition(self, value: Any) -> Any: + summary = self._to_jsonable(value) + if isinstance(summary, dict): + summary.pop("node_data", None) + return summary + + def _format_summary(self, detail: dict[str, Any]) -> str: + lines = [ + f"status: {detail.get('status')}", + f"expression: {detail.get('expression')}", + f"resolved_expression: {detail.get('resolved_expression')}", + ] + + reason = detail.get("reason") + if reason: + lines.append(f"reason: {reason}") + + resolved_values = detail.get("resolved_values") or {} + lines.append("resolved_values:") + if resolved_values: + for key, value in resolved_values.items(): + lines.append(f" {key}: {value}") + else: + lines.append(" ") + + lines.append("node_results:") + node_results = detail.get("node_results") or {} + if node_results: + for node_name, node_result in node_results.items(): + lines.append(f" - {node_name}") + if isinstance(node_result, dict): + lines.append(f" box_index: {node_result.get('box_index')}") + lines.append(f" extracted_text: {node_result.get('extracted_text')}") + if "value" in node_result: + lines.append(f" value: {node_result.get('value')}") + + recognition = node_result.get("recognition") + if isinstance(recognition, dict): + lines.append(f" recognition_type: {recognition.get('type')}") + lines.append(f" recognition_name: {recognition.get('name')}") + lines.append(f" recognition_algorithm: {recognition.get('algorithm')}") + lines.append(f" recognition_hit: {recognition.get('hit')}") + lines.append(f" recognition_box: {recognition.get('box')}") + else: + lines.append(f" {node_result}") + else: + lines.append(" ") + + return "\n".join(lines) + + def _to_jsonable(self, value: Any, depth: int = 0) -> Any: + if depth >= 6: + return self._safe_repr(value) + + if value is None or isinstance(value, (str, int, float, bool)): + return value + + if isinstance(value, dict): + return { + str(key): self._to_jsonable(item, depth + 1) + for key, item in value.items() + } + + if isinstance(value, OCRResult): + return { + "type": type(value).__name__, + "text": value.text, + } + + if isinstance(value, Iterable) and not isinstance(value, (str, bytes, bytearray)): + return [self._to_jsonable(item, depth + 1) for item in value] + + result: dict[str, Any] = {"type": type(value).__name__} + for attr in ( + "hit", + "text", + "box", + "score", + "detail", + "raw_detail", + "best_result", + "filtered_results", + "sub_results", + "all", + "algorithm", + "name", + "reco_id", + ): + if hasattr(value, attr): + attr_value = getattr(value, attr) + if attr_value is not None: + result[attr] = self._to_jsonable(attr_value, depth + 1) + + if len(result) == 1: + result["repr"] = self._safe_repr(value) + return result + + def _safe_repr(self, value: Any, limit: int = 240) -> str: + text = repr(value) + if len(text) <= limit: + return text + return f"{text[:limit]}..." + + def _normalize_expression(self, expression: str) -> str: + normalized = expression.replace("&&", " and ").replace("||", " or ") + normalized = re.sub(r"!(?!=)", " not ", normalized) + return normalized + + def _validate_ast(self, node: ast.AST, allowed_names: set[str]) -> None: + for child in ast.walk(node): + if isinstance( + child, + ( + ast.Expression, + ast.Load, + ast.BinOp, + ast.BoolOp, + ast.UnaryOp, + ast.Compare, + ast.Name, + ast.Constant, + ast.Add, + ast.Sub, + ast.Mult, + ast.Div, + ast.Mod, + ast.And, + ast.Or, + ast.Not, + ast.UAdd, + ast.USub, + ast.Eq, + ast.NotEq, + ast.Lt, + ast.LtE, + ast.Gt, + ast.GtE, + ), + ): + if isinstance(child, ast.Name) and child.id not in allowed_names: + raise ValueError(f"unexpected name {child.id}") + if isinstance(child, ast.Constant) and not isinstance( + child.value, (int, float, bool) + ): + raise ValueError("unsupported constant") + continue + raise ValueError(f"unsupported syntax {type(child).__name__}") diff --git a/Storage/customs/overflow65537/expression-recognition/README.md b/Storage/customs/overflow65537/expression-recognition/README.md new file mode 100644 index 0000000..0e6a8c6 --- /dev/null +++ b/Storage/customs/overflow65537/expression-recognition/README.md @@ -0,0 +1,91 @@ +# ExpressionRecognition 表达式识别 + +自定义识别器：在 pipeline 里用 `{节点名}` 引用其他节点的 OCR 数值，通过表达式做算术与逻辑比较，结果为 `true` 时识别命中。 + +适用场景：当前值是否达标、多路 OCR 数值加权求和后再判断等，不必为每种组合单独写节点。 + +## 文件说明 + +| 文件 | 说明 | +|------|------| +| `ExpressionRecognition.py` | 识别器实现 | +| `pipeline.json` | 最小 pipeline 示例 | +| `maahub_meta.json` | MaaHub 元信息 | + +## 注册方式 + +将 `ExpressionRecognition.py` 放入 agent 工程，在 agent 入口注册： + +```python +from maa.agent.agent_server import AgentServer +from ExpressionRecognition import ExpressionRecognition + +@AgentServer.custom_recognition("ExpressionRecognition") +class Agent_ExpressionRecognition(ExpressionRecognition): + pass +``` + +注册名 `"ExpressionRecognition"` 需与 pipeline 中 `custom_recognition` 字段一致。 + +## Pipeline 用法 + +在节点上使用 `Custom` 识别，通过 `custom_recognition_param.expression` 传入表达式： + +```json +"数值达标判断": { + "recognition": { + "type": "Custom", + "param": { + "custom_recognition": "ExpressionRecognition", + "custom_recognition_param": { + "expression": "{OCR-当前数值}>={OCR-目标数值}" + } + } + }, + "next": ["下一步节点"] +} +``` + +多节点加权示例： + +```json +"加权分数计算": { + "recognition": { + "type": "Custom", + "param": { + "custom_recognition": "ExpressionRecognition", + "custom_recognition_param": { + "expression": "{OCR-当前数值}*{OCR-倍率A}+{OCR-目标数值}*{OCR-倍率B}>={OCR-目标数值}" + } + } + } +} +``` + +完整可运行示例见同目录 `pipeline.json`。 + +## 表达式语法 + +- **占位符**：`{节点名}`，运行时对该节点执行识别，从 OCR 结果中提取第一个数字（支持整数与小数） +- **运算符**：`+` `-` `*` `/` `%`，比较 `>` `>=` `<` `<=` `==` `!=`，逻辑 `&&` `||` `!`（会转换为 Python 的 `and` / `or` / `not`） +- **返回值**：表达式必须求值为 **布尔值**；为 `true` 时返回固定 box `(0, 0, 100, 100)` 表示命中 + +被引用的节点需能 OCR 出数字文本。若节点使用 `And` 等多结果识别，可在该节点配置 `box_index` 指定取第几个子结果。 + +## 识别结果 detail + +`detail` 中包含调试信息，便于排查： + +| 字段 | 含义 | +|------|------| +| `status` | `success` / `expression evaluated to false` / `invalid expression` 等 | +| `expression` | 原始表达式 | +| `resolved_expression` | 占位符替换后的表达式 | +| `resolved_values` | 各占位符解析出的数值 | +| `node_results` | 各引用节点的识别摘要 | +| `summary` | 上述信息的可读文本 | + +## 依赖 + +- Python 3.12+ +- MaaFramework Agent SDK（`maa.custom_recognition`） diff --git a/Storage/customs/overflow65537/expression-recognition/maahub_meta.json b/Storage/customs/overflow65537/expression-recognition/maahub_meta.json new file mode 100644 index 0000000..009dd18 --- /dev/null +++ b/Storage/customs/overflow65537/expression-recognition/maahub_meta.json @@ -0,0 +1,20 @@ +{ + "id": "overflow65537/expression-recognition", + "title": "ExpressionRecognition 表达式识别", + "description": "自定义识别：引用其他 pipeline 节点的 OCR 数值，通过表达式做算术与逻辑比较，判断是否命中。", + "author": "overflow65537", + "source": "MAA_Punish", + "sourceGithub": "https://github.com/overflow65537/MaaHub", + "tags": ["custom", "recognition", "expression", "ocr", "python"], + "createdAt": "2026-06-28", + "updatedAt": "2026-06-28", + "version": "0.1.0", + "mfwVersion": "5.10.4", + "entry": "ExpressionRecognition.py", + "readme": "./README.md", + "status": "stable", + "type": "custom", + "language": "python", + "runtime": "python 3.12", + "dependencies": ["maa framework sdk"] +} diff --git a/Storage/customs/overflow65537/expression-recognition/pipeline.json b/Storage/customs/overflow65537/expression-recognition/pipeline.json new file mode 100644 index 0000000..ce3fe53 --- /dev/null +++ b/Storage/customs/overflow65537/expression-recognition/pipeline.json @@ -0,0 +1,67 @@ +{ + "OCR-当前数值": { + "recognition": { + "type": "OCR", + "param": { + "roi": [100, 200, 80, 40], + "expected": "^\\d*$", + "only_rec": true + } + } + }, + "OCR-目标数值": { + "recognition": { + "type": "OCR", + "param": { + "roi": [100, 100, 80, 40], + "expected": "^\\d*$", + "only_rec": true + } + } + }, + "OCR-倍率A": { + "recognition": { + "type": "OCR", + "param": { + "roi": [200, 200, 60, 40], + "expected": "^\\d*$", + "only_rec": true + } + } + }, + "OCR-倍率B": { + "recognition": { + "type": "OCR", + "param": { + "roi": [200, 260, 60, 40], + "expected": "^\\d*$", + "only_rec": true + } + } + }, + "数值达标判断": { + "recognition": { + "type": "Custom", + "param": { + "custom_recognition": "ExpressionRecognition", + "custom_recognition_param": { + "expression": "{OCR-当前数值}>={OCR-目标数值}" + } + } + }, + "next": ["下一步节点"] + }, + "加权分数计算": { + "recognition": { + "type": "Custom", + "param": { + "custom_recognition": "ExpressionRecognition", + "custom_recognition_param": { + "expression": "{OCR-当前数值}*{OCR-倍率A}+{OCR-目标数值}*{OCR-倍率B}>={OCR-目标数值}" + } + } + }, + "next": ["下一步节点"] + }, + "下一步节点": {} +}