安卓自动化之minicap截图

安卓自动化之minicap截图

关于安卓自动化使用找图方法点击时，最大的痛点是使用原生adb截图速度太慢了，大概需要3s的时间，再加上我们使用opencv的找图算法，时间就去都三秒多了，为了解决这一个痛点，我们就可以使用minicap，他是通过在安卓常驻一个miniservice，通过adb跟我们的程序进行通信，它实时获取屏幕信息转化为数据流发送给我们的程序，我们只需实时去接收数据并把它转化成图片即可
1.第一步当然是先在安卓上安装我们的miniservice

pip install uiautomator2 
python -m uiautomator2 init 
or
import uiautomator2 as u2
device =u2.connect()

我们在初始化的时候或者使用uiautomator2时会默认安装适合我们安卓架构的minicapservice
2.第二部使用adb命令启动安卓中的minicapservice

adb forward tcp:1717 localabstract:minicap
adb shell LD_LIBRARY_PATH=/data/local/tmp /data/local/tmp/minicap -P 6480x960@6480x960/0

6480x960@6480x960 是我们安卓设备的屏幕像素

adb shell wn size

3.第三步，编写接收数据，并转换成图片

import socket
import sys
import struct
from collections import OrderedDict


class Banner:
    def __init__(self):
        self.__banner = OrderedDict(
            [('version', 0),
             ('length', 0),
             ('pid', 0),
             ('realWidth', 0),
             ('realHeight', 0),
             ('virtualWidth', 0),
             ('virtualHeight', 0),
             ('orientation', 0),
             ('quirks', 0)
             ])

    def __setitem__(self, key, value):
        self.__banner[key] = value

    def __getitem__(self, key):
        return self.__banner[key]

    def keys(self):
        return self.__banner.keys()

    def __str__(self):
        return str(self.__banner)


class MiniCap:
    def __init__(self, host, port, banner):
        self.buffer_size = 4096
        self.host = host
        self.port = port
        self.banner = banner
        self.socket = None
        self.stop_flag = False

    def connect(self):
        """
        Connect to the specified host and port.

        Raises:
            socket.error: If there is any error creating the socket or connecting.
        """
        try:
            self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            self.socket.connect((self.host, self.port))
        except socket.error as e:
            print(f"Failed to connect to {self.host}:{self.port}. Error: {e}")
            raise e
    @classmethod
    def save_image(cls, data):
        file_name = 'received_image.jpg'  # 图片名
        with open(file_name, 'wb') as f:
            for b in data:
                f.write(b.to_bytes(1, 'big'))

    def consume(self,stop_flag):
        readBannerBytes = 0
        bannerLength = 24
        readFrameBytes = 0
        frameBodyLength = 0
        data = []
        while not stop_flag:
            try:
                chunk = self.socket.recv(self.buffer_size)
            except socket.error as e:
                print(e)
                sys.exit(1)
            cursor = 0
            buf_len = len(chunk)
            while cursor < buf_len:
                if readBannerBytes < bannerLength:
                    map(lambda i, val: self.banner.__setitem__(self.banner.keys()[i], val),
                        [i for i in range(len(self.banner.keys()))], struct.unpack("<2b5ibB", chunk))
                    cursor = buf_len
                    readBannerBytes = bannerLength
                elif readFrameBytes < 4:
                    frameBodyLength += (chunk[cursor] << (readFrameBytes * 8)) >> 0
                    cursor += 1
                    readFrameBytes += 1
                else:
                    if buf_len - cursor >= frameBodyLength:
                        data.extend(chunk[cursor:cursor + frameBodyLength])
                        self.save_image(data)
                        cursor += frameBodyLength
                        frameBodyLength = readFrameBytes = 0
                        data = []
                    else:
                        data.extend(chunk[cursor:buf_len])
                        frameBodyLength -= buf_len - cursor
                        readFrameBytes += buf_len - cursor
                        cursor = buf_len

    def run(self,stop_flag):
        """
        Connect to the specified host and port, then consume all frames from the socket.
        Raises:
            socket.error: If there is any error creating the socket or connecting.
        """
        try:
            self.connect()
            self.consume(stop_flag)
        except socket.error as e:
            print(f"Failed to connect to {self.host}:{self.port}. Error: {e}")
            raise e
 if '__main__' == __name__:
    # time.sleep(1)
    mc = MiniCap('localhost', 1717, Banner())
    mc.run(True)

当我们运行程序的时候就会在当前目录生成一个图片received_image.jpg，这个时候我们就完成了截图啦，但是这个图片实时更新的我们的其他程序是没有办法去读取图片的，这个时候就需要我们去截图啦，什么还要去截图，太麻烦了吧，不，准确的来说只是copy一下啦，截图就是copy一下，谁能有我截图快，快如电，嗖嗖嗖的

  def mini_cap_screen(self, local_path, time_out):
        """
        mini cap截图
        Args:
            local_path: 目标路径
            time_out: 超时时间（秒）
        Returns:
            local_path 或 None
        """
        src_path = "received_image.jpg"
        end_time = datetime.now() + timedelta(seconds=time_out)
        start_time = datetime.now()
        self.logger.info(f"mini cap 截图开始时间: {start_time}")

        while datetime.now() < end_time:
            if os.path.exists(src_path) and os.path.getsize(src_path) > 200000:
                os.makedirs(os.path.dirname(local_path), exist_ok=True)
                shutil.copy2(src_path, local_path)
                self.logger.info(f"Image copied to {local_path}")

                current_time = datetime.now()
                self.logger.info(f"mini cap 截图结束时间: {current_time}")
                self.logger.info(f"mini cap 截图耗时: {current_time - start_time}")

                return local_path
            time.sleep(0.1)

        self.logger.error("mini cap 截图失败，请检测mini cap service is running")
        return None

本人天下第一快，NO，我说的是程序，经过计算我们截图加opencv找图然后在模拟点击只需250毫秒，嘎嘎快