opt/tools/emulator/ys_operate.py

import os
import threading
import winreg

from tools.emulator.emulator import Emulator
from model.custom_struct import EmulatorInfo
from tools.ini_operate import ConfigIni
from tools.log import logger
from tools.utils import Utils
# from tools.window_operate import WindowOperate


class YS(Emulator):
    def __init__(self,system_type=9):
        self.install_path = self.get_ys_install_dir()
        super().__init__(self.install_path)
        self.emulator_type = 'ys'
        self.system_type = system_type
        if not os.path.exists(self.install_path + r'\NoxConsole.exe'):
            raise ValueError("模拟器安装路径错误")
        self.config_path = os.path.join(os.getenv('LOCALAPPDATA'), 'Nox')
        self._lock = threading.Lock()

    def set_install_path(self, path: str):
        self.install_path = path

    def get_install_path(self) -> str:
        return self.install_path

    def noc(self, cmd: str) -> str:
        return self._execute_cmd(r'\NoxConsole.exe" ' + cmd)

    def adb(self, cmd: str,_timeout=60) -> str:
        return self._execute_cmd(r'\nox_adb.exe" ' + cmd)

    def shell(self, index: int, cmd: str) -> str:
        return self.adb(f'-s {self.get_device_ip_by_index(index)} shell {cmd}')

    def start(self, index: int):
        super().start(index)
        self.noc(f'launch -index:{index}')

    def is_started(self, index: int) -> tuple[bool, EmulatorInfo | None]:
        emulator_info = None
        result = self.is_enter_android(index)
        if result:
            emulator_info = self.get_info(index)
        return result, emulator_info

    def is_exists(self, index: int) -> bool:
        return os.path.exists(os.path.join(self.get_install_path(), 'BignoxVMS', f'Nox_{index}'))

    def close(self, index: int):
        super().close(index)
        self.noc(f'quit -index:{index}')

    def close_all(self):
        self.noc(f'quitall')

    def add(self, name: str):
        self.noc(f'add -name:{name} -systemtype:{self.system_type}')

    def remove(self, index: int):
        self.noc(f'remove -index:{index}')

    def rename(self, index: int, name: str):
        self.noc(f'rename -index:{index} -title:{name}')

    def restore(self, index: int, file_path: str):
        self.noc(f'restore -index:{index} -file:{file_path}')

    def is_enter_android(self, index: int) -> bool:
        return self.shell(index, 'getprop sys.boot_completed').strip() == '1'

    def read_emulator_file(self, index: int, file_path: str) -> str:
        """
        读取模拟器文本文件

        :param index: 模拟器索引
        :param file_path:模拟器文件路径
        :return:
        """
        return super().read_emulator_file(index, file_path)

    def write_emulator_file(self, index: int, file_path: str, content: str) -> bool:
        """
        写入模拟器文本文件

        :param index: 模拟器索引
        :param file_path:模拟器文件路径
        :param content: 写入内容
        :return:
        """
        return super().write_emulator_file(index, file_path, content)

    def get_list(self) -> list[EmulatorInfo]:
        """
        获取模拟器列表

        :return:
        """

        info_list = []
        info_str = self.noc('list').strip()
        for line in info_str.split("\n"):
            tmp_list = line.split(',')
            if len(tmp_list) >= 7:
                info = EmulatorInfo(index=int(tmp_list[0]), name=tmp_list[2], top_handle=int(tmp_list[3], 16),
                                bind_handle=int(tmp_list[4], 16), is_enter_android=None, pid=int(tmp_list[5]),
                                vbox_pid=int(tmp_list[6]), width=None, height=None, dpi=None)
                info_list.append(info)
        return info_list

    def get_info(self, index: int) -> EmulatorInfo | None:
        info_str = self.noc('list').strip()
        for line in info_str.split("\n"):
            if line:
                tmp_list = line.split(',')
                # if int(tmp_list[0]) == index:
                #     tmp_handle = WindowOperate.find_subwindow_by_title_and_class(int(tmp_list[3], 16),'sub','subWin')
                #     bind_handle = WindowOperate.get_parent_window(tmp_handle)
                #     return EmulatorInfo(index=int(tmp_list[0]), name=tmp_list[2], top_handle=int(tmp_list[3], 16),
                #                     bind_handle=bind_handle, is_enter_android=None, pid=int(tmp_list[5]),
                #                     vbox_pid=int(tmp_list[6]), width=None, height=None, dpi=None)
        return None

    def modify(self,index,cpu=None, memory=None, resolution=None, manufacturer=None, model=None, pnumber=None,
               imei=None, imsi=None, simserial=None, androidid=None, mac=None, autorotate=None, lockwindow=None, root=None):
        """
        修改模拟器机型
        :param index: 模拟器索引
        :param args:
            [--resolution <w,h,dpi>]
            [--cpu <1 | 2 | 3 | 4>]
            [--memory <256 | 512 | 768 | 1024 | 1536 | 2048 | 4096 | 8192>]
            [--manufacturer asus]
            [--model ASUS_Z00DUO]
            [--pnumber 13800000000]
            [--imei <auto | 865166023949731>]
            [--imsi <auto | 460000000000000>]
            [--simserial <auto | 89860000000000000000>]
            [--androidid <auto | 0123456789abcdef>]
            [--mac <auto | 000000000000>]
            [--autorotate <1 | 0>  自动旋转
            [--lockwindow <1 | 0>  锁定窗口大小
            [--root <1 | 0>
        """
        formatted_args = " ".join([f"-{key}:{value}"  for key, value in  locals().items() if value and key not in ['index', 'self']])
        self.noc(f'modify -index:{index} {formatted_args}')
        self.set_config(index, 'vm_androidid', androidid)


    def get_config(self,index:int, key:str):
        try:
            config_file = os.path.join(self.config_path, f'clone_Nox_{index}_conf.ini')
            if os.path.exists(config_file):
                config = ConfigIni(config_file)
                return config.get('setting', key)
            return None
        except Exception as e:
            logger.exception("get_config error: {e}")
            return None

    def set_config(self,index:int, key:str,value:str):
        try:
            config_file = os.path.join(self.config_path, f'clone_Nox_{index}_conf.ini')
            if os.path.exists(config_file):
                config = ConfigIni(config_file)
                config.set('setting', key, value)
        except Exception as e:
            logger.exception("get_config error: {e}")

    def get_manufacturer(self, index: int):
        return self.shell(index, 'getprop persist.nox.manufacturer').strip()

    def get_model(self, index: int):
        return self.shell(index, 'getprop persist.nox.model').strip()

    def get_android_version(self, index: int):
        return self.shell(index, 'getprop ro.build.version.release').strip()

    def get_android_id(self, index: int):
        return self.shell(index, 'getprop persist.nox.androidid').strip()

    def get_IMEI(self, index: int):
        return self.shell(index, 'getprop persist.nox.modem.imei').strip()

    def get_IMSI(self, index: int):
        return self.shell(index, 'getprop persist.nox.modem.imsi').strip()

    def get_mac(self, index: int):
        return self.shell(index, 'cat /sys/class/net/wlan0/address').strip()

    def get_resolution(self, index: int):
        return self.shell(index, 'wm size')

    def get_sim_serial(self, index: int):
        return self.shell(index, 'persist.nox.modem.serial').strip()

    def get_phone_number(self, index: int):
        return self.shell(index, 'persist.nox.modem.phonumber').strip()

    def get_ip(self, index: int):
        tmp_str = self.shell(index, 'ifconfig eth0')
        return Utils.extract_between_strings(tmp_str, 'addr:', ' ').strip()

    def set_share_dir(self, index: int, path: str, dir_type=None) -> bool:
        logger.info(f"Setting shared directory for emulator {index} to {path}")
        try:
            config_file = os.path.join(self.config_path, f'clone_Nox_{index}_conf.ini')
            if os.path.exists(config_file):
                config = ConfigIni(config_file)
                config.set('setting', 'share_path', path)
                return True
            return False
        except FileNotFoundError:
            logger.exception("The file was not found.")
        except Exception as e:
            logger.exception(f"An unexpected error occurred: {e}")
        return False

    def get_share_dir(self, index: int):
        try:
            config_file = os.path.join(self.config_path, f'clone_Nox_{index}_conf.ini')
            if os.path.exists(config_file):
                config = ConfigIni(config_file)
                share_path = config.get('setting', 'share_path')
                if share_path:
                    return True, share_path
                return False, None
            return False, None
        except FileNotFoundError:
            logger.exception("The file was not found.")
        except Exception as e:
            logger.exception(f"An unexpected error occurred: {e}")
        return False

    def start_app(self, index, packagename):
        return self.noc(f"runapp -index:{index} -packagename:{packagename}")

    @staticmethod
    def get_ys_install_dir() -> str:
        temp_path = Utils.get_registry_value(r"SOFTWARE\WOW6432Node\DuoDianOnline\SetupInfo", "InstallPath",winreg.HKEY_LOCAL_MACHINE)
        if temp_path:
            install_path = os.path.join(temp_path, "bin")
        else:
            install_path = Utils.get_install_dir_from_shortcut(r'\夜神模拟器.lnk').replace(r'\Nox.exe', "")
        return install_path

    @staticmethod
    def get_device_ip_by_index(index: int):
        return '127.0.0.1:' + str(62024 + index)

    @staticmethod
    def get_index_by_device_ip(device_ip: str):
        return int(device_ip.replace('127.0.0.1:', '')) - 62024