feat(core): 新增目标截图间隔功能

可以通过设置目标截图间隔来限制截图速度,间接限制脚本运行速度。
This commit is contained in:
XcantloadX 2025-07-05 22:14:13 +08:00
parent 0e183b0ca6
commit c8fbf80640
5 changed files with 71 additions and 6 deletions

View File

@ -145,6 +145,18 @@ class KotoneBot:
"""
raise NotImplementedError('Implement `_create_device` before using Kotonebot.')
def _on_init_context(self) -> None:
"""
初始化 Context 的钩子方法子类可以重写此方法来自定义初始化逻辑
默认实现调用 init_context 而不传入 target_screenshot_interval
"""
d = self._on_create_device()
init_context(
config_path=self.config_path,
config_type=self.config_type,
target_device=d
)
def _on_after_init_context(self):
"""
抽象方法 init_context() 被调用后立即执行
@ -155,8 +167,7 @@ class KotoneBot:
"""
按优先级顺序运行所有任务
"""
d = self._on_create_device()
init_context(config_path=self.config_path, config_type=self.config_type, target_device=d)
self._on_init_context()
self._on_after_init_context()
vars.flow.clear_interrupt()

View File

@ -27,6 +27,7 @@ from cv2.typing import MatLike
from kotonebot.client.device import Device, AndroidDevice, WindowsDevice
from kotonebot.backend.flow_controller import FlowController
from kotonebot.util import Interval
import kotonebot.backend.image as raw_image
from kotonebot.backend.image import (
TemplateMatchResult,
@ -718,8 +719,20 @@ class Forwarded:
T_Device = TypeVar('T_Device', bound=Device)
class ContextDevice(Generic[T_Device], Device):
def __init__(self, device: T_Device):
def __init__(self, device: T_Device, target_screenshot_interval: float | None = None):
"""
:param device: 目标设备
:param target_screenshot_interval: `ContextDevice.target_screenshot_interval`
"""
self._device = device
self.target_screenshot_interval: float | None = target_screenshot_interval
"""
目标截图间隔可用于限制截图速度若两次截图实际间隔小于该值则会自动等待
None 时不限制截图速度
"""
self._screenshot_interval: Interval | None = None
if self.target_screenshot_interval is not None:
self._screenshot_interval = Interval(self.target_screenshot_interval)
def screenshot(self, *, force: bool = False):
"""
@ -734,6 +747,9 @@ class ContextDevice(Generic[T_Device], Device):
img = current._inherit_screenshot
current._inherit_screenshot = None
else:
if self._screenshot_interval is not None:
self._screenshot_interval.wait()
if next_wait == 'screenshot':
delta = time.time() - last_screenshot_time
if delta < next_wait_time:
@ -780,7 +796,8 @@ class Context(Generic[T]):
self,
config_path: str,
config_type: Type[T],
device: Device
device: Device,
target_screenshot_interval: float | None = None
):
self.__ocr = ContextOcr(self)
self.__image = ContextImage(self)
@ -788,7 +805,7 @@ class Context(Generic[T]):
self.__vars = ContextGlobalVars()
self.__debug = ContextDebug(self)
self.__config = ContextConfig[T](self, config_path, config_type)
self.__device = ContextDevice(device)
self.__device = ContextDevice(device, target_screenshot_interval)
def inject(
self,
@ -900,6 +917,7 @@ def init_context(
config_type: Type[T] = dict[str, Any],
force: bool = False,
target_device: Device,
target_screenshot_interval: float | None = None,
):
"""
初始化 Context 模块
@ -911,6 +929,7 @@ def init_context(
:param force: 是否强制重新初始化
若为 `True`则忽略已存在的 Context 实例并重新创建一个新的实例
:param target_device: 目标设备
:param target_screenshot_interval: `ContextDevice.target_screenshot_interval`
"""
global _c, device, ocr, image, color, vars, debug, config
if _c is not None and not force:
@ -919,6 +938,7 @@ def init_context(
config_path=config_path,
config_type=config_type,
device=target_device,
target_screenshot_interval=target_screenshot_interval,
)
device._FORWARD_getter = lambda: _c.device # type: ignore
ocr._FORWARD_getter = lambda: _c.ocr # type: ignore

View File

@ -50,6 +50,8 @@ class BackendConfig(ConfigBaseModel):
"""Windows 截图方式的 AutoHotkey 可执行文件路径,为 None 时使用默认路径"""
mumu_background_mode: bool = False
"""MuMu12 模拟器后台保活模式"""
target_screenshot_interval: float | None = None
"""最小截图间隔,单位为秒。为 None 时不限制截图速度。"""
class PushConfig(ConfigBaseModel):
"""推送配置。"""

View File

@ -37,7 +37,7 @@ ConfigKey = Literal[
'check_emulator', 'emulator_path',
'adb_emulator_name', 'emulator_args',
'_mumu_index', '_leidian_index',
'mumu_background_mode',
'mumu_background_mode', 'target_screenshot_interval',
# purchase
'purchase_enabled',
@ -761,6 +761,15 @@ class KotoneBotUI:
interactive=True
)
target_screenshot_interval = gr.Number(
label="最小截图间隔(秒)",
value=self.current_config.backend.target_screenshot_interval,
info=BackendConfig.model_fields['target_screenshot_interval'].description,
minimum=0,
step=0.1,
interactive=True
)
tab_mumu12.select(fn=partial(_update_emulator_tab_options, selected_index=0), inputs=[screenshot_impl], outputs=[screenshot_impl])
tab_leidian.select(fn=partial(_update_emulator_tab_options, selected_index=1), inputs=[screenshot_impl], outputs=[screenshot_impl])
tab_custom.select(fn=partial(_update_emulator_tab_options, selected_index=2), inputs=[screenshot_impl], outputs=[screenshot_impl])
@ -814,12 +823,14 @@ class KotoneBotUI:
# Common settings for all backend types
self.current_config.backend.screenshot_impl = data['screenshot_method']
self.current_config.backend.target_screenshot_interval = data['target_screenshot_interval']
self.current_config.keep_screenshots = data['keep_screenshots'] # This is a UserConfig field
return set_config, {
'adb_ip': adb_ip,
'adb_port': adb_port,
'screenshot_method': screenshot_impl, # screenshot_impl is the component
'target_screenshot_interval': target_screenshot_interval,
'keep_screenshots': keep_screenshots,
'check_emulator': check_emulator,
'emulator_path': emulator_path,

View File

@ -110,6 +110,27 @@ class Kaa(KotoneBot):
logger.exception('Failed to save error report:')
return ''
@override
def _on_init_context(self) -> None:
"""
初始化 Context从配置中读取 target_screenshot_interval
"""
from kotonebot.config.manager import load_config
from kotonebot.backend.context import init_context
# 加载配置以获取 target_screenshot_interval
config = load_config(self.config_path, type=self.config_type)
user_config = config.user_configs[0] # HACK: 硬编码
target_screenshot_interval = user_config.backend.target_screenshot_interval
d = self._on_create_device()
init_context(
config_path=self.config_path,
config_type=self.config_type,
target_device=d,
target_screenshot_interval=target_screenshot_interval
)
@override
def _on_after_init_context(self):
if self.backend_instance is None: