mobile-rpa/control/ui/settings_control.py

import logging
import subprocess
from typing import List

import tkinter as tk
from adbutils import adb_path

from control.base.base_control import BaseControl
from control.plat.deepcoin_control import DeepCoinControl


class SettingsControl(BaseControl):

    def __init__(self, name: str):

        """
        设置相关的控制器
        :param name:
        """
        super().__init__(name)

    def init_adb(self, commands: List[str], *args, **kwargs):

        def func():
            adb = adb_path()
            logging.info(f'adb_path:{adb}')
            log_widget = args[0] if args else None
            _commons = list(commands)
            _commons.append(f'adb devices')
            for command in _commons:
                command = command.strip()
                if command == '':
                    continue
                logging.info(f'初始化[{self.name}]：{command}')
                try:
                    cmd = command.split(' ')
                    if cmd[1] == 'connect':
                        self._init_connect(cmd)
                    cmd[0] = adb
                    logging.info(f"执行命令：{cmd}")
                    result = subprocess.run(cmd, capture_output=True, text=True, check=True)
                    logging.info(f"命令执行结果：\n{result.stdout}")
                    if log_widget:
                        log_widget.insert(tk.END, f"命令【{command}】执行成功:\n{result.stdout}")



                except subprocess.CalledProcessError as cpe:
                    logging.info(f"命令 '{command}' 执行失败: {cpe}")
                    if log_widget:
                        log_widget.insert(tk.END, f"{cpe}\n")

        top = kwargs['ui']

        if top:
            self.loadding(top, func)
        else:
            func()

    def _init_connect(self, cmds: List[str]):
        """
        初始化一些连接,将信息保存到 connect_dict
        :param cmds: 要执行的命令行 ['adb', 'connect', '127.0.0.1:64079']
        :return:
        """
        cmd = list(cmds).pop()
        d = self.connect_adb(cmd)
        info = d.info
        logging.info(f'设备信息{info}')
        pkg = info['currentPackageName']
        self.connect_dict[cmd] = {
            'd': d,
            'info': info,
            'serial': cmd,
            'status': 'online',
            'pkg': self.app_pkg[pkg],
            'name': str(d.shell('settings get global device_name').output).strip(),
        }
        if pkg == 'com.niocpeed.dna':
            self.connect_dict[cmd]['control'] = DeepCoinControl(serial=cmd, ctx=self.connect_dict[cmd])

        pass

    def devices_list(self) -> List[str]:
        """
        获取设备列表
        :param command:
        :return:
        """
        command = [adb_path(), 'devices']
        logging.info(f"获取设备列表：{command}")
        result = subprocess.run(command, capture_output=True, text=True, check=True)
        devices = result.stdout.splitlines()
        logging.info(f"设备列表：{devices}")
        a = [x for x in devices if x and ':' in x]
        # 去除每一项中的 \tdevice
        a = [x.split('\t')[0] for x in a]
        return a