diff --git a/aao/app.py b/aao/app.py index e402bf9..eb7c370 100644 --- a/aao/app.py +++ b/aao/app.py @@ -53,6 +53,7 @@ from aao.measure.overlay import OverlayWindow # noqa: E402 from aao.measure.worker import MeasurementWorker # noqa: E402 from aao.timeline.editor_window import EditorWindow # noqa: E402 +from aao.types import MeasureState # noqa: E402 from aao.ui.about_page import AboutPage # noqa: E402 from aao.ui.background import BackgroundContainer # noqa: E402 from aao.ui.calibration_page import CalibrationPage # noqa: E402 @@ -493,7 +494,7 @@ def _reset_measure_timer(self, node_name: str) -> None: logger.debug("计时器重置（pipeline 节点: %s）", node_name) self.worker.request_reset_timer() - def _on_measure_state(self, state: dict) -> None: + def _on_measure_state(self, state: MeasureState) -> None: from aao.core.timing.time_source import format_timer total = state.get("totalElapsedFrames", 0) diff --git a/aao/core/avatar.py b/aao/core/avatar.py index 56b0404..c1b851e 100644 --- a/aao/core/avatar.py +++ b/aao/core/avatar.py @@ -11,7 +11,7 @@ import json import time from pathlib import Path -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, TypedDict, cast import numpy as np @@ -25,6 +25,12 @@ _DETAIL_WAIT = 1 # 等详情页打开 +class Slot(TypedDict): + flag_rect: tuple[int, int, int, int] + avatar_rect: tuple[int, int, int, int] + click_pos: tuple[int, int] + + def _avatar_dir() -> Path: from aao.utils.runtime_paths import project_root @@ -39,7 +45,14 @@ def _match_avatar_roi_offset() -> tuple[int, int, int, int]: path = project_root() / "resource" / "base" / "pipeline" / "reco.json" raw = json.loads(path.read_text(encoding="utf-8")) - offset = raw["MatchAvatar"].get("roi_offset", [0, 0, 0, 0]) + if not isinstance(raw, dict): + return (0, 0, 0, 0) + data = cast(dict[str, object], raw) + match_avatar = data.get("MatchAvatar") + if not isinstance(match_avatar, dict): + return (0, 0, 0, 0) + match_avatar = cast(dict[str, object], match_avatar) + offset = match_avatar.get("roi_offset", [0, 0, 0, 0]) return tuple(int(v) for v in offset) # type: ignore[return-value] @@ -56,7 +69,9 @@ def _get_char_id(oper_name: str) -> str: if not mapping_path.exists(): return "" mapping = json.loads(mapping_path.read_text(encoding="utf-8")) - return mapping.get(oper_name, "") + if not isinstance(mapping, dict): + return "" + return str(cast(dict[str, object], mapping).get(oper_name, "")) def _normalize_name(name: str) -> str: @@ -71,14 +86,14 @@ def _normalize_name(name: str) -> str: def detect_slots( context: Context, image: np.ndarray, -) -> list[dict]: +) -> list[Slot]: """用 pipeline 节点 DetectSlots 检测待部署区所有干员槽位。""" reco_detail = context.run_recognition("DetectSlots", image) if not reco_detail or not reco_detail.hit: return [] - slots = [] + slots: list[Slot] = [] for result in reco_detail.all_results: box = getattr(result, "box", None) if box is None: @@ -217,7 +232,7 @@ def _ocr_oper_name(context: Context, detail_img: np.ndarray) -> str | None: def _save_avatar_from_image( image: np.ndarray, - slot: dict, + slot: Slot, oper_name: str, ) -> bool: """从截图截取槽位头像并存盘。""" diff --git a/aao/core/geometry/map_loader.py b/aao/core/geometry/map_loader.py index b773d94..5cd8ace 100644 --- a/aao/core/geometry/map_loader.py +++ b/aao/core/geometry/map_loader.py @@ -10,7 +10,9 @@ import json from pathlib import Path +from typing import cast +from aao.types import JsonObject from aao.utils.logger import logger from aao.utils.runtime_paths import project_root @@ -25,7 +27,8 @@ def _level_codes() -> dict[str, str]: path = project_root() / "data" / "level_codes.json" if not path.exists(): return {} - return json.loads(path.read_text(encoding="utf-8")) + data = json.loads(path.read_text(encoding="utf-8")) + return cast(dict[str, str], data) if isinstance(data, dict) else {} def find_map_file(code: str) -> Path | None: @@ -52,17 +55,21 @@ def find_map_file(code: str) -> Path | None: return None -def load_map(code: str) -> dict | None: +def load_map(code: str) -> JsonObject | None: """加载关卡数据。code 如 '1-7'。""" path = find_map_file(code) if path is None: logger.error("未找到关卡 %s 的地图数据", code) return None data = json.loads(path.read_text(encoding="utf-8")) - logger.info( - "加载关卡 %s (%s): %dx%d", code, data.get("name", "?"), data["height"], data["width"] - ) - return data + if not isinstance(data, dict): + logger.error("关卡 %s 的地图数据格式无效", code) + return None + name = str(cast(JsonObject, data).get("name", "?")) + height = int(cast(JsonObject, data)["height"]) + width = int(cast(JsonObject, data)["width"]) + logger.info("加载关卡 %s (%s): %dx%d", code, name, height, width) + return cast(JsonObject, data) def list_codes() -> list[str]: diff --git a/aao/core/geometry/view.py b/aao/core/geometry/view.py index 932271f..0ec6620 100644 --- a/aao/core/geometry/view.py +++ b/aao/core/geometry/view.py @@ -13,6 +13,8 @@ import numpy as np +from aao.types import JsonObject + # 投影参数（prts-plus ViewCalculationConfig） _FROM_RATIO = 9 / 16 _NEAR = 0.3 @@ -69,7 +71,7 @@ def _build_matrix(view_offset: list[float], side: bool) -> np.ndarray: def transform_map_to_view( - level_data: dict, + level_data: JsonObject, side: bool = False, ) -> list[list[tuple[float, float]]]: """把关卡数据投影为屏幕坐标（0-1 比例）。 diff --git a/aao/core/timing/tick.py b/aao/core/timing/tick.py index 2090a95..c19da99 100644 --- a/aao/core/timing/tick.py +++ b/aao/core/timing/tick.py @@ -12,6 +12,7 @@ from __future__ import annotations import bisect +from typing import cast import numpy as np @@ -245,9 +246,11 @@ def detect_negative_cost(frame: np.ndarray) -> bool: region = frame[y : y + h, x : x + w].astype(np.int16) # (h, w, 3) BGR white = (region > config.WHITE_THRESHOLD).all(axis=2) # (h, w) bool + white_rows = cast(np.ndarray, white) # 逐行求最长连续纯白 run：对每行做行内累计，遇非白清零，取全局最大。 - for row in white: # pyright: ignore[reportGeneralTypeIssues] + for row_index in range(int(white_rows.shape[0])): + row = cast(np.ndarray, white_rows[row_index]) if not row.any(): continue # 累计连续 True 长度：cumsum 在非白处“断点”归零的经典向量化写法。 diff --git a/aao/measure/api_server.py b/aao/measure/api_server.py index 91c6d43..b0dbd8e 100644 --- a/aao/measure/api_server.py +++ b/aao/measure/api_server.py @@ -15,6 +15,7 @@ import websockets +from aao.types import MeasureState from aao.utils.logger import logger DEFAULT_PORT = 2606 @@ -25,7 +26,7 @@ class ApiServer: def __init__( self, - get_state: Callable[[], dict], + get_state: Callable[[], MeasureState], host: str = "localhost", port: int = DEFAULT_PORT, rate_hz: float = 60.0, @@ -34,7 +35,7 @@ def __init__( self.host = host self.port = port self.rate_hz = rate_hz - self._clients: set = set() + self._clients: set[Any] = set() self._thread: threading.Thread | None = None def start(self) -> None: diff --git a/aao/measure/overlay.py b/aao/measure/overlay.py index 9ba534b..2fdda45 100644 --- a/aao/measure/overlay.py +++ b/aao/measure/overlay.py @@ -20,6 +20,7 @@ from PySide6.QtWidgets import QApplication, QLabel, QSizeGrip, QVBoxLayout, QWidget from aao.core.timing.time_source import format_timer +from aao.types import MeasureState from aao.ui import floating_state from aao.ui.window_snap import ( create_snap_follow, @@ -131,7 +132,7 @@ def reset_layout(self, x: int, y: int) -> None: self.move(x, y) self._save_window_state() - def on_state(self, state: dict) -> None: + def on_state(self, state: MeasureState) -> None: running = state.get("isRunning", False) cf = state.get("currentFrame") total = state.get("totalFramesInCycle", 0) diff --git a/aao/measure/worker.py b/aao/measure/worker.py index 70140e5..8263aac 100644 --- a/aao/measure/worker.py +++ b/aao/measure/worker.py @@ -15,6 +15,7 @@ from aao.core.timing.calibration import FullCalibrationData from aao.core.timing.time_source import TimeSource +from aao.types import MeasureState from aao.utils.logger import logger if TYPE_CHECKING: @@ -47,13 +48,13 @@ def __init__( self._running = False self._reset_requested = False self._consecutive_errors = 0 - self._latest: dict = {} + self._latest: MeasureState = {} self._lock = threading.Lock() @property - def latest_state(self) -> dict: + def latest_state(self) -> MeasureState: with self._lock: - return dict(self._latest) + return self._latest.copy() def run(self) -> None: self._running = True @@ -98,7 +99,7 @@ def run(self) -> None: time.sleep(backoff) continue - state = { + state: MeasureState = { "isRunning": self.time_source.is_running, "currentFrame": self.time_source.current_frame_in_cycle, "totalFramesInCycle": self.time_source.total_frames_in_cycle, diff --git a/aao/resources/updater.py b/aao/resources/updater.py index fba019f..8a83dc2 100644 --- a/aao/resources/updater.py +++ b/aao/resources/updater.py @@ -25,9 +25,11 @@ from collections.abc import Callable from dataclasses import dataclass from pathlib import Path +from typing import cast from aao import __version__ from aao.resources import syncer +from aao.types import JsonObject from aao.utils.logger import logger, setup_logging from aao.utils.runtime_paths import is_frozen, project_root @@ -93,7 +95,8 @@ def check_software(self) -> ReleaseInfo | None: if token: req.add_header("Authorization", f"Bearer {token}") with urllib.request.urlopen(req, timeout=10) as resp: - data = json.loads(resp.read()) + raw = json.loads(resp.read()) + data = cast(JsonObject, raw) if isinstance(raw, dict) else {} except Exception as e: # noqa: BLE001 logger.warning("检查更新失败: %s", e) return None @@ -140,14 +143,12 @@ def _fetch_changelog_since(self, current: str) -> str: try: req = _make_request(_RELEASES_LIST_API, _settings_github_token()) with urllib.request.urlopen(req, timeout=10) as resp: - releases = json.loads(resp.read()) + raw = json.loads(resp.read()) + releases = cast(list[JsonObject], raw) if isinstance(raw, list) else [] except Exception as e: # noqa: BLE001 logger.warning("拉取 release 列表失败，降级为单版 changelog: %s", e) return "" - if not isinstance(releases, list): - return "" - # 筛 version > current 的，按版本降序排 entries: list[tuple[str, str]] = [] # (version, body) for r in releases: @@ -209,13 +210,13 @@ def update_resources( progress_cb(r.message) return results - def update_all(self, progress_cb: Callable[[str], None] | None = None) -> dict: + def update_all(self, progress_cb: Callable[[str], None] | None = None) -> JsonObject: """检查软件更新 + 更新资源。 Returns: {"software": ReleaseInfo | None, "resources": [SyncResult]} """ - result: dict = {} + result: JsonObject = {} if progress_cb: progress_cb("检查软件更新...") @@ -417,7 +418,7 @@ def apply_update(self, zip_path: Path) -> None: _DOWNLOAD_BACKOFF_SEC = 1.0 -def _pick_win_asset(assets: list) -> AssetInfo | None: +def _pick_win_asset(assets: list[JsonObject]) -> AssetInfo | None: """从 release assets 里选 win-x64 zip。""" for a in assets: name = str(a.get("name", "")).lower() @@ -437,7 +438,9 @@ def _settings_proxy() -> str | None: path = project_root() / "config" / "settings.json" if not path.exists(): return None - proxy = json.loads(path.read_text(encoding="utf-8")).get("proxy", "") + raw = json.loads(path.read_text(encoding="utf-8")) + data = cast(JsonObject, raw) if isinstance(raw, dict) else {} + proxy = data.get("proxy", "") return str(proxy).strip() or None except Exception: # noqa: BLE001 return None @@ -450,7 +453,9 @@ def _settings_github_token() -> str | None: path = project_root() / "config" / "settings.json" if not path.exists(): return None - enc = json.loads(path.read_text(encoding="utf-8")).get("github_token_enc", "") + raw = json.loads(path.read_text(encoding="utf-8")) + data = cast(JsonObject, raw) if isinstance(raw, dict) else {} + enc = data.get("github_token_enc", "") return decrypt_text(str(enc)) if enc else None except Exception: # noqa: BLE001 return None diff --git a/aao/timeline/editor_window.py b/aao/timeline/editor_window.py index 5bc24ea..651cde2 100644 --- a/aao/timeline/editor_window.py +++ b/aao/timeline/editor_window.py @@ -12,6 +12,7 @@ from __future__ import annotations import json +from typing import cast from PySide6.QtCore import QEvent, QObject, QRect, Qt, Signal from PySide6.QtGui import QColor, QFont, QPainter, QPalette @@ -316,6 +317,7 @@ def _restore_side_panel_state(self) -> None: s = load_settings() states = s.get("collapsible_sections", {}) if isinstance(states, dict) and "timeline_side_panel" in states: + states = cast(dict[str, object], states) self._set_side_panel_visible(bool(states["timeline_side_panel"])) def _style_frame_label(self) -> None: diff --git a/aao/timeline/model.py b/aao/timeline/model.py index b86c5c0..b911d63 100644 --- a/aao/timeline/model.py +++ b/aao/timeline/model.py @@ -17,11 +17,19 @@ from __future__ import annotations from dataclasses import dataclass, field -from typing import Any +from typing import Any, cast from aao.core.battle.action import ActionType, DirectionType +def _empty_actions() -> list[TimelineAction]: + return [] + + +def _empty_candidates() -> list[str]: + return [] + + @dataclass class TimelineAction: """时间轴上的单个动作标记。""" @@ -90,9 +98,9 @@ class Timeline: map_code: str = "" coordinate: str = "frame" # "frame" | "time" - actions: list[TimelineAction] = field(default_factory=list) + actions: list[TimelineAction] = field(default_factory=_empty_actions) name: str = "" # 用户可读名称 - candidates: list[str] = field(default_factory=list) # 候选干员/装置名 + candidates: list[str] = field(default_factory=_empty_candidates) # 候选干员/装置名 calibration_profile: str = "" # 打轴时用的校准 profile 文件名 speed_mode: str = "auto" # "auto"（自动变速）或 "manual"（手动变速） @@ -112,14 +120,22 @@ def to_dict(self) -> dict[str, Any]: @classmethod def from_dict(cls, d: dict[str, Any]) -> Timeline: + raw_candidates = d.get("candidates", []) + candidates = cast(list[object], raw_candidates) if isinstance(raw_candidates, list) else [] + raw_actions = d.get("actions", []) + actions = cast(list[object], raw_actions) if isinstance(raw_actions, list) else [] return cls( - map_code=d.get("map_code", ""), - coordinate=d.get("coordinate", "frame"), - name=d.get("name", ""), - candidates=d.get("candidates", []), - calibration_profile=d.get("calibration_profile", ""), - speed_mode=d.get("speed_mode", "auto"), - actions=[TimelineAction.from_dict(a) for a in d.get("actions", [])], + map_code=str(d.get("map_code", "")), + coordinate=str(d.get("coordinate", "frame")), + name=str(d.get("name", "")), + candidates=[str(v) for v in candidates], + calibration_profile=str(d.get("calibration_profile", "")), + speed_mode=str(d.get("speed_mode", "auto")), + actions=[ + TimelineAction.from_dict(cast(dict[str, Any], a)) + for a in actions + if isinstance(a, dict) + ], ) def sorted(self) -> None: diff --git a/aao/types.py b/aao/types.py new file mode 100644 index 0000000..d268c85 --- /dev/null +++ b/aao/types.py @@ -0,0 +1,15 @@ +"""Shared typing aliases for dynamic JSON and runtime state payloads.""" + +from __future__ import annotations + +from typing import Any, TypedDict + +type JsonObject = dict[str, Any] + + +class MeasureState(TypedDict, total=False): + isRunning: bool + currentFrame: int | None + totalFramesInCycle: int + totalElapsedFrames: int + activeProfile: str diff --git a/aao/ui/calibration_page.py b/aao/ui/calibration_page.py index ce07a81..0a6dd54 100644 --- a/aao/ui/calibration_page.py +++ b/aao/ui/calibration_page.py @@ -172,7 +172,11 @@ def _on_start(self) -> None: self._thread = QThread() self._worker.moveToThread(self._thread) self._thread.started.connect(self._worker.run) - self._worker.progress.connect(lambda p: self.progress.setValue(int(p * 100))) + + def _on_progress(p: float) -> None: + self.progress.setValue(int(p * 100)) + + self._worker.progress.connect(_on_progress) self._worker.log.connect(self.txt_log.append) self._worker.finished_ok.connect(self._on_done) self._worker.failed.connect(self._on_failed) diff --git a/aao/ui/farm_page.py b/aao/ui/farm_page.py index 7f55914..aa32251 100644 --- a/aao/ui/farm_page.py +++ b/aao/ui/farm_page.py @@ -8,7 +8,7 @@ from __future__ import annotations -from typing import TYPE_CHECKING, Any +from typing import TYPE_CHECKING, Any, cast from PySide6.QtCore import Qt, QThread, Signal from PySide6.QtWidgets import ( @@ -294,6 +294,7 @@ def _load_advanced_settings(self) -> None: self.spin_accept_late.setValue(s.get("accept_late_frames", config.ACCEPT_LATE_FRAMES)) states = s.get("collapsible_sections", {}) if isinstance(states, dict) and "farm_advanced_params" in states: + states = cast(dict[str, object], states) self._advanced_box.set_expanded(bool(states["farm_advanced_params"])) def _on_advanced_param_changed(self) -> None: diff --git a/aao/ui/farm_worker.py b/aao/ui/farm_worker.py index c34c607..de1dc6d 100644 --- a/aao/ui/farm_worker.py +++ b/aao/ui/farm_worker.py @@ -17,6 +17,7 @@ from PySide6.QtCore import QObject, Signal +from aao.types import JsonObject from aao.utils.jsonc import load as load_jsonc from custom.action.executor import ExecuteTimeline, RoundResult from custom.reco.click_stage import get_attempt_count, reset_attempt_count @@ -114,7 +115,7 @@ def _on_node(node_name: str) -> None: tl_param = json.dumps({"timeline_path": self._timeline_path}, ensure_ascii=False) pipeline["Farm@ClickStage"]["custom_recognition_param"] = tl_param - exec_param: dict = {"timeline_path": self._timeline_path} + exec_param: JsonObject = {"timeline_path": self._timeline_path} if self._profile: exec_param["calibration"] = self._profile pipeline["Farm@Execute"]["custom_action_param"] = json.dumps(exec_param, ensure_ascii=False) diff --git a/aao/ui/floating_state.py b/aao/ui/floating_state.py index d1e4641..9687d99 100644 --- a/aao/ui/floating_state.py +++ b/aao/ui/floating_state.py @@ -12,7 +12,7 @@ from __future__ import annotations -from typing import Any +from typing import Any, cast from PySide6.QtWidgets import QWidget @@ -26,8 +26,9 @@ def clear_all() -> None: def load_state(window_id: str) -> dict[str, Any]: s = _load_settings() data = s.get("floating_windows", {}) - item = data.get(window_id, {}) if isinstance(data, dict) else {} - return item if isinstance(item, dict) else {} + windows = cast(dict[str, Any], data) if isinstance(data, dict) else {} + item = windows.get(window_id, {}) + return cast(dict[str, Any], item) if isinstance(item, dict) else {} def save_state(window_id: str, patch: dict[str, Any]) -> None: @@ -35,9 +36,11 @@ def save_state(window_id: str, patch: dict[str, Any]) -> None: windows = s.get("floating_windows", {}) if not isinstance(windows, dict): windows = {} + windows = cast(dict[str, Any], windows) item = windows.get(window_id, {}) if not isinstance(item, dict): item = {} + item = cast(dict[str, Any], item) item.update(patch) windows[window_id] = item s["floating_windows"] = windows @@ -55,7 +58,8 @@ def restore_geometry(window_id: str, widget: QWidget) -> None: return if not isinstance(g, list | tuple): return - x, y, w, h = [int(v) for v in g] + values = [_to_int(v) for v in cast(list[object] | tuple[object, ...], g)] + x, y, w, h = values widget.setGeometry(x, y, w, h) @@ -73,19 +77,29 @@ def save_follow(window_id: str, follow: dict[str, Any] | None) -> None: def load_follow(window_id: str) -> dict[str, Any] | None: value = load_state(window_id).get("follow") - return value if isinstance(value, dict) else None + return cast(dict[str, Any], value) if isinstance(value, dict) else None def _valid_geometry(value: object) -> bool: - if not isinstance(value, list | tuple) or len(value) != 4: + if not isinstance(value, list | tuple): + return False + seq = cast(list[object] | tuple[object, ...], value) + if len(seq) != 4: return False try: - _x, _y, w, h = [int(v) for v in value] + values = [_to_int(v) for v in seq] + _x, _y, w, h = values except (TypeError, ValueError): return False return w > 0 and h > 0 +def _to_int(value: object) -> int: + if isinstance(value, str | int | float): + return int(value) + raise TypeError(f"invalid integer value: {value!r}") + + def _load_settings() -> dict[str, Any]: from aao.ui.settings_page import load_settings diff --git a/aao/ui/map_picker.py b/aao/ui/map_picker.py index d79cdf1..2f485db 100644 --- a/aao/ui/map_picker.py +++ b/aao/ui/map_picker.py @@ -10,6 +10,8 @@ from __future__ import annotations +from typing import Any, cast + from PySide6.QtCore import Signal from PySide6.QtGui import QBrush, QColor, QFont, QMouseEvent, QPainter, QPen from PySide6.QtWidgets import ( @@ -26,6 +28,7 @@ from aao.core.geometry.convert_pos import tile_position_to_str from aao.core.geometry.map_loader import load_map +from aao.types import JsonObject _CELL = 40 _COLORS = { @@ -43,12 +46,12 @@ class _MapGrid(QGraphicsView): picked = Signal(str) # 棋盘记号，如 "D2" - def __init__(self, map_data: dict): + def __init__(self, map_data: JsonObject): super().__init__() self._map_data = map_data - self._height = map_data["height"] - self._width = map_data["width"] - self._tiles = map_data["tiles"] + self._height = int(map_data["height"]) + self._width = int(map_data["width"]) + self._tiles = cast(list[list[dict[str, Any]]], map_data["tiles"]) self._selected: tuple[int, int] | None = None # (col, row) self._cell_items: dict[tuple[int, int], QGraphicsRectItem] = {} diff --git a/aao/ui/settings_page.py b/aao/ui/settings_page.py index c1eac14..463d70a 100644 --- a/aao/ui/settings_page.py +++ b/aao/ui/settings_page.py @@ -9,7 +9,7 @@ import json import threading from pathlib import Path -from typing import TYPE_CHECKING, Any +from typing import TYPE_CHECKING, Any, cast from PySide6.QtCore import QObject, Qt, QThread, QTimer, Signal from PySide6.QtWidgets import ( @@ -33,6 +33,7 @@ from aao import __version__ from aao.resources.syncer import sync_all from aao.resources.updater import ReleaseInfo, UpdateChecker +from aao.types import JsonObject from aao.ui import theme from aao.ui.collapsible_box import CollapsibleBox from aao.ui.scrollbar_style import apply_themed_scrollbar @@ -40,24 +41,26 @@ from aao.utils.runtime_paths import project_root if TYPE_CHECKING: + from maa.toolkit import DesktopWindow from PySide6.QtGui import QImage, QShowEvent -def _settings_path(): +def _settings_path() -> Path: return project_root() / "config" / "settings.json" -def load_settings() -> dict: +def load_settings() -> JsonObject: p = _settings_path() if not p.exists(): return {} try: - return json.loads(p.read_text(encoding="utf-8")) + data = json.loads(p.read_text(encoding="utf-8")) + return cast(JsonObject, data) if isinstance(data, dict) else {} except (OSError, ValueError): return {} -def save_settings(data: dict) -> None: +def save_settings(data: JsonObject) -> None: p = _settings_path() p.parent.mkdir(parents=True, exist_ok=True) p.write_text(json.dumps(data, indent=2, ensure_ascii=False), encoding="utf-8") @@ -208,7 +211,11 @@ def _add_collapsible( box = CollapsibleBox(title) box.set_summary(summary) box.add_widget(widget) - box.toggled.connect(lambda expanded, k=key: self._save_collapsible_state(k, expanded)) + + def _on_toggled(expanded: bool, k: str = key) -> None: + self._save_collapsible_state(k, expanded) + + box.toggled.connect(_on_toggled) self._collapsibles[key] = box root.addWidget(box) return box @@ -387,7 +394,7 @@ def _build_ui(self) -> None: self.btn_token_eye.clicked.connect(self._toggle_token_visible) self.btn_export_log.clicked.connect(self._on_export_log) - self._windows: list = [] # DesktopWindow 列表（与 list_windows 行对应） + self._windows: list[DesktopWindow] = [] # DesktopWindow 列表（与 list_windows 行对应） self._preview_thread: QThread | None = None self._preview_worker: _PreviewWorker | None = None @@ -451,7 +458,7 @@ def _refresh_windows(self) -> None: self.list_windows.setCurrentRow(saved_row) self.lbl_win_status.setText(f"找到 {len(wins)} 个窗口，选中后可预览/设为默认") - def _selected_window(self): + def _selected_window(self) -> DesktopWindow | None: row = self.list_windows.currentRow() if row < 0 or row >= len(self._windows): return None @@ -613,6 +620,7 @@ def _load(self) -> None: self.lbl_bg_val.setText(f"{self.slider_bg.value()}%") states = s.get("collapsible_sections", {}) if isinstance(states, dict): + states = cast(dict[str, object], states) for key, box in self._collapsibles.items(): if key in states: box.set_expanded(bool(states[key])) diff --git a/aao/ui/window_snap.py b/aao/ui/window_snap.py index 268048a..912fcdb 100644 --- a/aao/ui/window_snap.py +++ b/aao/ui/window_snap.py @@ -15,6 +15,7 @@ import sys import weakref from dataclasses import dataclass +from typing import cast from PySide6.QtCore import QPoint, QRect from PySide6.QtWidgets import QApplication, QWidget @@ -115,10 +116,13 @@ def follow_from_dict(data: dict[str, object] | None) -> SnapFollow | None: return None if target_id is not None and not isinstance(target_id, str): return None - if not isinstance(offset, list | tuple) or len(offset) != 2: + if not isinstance(offset, list | tuple): + return None + pair = cast(list[object] | tuple[object, ...], offset) + if len(pair) != 2: return None try: - point = QPoint(int(offset[0]), int(offset[1])) + point = QPoint(_to_int(pair[0]), _to_int(pair[1])) except (TypeError, ValueError): return None widget_ref = None @@ -127,6 +131,12 @@ def follow_from_dict(data: dict[str, object] | None) -> SnapFollow | None: return SnapFollow(str(kind), target_id, point, widget_ref) +def _to_int(value: object) -> int: + if isinstance(value, str | int | float): + return int(value) + raise TypeError(f"invalid integer value: {value!r}") + + def follow_top_left(follow: SnapFollow) -> QPoint | None: """根据跟随关系返回新的 top-left；目标不可用时返回 None。""" rect: QRect | None = None diff --git a/aao/utils/jsonc.py b/aao/utils/jsonc.py index f327e0a..0a38c10 100644 --- a/aao/utils/jsonc.py +++ b/aao/utils/jsonc.py @@ -7,16 +7,16 @@ from __future__ import annotations from pathlib import Path -from typing import Any +from typing import Any, cast import json5 def loads(text: str) -> Any: """解析 JSONC/JSON5 字符串。""" - return json5.loads(text) + return cast(Any, json5.loads(text)) def load(path: Path | str) -> Any: """从文件解析 JSONC/JSON5。""" - return json5.loads(Path(path).read_text(encoding="utf-8")) + return cast(Any, json5.loads(Path(path).read_text(encoding="utf-8"))) diff --git a/aao/utils/logger.py b/aao/utils/logger.py index 2d0f299..8067838 100644 --- a/aao/utils/logger.py +++ b/aao/utils/logger.py @@ -13,10 +13,11 @@ from __future__ import annotations +import io import sys from collections.abc import Callable from pathlib import Path -from typing import Any +from typing import Any, cast from loguru import logger as _loguru_logger @@ -93,10 +94,11 @@ def setup_logging(level: str = "INFO", log_dir: str | Path = _DEFAULT_LOG_DIR) - """ # windowed 打包(console=False)下 sys.stdout 为 None，跳过控制台 sink # （日志仍写文件 + UI 面板） - has_console = sys.stdout is not None - if has_console: + console = sys.stdout + has_console = console is not None + if isinstance(console, io.TextIOWrapper): try: - sys.stdout.reconfigure(encoding="utf-8") # pyright: ignore[reportAttributeAccessIssue] + console.reconfigure(encoding="utf-8") except AttributeError: pass @@ -108,7 +110,7 @@ def setup_logging(level: str = "INFO", log_dir: str | Path = _DEFAULT_LOG_DIR) - # 用户显示 sink：控制台，仅消息，去来源（windowed 下无控制台则跳过） if has_console: _loguru_logger.add( - sys.stdout, + cast(Any, console), format=_CONSOLE_FORMAT, level=level, colorize=True, diff --git a/custom/action/executor.py b/custom/action/executor.py index d6919e0..880bd45 100644 --- a/custom/action/executor.py +++ b/custom/action/executor.py @@ -23,6 +23,7 @@ import time from collections.abc import Callable from dataclasses import dataclass +from typing import Any, cast import numpy as np from maa.context import Context @@ -38,6 +39,7 @@ from aao.core.geometry.view import transform_map_to_view from aao.core.timing.calibration import load as load_calibration from aao.core.timing.time_source import TimeSource +from aao.types import JsonObject from aao.utils.logger import logger from custom.registry import custom_action @@ -78,16 +80,22 @@ class ExecuteTimeline(CustomAction): def run(self, context: Context, argv: CustomAction.RunArg) -> CustomAction.RunResult: try: - params = json.loads(argv.custom_action_param) if argv.custom_action_param else {} + raw_params: Any = ( + json.loads(argv.custom_action_param) if argv.custom_action_param else {} + ) # MAA 可能双重 JSON 编码 custom_action_param - if isinstance(params, str): - params = json.loads(params) + if isinstance(raw_params, str): + raw_params = json.loads(raw_params) + if not isinstance(raw_params, dict): + logger.error("ExecuteTimeline 参数必须是 JSON object: %r", raw_params) + return CustomAction.RunResult(success=False) + params = cast(JsonObject, raw_params) return self._execute(context, params) except Exception: logger.exception("ExecuteTimeline 异常") return CustomAction.RunResult(success=False) - def _execute(self, context: Context, params: dict) -> CustomAction.RunResult: + def _execute(self, context: Context, params: JsonObject) -> CustomAction.RunResult: ctrl = context.tasker.controller # 优先 timeline_path（从文件加载，文件内含 map_code），兼容显式 timeline 数组 @@ -95,7 +103,7 @@ def _execute(self, context: Context, params: dict) -> CustomAction.RunResult: tl_calib = "" tl_speed_mode = "auto" if timeline_path: - tl = self._load_timeline_file(timeline_path) + tl = self._load_timeline_file(str(timeline_path)) if tl is None: return CustomAction.RunResult(success=False) raw_actions = tl.get("actions", []) @@ -222,7 +230,7 @@ def _execute(self, context: Context, params: dict) -> CustomAction.RunResult: # 漏怪 = 本局失败（farm pipeline 会走放弃重试） return CustomAction.RunResult(success=not self._leaked and not self._abort_reason) - def _load_timeline_file(self, path: str) -> dict | None: + def _load_timeline_file(self, path: str) -> JsonObject | None: """加载时间轴 JSON（纯文件名→config/timelines/，带路径→相对项目根）。""" from custom.reco.click_stage import resolve_timeline_path @@ -235,13 +243,18 @@ def _load_timeline_file(self, path: str) -> dict | None: except (OSError, ValueError): logger.exception("时间轴文件解析失败: %s", p) return None - n = len(data.get("actions", [])) - logger.info("加载时间轴 %s（map_code=%s, %d 动作）", p, data.get("map_code"), n) - return data + if not isinstance(data, dict): + logger.error("时间轴文件必须是 JSON object: %s", p) + return None + obj = cast(JsonObject, data) + actions = obj.get("actions", []) + n = len(cast(list[object], actions)) if isinstance(actions, list) else 0 + logger.info("加载时间轴 %s（map_code=%s, %d 动作）", p, obj.get("map_code"), n) + return obj - def _parse_actions(self, raw: list[dict], map_data: dict) -> list[Action]: + def _parse_actions(self, raw: list[JsonObject], map_data: JsonObject) -> list[Action]: """解析 JSON 动作列表 → Action 对象（含投影坐标 + 目标帧）。""" - h, w = map_data["height"], map_data["width"] + h, w = int(map_data["height"]), int(map_data["width"]) front = transform_map_to_view(map_data, side=False) side = transform_map_to_view(map_data, side=True) @@ -258,14 +271,16 @@ def _parse_actions(self, raw: list[dict], map_data: dict) -> list[Action]: tick_val = item.get("tick") target_frame = (cost_val or 0) * config.TICK_MAX_DEFAULT + (tick_val or 0) + action_type = item.get("action_type") + direction = item.get("direction") a = Action( cost=cost_val, tick=tick_val, time=item.get("time"), - action_type=ActionType(item["action_type"]) if "action_type" in item else None, + action_type=ActionType(action_type) if action_type is not None else None, oper=item.get("oper"), pos=item.get("pos"), - direction=DirectionType(item["direction"]) if "direction" in item else None, + direction=DirectionType(direction) if direction is not None else None, alias=item.get("alias"), ) if not a.is_valid(): diff --git a/custom/action/key_press.py b/custom/action/key_press.py index fab9741..600b416 100644 --- a/custom/action/key_press.py +++ b/custom/action/key_press.py @@ -13,11 +13,13 @@ import json import time +from typing import Any, cast from maa.context import Context from maa.custom_action import CustomAction from aao.core import afa_hotkey +from aao.types import JsonObject from aao.utils.logger import logger from custom.registry import custom_action @@ -30,9 +32,13 @@ def run(self, context: Context, argv: CustomAction.RunArg) -> CustomAction.RunRe raw = argv.custom_action_param logger.info("KeyPress 收到参数: %r", raw) try: - params = json.loads(raw) if raw else {} - if isinstance(params, str): # 双重 JSON 编码 - params = json.loads(params) + raw_params: Any = json.loads(raw) if raw else {} + if isinstance(raw_params, str): # 双重 JSON 编码 + raw_params = json.loads(raw_params) + if not isinstance(raw_params, dict): + logger.error("KeyPress 参数必须是 JSON object: %r", raw_params) + return CustomAction.RunResult(success=False) + params = cast(JsonObject, raw_params) key = params.get("key") interval_ms = int(params.get("interval_ms", 0)) @@ -40,7 +46,7 @@ def run(self, context: Context, argv: CustomAction.RunArg) -> CustomAction.RunRe logger.error("KeyPress 缺少 key，params=%r", params) return CustomAction.RunResult(success=False) - keys = key if isinstance(key, list) else [key] + keys: list[Any] = cast(list[Any], key) if isinstance(key, list) else [key] for i, vk in enumerate(keys): if i and interval_ms: time.sleep(interval_ms / 1000.0) diff --git a/custom/outcome.py b/custom/outcome.py index 57a6ae3..73c6c2f 100644 --- a/custom/outcome.py +++ b/custom/outcome.py @@ -84,7 +84,8 @@ def make_sink( 返回 should_debug=False：不额外开启 debug mode，避免额外开销。 """ global _on_outcome_cb - from maa.context import Context, ContextEventSink, NotificationType + from maa.context import Context, ContextEventSink + from maa.event_sink import NotificationType _on_outcome_cb = on_outcome diff --git a/custom/reco/click_stage.py b/custom/reco/click_stage.py index 2029874..02b0427 100644 --- a/custom/reco/click_stage.py +++ b/custom/reco/click_stage.py @@ -9,10 +9,12 @@ import json import time from pathlib import Path +from typing import cast from maa.context import Context from maa.custom_recognition import CustomRecognition +from aao.types import JsonObject from aao.utils.logger import logger from aao.utils.runtime_paths import project_root from custom.registry import custom_recognition @@ -65,7 +67,7 @@ def resolve_timeline_path(timeline_path: str) -> Path: return _TIMELINE_DIR / timeline_path -def _load_timeline_data(timeline_path: str | None) -> dict | None: +def _load_timeline_data(timeline_path: str | None) -> JsonObject | None: """加载 timeline JSON。""" if not timeline_path: logger.error("timeline_path 为空") @@ -75,7 +77,8 @@ def _load_timeline_data(timeline_path: str | None) -> dict | None: logger.error("时间轴文件不存在: %s", p) return None try: - return json.loads(p.read_text(encoding="utf-8")) + data = json.loads(p.read_text(encoding="utf-8")) + return cast(JsonObject, data) if isinstance(data, dict) else None except (OSError, ValueError): logger.exception("时间轴文件解析失败: %s", p) return None @@ -89,7 +92,7 @@ def read_map_code(timeline_path: str | None) -> str | None: mc = data.get("map_code") if not mc: logger.error("时间轴文件无 map_code") - return mc + return str(mc) if mc else None def read_stage_text(timeline_path: str | None) -> str | None: @@ -97,7 +100,8 @@ def read_stage_text(timeline_path: str | None) -> str | None: data = _load_timeline_data(timeline_path) if not data: return None - return data.get("stage_text") or data.get("map_code") + text = data.get("stage_text") or data.get("map_code") + return str(text) if text else None @custom_recognition("ClickStage") diff --git a/pyproject.toml b/pyproject.toml index b799e99..9160d56 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -31,6 +31,7 @@ packages = ["custom", "aao"] [tool.ruff] target-version = "py313" line-length = 100 +extend-exclude = ["typings"] [tool.ruff.lint] select = ["E", "F", "I", "UP", "B"] @@ -41,18 +42,6 @@ extraPaths = ["custom", "aao"] pythonVersion = "3.13" typeCheckingMode = "strict" -# strict 下关闭因 maafw 第三方库存根缺失导致的类型未知噪音 -# （reportMissingTypeStubs / reportUnknown* / reportMissingTypeArgument）， -# 保留 strict 的其余实质检查（privateUsage / unnecessaryComparison / unusedFunction 等）。 -reportMissingTypeStubs = "none" -reportUnknownMemberType = "none" -reportUnknownArgumentType = "none" -reportUnknownVariableType = "none" -reportUnknownParameterType = "none" -reportUnknownLambdaType = "none" -reportMissingTypeArgument = "none" - [tool.pytest.ini_options] testpaths = ["tests"] pythonpath = ["."] - diff --git a/typings/maa/__init__.pyi b/typings/maa/__init__.pyi new file mode 100644 index 0000000..ec89304 --- /dev/null +++ b/typings/maa/__init__.pyi @@ -0,0 +1,10 @@ +""" +This type stub file was generated by pyright. +""" + +import os +from pathlib import Path +from .library import Library + +env_path = ... +path = ... diff --git a/typings/maa/agent/__init__.pyi b/typings/maa/agent/__init__.pyi new file mode 100644 index 0000000..8d925e0 --- /dev/null +++ b/typings/maa/agent/__init__.pyi @@ -0,0 +1,10 @@ +""" +This type stub file was generated by pyright. +""" + +import os +from pathlib import Path +from ..library import Library + +env_path = ... +path = ... diff --git a/typings/maa/agent/agent_server.pyi b/typings/maa/agent/agent_server.pyi new file mode 100644 index 0000000..389c47c --- /dev/null +++ b/typings/maa/agent/agent_server.pyi @@ -0,0 +1,15 @@ +from __future__ import annotations + +from maa.custom_action import CustomAction +from maa.custom_recognition import CustomRecognition + + +class AgentServer: + @classmethod + def register_custom_action(cls, name: str, action: CustomAction) -> bool: ... + @classmethod + def register_custom_recognition(cls, name: str, recognition: CustomRecognition) -> bool: ... + @classmethod + def start_up(cls, identifier: str) -> bool: ... + @classmethod + def join(cls) -> None: ... diff --git a/typings/maa/agent_client.pyi b/typings/maa/agent_client.pyi new file mode 100644 index 0000000..b25b251 --- /dev/null +++ b/typings/maa/agent_client.pyi @@ -0,0 +1,175 @@ +""" +This type stub file was generated by pyright. +""" + +from typing import Optional +from .controller import Controller +from .define import * +from .resource import Resource +from .tasker import Tasker + +class AgentClient: + """Agent 客户端 / Agent client + + 用于连接到 AgentServer，将自定义识别器和动作的执行委托给独立进程。 + 这允许将 MaaFW 本体与 Custom 逻辑分离至独立进程运行。 + Used to connect to AgentServer, delegating custom recognition and action execution to a separate + process. + This allows separating MaaFW core from Custom logic into independent processes. + """ + _handle: MaaAgentClientHandle + def __init__(self, identifier: Optional[str] = ..., *, _handle: Optional[MaaAgentClientHandle] = ...) -> None: + """创建 Agent 客户端 / Create Agent client + + 默认使用 IPC 模式；如果 identifier 为纯数字字符串，则会将其视为 TCP 端口号并监听 127.0.0.1。 + 在不支持 AF_UNIX 的旧版 Windows（Build 17063 之前）上也会自动回退到 TCP 模式。 + Uses IPC mode by default; if identifier is a numeric string, it is treated as a TCP port + and listens on 127.0.0.1. On older Windows versions that don't support AF_UNIX + (before Build 17063), it will also automatically fall back to TCP mode. + + Args: + identifier: 可选的连接标识符；纯数字字符串会被视为 TCP 端口号 / + Optional connection identifier; numeric strings are treated as TCP ports + _handle: 内部使用，直接传入已创建的句柄 / Internal use, directly pass in an already created handle + + Raises: + RuntimeError: 如果创建失败 + """ + ... + + @classmethod + def create_tcp(cls, port: int = ...) -> AgentClient: + """创建使用 TCP 连接的 Agent 客户端 / Create Agent client with TCP connection + + 客户端会监听 127.0.0.1 上的指定端口。如果传入 0 则自动选择可用端口。 + AgentServer 端使用 identifier 属性获取的端口号作为 identifier 即可通过 TCP 连接。 + The client listens on 127.0.0.1 at the specified port. If 0 is passed, an available port is + automatically selected. + AgentServer can use the port number from the identifier property to connect via TCP. + + Args: + port: TCP 端口号 (0-65535)，0 表示自动选择 / TCP port number (0-65535), 0 means auto-select + + Returns: + AgentClient: TCP 模式的客户端实例 / Client instance in TCP mode + + Raises: + RuntimeError: 如果创建失败 + ValueError: 如果端口号无效 + """ + ... + + def __del__(self): # -> None: + ... + + @property + def identifier(self) -> Optional[str]: + """获取连接标识符 / Get connection identifier + + Returns: + Optional[str]: 连接标识符，如果未设置则返回 None / Connection identifier, or None if not set + """ + ... + + def bind(self, resource: Resource) -> bool: + """绑定资源 / Bind resource + + 将 AgentServer 中注册的自定义识别器和动作绑定到资源上。 + Bind custom recognitions and actions registered in AgentServer to the resource. + + Args: + resource: 资源对象 / Resource object + + Returns: + bool: 是否成功 / Whether successful + """ + ... + + def register_sink(self, resource: Resource, controller: Controller, tasker: Tasker) -> bool: + """注册事件监听器 / Register event sinks + + 将资源、控制器、任务器的事件转发给 AgentServer。 + Forward resource, controller, and tasker events to AgentServer. + + Args: + resource: 资源对象 / Resource object + controller: 控制器对象 / Controller object + tasker: 任务器对象 / Tasker object + + Returns: + bool: 是否全部注册成功 / Whether all registrations succeeded + """ + ... + + def connect(self) -> bool: + """连接到 AgentServer / Connect to AgentServer + + Returns: + bool: 是否成功 / Whether successful + """ + ... + + def disconnect(self) -> bool: + """断开与 AgentServer 的连接 / Disconnect from AgentServer + + Returns: + bool: 是否成功 / Whether successful + """ + ... + + @property + def connected(self) -> bool: + """判断是否已连接 / Check if connected + + Returns: + bool: 是否已连接 / Whether connected + """ + ... + + @property + def alive(self) -> bool: + """判断连接是否存活 / Check if connection is alive + + Returns: + bool: 连接是否存活 / Whether connection is alive + """ + ... + + def set_timeout(self, milliseconds: int) -> bool: + """设置超时时间 / Set timeout + + Args: + milliseconds: 超时时间（毫秒） / Timeout in milliseconds + + Returns: + bool: 是否成功 / Whether successful + """ + ... + + @property + def custom_recognition_list(self) -> list[str]: + """获取已注册的自定义识别器列表 / Get registered custom recognizer list + + Returns: + list[str]: 自定义识别器名列表 / List of custom recognizer names + + Raises: + RuntimeError: 如果获取失败 + """ + ... + + @property + def custom_action_list(self) -> list[str]: + """获取已注册的自定义操作列表 / Get registered custom action list + + Returns: + list[str]: 自定义操作名列表 / List of custom action names + + Raises: + RuntimeError: 如果获取失败 + """ + ... + + _api_properties_initialized: bool = ... + + diff --git a/typings/maa/buffer.pyi b/typings/maa/buffer.pyi new file mode 100644 index 0000000..d56e72c --- /dev/null +++ b/typings/maa/buffer.pyi @@ -0,0 +1,304 @@ +""" +This type stub file was generated by pyright. +""" + +import numpy +from typing import Optional, Union +from .define import * + +class StringBuffer: + """字符串缓冲区 / String buffer + + 用于在 Python 和 C API 之间传递字符串数据。 + Used to pass string data between Python and C API. + """ + _handle: MaaStringBufferHandle + _own: bool + def __init__(self, handle: Optional[MaaStringBufferHandle] = ...) -> None: + ... + + def __del__(self): # -> None: + ... + + def get(self) -> str: + """获取缓冲区内容 / Get buffer content + + Returns: + str: 字符串内容 / String content + """ + ... + + def set(self, value: Union[str, bytes]) -> bool: + """设置缓冲区内容 / Set buffer content + + Args: + value: 字符串或字节数据 / String or bytes data + + Returns: + bool: 是否成功 / Whether successful + """ + ... + + @property + def empty(self) -> bool: + """判断缓冲区是否为空 / Check if buffer is empty + + Returns: + bool: 是否为空 / Whether empty + """ + ... + + def clear(self) -> bool: + """清空缓冲区 / Clear buffer + + Returns: + bool: 是否成功 / Whether successful + """ + ... + + _api_properties_initialized: bool = ... + + +class StringListBuffer: + """字符串列表缓冲区 / String list buffer + + 用于在 Python 和 C API 之间传递字符串列表数据。 + Used to pass string list data between Python and C API. + """ + _handle: MaaStringListBufferHandle + _own: bool + def __init__(self, handle: Optional[MaaStringListBufferHandle] = ...) -> None: + ... + + def __del__(self): # -> None: + ... + + def get(self) -> list[str]: + """获取字符串列表 / Get string list + + Returns: + List[str]: 字符串列表 / String list + """ + ... + + def set(self, value: list[str]) -> bool: + """设置字符串列表 / Set string list + + Args: + value: 字符串列表 / String list + + Returns: + bool: 是否成功 / Whether successful + """ + ... + + def append(self, value: str) -> bool: + """追加字符串 / Append string + + Args: + value: 要追加的字符串 / String to append + + Returns: + bool: 是否成功 / Whether successful + """ + ... + + def remove(self, index: int) -> bool: + """移除指定索引的字符串 / Remove string at index + + Args: + index: 要移除的索引 / Index to remove + + Returns: + bool: 是否成功 / Whether successful + """ + ... + + def clear(self) -> bool: + """清空列表 / Clear list + + Returns: + bool: 是否成功 / Whether successful + """ + ... + + _api_properties_initialized: bool = ... + + +class ImageBuffer: + """图像缓冲区 / Image buffer + + 用于在 Python 和 C API 之间传递图像数据。图像格式为 BGR，与 OpenCV 兼容。 + Used to pass image data between Python and C API. Image format is BGR, compatible with OpenCV. + """ + _handle: MaaImageBufferHandle + _own: bool + def __init__(self, c_handle: Optional[MaaImageBufferHandle] = ...) -> None: + ... + + def __del__(self): # -> None: + ... + + def get(self) -> numpy.ndarray: + """获取图像数据 / Get image data + + Returns: + numpy.ndarray: BGR 格式图像，形状为 (height, width, channels) + BGR format image with shape (height, width, channels) + """ + ... + + def set(self, value: numpy.ndarray) -> bool: + """设置图像数据 / Set image data + + Args: + value: BGR 格式图像，形状为 (height, width, channels) + BGR format image with shape (height, width, channels) + + Returns: + bool: 是否成功 / Whether successful + + Raises: + TypeError: 如果 value 不是 numpy.ndarray + """ + ... + + def resize(self, width: int = ..., height: int = ...) -> bool: + """调整图像尺寸 / Resize image + + Args: + width: 目标宽度，0 表示按高度等比缩放 / Target width, 0 to scale by height + height: 目标高度，0 表示按宽度等比缩放 / Target height, 0 to scale by width + + Returns: + bool: 是否成功 / Whether successful + """ + ... + + @property + def empty(self) -> bool: + """判断缓冲区是否为空 / Check if buffer is empty + + Returns: + bool: 是否为空 / Whether empty + """ + ... + + def clear(self) -> bool: + """清空缓冲区 / Clear buffer + + Returns: + bool: 是否成功 / Whether successful + """ + ... + + _api_properties_initialized: bool = ... + + +class ImageListBuffer: + """图像列表缓冲区 / Image list buffer + + 用于在 Python 和 C API 之间传递图像列表数据。 + Used to pass image list data between Python and C API. + """ + _handle: MaaImageListBufferHandle + _own: bool + def __init__(self, c_handle: Optional[MaaImageListBufferHandle] = ...) -> None: + ... + + def __del__(self): # -> None: + ... + + def get(self) -> list[numpy.ndarray]: + """获取图像列表 / Get image list + + Returns: + List[numpy.ndarray]: 图像列表 / Image list + """ + ... + + def set(self, value: list[numpy.ndarray]) -> bool: + """设置图像列表 / Set image list + + Args: + value: 图像列表 / Image list + + Returns: + bool: 是否成功 / Whether successful + """ + ... + + def append(self, value: numpy.ndarray) -> bool: + """追加图像 / Append image + + Args: + value: 要追加的图像 / Image to append + + Returns: + bool: 是否成功 / Whether successful + """ + ... + + def remove(self, index: int) -> bool: + """移除指定索引的图像 / Remove image at index + + Args: + index: 要移除的索引 / Index to remove + + Returns: + bool: 是否成功 / Whether successful + """ + ... + + def clear(self) -> bool: + """清空列表 / Clear list + + Returns: + bool: 是否成功 / Whether successful + """ + ... + + _api_properties_initialized: bool = ... + + +class RectBuffer: + """矩形缓冲区 / Rectangle buffer + + 用于在 Python 和 C API 之间传递矩形数据（x, y, width, height）。 + Used to pass rectangle data (x, y, width, height) between Python and C API. + """ + _handle: MaaRectHandle + _own: bool + def __init__(self, c_handle: Optional[MaaRectHandle] = ...) -> None: + ... + + def __del__(self): # -> None: + ... + + def get(self) -> Rect: + """获取矩形数据 / Get rectangle data + + Returns: + Rect: 矩形对象 (x, y, width, height) / Rectangle object (x, y, width, height) + """ + ... + + def set(self, value: RectType) -> bool: + """设置矩形数据 / Set rectangle data + + Args: + value: 矩形数据，可以是 Rect、tuple、list 或 numpy.ndarray + Rectangle data, can be Rect, tuple, list, or numpy.ndarray + + Returns: + bool: 是否成功 / Whether successful + + Raises: + ValueError: 如果数据格式不正确 + TypeError: 如果类型不支持 + """ + ... + + _api_properties_initialized: bool = ... + + diff --git a/typings/maa/context.pyi b/typings/maa/context.pyi new file mode 100644 index 0000000..5fbff79 --- /dev/null +++ b/typings/maa/context.pyi @@ -0,0 +1,378 @@ +""" +This type stub file was generated by pyright. +""" + +import numpy +from dataclasses import dataclass +from typing import Any, Optional +from .define import * +from .event_sink import EventSink, NotificationType +from .job import TaskJob +from .pipeline import JActionParam, JActionType, JNodeAttr, JPipelineData, JRecognitionParam, JRecognitionType, JWaitFreezes +from .tasker import Tasker + +class Context: + _handle: MaaContextHandle + _tasker: Tasker + def __init__(self, handle: MaaContextHandle) -> None: + ... + + def __del__(self): # -> None: + ... + + def run_task(self, entry: str, pipeline_override: Optional[dict[str, Any]] = ...) -> Optional[TaskDetail]: + """同步执行任务 / Synchronously execute task + + Args: + entry: 任务入口 / Task entry + pipeline_override: 用于覆盖的 json / JSON for overriding + + Returns: + Optional[TaskDetail]: 任务详情，执行失败则返回 None / Task detail, or None if execution failed + """ + ... + + def run_recognition(self, entry: str, image: numpy.ndarray, pipeline_override: Optional[dict[str, Any]] = ...) -> Optional[RecognitionDetail]: + """同步执行识别逻辑 / Synchronously execute recognition logic + + 不会执行后续操作, 不会执行后续 next + Will not execute subsequent operations or next steps + + Args: + entry: 任务名 / Task name + image: 前序截图 / Previous screenshot + pipeline_override: 用于覆盖的 json / JSON for overriding + + Returns: + Optional[RecognitionDetail]: 识别结果。无论是否命中，只要尝试进行了识别，就会返回； + 请通过 RecognitionDetail.hit 判断是否命中。只在未能启动识别流程时（如 entry 不存在、node disabled、image + 为空等），才可能返回 + None。 + Recognition detail. It always returns as long as recognition was attempted; + use RecognitionDetail.hit to determine hit. Only return None if the recognition process + fails to start + (e.g., entry does not exist, node is disabled, image is empty). + """ + ... + + def run_action(self, entry: str, box: RectType = ..., reco_detail: str = ..., pipeline_override: Optional[dict[str, Any]] = ...) -> Optional[ActionDetail]: + """同步执行操作逻辑 / Synchronously execute action logic + + 不会执行后续 next + Will not execute subsequent next steps + + Args: + entry: 任务名 / Task name + box: 前序识别位置 / Previous recognition position + reco_detail: 前序识别详情 / Previous recognition details + pipeline_override: 用于覆盖的 json / JSON for overriding + + Returns: + Optional[ActionDetail]: 操作结果。无论动作是否成功，只要尝试执行了动作，就会返回； + 请通过 ActionDetail.success 判断是否执行成功。只在未能启动动作流程时（如 entry 不存在、node disabled + 等），才可能返回 None。 + Action detail. It always returns as long as the action was attempted; + use ActionDetail.success to determine success. Only return None if the action flow fails + to start + (e.g., entry does not exist, node is disabled, etc.). + """ + ... + + def run_recognition_direct(self, reco_type: JRecognitionType, reco_param: JRecognitionParam, image: numpy.ndarray) -> Optional[RecognitionDetail]: + """同步执行识别 / Synchronously execute recognition + + 直接使用识别类型和参数执行，无需通过 pipeline entry。 + Execute directly with recognition type and parameters, without requiring a pipeline entry. + + Args: + reco_type: 识别类型 / Recognition type + reco_param: 识别参数 / Recognition parameters + image: 前序截图 / Previous screenshot + + Returns: + Optional[RecognitionDetail]: 识别结果。无论是否命中，只要尝试进行了识别，就会返回； + 请通过 RecognitionDetail.hit 判断是否命中。只在未能启动识别流程时才可能返回 None。 + Recognition detail. It always returns as long as recognition was attempted; + use RecognitionDetail.hit to determine hit. Only return None if the recognition process + fails to start. + """ + ... + + def run_action_direct(self, action_type: JActionType, action_param: JActionParam, box: RectType = ..., reco_detail: str = ...) -> Optional[ActionDetail]: + """同步执行操作 / Synchronously execute action + + 直接使用操作类型和参数执行，无需通过 pipeline entry。 + Execute directly with action type and parameters, without requiring a pipeline entry. + + Args: + action_type: 操作类型 / Action type + action_param: 操作参数 / Action parameters + box: 前序识别位置 / Previous recognition position + reco_detail: 前序识别详情 / Previous recognition details + + Returns: + Optional[ActionDetail]: 操作结果。无论动作是否成功，只要尝试执行了动作，就会返回； + 请通过 ActionDetail.success 判断是否执行成功。只在未能启动动作流程时才可能返回 None。 + Action detail. It always returns as long as the action was attempted; + use ActionDetail.success to determine success. Only return None if the action flow fails + to start. + """ + ... + + def override_pipeline(self, pipeline_override: dict[str, Any]) -> bool: + """覆盖 pipeline / Override pipeline_override + + Args: + pipeline_override: 用于覆盖的 json / JSON for overriding + + Returns: + bool: 是否成功 / Whether successful + """ + ... + + def override_next(self, name: str, next_list: list[str]) -> bool: + """覆盖任务的 next 列表 / Override the next list of task + + 如果节点不存在，此方法会失败 + This method will fail if the node does not exist + + Args: + name: 任务名 / Task name + next_list: next 列表 / Next list + + Returns: + bool: 成功返回 True，如果节点不存在则返回 False / Returns True on success, False if node does not exist + """ + ... + + def override_image(self, image_name: str, image: numpy.ndarray) -> bool: + """覆盖图片 / Override the image corresponding to image_name + + Args: + image_name: 图片名 / Image name + image: 图片数据 / Image data + + Returns: + bool: 是否成功 / Whether successful + """ + ... + + def get_node_data(self, name: str) -> Optional[dict[str, Any]]: + """获取任务当前的定义 / Get the current definition of task + + Args: + name: 任务名 / Task name + + Returns: + Optional[Dict]: 任务定义字典，如果不存在则返回 None / Task definition dict, or None if not exists + """ + ... + + def get_node_object(self, name: str) -> Optional[JPipelineData]: + """获取任务当前的定义（解析为对象） / Get the current definition of task (parsed as object) + + Args: + name: 任务名 / Task name + + Returns: + Optional[JPipelineData]: 任务定义对象，如果不存在则返回 None + Task definition object, or None if not exists + """ + ... + + @property + def tasker(self) -> Tasker: + """获取实例 / Get instance + + Returns: + Tasker: 实例对象 / Instance object + """ + ... + + def get_task_job(self) -> TaskJob: + """获取对应任务号的任务作业 / Get task job for corresponding task id + + Returns: + TaskJob: 任务作业对象 / Task job object + + Raises: + ValueError: 如果任务 id 为 None + """ + ... + + def clone(self) -> Context: + """复制上下文 / Clone context + + Returns: + Context: 复制的上下文对象 / Cloned context object + + Raises: + ValueError: 如果克隆失败 + """ + ... + + def set_anchor(self, anchor_name: str, node_name: str) -> bool: + """设置锚点 / Set anchor + + Args: + anchor_name: 锚点名称 / Anchor name + node_name: 节点名称 / Node name + + Returns: + bool: 是否成功 / Whether successful + """ + ... + + def get_anchor(self, anchor_name: str) -> Optional[str]: + """获取锚点对应的节点名 / Get node name for anchor + + Args: + anchor_name: 锚点名称 / Anchor name + + Returns: + Optional[str]: 节点名称，如果不存在则返回 None / Node name, or None if not exists + """ + ... + + def get_hit_count(self, node_name: str) -> int: + """获取节点命中计数 / Get hit count for node + + Args: + node_name: 节点名称 / Node name + + Returns: + int: 命中计数 / Hit count + """ + ... + + def clear_hit_count(self, node_name: str) -> bool: + """清除节点命中计数 / Clear hit count for node + + Args: + node_name: 节点名称 / Node name + + Returns: + bool: 是否成功 / Whether successful + """ + ... + + def wait_freezes(self, time: int = ..., box: Optional[tuple[int, int, int, int]] = ..., wait_freezes_param: Optional[JWaitFreezes] = ...) -> bool: + """等待画面静止 / Wait for screen to stabilize (freeze) + + Args: + time: 等待时间（毫秒） / Wait time in milliseconds + box: 识别命中的区域 (x, y, w, h)，用于 target 为 Self 时计算 ROI + Recognition hit box, used when target is Self to calculate ROI + wait_freezes_param: 等待参数，使用 JWaitFreezes。支持 time, target, target_offset, threshold, + method, rate_limit, timeout + + Wait parameters, use JWaitFreezes. Supports time, target, + target_offset, threshold, method, rate_limit, timeout + + Returns: + bool: 是否成功 / Whether successful + + Note: + - time 和 wait_freezes_param.time 互斥，不能同时为非零或同时为零 + time and wait_freezes_param.time are mutually exclusive + """ + ... + + _api_properties_initialized: bool = ... + + +class ContextEventSink(EventSink): + @dataclass + class NodeWaitFreezesDetail: + task_id: int + wf_id: int + name: str + phase: str + roi: tuple[int, int, int, int] + param: dict[str, Any] + reco_ids: list[int] + elapsed: Optional[int] + focus: Any + ... + + + def on_node_wait_freezes(self, context: Context, noti_type: NotificationType, detail: NodeWaitFreezesDetail): # -> None: + ... + + @dataclass + class NodeNextListDetail: + task_id: int + name: str + next_list: list[JNodeAttr] + focus: Any + ... + + + def on_node_next_list(self, context: Context, noti_type: NotificationType, detail: NodeNextListDetail): # -> None: + ... + + @dataclass + class NodeRecognitionDetail: + task_id: int + reco_id: int + name: str + focus: Any + anchor: Optional[str] = ... + + + def on_node_recognition(self, context: Context, noti_type: NotificationType, detail: NodeRecognitionDetail): # -> None: + ... + + @dataclass + class NodeActionDetail: + task_id: int + action_id: int + name: str + focus: Any + ... + + + def on_node_action(self, context: Context, noti_type: NotificationType, detail: NodeActionDetail): # -> None: + ... + + @dataclass + class NodePipelineNodeDetail: + task_id: int + node_id: int + name: str + focus: Any + ... + + + def on_node_pipeline_node(self, context: Context, noti_type: NotificationType, detail: NodePipelineNodeDetail): # -> None: + ... + + @dataclass + class NodeRecognitionNodeDetail: + task_id: int + node_id: int + name: str + focus: Any + ... + + + def on_node_recognition_node(self, context: Context, noti_type: NotificationType, detail: NodeRecognitionNodeDetail): # -> None: + ... + + @dataclass + class NodeActionNodeDetail: + task_id: int + node_id: int + name: str + focus: Any + ... + + + def on_node_action_node(self, context: Context, noti_type: NotificationType, detail: NodeActionNodeDetail): # -> None: + ... + + def on_raw_notification(self, context: Context, msg: str, details: dict[str, Any]) -> None: + ... + + + diff --git a/typings/maa/controller.pyi b/typings/maa/controller.pyi new file mode 100644 index 0000000..4bbe34d --- /dev/null +++ b/typings/maa/controller.pyi @@ -0,0 +1,773 @@ +""" +This type stub file was generated by pyright. +""" + +import ctypes +import numpy +from abc import abstractmethod +from collections.abc import Sequence +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Optional, Union +from .define import * +from .event_sink import EventSink, NotificationType +from .job import Job, JobWithResult + +__all__ = ["AdbController", "DbgController", "ReplayController", "RecordController", "PlayCoverController", "Win32Controller", "GamepadController", "WlRootsController", "AndroidNativeController", "CustomController"] +class Controller: + _handle: MaaControllerHandle + _own: bool + def __init__(self, handle: Optional[MaaControllerHandle] = ...) -> None: + ... + + def __del__(self): # -> None: + ... + + def post_connection(self) -> Job: + """异步连接设备 / Asynchronously connect device + + 这是一个异步操作，会立即返回一个 Job 对象 + This is an asynchronous operation that immediately returns a Job object + + Returns: + Job: 作业对象，可通过 status/wait 查询状态 / Job object, can query status via status/wait + """ + ... + + def post_click(self, x: int, y: int, contact: int = ..., pressure: int = ...) -> Job: + """异步点击 / Asynchronously click + + 这是一个异步操作，会立即返回一个 Job 对象 + This is an asynchronous operation that immediately returns a Job object + + Args: + x: x 坐标 / x coordinate + y: y 坐标 / y coordinate + contact: 触点编号 (Adb 控制器: 手指编号; Win32 控制器: 鼠标按键 0:左键, 1:右键, 2:中键) + Contact number (Adb controller: finger number; Win32 controller: mouse button 0:left, + 1:right, 2:middle) + pressure: 触点力度 / Contact pressure + + Returns: + Job: 作业对象，可通过 status/wait 查询状态 / Job object, can query status via status/wait + """ + ... + + def post_swipe(self, x1: int, y1: int, x2: int, y2: int, duration: int, contact: int = ..., pressure: int = ...) -> Job: + """滑动 / Swipe + + Args: + x1: 起点 x 坐标 / Start x coordinate + y1: 起点 y 坐标 / Start y coordinate + x2: 终点 x 坐标 / End x coordinate + y2: 终点 y 坐标 / End y coordinate + duration: 滑动时长(毫秒) / Swipe duration in milliseconds + contact: 触点编号 (Adb 控制器: 手指编号; Win32 控制器: 鼠标按键 0:左键, 1:右键, 2:中键) + Contact number (Adb controller: finger number; Win32 controller: mouse button 0:left, + 1:right, 2:middle) + pressure: 触点力度 / Contact pressure + + Returns: + Job: 作业对象 / Job object + """ + ... + + def post_press_key(self, key: int) -> Job: + """ + Deprecated: Use post_click_key instead. + """ + ... + + def post_click_key(self, key: int) -> Job: + """单击按键 / Click key + + Args: + key: 虚拟键码 / Virtual key code + + Returns: + Job: 作业对象 / Job object + """ + ... + + def post_key_down(self, key: int) -> Job: + """按下键 / Key down + + Args: + key: 虚拟键码 / Virtual key code + + Returns: + Job: 作业对象 / Job object + """ + ... + + def post_key_up(self, key: int) -> Job: + """抬起键 / Key up + + Args: + key: 虚拟键码 / Virtual key code + + Returns: + Job: 作业对象 / Job object + """ + ... + + def post_input_text(self, text: str) -> Job: + """输入文本 / Input text + + Args: + text: 要输入的文本 / Text to input + + Returns: + Job: 作业对象 / Job object + """ + ... + + def post_start_app(self, intent: str) -> Job: + """启动应用 / Start app + + Args: + intent: 目标应用 (Adb 控制器: package name 或 activity) + Target app (Adb controller: package name or activity) + + Returns: + Job: 作业对象 / Job object + """ + ... + + def post_stop_app(self, intent: str) -> Job: + """关闭应用 / Stop app + + Args: + intent: 目标应用 (Adb 控制器: package name) / Target app (Adb controller: package name) + + Returns: + Job: 作业对象 / Job object + """ + ... + + def post_touch_down(self, x: int, y: int, contact: int = ..., pressure: int = ...) -> Job: + """按下 / Touch down + + Args: + x: x 坐标 / x coordinate + y: y 坐标 / y coordinate + contact: 触点编号 (Adb 控制器: 手指编号; Win32 控制器: 鼠标按键 0:左键, 1:右键, 2:中键) + Contact number (Adb controller: finger number; Win32 controller: mouse button 0:left, + 1:right, 2:middle) + pressure: 触点力度 / Contact pressure + + Returns: + Job: 作业对象 / Job object + """ + ... + + def post_touch_move(self, x: int, y: int, contact: int = ..., pressure: int = ...) -> Job: + """移动 / Move + + Args: + x: x 坐标 / x coordinate + y: y 坐标 / y coordinate + contact: 触点编号 (Adb 控制器: 手指编号; Win32 控制器: 鼠标按键 0:左键, 1:右键, 2:中键) + Contact number (Adb controller: finger number; Win32 controller: mouse button 0:left, + 1:right, 2:middle) + pressure: 触点力度 / Contact pressure + + Returns: + Job: 作业对象 / Job object + """ + ... + + def post_touch_up(self, contact: int = ...) -> Job: + """抬起 / Touch up + + Args: + contact: 触点编号 (Adb 控制器: 手指编号; Win32 控制器: 鼠标按键 0:左键, 1:右键, 2:中键) + Contact number (Adb controller: finger number; Win32 controller: mouse button 0:left, + 1:right, 2:middle) + + Returns: + Job: 作业对象 / Job object + """ + ... + + def post_relative_move(self, dx: int, dy: int) -> Job: + """异步执行一次相对位移 (当前仅 Win32 controller 支持) + Asynchronously execute a relative movement (Currently only supported by Win32 controller) + + Args: + dx: x 方向移动偏移 / x axis offset + dy: y 方向移动偏移 / y axis offset + + Returns: + Job: 作业对象 / Job object + """ + ... + + def set_mouse_lock_follow(self, enabled: bool) -> bool: + """设置鼠标锁定跟随模式 (Win32 controller) / Set mouse lock follow mode (Win32 controller) + + Args: + enabled: 是否开启 / Whether to enable + + Returns: + bool: 是否设置成功 / Whether the setting was successful + + Note: + 适用于 TPS/FPS 等在后台锁定鼠标的游戏 / For TPS/FPS games that lock the mouse in the background + """ + ... + + def set_background_managed_keys(self, keys: Sequence[int]) -> bool: + ... + + def post_scroll(self, dx: int, dy: int) -> Job: + """滚动 / Scroll + + Args: + dx: 水平滚动距离，正值向右滚动，负值向左滚动 + Horizontal scroll distance, positive for right, negative for left + dy: 垂直滚动距离，正值向上滚动，负值向下滚动 / Vertical scroll distance, positive for up, negative for down + + Returns: + Job: 作业对象 / Job object + + Note: + Win32 控制器和实现了 scroll 的自定义控制器支持滚动操作 + Win32 controllers and custom controllers that implement scroll support this operation + 建议使用 120 的整数倍（WHEEL_DELTA）以获得最佳兼容性 / Using multiples of 120 (WHEEL_DELTA) is recommended + """ + ... + + def post_screencap(self) -> JobWithResult[numpy.ndarray]: + """截图 / Screenshot + + Returns: + JobWithResult: 作业对象，可通过 result 获取截图 / Job object, can get screenshot via result + """ + ... + + @property + def cached_image(self) -> numpy.ndarray: + """获取最新一次截图 / Get the latest screenshot + + Returns: + numpy.ndarray: 截图图像 / Screenshot image + + Raises: + RuntimeError: 如果获取失败 + + Note: + 返回的图像是经过缩放的，尺寸根据截图目标尺寸设置（长边/短边）而定，可能与设备原始分辨率不同。 + 使用 resolution 属性可获取设备的原始（未缩放）分辨率。 + + The returned image is scaled according to the screenshot target size settings (long side + short side). + The image dimensions may differ from the raw device resolution. + Use the resolution property to get the raw (unscaled) device resolution. + """ + ... + + def post_shell(self, cmd: str, timeout: int = ...) -> JobWithResult[str]: + """执行 shell 命令 (仅 ADB 控制器) / Execute shell command (ADB only) + + Args: + cmd: shell 命令 / shell command + timeout: 超时时间（毫秒），默认 20000，设置为 -1 表示无限等待 + Timeout in milliseconds, default 20000, set to -1 for infinite wait + + Returns: + JobWithResult: 作业对象，可通过 result 获取命令输出 / Job object, can get output via result + """ + ... + + def post_inactive(self) -> Job: + """设置控制器为不活跃状态 / Set controller to inactive state + + 对于 Win32 控制器，这会恢复窗口位置（取消置顶）并解除输入阻断。 + 对于其他控制器，这是一个空操作，总是成功。 + + For Win32 controllers, this restores window position (removes topmost) and unblocks user + input. + For other controllers, this is a no-op that always succeeds. + + Returns: + Job: 作业对象，可通过 status/wait 查询状态 / Job object, can query status via status/wait + """ + ... + + @property + def shell_output(self) -> str: + """获取最近一次 shell 命令输出 / Get the latest shell command output + + Returns: + str: shell 命令输出 / shell command output + + Raises: + RuntimeError: 如果获取失败 + """ + ... + + @property + def connected(self) -> bool: + """判断是否已连接 / Check if connected + + Returns: + bool: 是否已连接 / Whether connected + """ + ... + + @property + def uuid(self) -> str: + """获取设备 uuid / Get device uuid + + Returns: + str: 设备 uuid / Device uuid + + Raises: + RuntimeError: 如果获取失败 + """ + ... + + @property + def info(self) -> dict[str, Any]: + """获取控制器信息 / Get controller information + + Returns: + Dict[str, Any]: 控制器信息，包含类型、构造参数等 + Controller information including type, constructor parameters, etc. + + Raises: + RuntimeError: 如果获取失败 + """ + ... + + @property + def resolution(self) -> tuple[int, int]: + """获取设备原始（未缩放）分辨率 / Get the raw (unscaled) device resolution + + Returns: + Tuple[int, int]: (宽度, 高度)，获取失败时返回 (0, 0) / (width, height), returns (0, 0) on failure + + Note: + 返回的是设备屏幕的实际分辨率，未经任何缩放处理。 + 而通过 cached_image 获取的截图是经过缩放的，其尺寸可能与此原始分辨率不同。 + 需要在首次截图后才能获取到有效值，否则返回 (0, 0)。 + + This returns the actual device screen resolution before any scaling. + The screenshot obtained via cached_image is scaled according to the screenshot target + size settings, + so its dimensions may differ from this raw resolution. + Valid values are only available after the first screenshot is taken, otherwise returns + (0, 0). + """ + ... + + def set_screenshot_target_long_side(self, long_side: int) -> bool: + """设置截图缩放长边到指定长度 / Set screenshot scaling long side to specified length + + Args: + long_side: 长边长度 / Long side length + + Returns: + bool: 是否成功 / Whether successful + """ + ... + + def set_screenshot_target_short_side(self, short_side: int) -> bool: + """设置截图缩放短边到指定长度 / Set screenshot scaling short side to specified length + + Args: + short_side: 短边长度 / Short side length + + Returns: + bool: 是否成功 / Whether successful + """ + ... + + def set_screenshot_use_raw_size(self, enable: bool) -> bool: + """设置截图不缩放 / Set screenshot use raw size without scaling + + 注意：此选项可能导致在不同分辨率的设备上坐标不正确 + Note: This option may cause incorrect coordinates on devices with different resolutions + + Args: + enable: 是否启用 / Whether to enable + + Returns: + bool: 是否成功 / Whether successful + """ + ... + + def set_screenshot_resize_method(self, method: int) -> bool: + """设置截图缩放插值方法 / Set screenshot resize interpolation method + + 值对应 cv::InterpolationFlags / Value corresponds to cv::InterpolationFlags: + INTER_NEAREST=0, INTER_LINEAR=1, INTER_CUBIC=2, INTER_AREA=3, INTER_LANCZOS4=4 + + Args: + method: 插值方法 / Interpolation method (default: 3, INTER_AREA) + + Returns: + bool: 是否成功 / Whether successful + """ + ... + + _sink_holder: dict[int, ControllerEventSink] = ... + def add_sink(self, sink: ControllerEventSink) -> Optional[int]: + """添加控制器事件监听器 / Add controller event listener + + Args: + sink: 事件监听器 / Event sink + + Returns: + Optional[int]: 监听器 id，失败返回 None / Listener id, or None if failed + """ + ... + + def remove_sink(self, sink_id: int) -> None: + """移除控制器事件监听器 / Remove controller event listener + + Args: + sink_id: 监听器 id / Listener id + """ + ... + + def clear_sinks(self) -> None: + """清除所有控制器事件监听器 / Clear all controller event listeners""" + ... + + _api_properties_initialized: bool = ... + + +class AdbController(Controller): + """Adb 控制器 / Adb controller + + 截图方式和输入方式会在启动时进行测速, 选择最快的方案 + Screenshot and input methods will be speed tested at startup, selecting the fastest option + """ + AGENT_BINARY_PATH = ... + def __init__(self, adb_path: Union[str, Path], address: str, screencap_methods: int = ..., input_methods: int = ..., config: Optional[dict[str, Any]] = ..., agent_path: Union[str, Path] = ...) -> None: + """创建 Adb 控制器 / Create Adb controller + + Args: + adb_path: adb 路径 / adb path + address: 连接地址 / connection address + screencap_methods: 所有可使用的截图方式 / all available screenshot methods + input_methods: 所有可使用的输入方式 / all available input methods + config: 额外配置 / extra config + agent_path: MaaAgentBinary 路径 / MaaAgentBinary path + + Raises: + RuntimeError: 如果创建失败 + """ + ... + + + +class Win32Controller(Controller): + """Win32 控制器 / Win32 controller""" + def __init__(self, hWnd: Union[ctypes.c_void_p, int, None], screencap_method: int = ..., mouse_method: int = ..., keyboard_method: int = ...) -> None: + """创建 Win32 控制器 / Create Win32 controller + + Args: + hWnd: 窗口句柄 / window handle + screencap_method: 使用的截图方式 / screenshot method used + mouse_method: 使用的鼠标输入方式 / mouse input method used + keyboard_method: 使用的键盘输入方式 / keyboard input method used + + Raises: + RuntimeError: 如果创建失败 + """ + ... + + + +class MacOSController(Controller): + """MacOS 控制器 / MacOS controller""" + def __init__(self, window_id: int, screencap_method: int = ..., input_method: int = ...) -> None: + """创建 MacOS 控制器 / Create MacOS controller + + Args: + window_id: 窗口 ID / window ID + screencap_method: 使用的截图方式 / screenshot method used + input_method: 使用的输入方式 / input method used + + Raises: + RuntimeError: 如果创建失败 + """ + ... + + + +class AndroidNativeController(Controller): + """Android Native 控制器 / Android native controller""" + def __init__(self, config: dict[str, Any]) -> None: + """创建 Android Native 控制器 / Create Android native controller + + Args: + config: 控制器配置 JSON 对象 / controller config JSON object + + Raises: + RuntimeError: 如果创建失败 + """ + ... + + + +class PlayCoverController(Controller): + """PlayCover 控制器 / PlayCover controller + + 用于在 macOS 上控制通过 PlayCover 运行的 iOS 应用 + For controlling iOS apps running via PlayCover on macOS + """ + def __init__(self, address: str, uuid: str) -> None: + """创建 PlayCover 控制器 / Create PlayCover controller + + Args: + address: PlayTools 服务地址 (host:port) / PlayTools service endpoint (host:port) + uuid: 目标应用 bundle identifier / Target app bundle identifier + + Raises: + RuntimeError: 如果创建失败 + """ + ... + + + +class WlRootsController(Controller): + """WlRoots 控制器 / WlRoots controller + + 用于在 Linux 上控制在 wlroots 合成器中运行的应用 + For controlling apps running in wlroots compositor on Linux + """ + def __init__(self, wlr_socket_path: str, use_win32_vk_code: bool = ...) -> None: + """创建 WlRoots 控制器 / Create WlRoots controller + + Args: + wlr_socket_path: Wayland Socket 路径 / Wayland Socket Path + use_win32_vk_code: 为 True 时按键被视为 Win32 VK 键码并转换为 Linux evdev 码； + 默认 False，按原始 evdev 码处理 + / When True, key codes are interpreted as Win32 Virtual-Key codes and translated + to Linux evdev codes internally; default False passes raw evdev codes through + + Raises: + RuntimeError: 如果创建失败 + """ + ... + + + +class DbgController(Controller): + """调试控制器，轮播图片截图，基本输入操作直接返回成功 + + Debug controller that cycles through images from a directory""" + def __init__(self, read_path: Union[str, Path]) -> None: + """创建调试控制器 / Create debug controller + + Args: + read_path: 图片目录（或单个图片文件）路径。连接时加载所有图片，截图时轮播。 + + Path to a directory of images (or a single image file). Images are loaded on + connect and cycled on screencap. + + Raises: + RuntimeError: 如果创建失败 + """ + ... + + + +class ReplayController(Controller): + """回放控制器，用于回放录制文件 / Replay controller for replaying recorded operations""" + def __init__(self, recording_path: Union[str, Path]) -> None: + """创建回放控制器 / Create replay controller + + Args: + recording_path: 录制 JSONL 文件路径，由 RecordController 写入。截图路径基于该文件所在目录解析。 + + Path to the recording JSONL file written by RecordController. Screenshot + paths are resolved relative to this file's parent directory. + + Raises: + RuntimeError: 如果创建失败 + """ + ... + + + +class RecordController(Controller): + """录制控制器，包装现有控制器并记录所有操作 + Record controller that wraps an existing controller and records all operations""" + def __init__(self, inner: Controller, recording_path: Union[str, Path]) -> None: + """创建录制控制器 / Create record controller + + Args: + inner: 被包装的内部控制器 / The inner controller to wrap + recording_path: 录制 JSONL 文件输出路径。截图会保存到同目录下的 "{stem}-Screenshot" 文件夹。 + + Path to the recording JSONL file to write. Screenshots are saved to a + "{stem}-Screenshot" folder in the same directory. + + Raises: + RuntimeError: 如果创建失败 + """ + ... + + + +class GamepadController(Controller): + """虚拟手柄控制器 (仅 Windows) / Virtual gamepad controller (Windows only) + + 通过 ViGEm 模拟 Xbox 360 或 DualShock 4 手柄，用于控制需要手柄输入的游戏。 + Emulates Xbox 360 or DualShock 4 gamepad via ViGEm for controlling games that require gamepad + input. + + 需要安装 ViGEm Bus Driver: https://github.com/ViGEm/ViGEmBus/releases + Requires ViGEm Bus Driver: https://github.com/ViGEm/ViGEmBus/releases + + 手柄操作映射: + - click_key/key_down/key_up: 数字按键 (使用 MaaGamepadButtonEnum) + - touch_down/touch_move/touch_up: 摇杆和扳机 (contact 使用 MaaGamepadContactEnum) + - contact 0: 左摇杆 (x, y: -32768~32767) + - contact 1: 右摇杆 (x, y: -32768~32767) + - contact 2: 左扳机 (pressure: 0~255) + - contact 3: 右扳机 (pressure: 0~255) + """ + def __init__(self, hWnd: Union[ctypes.c_void_p, int, None], gamepad_type: int = ..., screencap_method: int = ...) -> None: + """创建虚拟手柄控制器 / Create virtual gamepad controller + + Args: + hWnd: 窗口句柄，用于截图 (可为 None，不需要截图时) + Window handle for screencap (can be None if screencap not needed) + gamepad_type: 手柄类型 (MaaGamepadTypeEnum.Xbox360 或 MaaGamepadTypeEnum.DualShock4) + Gamepad type + screencap_method: 截图方式 (当 hWnd 不为 None 时使用) + Screencap method (used when hWnd is not None) + + Raises: + RuntimeError: 如果创建失败 + """ + ... + + + +class CustomController(Controller): + _callbacks: MaaCustomControllerCallbacks + def __init__(self) -> None: + ... + + @property + def c_handle(self) -> Any: + ... + + @property + def c_arg(self) -> ctypes.c_void_p: + ... + + @abstractmethod + def connect(self) -> bool: + ... + + def connected(self) -> bool: + """检查是否已连接（可选实现，默认返回 True）""" + ... + + @abstractmethod + def request_uuid(self) -> str: + ... + + def get_features(self) -> int: + ... + + @abstractmethod + def start_app(self, intent: str) -> bool: + ... + + @abstractmethod + def stop_app(self, intent: str) -> bool: + ... + + @abstractmethod + def screencap(self) -> numpy.ndarray: + ... + + @abstractmethod + def click(self, x: int, y: int) -> bool: + ... + + @abstractmethod + def swipe(self, x1: int, y1: int, x2: int, y2: int, duration: int) -> bool: + ... + + @abstractmethod + def touch_down(self, contact: int, x: int, y: int, pressure: int) -> bool: + ... + + @abstractmethod + def touch_move(self, contact: int, x: int, y: int, pressure: int) -> bool: + ... + + @abstractmethod + def touch_up(self, contact: int) -> bool: + ... + + @abstractmethod + def click_key(self, keycode: int) -> bool: + ... + + @abstractmethod + def input_text(self, text: str) -> bool: + ... + + @abstractmethod + def key_down(self, keycode: int) -> bool: + ... + + @abstractmethod + def key_up(self, keycode: int) -> bool: + ... + + def scroll(self, dx: int, dy: int) -> bool: + ... + + def relative_move(self, dx: int, dy: int) -> bool: + ... + + def shell(self, cmd: str, timeout: int) -> Optional[str]: + ... + + def inactive(self) -> bool: + """设置控制器为不活跃状态（可选实现，默认返回 True）""" + ... + + def get_custom_info(self) -> dict[str, Any]: + """获取自定义控制器的额外信息（可选实现，默认返回空字典） + + Get custom controller's extra info (optional, returns empty dict by default) + + Returns: + Dict[str, Any]: 额外信息，将与基础信息合并 / Extra info, will be merged with base info + """ + ... + + + +class ControllerEventSink(EventSink): + @dataclass + class ControllerActionDetail: + ctrl_id: int + uuid: str + action: str + param: dict[str, Any] + info: dict[str, Any] + ... + + + def on_controller_action(self, controller: Controller, noti_type: NotificationType, detail: ControllerActionDetail): # -> None: + ... + + def on_raw_notification(self, controller: Controller, msg: str, details: dict[str, Any]) -> None: + ... + + + diff --git a/typings/maa/custom_action.pyi b/typings/maa/custom_action.pyi new file mode 100644 index 0000000..5390aed --- /dev/null +++ b/typings/maa/custom_action.pyi @@ -0,0 +1,85 @@ +""" +This type stub file was generated by pyright. +""" + +import ctypes +from abc import ABC, abstractmethod +from dataclasses import dataclass +from typing import Any, Union +from .context import Context +from .define import * + +class CustomAction(ABC): + """自定义动作基类 / Custom action base class + + 用于实现自定义的 Pipeline 动作。继承此类并实现 run 方法， + 然后通过 Resource.register_custom_action 或 AgentServer.register_custom_action 注册。 + Used to implement custom Pipeline actions. Inherit this class and implement the run method, + then register via Resource.register_custom_action or AgentServer.register_custom_action. + + Example: + class MyAction(CustomAction): + def run(self, context, argv): + context.tasker.controller.post_click(100, 100).wait() + return True + """ + _handle: Any + def __init__(self) -> None: + ... + + @dataclass + class RunArg: + """run 方法的参数 / Arguments for run method + + Attributes: + task_detail: 当前任务详情 / Current task detail + node_name: 当前节点名 / Current node name + custom_action_name: 自定义动作名 / Custom action name + custom_action_param: 自定义动作参数 (JSON 字符串) / Custom action parameter (JSON string) + reco_detail: 前序识别详情 / Previous recognition detail + box: 前序识别位置 / Previous recognition box + """ + task_detail: TaskDetail + node_name: str + custom_action_name: str + custom_action_param: str + reco_detail: RecognitionDetail + box: Rect + ... + + + @dataclass + class RunResult: + """run 方法的返回结果 / Return result of run method + + Attributes: + success: 动作是否执行成功 / Whether the action executed successfully + """ + success: bool + ... + + + @abstractmethod + def run(self, context: Context, argv: RunArg) -> Union[RunResult, bool]: + """执行自定义动作 / Execute custom action + + Args: + context: 任务上下文，可用于执行其他操作 / Task context, can be used to execute other operations + argv: 动作参数 / Action arguments + + Returns: + Union[RunResult, bool]: 执行结果，可返回 RunResult 或 bool + Execution result, can return RunResult or bool + """ + ... + + @property + def c_handle(self) -> Any: + ... + + @property + def c_arg(self) -> ctypes.c_void_p: + ... + + + diff --git a/typings/maa/custom_recognition.pyi b/typings/maa/custom_recognition.pyi new file mode 100644 index 0000000..657b592 --- /dev/null +++ b/typings/maa/custom_recognition.pyi @@ -0,0 +1,93 @@ +""" +This type stub file was generated by pyright. +""" + +import ctypes +import numpy +from abc import ABC, abstractmethod +from dataclasses import dataclass +from typing import Any, Optional, Union +from .context import Context +from .define import * + +class CustomRecognition(ABC): + """自定义识别器基类 / Custom recognition base class + + 用于实现自定义的 Pipeline 识别算法。继承此类并实现 analyze 方法， + 然后通过 Resource.register_custom_recognition 或 AgentServer.register_custom_recognition 注册。 + Used to implement custom Pipeline recognition algorithms. Inherit this class and implement the + analyze method, + then register via Resource.register_custom_recognition or + AgentServer.register_custom_recognition. + + Example: + class MyRecognition(CustomRecognition): + def analyze(self, context, argv): + # 返回识别到的位置，或 None 表示未识别到 + return (100, 100, 50, 50) + """ + _handle: Any + def __init__(self) -> None: + ... + + @dataclass + class AnalyzeArg: + """analyze 方法的参数 / Arguments for analyze method + + Attributes: + task_detail: 当前任务详情 / Current task detail + node_name: 当前节点名 / Current node name + custom_recognition_name: 自定义识别器名 / Custom recognition name + custom_recognition_param: 自定义识别器参数 (JSON 字符串) + Custom recognition parameter (JSON string) + image: 待识别的图像 (BGR 格式) / Image to recognize (BGR format) + roi: 识别区域 / Recognition region of interest + """ + task_detail: TaskDetail + node_name: str + custom_recognition_name: str + custom_recognition_param: str + image: numpy.ndarray + roi: Rect + ... + + + @dataclass + class AnalyzeResult: + """analyze 方法的返回结果 / Return result of analyze method + + Attributes: + box: 识别到的位置，None 表示未识别到 / Recognized position, None means not recognized + detail: 识别详情，会被记录到识别结果中 / Recognition details, will be recorded in recognition result + """ + box: Optional[RectType] + detail: dict[str, Any] + ... + + + @abstractmethod + def analyze(self, context: Context, argv: AnalyzeArg) -> Union[AnalyzeResult, Optional[RectType]]: + """执行自定义识别 / Execute custom recognition + + Args: + context: 任务上下文，可用于执行其他操作 / Task context, can be used to execute other operations + argv: 识别参数 / Recognition arguments + + Returns: + Union[AnalyzeResult, Optional[RectType]]: 识别结果。可返回 AnalyzeResult、RectType 或 None。 + 返回 None 表示未识别到。 + Recognition result. Can return AnalyzeResult, RectType, or None. + Return None means not recognized. + """ + ... + + @property + def c_handle(self) -> Any: + ... + + @property + def c_arg(self) -> ctypes.c_void_p: + ... + + + diff --git a/typings/maa/define.pyi b/typings/maa/define.pyi new file mode 100644 index 0000000..37de621 --- /dev/null +++ b/typings/maa/define.pyi @@ -0,0 +1,715 @@ +""" +This type stub file was generated by pyright. +""" + +import ctypes +import numpy +from collections.abc import Iterator +from dataclasses import dataclass +from enum import IntEnum +from typing import Any, Callable, Optional, Union +from strenum import StrEnum + +__all__ = ["MaaBool", "MaaSize", "MaaNullSize", "MaaId", "MaaCtrlId", "MaaResId", "MaaTaskId", "MaaRecoId", "MaaActId", "MaaNodeId", "MaaWfId", "MaaSinkId", "MaaInvalidId", "MaaStatus", "MaaLoggingLevel", "MaaOptionValueSize", "MaaOptionValue", "MaaOption", "MaaGlobalOption", "MaaCtrlOption", "MaaResOption", "MaaStringBufferHandle", "MaaImageBufferHandle", "MaaRectHandle", "MaaStringListBufferHandle", "MaaImageListBufferHandle", "MaaResourceHandle", "MaaControllerHandle", "MaaTaskerHandle", "MaaContextHandle", "MaaAgentClientHandle", "MaaToolkitAdbDeviceListHandle", "MaaToolkitAdbDeviceHandle", "MaaToolkitDesktopWindowListHandle", "MaaToolkitDesktopWindowHandle", "MaaMacOSPermission", "MaaAdbScreencapMethod", "MaaAdbInputMethod", "MaaWin32ScreencapMethod", "MaaWin32InputMethod", "MaaMacOSScreencapMethod", "MaaMacOSInputMethod", "MaaGamepadType", "MaaControllerFeature", "FUNCTYPE", "MaaEventCallback", "MaaCustomRecognitionCallback", "MaaCustomActionCallback", "MaaCustomControllerCallbacks", "MaaStatusEnum", "MaaGlobalOptionEnum", "MaaCtrlOptionEnum", "MaaInferenceDeviceEnum", "MaaInferenceExecutionProviderEnum", "MaaResOptionEnum", "MaaAdbScreencapMethodEnum", "MaaAdbInputMethodEnum", "MaaWin32ScreencapMethodEnum", "MaaWin32InputMethodEnum", "MaaMacOSScreencapMethodEnum", "MaaMacOSInputMethodEnum", "MaaGamepadTypeEnum", "MaaGamepadButtonEnum", "MaaGamepadContactEnum", "MaaControllerFeatureEnum", "MaaMacOSPermissionEnum", "AlgorithmEnum", "ActionEnum", "LoggingLevelEnum", "Status", "Point", "Rect", "PointType", "RectType", "BoxAndScoreResult", "TemplateMatchResult", "BoxAndCountResult", "FeatureMatchResult", "ColorMatchResult", "OCRResult", "NeuralNetworkResult", "NeuralNetworkClassifyResult", "NeuralNetworkDetectResult", "CustomRecognitionResult", "AndRecognitionResult", "OrRecognitionResult", "RecognitionResult", "RecognitionDetail", "ClickActionResult", "LongPressActionResult", "SwipeActionResult", "MultiSwipeActionResult", "ClickKeyActionResult", "LongPressKeyActionResult", "InputTextActionResult", "AppActionResult", "ScrollActionResult", "TouchActionResult", "ShellActionResult", "ActionResult", "ActionDetail", "WaitFreezesDetail", "NodeDetail", "TaskDetail", "AlgorithmResultDict", "ActionResultDict"] +MaaBool = ctypes.c_uint8 +MaaSize = ctypes.c_size_t +MaaNullSize = MaaSize(-1) +MaaId = ctypes.c_int64 +MaaCtrlId = MaaId +MaaResId = MaaId +MaaTaskId = MaaId +MaaRecoId = MaaId +MaaActId = MaaId +MaaNodeId = MaaId +MaaWfId = MaaId +MaaSinkId = MaaId +MaaInvalidId = MaaId(0) +MaaStringBufferHandle = ctypes.c_void_p +MaaImageBufferHandle = ctypes.c_void_p +MaaRectHandle = ctypes.c_void_p +MaaStringListBufferHandle = ctypes.c_void_p +MaaImageListBufferHandle = ctypes.c_void_p +MaaResourceHandle = ctypes.c_void_p +MaaControllerHandle = ctypes.c_void_p +MaaTaskerHandle = ctypes.c_void_p +MaaContextHandle = ctypes.c_void_p +MaaStatus = ctypes.c_int32 +class MaaStatusEnum(IntEnum): + invalid = ... + pending = ... + running = ... + succeeded = ... + failed = ... + + +MaaLoggingLevel = ctypes.c_int32 +MaaOptionValueSize = ctypes.c_uint64 +MaaOptionValue = ctypes.c_void_p +MaaOption = ctypes.c_int32 +MaaGlobalOption = MaaOption +MaaCtrlOption = MaaOption +MaaResOption = MaaOption +class MaaGlobalOptionEnum(IntEnum): + Invalid = ... + LogDir = ... + SaveDraw = ... + StdoutLevel = ... + ShowHitDraw = ... + DebugMode = ... + SaveOnError = ... + DrawQuality = ... + RecoImageCacheLimit = ... + + +class MaaCtrlOptionEnum(IntEnum): + Invalid = ... + ScreenshotTargetLongSide = ... + ScreenshotTargetShortSide = ... + ScreenshotUseRawSize = ... + MouseLockFollow = ... + ScreenshotResizeMethod = ... + BackgroundManagedKeys = ... + + +class MaaInferenceDeviceEnum(IntEnum): + CPU = ... + Auto = ... + + +class MaaInferenceExecutionProviderEnum(IntEnum): + Auto = ... + CPU = ... + DirectML = ... + CoreML = ... + CUDA = ... + + +class MaaResOptionEnum(IntEnum): + Invalid = ... + InferenceDevice = ... + InferenceExecutionProvider = ... + + +MaaAdbScreencapMethod = ctypes.c_uint64 +class MaaAdbScreencapMethodEnum(IntEnum): + """ + Adb screencap method flags. + + Use bitwise OR to set the methods you need. + MaaFramework will test all provided methods and use the fastest available one. + + Default: All methods except RawByNetcat, MinicapDirect, MinicapStream + + Note: MinicapDirect and MinicapStream use lossy JPEG encoding, which may + significantly reduce template matching accuracy. Not recommended. + + | Method | Speed | Compatibility | Encoding | Notes | + |-----------------------|------------|---------------|----------|-----------------------------------| + | EncodeToFileAndPull | Slow | High | Lossless | | + | Encode | Slow | High | Lossless | | + | RawWithGzip | Medium | High | Lossless | | + | RawByNetcat | Fast | Low | Lossless | | + | MinicapDirect | Fast | Low | Lossy | | + | MinicapStream | Very Fast | Low | Lossy | | + | EmulatorExtras | Very Fast | Low | Lossless | Emulators only: MuMu 12, LDPlayer 9, Androws | + """ + Null = ... + EncodeToFileAndPull = ... + Encode = ... + RawWithGzip = ... + RawByNetcat = ... + MinicapDirect = ... + MinicapStream = ... + EmulatorExtras = ... + All = ... + Default = ... + + +MaaAdbInputMethod = ctypes.c_uint64 +class MaaAdbInputMethodEnum(IntEnum): + """ + Adb input method flags. + + Use bitwise OR to set the methods you need. + MaaFramework will select the first available method according to priority. + + Priority (high to low): EmulatorExtras > Maatouch > MinitouchAndAdbKey > AdbShell + + Default: All methods except EmulatorExtras + + | Method | Speed | Compatibility | Notes | + |----------------------|-------|---------------|---------------------------------------| + | AdbShell | Slow | High | | + | MinitouchAndAdbKey | Fast | Medium | Key press still uses AdbShell | + | Maatouch | Fast | Medium | | + | EmulatorExtras | Fast | Low | Emulators only: MuMu 12 | + """ + Null = ... + AdbShell = ... + MinitouchAndAdbKey = ... + Maatouch = ... + EmulatorExtras = ... + All = ... + Default = ... + + +MaaWin32ScreencapMethod = ctypes.c_uint64 +class MaaWin32ScreencapMethodEnum(IntEnum): + """ + Win32 screencap method flags. + + Use bitwise OR to set the methods you need. + MaaFramework will test all provided methods and use the fastest available one. + + No default value. Client should choose one as default. + + Predefined combinations: + - Foreground: DXGI_DesktopDup_Window | ScreenDC + - Background: FramePool | PrintWindow + + Different applications use different rendering methods, there is no universal solution. + + | Method | Speed | Compat | Admin | Bg | Notes | + |------------------------|-----------|--------|-------|----|---------------------------------| + | GDI | Fast | Medium | No | No | | + | FramePool | Very Fast | Medium | No | Yes| Requires Windows 10 1903+ | + | DXGI_DesktopDup | Very Fast | Low | No | No | Desktop duplication (fullscreen)| + | DXGI_DesktopDup_Window | Very Fast | Low | No | No | Desktop duplication then crop | + | PrintWindow | Medium | Medium | No | Yes| | + | ScreenDC | Fast | High | No | No | | + + Note: FramePool and PrintWindow support pseudo-minimize. Other methods still fail + when the target window is minimized. + """ + Null = ... + GDI = ... + FramePool = ... + DXGI_DesktopDup = ... + DXGI_DesktopDup_Window = ... + PrintWindow = ... + ScreenDC = ... + All = ... + Foreground = ... + Background = ... + + +MaaWin32InputMethod = ctypes.c_uint64 +class MaaWin32InputMethodEnum(IntEnum): + """ + Win32 input method. + + No bitwise OR, select ONE method only. + + No default value. Client should choose one as default. + + Different applications process input differently, there is no universal solution. + + | Method | Compat | Admin | Seize Mouse | Bg | Notes | + |--------------------------|--------|-------|-------------|----|---------------------------------------| + | Seize | High | No | Yes | No | | + | SendMessage | Medium | Maybe | No | Yes| | + | PostMessage | Medium | Maybe | No | Yes| | + | LegacyEvent | Low | No | Yes | No | | + | PostThreadMessage | Low | Maybe | No | Yes| | + | SendMessageWithCursorPos | Medium | Maybe | Briefly | Yes| Moves cursor to target, then restores | + | PostMessageWithCursorPos | Medium | Maybe | Briefly | Yes| Moves cursor to target, then restores | + | SendMessageWithWindowPos | Medium | Maybe | No | Yes| Moves window to align w/ cursor, rest.| + | PostMessageWithWindowPos | Medium | Maybe | No | Yes| Moves window to align w/ cursor, rest.| + + Note: + - Admin rights mainly depend on the target application's privilege level. + If the target runs as admin, MaaFramework should also run as admin for compatibility. + - "WithCursorPos" methods briefly move the cursor to target position, send message, + then restore cursor position. This "briefly" seizes the mouse but won't block user operations. + - "WithWindowPos" methods briefly move the window so the target aligns with the current cursor + position, send message, then restore the window position. The cursor is not moved. + """ + Null = ... + Seize = ... + SendMessage = ... + PostMessage = ... + LegacyEvent = ... + PostThreadMessage = ... + SendMessageWithCursorPos = ... + PostMessageWithCursorPos = ... + SendMessageWithWindowPos = ... + PostMessageWithWindowPos = ... + + +MaaMacOSScreencapMethod = ctypes.c_uint64 +class MaaMacOSScreencapMethodEnum(IntEnum): + """ + MacOS screencap method. + + No bitwise OR, select ONE method only. + + No default value. Client should choose one as default. + + | Method | Speed | Compatibility | Background Support | Notes | + |-----------------|-----------|---------------|--------------------|------------------------| + | ScreenCaptureKit| Very Fast | High | Yes | Requires macOS 12.3+ | + """ + Null = ... + ScreenCaptureKit = ... + + +MaaMacOSInputMethod = ctypes.c_uint64 +class MaaMacOSInputMethodEnum(IntEnum): + """ + MacOS input method. + + No bitwise OR, select ONE method only. + + No default value. Client should choose one as default. + + | Method | Compatibility | Background Support | Notes | + |-------------|---------------|--------------------|----------------------------| + | GlobalEvent | High | No | Global event injection | + | PostToPid | Medium | Yes | Post event to specific PID | + """ + Null = ... + GlobalEvent = ... + PostToPid = ... + + +MaaGamepadType = ctypes.c_uint64 +class MaaGamepadTypeEnum(IntEnum): + """ + Virtual gamepad type for GamepadController (Windows only). + + No bitwise OR, select ONE type only. + + Requires ViGEm Bus Driver to be installed. + + | Type | Description | + |-------------|---------------------------------------| + | Xbox360 | Microsoft Xbox 360 Controller (wired) | + | DualShock4 | Sony DualShock 4 Controller (wired) | + """ + Xbox360 = ... + DualShock4 = ... + + +class MaaGamepadButtonEnum(IntEnum): + """ + Gamepad button flags (XUSB protocol values). + + Use bitwise OR to combine multiple buttons. + DS4 face buttons are aliases to Xbox face buttons. + """ + DPAD_UP = ... + DPAD_DOWN = ... + DPAD_LEFT = ... + DPAD_RIGHT = ... + START = ... + BACK = ... + LEFT_THUMB = ... + RIGHT_THUMB = ... + LB = ... + RB = ... + GUIDE = ... + A = ... + B = ... + X = ... + Y = ... + CROSS = ... + CIRCLE = ... + SQUARE = ... + TRIANGLE = ... + L1 = ... + R1 = ... + L3 = ... + R3 = ... + OPTIONS = ... + SHARE = ... + PS = ... + TOUCHPAD = ... + + +class MaaGamepadContactEnum(IntEnum): + """ + Gamepad contact (analog stick or trigger) mapping for touch_down/touch_move/touch_up. + """ + LEFT_STICK = ... + RIGHT_STICK = ... + LEFT_TRIGGER = ... + RIGHT_TRIGGER = ... + + +MaaControllerFeature = ctypes.c_uint64 +class MaaControllerFeatureEnum(IntEnum): + Null = ... + UseMouseDownAndUpInsteadOfClick = ... + UseKeyboardDownAndUpInsteadOfClick = ... + + +FUNCTYPE = ... +MaaEventCallback = ... +MaaCustomRecognitionCallback = ... +MaaCustomActionCallback = ... +MaaToolkitAdbDeviceListHandle = ctypes.c_void_p +MaaToolkitAdbDeviceHandle = ctypes.c_void_p +MaaToolkitDesktopWindowListHandle = ctypes.c_void_p +MaaToolkitDesktopWindowHandle = ctypes.c_void_p +MaaMacOSPermission = ctypes.c_int32 +class MaaMacOSPermissionEnum(IntEnum): + ScreenCapture = ... + Accessibility = ... + + +MaaAgentClientHandle = ctypes.c_void_p +class MaaCustomControllerCallbacks(ctypes.Structure): + ConnectFunc = ... + ConnectedFunc = ... + RequestUuidFunc = ... + GetFeaturesFunc = ... + StartAppFunc = ... + StopAppFunc = ... + ScreencapFunc = ... + ClickFunc = ... + SwipeFunc = ... + TouchDownFunc = ... + TouchMoveFunc = ... + TouchUpFunc = ... + ClickKeyFunc = ... + InputTextFunc = ... + KeyDownFunc = ... + KeyUpFunc = ... + ScrollFunc = ... + RelativeMoveFunc = ... + ShellFunc = ... + InactiveFunc = ... + GetInfoFunc = ... + _fields_ = ... + + +class Status: + _status: MaaStatusEnum + def __init__(self, status: Union[MaaStatus, MaaStatusEnum, int]) -> None: + ... + + @property + def done(self) -> bool: + ... + + @property + def succeeded(self) -> bool: + ... + + @property + def failed(self) -> bool: + ... + + @property + def pending(self) -> bool: + ... + + @property + def running(self) -> bool: + ... + + + +@dataclass +class Point: + x: int = ... + y: int = ... + def __add__(self, other: Union[Point, tuple[int, int], list[int],]) -> Point: + ... + + def __iter__(self) -> Iterator[int]: + ... + + def __getitem__(self, key: int) -> int: + ... + + + +@dataclass +class Rect: + x: int = ... + y: int = ... + w: int = ... + h: int = ... + def __add__(self, other: Union[Rect, tuple[int, int, int, int], list[int],]) -> Rect: + ... + + def __iter__(self) -> Iterator[int]: + ... + + def __getitem__(self, key: int) -> int: + ... + + + +PointType = Union[Point, list[int], numpy.ndarray, tuple[int, int],] +RectType = Union[Rect, list[int], numpy.ndarray, tuple[int, int, int, int],] +class AlgorithmEnum(StrEnum): + DirectHit = ... + TemplateMatch = ... + FeatureMatch = ... + ColorMatch = ... + OCR = ... + NeuralNetworkClassify = ... + NeuralNetworkDetect = ... + And = ... + Or = ... + Custom = ... + + +class ActionEnum(StrEnum): + DoNothing = ... + Click = ... + LongPress = ... + Swipe = ... + MultiSwipe = ... + ClickKey = ... + LongPressKey = ... + InputText = ... + StartApp = ... + StopApp = ... + Scroll = ... + TouchDown = ... + TouchMove = ... + TouchUp = ... + KeyDown = ... + KeyUp = ... + StopTask = ... + Command = ... + Shell = ... + Custom = ... + + +@dataclass +class BoxAndScoreResult: + box: Rect + score: float + ... + + +TemplateMatchResult = BoxAndScoreResult +@dataclass +class BoxAndCountResult: + box: Rect + count: int + ... + + +FeatureMatchResult = BoxAndCountResult +ColorMatchResult = BoxAndCountResult +@dataclass +class OCRResult(BoxAndScoreResult): + text: str + ... + + +@dataclass +class NeuralNetworkResult(BoxAndScoreResult): + cls_index: int + label: str + box: Rect + score: float + ... + + +NeuralNetworkClassifyResult = NeuralNetworkResult +NeuralNetworkDetectResult = NeuralNetworkResult +@dataclass +class CustomRecognitionResult: + box: Rect + detail: Union[str, dict[str, Any]] + ... + + +@dataclass +class AndRecognitionResult: + """And 算法识别结果，包含所有子识别的完整详情""" + sub_results: list[RecognitionDetail] + ... + + +@dataclass +class OrRecognitionResult: + """Or 算法识别结果，包含已执行子识别的完整详情""" + sub_results: list[RecognitionDetail] + ... + + +RecognitionResult = Union[TemplateMatchResult, FeatureMatchResult, ColorMatchResult, OCRResult, NeuralNetworkClassifyResult, NeuralNetworkDetectResult, AndRecognitionResult, OrRecognitionResult, CustomRecognitionResult,] +AlgorithmResultDict = ... +@dataclass +class RecognitionDetail: + reco_id: int + name: str + algorithm: Union[AlgorithmEnum, str] + hit: bool + box: Optional[Rect] + all_results: list[RecognitionResult] + filtered_results: list[RecognitionResult] + best_result: Optional[RecognitionResult] + raw_detail: dict[str, Any] + raw_image: numpy.ndarray + draw_images: list[numpy.ndarray] + ... + + +@dataclass +class ClickActionResult: + point: Point + contact: int + pressure: int + ... + + +@dataclass +class LongPressActionResult: + point: Point + duration: int + contact: int + pressure: int + ... + + +@dataclass +class SwipeActionResult: + begin: Point + end: list[Point] + end_hold: list[int] + duration: list[int] + only_hover: bool + starting: int + contact: int + pressure: int + ... + + +@dataclass +class MultiSwipeActionResult: + swipes: list[SwipeActionResult] + ... + + +@dataclass +class ClickKeyActionResult: + keycode: list[int] + ... + + +@dataclass +class LongPressKeyActionResult: + keycode: list[int] + duration: int + ... + + +@dataclass +class InputTextActionResult: + text: str + ... + + +@dataclass +class AppActionResult: + package: str + ... + + +@dataclass +class ScrollActionResult: + point: Point + dx: int + dy: int + ... + + +@dataclass +class TouchActionResult: + contact: int + point: Point + pressure: int + ... + + +@dataclass +class ShellActionResult: + cmd: str + shell_timeout: int + success: bool + output: str + ... + + +ActionResult = Union[ClickActionResult, LongPressActionResult, SwipeActionResult, MultiSwipeActionResult, ClickKeyActionResult, LongPressKeyActionResult, InputTextActionResult, AppActionResult, ScrollActionResult, TouchActionResult, ShellActionResult, None,] +ActionResultDict = ... +@dataclass +class ActionDetail: + action_id: int + name: str + action: Union[ActionEnum, str] + box: Rect + success: bool + result: Optional[ActionResult] + raw_detail: dict[str, Any] + ... + + +@dataclass +class WaitFreezesDetail: + wf_id: int + name: str + phase: str + success: bool + elapsed_ms: int + reco_id_list: list[int] + roi: Rect + ... + + +@dataclass +class NodeDetail: + node_id: int + name: str + recognition: Optional[RecognitionDetail] + action: Optional[ActionDetail] + completed: bool + ... + + +class TaskDetail: + """任务详情 / Task detail + + nodes 属性为惰性加载，仅在首次访问时才通过 IPC 获取各节点详情并缓存结果。 + The nodes property is lazily loaded: node details are fetched via IPC + only on the first access and cached thereafter. + + Attributes: + task_id: 任务 ID / Task ID + entry: 入口节点名 / Entry node name + node_id_list: 节点 ID 列表（轻量，无 IPC 开销）/ Node ID list (lightweight, no IPC cost) + status: 任务状态 / Task status + nodes: 节点详情列表（惰性加载）/ Node detail list (lazily loaded) + """ + __slots__ = ... + def __init__(self, task_id: int, entry: str, node_id_list: list[int], status: Status, node_detail_func: Optional[Callable[[int], Optional[NodeDetail]]] = ...) -> None: + ... + + @property + def nodes(self) -> list[NodeDetail]: + ... + + def __repr__(self) -> str: + ... + + + +class LoggingLevelEnum(IntEnum): + Off = ... + Fatal = ... + Error = ... + Warn = ... + Info = ... + Debug = ... + Trace = ... + All = ... + + diff --git a/typings/maa/event_sink.pyi b/typings/maa/event_sink.pyi new file mode 100644 index 0000000..a739bf2 --- /dev/null +++ b/typings/maa/event_sink.pyi @@ -0,0 +1,58 @@ +""" +This type stub file was generated by pyright. +""" + +import ctypes +from enum import IntEnum +from typing import Any + +class NotificationType(IntEnum): + """通知类型枚举 / Notification type enumeration + + 用于标识事件回调的状态类型。 + Used to identify the status type of event callbacks. + + Attributes: + Unknown: 未知类型 / Unknown type + Starting: 开始 / Starting + Succeeded: 成功 / Succeeded + Failed: 失败 / Failed + """ + Unknown = ... + Starting = ... + Succeeded = ... + Failed = ... + + +class EventSink: + """事件监听器基类 / Event sink base class + + 用于接收 MaaFramework 各种事件回调的基类。 + 派生类包括 ResourceEventSink、ControllerEventSink、TaskerEventSink、ContextEventSink。 + Base class for receiving various event callbacks from MaaFramework. + Derived classes include ResourceEventSink, ControllerEventSink, TaskerEventSink, + ContextEventSink. + """ + def on_unknown_notification(self, instance: Any, msg: str, details: dict[str, Any]) -> None: + """处理未知类型的通知 / Handle unknown notification + + 当收到无法识别的通知时调用。 + Called when an unrecognized notification is received. + + Args: + instance: 相关实例对象 / Related instance object + msg: 消息类型 / Message type + details: 消息详情 / Message details + """ + ... + + @property + def c_callback(self) -> Any: + ... + + @property + def c_callback_arg(self) -> ctypes.c_void_p: + ... + + + diff --git a/typings/maa/job.pyi b/typings/maa/job.pyi new file mode 100644 index 0000000..8d4be79 --- /dev/null +++ b/typings/maa/job.pyi @@ -0,0 +1,162 @@ +""" +This type stub file was generated by pyright. +""" + +from typing import Any, Callable, Generic, Optional, TypeVar +from .define import * + +TResult = TypeVar("TResult") +class Job: + """异步作业句柄 / Asynchronous job handle + + 用于跟踪和管理异步操作的状态，如资源加载、控制器连接等。 + Used to track and manage the status of asynchronous operations + such as resource loading, controller connection, etc. + """ + _job_id: MaaId + def __init__(self, job_id: MaaId, status_func: Callable[[int], MaaStatus], wait_func: Callable[[int], MaaStatus]) -> None: + ... + + @property + def job_id(self) -> int: + """获取作业 ID / Get job ID + + Returns: + int: 作业 ID / Job ID + """ + ... + + def wait(self) -> Job: + """等待作业完成 / Wait for job completion + + 阻塞当前线程直到作业完成 + Blocks the current thread until the job is done + + Returns: + Job: 返回自身，支持链式调用 / Returns self for method chaining + """ + ... + + @property + def status(self) -> Status: + """获取作业状态 / Get job status + + Returns: + Status: 作业状态 / Job status + """ + ... + + @property + def done(self) -> bool: + """判断作业是否已完成 / Check if job is done + + Returns: + bool: 是否已完成（成功或失败） / Whether done (succeeded or failed) + """ + ... + + @property + def succeeded(self) -> bool: + """判断作业是否成功 / Check if job succeeded + + Returns: + bool: 是否成功 / Whether succeeded + """ + ... + + @property + def failed(self) -> bool: + """判断作业是否失败 / Check if job failed + + Returns: + bool: 是否失败 / Whether failed + """ + ... + + @property + def pending(self) -> bool: + """判断作业是否等待中 / Check if job is pending + + Returns: + bool: 是否等待中 / Whether pending + """ + ... + + @property + def running(self) -> bool: + """判断作业是否运行中 / Check if job is running + + Returns: + bool: 是否运行中 / Whether running + """ + ... + + + +class JobWithResult(Job, Generic[TResult]): + """带结果的异步作业句柄 / Asynchronous job handle with result + + 继承自 Job，额外提供获取作业结果的功能。 + Inherits from Job, additionally provides the ability to get job result. + """ + def __init__(self, job_id: MaaId, status_func: Callable[[int], MaaStatus], wait_func: Callable[[int], MaaStatus], get_func: Callable[[int], TResult]) -> None: + ... + + def wait(self) -> JobWithResult[TResult]: + """等待作业完成 / Wait for job completion + + Returns: + JobWithResult: 返回自身，支持链式调用 / Returns self for method chaining + """ + ... + + def get(self, wait: bool = ...) -> TResult: + """获取作业结果 / Get job result + + Args: + wait: 是否在获取结果前等待作业完成，默认为 False。建议先显式调用 wait()（或传入 wait=True）， + 确保异步操作已完成后再获取结果 / Whether to wait for job completion before getting result, + default is False. It's recommended to call wait() first (or pass wait=True) to + ensure the + async operation is finished before getting the result. + + Returns: + 作业执行结果，类型取决于具体作业 / Job execution result, type depends on the specific job + """ + ... + + + +class TaskJob(JobWithResult[Optional["TaskDetail"]]): + """任务作业句柄 / Task job handle + + 继承自 JobWithResult，额外提供任务相关的操作。 + Inherits from JobWithResult, additionally provides task-related operations. + """ + def __init__(self, job_id: MaaId, status_func: Callable[[int], MaaStatus], wait_func: Callable[[int], MaaStatus], get_func: Callable[[int], Optional[TaskDetail]], override_pipeline_func: Callable[[int, bytes], bool]) -> None: + ... + + def wait(self) -> TaskJob: + """等待作业完成 / Wait for job completion + + Returns: + TaskJob: 返回自身，支持链式调用 / Returns self for method chaining + """ + ... + + def override_pipeline(self, pipeline_override: dict[str, Any]) -> bool: + """覆盖此任务的 pipeline / Override pipeline for this task + + 在任务执行期间动态修改 pipeline 配置 + Dynamically modify pipeline configuration during task execution + + Args: + pipeline_override: 用于覆盖的 json / JSON for overriding + + Returns: + bool: 是否成功 / Whether successful + """ + ... + + + diff --git a/typings/maa/library.pyi b/typings/maa/library.pyi new file mode 100644 index 0000000..d5453c4 --- /dev/null +++ b/typings/maa/library.pyi @@ -0,0 +1,105 @@ +""" +This type stub file was generated by pyright. +""" + +import ctypes +import pathlib +from typing import Optional +from .define import * + +class Library: + """库加载管理器 / Library loading manager + + 管理 MaaFramework 各动态库的加载和访问。 + Manages loading and access to MaaFramework dynamic libraries. + """ + _is_agent_server: bool = ... + _framework: Optional[ctypes.CDLL] = ... + _toolkit: Optional[ctypes.CDLL] = ... + _agent_client: Optional[ctypes.CDLL] = ... + _agent_server: Optional[ctypes.CDLL] = ... + _lib_type: Optional[type[ctypes.CDLL]] = ... + framework_libpath: Optional[pathlib.Path] = ... + toolkit_libpath: Optional[pathlib.Path] = ... + agent_client_libpath: Optional[pathlib.Path] = ... + agent_server_libpath: Optional[pathlib.Path] = ... + @classmethod + def open(cls, path: pathlib.Path, agent_server: bool = ...) -> None: + """打开并加载库 / Open and load libraries + + Args: + path: 库文件所在目录 / Directory containing library files + agent_server: 是否以 AgentServer 模式加载 / Whether to load in AgentServer mode + + Raises: + FileNotFoundError: 如果路径不存在 + """ + ... + + @classmethod + def framework(cls) -> ctypes.CDLL: + """获取 MaaFramework 库 / Get MaaFramework library + + Returns: + (ctypes.CDLL | ctypes.WinDLL): MaaFramework 动态库对象 / MaaFramework dynamic library object + """ + ... + + @classmethod + def toolkit(cls) -> ctypes.CDLL: + """获取 MaaToolkit 库 / Get MaaToolkit library + + Returns: + (ctypes.CDLL | ctypes.WinDLL): MaaToolkit 动态库对象 / MaaToolkit dynamic library object + + Raises: + ValueError: 如果在 AgentServer 模式下调用 + """ + ... + + @classmethod + def agent_client(cls) -> ctypes.CDLL: + """获取 MaaAgentClient 库 / Get MaaAgentClient library + + Returns: + (ctypes.CDLL | ctypes.WinDLL): MaaFramework 动态库对象 / MaaFramework dynamic library object + + Raises: + ValueError: 如果在 AgentServer 模式下调用 + """ + ... + + @classmethod + def agent_server(cls) -> ctypes.CDLL: + """获取 MaaAgentServer 库 / Get MaaAgentServer library + + Returns: + (ctypes.CDLL | ctypes.WinDLL): MaaAgentServer 动态库对象 + MaaAgentServer dynamic library object + + Raises: + ValueError: 如果不在 AgentServer 模式下调用 + """ + ... + + @classmethod + def is_agent_server(cls) -> bool: + """判断是否为 AgentServer 模式 / Check if in AgentServer mode + + Returns: + bool: 是否为 AgentServer 模式 / Whether in AgentServer mode + """ + ... + + @classmethod + def version(cls) -> str: + """获取 MaaFramework 版本 / Get MaaFramework version + + Returns: + str: 版本字符串 / Version string + """ + ... + + _api_properties_initialized: bool = ... + + diff --git a/typings/maa/pipeline.pyi b/typings/maa/pipeline.pyi new file mode 100644 index 0000000..d65807a --- /dev/null +++ b/typings/maa/pipeline.pyi @@ -0,0 +1,344 @@ +""" +This type stub file was generated by pyright. +""" + +from dataclasses import dataclass +from typing import Any, Optional, Union +from strenum import StrEnum + +JRect = tuple[int, int, int, int] +JTarget = Union[bool, str, JRect] +class JRecognitionType(StrEnum): + DirectHit = ... + TemplateMatch = ... + FeatureMatch = ... + ColorMatch = ... + OCR = ... + NeuralNetworkClassify = ... + NeuralNetworkDetect = ... + And = ... + Or = ... + Custom = ... + + +class JActionType(StrEnum): + DoNothing = ... + Click = ... + LongPress = ... + Swipe = ... + MultiSwipe = ... + TouchDown = ... + TouchMove = ... + TouchUp = ... + ClickKey = ... + LongPressKey = ... + KeyDown = ... + KeyUp = ... + InputText = ... + StartApp = ... + StopApp = ... + StopTask = ... + Scroll = ... + Command = ... + Shell = ... + Screencap = ... + Custom = ... + + +@dataclass +class JDirectHit: + roi: JTarget = ... + roi_offset: JRect = ... + + +@dataclass +class JTemplateMatch: + template: list[str] + roi: JTarget = ... + roi_offset: JRect = ... + threshold: list[float] = ... + order_by: str = ... + index: int = ... + method: int = ... + green_mask: bool = ... + + +@dataclass +class JFeatureMatch: + template: list[str] + roi: JTarget = ... + roi_offset: JRect = ... + detector: str = ... + order_by: str = ... + count: int = ... + index: int = ... + green_mask: bool = ... + ratio: float = ... + + +@dataclass +class JColorMatch: + lower: list[list[int]] + upper: list[list[int]] + roi: JTarget = ... + roi_offset: JRect = ... + order_by: str = ... + method: int = ... + count: int = ... + index: int = ... + connected: bool = ... + + +@dataclass +class JOCR: + expected: list[str] = ... + roi: JTarget = ... + roi_offset: JRect = ... + threshold: float = ... + replace: list[list[str]] = ... + order_by: str = ... + index: int = ... + only_rec: bool = ... + model: str = ... + color_filter: str = ... + + +@dataclass +class JNeuralNetworkClassify: + model: str + expected: list[int] = ... + roi: JTarget = ... + roi_offset: JRect = ... + labels: list[str] = ... + order_by: str = ... + index: int = ... + + +@dataclass +class JNeuralNetworkDetect: + model: str + expected: list[int] = ... + roi: JTarget = ... + roi_offset: JRect = ... + labels: list[str] = ... + threshold: list[float] = ... + order_by: str = ... + index: int = ... + + +@dataclass +class JCustomRecognition: + custom_recognition: str + roi: JTarget = ... + roi_offset: JRect = ... + custom_recognition_param: Any = ... + + +@dataclass +class JAnd: + all_of: list[Any] = ... + box_index: int = ... + + +@dataclass +class JOr: + any_of: list[Any] = ... + + +JRecognitionParam = Union[JDirectHit, JTemplateMatch, JFeatureMatch, JColorMatch, JOCR, JNeuralNetworkClassify, JNeuralNetworkDetect, JAnd, JOr, JCustomRecognition,] +@dataclass +class JDoNothing: + ... + + +@dataclass +class JClick: + target: JTarget = ... + target_offset: JRect = ... + contact: int = ... + pressure: int = ... + + +@dataclass +class JLongPress: + target: JTarget = ... + target_offset: JRect = ... + duration: int = ... + contact: int = ... + pressure: int = ... + + +@dataclass +class JSwipe: + starting: int = ... + begin: JTarget = ... + begin_offset: JRect = ... + end: list[JTarget] = ... + end_offset: list[JRect] = ... + end_hold: list[int] = ... + duration: list[int] = ... + only_hover: bool = ... + contact: int = ... + pressure: int = ... + + +@dataclass +class JMultiSwipe: + swipes: list[JSwipe] + ... + + +@dataclass +class JTouch: + contact: int = ... + target: JTarget = ... + target_offset: JRect = ... + pressure: int = ... + + +@dataclass +class JTouchUp: + contact: int = ... + + +@dataclass +class JClickKey: + key: list[int] + ... + + +@dataclass +class JLongPressKey: + key: list[int] + duration: int = ... + + +@dataclass +class JKey: + key: int + ... + + +@dataclass +class JInputText: + input_text: str + ... + + +@dataclass +class JStartApp: + package: str + ... + + +@dataclass +class JStopApp: + package: str + ... + + +@dataclass +class JStopTask: + ... + + +@dataclass +class JScroll: + target: JTarget = ... + target_offset: JRect = ... + dx: int = ... + dy: int = ... + + +@dataclass +class JCommand: + exec: str + args: list[str] = ... + detach: bool = ... + + +@dataclass +class JShell: + cmd: str + shell_timeout: int = ... + + +@dataclass +class JScreencap: + filename: str = ... + format: str = ... + quality: int = ... + + +@dataclass +class JCustomAction: + custom_action: str + target: JTarget = ... + custom_action_param: Any = ... + target_offset: JRect = ... + + +JActionParam = Union[JDoNothing, JClick, JLongPress, JSwipe, JMultiSwipe, JTouch, JTouchUp, JClickKey, JLongPressKey, JKey, JInputText, JStartApp, JStopApp, JStopTask, JScroll, JCommand, JShell, JScreencap, JCustomAction,] +@dataclass +class JRecognition: + type: JRecognitionType + param: JRecognitionParam + ... + + +@dataclass +class JAction: + type: JActionType + param: JActionParam + ... + + +@dataclass +class JNodeAttr: + name: str + jump_back: bool = ... + anchor: bool = ... + + +@dataclass +class JWaitFreezes: + time: int = ... + target: JTarget = ... + target_offset: JRect = ... + threshold: float = ... + method: int = ... + rate_limit: int = ... + timeout: int = ... + + +@dataclass +class JPipelineData: + recognition: JRecognition + action: JAction + next: list[JNodeAttr] = ... + rate_limit: int = ... + timeout: int = ... + on_error: list[JNodeAttr] = ... + anchor: dict[str, str] = ... + inverse: bool = ... + enabled: bool = ... + pre_delay: int = ... + post_delay: int = ... + pre_wait_freezes: Optional[JWaitFreezes] = ... + post_wait_freezes: Optional[JWaitFreezes] = ... + repeat: int = ... + repeat_delay: int = ... + repeat_wait_freezes: Optional[JWaitFreezes] = ... + max_hit: int = ... + focus: Any = ... + attach: dict[str, Any] = ... + + +class JPipelineParser: + @classmethod + def parse_pipeline_data(cls, pipeline_data: Union[str, dict[str, Any]]) -> JPipelineData: + """Parse JSON string to JPipelineData dataclass with proper variant types.""" + ... + + + diff --git a/typings/maa/resource.pyi b/typings/maa/resource.pyi new file mode 100644 index 0000000..62f7412 --- /dev/null +++ b/typings/maa/resource.pyi @@ -0,0 +1,439 @@ +""" +This type stub file was generated by pyright. +""" + +import pathlib +import numpy +from dataclasses import dataclass +from typing import Any, Callable, Optional, TYPE_CHECKING, Union +from .define import * +from .event_sink import EventSink, NotificationType +from .job import Job +from .pipeline import JActionParam, JActionType, JPipelineData, JRecognitionParam, JRecognitionType +from .custom_action import CustomAction +from .custom_recognition import CustomRecognition + +if TYPE_CHECKING: + ... +class Resource: + _handle: MaaResourceHandle + _own: bool + def __init__(self, handle: Optional[MaaResourceHandle] = ...) -> None: + """创建资源 / Create resource + + Args: + handle: 可选的外部句柄 / Optional external handle + + Raises: + RuntimeError: 如果创建失败 + """ + ... + + def __del__(self): # -> None: + ... + + def post_bundle(self, path: Union[pathlib.Path, str]) -> Job: + """异步加载资源 / Asynchronously load resources from path + + 这是一个异步操作，会立即返回一个 Job 对象 + This is an asynchronous operation that immediately returns a Job object + + Args: + path: 资源路径 / Resource path + + Returns: + Job: 作业对象，可通过 status/wait 查询状态 / Job object, can query status via status/wait + """ + ... + + def post_ocr_model(self, path: Union[pathlib.Path, str]) -> Job: + """异步加载 OCR 模型 / Asynchronously load OCR model from path + + 这是一个异步操作，会立即返回一个 Job 对象 + This is an asynchronous operation that immediately returns a Job object + + Args: + path: OCR 模型目录路径 / OCR model directory path + + Returns: + Job: 作业对象，可通过 status/wait 查询状态 / Job object, can query status via status/wait + """ + ... + + def post_pipeline(self, path: Union[pathlib.Path, str]) -> Job: + """异步加载 Pipeline / Asynchronously load pipeline from path + + 这是一个异步操作，会立即返回一个 Job 对象 + This is an asynchronous operation that immediately returns a Job object + + 支持加载目录或单个 json/jsonc 文件 + Supports loading a directory or a single json/jsonc file + + Args: + path: Pipeline 目录或文件路径 / Pipeline directory or file path + + Returns: + Job: 作业对象，可通过 status/wait 查询状态 / Job object, can query status via status/wait + """ + ... + + def post_image(self, path: Union[pathlib.Path, str]) -> Job: + """异步加载图片资源 / Asynchronously load image resources from path + + 这是一个异步操作，会立即返回一个 Job 对象 + This is an asynchronous operation that immediately returns a Job object + + 支持加载目录或单个图片文件 + Supports loading a directory or a single image file + + Args: + path: 图片目录或文件路径 / Image directory or file path + + Returns: + Job: 作业对象，可通过 status/wait 查询状态 / Job object, can query status via status/wait + """ + ... + + def override_pipeline(self, pipeline_override: dict[str, Any]) -> bool: + """覆盖 pipeline / Override pipeline_override + + Args: + pipeline_override: 用于覆盖的 json / JSON for overriding + + Returns: + bool: 是否成功 / Whether successful + """ + ... + + def override_next(self, name: str, next_list: list[str]) -> bool: + """覆盖任务的 next 列表 / Override the next list of task + + 注意：此方法会直接设置 next 列表，即使节点不存在也会创建 + Note: This method directly sets the next list, creating the node if it doesn't exist + + Args: + name: 任务名 / Task name + next_list: next 列表 / Next list + + Returns: + bool: 总是返回 True / Always returns True + """ + ... + + def override_image(self, image_name: str, image: numpy.ndarray) -> bool: + """覆盖图片 / Override the image corresponding to image_name + + Args: + image_name: 图片名 / Image name + image: 图片数据 / Image data + + Returns: + bool: 总是返回 True / Always returns True + """ + ... + + def get_node_data(self, name: str) -> Optional[dict[str, Any]]: + """获取任务当前的定义 / Get the current definition of task + + Args: + name: 任务名 / Task name + + Returns: + Optional[Dict]: 任务定义字典，如果不存在则返回 None / Task definition dict, or None if not exists + """ + ... + + def get_node_object(self, name: str) -> Optional[JPipelineData]: + """获取任务当前的定义（解析为对象） / Get the current definition of task (parsed as object) + + Args: + name: 任务名 / Task name + + Returns: + Optional[JPipelineData]: 任务定义对象，如果不存在则返回 None + Task definition object, or None if not exists + """ + ... + + def get_default_recognition_param(self, reco_type: JRecognitionType) -> Optional[JRecognitionParam]: + """获取指定识别类型的默认参数 / Get default parameters for specified recognition type + + Args: + reco_type: 识别类型 / Recognition type + + Returns: + Optional[JRecognitionParam]: 默认参数对象，如果不存在则返回 None + Default parameter object, or None if not exists + """ + ... + + def get_default_action_param(self, action_type: JActionType) -> Optional[JActionParam]: + """获取指定动作类型的默认参数 / Get default parameters for specified action type + + Args: + action_type: 动作类型 / Action type + + Returns: + Optional[JActionParam]: 默认参数对象，如果不存在则返回 None + Default parameter object, or None if not exists + """ + ... + + @property + def loaded(self) -> bool: + """判断是否加载正常 / Check if resources loaded normally + + Returns: + bool: 是否已加载 / Whether loaded + """ + ... + + def clear(self) -> bool: + """清除已加载内容 / Clear loaded content + + 如果资源正在加载中，此方法会失败 + This method will fail if resources are currently loading + + Returns: + bool: 成功返回 True，如果正在加载中则返回 False / Returns True on success, False if currently loading + """ + ... + + def use_cpu(self) -> bool: + """使用 CPU 进行推理 / Use CPU for inference + + Returns: + bool: 是否成功 / Whether successful + """ + ... + + def use_directml(self, device_id: int = ...) -> bool: + """使用 DirectML 进行推理 / Use DirectML for inference + + Args: + device_id: 设备 id，默认为自动选择 / Device id, default is Auto + + Returns: + bool: 是否成功 / Whether successful + """ + ... + + def use_coreml(self, coreml_flag: int = ...) -> bool: + """使用 CoreML 进行推理 / Use CoreML for inference + + Args: + coreml_flag: CoreML 标志，默认为自动选择 / CoreML flag, default is Auto + + Returns: + bool: 是否成功 / Whether successful + """ + ... + + def use_auto_ep(self) -> bool: + """自动选择推理执行提供者 / Auto select inference execution provider + + Returns: + bool: 是否成功 / Whether successful + """ + ... + + def set_gpu(self, gpu_id: int) -> bool: + """ + Deprecated, please use `use_directml`, `use_coreml` or `use_cuda` instead. + """ + ... + + def set_cpu(self) -> bool: + """ + Deprecated, please use `use_cpu` instead. + """ + ... + + def set_auto_device(self) -> bool: + """ + Deprecated, please use `use_auto_ep` instead. + """ + ... + + def custom_recognition(self, name: str) -> Callable[[type[CustomRecognition]], type[CustomRecognition]]: + """自定义识别器装饰器 / Custom recognition decorator + + Args: + name: 识别器名称，需与 Pipeline 中的 custom_recognition 字段匹配 + Recognition name, should match the custom_recognition field in Pipeline + + Returns: + 装饰器函数 / Decorator function + """ + ... + + def register_custom_recognition(self, name: str, recognition: CustomRecognition) -> bool: + """注册自定义识别器 / Register a custom recognizer + + Args: + name: 名称 / Name + recognition: 自定义识别器 / Custom recognizer + + Returns: + bool: 是否成功 / Whether successful + """ + ... + + def unregister_custom_recognition(self, name: str) -> bool: + """移除自定义识别器 / Remove the custom recognizer + + Args: + name: 名称 / Name + + Returns: + bool: 是否成功 / Whether successful + """ + ... + + def clear_custom_recognition(self) -> bool: + """移除所有自定义识别器 / Remove all custom recognizers + + Returns: + bool: 是否成功 / Whether successful + """ + ... + + def custom_action(self, name: str) -> Callable[[type[CustomAction]], type[CustomAction]]: + """自定义动作装饰器 / Custom action decorator + + Args: + name: 动作名称，需与 Pipeline 中的 custom_action 字段匹配 + Action name, should match the custom_action field in Pipeline + + Returns: + 装饰器函数 / Decorator function + """ + ... + + def register_custom_action(self, name: str, action: CustomAction) -> bool: + """注册自定义操作 / Register a custom action + + Args: + name: 名称 / Name + action: 自定义操作 / Custom action + + Returns: + bool: 是否成功 / Whether successful + """ + ... + + def unregister_custom_action(self, name: str) -> bool: + """移除自定义操作 / Remove the custom action + + Args: + name: 名称 / Name + + Returns: + bool: 是否成功 / Whether successful + """ + ... + + def clear_custom_action(self) -> bool: + """移除所有自定义操作 / Remove all custom actions + + Returns: + bool: 是否成功 / Whether successful + """ + ... + + @property + def node_list(self) -> list[str]: + """获取任务列表 / Get task list + + Returns: + list[str]: 任务名列表 / List of task names + + Raises: + RuntimeError: 如果获取失败 + """ + ... + + @property + def custom_recognition_list(self) -> list[str]: + """获取已注册的自定义识别器列表 / Get registered custom recognizer list + + Returns: + list[str]: 自定义识别器名列表 / List of custom recognizer names + + Raises: + RuntimeError: 如果获取失败 + """ + ... + + @property + def custom_action_list(self) -> list[str]: + """获取已注册的自定义操作列表 / Get registered custom action list + + Returns: + list[str]: 自定义操作名列表 / List of custom action names + + Raises: + RuntimeError: 如果获取失败 + """ + ... + + @property + def hash(self) -> str: + """获取资源 hash / Get resource hash + + Returns: + str: 资源 hash / Resource hash + + Raises: + RuntimeError: 如果获取失败 + """ + ... + + _sink_holder: dict[int, ResourceEventSink] = ... + def add_sink(self, sink: ResourceEventSink) -> Optional[int]: + """添加资源事件监听器 / Add resource event listener + + Args: + sink: 事件监听器 / Event sink + + Returns: + Optional[int]: 监听器 id，失败返回 None / Listener id, or None if failed + """ + ... + + def remove_sink(self, sink_id: int) -> None: + """移除资源事件监听器 / Remove resource event listener + + Args: + sink_id: 监听器 id / Listener id + """ + ... + + def clear_sinks(self) -> None: + """清除所有资源事件监听器 / Clear all resource event listeners""" + ... + + def set_inference(self, execution_provider: int, device_id: int) -> bool: + ... + + _api_properties_initialized: bool = ... + + +class ResourceEventSink(EventSink): + @dataclass + class ResourceLoadingDetail: + res_id: int + path: str + type: str + hash: str + ... + + + def on_resource_loading(self, resource: Resource, noti_type: NotificationType, detail: ResourceLoadingDetail): # -> None: + ... + + def on_raw_notification(self, resource: Resource, msg: str, details: dict[str, Any]) -> None: + ... + + + diff --git a/typings/maa/tasker.pyi b/typings/maa/tasker.pyi new file mode 100644 index 0000000..34883c9 --- /dev/null +++ b/typings/maa/tasker.pyi @@ -0,0 +1,430 @@ +""" +This type stub file was generated by pyright. +""" + +import numpy +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Optional, TYPE_CHECKING, Union +from .controller import Controller +from .define import * +from .event_sink import EventSink, NotificationType +from .job import Job, TaskJob +from .pipeline import JActionParam, JActionType, JRecognitionParam, JRecognitionType +from .resource import Resource +from .context import ContextEventSink + +if TYPE_CHECKING: + ... +class Tasker: + _handle: MaaTaskerHandle + _own: bool + def __init__(self, handle: Optional[MaaTaskerHandle] = ...) -> None: + """创建实例 / Create instance + + Args: + handle: 可选的外部句柄 / Optional external handle + + Raises: + RuntimeError: 如果创建失败 + """ + ... + + def __del__(self): # -> None: + ... + + def bind(self, resource: Resource, controller: Controller) -> bool: + """关联资源和控制器 / Bind resource and controller + + Args: + resource: 资源对象 / Resource object + controller: 控制器对象 / Controller object + + Returns: + bool: 是否成功 / Whether successful + """ + ... + + @property + def resource(self) -> Resource: + """获取关联的资源 / Get bound resource + + Returns: + Resource: 资源对象 / Resource object + + Raises: + RuntimeError: 如果获取失败 + """ + ... + + @property + def controller(self) -> Controller: + """获取关联的控制器 / Get bound controller + + Returns: + Controller: 控制器对象 / Controller object + + Raises: + RuntimeError: 如果获取失败 + """ + ... + + @property + def inited(self) -> bool: + """判断是否正确初始化 / Check if initialized correctly + + Returns: + bool: 是否已正确初始化 / Whether correctly initialized + """ + ... + + def post_task(self, entry: str, pipeline_override: Optional[dict[str, Any]] = ...) -> TaskJob: + """异步执行任务 / Asynchronously execute task + + 这是一个异步操作，会立即返回一个 TaskJob 对象 + This is an asynchronous operation that immediately returns a TaskJob object + + Args: + entry: 任务入口 / Task entry + pipeline_override: 用于覆盖的 json / JSON for overriding + + Returns: + TaskJob: 任务作业对象，可通过 status/wait 查询状态，通过 get() 获取结果，通过 override_pipeline() 动态修改 + pipeline + + Task job object, can query status via status/wait, get result via get(), modify + pipeline via override_pipeline() + """ + ... + + def post_recognition(self, reco_type: JRecognitionType, reco_param: JRecognitionParam, image: numpy.ndarray) -> TaskJob: + """异步执行识别 / Asynchronously execute recognition + + Args: + reco_type: 识别类型 / Recognition type + reco_param: 识别参数 / Recognition parameters + image: 前序截图 / Previous screenshot + + Returns: + TaskJob: 任务作业对象 / Task job object + """ + ... + + def post_action(self, action_type: JActionType, action_param: JActionParam, box: RectType = ..., reco_detail: str = ...) -> TaskJob: + """异步执行操作 / Asynchronously execute action + + Args: + action_type: 操作类型 / Action type + action_param: 操作参数 / Action parameters + box: 前序识别位置 / Previous recognition position + reco_detail: 前序识别详情 / Previous recognition details + + Returns: + TaskJob: 任务作业对象 / Task job object + """ + ... + + @property + def running(self) -> bool: + """判断实例是否还在运行 / Check if instance is still running + + Returns: + bool: 是否正在运行 / Whether running + """ + ... + + def post_stop(self) -> Job: + """异步停止实例 / Asynchronously stop instance + + 这是一个异步操作，会立即返回一个 Job 对象 + 停止操作会中断当前运行的任务，并停止资源加载和控制器操作 + This is an asynchronous operation that immediately returns a Job object + The stop operation will interrupt the currently running task and stop resource loading and + controller operations + + Returns: + Job: 作业对象，可通过 status/wait 查询状态 / Job object, can query status via status/wait + """ + ... + + @property + def stopping(self) -> bool: + """判断实例是否正在停止中(尚未停止) / Check if instance is stopping (not yet stopped) + + Returns: + bool: 是否正在停止 / Whether stopping + """ + ... + + def get_latest_node(self, name: str) -> Optional[NodeDetail]: + """获取任务的最新节点号 / Get latest node id for task + + Args: + name: 任务名 / Task name + + Returns: + Optional[NodeDetail]: 节点详情，如果不存在则返回 None / Node detail, or None if not exists + """ + ... + + def clear_cache(self) -> bool: + """清理所有可查询的信息 / Clear all queryable information + + Returns: + bool: 是否成功 / Whether successful + """ + ... + + def override_pipeline(self, task_id: int, pipeline_override: dict[str, Any]) -> bool: + """覆盖指定任务的 pipeline / Override pipeline for specified task + + 在任务执行期间动态修改 pipeline 配置 + Dynamically modify pipeline configuration during task execution + + Args: + task_id: 任务 ID / Task ID + pipeline_override: 用于覆盖的 json / JSON for overriding + + Returns: + bool: 是否成功 / Whether successful + """ + ... + + _sink_holder: dict[int, EventSink] = ... + def add_sink(self, sink: TaskerEventSink) -> Optional[int]: + """添加实例事件监听器 / Add instance event listener + + Args: + sink: 事件监听器 / Event sink + + Returns: + Optional[int]: 监听器 id，失败返回 None / Listener id, or None if failed + """ + ... + + def remove_sink(self, sink_id: int) -> None: + """移除实例事件监听器 / Remove instance event listener + + Args: + sink_id: 监听器 id / Listener id + """ + ... + + def clear_sinks(self) -> None: + """清除所有实例事件监听器 / Clear all instance event listeners""" + ... + + def add_context_sink(self, sink: ContextEventSink) -> Optional[int]: + """添加上下文事件监听器 / Add context event listener + + Args: + sink: 上下文事件监听器 / Context event sink + + Returns: + Optional[int]: 监听器 id，失败返回 None / Listener id, or None if failed + """ + ... + + def remove_context_sink(self, sink_id: int) -> None: + """移除上下文事件监听器 / Remove context event listener + + Args: + sink_id: 监听器 id / Listener id + """ + ... + + def clear_context_sinks(self) -> None: + """清除所有上下文事件监听器 / Clear all context event listeners""" + ... + + def get_recognition_detail(self, reco_id: int) -> Optional[RecognitionDetail]: + """获取识别信息 / Get recognition info + + Args: + reco_id: 识别号 / Recognition id + + Returns: + Optional[RecognitionDetail]: 识别详情，如果不存在则返回 None + Recognition detail, or None if not exists + """ + ... + + def get_action_detail(self, action_id: int) -> Optional[ActionDetail]: + """获取操作信息 / Get action info + + Args: + action_id: 操作号 / Action id + + Returns: + Optional[ActionDetail]: 操作详情，如果不存在则返回 None / Action detail, or None if not exists + """ + ... + + def get_wait_freezes_detail(self, wf_id: int) -> Optional[WaitFreezesDetail]: + """获取等待画面静止信息 / Get wait freezes info + + Args: + wf_id: Wait Freezes ID + + Returns: + Optional[WaitFreezesDetail]: 等待画面静止详情 / Wait freezes detail, or None if not exists + """ + ... + + def get_node_detail(self, node_id: int) -> Optional[NodeDetail]: + """获取节点信息 / Get node info + + Args: + node_id: 节点号 / Node id + + Returns: + Optional[NodeDetail]: 节点详情，如果不存在则返回 None / Node detail, or None if not exists + """ + ... + + def get_task_detail(self, task_id: int) -> Optional[TaskDetail]: + """获取任务信息 / Get task info + + Args: + task_id: 任务号 / Task id + + Returns: + Optional[TaskDetail]: 任务详情，如果不存在则返回 None / Task detail, or None if not exists + """ + ... + + @staticmethod + def set_log_dir(path: Union[Path, str]) -> bool: + """设置日志路径 / Set the log path + + Args: + path: 日志路径 / Log path + + Returns: + bool: 是否成功 / Whether successful + """ + ... + + @staticmethod + def set_save_draw(save_draw: bool) -> bool: + """设置是否将识别保存到日志路径/vision中 / Set whether to save recognition results to log path/vision + + 开启后 RecoDetail 将可以获取到 draws / When enabled, RecoDetail can retrieve draws + + Args: + save_draw: 是否保存 / Whether to save + + Returns: + bool: 是否成功 / Whether successful + """ + ... + + @staticmethod + def set_recording(recording: bool) -> bool: + """ + Deprecated + """ + ... + + @staticmethod + def set_stdout_level(level: LoggingLevelEnum) -> bool: + """设置日志输出到 stdout 中的级别 / Set the log output level to stdout + + Args: + level: 日志级别 / Logging level + + Returns: + bool: 是否成功 / Whether successful + """ + ... + + @staticmethod + def set_debug_mode(debug_mode: bool) -> bool: + """设置是否启用调试模式 / Set whether to enable debug mode + + 调试模式下, RecoDetail 将可以获取到 raw/draws; 所有任务都会被视为 focus 而产生回调 + In debug mode, RecoDetail can retrieve raw/draws; all tasks are treated as focus and produce + callbacks + + Args: + debug_mode: 是否启用调试模式 / Whether to enable debug mode + + Returns: + bool: 是否成功 / Whether successful + """ + ... + + @staticmethod + def set_save_on_error(save_on_error: bool) -> bool: + """设置是否在错误时保存截图到日志路径/on_error中 + Set whether to save screenshot on error to log path/on_error + + Args: + save_on_error: 是否保存 / Whether to save + + Returns: + bool: 是否成功 / Whether successful + """ + ... + + @staticmethod + def set_draw_quality(quality: int) -> bool: + """设置识别可视化图像的 JPEG 质量 / Set the JPEG quality for recognition visualization images + + Args: + quality: JPEG 质量（0-100），默认 85 / JPEG quality (0-100), default 85 + + Returns: + bool: 是否成功 / Whether successful + """ + ... + + @staticmethod + def set_reco_image_cache_limit(limit: int) -> bool: + """设置识别图像缓存数量限制 / Set the recognition image cache limit + + Args: + limit: 缓存数量限制，默认 4096 / Cache limit, default 4096 + + Returns: + bool: 是否成功 / Whether successful + """ + ... + + @staticmethod + def load_plugin(path: Union[Path, str]) -> bool: + """加载插件 / Load plugin + + 可以使用完整路径或仅使用名称, 仅使用名称时会在系统目录和当前目录中搜索. 也可以递归搜索目录中的插件 + Can use full path or name only. When using name only, will search in system directory and + current directory. Can also recursively search for plugins in a directory + + Args: + path: 插件库路径或名称 / Plugin library path or name + + Returns: + bool: 是否成功 / Whether successful + """ + ... + + _api_properties_initialized: bool = ... + + +class TaskerEventSink(EventSink): + @dataclass + class TaskerTaskDetail: + task_id: int + entry: str + uuid: str + hash: str + ... + + + def on_tasker_task(self, tasker: Tasker, noti_type: NotificationType, detail: TaskerTaskDetail) -> None: + ... + + def on_raw_notification(self, tasker: Tasker, msg: str, details: dict[str, Any]) -> None: + ... + + + diff --git a/typings/maa/toolkit.pyi b/typings/maa/toolkit.pyi new file mode 100644 index 0000000..2602504 --- /dev/null +++ b/typings/maa/toolkit.pyi @@ -0,0 +1,141 @@ +""" +This type stub file was generated by pyright. +""" + +import ctypes +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Optional, Union +from .define import * + +@dataclass +class AdbDevice: + """ADB 设备信息 / ADB device information + + 通过 Toolkit.find_adb_devices 获取。 + Obtained via Toolkit.find_adb_devices. + + Attributes: + name: 设备名称 / Device name + adb_path: adb 可执行文件路径 / Path to adb executable + address: 设备地址 (如 127.0.0.1:5555) / Device address (e.g., 127.0.0.1:5555) + screencap_methods: 可用的截图方式位掩码 / Available screenshot methods bitmask + input_methods: 可用的输入方式位掩码 / Available input methods bitmask + config: 额外配置信息 / Extra configuration + """ + name: str + adb_path: Path + address: str + screencap_methods: int + input_methods: int + config: dict[str, Any] + ... + + +@dataclass +class DesktopWindow: + """桌面窗口信息 / Desktop window information + + 通过 Toolkit.find_desktop_windows 获取。 + Obtained via Toolkit.find_desktop_windows. + + Attributes: + hwnd: 窗口句柄 / Window handle + class_name: 窗口类名 / Window class name + window_name: 窗口标题 / Window title + """ + hwnd: ctypes.c_void_p + class_name: str + window_name: str + ... + + +class Toolkit: + """工具包 / Toolkit + + 提供设备发现、配置初始化等辅助功能。 + Provides auxiliary functions such as device discovery and configuration initialization. + """ + @staticmethod + def init_option(user_path: Union[str, Path], default_config: Optional[dict[str, Any]] = ...) -> bool: + """从 user_path 中加载全局配置 / Load global config from user_path + + Args: + user_path: 配置存储路径 / Config storage path + default_config: 默认配置 / Default config + + Returns: + bool: 是否成功 / Whether successful + """ + ... + + @staticmethod + def find_adb_devices(specified_adb: Optional[Union[str, Path]] = ...) -> list[AdbDevice]: + """搜索所有已知安卓模拟器 / Search all known Android emulators + + Args: + specified_adb: 可选，指定 adb 路径进行搜索 / Optional, search using specified adb path + + Returns: + List[AdbDevice]: 设备列表 / Device list + """ + ... + + @staticmethod + def find_desktop_windows() -> list[DesktopWindow]: + """查询所有窗口信息 / Query all window info + + Returns: + List[DesktopWindow]: 窗口列表 / Window list + """ + ... + + @staticmethod + def macos_check_permission(perm: MaaMacOSPermissionEnum) -> bool: + """检查 macOS 权限 / Check macOS permission + + 检查应用是否已获得指定的 macOS 系统权限。 + Check if the application has been granted the specified macOS system permission. + + Args: + perm: 权限类型 / Permission type + + Returns: + bool: 是否已授权 / Whether permission is granted + """ + ... + + @staticmethod + def macos_request_permission(perm: MaaMacOSPermissionEnum) -> bool: + """请求 macOS 权限 / Request macOS permission + + 向用户请求指定的 macOS 系统权限。系统可能会弹出授权对话框。 + Request the specified macOS system permission from user. System may show an authorization + dialog. + + Args: + perm: 权限类型 / Permission type + + Returns: + bool: 是否成功请求（不代表已授权）/ Whether request succeeded (doesn't mean granted) + """ + ... + + @staticmethod + def macos_reveal_permission_settings(perm: MaaMacOSPermissionEnum) -> bool: + """打开 macOS 权限设置 / Open macOS permission settings + + 打开系统偏好设置中对应权限的设置页面。 + Open the corresponding permission settings page in System Preferences. + + Args: + perm: 权限类型 / Permission type + + Returns: + bool: 是否成功打开 / Whether successfully opened + """ + ... + + _api_properties_initialized: bool = ... + +