mower-ng/mower/utils/device/method/mumumanager.py

import json
import time
from datetime import datetime
from functools import wraps
from typing import Literal

from mower.utils import config
from mower.utils.csleep import MowerExit, csleep
from mower.utils.device.emulator import restart_emulator
from mower.utils.log import logger

from .utils import subprocess_run


def retry_mumumanager(func):
    @wraps(func)
    def retry_wrapper(self, *args, **kwargs):
        for _ in range(4):
            try:
                return func(self, *args, **kwargs)
            except MowerExit:
                raise
            except ValueError as e:
                logger.debug_exception(e)
                if _ < 3:
                    time.sleep(10)
                else:
                    restart_emulator()
            except Exception as e:
                logger.debug_exception(e)

    return retry_wrapper


class MuMuManager:
    @property
    def manager_path(self) -> str:
        return config.conf.emulator.emulator_folder + "\\MuMuManager.exe"

    @property
    def index(self) -> str:
        return config.conf.emulator.index

    def __init__(self):
        self.disable_log()

    def disable_log(self):
        cmd = [self.manager_path, "log", "off"]
        subprocess_run(cmd)

    def load_json(self, data: str) -> dict:
        try:
            return json.loads(data)
        except json.JSONDecodeError as e:
            logger.error(f"JSON 解析失败: {e}")
            return {}

    def get_emulator_info(self):
        cmd = [self.manager_path, "info", "-v", str(self.index)]
        output = subprocess_run(cmd)
        return self.load_json(output)

    @retry_mumumanager
    def get_field_value(self, data, field_name):
        """
        从模拟器信息中提取指定字段的值
        Args:
            field_name (str): 字段名称

        Returns:
            str, int, bool, or None: 字段: 字段值或错误消息
        """
        # 获取模拟器信息
        if not data:
            logger.error("未能获取数据")
            return

        # 提取指定字段的值
        value = data.get(field_name, None)
        if value is None:
            raise ValueError(f"字段 '{field_name}' 不存在")

        return value

    def emulator_status(self) -> Literal["running", "launching", "stopped"]:
        data = self.get_emulator_info()
        android = self.get_field_value(data, "is_android_started")
        process = self.get_field_value(data, "is_process_started")
        if android:
            return "running"
        if process:
            return "launching"
        return "stopped"

    def get_setting_info(self):
        cmd = [self.manager_path, "setting", "-v", str(self.index), "-a"]
        output = subprocess_run(cmd)
        return self.load_json(output)

    def app_kept_alive(self) -> bool:
        return self.get_field_value(self.get_setting_info(), "app_keptlive")

    def emulator_version(self) -> list[int]:
        version = self.get_field_value(self.get_setting_info(), "core_version")
        logger.debug(f"MuMu12模拟器版本: {version}")
        parts = [int(x) for x in version.split(".")[:3]]  # 4.1.31.3724取出[4, 1, 31]
        return parts

    def set_app_kept_alive_false(self) -> None:
        cmd = [
            self.manager_path,
            "setting",
            "-v",
            str(self.index),
            "-k",
            "app_keptlive",
            "-val",
            "false",
        ]
        subprocess_run(cmd)

    def launch(self) -> None:
        cmd = [
            self.manager_path,
            "control",
            "-v",
            str(self.index),
            "app",
            "launch",
            "-pkg",
            config.conf.APPNAME,
        ]
        subprocess_run(cmd)

    def exit(self) -> None:
        cmd = [
            self.manager_path,
            "control",
            "-v",
            str(self.index),
            "app",
            "close",
            "-pkg",
            config.conf.APPNAME,
        ]
        subprocess_run(cmd)

    def home(self) -> None:
        cmd = [
            self.manager_path,
            "control",
            "-v",
            str(self.index),
            "tool",
            "func",
            "-n",
            "go_home",
        ]
        subprocess_run(cmd)

    def current_focus(self) -> str:
        cmd = [self.manager_path, "control", "-v", str(self.index), "app", "info", "-i"]
        output = subprocess_run(cmd)
        data = self.load_json(output)
        return self.get_field_value(data, "active")

    def check_current_focus(self) -> bool:
        """check if the application is in the foreground"""
        update = False
        start_time = datetime.now()
        while True:
            try:
                focus = self.current_focus()
                if focus != config.conf.APPNAME:
                    if (datetime.now() - start_time).total_seconds() > 40:
                        self.exit()  # 应用卡死
                        start_time = datetime.now()
                    self.launch()
                    update = True
                    csleep(1)
                    continue
                return update
            except MowerExit:
                raise
            except Exception as e:
                logger.exception(e)
                update = True

    def check_device_screen(self) -> bool:
        data = self.get_setting_info()
        width = int(float(self.get_field_value(data, "resolution_width.custom")))
        height = int(float(self.get_field_value(data, "resolution_height.custom")))
        dpi = int(float(self.get_field_value(data, "resolution_dpi.custom")))
        if (
            config.conf.screencap_strategy.startswith("droidcast")
            and self.app_kept_alive()
        ):
            logger.error("MuMu12的droidcast截图策略下，需要关闭应用后台保活")
            try:
                self.set_app_kept_alive_false()
                logger.info("关闭应用后台保活成功")
                restart_emulator()
            except Exception as e:
                logger.debug_exception(e)
                raise MowerExit

        if width == 1920 and height == 1080 and dpi == 280:
            return True
        logger.error("mower-ng仅支持1920x1080分辨率、280DPI，请修改模拟器设置")
        return False

    def kill_server(self) -> None:
        pass

    def reset_when_exit(self) -> None:
        pass

    def launch_emulator(self) -> None:
        cmd = [self.manager_path, "control", "-v", str(self.index), "launch"]
        subprocess_run(cmd)

    def shutdown_emulator(self) -> None:
        cmd = [self.manager_path, "control", "-v", str(self.index), "shutdown"]
        subprocess_run(cmd)

    def restart_emulator(self, stop: bool = True, start: bool = True) -> None:
        if stop and self.emulator_status() == "running":
            logger.info("关闭MuMu12模拟器")
            self.shutdown_emulator()
            csleep(3)
        if start:
            start_time = datetime.now()
            while (
                datetime.now() - start_time
            ).total_seconds() < config.conf.emulator.wait_time:
                status = self.emulator_status()
                if status == "stopped":
                    logger.info("启动MuMu12模拟器")
                    self.launch_emulator()
                elif status == "running":
                    if config.conf.emulator.hotkey:
                        hotkey = config.conf.emulator.hotkey
                        hotkey = hotkey.split("+")
                        import pyautogui

                        pyautogui.FAILSAFE = False
                        pyautogui.hotkey(*hotkey)
                    csleep(5)  # adb不一定能完全连
                    return True
                csleep(5)
            logger.error("启动模拟器超时")
            if self.emulator_status() == "launching":
                self.shutdown_emulator()
            return False