feat(task): 实现自动竞赛任务

kotonebot/__init__.py

@@ -1,4 +1,5 @@
 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, cropped
+from .backend.util import Rect, fuzz, regex, contains, grayscaled, grayscale_cached, cropper, x, y, cropped, UnrecoverableError
 from .backend.core import task, action
+from .ui import user

kotonebot/backend/util.py

@@ -15,6 +15,9 @@ class TaskInfo(NamedTuple):
     description: str
     entry: Callable[[], None]
 
+class UnrecoverableError(Exception):
+    pass
+
 Rect = typing.Sequence[int]
 """左上X, 左上Y, 宽度, 高度"""
 

kotonebot/tasks/acquire_activity_funds.py

@@ -14,7 +14,7 @@ def acquire_activity_funds():
     if image.find(R.Daily.TextActivityFundsMax):
         logger.info('Activity funds maxed out.')
         device.click()
-        device.click(image.expect_wait(R.InPurodyuusu.ButtonClose, timeout=2))
+        device.click(image.expect_wait(R.Common.ButtonClose, timeout=2))
         logger.info('Activity funds acquired.')
     else:
         logger.info('Activity funds not maxed out. No action needed.')

kotonebot/tasks/actions/in_purodyuusu.py

@@ -441,7 +441,7 @@ def exam():
         sleep(9) # TODO: 采用更好的方式检测练习结束
     
     # 点击“次へ”
-    device.click(image.expect_wait(R.InPurodyuusu.NextBtn))
+    device.click(image.expect_wait(R.Common.ButtonNext))
     while ocr.wait_for(contains("メモリー"), timeout=7):
         device.click_center()
 
@@ -490,7 +490,7 @@ def produce_end():
                 continue
             device.click()
         # 记忆封面保存失败提示
-        elif image.find(R.InPurodyuusu.ButtonClose):
+        elif image.find(R.Common.ButtonClose):
             logger.info("Memory cover save failed. Click to close.")
             device.click()
         # 结算完毕

kotonebot/tasks/actions/loading.py

@@ -1,41 +1,47 @@
 import time
 from time import sleep
+from logging import getLogger
 
 import cv2
 import numpy as np
 
 from kotonebot import image, device, debug
+from kotonebot.backend.debug import result
+
+logger = getLogger(__name__)
 
 def loading() -> bool:
     """检测是否在场景加载页面"""
     img = device.screenshot()
     # 二值化图片
     _, img = cv2.threshold(img, 127, 255, cv2.THRESH_BINARY)
-    debug.show(img)
     # 裁剪上面 10%
     img = img[:int(img.shape[0] * 0.1), :]
-    debug.show(img)
     # 判断图片中颜色数量是否 <= 2
     # https://stackoverflow.com/questions/56606294/count-number-of-unique-colours-in-image
     b,g,r = cv2.split(img)
     shiftet_im = b.astype(np.int64) + 1000 * (g.astype(np.int64) + 1) + 1000 * 1000 * (r.astype(np.int64) + 1)
-    return len(np.unique(shiftet_im)) <= 2
+    ret = len(np.unique(shiftet_im)) <= 2
+    result('tasks.actions.loading', img, f'result={ret}')
+    return ret
 
-def wait_loading_start(timeout: float = 10):
+def wait_loading_start(timeout: float = 60):
     """等待加载开始"""
     start_time = time.time()
     while not loading():
         if time.time() - start_time > timeout:
             raise TimeoutError('加载超时')
-        sleep(0.5)
+        logger.debug('Not loading...')
+        sleep(1)
 
-def wait_loading_end(timeout: float = 10):
+def wait_loading_end(timeout: float = 60):
     """等待加载结束"""
     start_time = time.time()
     while loading():
         if time.time() - start_time > timeout:
             raise TimeoutError('加载超时')
-        sleep(0.5)
+        logger.debug('Loading...')
+        sleep(1)
 
 if __name__ == '__main__':
     print(loading())

kotonebot/tasks/actions/scenes.py

@@ -1,9 +1,11 @@
 import logging
 import time
+from time import sleep
 from typing import Callable
 
 from .. import R
-from kotonebot import device, image, ocr, action, cropper, x, y
+from .loading import loading
+from kotonebot import device, image, action, cropped, UnrecoverableError
 
 logger = logging.getLogger(__name__)
 
@@ -32,18 +34,38 @@ def until(
 
 @action
 def at_home() -> bool:
-    with device.hook(cropper(y(from_=0.7))):
+    with cropped(device, y1=0.7):
         return image.find(R.Daily.ButtonHomeCurrent) is not None
 
 @action
 def at_shop() -> bool:
-    with device.hook(cropper(y(to=0.3))):
+    with cropped(device, y2=0.3):
         return image.find(R.Daily.IconShopTitle) is not None
 
 @action
 def goto_home():
+    """
+    从其他场景返回首页。
+
+    前置条件：无 \n
+    结束状态：位于首页
+    """
     logger.info("Going home.")
-    device.click(image.expect(R.Common.ButtonToolbarHome, transparent=True, threshold=0.9999, colored=True))
+    with cropped(device, y1=0.7):
+        if image.find(
+            R.Common.ButtonToolbarHome,
+            transparent=True,
+            threshold=0.9999,
+            colored=True
+        ):
+            device.click()
+            while loading():
+                sleep(0.5)
+        elif image.find(R.Common.ButtonHome):
+            device.click()
+        else:
+            raise UnrecoverableError("Failed to go home.")
+    image.expect_wait(R.Daily.ButtonHomeCurrent, timeout=20)
 
 @action
 def goto_shop():

kotonebot/tasks/contest.py

@@ -0,0 +1,117 @@
+"""竞赛"""
+import logging
+from time import sleep
+from gettext import gettext as _
+
+from . import R
+from .actions.scenes import at_home, goto_home
+from .actions.loading import wait_loading_end
+from kotonebot import device, image, ocr, action, task, user
+
+logger = logging.getLogger(__name__)
+
+@action
+def goto_contest():
+    """
+    前置条件：位于首页 \n
+    结束状态：位于竞赛界面，且已经点击了各种奖励领取提示
+    """
+    device.click(image.expect(R.Common.ButtonContest))
+    device.click(image.expect_wait(R.Daily.TextContest, colored=True, transparent=True, threshold=0.9999))
+    sleep(0.5)
+    wait_loading_end()
+    while not image.find(R.Daily.ButtonContestRanking):
+        # [screenshots/contest/acquire1.png]
+        # [screenshots/contest/acquire2.png]
+        device.click_center()
+        sleep(1)
+    # [screenshots/contest/main.png]
+
+@action
+def pick_and_contest() -> bool:
+    """
+    选择并挑战
+
+    前置条件：位于竞赛界面 \n
+    结束状态：位于竞赛界面
+
+    :return: 如果返回假，说明今天挑战次数已经用完了
+    """
+    image.expect_wait(R.Daily.ButtonContestRanking)
+    sleep(1) # 等待动画
+    logger.info('Randomly pick a contestant and start challenge.')
+    # 随机选一个对手 [screenshots/contest/main.png]
+    logger.debug('Clicking on contestant.')
+    contestant = image.wait_for(R.Daily.TextContestOverallStats, timeout=2)
+    if contestant is None:
+        logger.info('No contestant found. Today\'s challenge points used up.')
+        return False
+    device.click(contestant)
+    # 挑战开始 [screenshots/contest/start1.png]
+    logger.debug('Clicking on start button.')
+    device.click(image.expect_wait(R.Daily.ButtonContestStart))
+    sleep(1)
+    # 记忆未编成 [screenshots/contest/no_memo.png]
+    if image.find(R.Daily.TextContestNoMemory):
+        logger.debug('Memory not set. Using auto-compilation.')
+        user.warning(_('记忆未编成。将使用自动编成。'), once=True)
+        device.click(image.expect(R.Daily.ButtonContestChallenge))
+    # 进入挑战页面 [screenshots/contest/contest1.png]
+    # [screenshots/contest/contest2.png]
+    image.expect_wait(R.Daily.ButtonContestChallengeStart)
+    # 勾选跳过所有
+    if image.find(R.Common.CheckboxUnchecked):
+        logger.debug('Checking skip all.')
+        device.click()
+        sleep(0.5)
+    # 点击 SKIP
+    logger.debug('Clicking on SKIP.')
+    device.click(image.expect(R.Daily.ButtonIconSkip, colored=True, transparent=True, threshold=0.999))
+    while not image.wait_for(R.Common.ButtonNextNoIcon, timeout=2):
+        device.click_center()
+        logger.debug('Waiting for the result.')
+    # [screenshots/contest/after_contest1.png]
+    # 点击 次へ [screenshots/contest/after_contest2.png]
+    logger.debug('Challenge finished. Clicking on next.')
+    device.click()
+    # 点击 終了 [screenshots/contest/after_contest3.png]
+    logger.debug('Clicking on end.')
+    device.click(image.expect_wait(R.Common.ButtonEnd))
+    # 可能出现的奖励弹窗 [screenshots/contest/after_contest4.png]
+    sleep(1)
+    if image.find(R.Common.ButtonClose):
+        logger.debug('Clicking on close.')
+        device.click()
+    # 等待返回竞赛界面
+    wait_loading_end()
+    image.expect_wait(R.Daily.ButtonContestRanking)
+    logger.info('Challenge finished.')
+    return True
+
+@task('竞赛')
+def contest():
+    """"""
+    logger.info('Contest started.')
+    if not at_home():
+        goto_home()
+    goto_contest()
+    while pick_and_contest():
+        sleep(1.3)
+    goto_home()
+    logger.info('Contest all finished.')
+    
+
+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')
+    logging.getLogger('kotonebot').setLevel(logging.DEBUG)
+    logger.setLevel(logging.DEBUG)
+    init_context()
+    
+    # if image.find(R.Common.CheckboxUnchecked):
+    #     logger.debug('Checking skip all.')
+    #     device.click()
+    #     sleep(0.5)
+    # device.click(image.expect(R.Daily.ButtonIconSkip, colored=True, transparent=True, threshold=0.999))
+    contest()

kotonebot/ui/user.py

@@ -0,0 +1,24 @@
+"""消息框、通知、推送等 UI 相关函数"""
+
+def ask(
+    question: str,
+    options: list[str],
+    *,
+    timeout: float = -1,
+) -> bool:
+    """
+    询问用户
+    """
+    raise NotImplementedError
+
+def warning(
+    message: str,
+    once: bool = False
+):
+    """
+    警告信息。
+
+    :param message: 消息内容
+    :param once: 每次运行是否只显示一次。
+    """
+    pass