opt/tools/emulator/ld_operate.py

import json
import os

from typing import TextIO
from model.custom_struct import EmulatorInfo
from tools.emulator.emulator import Emulator
from tools.utils import Utils


class LD(Emulator):
    def __init__(self, system_type: int):
        self.install_path = self.get_ld_install_dir(system_type)
        super().__init__(self.install_path)
        self.emulator_type = 'ld'
        self.system_type = system_type

        if not os.path.exists(self.install_path + r'\dnconsole.exe'):
            raise ValueError(f"{self.emulator_type}-{system_type},安装路径错误")

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

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

    def dnc(self, cmd: str) -> str:
        return self._execute_cmd('dnconsole.exe" ' + cmd)

    def shell(self, index: int, cmd: str) -> str:
        return self._execute_cmd(f'ld.exe" -s {index} "{cmd}"').rstrip('\n')

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

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

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

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

    def add(self, name: str):
        self.dnc(f'add --name {name}')

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

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

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

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

    def is_running(self, index: int) -> bool:
        return self.dnc(f'isrunning --index {index}') == 'running'

    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.dnc(f'modify --index {index} {formatted_args}')

    def get_manufacturer(self, index:int):
        return self.shell(index, 'getprop ro.product.manufacturer')

    def get_model(self,index:int):
        return self.shell(index, 'getprop ro.product.model')

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

    def get_android_id(self, index:int):
        return self.shell(index, 'getprop phone.androidid')

    def get_IMEI(self, index:int):
        return self.shell(index, 'getprop phone.imei')

    def get_IMSI(self, index:int):
        return self.shell(index, 'getprop phone.imsi')

    def get_mac(self, index:int):
        if self.system_type == 9:
            return self.shell(index, 'cat /sys/class/net/wlan0/address')
        else:
            return self.shell(index, 'cat /sys/class/net/eth0/address')

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

    def get_sim_serial(self, index:int):
        return self.shell(index, 'getprop phone.simserial')

    def get_phone_number(self, index:int):
        return self.shell(index, 'getprop phone.number')

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

    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]:
        emu_list = []
        info_str = self.dnc('list2').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[1], top_handle=int(tmp_list[2]),
                    bind_handle=int(tmp_list[3]), is_enter_android=int(tmp_list[4]),
                    pid=int(tmp_list[5]), vbox_pid=int(tmp_list[6]), width=None,
                    height=None, dpi=None
                )
                emu_list.append(info)
            elif len(tmp_list) == 10:
                info = EmulatorInfo(
                    index=int(tmp_list[0]), name=tmp_list[1], top_handle=int(tmp_list[2]),
                    bind_handle=int(tmp_list[3]), is_enter_android=int(tmp_list[4]),
                    pid=int(tmp_list[5]), vbox_pid=int(tmp_list[6]), width=int(tmp_list[7]),
                    height=int(tmp_list[8]), dpi=int(tmp_list[9])
                )
                emu_list.append(info)
        return emu_list

    def get_info(self, index: int) -> EmulatorInfo | None:
        info_str = self.dnc('list2').strip()
        for line in info_str.split("\n"):
            if line:
                tmp_list = line.split(',')
                if int(tmp_list[0]) == index:
                    if len(tmp_list) == 7:
                        return EmulatorInfo(
                            index=int(tmp_list[0]), name=tmp_list[1], top_handle=int(tmp_list[2]),
                            bind_handle=int(tmp_list[3]), is_enter_android=int(tmp_list[4]),
                            pid=int(tmp_list[5]), vbox_pid=int(tmp_list[6]), width=None,
                            height=None, dpi=None
                        )
                    elif len(tmp_list) == 10:
                        return EmulatorInfo(
                            index=int(tmp_list[0]), name=tmp_list[1], top_handle=int(tmp_list[2]),
                            bind_handle=int(tmp_list[3]), is_enter_android=int(tmp_list[4]),
                            pid=int(tmp_list[5]), vbox_pid=int(tmp_list[6]), width=int(tmp_list[7]),
                            height=int(tmp_list[8]), dpi=int(tmp_list[9])
                        )
                    raise f"Emulator {index} not found"
        return None

    def set_share_dir(self, index: int, path: str, dir_type: int = 1) -> bool:
        """
        设置模拟器的共享目录。

        :param index: 模拟器索引
        :param path: 共享目录路径
        :param dir_type: 共享目录类型（1: 图片, 2: 应用, 3: 杂项）
        :return: 成功返回 True，失败返回 False
        """

        # 共享目录类型映射
        key_map = {
            1: 'statusSettings.sharedPictures',
            2: 'statusSettings.sharedApplications',
            3: 'statusSettings.sharedMisc'
        }

        # 获取对应的 key
        key = key_map.get(dir_type, 'statusSettings.sharedPictures')

        # 处理路径分隔符
        path = path.replace("\\", "/")

        # 拼接配置文件路径
        config_file = os.path.join(self.install_path, 'vms', 'config', f'leidian{index}.config')

        if not os.path.exists(config_file):
            raise FileNotFoundError(f"Config file not found: {config_file}")

        try:
            # 读取配置文件
            with open(config_file, 'r', encoding='utf-8') as file:
                config = json.load(file)

            # 设置共享目录
            config[key] = path

            # 写回配置文件
            with open(config_file, 'w', encoding='utf-8') as file: # type: TextIO
                json.dump(config, file, ensure_ascii=False, indent=4)

            return True
        except FileNotFoundError:
            raise FileNotFoundError(f"Config file not found: {config_file}")
        except json.JSONDecodeError:
            raise json.JSONDecodeError(f"Error decoding JSON from {config_file}")
        except Exception as e:
            raise Exception(f"An unexpected error occurred: {e}")

    def get_share_dir(self, index: int, dir_type: int = 1):
        key_map = {
            1: 'statusSettings.sharedPictures',
            2: 'statusSettings.sharedApplications',
            3: 'statusSettings.sharedMisc'
        }
        key = key_map.get(dir_type, 'statusSettings.sharedPictures')
        config_file = self.install_path + f'\\vms\\config\\leidian{index}.config'
        try:
            with open(config_file, 'r', encoding='utf-8') as file:
                config = json.load(file)
            return True, config[key]
        except FileNotFoundError:
            raise FileNotFoundError(f"Config file not found: {config_file}")
        except json.JSONDecodeError:
            raise json.JSONDecodeError(f"Error decoding JSON from {config_file}")
        except Exception as e:
            raise Exception(f"An unexpected error occurred: {e}")

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

    @staticmethod
    def get_ld_install_dir(_type: int) -> str:
        """
        获取雷电模拟器的安装路径。

        :param _type: 模拟器类型 (4 表示雷电模拟器4, 9 表示雷电模拟器9)
        :return: 安装路径字符串，若未找到则返回空字符串
        """
        install_path = ""

        if _type == 4:
            install_path = Utils.get_registry_value(r"SOFTWARE\leidian\ldplayer", "InstallDir")
            if not install_path:
                install_path = Utils.get_install_dir_from_shortcut("雷电模拟器4.lnk").replace("dnplayer.exe", "")
        if _type == 64:
            install_path = Utils.get_registry_value(r"SOFTWARE\leidian\ldplayer64", "InstallDir")
            if not install_path:
                install_path = Utils.get_install_dir_from_shortcut("雷电模拟器64.lnk").replace("dnplayer.exe", "")
        elif _type == 9:
            install_path = Utils.get_registry_value(r"SOFTWARE\leidian\ldplayer9", "InstallDir")
            if not install_path:
                install_path = Utils.get_install_dir_from_shortcut("雷电模拟器9.lnk").replace("dnplayer.exe", "")

        if not install_path:
            raise FileNotFoundError(f"雷电模拟器-{_type}:安装路径未找到")

        return install_path

    @staticmethod
    def get_device_ip_by_index(index: int):
        return 'emulator-' + str(5554 + 2 * index)

    @staticmethod
    def get_index_by_device_ip(device_ip: str):
        return (int(device_ip.replace('emulator-', '')) - 5554) // 2