diff --git a/README.md b/README.md index ce606ab2c..dad4cbd2b 100644 --- a/README.md +++ b/README.md @@ -77,6 +77,46 @@ OAS 在其基础上进行了如下优化： - [ppocr-onnx](https://github.com/triwinds/ppocr-onnx): OCR 库，基于 onnxruntime 和 PaddleOCR - [gurs](https://github.com/2833844911/gurs): 基于赛贝尔曲线模拟滑动轨迹, 引入其轨迹模拟人手滑动 +## macOS PlayCover / MaaTools（实验性支持） + +### 中文说明 + +本实验性接入借鉴了 MAA（明日方舟小助手）的 [macOS PlayCover / MaaTools 接入思路](https://docs.maa.plus/zh-cn/manual/device/macos.html)，并使用 [OASX](https://github.com/runhey/OASX) 作为界面。我们已从 OASX Flutter 源码编译出一个可在 macOS 本地运行的 app，用于本集成验证；本 PR 不上传或分发该二进制。此功能仍处于试验性阶段，不代表官方 release 已经包含 macOS app。 + +Flutter app 仅是界面，不内置 Python 解释器、OCR、本地 OAS 服务或其依赖。请在 macOS 主机上安装 Python 3.10，创建 venv，并手动安装 [requirements-macos-playcover.txt](requirements-macos-playcover.txt) 中的 pip 包： + +```bash +python3.10 -m venv .venv +source .venv/bin/activate +python -m pip install --upgrade pip +python -m pip install -r requirements-macos-playcover.txt +``` + +这份文件只是 pip 包清单，不包含 Python 解释器或 OCR 模型文件；这些内容也不会由 Flutter app 内置提供。 + +PlayCover 需要 MaaTools。请先按 [MAA macOS 手册](https://docs.maa.plus/zh-cn/manual/device/macos.html) 配置 PlayCover/MaaTools，然后在 OASX 现有下拉框中选择 PlayCover（`control=MacPlayTools`），`screenshot` 选择 `MacBGR`、`RGBA` 或 `MacSCK`，并将 `serial` 填为标题栏中的 `localhost:1718`。PlayCover、ADB 和 minitouch 是并列可选的控制方式；选择 `adb` 或 `minitouch` 时使用 Android 模拟器。 + +请在 PlayCover 中将游戏分辨率设置为 1280×720。OAS 要求使用该分辨率，否则无法正常识别和点击。 + +### English + +This experimental integration follows the [macOS PlayCover / MaaTools approach used by MAA (MaaAssistantArknights)](https://docs.maa.plus/zh-cn/manual/device/macos.html) and uses [OASX](https://github.com/runhey/OASX) as its UI. We compiled an app from the OASX Flutter source that runs locally on macOS for integration validation; this PR does not upload or distribute that binary. This remains experimental support and does not claim that an official release already includes a macOS app. + +The Flutter app is UI-only. It does not bundle a Python interpreter, OCR, the OAS service, or its dependencies. On the macOS host, install Python 3.10, create a venv, and manually install the pip packages in [requirements-macos-playcover.txt](requirements-macos-playcover.txt): + +```bash +python3.10 -m venv .venv +source .venv/bin/activate +python -m pip install --upgrade pip +python -m pip install -r requirements-macos-playcover.txt +``` + +That file is only a pip package list; it does not include the Python interpreter or OCR model files, and the Flutter app does not bundle or provide them. + +PlayCover requires MaaTools. First configure PlayCover/MaaTools according to the [MAA macOS manual](https://docs.maa.plus/zh-cn/manual/device/macos.html), then in OASX's existing dropdowns select PlayCover (`control=MacPlayTools`), choose `MacBGR`, `RGBA`, or `MacSCK` for `screenshot`, and set `serial` to `localhost:1718` shown in the title bar. PlayCover, ADB, and minitouch are parallel control options; selecting `adb` or `minitouch` uses the Android emulator. + +Set the game resolution in PlayCover to 1280×720. OAS requires this resolution; otherwise recognition and tapping will not work correctly. + ## 联系/加入我们 Contact/Join Us @@ -123,4 +163,4 @@ OAS 在其基础上进行了如下优化： ![](https://profile-counter.glitch.me/runhey-OnmyojiAutoScript/count.svg) - \ No newline at end of file + diff --git a/module/device/app_control.py b/module/device/app_control.py index 6e99279d7..b06e26db4 100644 --- a/module/device/app_control.py +++ b/module/device/app_control.py @@ -6,12 +6,13 @@ # from module.device.method.wsa import WSA from module.logger import logger - class AppControl(Adb, Uiautomator2): hierarchy: etree._Element _app_u2_family = ['uiautomator2', 'minitouch', 'scrcpy'] def app_is_running(self) -> bool: + if self.is_playcover: + return True method = self.config.script.device.control_method # if self.is_wsa: # package = self.app_current_wsa() @@ -25,6 +26,8 @@ def app_is_running(self) -> bool: return package == self.package def app_start(self): + if self.is_playcover: + return method = self.config.script.device.screenshot_method logger.info(f'App start: {self.package}') # if self.config.Emulator_Serial == 'wsa-0': @@ -35,6 +38,8 @@ def app_start(self): self.app_start_adb() def app_stop(self): + if self.is_playcover: + return method = self.config.script.device.screenshot_method logger.info(f'App stop: {self.package}') if method in AppControl._app_u2_family: @@ -47,6 +52,8 @@ def dump_hierarchy(self) -> etree._Element: Returns: etree._Element: Select elements with `self.hierarchy.xpath('//*[@text="Hermit"]')` for example. """ + if self.is_playcover: + return None method = self.config.script.device.screenshot_method if method in AppControl._app_u2_family: self.hierarchy = self.dump_hierarchy_uiautomator2() diff --git a/module/device/connection.py b/module/device/connection.py index e0d4072b5..16399e5cd 100644 --- a/module/device/connection.py +++ b/module/device/connection.py @@ -15,6 +15,7 @@ from module.base.decorator import Config, cached_property, del_cached_property from module.base.utils import ensure_time from module.device.connection_attr import ConnectionAttr +from module.device.method.playcover import PlayCoverClient from module.device.method.utils import ( RETRY_TRIES, remove_shell_warning, retry_sleep, handle_adb_error, PackageNotInstalled, @@ -95,6 +96,14 @@ def __init__(self, config): config (AzurLaneConfig, str): Name of the user config under ./config """ super().__init__(config) + if self.is_playcover: + self.playcover_client = PlayCoverClient( + self.serial, + screenshot_mode=self.config.script.device.screenshot_method, + ) + self.playcover_client.connect() + self.package = 'com.netease.onmyoji' + return if not self.is_over_http: self.detect_device() diff --git a/module/device/connection_attr.py b/module/device/connection_attr.py index cee77d2e9..7bd0c6020 100644 --- a/module/device/connection_attr.py +++ b/module/device/connection_attr.py @@ -13,6 +13,7 @@ from module.exception import RequestHumanTakeover from module.logger import logger + class ConnectionAttr: config: Config serial: str @@ -34,6 +35,12 @@ def __init__(self, config): else: self.config = config + self.serial = str(self.config.script.device.serial) + self.is_playcover = self.config.script.device.control_method == 'MacPlayTools' + if self.is_playcover: + logger.attr('PlayCover', self.serial) + return + # Init adb client logger.attr('AdbBinary', self.adb_binary) # Monkey patch to custom adb @@ -65,7 +72,6 @@ def __init__(self, config): # Parse custom serial # self.serial = str(self.config.Emulator_Serial) - self.serial = str(self.config.script.device.serial) self.serial_check() self.config.DEVICE_OVER_HTTP = self.is_over_http @@ -282,5 +288,3 @@ def u2(self) -> u2.Device: logger.attr('u2.Device', f'Device(atx_agent_url={device._get_atx_agent_url()})') return device - - diff --git a/module/device/control.py b/module/device/control.py index 43d4ecb85..f11283688 100644 --- a/module/device/control.py +++ b/module/device/control.py @@ -24,6 +24,7 @@ def click_methods(self): 'uiautomator2': self.click_uiautomator2, 'minitouch': self.click_minitouch, 'window_message': self.click_window_message if IS_WINDOWS else None, + 'MacPlayTools': self.click_playcover, # 'Hermit': self.click_hermit, # 'MaaTouch': self.click_maatouch, } @@ -35,7 +36,8 @@ def long_click_methods(self): 'uiautomator2': self.long_click_uiautomator2, 'minitouch': self.long_click_minitouch, 'window_message': self.long_click_window_message if IS_WINDOWS else None, - 'scrcpy': self.long_click_scrcpy + 'scrcpy': self.long_click_scrcpy, + 'MacPlayTools': self.long_click_playcover, # 'Hermit': self.click_hermit, # 'MaaTouch': self.click_maatouch, } @@ -81,6 +83,15 @@ def click(self, x: int, y: int, control_check=True, control_name='Click') -> Non ) method(x, y) + def click_playcover(self, x, y): + self.playcover_client.click(x, y) + + def long_click_playcover(self, x, y, duration=0.8): + self.playcover_client.long_click(x, y, duration=duration) + + def swipe_playcover(self, p1, p2, duration=0.1): + self.playcover_client.swipe(p1, p2, duration=duration) + def multi_click(self, button, n, interval=(0.1, 0.2)): """ @@ -163,6 +174,8 @@ def swipe(self, p1, p2, duration=(0.1, 0.2), control_name='SWIPE', distance_chec logger.info('Swipe %s -> %s, %s' % (point2str(*p1), point2str(*p2), duration)) elif method == 'scrcpy': logger.info('Swipe %s -> %s' % (point2str(*p1), point2str(*p2))) + elif method == 'MacPlayTools': + logger.info('Swipe %s -> %s' % (point2str(*p1), point2str(*p2))) # elif method == 'MaaTouch': # logger.info('Swipe %s -> %s' % (point2str(*p1), point2str(*p2))) else: @@ -192,6 +205,8 @@ def swipe(self, p1, p2, duration=(0.1, 0.2), control_name='SWIPE', distance_chec self.swipe_uiautomator2(p1, p2, duration=duration) elif method == 'scrcpy': self.swipe_scrcpy(p1, p2) + elif method == 'MacPlayTools': + self.swipe_playcover(p1, p2, duration=duration) # elif method == 'MaaTouch': # self.swipe_maatouch(p1, p2) else: diff --git a/module/device/method/playcover.py b/module/device/method/playcover.py new file mode 100644 index 000000000..4cfd99ae0 --- /dev/null +++ b/module/device/method/playcover.py @@ -0,0 +1,203 @@ +import socket +import struct +import time + +import cv2 +import numpy as np + + +class PlayCoverError(RuntimeError): + pass + + +class PlayCoverProtocolError(PlayCoverError): + pass + + +class PlayCoverClient: + HANDSHAKE = b'MAA\x00' + HANDSHAKE_OK = b'OKAY' + DEFAULT_TIMEOUT = 10.0 + MAX_FRAME_BYTES = 256 * 1024 * 1024 + TOUCH_BEGAN = 0 + TOUCH_MOVED = 1 + TOUCH_ENDED = 3 + + def __init__(self, address, *, screenshot_mode='MacBGR', + timeout=DEFAULT_TIMEOUT, socket_factory=socket.create_connection): + self.host, self.port = self._parse_address(address) + self.screenshot_mode = getattr(screenshot_mode, 'value', screenshot_mode) + self.timeout = float(timeout) + self.socket_factory = socket_factory + self._socket = None + self.version = None + self.width = None + self.height = None + + @staticmethod + def _parse_address(address): + text = str(address).strip() + if not text or '://' in text: + raise PlayCoverError(f'Invalid PlayCover address: {address!r}') + if text.isdigit(): + host, port_text = '127.0.0.1', text + else: + try: + host, port_text = text.rsplit(':', 1) + except ValueError as exc: + raise PlayCoverError(f'Invalid PlayCover address: {address!r}') from exc + try: + port = int(port_text) + except ValueError as exc: + raise PlayCoverError(f'Invalid PlayCover port: {port_text!r}') from exc + if not host or not 1 <= port <= 65535: + raise PlayCoverError(f'Invalid PlayCover address: {address!r}') + return host, port + + @property + def connected(self): + return self._socket is not None + + @property + def screen_size(self): + if self.width is None or self.height is None: + return None + return self.width, self.height + + def connect(self): + if self._socket is not None: + return self + sock = None + try: + sock = self.socket_factory((self.host, self.port), self.timeout) + sock.settimeout(self.timeout) + sock.sendall(self.HANDSHAKE) + if self.recv_exact(sock, 4) != self.HANDSHAKE_OK: + raise PlayCoverProtocolError('Invalid MaaTools handshake') + self._socket = sock + self._send(b'VERN') + self.version = self._read('>I')[0] + self.width, self.height = self._read_size() + return self + except PlayCoverError: + self._socket = None + self._close(sock) + raise + except OSError as exc: + self._socket = None + self._close(sock) + raise PlayCoverError('PlayCover connection failed') from exc + + def close(self): + sock, self._socket = self._socket, None + self.version = None + self.width = None + self.height = None + self._close(sock) + + def screenshot(self): + self._ensure_connected() + if self.screenshot_mode == 'MacBGR': + return self._screenshot_bgr() + if self.screenshot_mode in ('RGBA', 'MacSCK'): + return self._screenshot_rgba() + raise PlayCoverProtocolError( + f'Unsupported PlayCover screenshot mode: {self.screenshot_mode}' + ) + + def refresh_size(self): + if self._socket is None: + self.connect() + if self.screen_size is None: + self.width, self.height = self._read_size() + return self.width, self.height + + def click(self, x, y): + self.touch(self.TOUCH_BEGAN, x, y) + time.sleep(0.05) + self.touch(self.TOUCH_ENDED, x, y) + + def long_click(self, x, y, duration=0.8): + self.touch(self.TOUCH_BEGAN, x, y) + time.sleep(max(0, float(duration))) + self.touch(self.TOUCH_ENDED, x, y) + + def swipe(self, p1, p2, duration=0.2): + self.touch(self.TOUCH_BEGAN, p1[0], p1[1]) + self.touch(self.TOUCH_MOVED, p2[0], p2[1]) + time.sleep(max(0, float(duration))) + self.touch(self.TOUCH_ENDED, p2[0], p2[1]) + + def touch(self, phase, x, y): + self._ensure_connected() + width, height = self.screen_size + x = max(0, min(width - 1, int(x))) + y = max(0, min(height - 1, int(y))) + self._send(b'TUCH', bytes((int(phase),)) + struct.pack('>HH', x, y)) + + def _screenshot_bgr(self): + self._send(b'BGR\x01') + width, height, length = self._read('>III') + data = self._frame(width, height, length, 3) + self.width, self.height = width, height + image = np.frombuffer(data, dtype=np.uint8).reshape((height, width, 3)) + return cv2.cvtColor(image, cv2.COLOR_BGR2RGB) + + def _screenshot_rgba(self): + width, height = self.screen_size + self._send(b'SCRN') + length = self._read('>I')[0] + data = self._frame(width, height, length, 4) + image = np.frombuffer(data, dtype=np.uint8).reshape((height, width, 4)) + return cv2.cvtColor(image, cv2.COLOR_RGBA2RGB) + + def _read_size(self): + self._send(b'SIZE') + width, height = self._read('>HH') + if width <= 0 or height <= 0 or width * height * 4 > self.MAX_FRAME_BYTES: + raise PlayCoverProtocolError(f'Invalid MaaTools size: {width}x{height}') + return width, height + + def _frame(self, width, height, length, channels): + expected = width * height * channels + if width <= 0 or height <= 0 or length != expected or length > self.MAX_FRAME_BYTES: + raise PlayCoverProtocolError( + f'Invalid MaaTools frame length: {length}, expected {expected}' + ) + return self.recv_exact(self._socket, length) + + def _read(self, fmt): + return struct.unpack(fmt, self.recv_exact(self._socket, struct.calcsize(fmt))) + + def _send(self, command, payload=b''): + size = len(command) + len(payload) + if len(command) != 4 or size > 0xffff: + raise PlayCoverProtocolError('Invalid MaaTools command') + if self._socket is None: + raise PlayCoverError('PlayCover socket is not connected') + self._socket.sendall(struct.pack('>H', size) + command + payload) + + def _ensure_connected(self): + if self._socket is None: + self.connect() + + @staticmethod + def recv_exact(sock, size): + data = bytearray() + while len(data) < size: + try: + chunk = sock.recv(size - len(data)) + except OSError as exc: + raise PlayCoverError('PlayCover socket read failed') from exc + if not chunk: + raise PlayCoverError('PlayCover socket closed') + data.extend(chunk) + return bytes(data) + + @staticmethod + def _close(sock): + if sock is not None: + try: + sock.close() + except OSError: + pass diff --git a/module/device/screenshot.py b/module/device/screenshot.py index 90d29ffe0..3ae9acc24 100644 --- a/module/device/screenshot.py +++ b/module/device/screenshot.py @@ -45,9 +45,15 @@ def screenshot_methods(self): 'DroidCast_raw': self.screenshot_droidcast_raw, 'scrcpy': self.screenshot_scrcpy, 'window_background': self.screenshot_window_background if IS_WINDOWS else None, - 'nemu_ipc': self.screenshot_nemu_ipc + 'nemu_ipc': self.screenshot_nemu_ipc, + 'MacBGR': self.screenshot_playcover, + 'RGBA': self.screenshot_playcover, + 'MacSCK': self.screenshot_playcover, } + def screenshot_playcover(self): + return self.playcover_client.screenshot() + def screenshot(self): """ Returns: diff --git a/requirements-macos-playcover.txt b/requirements-macos-playcover.txt new file mode 100644 index 000000000..ba32fe34f --- /dev/null +++ b/requirements-macos-playcover.txt @@ -0,0 +1,61 @@ +# macOS PlayCover / MaaTools manual installation (Python 3.10) +# macOS PlayCover / MaaTools 手动安装（Python 3.10） +# +# Run these commands in a macOS terminal, outside the Flutter app: +# 请在 macOS 终端中、Flutter app 外执行以下命令： +# python3.10 -m venv .venv +# source .venv/bin/activate +# python -m pip install --upgrade pip +# python -m pip install -r requirements-macos-playcover.txt +# +# This is a pip package list only. It does not include the Python interpreter +# or OCR model files. The Flutter app is UI-only and does not bundle Python, +# OCR, the OAS service, or its runtime dependencies. +# 这只是 pip 包清单，不包含 Python 解释器或 OCR 模型文件。Flutter app 仅是 +# 界面，不内置 Python、OCR、本地 OAS 服务或服务依赖。 +# +# Unrelated Windows, instrumentation, and check-in packages are intentionally +# not part of this PlayCover PR. +# 本 PlayCover PR 有意不包含 Windows、注入及签到相关的无关依赖。 + +# Image processing and OCR / 图像处理与 OCR +numpy==1.24.3 +opencv-python==4.7.0.72 +Pillow==10.2.0 +ppocr-onnx==0.0.3.9 + +# OAS service and OASX communication / 本地 OAS 服务与 OASX 通信 +pydantic==2.10.0 +fastapi==0.104.1 +uvicorn==0.38.0 +websockets==13.1 +zerorpc==0.6.3 +paho-mqtt==1.6.1 +requests==2.31.0 + +# Cross-platform import-time compatibility / 跨平台 import-time 兼容 +# Existing OAS modules import these at module load time, including the +# ADB/uiautomator2/minitouch code paths. PlayCover transport itself does not +# use ADB; keep these packages only so the existing modules can be imported. +# OAS 现有模块加载时仍会导入这些包，包括 ADB/uiautomator2/minitouch 代码路径。 +# PlayCover 传输本身不使用 ADB；保留它们只是为了现有模块能够完成导入。 +adbutils==0.11.0 +uiautomator2==2.16.17 +uiautomator2cache==0.3.0.1 +wrapt==1.15.0 +lxml==5.0.0 + +# OAS runtime utilities / OAS 运行时工具 +psutil==6.1.1 +rich==13.3.5 +PyYAML==6.0 +tqdm==4.65.0 +anytree==2.8.0 +cn2an==0.5.23 +inflection==0.5.1 +onepush==1.3.0 +cryptography>=42.0.7 + +# Transitive dependencies (for example onnxruntime and pyzmq) are resolved +# by pip and are intentionally not expanded into this manual list. +# onnxruntime、pyzmq 等传递依赖由 pip 解析，本手动清单不逐项展开。 diff --git a/tasks/Script/config_device.py b/tasks/Script/config_device.py index feee62e8c..6ea11ec10 100644 --- a/tasks/Script/config_device.py +++ b/tasks/Script/config_device.py @@ -25,12 +25,16 @@ class ScreenshotMethod(str, Enum): SCRCPY = 'scrcpy' WINDOW_BACKGROUND = 'window_background' NEMU_IPC = 'nemu_ipc' + MacBGR = 'MacBGR' + RGBA = 'RGBA' + MacSCK = 'MacSCK' class ControlMethod(str, Enum): ADB = 'adb' UIAUTOMATOR2 = 'uiautomator2' MINITOUCH = 'minitouch' WINDOW_MESSAGE = 'window_message' + MacPlayTools = 'MacPlayTools' class EmulatorInfoType(str, Enum): # module.device.platform2.emulator_base.EmulatorBase diff --git a/tests/test_playcover_integration.py b/tests/test_playcover_integration.py new file mode 100644 index 000000000..57c7134ff --- /dev/null +++ b/tests/test_playcover_integration.py @@ -0,0 +1,134 @@ +import json +import unittest +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import Mock, PropertyMock, patch + + +ROOT = Path(__file__).resolve().parents[1] + + +class PlayCoverIntegrationTests(unittest.TestCase): + def test_template_defaults(self): + template = json.loads((ROOT / "config" / "template.json").read_text(encoding="utf-8")) + device = template["script"]["device"] + self.assertEqual(device["serial"], "auto") + self.assertEqual(device["screenshot_method"], "auto") + self.assertEqual(device["control_method"], "minitouch") + + def test_playcover_screenshot_passes_through_1280x720(self): + try: + import numpy as np + from module.device.screenshot import Screenshot + except ImportError as exc: + self.skipTest(f"optional OAS image dependencies unavailable: {exc}") + + image = np.zeros((720, 1280, 3), dtype=np.uint8) + fake = SimpleNamespace( + playcover_client=SimpleNamespace( + screenshot=lambda: image + ) + ) + self.assertIs(Screenshot.screenshot_playcover(fake), image) + + def test_macplaytools_passes_oas_coordinates_to_playcover(self): + try: + from module.device.control import Control + except ImportError as exc: + self.skipTest(f"optional OAS control dependencies unavailable: {exc}") + + client = SimpleNamespace( + click=Mock(), + long_click=Mock(), + swipe=Mock(), + ) + control = SimpleNamespace(playcover_client=client) + + Control.click_playcover(control, 640, 360) + Control.long_click_playcover(control, 640, 360, duration=1.2) + Control.swipe_playcover(control, (10, 20), (1279, 719), duration=0.3) + + client.click.assert_called_once_with(640, 360) + client.long_click.assert_called_once_with(640, 360, duration=1.2) + client.swipe.assert_called_once_with((10, 20), (1279, 719), duration=0.3) + + def test_macplaytools_selects_playcover(self): + try: + from module.device.connection import Connection + from module.device.connection_attr import ConnectionAttr + except ImportError as exc: + self.skipTest(f"optional OAS runtime dependencies unavailable: {exc}") + + device = SimpleNamespace( + serial="localhost:1718", + screenshot_method="MacBGR", + control_method="MacPlayTools", + ) + config = SimpleNamespace(script=SimpleNamespace(device=device)) + with patch("module.device.connection.PlayCoverClient") as playcover_client, \ + patch.object(Connection, "detect_device", side_effect=AssertionError("ADB detect called")), \ + patch.object(Connection, "adb_connect", side_effect=AssertionError("ADB connect called")), \ + patch.object(ConnectionAttr, "adb_client", new_callable=PropertyMock) as adb_client: + connection = Connection(config) + playcover_client.assert_called_once_with( + "localhost:1718", screenshot_mode="MacBGR" + ) + playcover_client.return_value.connect.assert_called_once_with() + adb_client.assert_not_called() + self.assertTrue(connection.is_playcover) + self.assertEqual(connection.package, "com.netease.onmyoji") + + def test_minitouch_connection_uses_adb_path(self): + try: + from tasks.Script.config_device import PackageName + from module.device.connection import Connection + from module.device.connection_attr import ConnectionAttr + except ImportError as exc: + self.skipTest(f"optional OAS runtime dependencies unavailable: {exc}") + + device = SimpleNamespace( + serial="localhost:1718", + screenshot_method="MacBGR", + control_method="minitouch", + package_name=PackageName.AUTO, + ) + config = SimpleNamespace(script=SimpleNamespace(device=device)) + with patch("module.device.connection.PlayCoverClient") as playcover_client, \ + patch.object(Connection, "detect_device", return_value=None) as detect_device, \ + patch.object(Connection, "adb_connect", return_value=None) as adb_connect, \ + patch.object(Connection, "detect_package", return_value=None), \ + patch.object(Connection, "adb", new_callable=PropertyMock) as adb, \ + patch.object(ConnectionAttr, "adb_client", new_callable=PropertyMock) as adb_client, \ + patch( + "module.device.connection_attr.deep_iter", + side_effect=[ + [([], {"type": "oc", "value": True})] * 3, + [], + ], + ): + connection = Connection(config) + + playcover_client.assert_not_called() + self.assertFalse(connection.is_playcover) + detect_device.assert_called_once_with() + adb_connect.assert_called_once_with("localhost:1718") + adb_client.assert_called_once_with() + adb.assert_called_once_with() + + def test_enum_values_exist_when_pydantic_is_available(self): + try: + from tasks.Script.config_device import ControlMethod, Device, ScreenshotMethod + except ImportError as exc: + self.skipTest(f"pydantic unavailable: {exc}") + config = Device() + self.assertEqual(config.serial, "auto") + self.assertEqual(config.screenshot_method, ScreenshotMethod.AUTO) + self.assertEqual(config.control_method, ControlMethod.MINITOUCH) + self.assertIn("MacBGR", [item.value for item in ScreenshotMethod]) + self.assertIn("RGBA", [item.value for item in ScreenshotMethod]) + self.assertIn("MacSCK", [item.value for item in ScreenshotMethod]) + self.assertIn("MacPlayTools", [item.value for item in ControlMethod]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_playcover_protocol.py b/tests/test_playcover_protocol.py new file mode 100644 index 000000000..0f2e4792a --- /dev/null +++ b/tests/test_playcover_protocol.py @@ -0,0 +1,132 @@ +import struct +import unittest +from unittest.mock import patch + +try: + import numpy as np + from module.device.method.playcover import ( + PlayCoverClient, + PlayCoverProtocolError, + ) +except ImportError as exc: + np = None + _IMPORT_ERROR = exc +else: + _IMPORT_ERROR = None + + +class FakeSocket: + def __init__(self, incoming=b"", chunk_size=3): + self.incoming = bytearray(incoming) + self.chunk_size = chunk_size + self.sent = bytearray() + self.timeout = None + self.closed = False + + def settimeout(self, timeout): + self.timeout = timeout + + def sendall(self, data): + if self.closed: + raise OSError("closed") + self.sent.extend(data) + + def recv(self, size): + if self.closed or not self.incoming: + return b"" + take = min(size, self.chunk_size, len(self.incoming)) + result = bytes(self.incoming[:take]) + del self.incoming[:take] + return result + + def close(self): + self.closed = True + + +def handshake_stream(width=1280, height=720, version=3): + return b"OKAY" + struct.pack(">I", version) + struct.pack(">HH", width, height) + + +def command_frames(data): + frames = [] + offset = 4 + while offset < len(data): + length = struct.unpack(">H", data[offset:offset + 2])[0] + offset += 2 + frames.append(bytes(data[offset:offset + length])) + offset += length + return frames + + +@unittest.skipIf(_IMPORT_ERROR is not None, f"optional image dependency unavailable: {_IMPORT_ERROR}") +class PlayCoverProtocolTests(unittest.TestCase): + def make_client(self, incoming, **kwargs): + sock = FakeSocket(incoming, chunk_size=kwargs.pop("chunk_size", 3)) + client = PlayCoverClient( + "localhost:1718", + socket_factory=lambda _address, _timeout: sock, + **kwargs, + ) + return client, sock + + def test_handshake_and_partial_recv(self): + client, sock = self.make_client(handshake_stream(), chunk_size=1) + client.connect() + self.assertEqual(bytes(sock.sent[:4]), b"MAA\x00") + self.assertEqual(command_frames(sock.sent), [b"VERN", b"SIZE"]) + self.assertEqual(client.screen_size, (1280, 720)) + self.assertEqual(sock.timeout, client.timeout) + + def test_bgr_is_decoded_to_rgb(self): + bgr = bytes((1, 2, 3, 10, 20, 30)) + incoming = handshake_stream(2, 1) + struct.pack(">III", 2, 1, len(bgr)) + bgr + client, sock = self.make_client(incoming) + image = client.screenshot() + self.assertEqual(image.shape, (1, 2, 3)) + self.assertEqual(image.tolist(), [[[3, 2, 1], [30, 20, 10]]]) + self.assertEqual(command_frames(sock.sent)[-1], b"BGR\x01") + + def test_scrn_decodes_rgba(self): + rgba = bytes((1, 2, 3, 255, 10, 20, 30, 255)) + incoming = handshake_stream(2, 1) + struct.pack(">I", len(rgba)) + rgba + client, sock = self.make_client(incoming, screenshot_mode="RGBA") + image = client.screenshot() + self.assertEqual(image.shape, (1, 2, 3)) + self.assertEqual(image.tolist(), [[[1, 2, 3], [10, 20, 30]]]) + self.assertEqual(command_frames(sock.sent)[-1], b"SCRN") + + def test_macsck_uses_scrn(self): + rgba = bytes((1, 2, 3, 255)) + incoming = handshake_stream(1, 1) + struct.pack(">I", len(rgba)) + rgba + client, sock = self.make_client(incoming, screenshot_mode="MacSCK") + client.screenshot() + self.assertEqual(command_frames(sock.sent)[-1], b"SCRN") + + def test_touch_phases_are_clamped(self): + client, sock = self.make_client(handshake_stream(10, 5)) + with patch("module.device.method.playcover.time.sleep", return_value=None): + client.click(-10, 99) + client.swipe((-1, -2), (99, 88), duration=0) + touch_frames = [frame for frame in command_frames(sock.sent) if frame[:4] == b"TUCH"] + self.assertEqual([frame[4] for frame in touch_frames], [0, 3, 0, 1, 3]) + for frame in touch_frames: + x, y = struct.unpack(">HH", frame[5:9]) + self.assertLessEqual(x, 9) + self.assertLessEqual(y, 4) + + def test_invalid_frame_length_raises(self): + incoming = handshake_stream(2, 1) + struct.pack(">III", 2, 1, 5) + b"12345" + client, _sock = self.make_client(incoming) + with self.assertRaises(PlayCoverProtocolError): + client.screenshot() + + def test_bare_port_uses_localhost(self): + sock = FakeSocket(handshake_stream()) + client = PlayCoverClient( + "1718", + socket_factory=lambda _address, _timeout: sock, + ) + self.assertEqual((client.host, client.port), ("127.0.0.1", 1718)) + +if __name__ == "__main__": + unittest.main()