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
from utils.control_util import loading


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 = kwargs.get('log')
            _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")
                        log_widget.see(tk.END)
            self.devices_list(*args, **kwargs)

        # loading 相对的元素
        top = kwargs['top'] if 'top' in kwargs else None

        if top:
            loading(top, func)
        else:
            func()

    def _init_connect(self, cmds: List[str]):
        """
        初始化一些连接,将信息保存到 connect_dict
        :param cmds: 要执行的命令行 ['adb', 'connect', '127.0.0.1:64079']
        :return:
        """
        try:
            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])
        except Exception as e:
            logging.error(f"无法连接:{e}")
        pass

    def test_device(self):
        """
        测试连接
        :return:
        """

        pass

    def devices_list(self, *args, **kwargs) -> List[str]:
        """
        获取设备列表
        :param command:
        :return:
        """
        log = kwargs.get('log')
        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}")
        if log:
            log.insert(tk.END, f"===检查结果:====\n{result.stdout}\n===============\n")
            log.see(tk.END)

        a = [x for x in devices if x and ':' in x]
        # 去除每一项中的 \tdevice 并且过滤offline数据
        a = [x.split('\t')[0] for x in a if x.split('\t')[1] != 'offline']

        for k, v in self.connect_dict.items():

            if k not in a:
                logging.info(f"移除掉线的Key:{k}")
                self.connect_dict.pop(k)

        return a