Skip to content

Instantly share code, notes, and snippets.

@likeablob
Last active May 8, 2024 08:54
Show Gist options
  • Save likeablob/e0b1fb7fb0259da82f7a3536f018ab6f to your computer and use it in GitHub Desktop.
Save likeablob/e0b1fb7fb0259da82f7a3536f018ab6f to your computer and use it in GitHub Desktop.
A simple minicap client with python3

minicap-python3-client

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)
version: "3"
services:
minicap:
build: .
volumes:
- ".:/workdir:rw"
working_dir: /workdir
user: 1000:1000
command: "cp -R /minicap ."
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
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)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment