diff --git a/README.md b/README.md index b780271f3..27b2afecd 100644 --- a/README.md +++ b/README.md @@ -102,6 +102,28 @@ OAS 在其基础上进行了如下优化： - [安装教程](https://runhey.github.io/OnmyojiAutoScript-website/docs/user-manual/installation): 保姆式安装手册,多翻翻有惊喜 - [开发文档](https://runhey.github.io/OnmyojiAutoScript-website/docs/development/preamble): 虽然迭代很多、年久失修，但入门开发必读，具体以源码为准 +## macOS PlayCover / MaaTools + +### 中文 + +macOS 用户可以通过 [PlayCover](https://github.com/PlayCover/PlayCover) 运行 iOS 版游戏，并使用 MaaTools 连接 OAS。环境准备可参考 [MAA macOS 手册](https://docs.maa.plus/zh-cn/manual/device/macos.html)。 + +1. 从 [OASX Flutter 源码](https://github.com/runhey/OASX) 编译并运行本地 app。 +2. 按 [`requirements-macos-playcover.txt`](requirements-macos-playcover.txt) 清单手动安装依赖。 +3. 在 PlayCover 中将游戏设置为 `1280x720`。 +4. 在 OASX 中选择 `PlayCover` / `MacPlayTools`，截图方式选择 `MacBGR`、`RGBA` 或 `MacSCK`；serial 示例为 `localhost:1718`。 +5. `ADB` 与 `minitouch` 仍然保留，可继续用于原有的 Android/模拟器配置。 + +### English + +On macOS, run the iOS game with [PlayCover](https://github.com/PlayCover/PlayCover) and connect OAS through MaaTools. See the [MAA macOS guide](https://docs.maa.plus/zh-cn/manual/device/macos.html) for environment setup. + +1. Build and run a local app from the [OASX Flutter source](https://github.com/runhey/OASX). +2. Manually install the dependencies listed in [`requirements-macos-playcover.txt`](requirements-macos-playcover.txt). +3. Set the game in PlayCover to `1280x720`. +4. In OASX, choose `PlayCover` / `MacPlayTools`, then select `MacBGR`, `RGBA`, or `MacSCK` as the screenshot method; example serial: `localhost:1718`. +5. `ADB` and `minitouch` remain available for existing Android/emulator configurations. + ## 鸣谢 Acknowledgements 感谢所有参与到开发/测试中的朋友们 @@ -123,4 +145,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 a9b2ae7f3..b5df0e752 100644 --- a/module/device/app_control.py +++ b/module/device/app_control.py @@ -17,6 +17,8 @@ def app_is_alive(self, package_name=None) -> bool: 这用于区分“应用被切到后台”和“应用已经被真正杀掉”两种情况。 """ + if getattr(self, 'is_playcover', False): + return True if not package_name: package_name = self.package @@ -31,6 +33,8 @@ def app_is_alive(self, package_name=None) -> bool: return bool(result) def app_is_running(self) -> bool: + if getattr(self, 'is_playcover', False): + return True method = self.config.script.device.control_method # if self.is_wsa: # package = self.app_current_wsa() @@ -44,6 +48,8 @@ def app_is_running(self) -> bool: return package == self.package def app_start(self): + if getattr(self, 'is_playcover', False): + return method = self.config.script.device.screenshot_method logger.info(f'App start: {self.package}') # if self.config.Emulator_Serial == 'wsa-0': @@ -54,6 +60,8 @@ def app_start(self): self.app_start_adb() def app_stop(self): + if getattr(self, 'is_playcover', False): + return method = self.config.script.device.screenshot_method logger.info(f'App stop: {self.package}') if method in AppControl._app_u2_family: @@ -66,6 +74,9 @@ def dump_hierarchy(self) -> etree._Element: Returns: etree._Element: Select elements with `self.hierarchy.xpath('//*[@text="Hermit"]')` for example. """ + if getattr(self, 'is_playcover', False): + self.hierarchy = etree.Element('hierarchy') + return self.hierarchy 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 b6e12ba0c..d5e26cfe5 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,16 @@ 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' + logger.attr('PackageName', self.package) + 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..03be04837 100644 --- a/module/device/connection_attr.py +++ b/module/device/connection_attr.py @@ -34,6 +34,13 @@ 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: + self.config.DEVICE_OVER_HTTP = False + 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 fa237bec2..882541bd6 100644 --- a/module/device/control.py +++ b/module/device/control.py @@ -34,6 +34,7 @@ def click_methods(self): 'ADB': self.click_adb, 'uiautomator2': self.click_uiautomator2, 'minitouch': self.click_minitouch, + 'MacPlayTools': self.click_playcover, # 'Hermit': self.click_hermit, # 'MaaTouch': self.click_maatouch, } @@ -47,7 +48,8 @@ def long_click_methods(self): 'ADB': self.long_click_adb, 'uiautomator2': self.long_click_uiautomator2, 'minitouch': self.long_click_minitouch, - 'scrcpy': self.long_click_scrcpy + 'scrcpy': self.long_click_scrcpy, + 'MacPlayTools': self.long_click_playcover, # 'Hermit': self.click_hermit, # 'MaaTouch': self.click_maatouch, } @@ -119,6 +121,15 @@ def long_click(self, x: int, y: int, duration=(0.5, 2), control_name='LongClick' elapsed = time.perf_counter() - start logger.info(f'{self._format_action_duration(elapsed)}Click {point2str(x, y)} @ {control_name} {duration}') + 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 swipe(self, p1, p2, duration=(0.1, 0.2), control_name='SWIPE', distance_check=True): self.handle_control_check(control_name) p1, p2 = ensure_int(p1, p2) @@ -133,6 +144,8 @@ def swipe(self, p1, p2, duration=(0.1, 0.2), control_name='SWIPE', distance_chec swipe_log = 'Swipe %s -> %s, %s' % (point2str(*p1), point2str(*p2), duration) elif method == 'scrcpy': swipe_log = 'Swipe %s -> %s' % (point2str(*p1), point2str(*p2)) + elif method == 'MacPlayTools': + swipe_log = 'Swipe %s -> %s, %s' % (point2str(*p1), point2str(*p2), duration) # elif method == 'MaaTouch': # logger.info('Swipe %s -> %s' % (point2str(*p1), point2str(*p2))) else: @@ -164,6 +177,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: @@ -204,7 +219,12 @@ def drag(self, p1, p2, segments=1, shake=(0, 15), point_random=(-10, -10, 10, 10 self.handle_control_check(name) p1, p2 = ensure_int(p1, p2) drag_log = 'Drag %s -> %s' % (point2str(*p1), point2str(*p2)) - method = self.config.script.emulator.control_method + configured_method = self.config.script.device.control_method + if configured_method == 'MacPlayTools' or getattr(self, 'is_playcover', False): + method = configured_method + self._invalidate_image_batch_cache() + else: + method = self.config.script.emulator.control_method start = time.perf_counter() if method == 'minitouch': self.drag_minitouch(p1, p2, point_random=point_random) @@ -214,6 +234,8 @@ def drag(self, p1, p2, segments=1, shake=(0, 15), point_random=(-10, -10, 10, 10 swipe_duration=swipe_duration, shake_duration=shake_duration) elif method == 'scrcpy': self.drag_scrcpy(p1, p2, point_random=point_random) + elif method == 'MacPlayTools': + self.playcover_client.swipe(p1, p2, duration=ensure_time(swipe_duration)) # elif method == 'MaaTouch': # self.drag_maatouch(p1, p2, point_random=point_random) 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 398282c34..9ad175879 100644 --- a/module/device/screenshot.py +++ b/module/device/screenshot.py @@ -50,6 +50,9 @@ def screenshot_methods(self): 'DroidCast': self.screenshot_droidcast, 'DroidCast_raw': self.screenshot_droidcast_raw, 'scrcpy': self.screenshot_scrcpy, + 'MacBGR': self.screenshot_playcover, + 'RGBA': self.screenshot_playcover, + 'MacSCK': self.screenshot_playcover, } if IS_WINDOWS: methods.update({ @@ -58,6 +61,15 @@ def screenshot_methods(self): }) return methods + def screenshot_playcover(self): + image = self.playcover_client.screenshot() + width, height = image_size(image) + if (width, height) != (1280, 720): + raise RequestHumanTakeover( + f'PlayCover screenshot is {width}x{height}; OAS requires 1280x720' + ) + return image + def screenshot(self): """ Returns: diff --git a/requirements-macos-playcover.txt b/requirements-macos-playcover.txt new file mode 100644 index 000000000..6390ca203 --- /dev/null +++ b/requirements-macos-playcover.txt @@ -0,0 +1,71 @@ +# OAS macOS PlayCover runtime, derived from requirements.txt. +# PlayCover transport does not use ADB, but these packages are retained for compatibility with existing module imports. +adbutils==0.11.0 +annotated-types==0.7.0 +anyio==3.7.1 +anytree==2.8.0 +cached-property==1.5.2 +certifi==2023.11.17 +cffi==1.16.0 +charset-normalizer==3.3.2 +click==8.1.7 +cn2an==0.5.23 +coloredlogs==15.0.1 +cryptography==42.0.8 +decorator==5.1.1 +filelock==3.13.1 +fastapi==0.104.1 +flatbuffers==23.5.26 +future==0.18.3 +gevent==23.9.1 +greenlet==3.0.3 +h11==0.14.0 +humanfriendly==10.0 +idna==3.6 +inflection==0.5.1 +lxml==5.1.0 +markdown-it-py==2.2.0 +mdurl==0.1.2 +mpmath==1.3.0 +msgpack==1.0.7 +numpy==1.24.3 +oas-checkin-biggod==0.0.1 +oashya==0.0.7 +onepush==1.3.0 +onnxruntime==1.16.3 +opencv-python==4.7.0.72 +packaging==20.9 +paho-mqtt==1.6.1 +pillow==10.2.0 +ppocr-onnx==0.0.3.9 +proces==0.1.7 +protobuf==4.25.1 +psutil==6.1.1 +pyclipper==1.3.0.post5 +pycparser==2.21 +pycryptodome==3.21.0 +pydantic==2.10.0 +pydantic-core==2.27.0 +pygments==2.17.2 +pyparsing==3.1.1 +python-multipart==0.0.9 +pyyaml==6.0 +pyzmq==25.1.2 +requests>=2.32.2,<3 +rich==13.3.5 +shapely==2.0.2 +six==1.16.0 +sniffio==1.3.0 +starlette==0.27.0 +sympy==1.12 +tqdm==4.65.0 +typing-extensions==4.12.2 +uiautomator2==2.16.17 +uiautomator2cache==0.3.0.1 +urllib3==2.1.0 +uvicorn==0.38.0 +websockets==13.1 +wrapt==1.15.0 +zerorpc==0.6.3 +zope-event==5.0 +zope-interface==6.1 diff --git a/tasks/Script/config_device.py b/tasks/Script/config_device.py index 9b4c49e5d..2d3122198 100644 --- a/tasks/Script/config_device.py +++ b/tasks/Script/config_device.py @@ -31,6 +31,9 @@ class ScreenshotMethod(str, Enum): SCRCPY = 'scrcpy' WINDOW_BACKGROUND = 'window_background' NEMU_IPC = 'nemu_ipc' + MAC_BGR = 'MacBGR' + RGBA = 'RGBA' + MAC_SCK = 'MacSCK' class ControlMethod(str, Enum): @@ -38,6 +41,7 @@ class ControlMethod(str, Enum): UIAUTOMATOR2 = 'uiautomator2' MINITOUCH = 'minitouch' WINDOW_MESSAGE = 'window_message' + MAC_PLAYTOOLS = 'MacPlayTools' class EmulatorInfoType(str, Enum): 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..e660170bc --- /dev/null +++ b/tests/test_playcover_protocol.py @@ -0,0 +1,182 @@ +import struct +import sys +import unittest +from types import ModuleType +from unittest.mock import patch + +try: + import cv2 + import numpy as np +except ImportError: + class _FakeImage: + def __init__(self, payload, shape): + self.payload = bytes(payload) + self.shape = tuple(shape) + + def reshape(self, shape): + self.shape = tuple(shape) + return self + + def tolist(self): + height, width, channels = self.shape + return [ + [ + list( + self.payload[ + (row * width + column) * channels: + (row * width + column + 1) * channels + ] + ) + for column in range(width) + ] + for row in range(height) + ] + + def _frombuffer(payload, dtype=None): + return _FakeImage(payload, (len(payload), 1, 1)) + + def _cvt_color(image, code): + height, width, channels = image.shape + pixels = [ + image.payload[offset:offset + channels] + for offset in range(0, len(image.payload), channels) + ] + if code == 1: + converted = b"".join(pixel[::-1] for pixel in pixels) + elif code == 2: + converted = b"".join(pixel[:3] for pixel in pixels) + else: + raise AssertionError(f"unexpected color conversion code: {code!r}") + return _FakeImage(converted, (height, width, 3)) + + np = ModuleType("numpy") + np.uint8 = object() + np.frombuffer = _frombuffer + cv2 = ModuleType("cv2") + cv2.COLOR_BGR2RGB = 1 + cv2.COLOR_RGBA2RGB = 2 + cv2.cvtColor = _cvt_color + sys.modules["numpy"] = np + sys.modules["cv2"] = cv2 + +from module.device.method.playcover import ( + PlayCoverClient, + PlayCoverProtocolError, +) + + +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 + + +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()