add method create_connection

openatx · Mar 26, 2021 · a1ea667 · a1ea667
1 parent 70d9fc5
commit a1ea667
Show file tree

Hide file tree

Showing 2 changed files with 172 additions and 53 deletions.
diff --git a/README.md b/README.md
@@ -4,6 +4,35 @@
 
 Python adb library for adb service (Only support Python3.6+)
 
+**Table of Contents**
+
+<!--ts-->
+   * [adbutils](#adbutils)
+   * [Install](#install)
+   * [Usage](#usage)
+      * [Connect ADB Server](#connect-adb-server)
+      * [List all the devices and get device object](#list-all-the-devices-and-get-device-object)
+      * [Connect remote device](#connect-remote-device)
+      * [adb forward and adb reverse](#adb-forward-and-adb-reverse)
+      * [Create socket connection to the device](#create-socket-connection-to-the-device)
+      * [Run shell command](#run-shell-command)
+      * [Transfer files](#transfer-files)
+      * [Extended Functions](#extended-functions)
+      * [Run in command line 命令行使用](#run-in-command-line-命令行使用)
+         * [Environment variables](#environment-variables)
+         * [Color Logcat](#color-logcat)
+      * [Experiment](#experiment)
+      * [Examples](#examples)
+   * [Develop](#develop)
+      * [Watch adb socket data](#watch-adb-socket-data)
+   * [Thanks](#thanks)
+   * [Ref](#ref)
+   * [LICENSE](#license)
+
+<!-- Added by: shengxiang, at: 2021年 3月26日 星期五 15时05分04秒 CST -->
+
+<!--te-->
+
 # Install
 ```
 pip3 install adbutils
@@ -79,6 +108,34 @@ for event in adb.track_devices():
 # When adb-server killed, AdbError will be raised
 ```
 
+## Create socket connection to the device
+
+For example
+
+```python
+# minitouch: https://github.com/openstf/minitouch
+c = d.create_connection("unix", "minitouch")
+print(c.recv(500))
+c.close()
+```
+
+```python
+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()
+```
+
+```python
+# read device file
+with d.create_connection(adbutils.Network.DEV, "/data/local/tmp/hello.txt") as c:
+    print(c.recv(500))
+```
+
+There are many other usage, see [SERVICES.TXT](https://cs.android.com/android/platform/superproject/+/master:packages/modules/adb/SERVICES.TXT;l=175) for more details
+
+Thanks for Pull Request from [@hfutxqd](https://github.com/openatx/adbutils/pull/27)
+
 ## Run shell command
 I assume there is only one device connected.
 
@@ -187,57 +244,6 @@ r.stop_and_pull("video.mp4") # stop recording and pull video to local, then remo
 
 For further usage, please read [mixin.py](adbutils/mixin.py) for details.
 
-## Examples
-Record video using screenrecord
-
-```python
-stream = d.shell("screenrecord /sdcard/s.mp4", stream=True)
-time.sleep(3) # record for 3 seconds
-with stream:
-	stream.send("\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
-```
-
-Reading Logcat
-
-```python
-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()
-```
-
-## Develop
-```sh
-git clone https://github.com/openatx/adbutils adbutils
-pip3 install -e adbutils # install as development mode
-```
-
-Now you can edit code in `adbutils` and test with
-
-```python
-import adbutils
-# .... test code here ...
-```
-
-Run tests requires one device connected to your computer
-
-```sh
-# change to repo directory
-cd adbutils
-
-pip3 install pytest
-pytest tests/
-```
-
 ## Run in command line 命令行使用
 
 ```bash
@@ -337,11 +343,83 @@ $ python -m adbutils --install-confirm -i some.apk
 
 For more usage, please see the code for details.
 
+## Examples
+Record video using screenrecord
+
+```python
+stream = d.shell("screenrecord /sdcard/s.mp4", stream=True)
+time.sleep(3) # record for 3 seconds
+with stream:
+	stream.send("\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
+```
+
+Reading Logcat
+
+```python
+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()
+```
+
+# Develop
+```sh
+git clone https://github.com/openatx/adbutils adbutils
+pip3 install -e adbutils # install as development mode
+```
+
+Now you can edit code in `adbutils` and test with
+
+```python
+import adbutils
+# .... test code here ...
+```
+
+Run tests requires one device connected to your computer
+
+```sh
+# change to repo directory
+cd adbutils
+
+pip3 install pytest
+pytest tests/
+```
+
+## Watch adb socket data
+Watch the adb socket data using `socat`
+
+```
+$ socat -t100 -x -v TCP-LISTEN:5577,reuseaddr,fork TCP4:localhost:5037
+```
+
+open another terminal, type the following command then you will see the socket data
+
+```bash
+$ export ANDROID_ADB_SERVER_PORT=5577
+$ adb devices
+```
+
+## Generate TOC
+```bash
+gh-md-toc --insert README.md
+```
+
+<https://github.com/ekalinin/github-markdown-toc>
+
 # Thanks
 - [swind pure-python-adb](https://github.com/Swind/pure-python-adb)
 - [openstf/adbkit](https://github.com/openstf/adbkit)
 - [ADB Source Code](https://github.com/aosp-mirror/platform_system_core/blob/master/adb)
-- ADB Protocols [OVERVIEW.TXT](https://github.com/aosp-mirror/platform_system_core/blob/master/adb/OVERVIEW.TXT) [SERVICES.TXT](https://github.com/aosp-mirror/platform_system_core/blob/master/adb/SERVICES.TXT) [SYNC.TXT](https://github.com/aosp-mirror/platform_system_core/blob/master/adb/SYNC.TXT)
+- ADB Protocols [OVERVIEW.TXT](https://cs.android.com/android/platform/superproject/+/master:packages/modules/adb/OVERVIEW.TXT) [SERVICES.TXT](https://cs.android.com/android/platform/superproject/+/master:packages/modules/adb/SERVICES.TXT) [SYNC.TXT](https://cs.android.com/android/platform/superproject/+/master:packages/modules/adb/SYNC.TXT)
 - [Awesome ADB](https://github.com/mzlogin/awesome-adb)
 - [JakeWharton/pidcat](https://github.com/JakeWharton/pidcat)
 

diff --git a/adbutils/__init__.py b/adbutils/__init__.py
@@ -4,16 +4,18 @@
 from __future__ import print_function
 
 import datetime
+import enum
 import json
 import os
 import re
 import socket
 import stat
 import struct
 import subprocess
+import typing
 from collections import namedtuple
 from contextlib import contextmanager
-from typing import Union, Iterator, Optional
+from typing import ContextManager, Union, Iterator, Optional
 
 import pkg_resources
 import six
@@ -32,6 +34,17 @@
     r'.*DisplayViewport{valid=true, .*orientation=(?P<orientation>\d+), .*deviceWidth=(?P<width>\d+), deviceHeight=(?P<height>\d+).*'
 )
 
+class Network(str, enum.Enum):
+    TCP = "tcp"
+    UNIX = "unix"
+
+    DEV = "dev"
+    LOCAL = "local"
+    LOCAL_RESERVED = "localreserved"
+    LOCAL_FILESYSTEM = "localfilesystem"
+    LOCAL_ABSTRACT = "localabstract" # same as UNIX
+
+
 DeviceEvent = namedtuple('DeviceEvent', ['present', 'serial', 'status'])
 ForwardItem = namedtuple("ForwardItem", ["serial", "local", "remote"])
 ReverseItem = namedtuple("ReverseItem", ["remote", "local"])
@@ -530,6 +543,34 @@ def reverse_list(self):
     def push(self, local: str, remote: str):
         self.adb_output("push", local, remote)
 
+    def create_connection(self, network: Network, address: Union[int, str]):
+        """
+        Used to connect a socket (unix of tcp) on the device
+
+        Returns:
+            socket object
+
+        Raises:
+            AssertionError, ValueError
+        """
+        c = self._client._connect()
+        c.send_command("host:transport:" + self._serial)
+        c.check_okay()
+        if network == Network.TCP:
+            assert isinstance(address, int)
+            c.send_command("tcp:" + str(address))
+            c.check_okay()
+        elif network in [Network.UNIX, Network.LOCAL_ABSTRACT]:
+            assert isinstance(address, str)
+            c.send_command("localabstract:" + address)
+            c.check_okay()
+        elif network in [Network.LOCAL_FILESYSTEM, Network.LOCAL, Network.DEV, Network.LOCAL_RESERVED]:
+            c.send_command(network + ":" + address)
+            c.check_okay()
+        else:
+            raise ValueError("Unsupported network type", network)
+        return c.conn
+
 
 class Sync():
     def __init__(self, adbclient: AdbClient, serial: str):