A simple minicap client with python3

01-yet-another-minicap-python3-client.md

minicap-python3-client

• The following are heavily based on the code from openstf/minicap#96.

minicap installation

Plan A: Use minicap-prebuilt

$ npm init -y
$ npm i -S minicap-prebuilt
$ ls node_modules/minicap-prebuilt/prebuilt/

Plan B: Build it by yourself

$ docker-compose run --rm minicap
$ ls minicap/

Device initialization for lazy guys

    def init_device(self, port):
        # Get ABI & SDK version
        res = subprocess.run(
            ['adb', 'shell', 'getprop', 'ro.product.cpu.abi'], stdout=subprocess.PIPE)
        abi = res.stdout.decode("utf8").replace("\n", "")
        res = subprocess.run(
            ['adb', 'shell', 'getprop', 'ro.build.version.sdk'], stdout=subprocess.PIPE)
        sdk = res.stdout.decode("utf8").replace("\n", "")

        # Copy binaries
        # basepath = "node_modules/minicap-prebuilt/prebuilt"
        basepath = "minicap/prebuilt"
        res = subprocess.run(
            ['adb', 'push', f'{basepath}/{abi}/bin/minicap', '/data/local/tmp/'], stdout=subprocess.PIPE, check=True)
        res = subprocess.run(
            ['adb', 'push', f'{basepath}/{abi}/lib/android-{sdk}/minicap.so', '/data/local/tmp/'], stdout=subprocess.PIPE, check=True)

        # Start minicap & forwarding
        res = subprocess.Popen(
            ['adb', 'shell', 'LD_LIBRARY_PATH=/data/local/tmp', '/data/local/tmp/minicap', '-P', '1080x2340@1080x2340/90'], stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
        while res.poll() is None:
            line = res.stdout.readline().decode("utf8")
            print("log", line)
            if "Publishing virtual display" in line:
                break
        subprocess.run(
            ['adb', 'forward', f'tcp:{port}', 'localabstract:minicap'], stdout=subprocess.PIPE)

docker-compose.yml

version: "3"
services:
  minicap:
    build: .
    volumes:
      - ".:/workdir:rw"
    working_dir: /workdir
    user: 1000:1000
    command: "cp -R /minicap ."

Dockerfile

FROM lakoo/android-ndk as builder

RUN apt-get update && apt-get -y install \
  make \
  && apt-get clean \
  && rm -rf /var/lib/apt/lists/*

RUN cd / && \
  git clone https://github.com/DeviceFarmer/minicap --recursive

WORKDIR /minicap
RUN export PATH=/opt/android-ndk-linux:${PATH} && make

FROM busybox:latest
COPY --from=builder /minicap /minicap

minicap-client-cv2.py

import socket
import threading
import cv2
import numpy as np
from queue import LifoQueue


class Minicap(object):
    def __init__(self, host='127.0.0.1', port=1313):
        # self.init_device(port)
        self.connection = socket.create_connection((host, port))

    def read_bytes(self, socket, length):
        data = bytearray()
        while length > 0:
            tmp = socket.recv(length)
            length -= len(tmp)
            data.extend(tmp)
        return bytearray(data)

    def read_frames(self):
        socket = self.connection
        version = self.read_bytes(socket, 1)[0]
        print("Version {}".format(version))
        banner_length = self.read_bytes(socket, 1)[0]
        banner_rest = self.read_bytes(socket, banner_length - 2)
        print("Banner length {}".format(banner_length))

        while True:
            frame_bytes = self.read_bytes(socket, 4)
            total = int.from_bytes(frame_bytes, byteorder="little")
            print("JPEG data: {}".format(total))
            jpeg_data = self.read_bytes(socket, total)
            yield jpeg_data


def minicapThread(minicap: Minicap, q: LifoQueue):
    for frame in minicap.read_frames():
        while not q.empty():  # FIXME: May be better to do this at Consumer
            q.get()
        q.put(frame)


minicap = Minicap()
q = LifoQueue()

threading.Thread(target=minicapThread, daemon=True,
                 args=(minicap, q, )).start()

while True:
    frame = q.get()
    img = np.array(bytearray(frame))
    img = cv2.imdecode(img, 1)

    cv2.imshow('capture', img)
    if cv2.waitKey(1) & 0xFF == ord('q'):
        cv2.destroyAllWindows()
        exit(0)

Alternatively you can also use; https://github.com/leng-yue/py-scrcpy-client

from queue import LifoQueue
import scrcpy
from adbutils import adb


def start_scrcpy():
    q = LifoQueue()
    client = scrcpy.Client(device=adb.device_list()[0])

    def on_frame(frame):
        if frame is not None:
            while not q.empty():
                q.get()
            q.put(frame)

    client.add_listener(scrcpy.EVENT_FRAME, on_frame)
    client.start(daemon_threaded=True)

    return client, q