feat(task): 启动游戏任务 & 收取活动费任务

kotonebot/__init__.py

@@ -1,4 +1,4 @@
 from .client.protocol import DeviceABC
 from .backend.context import ContextOcr, ContextImage, ContextDebug, device, ocr, image, debug
-from .backend.util import Rect, fuzz, regex, contains, grayscaled, grayscale_cached, cropper, x, y
+from .backend.util import Rect, fuzz, regex, contains, grayscaled, grayscale_cached, cropper, x, y, cropped
 from .backend.core import task, action

kotonebot/backend/context.py

@@ -137,7 +137,8 @@ class ContextImage:
             mask: MatLike | str | None = None,
             threshold: float = 0.9,
             timeout: float = 10,
-            colored: bool = False
+            colored: bool = False,
+            interval: float = 0.1
         ) -> TemplateMatchResult | None:
         """
         等待指定图像出现。
@@ -150,7 +151,7 @@ class ContextImage:
                 return ret
             if time.time() - start_time > timeout:
                 return None
-            time.sleep(0.1)
+            time.sleep(interval)
 
     def wait_for_any(
             self,

kotonebot/backend/core.py

@@ -1,4 +1,7 @@
-from typing import Callable, overload, Any
+from typing import Callable, ParamSpec, TypeVar, overload, Any
+
+P = ParamSpec('P')
+R = TypeVar('R')
 
 def task(
     name: str,
@@ -10,15 +13,15 @@ def task(
     :param name: 任务名称
     :param description: 任务描述。如果为 None，则使用函数的 docstring 作为描述。
     """
-    def _task_decorator(func: Callable):
+    def _task_decorator(func: Callable[P, R]) -> Callable[P, R]:
         return func
     return _task_decorator
 
 @overload
-def action(func: Callable[..., Any]) -> Callable[..., Any]: ...
+def action(func: Callable[P, R]) -> Callable[P, R]: ...
 
 @overload
-def action(name: str, description: str|None = None) -> Callable[..., Any]:
+def action(name: str, description: str|None = None) -> Callable[[Callable[P, R]], Callable[P, R]]:
     """
     `action` 装饰器，用于标记一个函数为动作函数。
 

kotonebot/backend/image.py

@@ -7,6 +7,7 @@ from .debug import result, debug, img
 import cv2
 import numpy as np
 from cv2.typing import MatLike, Rect, Point, Size
+from skimage.metrics import structural_similarity
 
 logger = getLogger(__name__)
 
@@ -477,3 +478,14 @@ def expect(
     else:
         return ret
 
+def similar(
+    image1: MatLike,
+    image2: MatLike,
+    threshold: float = 0.8
+) -> bool:
+    """
+    判断两张图像是否相似。输入的两张图片必须为相同尺寸。
+    """
+    if image1.shape != image2.shape:
+        raise ValueError('Expected two images with the same size.')
+    return structural_similarity(image1, image2, multichannel=True) >= threshold

kotonebot/client/device/adb.py

@@ -84,6 +84,13 @@ class AdbDevice(DeviceABC):
     @staticmethod
     def list_devices() -> list[str]:
         raise NotImplementedError
+    
+
+    def start_app(self, package_name: str) -> None:
+        self.device.shell(f"monkey -p {package_name} 1")
+    
+    def current_package(self) -> str:
+        raise NotImplementedError
 
         
 if __name__ == "__main__":

kotonebot/client/protocol.py

@@ -205,3 +205,14 @@ class DeviceABC(ABC):
         """
         ...
 
+    def current_package(self) -> str:
+        """
+        前台 APP 的包名
+        """
+        ...
+
+    def start_app(self, package_name: str) -> None:
+        """
+        启动某个 APP
+        """
+        ...

kotonebot/tasks/acquire_activity_funds.py

@@ -0,0 +1,29 @@
+"""收取活动费"""
+import logging
+
+from kotonebot import task, device, image, cropped
+from .actions.scenes import at_home, goto_home
+from . import R
+
+logger = logging.getLogger(__name__)
+
+@task('收取活动费')
+def acquire_activity_funds():
+    if not at_home():
+        goto_home()
+    if image.find(R.Daily.TextActivityFundsMax):
+        logger.info('Activity funds maxed out.')
+        device.click()
+        device.click(image.expect_wait(R.InPurodyuusu.ButtonClose, timeout=2))
+        logger.info('Activity funds acquired.')
+    else:
+        logger.info('Activity funds not maxed out. No action needed.')
+
+
+if __name__ == '__main__':
+    from kotonebot.backend.context import init_context
+    import logging
+    logging.basicConfig(level=logging.INFO, format='[%(asctime)s] [%(levelname)s] [%(name)s] [%(funcName)s] [%(lineno)d] %(message)s')
+    logger.setLevel(logging.DEBUG)
+    init_context()
+    acquire_activity_funds()

kotonebot/tasks/actions/scenes.py

@@ -38,11 +38,15 @@ def at_shop() -> bool:
     with device.hook(cropper(y(to=0.3))):
         return image.find(R.Daily.IconShopTitle) is not None
 
+@action
+def goto_home():
+    pass
+
 @action
 def goto_shop():
     """
     从首页进入 ショップ。
-    
+
     前置条件：位于首页 \n
     结束状态：位于商店页面
     """

kotonebot/tasks/start_game.py

@@ -0,0 +1,31 @@
+"""启动游戏，领取登录奖励，直到首页为止"""
+import logging
+from time import sleep
+from kotonebot import task, device, image, cropped
+from . import R
+
+logger = logging.getLogger(__name__)
+
+@task('启动游戏')
+def start_game():
+    """
+    启动游戏，直到游戏进入首页为止。
+    
+    执行前游戏必须处于未启动状态。
+    """
+    device.start_app('com.bandainamcoent.idolmaster_gakuen') # TODO: 包名放到配置文件里
+    # [screenshots/startup/1.png]
+    image.wait_for(R.Daily.ButonLinkData, timeout=30)
+    sleep(2)
+    device.click_center()
+    while not image.wait_for(R.Daily.ButtonHomeCurrent, timeout=3):
+        device.click_center()
+
+if __name__ == '__main__':
+    from kotonebot.backend.context import init_context
+    import logging
+    logging.basicConfig(level=logging.INFO, format='[%(asctime)s] [%(levelname)s] [%(name)s] [%(funcName)s] [%(lineno)d] %(message)s')
+    logger.setLevel(logging.DEBUG)
+    init_context()
+    start_game()
+