feat: support adb client

resolve #44

README.md

@@ -10,8 +10,7 @@
 
 - [x] 支持客户端
   - [x] DMM （前台）
-  - [ ] 安卓 ADB 连接（后台）
-  - [ ] 模拟器窗口（后台）
+  - [x] _实验性_ 安卓 ADB 连接（后台）开发基于 1080x1920 分辨率
 - [x] 团队赛 (Team race)
   - [x] 有胜利确定奖励时吃帕菲
 - [x] 日常赛 (Daily race)

auto_derby/__main__.py

@@ -38,8 +38,14 @@ def main():
         default=config.PLUGINS,
         help="plugin names to enable",
     )
+    parser.add_argument(
+        "--adb",
+        help="adb connect address like `127.0.0.1:5037`",
+        default=config.ADB_ADDRESS,
+    )
     args = parser.parse_args()
     job = avaliable_jobs.get(args.job)
+    adb_address = args.adb
     plugin.reload()
     plugins = args.plugin
     for i in plugins:
@@ -54,22 +60,28 @@ def main():
         )
         exit(1)
 
-    c = clients.DMMClient.find()
-    if not c:
-        if (
-            win32gui.MessageBox(
-                0, "Launch DMM umamusume?", "Can not found window", win32con.MB_YESNO
-            )
-            == 6
-        ):
-            webbrowser.open("dmmgameplayer://umamusume/cl/general/umamusume")
-            while not c:
-                time.sleep(1)
-                LOGGER.info("waiting game launch")
-                c = clients.DMMClient.find()
-            LOGGER.info("game window: %s", c.h_wnd)
-        else:
-            exit(1)
+    if adb_address:
+        c = clients.ADBClient(adb_address)
+    else:
+        c = clients.DMMClient.find()
+        if not c:
+            if (
+                win32gui.MessageBox(
+                    0,
+                    "Launch DMM umamusume?",
+                    "Can not found window",
+                    win32con.MB_YESNO,
+                )
+                == 6
+            ):
+                webbrowser.open("dmmgameplayer://umamusume/cl/general/umamusume")
+                while not c:
+                    time.sleep(1)
+                    LOGGER.info("waiting game launch")
+                    c = clients.DMMClient.find()
+                LOGGER.info("game window: %s", c.h_wnd)
+            else:
+                exit(1)
     c.setup()
     clients.set_current(c)
     job()

auto_derby/_config.py

@@ -7,11 +7,13 @@
 from auto_derby import plugin
 
 from . import ocr, single_mode, template, window
+from .clients import ADBClient
 
 
 class config:
     LOG_PATH = os.getenv("AUTO_DERBY_LOG_PATH", "auto_derby.log")
     PLUGINS = tuple(i for i in os.getenv("AUTO_DERBY_PLUGINS", "").split(",") if i)
+    ADB_ADDRESS = os.getenv("AUTO_DERBY_ADB_ADDRESS", "")
 
     single_mode_race_data_path = os.getenv(
         "AUTO_DERBY_SINGLE_MODE_RACE_DATA_PATH", "single_mode_races.json"
@@ -33,9 +35,11 @@ class config:
     use_legacy_screenshot = (
         os.getenv("AUTO_DERBY_USE_LEGACY_SCREENSHOT", "").lower() == "true"
     )
+    adb_key_path = os.getenv("AUTO_DERBY_ADB_KEY_PATH", ADBClient.key_path)
 
     @classmethod
     def apply(cls) -> None:
+        ADBClient.key_path = cls.adb_key_path
         ocr.g.data_path = cls.ocr_data_path
         ocr.g.image_path = cls.ocr_image_path
         plugin.g.path = cls.plugin_path

auto_derby/action.py

@@ -16,6 +16,7 @@ def resize_proxy() -> mathtools.ResizeProxy:
 
 def click(point: Tuple[int, int]):
     clients.current().click(point)
+    template.invalidate_screeshot()
 
 
 def count_image(*tmpl: Union[Text, template.Specification]) -> int:
@@ -87,11 +88,3 @@ def wheel(point: Tuple[int, int], delta: int) -> None:
 def drag(point: Tuple[int, int], *, dx: int = 0, dy: int = 0, duration: float = 0.03):
     clients.current().drag(point, dx=dx, dy=dy, duration=duration)
     template.invalidate_screeshot()
-
-
-def drag_through(
-    *points: Tuple[int, int], duration: float = 0.02
-) -> Iterator[Tuple[int, int]]:
-    for i in clients.current().drag_through(*points, duration=duration):
-        template.invalidate_screeshot()
-        yield i

auto_derby/clients/__init__.py

@@ -1,2 +1,3 @@
 from .client import Client, set_current, current
 from .dmm import DMMClient
+from .adb import ADBClient

auto_derby/clients/adb.py

@@ -0,0 +1,116 @@
+# -*- coding=UTF-8 -*-
+# pyright: strict
+
+from __future__ import annotations
+
+
+from pathlib import Path
+from typing import Text, Tuple
+
+import PIL.Image
+from adb_shell.adb_device import AdbDeviceTcp
+from adb_shell.auth.keygen import keygen
+from adb_shell.auth.sign_pythonrsa import PythonRSASigner
+
+from .client import Client
+
+import re
+
+import logging
+import time
+
+LOGGER = logging.getLogger(__name__)
+
+
+class ADBClient(Client):
+    key_path: Text = "adb.local.key"
+
+    def __init__(self, address: Text):
+        hostname, port = address.split(":", 2)
+        assert hostname, "invalid address: missing hostname: %s" % address
+        assert port, "invalid port: missing port: %s" % address
+
+        self.hostname = hostname
+        self.port = int(port)
+        self.device = AdbDeviceTcp(self.hostname, self.port)
+        self._height, self._width = 0, 0
+
+    @property
+    def width(self) -> int:
+        return self._width
+
+    @property
+    def height(self) -> int:
+        return self._height
+
+    def connect(self):
+        if not Path(ADBClient.key_path).exists():
+            keygen(self.key_path)
+        signer = PythonRSASigner.FromRSAKeyPath(self.key_path)
+        self.device.connect(rsa_keys=[signer])
+
+    def click(self, point: Tuple[int, int]) -> None:
+        x, y = point
+        res = self.device.shell(f"input tap {x} {y}")
+        assert not res, res
+        time.sleep(0.5)
+
+    def start_game(self):
+        self.device.shell(
+            "am start -n jp.co.cygames.umamusume/jp.co.cygames.umamusume_activity.UmamusumeActivity"
+        )
+
+    def load_size(self):
+        res = self.device.shell("wm size")
+        match = re.match(r"Physical size: (\d+)x(\d+)", res)
+        assert match, "unexpected command result: %s" % res
+        self._width = int(match.group(2))
+        self._height = int(match.group(1))
+        LOGGER.debug("screen size: width=%d height=%d", self.width, self.height)
+
+    def setup(self) -> None:
+        self.connect()
+        self.load_size()
+        self.start_game()
+
+    def screenshot(self) -> PIL.Image.Image:
+        # img_data = self.device.shell(
+        #     f"screencap -p",
+        #     decode=False,
+        #     transport_timeout_s=None,
+        # )
+        # img = PIL.Image.open(io.BytesIO(img_data))
+
+        # TODO: compare with png format screenshot
+        # https://stackoverflow.com/a/59470924
+        img_data = self.device.shell(
+            f"screencap",
+            decode=False,
+            transport_timeout_s=None,
+        )
+        width = int.from_bytes(img_data[0:4], "little")
+        height = int.from_bytes(img_data[4:8], "little")
+        pixel_format = int.from_bytes(img_data[8:12], "little")
+        # https://developer.android.com/reference/android/graphics/PixelFormat#RGBA_8888
+        assert pixel_format == 1, "unsupported pixel format: %s" % pixel_format
+        img = PIL.Image.frombuffer(
+            "RGBA", (width, height), img_data[12:], "raw", "RGBX", 0, 1
+        ).convert("RGBA")
+        return img
+
+    def drag(
+        self, point: Tuple[int, int], *, dx: int, dy: int, duration: float
+    ) -> None:
+        x1, y1 = point
+        x2, y2 = x1 + dx, y1 + dy
+        duration_ms = int(duration * 1e3)
+        duration_ms = max(100, duration_ms)  # drag not work if too fast
+        res = self.device.shell(
+            f"input swipe {x1} {y1} {x2} {y2} {duration_ms}",
+            read_timeout_s=10 + duration,
+        )
+        assert not res, res
+        time.sleep(0.5)
+
+    def wheel(self, point: Tuple[int, int], delta: int) -> None:
+        self.drag(point, dx=0, dy=round(delta * self.height * 0.05), duration=0.2)

auto_derby/clients/client.py

@@ -2,7 +2,7 @@
 # pyright: strict
 from __future__ import annotations
 from abc import abstractmethod, ABC
-from typing import Dict, Iterator, Literal, Tuple
+from typing import Dict, Literal, Tuple
 import PIL.Image
 
 
@@ -24,22 +24,18 @@ def setup(self) -> None:
     def screenshot(self) -> PIL.Image.Image:
         ...
 
+    # TODO: rename to tap
     @abstractmethod
     def click(self, point: Tuple[int, int]) -> None:
         ...
 
+    # TODO: refactor wheel and drag to swipe
     @abstractmethod
     def drag(
         self, point: Tuple[int, int], *, dx: int, dy: int, duration: float = 1
     ) -> None:
         ...
 
-    @abstractmethod
-    def drag_through(
-        self, *points: Tuple[int, int], duration: float = 0.02
-    ) -> Iterator[Tuple[int, int]]:
-        ...
-
     @abstractmethod
     def wheel(self, point: Tuple[int, int], delta: int) -> None:
         ...

auto_derby/clients/dmm.py

@@ -4,7 +4,7 @@
 
 import logging
 from ctypes import windll
-from typing import Dict, Iterator, Literal, Optional, Tuple
+from typing import Dict, Literal, Optional, Tuple
 
 import PIL.Image
 import PIL.ImageGrab
@@ -85,11 +85,6 @@ def drag(
     ) -> None:
         window.drag_at(self.h_wnd, point, dx=dx, dy=dy, duration=duration)
 
-    def drag_through(
-        self, *points: Tuple[int, int], duration: float = 0.02
-    ) -> Iterator[Tuple[int, int]]:
-        return window.drag_through_at(self.h_wnd, *points, duration=duration)
-
     def wheel(self, point: Tuple[int, int], delta: int) -> None:
         with window.recover_cursor(), window.recover_foreground(), window.topmost(
             self.h_wnd

auto_derby/jobs/nurturing.py

@@ -45,33 +45,29 @@ def _choose_race(ctx: Context, race1: race.Race) -> None:
         action.click(rp.vector2((100, 600), 466))
         if _current_race(ctx) == race1:
             return
-        for _ in action.drag_through(
+        action.wheel(
             rp.vector2((100, 600), 466),
-            rp.vector2((100, 500), 466),
-            rp.vector2((100, 490), 466),
-            rp.vector2((100, 490), 466),
-            duration=0.2,
-        ):
-            pass
+            -1,
+        )
 
 
 def _handle_training(ctx: Context) -> None:
     rp = action.resize_proxy()
     trainings: List[Training] = []
 
     action.wait_image(_TRAINING_CONFIRM)
+    dy = rp.vector(100, 466)
     for x, y in (
         rp.vector2((78, 700), 466),
         rp.vector2((158, 700), 466),
         rp.vector2((234, 700), 466),
         rp.vector2((314, 700), 466),
         rp.vector2((402, 700), 466),
     ):
-        action.drag((x, y - 100), dy=100)
+        action.drag((x, y - dy), dy=dy)
         action.wait_image(_TRAINING_CONFIRM)
         t = Training.from_training_scene(template.screenshot())
         trainings.append(t)
-
     expected_score = 15 + ctx.turn_count() * 10 / 24
     races_with_score = sorted(
         ((i, i.score(ctx)) for i in race.find(ctx)),

auto_derby/launcher/launcher.cs

@@ -137,6 +137,19 @@ public string Plugins
             }
         }
 
+        public string ADBAddress
+        {
+            get
+            {
+                return (string)key.GetValue("ADBAddress", "");
+            }
+            set
+            {
+                key.SetValue("ADBAddress", value);
+                OnPropertyChanged("ADBAddress");
+            }
+        }
+
         public bool Debug
         {
             get