import adbutils adb = adbutils.AdbClient(host="127.0.0.1", port=5037) for info in adb.list(extended=True): print(info.serial, info.state) # <serial> <device|offline> print(info.transport_id) # only list state=device print(adb.device_list()) # Set socket timeout to 10 (default None) adb = adbutils.AdbClient(host="127.0.0.1", port=5037, socket_timeout=10) print(adb.device_list())

from adbutils import adb for d in adb.device_list(): print(d.serial) # print device serial d = adb.device(serial="33ff22xx") # or d = adb.device(transport_id=24) # transport_id can be found in: adb devices -l # You do not need to offer serial if only one device connected # RuntimeError will be raised if multi device connected d = adb.device()

output = adb.connect("127.0.0.1:5555") print(output) # output: already connected to 127.0.0.1:5555 # connect with timeout try: adb.connect("127.0.0.1:5555", timeout=3.0) except AdbTimeout as e: print(e) adb.disconnect("127.0.0.1:5555") adb.disconnect("127.0.0.1:5555", raise_error=True) # if device is not present, AdbError will raise # wait-for-device adb.wait_for("127.0.0.1:5555", state="device") # wait for device online, state default value is "device" adb.wait_for("127.0.0.1:5555", state="disconnect") # wait device disconnect

# forward local address to remote address d.forward("tcp:9999", "localabstract:scrcpy") # forward remote address to local random port port = d.forward_port("localabstract:scrcpy") print(port) # 54622 (random) # remove forwarded connection d.forward_remove("tcp:9999") # use local address # remove all forwarded connections tied to specific device d.forward_remove_all() # reverse (forward) a remote address to a local address d.reverse("localabstract:scrcpy", "tcp:9999") # remove reversed connection d.reverse_remove("localabstract:scrcpy") # remove all reversed connection tied to specific device d.reverse_remove_all() # list all forwards for item in adb.forward_list(): print(item.serial, item.local, item.remote) # 8d1f93be tcp:10603 tcp:7912 # 12345678 tcp:10664 tcp:7912 # list only one device forwards for item in adb.forward_list("8d1f93be"): print(item.serial, item.local, item.remote) # 8d1f93be tcp:10603 tcp:7912 # 12345678 tcp:10664 tcp:7912 # list all reversed forwards for item in adb.reverse_list(): print(item.serial, item.local, item.remote) # 8d1f93be tcp:9999 localabstract:scrcpy # 12345678 tcp:9999 localabstract:scrcpy # 监控设备连接 track-devices for event in adb.track_devices(): print(event.present, event.serial, event.status) ## When plugin two device, output # True WWUDU16C22003963 device # True bf755cab device # False bf755cab absent # When adb-server killed, AdbError will be raised

# minitouch: https://github.com/openstf/minitouch c = d.create_connection("unix", "minitouch") print(c.recv(500)) c.close()

c = d.create_connection("tcp", 7912) # the second argument must be int c.send(b"GET / HTTP/1.1\r\nHost: localhost\r\n\r\n") print(c.recv(500)) c.close()

# read device file with d.create_connection(adbutils.Network.DEV, "/data/local/tmp/hello.txt") as c: print(c.recv(500))

import io from adbutils import adb d = adb.device() print(d.serial) # 获取序列号 # Argument support list, str serial = d.shell(["getprop", "ro.serial"]) # 获取Prop信息 # Same as serial = d.shell("getprop ro.serial") # Set timeout for shell command d.shell("sleep 1", timeout=0.5) # Should raise adbutils.AdbTimeout # Open shell as connection c = d.open_shell('cat') # AdbConnection, use c.conn to get socket.socket object c.send(b'hello\n') print(c.recv(100)) c.close() # Expect output: b'hello\n' # The advanced shell (returncode archieved by add command suffix: ;echo EXIT:$?) ret = d.shell2("echo 1") print(ret) # expect: ShellReturn(args='echo 1', returncode=0, output='1\n') # Advanced shell using the shell v2 protocol ret = d.shell2("echo 1; echo 2 1>&2", v2=True) print(ret) # expect: ShellReturn(command='echo 1; echo 2 1>&2', returncode=0, output='1\n2\n', stderr='2\n', stdout='1\n') # show property, also based on d.shell print(d.prop.name) # output example: surabaya d.prop.model d.prop.device d.prop.get("ro.product.model") d.prop.get("ro.product.model", cache=True) # a little faster, use cache data first d.get_serialno() # same as adb get-serialno d.get_devpath() # same as adb get-devpath d.get_state() # same as adb get-state

# Method 1 (Recommend) pil_image = d.screenshot() # default display_id=0, error_ok=True try: pil_image = d.screenshot(display_id=1, error_ok=False) except AdbError: print("failed to takeScreenshot") # Method 2 # adb exec-out screencap -p p.png png_data = d.shell("screencap -p", encoding=None) pathlib.Path("p.png").write_bytes(png_data)

d.sync.push(b"Hello Android", "/data/local/tmp/hi.txt") # 推送二进制文本 d.sync.push(io.BytesIO(b"Hello Android"), "/data/local/tmp/hi.txt") # 推送可读对象Readable object d.sync.push("/tmp/hi.txt", "/data/local/tmp/hi.txt") # 推送本地文件 d.sync.push(pathlib.Path("/tmp/hi.txt"), "/data/local/tmp/hi.txt") # 推送本地文件 # 读取文件 for chunk in d.sync.iter_content("/data/local/tmp/hi.txt"): print("Chunk", chunk) d.sync.push(b"Hello world", "/data/local/tmp/hi.txt") output = d.sync.read_text("/data/local/tmp/hi.txt", encoding="utf-8") # Expect output: "Hello world" output = d.sync.read_bytes("/data/local/tmp/hi.txt") # Expect output: b"Hello world" # 拷贝到本地 d.sync.pull("/data/local/tmp/hi.txt", "hi.txt") # 获取包的信息 info = d.app_info("com.example.demo") if info: print(info) # output example: # { # "version_name": "1.2.3", "version_code": "12", "signature": "0xff132",  # "first_install_time": datetime-object, "last_update_time": datetime-object, # }

# save screenshot pilimg = d.screenshot() pilimg.save("screenshot.jpg") # get current app info app_info = d.app_current() print(app_info.package) print(app_info.activity) print(app_info.pid) # might be 0 # install apk d.install("apidemo.apk") # use local path d.install("http://example.com/apidemo.apk") # install from url # raise AdbInstallError if something went wrong # simulate click d.click(100, 100) d.click(0.5, 0.5) # center, should be float and <= 1.0 # swipe from(10, 10) to(200, 200) 500ms d.swipe(10, 10, 200, 200, 0.5) # drag and drop from(10, 10) to(200, 200) 3000ms d.drag(10, 10, 200, 200, 3) d.list_packages() # example output: ["com.example.hello"] d.window_size() # example output: (1080, 1920) when phone is portrait # example output: (1920, 1080) when phone is landscape d.window_size(landscape=True) # force landscape mode # example output: (1920, 1080) d.rotation() -> int # example output: 0 # 0: natural, 1: left, 2: right, 3: upsidedown d.app_info("com.github.uiautomator") # example output: {"version_name": "1.1.7", "version_code": "1007"} d.keyevent("HOME") d.volume_up() d.volume_down() # default times=1, If you want to adjust the volume multiple times, you can use：d.volume_up(times=xxx) d.volume_mute() # device mute d.send_keys("hello world$%^&*") # simulate: adb shell input text "hello%sworld\%\^\&\*" d.open_browser("https://www.baidu.com") # 打开百度 # There still too many functions, please see source codes # check if screen is on d.is_screen_on() # 返回屏幕是否亮屏 True or False # adb root d.root() # adb tcpip <port> d.tcpip(5555) print(d.battery()) # get battery info BatteryInfo(ac_powered=False, usb_powered=False, wireless_powered=False, dock_powered=False, max_charging_current=0, max_charging_voltage=0, charge_counter=10000, status=4, health=2, present=True, level=100, scale=100, voltage=5000, temperature=25.0, technology='Li-ion') print(d.brightness_value) # get brightness value, return int value in 0-255 d.brightness_value = 100 # set brightness value # you can also set brightness mode from adbutils import BrightnessMode print(d.brightness_mode) # output BrightnessMode.AUTO or BrightnessMode.MANUAL d.brightness_mode = BrightnessMode.MANUAL # set brightness mode is manual d.brightness_mode = BrightnessMode.AUTO # set brightness mode is auto

d.start_recording("video.mp4") time.sleep(5) d.stop_recording()

# filter logcat to file logcat = d.logcat("logcat.txt", clear=True, re_filter=".*FA.*") # clear default False # do something else logcat.stop(timeout=3) # tell thread to stop write, wait for 3s, if not stopped, raise TimeoutError logcat.stop_nowait() # tell thread to stop write and close file

stream = d.shell("screenrecord /sdcard/s.mp4", stream=True) time.sleep(3) # record for 3 seconds with stream: stream.send(b"\003") # send Ctrl+C stream.read_until_close() start = time.time() print("Video total time is about", time.time() - start) d.sync.pull("/sdcard/s.mp4", "s.mp4") # pulling video

d.shell("logcat --clear") stream = d.shell("logcat", stream=True) with stream: f = stream.conn.makefile() for _ in range(100): # read 100 lines line = f.readline() print("Logcat:", line.rstrip()) f.close()

import adbutils # .... test code here ...